ligbox-ops-platform/scripts/sync-forgejo-milestones.py

#!/usr/bin/env python3
"""Cria milestones Forgejo por spec e associa issues (label spec-NNN).

Uso:
  python3 scripts/sync-forgejo-milestones.py [--dry-run]

Requer credenciais Forgejo (roger:805353 default CT130).
"""
from __future__ import annotations

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

REPO_ROOT = Path(__file__).resolve().parents[1]
SPECS_DIR = REPO_ROOT / "specs"
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")


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")
    import base64

    req.add_header("Authorization", "Basic " + base64.b64encode(f"{FORGEJO_USER}:{FORGEJO_TOKEN}".encode()).decode())
    try:
        with urllib.request.urlopen(req, timeout=120) 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[:400]}") from e


def spec_title(spec_dir: Path) -> tuple[str, str]:
    spec_id = SPEC_DIR.match(spec_dir.name).group(1)
    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()[:6]:
            if line.startswith("# "):
                title = line.lstrip("# ").strip()
                break
    return spec_id, title


def load_specs() -> list[tuple[str, str, str]]:
    items: list[tuple[str, str, str]] = []
    for spec_dir in sorted(SPECS_DIR.iterdir()):
        if not spec_dir.is_dir():
            continue
        m = SPEC_DIR.match(spec_dir.name)
        if not m:
            continue
        spec_id, title = spec_title(spec_dir)
        ms_title = f"Spec {spec_id} — {title[:80]}"
        desc = (
            f"**Spec:** `{spec_dir.name}`\n\n"
            f"Portal: https://spec.ligbox.com.br/specs/{spec_dir.name}/\n\n"
            f"Issues com label `spec-{spec_id}`."
        )
        items.append((spec_id, ms_title, desc))
    return items


def fetch_milestones() -> dict[str, int]:
    by_title: dict[str, int] = {}
    page = 1
    while True:
        batch = api("GET", f"/repos/{OWNER}/{REPO}/milestones?state=all&limit=50&page={page}")
        if not batch:
            break
        for ms in batch:
            by_title[ms["title"]] = ms["id"]
        if len(batch) < 50:
            break
        page += 1
    return by_title


def ensure_milestone(spec_id: str, title: str, desc: str, cache: dict[str, int], dry_run: bool) -> int:
    if title in cache:
        return cache[title]
    if dry_run:
        cache[title] = -1
        return -1
    ms = api("POST", f"/repos/{OWNER}/{REPO}/milestones", {"title": title, "description": desc, "state": "open"})
    cache[title] = ms["id"]
    return ms["id"]


def fetch_all_issues() -> list[dict]:
    issues: list[dict] = []
    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
            issues.append(issue)
        if len(batch) < 50:
            break
        page += 1
    return issues


def spec_id_from_issue(issue: dict) -> str | None:
    for lb in issue.get("labels", []):
        name = lb.get("name", "")
        if name.startswith("spec-"):
            return name.replace("spec-", "", 1)
    m = re.search(r"\[spec-(\d{3})\]", issue.get("title", ""))
    return m.group(1) if m else None


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--sleep", type=float, default=0.05)
    args = parser.parse_args()

    specs = load_specs()
    print(f"==> Specs: {len(specs)}")

    ms_cache = {} if args.dry_run else fetch_milestones()
    spec_to_ms: dict[str, int] = {}

    for spec_id, title, desc in specs:
        mid = ensure_milestone(spec_id, title, desc, ms_cache, args.dry_run)
        spec_to_ms[spec_id] = mid
        if not args.dry_run:
            time.sleep(args.sleep)

    print(f"==> Milestones: {len(spec_to_ms)}")

    issues = fetch_all_issues() if not args.dry_run else []
    print(f"==> Issues a associar: {len(issues)}")

    linked = skipped = 0
    for issue in issues:
        sid = spec_id_from_issue(issue)
        if not sid or sid not in spec_to_ms:
            skipped += 1
            continue
        mid = spec_to_ms[sid]
        if issue.get("milestone") and issue["milestone"].get("id") == mid:
            skipped += 1
            continue
        if args.dry_run:
            linked += 1
            continue
        api("PATCH", f"/repos/{OWNER}/{REPO}/issues/{issue['number']}", {"milestone": mid})
        linked += 1
        if linked % 100 == 0:
            print(f"  ... {linked} associadas")
        time.sleep(args.sleep)

    print(f"==> Concluído: milestones={len(spec_to_ms)} issues_linked={linked} skipped={skipped}")
    if not args.dry_run:
        open_ms = api("GET", f"/repos/{OWNER}/{REPO}/milestones?state=open&limit=1")
        print(f"==> URL: https://git.spec.ligbox.com.br/{OWNER}/{REPO}/milestones")
    return 0


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