E2E

README.md

@@ -23,12 +23,16 @@ uvicorn main:app --host 0.0.0.0 --port 8080 --reload --env-file .env
 ## Переменные окружения
 
 - `TELEGRAM_BOT_TOKEN` - обязательно
+- `TELEGRAM_WEBHOOK_SECRET` - обязательно, секрет для заголовка `X-Telegram-Bot-Api-Secret-Token`
 - `GITEA_BASE_URL` - обязательно для назначения ревьюера, например `https://git.example.com`
 - `GITEA_TOKEN` - обязательно для назначения/меток
-- `GITEA_WEBHOOK_SECRET` - необязательно, но рекомендуется
+- `GITEA_WEBHOOK_SECRET` - обязательно, общий секрет webhook от Gitea
 - `REMINDER_TIME` - необязательно, по умолчанию `09:00`
 - `AUTO_ASSIGNED_LABEL` - необязательно, по умолчанию `auto-assigned`
 - `ALLOW_SELF_ASSIGN` - необязательно, по умолчанию `false` (для локального теста можно `true`)
+- `TELEGRAM_ALLOWED_CHAT_IDS` - необязательно, CSV списка `chat_id`, которым можно управлять ботом
+- `TELEGRAM_ALLOWED_USERNAMES` - необязательно, CSV списка Telegram username (без `@`)
+- `GITEA_COUNT_REPOS` - необязательно, CSV репозиториев `org/repo` для расчета нагрузки ревьюеров
 - `BOT_DB_PATH` - необязательно, по умолчанию `botreviewer.sqlite3`
 
 ## Webhook-эндпоинты
@@ -36,7 +40,7 @@ uvicorn main:app --host 0.0.0.0 --port 8080 --reload --env-file .env
 - Обновления Telegram: `POST /telegram/webhook`
 - Webhook Gitea: `POST /gitea/webhook`
 
-Для Telegram укажите URL вашего сервиса с путем `/telegram/webhook`.
+Для Telegram укажите URL вашего сервиса с путем `/telegram/webhook` и `secret_token`.
 Для Gitea включите события `pull_request` и задайте общий секрет.
 
 ## Локальная разработка через Tuna
@@ -71,7 +75,7 @@ https://example.ru.tuna.am/health
 5. Настройте webhook в Telegram:
 
 ```text
-https://api.telegram.org/bot<TELEGRAM_BOT_TOKEN>/setWebhook?url=https://example.ru.tuna.am/telegram/webhook
+https://api.telegram.org/bot<TELEGRAM_BOT_TOKEN>/setWebhook?url=https://example.ru.tuna.am/telegram/webhook&secret_token=<TELEGRAM_WEBHOOK_SECRET>
 ```
 
 6. Проверьте, что webhook установлен:

main.py

@@ -1,4 +1,5 @@
 import asyncio
+from collections import defaultdict
 import hashlib
 import hmac
 import json
@@ -22,6 +23,22 @@ GITEA_WEBHOOK_SECRET = os.getenv("GITEA_WEBHOOK_SECRET", "")
 REMINDER_TIME = os.getenv("REMINDER_TIME", "09:00")  # Формат ЧЧ:ММ
 AUTO_ASSIGNED_LABEL = os.getenv("AUTO_ASSIGNED_LABEL", "auto-assigned")
 ALLOW_SELF_ASSIGN = os.getenv("ALLOW_SELF_ASSIGN", "false").lower() == "true"
+TELEGRAM_WEBHOOK_SECRET = os.getenv("TELEGRAM_WEBHOOK_SECRET", "")
+TELEGRAM_ALLOWED_CHAT_IDS = {
+    int(v.strip())
+    for v in os.getenv("TELEGRAM_ALLOWED_CHAT_IDS", "").split(",")
+    if v.strip().isdigit()
+}
+TELEGRAM_ALLOWED_USERNAMES = {
+    v.strip().lstrip("@").lower()
+    for v in os.getenv("TELEGRAM_ALLOWED_USERNAMES", "").split(",")
+    if v.strip()
+}
+GITEA_COUNT_REPOS = {
+    v.strip()
+    for v in os.getenv("GITEA_COUNT_REPOS", "").split(",")
+    if v.strip()
+}
 
 
 @dataclass
@@ -40,6 +57,26 @@ def parse_date(value: str) -> date:
     return datetime.strptime(value, "%Y-%m-%d").date()
 
 
