ligbox-ops-platform/scripts/sync-tasks-to-forgejo-issues.py

#!/usr/bin/env python3
"""Sincroniza tasks.md (Spec Kit) → issues Forgejo.

Uso:
  python3 scripts/sync-tasks-to-forgejo-issues.py [--dry-run] [--open-only]

Requer: FORGEJO_URL, FORGEJO_USER, FORGEJO_TOKEN (ou defaults CT130).
Idempotente: reutiliza issues pelo título canónico [spec-NNN] Txxx.
"""
from __future__ import annotations

import argparse
import json
import os
import re
import sys
import time
import urllib.error
import urllib.request
from dataclasses import dataclass
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[1]
SPECS_DIR = REPO_ROOT / "specs"
TASK_LINE = re.compile(r"^-\s+\[([ xX])\]\s+(.*)$")
SPEC_DIR = re.compile(r"^(\d{3})-")

FORGEJO_URL = os.environ.get("FORGEJO_URL", "http://10.10.10.130:3000").rstrip("/")
FORGEJO_USER = os.environ.get("FORGEJO_USER", "roger")
FORGEJO_TOKEN = os.environ.get("FORGEJO_TOKEN", "805353")
OWNER = os.environ.get("FORGEJO_OWNER", "ligbox")
REPO = os.environ.get("FORGEJO_REPO", "ligbox-ops-platform")


@dataclass
class TaskItem:
    spec_id: str
    spec_slug: str
    spec_title: str
    task_id: str
    text: str
    done: bool
    tasks_path: str

    @property
    def issue_title(self) -> str:
        tid = self.task_id or self.text[:60]
        return f"[spec-{self.spec_id}] {tid} — {self.text[:120]}"

    @property
    def issue_body(self) -> str:
        status = "done" if self.done else "open"
        return (
            f"## Spec {self.spec_id} — {self.spec_title}\n\n"
            f"| Campo | Valor |\n|-------|-------|\n"
            f"| **Task** | `{self.task_id or '—'}` |\n"
            f"| **Estado tasks.md** | `{status}` |\n"
            f"| **Ficheiro** | `{self.tasks_path}` |\n"
            f"| **Portal** | https://spec.ligbox.com.br/specs/{self.spec_slug}/ |\n\n"
            f"### Descrição\n\n{self.text}\n\n"
            f"---\n_Sincronizado automaticamente por `scripts/sync-tasks-to-forgejo-issues.py`_"
        )


def api(method: str, path: str, data: dict | None = None) -> dict | list:
    url = f"{FORGEJO_URL}/api/v1{path}"
    body = json.dumps(data).encode() if data is not None else None
    req = urllib.request.Request(url, data=body, method=method)
    req.add_header("Content-Type", "application/json")
    token = f"{FORGEJO_USER}:{FORGEJO_TOKEN}"
    import base64

    req.add_header("Authorization", "Basic " + base64.b64encode(token.encode()).decode())
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            raw = resp.read().decode()
            return json.loads(raw) if raw else {}
    except urllib.error.HTTPError as e:
        err = e.read().decode()
        raise RuntimeError(f"{method} {path} → HTTP {e.code}: {err[:500]}") from e


def spec_meta(spec_dir: Path) -> tuple[str, str]:
    m = SPEC_DIR.match(spec_dir.name)
    spec_id = m.group(1) if m else spec_dir.name[:3]
    title = spec_dir.name
    spec_md = spec_dir / "spec.md"
    if spec_md.is_file():
        for line in spec_md.read_text(encoding="utf-8", errors="replace").splitlines()[:8]:
            if line.startswith("# "):
                title = line.lstrip("# ").strip()
                break
    return spec_id, title


def parse_tasks(spec_dir: Path) -> list[TaskItem]:
    tasks_file = spec_dir / "tasks.md"
    if not tasks_file.is_file():
        return []
    spec_id, spec_title = spec_meta(spec_dir)
    rel = str(tasks_file.relative_to(REPO_ROOT))
    items: list[TaskItem] = []
    for line in tasks_file.read_text(encoding="utf-8", errors="replace").splitlines():
        m = TASK_LINE.match(line.strip())
        if not m:
            continue
        done = m.group(1).lower() == "x"
        raw = m.group(2).strip()
        tid_m = re.match(r"^(T\d+)\b", raw)
        task_id = tid_m.group(1) if tid_m else ""
        items.append(
            TaskItem(
                spec_id=spec_id,
                spec_slug=spec_dir.name,
                spec_title=spec_title,
                task_id=task_id,
                text=raw,
                done=done,
                tasks_path=rel,
            )
        )
    return items


