myxce/scripts/gitea-sync.py

#!/usr/bin/env python3
"""
xce-gitea-sync: Sync cc-sdd specs and OpenWolf buglog to Gitea issues.

Reads .kiro/specs/<feature>/ (spec.json, brief.md, requirements.md, tasks.md)
and .wolf/buglog.json as ground truth. Creates/updates Gitea milestones, issues,
and labels to mirror the local state.

Sync is one-way: files -> Gitea. Never Gitea -> files.
Sync state (issue numbers) is stored in the source files for idempotency.

Usage:
    python scripts/gitea-sync.py                    # sync all specs + bugs
    python scripts/gitea-sync.py --spec <name>      # sync one spec
    python scripts/gitea-sync.py --bugs             # sync bugs only
    python scripts/gitea-sync.py --dry-run           # show what would change

Environment:
    GITEA_URL    - base URL (default: https://gitea.shenrs.eu)
    GITEA_TOKEN  - API token (required: write:issue, write:milestone scopes)
    GITEA_OWNER  - repo owner (default: nexa)
    GITEA_REPO   - repo name (default: xce)
"""

import argparse
import datetime
import hashlib
import json
import os
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path


# ── Helpers ──────────────────────────────────────────────────────────────────

def die(msg):
    print(f"ERROR: {msg}", file=sys.stderr)
    sys.exit(1)


def warn(msg):
    print(f"WARN: {msg}", file=sys.stderr)


def _env(key, default=None, required=False):
    val = os.environ.get(key, default)
    if required and not val:
        die(f"{key} environment variable is required")
    return val


def _discover_gitea_token():
    """Discover Gitea API token from multiple sources.

    Priority:
    1. GITEA_TOKEN environment variable
    2. ~/.claude.json → mcpServers.gitea.env.GITEA_ACCESS_TOKEN
    3. GITEA_ACCESS_TOKEN environment variable
    """
    # Explicit override
    token = _env("GITEA_TOKEN")
    if token:
        return token

    # Try Claude Code MCP config
    claude_config_path = Path.home() / ".claude.json"
    if claude_config_path.exists():
        try:
            config = json.loads(claude_config_path.read_text())
            gitea_cfg = config.get("mcpServers", {}).get("gitea", {})
            token = gitea_cfg.get("env", {}).get("GITEA_ACCESS_TOKEN")
            if token:
                return token
        except (json.JSONDecodeError, OSError):
            pass

    # MCP env var
    return _env("GITEA_ACCESS_TOKEN")


# ── Configuration ────────────────────────────────────────────────────────────

GITEA_URL = _env("GITEA_URL", "https://gitea.shenrs.eu").rstrip("/")
GITEA_TOKEN = _discover_gitea_token()
GITEA_OWNER = _env("GITEA_OWNER", "nexa")
GITEA_REPO = _env("GITEA_REPO", "xce")

API_BASE = f"{GITEA_URL}/api/v1"
REPO_API = f"{API_BASE}/repos/{GITEA_OWNER}/{GITEA_REPO}"

# Project roots (relative to script or cwd)
SCRIPT_DIR = Path(__file__).resolve().parent
PROJECT_ROOT = SCRIPT_DIR.parent
SPECS_DIR = PROJECT_ROOT / ".kiro" / "specs"
BUGLOG_PATH = PROJECT_ROOT / ".wolf" / "buglog.json"


# ── Label Definitions ────────────────────────────────────────────────────────

LABELS = [
    # Spec phase labels
    {"name": "phase:requirements", "color": "6C5CE7",
     "description": "Spec requirements phase — EARS entries, acceptance criteria"},
    {"name": "phase:design", "color": "6C5CE7",
     "description": "Spec design phase — architecture, data flow, API contracts"},
    {"name": "phase:tasks", "color": "6C5CE7",
     "description": "Spec tasks phase — implementation task breakdown"},
    {"name": "phase:implementation", "color": "F39C12",
     "description": "Spec implementation phase — code in progress"},
    {"name": "phase:complete", "color": "27AE60",
     "description": "Spec complete — all tasks done, validated, merged"},
    # Top-level type labels
    {"name": "spec", "color": "3498DB",
     "description": "Spec-driven development tracked issue"},
    {"name": "bug", "color": "E74C3C",
     "description": "Bug found during development"},
    {"name": "bug:fixed", "color": "27AE60",
     "description": "Bug resolved and fix committed"},
    # State labels
    {"name": "blocked", "color": "E74C3C",
     "description": "Blocked on external dependency or upstream spec"},
]

# Map buglog tags to Gitea labels (prefix-based matching)
TAG_TO_LABEL = {
    "asan": "bug:asan",
    "ubsan": "bug:asan",
    "SEGV": "bug:crash",
    "double-free": "bug:crash",
    "heap-use-after-free": "bug:crash",
    "UAF": "bug:crash",
    "data-loss": "bug:data-loss",
    "race": "bug:race",
    "toctou": "bug:race",
    "data-race": "bug:race",
    "ha": "bug:ha",
    "raft": "bug:ha",
    "election": "bug:ha",
    "failover": "bug:ha",
}


def api_url(path):
    """Build full API URL. All paths are repo-scoped."""
    return f"{REPO_API}/{path.lstrip('/')}"


def request(method, path, body=None):
    """Make a Gitea API request. Returns (status, response_body) or None on 404."""
    url = api_url(path)
    data = json.dumps(body).encode("utf-8") if body else None

    req = urllib.request.Request(url, data=data, method=method)
    req.add_header("Authorization", f"token {GITEA_TOKEN}")
    req.add_header("Content-Type", "application/json")
    req.add_header("Accept", "application/json")

    try:
        with urllib.request.urlopen(req) as resp:
            return resp.status, json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        if e.code == 404:
            return None, None
        body_text = ""
        try:
            body_text = e.read().decode("utf-8")
        except Exception:
            pass
        raise RuntimeError(
            f"Gitea API error: {method} {path} -> {e.code}\n{body_text}"
        ) from e
    except urllib.error.URLError as e:
        raise RuntimeError(
            f"Gitea API unreachable: {method} {path}\n{e.reason}"
        ) from e


# ── GiteaAPI ─────────────────────────────────────────────────────────────────