app

  1"""
  2Webhook-сервер LLM code review бота.
  3
  4Слушает события pull_request из Gitea (opened/synchronize/reopened),
  5запускает ревью в фоне и постит результат через Gitea API.
  6"""
  7
  8from __future__ import annotations
  9
 10import hashlib
 11import hmac
 12import json
 13import logging
 14import os
 15from contextlib import asynccontextmanager
 16import sys
 17
 18from apscheduler.schedulers.asyncio import AsyncIOScheduler
 19from apscheduler.triggers.cron import CronTrigger
 20from e_lib import Config as EConfig
 21from e_lib.assistants import LangfuseCleanerAssistant
 22from fastapi import BackgroundTasks, FastAPI, Request
 23from fastapi.responses import JSONResponse
 24from loguru import logger
 25
 26from config import Config
 27from gitea_client import GiteaClient
 28from langfuse_client import get_langfuse
 29from models import AppConfig
 30from review import run_review
 31from reviewer import cache_prompt
 32
 33APP_VERSION = os.getenv("PRODUCTS_VERSION", "dev")
 34
 35config = Config.from_env()
 36client = GiteaClient(config)
 37
 38# Gitea 1.19 использует "synchronized" (в отличие от GitHub "synchronize").
 39PR_ACTIONS = ("opened", "synchronized", "synchronize", "reopened", "ready_for_review")
 40
 41CONFIG_PATH = os.getenv(
 42    "CONFIG_PATH",
 43    os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.yaml"),
 44)
 45
 46_scheduler: AsyncIOScheduler | None = None
 47
 48
 49class InterceptHandler(logging.Handler):
 50    """
 51    Перехватчик логов из стандартного logging в loguru.
 52
 53    Перенаправляет все сообщения из логгеров uvicorn и fastapi
 54    в форматированный вывод loguru.
 55    """
 56
 57    def emit(self, record):
 58        """
 59        Обрабатывает и перенаправляет запись лога в loguru.
 60
 61        Args:
 62            record: Запись лога из стандартного модуля logging
 63        """
 64        try:
 65            level = logger.level(record.levelname).name
 66        except ValueError:
 67            level = record.levelno
 68
 69        frame, depth = logging.currentframe(), 2
 70        while frame.f_code.co_filename == logging.__file__:
 71            frame = frame.f_back
 72            depth += 1
 73
 74        logger.opt(depth=depth, exception=record.exc_info).log(
 75            level, record.getMessage()
 76        )
 77
 78
 79def setup_logging():
 80    """
 81    Настраивает логирование приложения.
 82
 83    Перенаправляет логи uvicorn и fastapi в loguru с цветным
 84    форматированным выводом в stdout. Уровень DEBUG для всех
 85    сообщений приложения.
 86    """
 87    logger.remove()
 88
 89    logger.add(
 90        sys.stdout,
 91        format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> | <level>{message}</level>",
 92        level="DEBUG",
 93        colorize=True,
 94    )
 95
 96    intercept_handler = InterceptHandler()
 97
 98    root_logger = logging.getLogger()
 99    root_logger.handlers = [intercept_handler]