+def log_info(msg: str) -> None:
+    print(f"[info] {msg}")
+
+
+def log_warn(msg: str) -> None:
+    print(f"[warn] {msg}")
+
+
+def log_error(context: str, exc: Exception) -> None:
+    # Не логируем str(exc), чтобы не утекали токены/URL/чувствительные детали.
+    print(f"[error] {context}: {type(exc).__name__}")
+
+
+def is_user_allowed(chat_id: int, username: str | None) -> bool:
+    if not TELEGRAM_ALLOWED_CHAT_IDS and not TELEGRAM_ALLOWED_USERNAMES:
+        return True
+    username_norm = (username or "").lstrip("@").lower()
+    return chat_id in TELEGRAM_ALLOWED_CHAT_IDS or (username_norm and username_norm in TELEGRAM_ALLOWED_USERNAMES)
+
+
 class Storage:
     def __init__(self, db_path: str) -> None:
         self.db_path = db_path
@@ -256,6 +293,17 @@ class Storage:
             for r in rows
         ]
 
+    def get_known_repositories(self) -> list[str]:
+        with self._connect() as conn:
+            rows = conn.execute(
+                """
+                SELECT DISTINCT repo_full_name
+                FROM assignments
+                WHERE repo_full_name IS NOT NULL AND repo_full_name != ''
+                """
+            ).fetchall()
+        return [str(r["repo_full_name"]) for r in rows]
+
     def get_kv(self, key: str) -> str | None:
         with self._connect() as conn:
             row = conn.execute("SELECT value FROM kv WHERE key = ?", (key,)).fetchone()
@@ -313,7 +361,7 @@ class TelegramClient:
         except Exception as exc:
             # Не роняем обработку webhook, если Telegram API временно недоступен
             # или указан неверный токен/чат.
-            print(f"[telegram] Ошибка отправки сообщения: {exc}")
+            log_error("telegram send_message", exc)
 
 
 class GiteaClient:
@@ -350,6 +398,34 @@ class GiteaClient:
             if response.status_code >= 400:
                 return
 
+    async def get_pull_request(self, repo_full_name: str, pr_number: int) -> dict[str, Any] | None:
+        if not self.enabled:
+            return None
+        async with httpx.AsyncClient(timeout=20) as client:
+            response = await client.get(
+                f"{self.base_url}/api/v1/repos/{repo_full_name}/pulls/{pr_number}",
+                headers={"Authorization": f"token {self.token}"},
+            )
+            if response.status_code >= 400:
+                return None
+            return response.json()
+
+    async def list_open_pull_requests(self, repo_full_name: str, page: int = 1, limit: int = 50) -> list[dict[str, Any]]:
+        if not self.enabled:
+            return []
+        async with httpx.AsyncClient(timeout=20) as client:
+            response = await client.get(
+                f"{self.base_url}/api/v1/repos/{repo_full_name}/pulls",
+                headers={"Authorization": f"token {self.token}"},
+                params={"state": "open", "page": page, "limit": limit},
+            )
+            if response.status_code >= 400:
+                return []
+            data = response.json()
+            if not isinstance(data, list):
+                return []
+            return data
+
 
 storage = Storage(DB_PATH)
 telegram_client = TelegramClient(TELEGRAM_BOT_TOKEN)
@@ -359,34 +435,101 @@ reminder_task: asyncio.Task | None = None
 
 def verify_gitea_signature(raw_body: bytes, signature_header: str | None) -> bool:
     if not GITEA_WEBHOOK_SECRET:
-        return True
+        return False
     if not signature_header:
         return False
+    signature_value = signature_header.strip()
+    if signature_value.startswith("sha256="):
+        signature_value = signature_value.split("=", 1)[1]
     digest = hmac.new(
         GITEA_WEBHOOK_SECRET.encode("utf-8"),
         raw_body,
         hashlib.sha256,
     ).hexdigest()
-    return hmac.compare_digest(digest, signature_header)
+    return hmac.compare_digest(digest, signature_value)
 
 
 def choose_reviewer(candidates: list[RegisteredUser]) -> RegisteredUser:
-    counts: dict[str, int] = {}
-    for candidate in candidates:
-        counts[candidate.gitea_login] = storage.get_open_reviews_count(candidate.gitea_login)
+    counts = {
+        candidate.gitea_login: storage.get_open_reviews_count(candidate.gitea_login)
+        for candidate in candidates
+    }
     min_count = min(counts.values())
     shortlist = [c for c in candidates if counts[c.gitea_login] == min_count]
     return random.choice(shortlist)
 
 
