639 lines
21 KiB
Python
Executable file
639 lines
21 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Discover all origin branches, emit branch-status.json, optional edition rebuild.
|
|
|
|
Dynamic cascade for dossiers.ngi.hacktivism.ch:
|
|
|
|
• every origin/* tip (not a hard-coded ladder)
|
|
• parent = nearest other tip that is a git ancestor
|
|
• sticky bar reads this JSON and draws a semi-graphic tree
|
|
• optional --rebuild / REBUILD=1: worktree + make html into editions/<slug>/
|
|
only when .edition-tip differs (on demand; timer ~10min)
|
|
|
|
Env:
|
|
|
|
TDPD_ROOT tech-dossier-project-data checkout (may be git-dir only)
|
|
OUT branch-status.json path (live www on koopa)
|
|
EDITIONS_ROOT editions/ dir on www (required for rebuild)
|
|
REBUILD 1 → rebuild stale/missing freezes
|
|
FETCH 0 → skip git fetch (default 1)
|
|
FREEZE_CMD override build (default: make html); `{root}` `{branch}` `{slug}`
|
|
DEFAULT_BRANCH branch behind cookie flask (default feat/website-views)
|
|
HOST public host URL
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
ADMIN_LOG = Path(__file__).resolve().parents[2]
|
|
_DEFAULT_OUT = ADMIN_LOG / "sites" / "dossiers.ngi.hacktivism.ch" / "branch-status.json"
|
|
_PLACEHOLDER = (
|
|
ADMIN_LOG / "sites" / "dossiers.ngi.hacktivism.ch" / "_edition-placeholder" / "index.html"
|
|
)
|
|
HOST_DEFAULT = "https://dossiers.ngi.hacktivism.ch"
|
|
DEFAULT_TDPD = Path.home() / "git" / "src" / "ngi" / "tech-dossier-project-data"
|
|
INJECT = Path(__file__).resolve().parent / "inject-cascade-bar.py"
|
|
SLUG_RE = re.compile(r"[^A-Za-z0-9_-]+")
|
|
|
|
|
|
def env_path(name: str, default: Path | None = None) -> Path | None:
|
|
raw = os.environ.get(name, "").strip()
|
|
if raw:
|
|
return Path(raw).expanduser()
|
|
return default
|
|
|
|
|
|
def out_path() -> Path:
|
|
return env_path("OUT", _DEFAULT_OUT) # type: ignore[return-value]
|
|
|
|
|
|
def tdpd_root() -> Path:
|
|
return env_path("TDPD_ROOT", DEFAULT_TDPD) # type: ignore[return-value]
|
|
|
|
|
|
def editions_root() -> Path | None:
|
|
return env_path("EDITIONS_ROOT", None)
|
|
|
|
|
|
def host() -> str:
|
|
return os.environ.get("HOST", "").strip() or HOST_DEFAULT
|
|
|
|
|
|
def default_branch() -> str:
|
|
return os.environ.get("DEFAULT_BRANCH", "").strip() or "feat/website-views"
|
|
|
|
|
|
def edition_slug(branch: str) -> str:
|
|
return SLUG_RE.sub("-", branch.replace("/", "-")).strip("-") or "branch"
|
|
|
|
|
|
def git(repo: Path, *args: str, check: bool = True, text: bool = True) -> subprocess.CompletedProcess:
|
|
return subprocess.run(
|
|
["git", "-C", str(repo), *args],
|
|
check=check,
|
|
text=text,
|
|
capture_output=True,
|
|
)
|
|
|
|
|
|
def fetch_origin(repo: Path) -> None:
|
|
if os.environ.get("FETCH", "1").strip() == "0":
|
|
return
|
|
subprocess.run(
|
|
["git", "-C", str(repo), "fetch", "--prune", "origin"],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
|
|
def list_origin_branches(repo: Path) -> dict[str, str]:
|
|
"""branch name → full SHA for refs/remotes/origin/* (no HEAD)."""
|
|
cp = git(repo, "for-each-ref", "--format=%(refname:short)|%(objectname)", "refs/remotes/origin")
|
|
out: dict[str, str] = {}
|
|
for line in cp.stdout.splitlines():
|
|
if "|" not in line:
|
|
continue
|
|
short, sha = line.split("|", 1)
|
|
if short in ("origin", "origin/HEAD") or "->" in short:
|
|
continue
|
|
if not short.startswith("origin/"):
|
|
continue
|
|
name = short[len("origin/") :]
|
|
if name == "HEAD":
|
|
continue
|
|
out[name] = sha
|
|
return out
|
|
|
|
|
|
def is_ancestor(repo: Path, anc: str, desc: str) -> bool:
|
|
cp = git(repo, "merge-base", "--is-ancestor", anc, desc, check=False)
|
|
return cp.returncode == 0
|
|
|
|
|
|
def commits_between(repo: Path, a: str, b: str) -> int:
|
|
cp = git(repo, "rev-list", "--count", f"{a}..{b}", check=False)
|
|
if cp.returncode != 0:
|
|
return 10**9
|
|
try:
|
|
return int(cp.stdout.strip() or "0")
|
|
except ValueError:
|
|
return 10**9
|
|
|
|
|
|
def short_sha(sha: str) -> str:
|
|
return sha[:7] if sha else ""
|
|
|
|
|
|
def nearest_parents(repo: Path, tips: dict[str, str]) -> dict[str, str]:
|
|
"""For each branch, nearest other tip that is a strict ancestor."""
|
|
parents: dict[str, str] = {}
|
|
names = list(tips)
|
|
for b in names:
|
|
tip_b = tips[b]
|
|
cands: list[tuple[int, str]] = []
|
|
for a in names:
|
|
if a == b:
|
|
continue
|
|
tip_a = tips[a]
|
|
if tip_a == tip_b:
|
|
continue
|
|
if not is_ancestor(repo, tip_a, tip_b):
|
|
continue
|
|
cands.append((commits_between(repo, tip_a, tip_b), a))
|
|
parents[b] = min(cands)[1] if cands else ""
|
|
return parents
|
|
|
|
|
|
def children_map(parents: dict[str, str]) -> dict[str, list[str]]:
|
|
kids: dict[str, list[str]] = {b: [] for b in parents}
|
|
orphans: list[str] = []
|
|
for b, p in parents.items():
|
|
if p and p in kids:
|
|
kids[p].append(b)
|
|
else:
|
|
orphans.append(b)
|
|
for p in kids:
|
|
kids[p].sort()
|
|
orphans.sort()
|
|
# roots = orphans (parent missing or empty)
|
|
kids[""] = orphans
|
|
return kids
|
|
|
|
|
|
def walk_tree(
|
|
parents: dict[str, str],
|
|
tips: dict[str, str],
|
|
) -> list[dict]:
|
|
"""Flat rows in DFS order with semi-graphic prefix fields."""
|
|
kids = children_map(parents)
|
|
rows: list[dict] = []
|
|
|
|
def walk(name: str, prefix: str, is_last: bool, is_root: bool) -> None:
|
|
tip = tips[name]
|
|
if is_root:
|
|
branch_prefix = ""
|
|
child_prefix = ""
|
|
else:
|
|
branch_prefix = prefix + ("└── " if is_last else "├── ")
|
|
child_prefix = prefix + (" " if is_last else "│ ")
|
|
rows.append(
|
|
{
|
|
"branch": name,
|
|
"edition": edition_slug(name),
|
|
"tip": short_sha(tip),
|
|
"tip_full": tip,
|
|
"parent": parents.get(name, ""),
|
|
"prefix": branch_prefix,
|
|
"url": f"/_cascade/use/{edition_slug(name)}",
|
|
}
|
|
)
|
|
ch = kids.get(name, [])
|
|
for i, c in enumerate(ch):
|
|
walk(c, child_prefix, i == len(ch) - 1, False)
|
|
|
|
roots = kids.get("", [])
|
|
for i, r in enumerate(roots):
|
|
walk(r, "", i == len(roots) - 1, True)
|
|
return rows
|
|
|
|
|
|
def ensure_placeholder(editions: Path, slug: str, tip: str) -> None:
|
|
dest = editions / slug
|
|
dest.mkdir(parents=True, exist_ok=True)
|
|
idx = dest / "index.html"
|
|
if not idx.exists():
|
|
src = _PLACEHOLDER
|
|
if src.is_file():
|
|
idx.write_text(src.read_text(encoding="utf-8"), encoding="utf-8")
|
|
else:
|
|
idx.write_text(
|
|
"<!DOCTYPE html><html><body>"
|
|
'{% include "_cascade/bar.html" . %}'
|
|
"<h1>edition pending</h1></body></html>\n",
|
|
encoding="utf-8",
|
|
)
|
|
tip_path = dest / ".edition-tip"
|
|
if not tip_path.exists():
|
|
tip_path.write_text(tip + "\n", encoding="utf-8")
|
|
|
|
|
|
def tip_recorded(editions: Path, slug: str) -> str:
|
|
p = editions / slug / ".edition-tip"
|
|
if not p.is_file():
|
|
return ""
|
|
return p.read_text(encoding="utf-8").strip()
|
|
|
|
|
|
def tip_commit_time(repo: Path, tip: str) -> str:
|
|
"""Committer ISO time of tip (empty on failure)."""
|
|
cp = git(repo, "log", "-1", "--format=%cI", tip, check=False)
|
|
if cp.returncode != 0:
|
|
return ""
|
|
return (cp.stdout or "").strip()
|
|
|
|
|
|
def read_build_meta(editions: Path, slug: str) -> dict:
|
|
dest = editions / slug
|
|
out: dict = {}
|
|
at = dest / ".edition-built-at"
|
|
sec = dest / ".edition-build-seconds"
|
|
if at.is_file():
|
|
out["built_at"] = at.read_text(encoding="utf-8").strip()
|
|
if sec.is_file():
|
|
raw = sec.read_text(encoding="utf-8").strip()
|
|
try:
|
|
out["build_seconds"] = float(raw)
|
|
except ValueError:
|
|
pass
|
|
return out
|
|
|
|
|
|
def write_build_meta(editions: Path, slug: str, seconds: float) -> None:
|
|
dest = editions / slug
|
|
dest.mkdir(parents=True, exist_ok=True)
|
|
now = datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
|
|
(dest / ".edition-built-at").write_text(now + "\n", encoding="utf-8")
|
|
(dest / ".edition-build-seconds").write_text(f"{seconds:.3f}\n", encoding="utf-8")
|
|
|
|
|
|
def cleanup_stale_editions(editions: Path, tips: dict[str, str]) -> list[str]:
|
|
"""Drop edition dirs whose branch tip is gone (keep flask alias)."""
|
|
keep = {edition_slug(b) for b in tips} | {"flask"}
|
|
removed: list[str] = []
|
|
if not editions.is_dir():
|
|
return removed
|
|
for child in list(editions.iterdir()):
|
|
if not child.is_dir():
|
|
continue
|
|
if child.name in keep:
|
|
continue
|
|
shutil.rmtree(child, ignore_errors=True)
|
|
removed.append(child.name)
|
|
return removed
|
|
|
|
|
|
def cleanup_orphan_tmps(max_age_sec: int = 3600) -> int:
|
|
"""Remove leftover tdpd-* workdirs older than max_age_sec."""
|
|
bases: list[Path] = []
|
|
home_tmp = Path.home() / "tmp"
|
|
if home_tmp.is_dir():
|
|
bases.append(home_tmp)
|
|
bases.append(Path(tempfile.gettempdir()))
|
|
cutoff = time.time() - max_age_sec
|
|
n = 0
|
|
for base in bases:
|
|
try:
|
|
kids = list(base.glob("tdpd-*"))
|
|
except OSError:
|
|
continue
|
|
for path in kids:
|
|
if not path.is_dir():
|
|
continue
|
|
try:
|
|
if path.stat().st_mtime >= cutoff:
|
|
continue
|
|
except OSError:
|
|
continue
|
|
shutil.rmtree(path, ignore_errors=True)
|
|
n += 1
|
|
return n
|
|
|
|
|
|
def cleanup_repo(repo: Path) -> None:
|
|
git(repo, "worktree", "prune", check=False)
|
|
|
|
|
|
def _attach_venv(work: Path) -> None:
|
|
"""Worktrees lack .venv; reuse TDPD_VENV or sibling checkout .venv via symlink."""
|
|
if (work / ".venv" / "bin" / "python").exists():
|
|
return
|
|
candidates: list[Path] = []
|
|
raw = os.environ.get("TDPD_VENV", "").strip()
|
|
if raw:
|
|
candidates.append(Path(raw).expanduser())
|
|
root = tdpd_root()
|
|
candidates.append(root / ".venv")
|
|
# laptop SoT checkout even when TDPD_ROOT is git-dir-only on koopa
|
|
candidates.append(DEFAULT_TDPD / ".venv")
|
|
for venv in candidates:
|
|
py = venv / "bin" / "python"
|
|
if py.is_file():
|
|
dest = work / ".venv"
|
|
if dest.exists() or dest.is_symlink():
|
|
return
|
|
dest.symlink_to(venv.resolve())
|
|
return
|
|
|
|
|
|
def run_freeze(work: Path, branch: str, slug: str) -> None:
|
|
tmpl = os.environ.get("FREEZE_CMD", "").strip()
|
|
env = os.environ.copy()
|
|
env["TDPD_BRANCH"] = branch
|
|
env["TDPD_SLUG"] = slug
|
|
if tmpl:
|
|
cmd = tmpl.format(root=str(work), branch=branch, slug=slug)
|
|
subprocess.run(cmd, shell=True, check=True, cwd=str(work), env=env)
|
|
return
|
|
_attach_venv(work)
|
|
# Older branches hardcode `python3 -m website.freeze` without .venv;
|
|
# prefer the attached venv interpreter directly.
|
|
py = work / ".venv" / "bin" / "python"
|
|
freeze_mod = work / "website" / "freeze.py"
|
|
if py.is_file() and freeze_mod.is_file():
|
|
build = work / "website" / "build"
|
|
if build.exists():
|
|
shutil.rmtree(build)
|
|
subprocess.run(
|
|
[str(py), "-m", "website.freeze"],
|
|
check=True,
|
|
cwd=str(work),
|
|
env=env,
|
|
)
|
|
return
|
|
makefile = work / "Makefile"
|
|
if not makefile.is_file():
|
|
raise FileNotFoundError(f"no Makefile in {work}")
|
|
subprocess.run(["make", "html"], check=True, cwd=str(work), env=env)
|
|
|
|
|
|
def inject_bar(edition_dir: Path) -> None:
|
|
if not INJECT.is_file():
|
|
return
|
|
subprocess.run(
|
|
[sys.executable, str(INJECT), str(edition_dir)],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
|
|
def rebuild_edition(repo: Path, branch: str, tip: str, editions: Path) -> dict:
|
|
"""Return {status, seconds} — status skipped|built|placeholder|error:…"""
|
|
slug = edition_slug(branch)
|
|
dest = editions / slug
|
|
t0 = time.perf_counter()
|
|
force = os.environ.get("FORCE_REBUILD", "").strip() in ("1", "true", "yes")
|
|
if (
|
|
not force
|
|
and tip_recorded(editions, slug) == tip
|
|
and (dest / "index.html").is_file()
|
|
):
|
|
inject_bar(dest)
|
|
return {"status": "skipped", "seconds": round(time.perf_counter() - t0, 3)}
|
|
|
|
wt: Path | None = None
|
|
try:
|
|
tmp_base = Path.home() / "tmp" if (Path.home() / "tmp").is_dir() else Path(tempfile.gettempdir())
|
|
wt = Path(tempfile.mkdtemp(prefix=f"tdpd-{slug}-", dir=str(tmp_base)))
|
|
cp = git(repo, "worktree", "add", "--detach", str(wt), tip, check=False)
|
|
if cp.returncode != 0:
|
|
git(repo, "worktree", "prune", check=False)
|
|
cp = git(repo, "worktree", "add", "--detach", str(wt), tip, check=False)
|
|
if cp.returncode != 0:
|
|
ensure_placeholder(editions, slug, tip)
|
|
return {
|
|
"status": f"error:worktree:{cp.stderr.strip()[:200]}",
|
|
"seconds": round(time.perf_counter() - t0, 3),
|
|
}
|
|
|
|
run_freeze(wt, branch, slug)
|
|
build = wt / "website" / "build"
|
|
if not build.is_dir():
|
|
ensure_placeholder(editions, slug, tip)
|
|
return {"status": "error:no-build", "seconds": round(time.perf_counter() - t0, 3)}
|
|
|
|
dest.mkdir(parents=True, exist_ok=True)
|
|
keep_meta = {
|
|
".edition-tip",
|
|
".edition-built-at",
|
|
".edition-build-seconds",
|
|
}
|
|
for child in list(dest.iterdir()):
|
|
if child.name in keep_meta:
|
|
continue
|
|
if child.is_dir():
|
|
shutil.rmtree(child)
|
|
else:
|
|
child.unlink()
|
|
for child in build.iterdir():
|
|
target = dest / child.name
|
|
if child.is_dir():
|
|
shutil.copytree(child, target)
|
|
else:
|
|
shutil.copy2(child, target)
|
|
secs = round(time.perf_counter() - t0, 3)
|
|
(dest / ".edition-tip").write_text(tip + "\n", encoding="utf-8")
|
|
write_build_meta(editions, slug, secs)
|
|
inject_bar(dest)
|
|
return {"status": "built", "seconds": secs}
|
|
except Exception as exc: # noqa: BLE001 — hourly must continue
|
|
ensure_placeholder(editions, slug, tip)
|
|
return {
|
|
"status": f"error:{type(exc).__name__}:{exc}",
|
|
"seconds": round(time.perf_counter() - t0, 3),
|
|
}
|
|
finally:
|
|
if wt is not None:
|
|
git(repo, "worktree", "remove", "--force", str(wt), check=False)
|
|
shutil.rmtree(wt, ignore_errors=True)
|
|
|
|
|
|
def sync_flask_alias(editions: Path, tips: dict[str, str]) -> None:
|
|
"""Keep editions/flask in sync with DEFAULT_BRANCH freeze when present."""
|
|
src_branch = default_branch()
|
|
if src_branch not in tips:
|
|
return
|
|
src = editions / edition_slug(src_branch)
|
|
dst = editions / "flask"
|
|
tip = tips[src_branch]
|
|
if tip_recorded(editions, "flask") == tip and (dst / "index.html").is_file():
|
|
return
|
|
if not (src / "index.html").is_file():
|
|
ensure_placeholder(editions, "flask", tip)
|
|
return
|
|
if dst.exists():
|
|
shutil.rmtree(dst)
|
|
shutil.copytree(src, dst)
|
|
(dst / ".edition-tip").write_text(tip + "\n", encoding="utf-8")
|
|
# reuse src build meta if present
|
|
for name in (".edition-built-at", ".edition-build-seconds"):
|
|
s = src / name
|
|
if s.is_file():
|
|
shutil.copy2(s, dst / name)
|
|
inject_bar(dst)
|
|
|
|
|
|
def build_payload(repo: Path, tips: dict[str, str], rebuild: bool) -> dict:
|
|
parents = nearest_parents(repo, tips)
|
|
rows = walk_tree(parents, tips)
|
|
rebuild_status: dict[str, dict] = {}
|
|
rebuild_timing: dict | None = None
|
|
ed_root = editions_root()
|
|
cleanup_repo(repo)
|
|
n_tmp = cleanup_orphan_tmps()
|
|
if n_tmp:
|
|
print(f"cleanup_tmp removed={n_tmp}", file=sys.stderr)
|
|
if ed_root is not None:
|
|
ed_root.mkdir(parents=True, exist_ok=True)
|
|
removed = cleanup_stale_editions(ed_root, tips)
|
|
if removed:
|
|
print(f"cleanup_editions removed={removed}", file=sys.stderr)
|
|
if rebuild and ed_root is not None:
|
|
wall0 = time.perf_counter()
|
|
for name, tip in sorted(tips.items()):
|
|
# rebuild_edition itself skips when .edition-tip matches tip
|
|
rebuild_status[name] = rebuild_edition(repo, name, tip, ed_root)
|
|
sync_flask_alias(ed_root, tips)
|
|
cleanup_repo(repo)
|
|
cleanup_orphan_tmps(max_age_sec=600)
|
|
wall = round(time.perf_counter() - wall0, 3)
|
|
built_secs = [
|
|
v["seconds"]
|
|
for v in rebuild_status.values()
|
|
if v.get("status") == "built"
|
|
]
|
|
rebuild_timing = {
|
|
"wall_seconds": wall,
|
|
"branches": len(tips),
|
|
"built": sum(1 for v in rebuild_status.values() if v.get("status") == "built"),
|
|
"skipped": sum(1 for v in rebuild_status.values() if v.get("status") == "skipped"),
|
|
"errors": sum(
|
|
1 for v in rebuild_status.values() if str(v.get("status", "")).startswith("error")
|
|
),
|
|
"built_seconds_sum": round(sum(built_secs), 3) if built_secs else 0.0,
|
|
"built_seconds_max": round(max(built_secs), 3) if built_secs else 0.0,
|
|
}
|
|
print(
|
|
f"REBUILD_TIMING wall_seconds={wall} built={rebuild_timing['built']} "
|
|
f"skipped={rebuild_timing['skipped']} errors={rebuild_timing['errors']} "
|
|
f"built_seconds_sum={rebuild_timing['built_seconds_sum']} "
|
|
f"built_seconds_max={rebuild_timing['built_seconds_max']}",
|
|
file=sys.stderr,
|
|
)
|
|
elif ed_root is not None:
|
|
ed_root.mkdir(parents=True, exist_ok=True)
|
|
for name, tip in tips.items():
|
|
slug = edition_slug(name)
|
|
if not (ed_root / slug / "index.html").is_file():
|
|
ensure_placeholder(ed_root, slug, tip)
|
|
rebuild_status[name] = {"status": "placeholder", "seconds": 0.0}
|
|
if not (ed_root / "flask" / "index.html").is_file():
|
|
sync_flask_alias(ed_root, tips)
|
|
|
|
tip_times = {name: tip_commit_time(repo, tip) for name, tip in tips.items()}
|
|
branches_out = []
|
|
for row in rows:
|
|
name = row["branch"]
|
|
slug = row["edition"]
|
|
meta = read_build_meta(ed_root, slug) if ed_root is not None else {}
|
|
# prefer this-run built seconds when present
|
|
rb = rebuild_status.get(name) or {}
|
|
if rb.get("status") == "built" and "seconds" in rb:
|
|
meta = {
|
|
**meta,
|
|
"build_seconds": rb["seconds"],
|
|
}
|
|
entry = {
|
|
"branch": name,
|
|
"edition": slug,
|
|
"tip": row["tip"],
|
|
"tip_time": tip_times.get(name, ""),
|
|
"parent": row["parent"],
|
|
"prefix": row["prefix"],
|
|
"url": row["url"],
|
|
}
|
|
if meta.get("built_at"):
|
|
entry["built_at"] = meta["built_at"]
|
|
if "build_seconds" in meta:
|
|
entry["build_seconds"] = meta["build_seconds"]
|
|
branches_out.append(entry)
|
|
|
|
payload = {
|
|
"generated": datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds"),
|
|
"host": host(),
|
|
"default_branch": default_branch(),
|
|
"default_edition": "flask",
|
|
"branches": branches_out,
|
|
}
|
|
if rebuild_status:
|
|
payload["rebuild"] = rebuild_status
|
|
if rebuild_timing is not None:
|
|
payload["rebuild_timing"] = rebuild_timing
|
|
return payload
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
p = argparse.ArgumentParser(description=__doc__)
|
|
p.add_argument(
|
|
"--rebuild",
|
|
action="store_true",
|
|
help="rebuild stale editions into EDITIONS_ROOT",
|
|
)
|
|
p.add_argument(
|
|
"--force",
|
|
action="store_true",
|
|
help="rebuild every edition even when .edition-tip matches (timing)",
|
|
)
|
|
p.add_argument(
|
|
"--no-fetch",
|
|
action="store_true",
|
|
help="skip git fetch --prune origin",
|
|
)
|
|
return p.parse_args(argv)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
repo = tdpd_root()
|
|
if not (repo / ".git").exists() and not (repo / "HEAD").exists():
|
|
# allow bare / gitfile
|
|
gitfile = repo / ".git"
|
|
if not repo.is_dir():
|
|
print(f"TDPD_ROOT missing or not a dir: {repo}", file=sys.stderr)
|
|
return 1
|
|
if not gitfile.exists() and not (repo / "objects").exists():
|
|
print(f"TDPD_ROOT has no git metadata: {repo}", file=sys.stderr)
|
|
return 1
|
|
|
|
if args.no_fetch:
|
|
os.environ["FETCH"] = "0"
|
|
if args.force:
|
|
os.environ["FORCE_REBUILD"] = "1"
|
|
fetch_origin(repo)
|
|
|
|
tips = list_origin_branches(repo)
|
|
if not tips:
|
|
print("no origin/* branches found", file=sys.stderr)
|
|
return 1
|
|
|
|
rebuild = (
|
|
args.rebuild
|
|
or args.force
|
|
or os.environ.get("REBUILD", "").strip() in ("1", "true", "yes")
|
|
)
|
|
if rebuild and editions_root() is None:
|
|
print("REBUILD set but EDITIONS_ROOT missing", file=sys.stderr)
|
|
return 1
|
|
|
|
payload = build_payload(repo, tips, rebuild)
|
|
dest = out_path()
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
dest.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
|
|
print(f"wrote {dest} branches={len(payload['branches'])}", file=sys.stderr)
|
|
if "rebuild" in payload:
|
|
print(f"rebuild={payload['rebuild']}", file=sys.stderr)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|