def load_all_tasks(open_only: bool) -> list[TaskItem]:
    all_items: list[TaskItem] = []
    for spec_dir in sorted(SPECS_DIR.iterdir()):
        if not spec_dir.is_dir():
            continue
        for item in parse_tasks(spec_dir):
            if open_only and item.done:
                continue
            all_items.append(item)
    return all_items


def fetch_existing_titles() -> dict[str, int]:
    titles: dict[str, int] = {}
    page = 1
    while True:
        batch = api("GET", f"/repos/{OWNER}/{REPO}/issues?state=all&limit=50&page={page}")
        if not batch:
            break
        for issue in batch:
            if issue.get("pull_request"):
                continue
            titles[issue["title"]] = issue["number"]
        if len(batch) < 50:
            break
        page += 1
    return titles


def ensure_label(name: str, cache: dict[str, int]) -> int:
    if name in cache:
        return cache[name]
    labels = api("GET", f"/repos/{OWNER}/{REPO}/labels?limit=100")
    for lb in labels:
        cache[lb["name"]] = lb["id"]
        if lb["name"] == name:
            return lb["id"]
    created = api("POST", f"/repos/{OWNER}/{REPO}/labels", {"name": name, "color": "0052cc"})
    cache[name] = created["id"]
    return created["id"]


def apply_labels(issue_num: int, label_names: list[str], cache: dict[str, int], dry_run: bool) -> None:
    if dry_run:
        return
    ids = [ensure_label(n, cache) for n in label_names]
    api("POST", f"/repos/{OWNER}/{REPO}/issues/{issue_num}/labels", {"labels": ids})


def sync_issue(task: TaskItem, existing: dict[str, int], label_cache: dict[str, int], dry_run: bool) -> str:
    title = task.issue_title
    labels = [f"spec-{task.spec_id}", "done" if task.done else "open"]
    state = "closed" if task.done else "open"

    if title in existing:
        num = existing[title]
        if dry_run:
            return f"skip #{num} {title[:70]}"
        api(
            "PATCH",
            f"/repos/{OWNER}/{REPO}/issues/{num}",
            {"state": state, "body": task.issue_body},
        )
        apply_labels(num, labels, label_cache, dry_run)
        return f"update #{num}"

    if dry_run:
        return f"create {title[:70]}"
    issue = api(
        "POST",
        f"/repos/{OWNER}/{REPO}/issues",
        {"title": title, "body": task.issue_body},
    )
    num = issue["number"]
    apply_labels(num, labels, label_cache, dry_run)
    if task.done:
        api("PATCH", f"/repos/{OWNER}/{REPO}/issues/{num}", {"state": "closed"})
    existing[title] = num
    return f"create #{num}"


def update_repo_metadata(dry_run: bool) -> None:
    desc = (
        "Monorepo Ligbox Ops — Desk VM122, Wizard VM112, Finance VM123, Spec Kit. "
        "Fonte de verdade: specs/ + issues sincronizadas de tasks.md. "
        "Portal: https://spec.ligbox.com.br"
    )
    if dry_run:
        print("would update repo description")
        return
    api("PATCH", f"/repos/{OWNER}/{REPO}", {"description": desc, "has_issues": True})


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--open-only", action="store_true", help="Só tarefas [ ] abertas")
    parser.add_argument("--sleep", type=float, default=0.15, help="Pausa entre requests")
    args = parser.parse_args()

    tasks = load_all_tasks(args.open_only)
    print(f"==> Tasks encontradas: {len(tasks)} (open_only={args.open_only})")
    if not tasks:
        print("Nada a sincronizar.")
        return 0

    update_repo_metadata(args.dry_run)
    existing = {} if args.dry_run else fetch_existing_titles()
    print(f"==> Issues existentes: {len(existing)}")

    created = updated = 0
    label_cache: dict[str, int] = {}
    for i, task in enumerate(tasks, 1):
        result = sync_issue(task, existing, label_cache, args.dry_run)
        if result.startswith("create"):
            created += 1
        elif result.startswith("update"):
            updated += 1
        if i % 50 == 0:
            print(f"  ... {i}/{len(tasks)}")
        if not args.dry_run:
            time.sleep(args.sleep)

    print(f"==> Concluído: create={created} update={updated} total={len(tasks)}")
    if not args.dry_run:
        open_n = api("GET", f"/repos/{OWNER}/{REPO}")
        print(
            f"==> Repo: https://git.spec.ligbox.com.br/{OWNER}/{REPO}/issues "
            f"(open={open_n.get('open_issues_count')})"
        )
    return 0


if __name__ == "__main__":
    sys.exit(main())