252 lines
9.2 KiB
Python
252 lines
9.2 KiB
Python
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass
|
||
from typing import Any, Optional
|
||
from urllib.parse import quote
|
||
|
||
import httpx
|
||
|
||
from config import settings
|
||
|
||
|
||
@dataclass
|
||
class GitLabTask:
|
||
id: int
|
||
iid: int
|
||
title: str
|
||
reference: str
|
||
web_url: str
|
||
project_path: str
|
||
source: str = "unknown"
|
||
labels: list[str] | None = None
|
||
|
||
def to_description(self) -> str:
|
||
from services.description_formatter import format_professional_description
|
||
|
||
return format_professional_description(self)
|
||
|
||
|
||
class GitLabClient:
|
||
def __init__(self) -> None:
|
||
self.base_url = settings.gitlab_url.rstrip("/")
|
||
self.token = (settings.gitlab_token or "").strip().strip("\"'")
|
||
self._username: Optional[str] = None
|
||
|
||
def _headers(self) -> dict[str, str]:
|
||
return {"PRIVATE-TOKEN": self.token}
|
||
|
||
def _auth_error_message(self, response: httpx.Response) -> str:
|
||
detail = ""
|
||
try:
|
||
payload = response.json()
|
||
detail = (
|
||
payload.get("error_description")
|
||
or payload.get("message")
|
||
or payload.get("error")
|
||
or ""
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
detail = (response.text or "")[:200]
|
||
detail = str(detail).strip()
|
||
lower = detail.lower()
|
||
if response.status_code == 401:
|
||
if "revok" in lower:
|
||
return (
|
||
"GitLab token 已撤銷或失效。請到 GitLab → Preferences → Access Tokens "
|
||
"重新建立 Personal Access Token（勾選 api），再到設定頁貼上新 token。"
|
||
)
|
||
return (
|
||
"GitLab token 無效（401）。請確認已貼上新的 Personal Access Token，"
|
||
f"且 URL 為 {self.base_url}。{(' 詳情: ' + detail) if detail else ''}"
|
||
)
|
||
if response.status_code == 403:
|
||
return f"GitLab token 權限不足（403）。請勾選 api scope。{(' 詳情: ' + detail) if detail else ''}"
|
||
return f"GitLab API error {response.status_code}: {detail or response.reason_phrase}"
|
||
|
||
async def _get_current_username(self) -> str:
|
||
if self._username:
|
||
return self._username
|
||
async with httpx.AsyncClient(timeout=60.0, headers=self._headers()) as client:
|
||
response = await client.get(f"{self.base_url}/api/v4/user")
|
||
if response.status_code in {401, 403}:
|
||
raise ValueError(self._auth_error_message(response))
|
||
response.raise_for_status()
|
||
self._username = response.json()["username"]
|
||
return self._username
|
||
|
||
def _issue_to_task(self, issue: dict[str, Any], source: str) -> GitLabTask:
|
||
refs = issue.get("references") or {}
|
||
return GitLabTask(
|
||
id=issue["id"],
|
||
iid=issue["iid"],
|
||
title=issue["title"],
|
||
reference=refs.get("full") or f"#{issue['iid']}",
|
||
web_url=issue.get("web_url", ""),
|
||
project_path=refs.get("relative", ""),
|
||
source=source,
|
||
labels=issue.get("labels") or [],
|
||
)
|
||
|
||
async def _fetch_project_issues(
|
||
self,
|
||
project_path: str,
|
||
*,
|
||
source: str,
|
||
extra_params: Optional[dict[str, Any]] = None,
|
||
) -> list[GitLabTask]:
|
||
encoded = quote(project_path, safe="")
|
||
params: dict[str, Any] = {
|
||
"state": "opened",
|
||
"per_page": 100,
|
||
"order_by": "updated_at",
|
||
"sort": "desc",
|
||
}
|
||
if extra_params:
|
||
params.update(extra_params)
|
||
|
||
async with httpx.AsyncClient(timeout=60.0, headers=self._headers()) as client:
|
||
response = await client.get(
|
||
f"{self.base_url}/api/v4/projects/{encoded}/issues",
|
||
params=params,
|
||
)
|
||
response.raise_for_status()
|
||
issues = response.json()
|
||
|
||
return [self._issue_to_task(issue, source) for issue in issues]
|
||
|
||
async def get_board_bug_tasks(self) -> list[GitLabTask]:
|
||
username = await self._get_current_username()
|
||
labels = ",".join(settings.gitlab_board_label_list)
|
||
return await self._fetch_project_issues(
|
||
settings.gitlab_project_path,
|
||
source="board_bug",
|
||
extra_params={
|
||
"assignee_username": username,
|
||
"labels": labels,
|
||
},
|
||
)
|
||
|
||
async def _pick_current_milestone(
|
||
self,
|
||
active_milestones: list[dict[str, Any]],
|
||
closed_milestones: list[dict[str, Any]],
|
||
) -> Optional[dict[str, Any]]:
|
||
if settings.gitlab_milestone_title:
|
||
for milestone in active_milestones:
|
||
if milestone.get("title") == settings.gitlab_milestone_title:
|
||
return milestone
|
||
return {"title": settings.gitlab_milestone_title, "state": "configured"}
|
||
|
||
version_re = re.compile(r"^SCC_v", re.IGNORECASE)
|
||
active_versions = [
|
||
milestone
|
||
for milestone in active_milestones
|
||
if milestone.get("state") == "active"
|
||
and "backlog" not in milestone.get("title", "").lower()
|
||
and version_re.match(milestone.get("title", ""))
|
||
]
|
||
if not active_versions:
|
||
return None
|
||
|
||
closed_versions = [
|
||
milestone
|
||
for milestone in closed_milestones
|
||
if version_re.match(milestone.get("title", ""))
|
||
]
|
||
latest_closed_version = 0
|
||
if closed_versions:
|
||
latest_closed_version = max(
|
||
self._milestone_version_key(milestone.get("title", ""))
|
||
for milestone in closed_versions
|
||
)
|
||
|
||
# 目前衝刺 = 最新已關閉版本之後、版本號最小的 active milestone
|
||
# 例：a10 已關閉 → 選 a11（不是規劃中的 a12）
|
||
successors = [
|
||
milestone
|
||
for milestone in active_versions
|
||
if self._milestone_version_key(milestone.get("title", "")) > latest_closed_version
|
||
]
|
||
if successors:
|
||
successors.sort(key=lambda milestone: self._milestone_version_key(milestone.get("title", "")))
|
||
return successors[0]
|
||
|
||
active_versions.sort(key=lambda milestone: self._milestone_version_key(milestone.get("title", "")))
|
||
return active_versions[0]
|
||
|
||
def _milestone_version_key(self, title: str) -> int:
|
||
match = re.search(r"a(\d+)$", title, re.IGNORECASE)
|
||
return int(match.group(1)) if match else 0
|
||
|
||
async def get_milestone_tasks(self) -> tuple[Optional[str], list[GitLabTask]]:
|
||
encoded = quote(settings.gitlab_project_path, safe="")
|
||
username = await self._get_current_username()
|
||
|
||
async with httpx.AsyncClient(timeout=60.0, headers=self._headers()) as client:
|
||
active_resp = await client.get(
|
||
f"{self.base_url}/api/v4/projects/{encoded}/milestones",
|
||
params={"state": "active", "per_page": 50},
|
||
)
|
||
active_resp.raise_for_status()
|
||
active_milestones = active_resp.json()
|
||
|
||
closed_resp = await client.get(
|
||
f"{self.base_url}/api/v4/projects/{encoded}/milestones",
|
||
params={"state": "closed", "per_page": 20, "order_by": "updated_at", "sort": "desc"},
|
||
)
|
||
closed_resp.raise_for_status()
|
||
closed_milestones = closed_resp.json()
|
||
|
||
milestone = await self._pick_current_milestone(active_milestones, closed_milestones)
|
||
if not milestone:
|
||
return None, []
|
||
|
||
title = milestone["title"]
|
||
tasks = await self._fetch_project_issues(
|
||
settings.gitlab_project_path,
|
||
source="milestone",
|
||
extra_params={
|
||
"assignee_username": username,
|
||
"milestone": title,
|
||
},
|
||
)
|
||
return title, tasks
|
||
|
||
async def get_assigned_tasks(self) -> list[GitLabTask]:
|
||
if not self.token:
|
||
raise ValueError("GITLAB_TOKEN is not configured")
|
||
|
||
board_tasks = await self.get_board_bug_tasks()
|
||
milestone_title, milestone_tasks = await self.get_milestone_tasks()
|
||
|
||
merged: dict[int, GitLabTask] = {}
|
||
for task in board_tasks:
|
||
merged[task.id] = task
|
||
for task in milestone_tasks:
|
||
merged.setdefault(task.id, task)
|
||
|
||
tasks = list(merged.values())
|
||
tasks.sort(key=lambda t: (0 if t.source == "board_bug" else 1, t.reference))
|
||
return tasks
|
||
|
||
async def get_task_sources(self) -> dict[str, Any]:
|
||
board_tasks = await self.get_board_bug_tasks()
|
||
milestone_title, milestone_tasks = await self.get_milestone_tasks()
|
||
merged = await self.get_assigned_tasks()
|
||
return {
|
||
"board": {
|
||
"project": settings.gitlab_project_path,
|
||
"board_id": settings.gitlab_board_id,
|
||
"labels": settings.gitlab_board_label_list,
|
||
"count": len(board_tasks),
|
||
"tasks": [t.__dict__ for t in board_tasks],
|
||
},
|
||
"milestone": {
|
||
"title": milestone_title,
|
||
"count": len(milestone_tasks),
|
||
"tasks": [t.__dict__ for t in milestone_tasks],
|
||
},
|
||
"merged_count": len(merged),
|
||
"tasks": [t.__dict__ for t in merged],
|
||
} |