467 lines
17 KiB
Python
467 lines
17 KiB
Python
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import get_settings
|
|
from app.db.models import ProjectBinding, TaigaProject, WorkItem
|
|
from app.integrations.gitea import GiteaClient
|
|
from app.integrations.taiga import TaigaClient
|
|
from app.projects.commit_parser import parse_commit_message
|
|
from app.projects.structuring import (
|
|
format_gitea_body,
|
|
format_story_description,
|
|
slugify_branch,
|
|
structure_work_item,
|
|
)
|
|
|
|
|
|
class ProjectService:
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
self.settings = get_settings()
|
|
|
|
def sync_taiga_projects(self) -> list[dict[str, Any]]:
|
|
if not self.settings.taiga_configured:
|
|
raise ValueError("Taiga не настроена: задайте TAIGA_USERNAME и TAIGA_PASSWORD")
|
|
|
|
client = TaigaClient()
|
|
remote = client.list_projects()
|
|
now = datetime.now(timezone.utc)
|
|
|
|
for item in remote:
|
|
slug = item.get("slug") or ""
|
|
if not slug:
|
|
continue
|
|
existing = self.db.scalar(
|
|
select(TaigaProject).where(TaigaProject.slug == slug)
|
|
)
|
|
if existing:
|
|
existing.name = item.get("name", slug)
|
|
existing.taiga_id = item["id"]
|
|
existing.synced_at = now
|
|
else:
|
|
self.db.add(
|
|
TaigaProject(
|
|
taiga_id=item["id"],
|
|
name=item.get("name", slug),
|
|
slug=slug,
|
|
synced_at=now,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
return self.list_projects()
|
|
|
|
def list_projects(self) -> list[dict[str, Any]]:
|
|
stmt = (
|
|
select(TaigaProject, ProjectBinding)
|
|
.outerjoin(ProjectBinding, ProjectBinding.taiga_slug == TaigaProject.slug)
|
|
.order_by(TaigaProject.name)
|
|
)
|
|
rows = self.db.execute(stmt).all()
|
|
result = []
|
|
for taiga_proj, binding in rows:
|
|
result.append(
|
|
{
|
|
"taiga_id": taiga_proj.taiga_id,
|
|
"name": taiga_proj.name,
|
|
"slug": taiga_proj.slug,
|
|
"gitea_owner": binding.gitea_owner if binding else "",
|
|
"gitea_repo": binding.gitea_repo if binding else "",
|
|
"default_branch": binding.default_branch if binding else "main",
|
|
"gitea_configured": bool(binding and binding.gitea_owner and binding.gitea_repo),
|
|
}
|
|
)
|
|
return result
|
|
|
|
def bind_gitea(
|
|
self,
|
|
taiga_slug: str,
|
|
gitea_owner: str,
|
|
gitea_repo: str,
|
|
default_branch: str = "main",
|
|
) -> dict[str, Any]:
|
|
if not self.db.scalar(select(TaigaProject).where(TaigaProject.slug == taiga_slug)):
|
|
raise ValueError(f"Проект Taiga '{taiga_slug}' не найден. Сначала sync-taiga.")
|
|
|
|
binding = self.db.scalar(
|
|
select(ProjectBinding).where(ProjectBinding.taiga_slug == taiga_slug)
|
|
)
|
|
if binding:
|
|
binding.gitea_owner = gitea_owner
|
|
binding.gitea_repo = gitea_repo
|
|
binding.default_branch = default_branch
|
|
else:
|
|
binding = ProjectBinding(
|
|
taiga_slug=taiga_slug,
|
|
gitea_owner=gitea_owner,
|
|
gitea_repo=gitea_repo,
|
|
default_branch=default_branch,
|
|
)
|
|
self.db.add(binding)
|
|
self.db.commit()
|
|
|
|
for proj in self.list_projects():
|
|
if proj["slug"] == taiga_slug:
|
|
return proj
|
|
raise ValueError("Binding failed")
|
|
|
|
def _resolve_project(self, slug: str | None) -> tuple[TaigaProject, ProjectBinding | None]:
|
|
projects = self.db.scalars(select(TaigaProject).order_by(TaigaProject.name)).all()
|
|
if not projects:
|
|
raise ValueError("Нет проектов Taiga. Вызовите sync_taiga_projects.")
|
|
|
|
taiga_proj: TaigaProject | None = None
|
|
if slug:
|
|
taiga_proj = self.db.scalar(
|
|
select(TaigaProject).where(TaigaProject.slug == slug)
|
|
)
|
|
if not taiga_proj:
|
|
raise ValueError(f"Проект '{slug}' не найден")
|
|
else:
|
|
taiga_proj = projects[0]
|
|
|
|
binding = self.db.scalar(
|
|
select(ProjectBinding).where(ProjectBinding.taiga_slug == taiga_proj.slug)
|
|
)
|
|
return taiga_proj, binding
|
|
|
|
async def create_work_item(
|
|
self, raw_text: str, project_slug: str | None = None
|
|
) -> dict[str, Any]:
|
|
if not self.settings.taiga_configured:
|
|
raise ValueError("Taiga не настроена")
|
|
|
|
project_list = self.list_projects()
|
|
if not project_list:
|
|
self.sync_taiga_projects()
|
|
project_list = self.list_projects()
|
|
|
|
structured = await structure_work_item(raw_text, project_list)
|
|
slug = project_slug or structured.get("project_slug")
|
|
taiga_proj, binding = self._resolve_project(slug)
|
|
|
|
if binding and not (binding.gitea_owner and binding.gitea_repo):
|
|
binding = None
|
|
|
|
taiga = TaigaClient()
|
|
title = (structured.get("title") or raw_text).strip()[:500]
|
|
description = format_story_description(structured, raw_text)
|
|
tags = structured.get("tags") or []
|
|
issue_type = structured.get("issue_type", "feature")
|
|
if issue_type == "bug" and "bug" not in [t.lower() for t in tags]:
|
|
tags.append("bug")
|
|
|
|
story = taiga.create_userstory(
|
|
taiga_proj.taiga_id,
|
|
title,
|
|
description,
|
|
tags=tags,
|
|
)
|
|
|
|
subtasks = []
|
|
for child in structured.get("children") or []:
|
|
if isinstance(child, dict):
|
|
subtasks.append(
|
|
taiga.create_task(
|
|
taiga_proj.taiga_id,
|
|
story["id"],
|
|
child.get("title", "Подзадача"),
|
|
child.get("description", ""),
|
|
)
|
|
)
|
|
|
|
branch = f"feature/{story['ref']}-{slugify_branch(title)}"
|
|
gitea_issue_number = None
|
|
gitea_url = ""
|
|
|
|
if binding and self.settings.gitea_configured:
|
|
gitea = GiteaClient()
|
|
gitea_body = format_gitea_body(
|
|
structured,
|
|
raw_text,
|
|
story["ref"],
|
|
taiga.story_url(taiga_proj.taiga_id, story["ref"]),
|
|
branch,
|
|
)
|
|
if issue_type:
|
|
gitea_body = f"**Тип:** {issue_type}\n\n{gitea_body}"
|
|
issue = gitea.create_issue(
|
|
binding.gitea_owner,
|
|
binding.gitea_repo,
|
|
title,
|
|
gitea_body,
|
|
)
|
|
gitea_issue_number = issue["number"]
|
|
gitea_url = gitea.issue_url(
|
|
binding.gitea_owner, binding.gitea_repo, gitea_issue_number
|
|
)
|
|
|
|
work_item = WorkItem(
|
|
taiga_slug=taiga_proj.slug,
|
|
taiga_project_id=taiga_proj.taiga_id,
|
|
taiga_story_id=story["id"],
|
|
taiga_story_ref=story["ref"],
|
|
gitea_owner=binding.gitea_owner if binding else "",
|
|
gitea_repo=binding.gitea_repo if binding else "",
|
|
gitea_issue_number=gitea_issue_number,
|
|
suggested_branch=branch,
|
|
raw_text=raw_text,
|
|
title=title,
|
|
status="open",
|
|
)
|
|
self.db.add(work_item)
|
|
self.db.commit()
|
|
self.db.refresh(work_item)
|
|
|
|
return {
|
|
"ok": True,
|
|
"work_item_id": work_item.id,
|
|
"taiga": {
|
|
"ref": story["ref"],
|
|
"id": story["id"],
|
|
"subject": story["subject"],
|
|
"url": taiga.story_url(taiga_proj.taiga_id, story["ref"]),
|
|
},
|
|
"gitea": {
|
|
"number": gitea_issue_number,
|
|
"url": gitea_url,
|
|
},
|
|
"branch": branch,
|
|
"issue_type": issue_type,
|
|
"subtasks": [{"ref": t.get("ref"), "subject": t.get("subject")} for t in subtasks],
|
|
"questions": structured.get("questions") or [],
|
|
"project_slug": taiga_proj.slug,
|
|
}
|
|
|
|
def process_push(
|
|
self, owner: str, repo: str, commits: list[dict[str, Any]]
|
|
) -> list[dict[str, Any]]:
|
|
if not self.settings.taiga_configured:
|
|
return []
|
|
|
|
taiga = TaigaClient()
|
|
gitea = GiteaClient() if self.settings.gitea_configured else None
|
|
results: list[dict[str, Any]] = []
|
|
|
|
for commit in commits:
|
|
message = commit.get("message", "")
|
|
parsed = parse_commit_message(message)
|
|
sha = commit.get("id", "")[:8]
|
|
|
|
gitea_refs = set(parsed["gitea"])
|
|
taiga_story_refs = set(parsed["taiga_story"])
|
|
taiga_task_refs = set(parsed["taiga_task"])
|
|
|
|
linked_items = self.db.scalars(
|
|
select(WorkItem).where(
|
|
WorkItem.gitea_owner == owner,
|
|
WorkItem.gitea_repo == repo,
|
|
WorkItem.status == "open",
|
|
)
|
|
).all()
|
|
|
|
for item in linked_items:
|
|
if item.gitea_issue_number and item.gitea_issue_number in gitea_refs:
|
|
taiga_story_refs.add(item.taiga_story_ref)
|
|
if item.taiga_story_ref in taiga_story_refs and item.gitea_issue_number:
|
|
gitea_refs.add(item.gitea_issue_number)
|
|
|
|
for gitea_num in gitea_refs:
|
|
if gitea:
|
|
try:
|
|
gitea.close_issue(owner, repo, gitea_num)
|
|
except Exception as exc:
|
|
results.append({"error": f"gitea #{gitea_num}: {exc}"})
|
|
continue
|
|
|
|
for item in linked_items:
|
|
if item.gitea_issue_number == gitea_num:
|
|
try:
|
|
self._close_work_item(item, taiga)
|
|
results.append(
|
|
{
|
|
"commit": sha,
|
|
"closed": f"gitea #{gitea_num}, taiga #{item.taiga_story_ref}",
|
|
}
|
|
)
|
|
except Exception as exc:
|
|
results.append(
|
|
{"error": f"work item {item.id} (gitea #{gitea_num}): {exc}"}
|
|
)
|
|
|
|
for ref in taiga_story_refs:
|
|
project_id = self._project_id_for_ref(owner, repo, ref, linked_items)
|
|
if not project_id:
|
|
continue
|
|
story = taiga.get_by_ref(project_id, ref, kind="userstory")
|
|
if story and not story.get("is_closed"):
|
|
try:
|
|
taiga.close_userstory(story["id"], project_id)
|
|
results.append({"commit": sha, "closed": f"taiga #{ref}"})
|
|
except Exception as exc:
|
|
results.append({"error": f"taiga #{ref}: {exc}"})
|
|
for item in linked_items:
|
|
if item.taiga_story_ref == ref and item.status != "closed":
|
|
try:
|
|
self._close_work_item(item, taiga, close_gitea=bool(gitea))
|
|
except Exception as exc:
|
|
results.append(
|
|
{"error": f"work item {item.id} (taiga #{ref}): {exc}"}
|
|
)
|
|
|
|
for ref in taiga_task_refs:
|
|
binding = self.db.scalar(
|
|
select(ProjectBinding).where(
|
|
ProjectBinding.gitea_owner == owner,
|
|
ProjectBinding.gitea_repo == repo,
|
|
)
|
|
)
|
|
if not binding:
|
|
continue
|
|
taiga_proj = self.db.scalar(
|
|
select(TaigaProject).where(TaigaProject.slug == binding.taiga_slug)
|
|
)
|
|
if not taiga_proj:
|
|
continue
|
|
task = taiga.get_by_ref(taiga_proj.taiga_id, ref, kind="task")
|
|
if task and not task.get("is_closed"):
|
|
try:
|
|
taiga.close_task(task["id"], taiga_proj.taiga_id)
|
|
results.append({"commit": sha, "closed": f"taiga task #{ref}"})
|
|
except Exception as exc:
|
|
results.append({"error": f"taiga task #{ref}: {exc}"})
|
|
|
|
self.db.commit()
|
|
return results
|
|
|
|
def _project_id_for_ref(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
ref: int,
|
|
items: list[WorkItem],
|
|
) -> int | None:
|
|
for item in items:
|
|
if item.taiga_story_ref == ref:
|
|
return item.taiga_project_id
|
|
binding = self.db.scalar(
|
|
select(ProjectBinding).where(
|
|
ProjectBinding.gitea_owner == owner,
|
|
ProjectBinding.gitea_repo == repo,
|
|
)
|
|
)
|
|
if binding:
|
|
taiga_proj = self.db.scalar(
|
|
select(TaigaProject).where(TaigaProject.slug == binding.taiga_slug)
|
|
)
|
|
return taiga_proj.taiga_id if taiga_proj else None
|
|
return None
|
|
|
|
def _close_work_item(
|
|
self,
|
|
item: WorkItem,
|
|
taiga: TaigaClient,
|
|
*,
|
|
close_gitea: bool = True,
|
|
) -> None:
|
|
if item.status == "closed":
|
|
return
|
|
story = taiga.get_by_ref(item.taiga_project_id, item.taiga_story_ref, kind="userstory")
|
|
if story:
|
|
taiga.close_userstory(story["id"], item.taiga_project_id)
|
|
if (
|
|
close_gitea
|
|
and item.gitea_issue_number
|
|
and self.settings.gitea_configured
|
|
):
|
|
GiteaClient().close_issue(
|
|
item.gitea_owner, item.gitea_repo, item.gitea_issue_number
|
|
)
|
|
item.status = "closed"
|
|
item.closed_at = datetime.now(timezone.utc)
|
|
|
|
def list_taiga_open_tasks(
|
|
self,
|
|
project_slug: str | None = None,
|
|
limit: int = 20,
|
|
) -> dict[str, Any]:
|
|
if not self.settings.taiga_configured:
|
|
raise ValueError("Taiga не настроена")
|
|
|
|
projects = self.list_projects()
|
|
if not projects:
|
|
projects = self.sync_taiga_projects()
|
|
|
|
if project_slug:
|
|
projects = [p for p in projects if p["slug"] == project_slug]
|
|
if not projects:
|
|
raise ValueError(
|
|
f"Проект '{project_slug}' не найден. Вызови sync_taiga_projects."
|
|
)
|
|
|
|
client = TaigaClient()
|
|
blocks: list[dict[str, Any]] = []
|
|
|
|
for proj in projects:
|
|
stories = client.list_open_userstories(proj["taiga_id"], limit=limit)
|
|
tasks = client.list_open_tasks(proj["taiga_id"], limit=limit)
|
|
blocks.append(
|
|
{
|
|
"slug": proj["slug"],
|
|
"name": proj["name"],
|
|
"taiga_id": proj["taiga_id"],
|
|
"stories": [
|
|
{
|
|
"ref": s.get("ref"),
|
|
"subject": s.get("subject", ""),
|
|
"url": client.story_url(proj["taiga_id"], s.get("ref", 0)),
|
|
}
|
|
for s in stories
|
|
],
|
|
"tasks": [
|
|
{
|
|
"ref": t.get("ref"),
|
|
"subject": t.get("subject", ""),
|
|
"user_story": t.get("user_story"),
|
|
}
|
|
for t in tasks
|
|
],
|
|
}
|
|
)
|
|
|
|
total_stories = sum(len(b["stories"]) for b in blocks)
|
|
total_tasks = sum(len(b["tasks"]) for b in blocks)
|
|
return {
|
|
"projects": blocks,
|
|
"total_stories": total_stories,
|
|
"total_tasks": total_tasks,
|
|
}
|
|
|
|
def list_work_items(self, limit: int = 30, status: str | None = None) -> list[dict[str, Any]]:
|
|
stmt = select(WorkItem).order_by(WorkItem.created_at.desc()).limit(limit)
|
|
if status:
|
|
stmt = stmt.where(WorkItem.status == status)
|
|
items = self.db.scalars(stmt).all()
|
|
settings = get_settings()
|
|
return [
|
|
{
|
|
"id": i.id,
|
|
"title": i.title,
|
|
"status": i.status,
|
|
"taiga_slug": i.taiga_slug,
|
|
"taiga_ref": i.taiga_story_ref,
|
|
"gitea_issue": i.gitea_issue_number,
|
|
"branch": i.suggested_branch,
|
|
"taiga_url": f"{settings.taiga_public_url}/project/0/{i.taiga_project_id}/us/{i.taiga_story_ref}",
|
|
"gitea_url": (
|
|
f"{settings.gitea_public_url}/{i.gitea_owner}/{i.gitea_repo}/issues/{i.gitea_issue_number}"
|
|
if i.gitea_issue_number
|
|
else ""
|
|
),
|
|
"created_at": i.created_at.isoformat() if i.created_at else None,
|
|
}
|
|
for i in items
|
|
]
|