-def get_pr_participant_chat_ids(
-    repo_full_name: str, pr_number: int, exclude_login: str | None = None
-) -> list[int]:
-    """Возвращает chat_id участников PR (автор и ревьюеры), зарегистрированных в боте. exclude_login не включается."""
-    rows = storage.get_assignments_for_pr(repo_full_name, pr_number)
-    if not rows:
-        return []
+async def choose_reviewer_with_remote_load(candidates: list[RegisteredUser], current_repo_full_name: str) -> RegisteredUser:
+    candidate_logins = {candidate.gitea_login for candidate in candidates}
+    repos_to_scan = set(storage.get_known_repositories()) | GITEA_COUNT_REPOS | {current_repo_full_name}
+    counts: dict[str, int] = defaultdict(int)
+
+    if not gitea_client.enabled:
+        return choose_reviewer(candidates)
+
+    for repo in repos_to_scan:
+        page = 1
+        while True:
+            pulls = await gitea_client.list_open_pull_requests(repo, page=page, limit=50)
+            if not pulls:
+                break
+            for pull in pulls:
+                reviewer_logins = extract_reviewer_logins_from_pull_request(pull)
+                for login in reviewer_logins:
+                    if login in candidate_logins:
+                        counts[login] += 1
+            if len(pulls) < 50:
+                break
+            page += 1
+
+    for candidate in candidates:
+        counts[candidate.gitea_login] += 0
+
+    min_count = min(counts.values())
+    shortlist = [c for c in candidates if counts[c.gitea_login] == min_count]
+    return random.choice(shortlist)
+
+
+def extract_logins_from_pull_request(pull_request: dict[str, Any] | None) -> set[str]:
     logins: set[str] = set()
+    if not pull_request:
+        return logins
+    author_login = ((pull_request.get("user") or {}).get("login") or "").strip()
+    if author_login:
+        logins.add(author_login)
+    for key in ("requested_reviewers", "reviewers"):
+        for reviewer in pull_request.get(key) or []:
+            login = ((reviewer or {}).get("login") or "").strip()
+            if login:
+                logins.add(login)
+    return logins
+
+
+def extract_reviewer_logins_from_pull_request(pull_request: dict[str, Any] | None) -> set[str]:
+    logins: set[str] = set()
+    if not pull_request:
+        return logins
+    for key in ("requested_reviewers", "reviewers"):
+        for reviewer in pull_request.get(key) or []:
+            login = ((reviewer or {}).get("login") or "").strip()
+            if login:
+                logins.add(login)
+    return logins
+
+
+async def get_pr_participant_chat_ids(
+    repo_full_name: str,
+    pr_number: int,
+    pull_request: dict[str, Any] | None = None,
+    exclude_login: str | None = None,
+) -> list[int]:
+    """Возвращает chat_id участников PR (автор и ревьюеры), зарегистрированных в боте."""
+    logins = extract_logins_from_pull_request(pull_request)
+    if len(logins) <= 1:
+        remote_pull = await gitea_client.get_pull_request(repo_full_name, pr_number)
+        logins.update(extract_logins_from_pull_request(remote_pull))
+
+    rows = storage.get_assignments_for_pr(repo_full_name, pr_number)
     for r in rows:
         logins.add(r["author_login"])
         logins.add(r["reviewer_login"])
