#!/usr/bin/env python3 """ TP36 (reframed) — teilprojekte.md Index/Status → Gitea Issues (admin/bf-infra). Idempotenter Sync (get-or-create, Match ueber "TPnn" im Titel): - Archiv-Header in docs/teilprojekte-archiv.md -> closed Issues (nur TPs, die in Gitea NICHT offen sind; ein reopentes TP mit Archiv-Zeiger bleibt offen). - Labels status:* (Marker) + thema:* (Gruppierung) werden vorab get-or-create. SSoT-Modell (seit TP36-Cutover 2026-09-07): Gitea fuehrt Index+Status — aktive TPs werden DIREKT als Gitea-Issue angelegt/gepflegt, NICHT mehr aus einer Datei-Tabelle abgeleitet (die Quick-Index-Tabelle wurde aus teilprojekte.md entfernt). Dieses Werkzeug haelt nur noch die ARCHIVIERTEN (closed) TPs mit teilprojekte-archiv.md idempotent abgeglichen und fasst offene Issues nie an. Die per-TP-Prosa bleibt in docs/teilprojekte.md / teilprojekte-archiv.md. Token: ~/.hermes/.env (GITEA_ACCESS_TOKEN, GITEA_HOST) — wird NIE ausgegeben. Aufruf: python3 bin/gitea-tp-sync.py # Dry-Run (default) — zeigt, was passieren wuerde python3 bin/gitea-tp-sync.py --commit # legt Labels + Issues an / gleicht sie ab """ import json import os import re import sys import urllib.request import urllib.error from pathlib import Path REPO = Path(__file__).resolve().parent.parent ACTIVE = REPO / "docs" / "teilprojekte.md" ARCHIV = REPO / "docs" / "teilprojekte-archiv.md" HERMES = Path.home() / ".hermes" / ".env" OWNER_REPO = "admin/bf-infra" COMMIT = "--commit" in sys.argv # Marker -> (status-label, gitea-state) MARKER = { "🔜": ("status:todo", "open"), "⏳": ("status:wip", "open"), "⏸️": ("status:paused", "open"), "💡": ("status:idea", "open"), "🟢": ("status:concept", "open"), "✅": ("status:done", "closed"), "❌": ("status:dropped", "closed"), } LABEL_COLORS = { "status:todo": "#ededed", "status:wip": "#fbca04", "status:paused": "#c5def5", "status:idea": "#d4c5f9", "status:concept": "#bfdadc", "status:done": "#0e8a16", "status:dropped": "#e11d21", "thema:firewall-dns": "#b60205", "thema:wlan": "#1d76db", "thema:cloudflare-proxy": "#f9a825", "thema:nautobot-doku": "#5319e7", "thema:infra": "#0e8a16", "thema:ipv6-matter": "#006b75", "thema:identity": "#e99695", "thema:backlog": "#cccccc", } # TP-Nummer -> Themen-Label (aus dem Statusbaum-Gruppierung; nur aktive TPs brauchen ein Board-Thema) THEME = { "5.1": "firewall-dns", "23": "firewall-dns", "25": "wlan", "28": "wlan", "13": "wlan", "17.1": "wlan", "17.3": "wlan", "16": "cloudflare-proxy", "16.2": "cloudflare-proxy", "16.3": "cloudflare-proxy", "31": "cloudflare-proxy", "41": "cloudflare-proxy", "14.4": "cloudflare-proxy", "42": "cloudflare-proxy", "43": "cloudflare-proxy", "44": "cloudflare-proxy", "32": "nautobot-doku", "34": "nautobot-doku", "35": "nautobot-doku", "36": "nautobot-doku", "37": "nautobot-doku", "30": "infra", "40": "infra", "20.1": "infra", "18": "infra", "24": "infra", "38": "infra", "11": "infra", "12": "ipv6-matter", "12.1": "ipv6-matter", "8": "identity", "29": "backlog", "33": "backlog", } def hermes_env(): tok = host = None for line in HERMES.read_text().splitlines(): m = re.match(r"\s*(GITEA_ACCESS_TOKEN|GITEA_HOST)\s*=\s*(.+?)\s*$", line) if not m: continue if m.group(1) == "GITEA_ACCESS_TOKEN": tok = m.group(2).strip().strip('"').strip("'") else: host = m.group(2).strip().strip('"').strip("'") if not tok: sys.exit("GITEA_ACCESS_TOKEN nicht in %s gefunden." % HERMES) host = (host or "https://gitea.emsenhuber.net").rstrip("/") if not host.startswith("http"): host = "https://" + host return tok, host TOKEN, HOST = hermes_env() API = HOST + "/api/v1" def api(method, path, body=None): url = API + path data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method, headers={ "Authorization": "token " + TOKEN, "Accept": "application/json", "Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=30) as r: return r.status, json.load(r) except urllib.error.HTTPError as e: return e.code, (json.load(e) if e.headers.get("content-type", "").startswith("application/json") else {}) # --------------------------------------------------------------------------- parse def parse_archiv(exclude_nums): """Archiv-Header -> archivierte TPs. exclude_nums = TP-Nummern, die in Gitea OFFEN sind (reopente TPs mit Archiv-Zeiger) -> die bleiben offen, nicht als closed abgleichen.""" out = [] hdr = re.compile(r"^#{2,4}\s*(TP[\d.\-A-Za-z]+)\s*—\s*(.+?)\s*▸", ) seen = set() for line in ARCHIV.read_text().splitlines(): m = hdr.match(line) if not m: continue tp, title = m.groups() num = tp[2:] if num in exclude_nums or num in seen: continue seen.add(num) short = re.sub(r"\*\*|`", "", title).strip()[:80] out.append({"tp": tp, "num": num, "marker": "✅", "title": short, "date": "", "src": "archiv"}) return out # --------------------------------------------------------------------------- gitea sync def ensure_labels(existing): by_name = {l["name"]: l for l in existing} for name, color in LABEL_COLORS.items(): if name in by_name: continue if COMMIT: api("POST", f"/repos/{OWNER_REPO}/labels", {"name": name, "color": color.lstrip("#")}) print(f" [{'+' if COMMIT else 'DRY'}] label {name}") st, cur = api("GET", f"/repos/{OWNER_REPO}/labels?limit=100") return {l["name"]: l["id"] for l in (cur if isinstance(cur, list) else [])} def all_issues(): out, page = [], 1 while True: st, d = api("GET", f"/repos/{OWNER_REPO}/issues?state=all&type=issues&limit=50&page={page}") if not isinstance(d, list) or not d: break out += d page += 1 return out def main(): print(f"Gitea {HOST} repo {OWNER_REPO} mode={'COMMIT' if COMMIT else 'DRY-RUN'}") st, existing_labels = api("GET", f"/repos/{OWNER_REPO}/labels?limit=100") label_id = ensure_labels(existing_labels if isinstance(existing_labels, list) else []) issues = all_issues() by_tp = {} open_nums = set() for iss in issues: # Titel ist "TPxx — …"; volle ID inkl. Buchstaben/Bindestrich (TP14.2a, TP-R) bis zum " — " m = re.match(r"^(TP[\w.\-]*?)\s+—", iss["title"]) if m: by_tp[m.group(1)] = iss if iss.get("state") == "open": open_nums.add(m.group(1)[2:]) # Aktive TPs sind seit dem TP36-Cutover SSoT direkt in Gitea -> nur das Archiv abgleichen, # und offene (reopente) TPs davon ausschliessen, damit sie nicht faelschlich geschlossen werden. items = parse_archiv(open_nums) print(f"Geparst: {len(items)} archivierte TP (aktive TPs = SSoT in Gitea: " f"{len(open_nums)} offen, nicht angetastet)\n") created = updated = 0 for it in items: status_label, state = MARKER.get(it["marker"], ("status:todo", "open")) labels = [status_label] th = THEME.get(it["num"]) if th: labels.append(f"thema:{th}") title = f"{it['tp']} — {it['title']}" body = (f"Steuerungs-SSoT: Index/Status hier in Gitea; **Prosa/Detail** im Repo unter " f"`docs/{'teilprojekte-archiv.md' if it['src']=='archiv' else 'teilprojekte.md'}` " f"(Abschnitt `{it['tp']}`).\n\n" f"- Status: {it['marker']} ({status_label})\n" + (f"- Abschluss: {it['date']}\n" if it["date"] else "")) want_ids = [label_id[l] for l in labels if l in label_id] cur = by_tp.get(it["tp"]) if cur is None: print(f" [{'+' if COMMIT else 'DRY'}] NEW {title} {labels} state={state}") if COMMIT: sc, res = api("POST", f"/repos/{OWNER_REPO}/issues", {"title": title, "body": body, "labels": want_ids}) if state == "closed" and isinstance(res, dict) and res.get("number"): api("PATCH", f"/repos/{OWNER_REPO}/issues/{res['number']}", {"state": "closed"}) created += 1 else: have = {l["name"] for l in cur.get("labels", [])} drift = (have != set(labels)) or (cur.get("state") != state) if drift: print(f" [{'~' if COMMIT else 'DRY'}] UPD #{cur['number']} {title} " f"{sorted(have)}->{labels} state {cur.get('state')}->{state}") if COMMIT: api("PATCH", f"/repos/{OWNER_REPO}/issues/{cur['number']}", {"labels": want_ids, "state": state}) updated += 1 print(f"\n{'Angelegt' if COMMIT else 'Wuerde anlegen'}: {created} · " f"{'Abgeglichen' if COMMIT else 'Wuerde abgleichen'}: {updated}") if not COMMIT: print("Dry-Run — nichts geschrieben. Mit --commit ausfuehren.") if __name__ == "__main__": main()