"""FlowDeck — forge (Gitea/GitHub) issues importer (v5.6.0, Phase 4). Pulls a repository's issues, labels and milestones through a forge adapter and normalizes them into FlowDeck collections. """ from __future__ import annotations from typing import Any from app.services.importers.base import ImportResult, make_collection def _label_names(issue: dict) -> list[str]: labels = issue.get("labels") or [] names: list[str] = [] for label in labels: if isinstance(label, dict): name = label.get("name") or label.get("title") else: name = str(label) if name and name not in names: names.append(name) return names def _milestone_name(issue: dict) -> str: milestone = issue.get("milestone") if isinstance(milestone, dict): return str(milestone.get("title") or milestone.get("name") or "") return str(milestone or "") def _assignees(issue: dict) -> list[str]: out: list[str] = [] for key in ("assignees", "assignee"): value = issue.get(key) if isinstance(value, list): for a in value: login = a.get("login") if isinstance(a, dict) else str(a) if login and login not in out: out.append(login) elif isinstance(value, dict): login = value.get("login") if login and login not in out: out.append(login) elif isinstance(value, str) and value and value not in out: out.append(value) return out _ISSUE_SCHEMA = [ {"name": "Title", "type": "title"}, {"name": "Number", "type": "number"}, {"name": "State", "type": "select", "options": [ {"name": "open", "color": "green"}, {"name": "closed", "color": "red"}, ]}, {"name": "Labels", "type": "multi_select"}, {"name": "Milestone", "type": "text"}, {"name": "Assignee", "type": "text"}, {"name": "Created", "type": "date"}, {"name": "Updated", "type": "date"}, {"name": "URL", "type": "url"}, {"name": "Body", "type": "text"}, ] def build_issues_result( issues: list[dict], *, labels: list[dict] | None = None, milestones: list[dict] | None = None, owner: str = "", repo: str = "", provider: str = "", ) -> ImportResult: """Normalize forge issues/labels/milestones into an ImportResult.""" result = ImportResult(source=f"forge:{provider}" if provider else "forge") name = f"{owner}/{repo} issues".strip("/ ") or "Issues" rows: list[dict] = [] for issue in issues: if issue.get("pull_request"): continue props: dict[str, Any] = {} if issue.get("number") is not None: props["Number"] = issue["number"] if issue.get("state"): props["State"] = issue["state"] label_names = _label_names(issue) if label_names: props["Labels"] = label_names milestone = _milestone_name(issue) if milestone: props["Milestone"] = milestone assignees = _assignees(issue) if assignees: props["Assignee"] = ", ".join(assignees) if issue.get("created_at"): props["Created"] = issue["created_at"] if issue.get("updated_at"): props["Updated"] = issue["updated_at"] if issue.get("html_url"): props["URL"] = issue["html_url"] if issue.get("body"): props["Body"] = issue["body"] title = issue.get("title") or f"#{issue.get('number', '')}".strip() rows.append({"title": title[:200], "properties": props}) result.pages.append(make_collection( name, _ISSUE_SCHEMA, rows, source_path=f"{provider}:{owner}/{repo}:issues", external_id=f"{provider}:{owner}/{repo}:issues", )) result.stats["rows"] = len(rows) if labels: label_schema = [ {"name": "Name", "type": "title"}, {"name": "Color", "type": "text"}, {"name": "Description", "type": "text"}, ] label_rows = [{ "title": (lbl.get("name") or lbl.get("title") or "Label")[:200], "properties": { k: v for k, v in ( ("Color", lbl.get("color")), ("Description", lbl.get("description")), ) if v }, } for lbl in labels] result.pages.append(make_collection( f"{owner}/{repo} labels".strip("/ "), label_schema, label_rows, source_path=f"{provider}:{owner}/{repo}:labels", external_id=f"{provider}:{owner}/{repo}:labels", )) if milestones: ms_schema = [ {"name": "Title", "type": "title"}, {"name": "State", "type": "select", "options": [ {"name": "open", "color": "green"}, {"name": "closed", "color": "red"}, ]}, {"name": "Due date", "type": "date"}, {"name": "Description", "type": "text"}, ] ms_rows = [] for ms in milestones: props: dict[str, Any] = {} if ms.get("state"): props["State"] = ms["state"] if ms.get("due_on"): props["Due date"] = ms["due_on"] if ms.get("description"): props["Description"] = ms["description"] ms_rows.append({"title": (ms.get("title") or "Milestone")[:200], "properties": props}) result.pages.append(make_collection( f"{owner}/{repo} milestones".strip("/ "), ms_schema, ms_rows, source_path=f"{provider}:{owner}/{repo}:milestones", external_id=f"{provider}:{owner}/{repo}:milestones", )) return result.finalize() class GiteaForgeAdapter: """Adapts a :class:`GiteaClient` to the ``list_*`` interface used here.""" def __init__(self, client) -> None: self._client = client async def list_issues(self, owner: str, repo: str, state: str = "all") -> list[dict]: issues: list[dict] = [] for page in range(1, 6): batch = await self._client.get_issues(owner, repo, state=state, page=page, limit=50) if not batch: break issues.extend(batch) if len(batch) < 50: break return issues async def list_labels(self, owner: str, repo: str) -> list[dict]: return await self._client.get_labels(owner, repo) async def list_milestones(self, owner: str, repo: str, state: str = "all") -> list[dict]: return await self._client.get_milestones(owner, repo, state=state) async def list_repo_files(self, owner: str, repo: str, path: str = "") -> list[dict]: """Recursively flatten Gitea repo contents into file entries.""" files: list[dict] = [] pending = [path.strip("/")] while pending and len(files) < 5000: current = pending.pop() items = await self._client.get_repo_contents(owner, repo, current) for item in items: if item.get("type") == "dir": pending.append(item.get("path") or item.get("name")) elif item.get("type") == "file": files.append({"path": item.get("path") or item.get("name"), "size": item.get("size", 0)}) return files async def get_file_content(self, owner: str, repo: str, path: str) -> str: return await self._client.get_file_content(owner, repo, path) async def fetch_forge_issues( adapter, owner: str, repo: str, *, provider: str = "", state: str = "all", include_labels: bool = True, include_milestones: bool = True, ) -> ImportResult: """Fetch issues (and optionally labels/milestones) then normalize them.""" issues = await adapter.list_issues(owner, repo, state) labels = None milestones = None if include_labels: try: labels = await adapter.list_labels(owner, repo) except Exception: # noqa: BLE001 - labels are optional labels = None if include_milestones: try: milestones = await adapter.list_milestones(owner, repo, state) except Exception: # noqa: BLE001 milestones = None return build_issues_result( issues, labels=labels, milestones=milestones, owner=owner, repo=repo, provider=provider, )