from typing import Any from sqlalchemy.orm import Session from app.config import get_settings from app.integrations.taiga import TaigaClient from app.projects.service import ProjectService MAX_PROJECTS_IN_CONTEXT = 8 MAX_OPEN_PER_PROJECT = 5 def get_projects_snapshot(db: Session) -> dict[str, Any]: settings = get_settings() service = ProjectService(db) if not settings.taiga_configured: return {"configured": False, "projects": [], "open_items": [], "taiga_open": []} projects = service.list_projects() if not projects: try: projects = service.sync_taiga_projects() except Exception: projects = [] open_items = service.list_work_items(limit=15, status="open") taiga_open: list[dict[str, Any]] = [] try: client = TaigaClient() for proj in projects[:MAX_PROJECTS_IN_CONTEXT]: stories = client.list_open_userstories( proj["taiga_id"], limit=MAX_OPEN_PER_PROJECT ) if not stories: continue taiga_open.append( { "slug": proj["slug"], "name": proj["name"], "stories": [ { "ref": s.get("ref"), "subject": s.get("subject", "")[:120], } for s in stories ], } ) except Exception: pass return { "configured": True, "projects": projects, "open_items": open_items, "taiga_open": taiga_open, } def format_projects_context(snapshot: dict[str, Any]) -> str: if not snapshot.get("configured"): return "[Taiga/Gitea]\nНе настроено (нет TAIGA_USERNAME/PASSWORD в .env)." lines = ["[Проекты и задачи — актуальный снимок для контекста]"] projects = snapshot.get("projects") or [] if not projects: lines.append("Проекты Taiga: кэш пуст. Попроси sync_taiga_projects или проверь подключение.") else: lines.append("Проекты Taiga:") for p in projects[:MAX_PROJECTS_IN_CONTEXT]: gitea = ( f"{p.get('gitea_owner')}/{p.get('gitea_repo')}" if p.get("gitea_configured") else "Gitea не привязан" ) lines.append(f"- `{p.get('slug')}`: {p.get('name')} · {gitea}") open_items = snapshot.get("open_items") or [] if open_items: lines.append("") lines.append("Локальные work items (созданные ассистентом, открытые):") for item in open_items[:10]: gitea_part = f", gitea #{item.get('gitea_issue')}" if item.get("gitea_issue") else "" lines.append( f"- taiga #{item.get('taiga_ref')} {item.get('title')} " f"({item.get('taiga_slug')}{gitea_part})" ) taiga_open = snapshot.get("taiga_open") or [] if taiga_open: lines.append("") lines.append("Открытые user stories в Taiga:") for block in taiga_open: lines.append(f" [{block.get('slug')}]") for story in block.get("stories", []): lines.append(f" - #{story.get('ref')} {story.get('subject')}") elif projects: lines.append("") lines.append("Открытые user stories в Taiga: нет или не удалось загрузить.") lines.append("") lines.append( "Для создания фичи/бага из вольного текста используй create_work_item. " "Не выдумывай номера задач — опирайся на список выше." ) return "\n".join(lines)