@@ -397,19 +540,36 @@ def get_pr_participant_chat_ids(
         user = storage.get_user_by_gitea_login(login)
         if user:
             chat_ids.append(user.telegram_chat_id)
+        else:
+            log_warn(f"[notify] Пользователь '{login}' не зарегистрирован в боте, уведомление пропущено.")
     return chat_ids
 
 
+async def notify_user_by_login(login: str, text: str) -> None:
+    user = storage.get_user_by_gitea_login(login)
+    if not user:
+        log_warn(f"[notify] Пользователь '{login}' не зарегистрирован в боте, уведомление пропущено.")
+        return
+    await telegram_client.send_message(user.telegram_chat_id, text)
+
+
 async def handle_telegram_command(update: dict[str, Any]) -> None:
     message = update.get("message") or {}
     chat = message.get("chat") or {}
     user = message.get("from") or {}
     text = (message.get("text") or "").strip()
     chat_id = chat.get("id")
+    chat_type = chat.get("type")
     username = user.get("username")
 
     if not chat_id or not text.startswith("/"):
         return
+    if chat_type != "private":
+        log_warn(f"[telegram] Игнорирую команду из не-личного чата id={chat_id}, type={chat_type}")
+        return
+    if not is_user_allowed(int(chat_id), username):
+        log_warn(f"[telegram] Пользователь chat_id={chat_id}, username={username} не входит в allowlist.")
+        return
 
     parts = text.split()
     cmd = parts[0].lower()
@@ -436,6 +596,7 @@ async def handle_telegram_command(update: dict[str, Any]) -> None:
             return
         email = parts[1].lower()
         gitea_login = parts[2]
+        try:
             storage.upsert_user(
                 RegisteredUser(
                     telegram_chat_id=int(chat_id),
@@ -444,6 +605,16 @@ async def handle_telegram_command(update: dict[str, Any]) -> None:
                     gitea_login=gitea_login,
                 )
             )
+        except sqlite3.IntegrityError:
+            log_warn(
+                f"[telegram] Конфликт регистрации chat_id={chat_id}, "
+                f"email={email}, gitea_login={gitea_login}"
+            )
+            await telegram_client.send_message(
+                chat_id,
+                "Этот email или gitea_login уже привязан к другому Telegram-аккаунту.",
+            )
+            return
         await telegram_client.send_message(chat_id, "Пользователь зарегистрирован.")
         return
 
@@ -497,6 +668,7 @@ async def handle_telegram_command(update: dict[str, Any]) -> None:
     if cmd == "/myreviews":
         me = storage.get_user_by_chat_id(int(chat_id))
         if me is None:
+            log_warn(f"[telegram] /myreviews для незарегистрированного chat_id={chat_id}")
             await telegram_client.send_message(
                 chat_id,
                 "Учетная запись не найдена. Сначала выполните /register.",
@@ -516,8 +688,6 @@ async def handle_telegram_command(update: dict[str, Any]) -> None:
 async def handle_pr_opened(payload: dict[str, Any]) -> None:
     pull_request = payload.get("pull_request") or {}
     requested_reviewers = pull_request.get("requested_reviewers") or []
-    if requested_reviewers:
-        return
 
     repository = payload.get("repository") or {}
     repo_full_name = repository.get("full_name")
@@ -526,7 +696,26 @@ async def handle_pr_opened(payload: dict[str, Any]) -> None:
 
     if not repo_full_name or not pr_number or not author_login:
         return
+    requested_reviewer_logins = [
+        reviewer.get("login")
+        for reviewer in requested_reviewers
+        if (reviewer or {}).get("login")
+    ]
+    if requested_reviewer_logins:
+        for reviewer_login in requested_reviewer_logins:
+            storage.add_assignment(
+                repo_full_name=repo_full_name,
+                pr_number=int(pr_number),
+                author_login=author_login,
+                reviewer_login=reviewer_login,
+            )
+            await notify_user_by_login(
+                reviewer_login,
+                f"Назначено новое ревью: {repo_full_name}#{pr_number} (автор: {author_login})",
+            )
+        return
     if not gitea_client.enabled:
+        log_warn("[gitea] Клиент не настроен, автоназначение пропущено.")
         return
 
     candidates = storage.get_available_candidates(
@@ -535,9 +724,10 @@ async def handle_pr_opened(payload: dict[str, Any]) -> None:
         allow_self_assign=ALLOW_SELF_ASSIGN,
     )
     if not candidates:
+        log_warn(f"[assign] Нет доступных кандидатов для {repo_full_name}#{pr_number}.")
         return
 
-    reviewer = choose_reviewer(candidates)
+    reviewer = await choose_reviewer_with_remote_load(candidates, repo_full_name)
     await gitea_client.assign_reviewer(repo_full_name, int(pr_number), reviewer.gitea_login)
     await gitea_client.add_label(repo_full_name, int(pr_number), AUTO_ASSIGNED_LABEL)
     storage.add_assignment(
@@ -561,7 +751,11 @@ async def handle_pr_closed(payload: dict[str, Any]) -> None:
     action = payload.get("action")
     if not repo_full_name or not pr_number:
         return
-    participant_chat_ids = get_pr_participant_chat_ids(repo_full_name, int(pr_number))
+    participant_chat_ids = await get_pr_participant_chat_ids(
+        repo_full_name,
+        int(pr_number),
+        pull_request=pull_request,
+    )
     storage.close_assignments_for_pr(repo_full_name, int(pr_number))
     status_label = "смержен" if action == "merged" else "закрыт"
     title = (pull_request.get("title") or "").strip() or f"#{pr_number}"
@@ -578,14 +772,15 @@ async def handle_pr_synchronize(payload: dict[str, Any]) -> None:
     pr_number = pull_request.get("number")
     if not repo_full_name or not pr_number:
         return
-    rows = storage.get_assignments_for_pr(repo_full_name, int(pr_number))
     author_login = (pull_request.get("user") or {}).get("login") or ""
     msg = f"В PR {repo_full_name}#{pr_number} добавлены новые коммиты (автор: {author_login})."
-    for r in rows:
-        reviewer_login = r["reviewer_login"]
-        user = storage.get_user_by_gitea_login(reviewer_login)
-        if user:
-            await telegram_client.send_message(user.telegram_chat_id, msg)
+    for chat_id in await get_pr_participant_chat_ids(
+        repo_full_name,
+        int(pr_number),
+        pull_request=pull_request,
+        exclude_login=author_login,
+    ):
+        await telegram_client.send_message(chat_id, msg)
 
 
 async def handle_issue_comment(payload: dict[str, Any]) -> None:
@@ -602,7 +797,11 @@ async def handle_issue_comment(payload: dict[str, Any]) -> None:
     if not repo_full_name or not pr_number:
         return
     msg = f"Новый комментарий в PR {repo_full_name}#{pr_number}: {body}"
-    for chat_id in get_pr_participant_chat_ids(repo_full_name, int(pr_number), exclude_login=commenter_login):
+    for chat_id in await get_pr_participant_chat_ids(
+        repo_full_name,
+        int(pr_number),
+        exclude_login=commenter_login,
+    ):
         await telegram_client.send_message(chat_id, msg)
 
 
@@ -624,7 +823,12 @@ async def handle_pr_reviewed(payload: dict[str, Any]) -> None:
     else:
         status_text = state or "обновлён"
     msg = f"PR {repo_full_name}#{pr_number}: ревью {status_text} ({reviewer_login})."
-    for chat_id in get_pr_participant_chat_ids(repo_full_name, int(pr_number), exclude_login=reviewer_login):
+    for chat_id in await get_pr_participant_chat_ids(
+        repo_full_name,
+        int(pr_number),
+        pull_request=pull_request,
+        exclude_login=reviewer_login,
+    ):
         await telegram_client.send_message(chat_id, msg)
 
 
@@ -682,15 +886,25 @@ async def health() -> dict[str, str]:
 
 
 @app.post("/telegram/webhook")
-async def telegram_webhook(update: dict[str, Any]) -> dict[str, str]:
+async def telegram_webhook(
+    update: dict[str, Any],
+    x_telegram_bot_api_secret_token: str | None = Header(default=None),
+) -> dict[str, str]:
     if not TELEGRAM_BOT_TOKEN:
         raise HTTPException(status_code=500, detail="Не задан TELEGRAM_BOT_TOKEN")
+    if not TELEGRAM_WEBHOOK_SECRET:
+        raise HTTPException(status_code=500, detail="Не задан TELEGRAM_WEBHOOK_SECRET")
+    if not x_telegram_bot_api_secret_token or not hmac.compare_digest(
+        TELEGRAM_WEBHOOK_SECRET,
+        x_telegram_bot_api_secret_token,
+    ):
+        raise HTTPException(status_code=401, detail="Некорректный telegram webhook secret")
     try:
         await handle_telegram_command(update)
     except Exception as exc:
         # Telegram повторяет доставку при 5xx, поэтому здесь всегда отвечаем 200
         # и логируем причину для отладки.
-        print(f"[telegram] Ошибка обработки апдейта: {exc}")
+        log_error("telegram webhook update handling", exc)
     return {"ok": "true"}
 
 
@@ -700,11 +914,18 @@ async def gitea_webhook(
     x_gitea_event: str | None = Header(default=None),
     x_gitea_signature: str | None = Header(default=None),
 ) -> dict[str, str]:
+    if not GITEA_WEBHOOK_SECRET:
+        raise HTTPException(status_code=500, detail="Не задан GITEA_WEBHOOK_SECRET")
     body = await request.body()
     if not verify_gitea_signature(body, x_gitea_signature):
         raise HTTPException(status_code=401, detail="Некорректная подпись")
+    try:
         payload = json.loads(body.decode("utf-8"))
+    except json.JSONDecodeError as exc:
+        log_error("gitea webhook invalid json", exc)
+        raise HTTPException(status_code=400, detail="Некорректный JSON") from exc
 
+    try:
         if x_gitea_event == "issue_comment":
             await handle_issue_comment(payload)
             return {"ok": "true"}
@@ -724,6 +945,10 @@ async def gitea_webhook(
             elif action == "reviewed" and payload.get("review"):
                 await handle_pr_reviewed(payload)
             return {"ok": "true"}
+    except Exception as exc:
+        # Не падаем на обработчиках событий и логируем безопасно.
+        log_error("gitea webhook event handling", exc)
+        return {"ok": "true"}
 
     return {"ok": "ignored"}
 

requirements.txt

@@ -1,3 +1,4 @@
 fastapi
 uvicorn
 httpx
+dotenv