From cb4bb4a39904f156ef670aef40d971194262df08 Mon Sep 17 00:00:00 2001 From: Raykov-MS Date: Sat, 28 Feb 2026 20:07:50 +0300 Subject: [PATCH] E2E --- README.md | 10 +- main.py | 327 +++++++++++++++++++++++++++++++++++++++-------- requirements.txt | 1 + 3 files changed, 284 insertions(+), 54 deletions(-) diff --git a/README.md b/README.md index 869c16e..d1785d2 100644 --- a/README.md +++ b/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/setWebhook?url=https://example.ru.tuna.am/telegram/webhook +https://api.telegram.org/bot/setWebhook?url=https://example.ru.tuna.am/telegram/webhook&secret_token= ``` 6. Проверьте, что webhook установлен: diff --git a/main.py b/main.py index d6c60f1..0d88d1b 100644 --- a/main.py +++ b/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,14 +596,25 @@ async def handle_telegram_command(update: dict[str, Any]) -> None: return email = parts[1].lower() gitea_login = parts[2] - storage.upsert_user( - RegisteredUser( - telegram_chat_id=int(chat_id), - telegram_username=username, - email=email, - gitea_login=gitea_login, + try: + storage.upsert_user( + RegisteredUser( + telegram_chat_id=int(chat_id), + telegram_username=username, + email=email, + 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,29 +914,40 @@ 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="Некорректная подпись") - payload = json.loads(body.decode("utf-8")) + 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 - if x_gitea_event == "issue_comment": - await handle_issue_comment(payload) - return {"ok": "true"} + try: + if x_gitea_event == "issue_comment": + await handle_issue_comment(payload) + return {"ok": "true"} - if x_gitea_event == "pull_request_review" and payload.get("review"): - await handle_pr_reviewed(payload) - return {"ok": "true"} - - if x_gitea_event == "pull_request": - action = payload.get("action") - if action == "opened": - await handle_pr_opened(payload) - elif action in ("closed", "merged"): - await handle_pr_closed(payload) - elif action == "synchronize": - await handle_pr_synchronize(payload) - elif action == "reviewed" and payload.get("review"): + if x_gitea_event == "pull_request_review" and payload.get("review"): await handle_pr_reviewed(payload) + return {"ok": "true"} + + if x_gitea_event == "pull_request": + action = payload.get("action") + if action == "opened": + await handle_pr_opened(payload) + elif action in ("closed", "merged"): + await handle_pr_closed(payload) + elif action == "synchronize": + await handle_pr_synchronize(payload) + 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"} diff --git a/requirements.txt b/requirements.txt index d23d558..0224a44 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,4 @@ fastapi uvicorn httpx +dotenv