feat(forgejo): sincronizar 852 tasks.md → issues documentadas
Script idempotente sync-tasks-to-forgejo-issues.py; docs CT130 com URLs directas org/repo/issues e nota sobre tab organizations sem login. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Ligbox Spec Hub
parent
9541201431
commit
a50901c5a1
3 changed files with 276 additions and 4 deletions
|
|
@ -81,6 +81,7 @@ GitHub **não substitui** Forgejo; é réplica read-only ou bidirecional acordad
|
|||
- [ ] `refresh-spec-driver.sh` executado no CT130
|
||||
- [ ] Spec visível em https://spec.ligbox.com.br/specs/
|
||||
- [ ] Commit + push Forgejo
|
||||
- [ ] Se `tasks.md` mudou → `python3 scripts/sync-tasks-to-forgejo-issues.py`
|
||||
- [ ] Push/mirror GitHub
|
||||
|
||||
## 7. Falhas conhecidas a evitar
|
||||
|
|
|
|||
|
|
@ -18,6 +18,9 @@
|
|||
| https://spec.ligbox.com.br/specs/ | Browser Spec Kit |
|
||||
| https://spec.ligbox.com.br/vault/ | Browser Obsidian |
|
||||
| https://git.spec.ligbox.com.br | Forgejo Git |
|
||||
| https://git.spec.ligbox.com.br/ligbox | **Org Ligbox** (repos) |
|
||||
| https://git.spec.ligbox.com.br/ligbox/ligbox-ops-platform | **Repo principal** |
|
||||
| https://git.spec.ligbox.com.br/ligbox/ligbox-ops-platform/issues | **Issues** (sync `tasks.md`) |
|
||||
|
||||
## Stack local
|
||||
|
||||
|
|
@ -32,10 +35,19 @@
|
|||
|
||||
## Repos Forgejo (org `ligbox`)
|
||||
|
||||
| Repo | Conteúdo |
|
||||
|------|----------|
|
||||
| `ligbox-ops-platform` | Monorepo — **todas** VMs (deploy/ por VM) |
|
||||
| `obsidian-vault` | Notas, LAPTOP, anais |
|
||||
| Repo | Conteúdo | Issues |
|
||||
|------|----------|--------|
|
||||
| [`ligbox-ops-platform`](https://git.spec.ligbox.com.br/ligbox/ligbox-ops-platform) | Monorepo — **todas** VMs | [852 issues](https://git.spec.ligbox.com.br/ligbox/ligbox-ops-platform/issues) (sync `tasks.md`) |
|
||||
| [`obsidian-vault`](https://git.spec.ligbox.com.br/ligbox/obsidian-vault) | Notas, LAPTOP, anais | — |
|
||||
|
||||
**Navegação Forgejo:** a homepage `/?repo-search-tab=organizations` sem login mostra landing vazia.
|
||||
Usar directamente: **https://git.spec.ligbox.com.br/ligbox** ou **Explore → Repositories**.
|
||||
|
||||
**Sincronizar issues** (após alterar `tasks.md`):
|
||||
|
||||
```bash
|
||||
python3 scripts/sync-tasks-to-forgejo-issues.py
|
||||
```
|
||||
|
||||
## Utilizadores
|
||||
|
||||
|
|
|
|||
259
scripts/sync-tasks-to-forgejo-issues.py
Executable file
259
scripts/sync-tasks-to-forgejo-issues.py
Executable file
|
|
@ -0,0 +1,259 @@
|
|||
#!/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())
|
||||
Loading…
Reference in a new issue