feat: AI-аудит кода через Gitea Actions + DeepSeek V4.1 Flash
This commit is contained in:
@@ -0,0 +1,259 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
AI Code Review for Filwords PL Backend
|
||||
|
||||
Читает Python-файлы проекта, отправляет в DeepSeek V4.1 Flash (Together AI),
|
||||
получает структурированное ТЗ и создаёт Issue в Gitea.
|
||||
|
||||
Запускается через Gitea Actions (workflow_dispatch — вручную из UI).
|
||||
|
||||
(20.09.2026) Первая версия — конвейер без tool-calling, предсказуемо по токенам.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
|
||||
|
||||
# ============ Настройки ============
|
||||
|
||||
TOGETHER_API_KEY = os.environ.get("TOGETHER_API_KEY")
|
||||
GITEA_BOT_TOKEN = os.environ.get("BOT_TOKEN")
|
||||
GITEA_INSTANCE = os.environ.get("GITEA_INSTANCE", "https://git.filwordspl.com")
|
||||
GITEA_REPO = os.environ.get("GITEA_REPO", "Drobysh/filwordspl-backend")
|
||||
|
||||
MODEL = "deepseek-ai/DeepSeek-V4.1-Flash"
|
||||
MAX_OUTPUT_TOKENS = 4000
|
||||
|
||||
# Что игнорировать при сборе кода
|
||||
IGNORE_DIRS = {
|
||||
"venv", ".venv", "__pycache__", ".git", ".gitea",
|
||||
"tests", "archive", "uploads", "tts_cache",
|
||||
"node_modules", "dist",
|
||||
}
|
||||
IGNORE_FILES = {"ai_review.py", "conftest.py"}
|
||||
|
||||
# Расширения, которые читаем
|
||||
INCLUDE_EXTENSIONS = {".py"}
|
||||
|
||||
|
||||
# ============ Проверка окружения ============
|
||||
|
||||
def check_env() -> None:
|
||||
missing = []
|
||||
if not TOGETHER_API_KEY:
|
||||
missing.append("TOGETHER_API_KEY")
|
||||
if not GITEA_BOT_TOKEN:
|
||||
missing.append("BOT_TOKEN")
|
||||
if missing:
|
||||
print(f"❌ Отсутствуют переменные окружения: {', '.join(missing)}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
# ============ Сбор кода ============
|
||||
|
||||
def collect_code(repo_root: Path) -> str:
|
||||
"""Собирает все .py файлы в один текст с заголовками."""
|
||||
chunks: list[str] = []
|
||||
total_files = 0
|
||||
total_size = 0
|
||||
|
||||
for path in sorted(repo_root.rglob("*")):
|
||||
if not path.is_file():
|
||||
continue
|
||||
if path.suffix not in INCLUDE_EXTENSIONS:
|
||||
continue
|
||||
|
||||
# Пропускаем игнорируемые папки
|
||||
rel = path.relative_to(repo_root)
|
||||
if any(part in IGNORE_DIRS for part in rel.parts):
|
||||
continue
|
||||
if path.name in IGNORE_FILES:
|
||||
continue
|
||||
|
||||
try:
|
||||
content = path.read_text(encoding="utf-8")
|
||||
except Exception as e:
|
||||
print(f"⚠️ Не прочитал {rel}: {e}", file=sys.stderr)
|
||||
continue
|
||||
|
||||
# Пропускаем слишком большие файлы (>100 KB)
|
||||
if len(content) > 100_000:
|
||||
print(f"⚠️ Пропущен большой файл: {rel} ({len(content)} символов)", file=sys.stderr)
|
||||
continue
|
||||
|
||||
chunks.append(f"### FILE: {rel}\n```python\n{content}\n```\n")
|
||||
total_files += 1
|
||||
total_size += len(content)
|
||||
|
||||
print(f"📂 Собрано файлов: {total_files}, символов: {total_size:,}")
|
||||
return "\n".join(chunks)
|
||||
|
||||
|
||||
# ============ Промпт для модели ============
|
||||
|
||||
SYSTEM_PROMPT = """Ты — опытный техлид и ревьюер Python-кода (FastAPI, asyncpg, PostgreSQL).
|
||||
|
||||
Твоя задача — проанализировать код и выдать СТРУКТУРИРОВАННОЕ ТЗ для разработчика.
|
||||
|
||||
ПРАВИЛА:
|
||||
1. НЕ переписывай код. Только описывай, что не так и что сделать.
|
||||
2. Каждый пункт — КОНКРЕТНЫЙ: файл → строка/функция → проблема → что сделать.
|
||||
3. Без воды. Без общих слов. Без объяснений «что такое XSS» — только факт и фикс.
|
||||
4. Приоритеты: 🔴 критично / 🟡 важно / 🟢 желательно.
|
||||
5. Если проблем мало — лучше меньше пунктов, но точных.
|
||||
6. Максимум 20 пунктов — только самое важное.
|
||||
|
||||
ФОРМАТ ОТВЕТА (markdown):
|
||||
|
||||
## 🔴 Критично
|
||||
|
||||
### 1. [короткое название]
|
||||
- **Файл:** `path/to/file.py:123`
|
||||
- **Проблема:** что не так
|
||||
- **Что сделать:** конкретное действие
|
||||
- **Критерий:** как проверить, что исправлено
|
||||
|
||||
## 🟡 Важно
|
||||
(то же самое)
|
||||
|
||||
## 🟢 Желательно
|
||||
(то же самое)
|
||||
|
||||
Если проблем нет — напиши: "✅ Критичных проблем не найдено".
|
||||
"""
|
||||
|
||||
|
||||
USER_PROMPT_TEMPLATE = """Проанализируй код проекта Filwords PL (backend).
|
||||
|
||||
Это FastAPI + PostgreSQL + asyncpg. Есть:
|
||||
- routers/ — эндпоинты
|
||||
- services/ — бизнес-логика
|
||||
- database.py — схема БД
|
||||
- auth.py — HMAC-SHA256 initData
|
||||
|
||||
Ищи:
|
||||
1. Баги и логические ошибки
|
||||
2. Проблемы безопасности (утечки, инъекции, неправильная валидация)
|
||||
3. N+1 запросы к БД
|
||||
4. Отсутствие обработки ошибок
|
||||
5. Дублирование кода
|
||||
6. Мёртвый код
|
||||
7. Слабые места архитектуры
|
||||
|
||||
Вот код:
|
||||
|
||||
{code}
|
||||
"""
|
||||
|
||||
|
||||
# ============ Запрос к Together AI ============
|
||||
|
||||
def call_together(code: str) -> str:
|
||||
"""Отправляет код в DeepSeek V4.1 Flash, возвращает текст ответа."""
|
||||
print("🤖 Отправляю код в DeepSeek V4.1 Flash...")
|
||||
|
||||
user_prompt = USER_PROMPT_TEMPLATE.format(code=code)
|
||||
|
||||
response = httpx.post(
|
||||
"https://api.together.xyz/v1/chat/completions",
|
||||
headers={
|
||||
"Authorization": f"Bearer {TOGETHER_API_KEY}",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
json={
|
||||
"model": MODEL,
|
||||
"messages": [
|
||||
{"role": "system", "content": SYSTEM_PROMPT},
|
||||
{"role": "user", "content": user_prompt},
|
||||
],
|
||||
"max_tokens": MAX_OUTPUT_TOKENS,
|
||||
"temperature": 0.3,
|
||||
},
|
||||
timeout=300.0,
|
||||
)
|
||||
|
||||
if response.status_code != 200:
|
||||
print(f"❌ Together AI вернул {response.status_code}: {response.text[:500]}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
data = response.json()
|
||||
result = data["choices"][0]["message"]["content"]
|
||||
|
||||
usage = data.get("usage", {})
|
||||
print(f"✅ Ответ получен. Токены: вход={usage.get('prompt_tokens', 0)}, выход={usage.get('completion_tokens', 0)}")
|
||||
|
||||
return result
|
||||
|
||||
|
||||
# ============ Создание Issue в Gitea ============
|
||||
|
||||
def create_issue(title: str, body: str) -> str:
|
||||
"""Создаёт Issue в Gitea, возвращает URL."""
|
||||
print(f"📝 Создаю Issue в Gitea...")
|
||||
|
||||
response = httpx.post(
|
||||
f"{GITEA_INSTANCE}/api/v1/repos/{GITEA_REPO}/issues",
|
||||
headers={
|
||||
"Authorization": f"token {GITEA_BOT_TOKEN}",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
json={
|
||||
"title": title,
|
||||
"body": body,
|
||||
},
|
||||
timeout=30.0,
|
||||
)
|
||||
|
||||
if response.status_code not in (200, 201):
|
||||
print(f"❌ Gitea вернул {response.status_code}: {response.text[:500]}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
data = response.json()
|
||||
issue_url = data.get("html_url", "")
|
||||
print(f"✅ Issue создан: {issue_url}")
|
||||
return issue_url
|
||||
|
||||
|
||||
# ============ Main ============
|
||||
|
||||
def main() -> None:
|
||||
check_env()
|
||||
|
||||
repo_root = Path(".").resolve()
|
||||
print(f"📁 Репозиторий: {repo_root}")
|
||||
|
||||
code = collect_code(repo_root)
|
||||
if not code:
|
||||
print("❌ Не найдено файлов для анализа", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
analysis = call_together(code)
|
||||
|
||||
from datetime import datetime
|
||||
date_str = datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")
|
||||
title = f"🤖 AI Review: список задач ({date_str})"
|
||||
|
||||
body = f"""# 🤖 AI Code Review
|
||||
|
||||
**Дата:** {date_str}
|
||||
**Модель:** `{MODEL}`
|
||||
**Репозиторий:** `{GITEA_REPO}`
|
||||
|
||||
---
|
||||
|
||||
{analysis}
|
||||
|
||||
---
|
||||
|
||||
_Этот Issue создан автоматически. Исправления делаются вручную._
|
||||
"""
|
||||
|
||||
create_issue(title, body)
|
||||
print("🎉 Готово!")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user