669 lines
22 KiB
Python
Executable file
669 lines
22 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 diff_numstat(repo: Path, parent_tip: str, tip: str) -> tuple[int, int]:
|
||
"""Insertions/deletions of tip vs parent tip (git diff --numstat)."""
|
||
if not parent_tip or not tip or parent_tip == tip:
|
||
return 0, 0
|
||
cp = git(repo, "diff", "--numstat", main_tip, tip, check=False)
|
||
if cp.returncode != 0:
|
||
return 0, 0
|
||
plus = minus = 0
|
||
for line in (cp.stdout or "").splitlines():
|
||
parts = line.split(" ")
|
||
if len(parts) < 3:
|
||
continue
|
||
a, d = parts[0], parts[1]
|
||
if a == "-" or d == "-":
|
||
continue
|
||
try:
|
||
plus += int(a)
|
||
minus += int(d)
|
||
except ValueError:
|
||
continue
|
||
return plus, minus
|
||
|
||
|
||
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"]
|
||
# sticky bar: +grün / −rot vs parent tip; omit when no parent
|
||
parent = row["parent"]
|
||
if parent and parent in tips:
|
||
plus, minus = diff_numstat(repo, tips[parent], row["tip"])
|
||
entry["vs_parent_plus"] = plus
|
||
entry["vs_parent_minus"] = minus
|
||
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())
|