100    root_logger.setLevel(logging.INFO)
101
102    for logger_name in ["uvicorn", "uvicorn.access", "uvicorn.error", "fastapi"]:
103        logging_logger = logging.getLogger(logger_name)
104        logging_logger.handlers = [intercept_handler]
105        logging_logger.setLevel(logging.INFO)
106        logging_logger.propagate = False
107
108
109def _to_cron(time_str: str) -> str:
110    """Преобразует время 'ЧЧ:ММ' в cron-выражение."""
111    h, m = time_str.split(":")
112    return f"{m} {h} * * *"
113
114
115@asynccontextmanager
116async def lifespan(_: FastAPI):
117    """
118    Инициализирует Langfuse и планирует фонового ассистента очистки трейсов
119    (LangfuseCleanerAssistant из e_lib) по конфигу e_lib.Config (config.yaml).
120    """
121    global _scheduler
122    setup_logging()
123    products_log = None
124    try:
125        from e_lib.product_log import ProductsLog
126
127        products_log = ProductsLog("llm-code-review", APP_VERSION, logger=logger)
128        products_log.post_started()
129    except Exception as e:
130        logger.warning(f"ProductsLog: {e}")
131    try:
132        cache_prompt()
133        get_langfuse()
134        econfig = EConfig(CONFIG_PATH, model=AppConfig)
135        langfuse_cleaner_cfg = econfig.get("Assistants", {}).get("LangfuseCleaner", {})
136        if langfuse_cleaner_cfg.get("ENABLED"):
137            lf_cleaner = LangfuseCleanerAssistant(
138                project_slug=langfuse_cleaner_cfg.get("PROJECT", ""),
139                period=langfuse_cleaner_cfg.get("RETENTION_DAYS", 7),
140                logger=logger,
141            )
142            _scheduler = AsyncIOScheduler()
143            _scheduler.add_job(
144                lf_cleaner.run,
145                CronTrigger.from_crontab(_to_cron(langfuse_cleaner_cfg["TIME"])),
146                id="langfuse_cleaner",
147                replace_existing=True,
148            )
149            _scheduler.start()
150            logger.info(f"Ассистент '{langfuse_cleaner_cfg.get('NAME')}' включён")
151    except Exception as e:
152        logger.warning(f"Langfuse: ошибка инициализации: {e}")
153    yield
154    if _scheduler:
155        _scheduler.shutdown(wait=False)
156        logger.info("Приложение остановлено")
157    try:
158        if products_log:
159            products_log.post_shutdown()
160    except Exception as e:
161        logger.warning(f"ProductsLog: {e}")
162
163
164app = FastAPI(title="LLM Code Review", lifespan=lifespan)
165
166
167def _verify_signature(payload: bytes, signature: str | None) -> bool:
168    """Проверяет X-Gitea-Signature (HMAC-SHA256 от тела запроса)."""
169    if not config.webhook_secret:
170        return True
171    if not signature:
172        return False
173    expected = hmac.new(
174        config.webhook_secret.encode(), payload, hashlib.sha256
175    ).hexdigest()
176    return hmac.compare_digest(expected, signature)
177
178
179def _handle_pull_request(data: dict) -> None:
180    """Основная работа: достаёт данные PR и запускает ревью."""
181    try:
182        repo_full = data.get("repository", {}).get("full_name", "")
183        if "/" not in repo_full:
184            logger.warning(f"Webhook: нет full_name репозитория: {repo_full!r}")
185            return
186        owner, repo = repo_full.split("/", 1)
187        index = data.get("number") or data.get("pull_request", {}).get("number")
188        if not index:
189            logger.warning("Webhook: нет номера PR")
190            return
191        head_sha = data.get("pull_request", {}).get("head", {}).get("sha")
192        outcome = run_review(client, config, owner, repo, int(index), head_sha)
193        if outcome.skipped:
194            logger.info(f"Ревью {owner}/{repo}#{index} пропущено: {outcome.reason}")
195    except Exception:
196        logger.exception("Ошибка при обработке webhook pull_request")
197
198
199@app.get("/health")
200def health() -> dict:
201    """Health-check для Docker healthcheck."""
202    return {"status": "ok"}
203
204
205@app.post("/webhook")
206async def webhook(request: Request, background_tasks: BackgroundTasks):
207    """Принимает webhook от Gitea и ставит обработку в фон."""
208    payload = await request.body()
209
210    event = request.headers.get("x-gitea-event", "")
211    signature = request.headers.get("x-gitea-signature")
212
213    if not _verify_signature(payload, signature):
214        logger.warning("Webhook: неверная подпись")
215        return JSONResponse(status_code=403, content={"status": "forbidden"})
216
217    if event != "pull_request":
218        return {"status": "ignored", "event": event}
219
220    try:
221        data = json.loads(payload)
222    except json.JSONDecodeError as e:
223        logger.warning(f"Webhook: битый JSON: {e}")
224        return JSONResponse(status_code=400, content={"status": "bad json"})
225
226    action = data.get("action", "")
227    if action not in PR_ACTIONS:
228        return {"status": "ignored", "action": action}
229
230    if data.get("pull_request", {}).get("draft"):
231        return {"status": "ignored", "reason": "draft"}
232
233    background_tasks.add_task(_handle_pull_request, data)
234    return {"status": "accepted", "action": action}