574 lines
22 KiB
Python
Executable File
574 lines
22 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Vendor or test upstream README/Compose snapshots.
|
|
|
|
This script intentionally uses only the Python standard library. It parses the
|
|
limited YAML shape used by docs/upstream-sources.yml instead of requiring PyYAML.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urlparse
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
INVENTORY = ROOT / "docs" / "upstream-sources.yml"
|
|
COMPOSE_BLOCK_LANGS = {"", "yaml", "yml", "docker-compose", "compose"}
|
|
NORMALIZED_COMPOSE_NAME = "docker-compose.yml"
|
|
DERIVED_TEMPLATE_MARKER = "nbops: derived-from-upstream"
|
|
DERIVED_TEMPLATE_NAMES = ("compose.yaml.j2", "compose.override.yaml.j2")
|
|
|
|
|
|
def parse_inline_list(value: str) -> list[str]:
|
|
value = value.strip()
|
|
if value == "[]":
|
|
return []
|
|
if not (value.startswith("[") and value.endswith("]")):
|
|
return [value.strip().strip("\"'")]
|
|
inner = value[1:-1].strip()
|
|
if not inner:
|
|
return []
|
|
return [item.strip().strip("\"'") for item in inner.split(",")]
|
|
|
|
|
|
def parse_inventory(path: Path) -> dict[str, dict]:
|
|
sources: dict[str, dict] = {}
|
|
current_role: str | None = None
|
|
current_repo: dict | None = None
|
|
in_sources = False
|
|
in_repositories = False
|
|
|
|
for raw_line in path.read_text().splitlines():
|
|
if not raw_line.strip() or raw_line.lstrip().startswith("#"):
|
|
continue
|
|
|
|
if raw_line == "sources:":
|
|
in_sources = True
|
|
continue
|
|
if not in_sources:
|
|
continue
|
|
|
|
role_match = re.match(r"^ ([A-Za-z0-9_]+):$", raw_line)
|
|
if role_match:
|
|
current_role = role_match.group(1)
|
|
sources[current_role] = {"repositories": []}
|
|
current_repo = None
|
|
in_repositories = False
|
|
continue
|
|
|
|
if current_role is None:
|
|
continue
|
|
|
|
role_key = re.match(r"^ ([A-Za-z0-9_]+):\s*(.*)$", raw_line)
|
|
if role_key:
|
|
key, value = role_key.groups()
|
|
if key == "repositories":
|
|
in_repositories = True
|
|
current_repo = None
|
|
continue
|
|
if key == "notes":
|
|
continue
|
|
sources[current_role][key] = value.strip().strip("\"'")
|
|
continue
|
|
|
|
if in_repositories:
|
|
repo_start = re.match(r"^ - purpose:\s*(.*)$", raw_line)
|
|
if repo_start:
|
|
current_repo = {"purpose": repo_start.group(1).strip().strip("\"'")}
|
|
sources[current_role]["repositories"].append(current_repo)
|
|
continue
|
|
|
|
repo_key = re.match(r"^ ([A-Za-z0-9_]+):\s*(.*)$", raw_line)
|
|
if repo_key and current_repo is not None:
|
|
key, value = repo_key.groups()
|
|
value = value.strip()
|
|
if key in {"readme_paths", "compose_paths", "compose_urls"}:
|
|
current_repo[key] = parse_inline_list(value)
|
|
else:
|
|
current_repo[key] = value.strip("\"'")
|
|
|
|
for role, config in sources.items():
|
|
if "vendor_dir" not in config:
|
|
raise ValueError(f"{role}: missing vendor_dir")
|
|
config.setdefault("local_compose_file", "")
|
|
for repo in config["repositories"]:
|
|
if "purpose" not in repo:
|
|
raise ValueError(f"{role}: repository missing purpose")
|
|
repo.setdefault("readme_paths", [])
|
|
repo.setdefault("compose_paths", [])
|
|
repo.setdefault("compose_urls", [])
|
|
needs_clone = bool(repo["readme_paths"] or repo["compose_paths"])
|
|
has_clone_source = "url" in repo or "ref" in repo
|
|
if needs_clone or has_clone_source:
|
|
for key in ("url", "ref"):
|
|
if key not in repo:
|
|
raise ValueError(f"{role}: repository missing {key}")
|
|
|
|
return sources
|
|
|
|
|
|
def run_capture(command: list[str]) -> tuple[bool, str]:
|
|
completed = subprocess.run(
|
|
command,
|
|
cwd=ROOT,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
check=False,
|
|
)
|
|
return completed.returncode == 0, completed.stdout
|
|
|
|
|
|
def copy_if_present(src_root: Path, relative_path: str, dest_root: Path, apply: bool) -> str:
|
|
src = src_root / relative_path
|
|
if not src.is_file():
|
|
return "missing"
|
|
if not apply:
|
|
return "found"
|
|
|
|
dest = dest_root / relative_path
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(src, dest)
|
|
return "copied"
|
|
|
|
|
|
def copy_compose_if_present(src_root: Path, relative_path: str, dest_root: Path, apply: bool) -> str:
|
|
src = src_root / relative_path
|
|
if not src.is_file():
|
|
return "missing"
|
|
if not looks_like_compose(src.read_text(errors="replace")):
|
|
return "invalid"
|
|
if not apply:
|
|
return "found"
|
|
|
|
dest = dest_root / NORMALIZED_COMPOSE_NAME
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(src, dest)
|
|
return "copied"
|
|
|
|
|
|
def copy_local_file_if_present(relative_path: str, dest_root: Path, apply: bool) -> str:
|
|
src = ROOT / relative_path
|
|
if not src.is_file():
|
|
return "missing"
|
|
if not looks_like_compose(src.read_text(errors="replace")):
|
|
return "invalid"
|
|
if not apply:
|
|
return "found"
|
|
|
|
dest = dest_root / "local" / src.name
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(src, dest)
|
|
return "copied"
|
|
|
|
|
|
def replace_directory(src: Path, dest: Path) -> None:
|
|
if dest.exists():
|
|
shutil.rmtree(dest)
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(src), str(dest))
|
|
|
|
|
|
def preferred_override_exists(snapshot_dir: Path) -> bool:
|
|
return (snapshot_dir / "docker-compose.override.yml").is_file()
|
|
|
|
|
|
def fetch_compose_url(url: str) -> tuple[str, str, str]:
|
|
parsed = urlparse(url)
|
|
if parsed.scheme not in {"http", "https"}:
|
|
return "invalid", "", "unsupported URL scheme"
|
|
|
|
try:
|
|
request = Request(url, headers={"User-Agent": "nbops-upstream-snapshots/1.0"})
|
|
with urlopen(request, timeout=30) as response:
|
|
charset = response.headers.get_content_charset() or "utf-8"
|
|
body = response.read().decode(charset, errors="replace")
|
|
except HTTPError as exc:
|
|
return "missing", "", f"HTTP {exc.code}"
|
|
except (OSError, TimeoutError, URLError) as exc:
|
|
return "missing", "", str(exc)
|
|
|
|
if not looks_like_compose(body):
|
|
return "invalid", body, "response is not Compose-shaped YAML"
|
|
|
|
return "found", body, ""
|
|
|
|
|
|
def extract_fenced_blocks(markdown: str) -> list[tuple[str, str]]:
|
|
pattern = re.compile(r"^```([^\n`]*)\n(.*?)^```", re.MULTILINE | re.DOTALL)
|
|
blocks = []
|
|
for match in pattern.finditer(markdown):
|
|
lang = match.group(1).strip().split(maxsplit=1)[0].lower() if match.group(1).strip() else ""
|
|
blocks.append((lang, match.group(2)))
|
|
return blocks
|
|
|
|
|
|
def looks_like_compose(text: str) -> bool:
|
|
lines = text.splitlines()
|
|
for index, line in enumerate(lines):
|
|
if re.match(r"^services:\s*(#.*)?$", line):
|
|
return any(re.match(r"^ [A-Za-z0-9_.-]+:\s*(#.*)?$", later) for later in lines[index + 1 :])
|
|
return False
|
|
|
|
|
|
def readme_compose_candidates(readme_path: Path) -> list[dict]:
|
|
if not readme_path.is_file():
|
|
return []
|
|
|
|
candidates = []
|
|
for index, (lang, body) in enumerate(extract_fenced_blocks(readme_path.read_text()), start=1):
|
|
if lang not in COMPOSE_BLOCK_LANGS:
|
|
continue
|
|
if not looks_like_compose(body):
|
|
continue
|
|
candidates.append({"index": index, "lang": lang or "untagged", "body": body})
|
|
return candidates
|
|
|
|
|
|
def print_table(rows: list[list[str]]) -> None:
|
|
headers = ["Role", "Status", "Repos", "README", "Compose", "README parse", "Notes"]
|
|
widths = [len(header) for header in headers]
|
|
for row in rows:
|
|
for index, value in enumerate(row):
|
|
widths[index] = max(widths[index], len(str(value)))
|
|
|
|
def fmt(row: list[str]) -> str:
|
|
return " ".join(str(value).ljust(widths[index]) for index, value in enumerate(row))
|
|
|
|
print(fmt(headers))
|
|
print(" ".join("-" * width for width in widths))
|
|
for row in rows:
|
|
print(fmt(row))
|
|
|
|
|
|
def progress_enabled() -> bool:
|
|
return sys.stderr.isatty()
|
|
|
|
|
|
def render_progress(current: int, total: int, label: str) -> None:
|
|
if not progress_enabled() or total <= 0:
|
|
return
|
|
width = 28
|
|
filled = min(width, int(width * current / total))
|
|
bar = "#" * filled + "-" * (width - filled)
|
|
sys.stderr.write(f"\r\033[K[{bar}] {current}/{total} {label}")
|
|
sys.stderr.flush()
|
|
|
|
|
|
def clear_progress() -> None:
|
|
if progress_enabled():
|
|
sys.stderr.write("\r\033[K")
|
|
sys.stderr.flush()
|
|
|
|
|
|
def snapshot_has_content(path: Path) -> bool:
|
|
return path.exists() and any(path.iterdir())
|
|
|
|
|
|
def split_filter_values(values: list[str], csv_value: str) -> set[str]:
|
|
filters: set[str] = set()
|
|
for value in values:
|
|
filters.update(item.strip() for item in value.split(",") if item.strip())
|
|
filters.update(item.strip() for item in csv_value.split(",") if item.strip())
|
|
return filters
|
|
|
|
|
|
def filtered_repositories(config: dict, upstream_filter: set[str]) -> list[dict]:
|
|
repos = config["repositories"]
|
|
if not upstream_filter:
|
|
return repos
|
|
return [repo for repo in repos if repo["purpose"] in upstream_filter]
|
|
|
|
|
|
def clean_derived_templates(role: str) -> list[list[str]]:
|
|
rows: list[list[str]] = []
|
|
template_dir = ROOT / "roles" / role / "templates"
|
|
for name in DERIVED_TEMPLATE_NAMES:
|
|
path = template_dir / name
|
|
if not path.is_file():
|
|
continue
|
|
text = path.read_text(errors="replace")
|
|
if DERIVED_TEMPLATE_MARKER not in text:
|
|
rows.append([role, "template", str(path), "kept: unmarked"])
|
|
continue
|
|
path.unlink()
|
|
rows.append([role, "template", str(path), "removed"])
|
|
return rows
|
|
|
|
|
|
def snapshot_source(vendor_dir: Path, role: str, status: str, notes: list[str], apply: bool) -> None:
|
|
if not apply:
|
|
return
|
|
text = [
|
|
"# Upstream Snapshot",
|
|
"",
|
|
f"Role: `{role}`",
|
|
"Snapshot source inventory: `docs/upstream-sources.yml`",
|
|
f"Snapshot status: `{status}`",
|
|
"",
|
|
"Fetched repositories:",
|
|
*notes,
|
|
"",
|
|
"These files are reference snapshots only. Deployments are generated from role templates.",
|
|
"",
|
|
]
|
|
vendor_dir.mkdir(parents=True, exist_ok=True)
|
|
(vendor_dir / "SOURCE.md").write_text("\n".join(text))
|
|
|
|
|
|
def clean_upstream_directories(selected: dict[str, dict], upstream_filter: set[str]) -> int:
|
|
rows: list[list[str]] = []
|
|
removed = 0
|
|
total = sum(len(filtered_repositories(config, upstream_filter)) for config in selected.values())
|
|
current = 0
|
|
|
|
for role, config in selected.items():
|
|
vendor_dir = ROOT / config["vendor_dir"]
|
|
repos = filtered_repositories(config, upstream_filter)
|
|
for repo in repos:
|
|
current += 1
|
|
purpose_dir = vendor_dir / repo["purpose"]
|
|
render_progress(current, total, f"clean {role}/{repo['purpose']}")
|
|
if purpose_dir.is_dir():
|
|
shutil.rmtree(purpose_dir)
|
|
removed += 1
|
|
result = "removed"
|
|
else:
|
|
result = "missing"
|
|
rows.append([role, repo["purpose"], str(purpose_dir), result])
|
|
rows.extend(clean_derived_templates(role))
|
|
|
|
clear_progress()
|
|
headers = ["Role", "Purpose", "Path", "Result"]
|
|
widths = [len(header) for header in headers]
|
|
for row in rows:
|
|
for index, value in enumerate(row):
|
|
widths[index] = max(widths[index], len(value))
|
|
|
|
def fmt(row: list[str]) -> str:
|
|
return " ".join(value.ljust(widths[index]) for index, value in enumerate(row))
|
|
|
|
print(fmt(headers))
|
|
print(" ".join("-" * width for width in widths))
|
|
for row in rows:
|
|
print(fmt(row))
|
|
print(f"\nRemoved {removed} upstream purpose directories.")
|
|
return 0
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Test or vendor upstream README/Compose snapshots.")
|
|
parser.add_argument("roles", nargs="*", help="Role names to process. Defaults to all roles.")
|
|
parser.add_argument("--apply", action="store_true", help="Copy fetched files into role upstream directories.")
|
|
parser.add_argument("--no-parse-readme", action="store_true", help="Do not extract Compose YAML from README blocks.")
|
|
parser.add_argument(
|
|
"--clean-upstreams",
|
|
action="store_true",
|
|
help="Delete upstream/<purpose> directories and marked derived templates for selected roles, then exit.",
|
|
)
|
|
parser.add_argument(
|
|
"--clean-purposes",
|
|
dest="clean_upstreams",
|
|
action="store_true",
|
|
help=argparse.SUPPRESS,
|
|
)
|
|
parser.add_argument(
|
|
"--upstream",
|
|
action="append",
|
|
default=[],
|
|
help="Limit processing to an upstream purpose name such as container or application. Repeatable.",
|
|
)
|
|
parser.add_argument(
|
|
"--upstreams",
|
|
default="",
|
|
help="Comma-separated upstream purpose names to process.",
|
|
)
|
|
parser.add_argument("--list", action="store_true", help="List roles in the upstream inventory and exit.")
|
|
parser.add_argument("--validate-inventory", action="store_true", help="Parse the inventory and exit without cloning.")
|
|
args = parser.parse_args()
|
|
|
|
sources = parse_inventory(INVENTORY)
|
|
|
|
if args.list:
|
|
print("\n".join(sorted(sources)))
|
|
return 0
|
|
|
|
if args.validate_inventory:
|
|
print(f"inventory ok: {len(sources)} roles")
|
|
return 0
|
|
|
|
selected = {role: config for role, config in sources.items() if not args.roles or role in args.roles}
|
|
if not selected:
|
|
print(f"no matching roles found for: {', '.join(args.roles)}", file=sys.stderr)
|
|
return 2
|
|
upstream_filter = split_filter_values(args.upstream, args.upstreams)
|
|
if upstream_filter:
|
|
known_upstreams = {repo["purpose"] for config in selected.values() for repo in config["repositories"]}
|
|
unknown_upstreams = sorted(upstream_filter - known_upstreams)
|
|
if unknown_upstreams:
|
|
print(f"no matching upstream purpose found for: {', '.join(unknown_upstreams)}", file=sys.stderr)
|
|
return 2
|
|
if args.clean_upstreams:
|
|
return clean_upstream_directories(selected, upstream_filter)
|
|
|
|
rows: list[list[str]] = []
|
|
failures: list[str] = []
|
|
total_steps = sum(len(filtered_repositories(config, upstream_filter)) for config in selected.values())
|
|
if not upstream_filter:
|
|
total_steps += sum(1 for config in selected.values() if config["local_compose_file"])
|
|
current_step = 0
|
|
|
|
for role, config in selected.items():
|
|
vendor_dir = ROOT / config["vendor_dir"]
|
|
if args.apply:
|
|
vendor_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
repo_count = 0
|
|
readme_found = readme_missing = 0
|
|
compose_found = compose_missing = 0
|
|
readme_candidates = readme_ambiguous = 0
|
|
notes: list[str] = []
|
|
snapshot_notes: list[str] = []
|
|
|
|
for repo in filtered_repositories(config, upstream_filter):
|
|
current_step += 1
|
|
render_progress(current_step, total_steps, f"{role}/{repo['purpose']}")
|
|
repo_count += 1
|
|
with tempfile.TemporaryDirectory(prefix=f"nbops-upstream-{role}-") as tmp:
|
|
tmpdir = Path(tmp)
|
|
clone_dir = tmpdir / "repo"
|
|
purpose_dest = vendor_dir / repo["purpose"]
|
|
purpose_snapshot_dir = purpose_dest
|
|
if args.apply:
|
|
purpose_snapshot_dir = tmpdir / "__snapshot__" / repo["purpose"]
|
|
purpose_snapshot_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
ok = False
|
|
if "url" in repo:
|
|
ok, output = run_capture(["git", "clone", "--depth", "1", "--branch", repo["ref"], repo["url"], str(clone_dir)])
|
|
if not ok:
|
|
notes.append(f"{repo['purpose']}: clone failed")
|
|
failures.append(f"{role}/{repo['purpose']}: {output.splitlines()[-1] if output.splitlines() else 'clone failed'}")
|
|
|
|
if ok:
|
|
for path in repo["readme_paths"]:
|
|
result = copy_if_present(clone_dir, path, purpose_snapshot_dir, args.apply)
|
|
if result == "missing":
|
|
readme_missing += 1
|
|
else:
|
|
readme_found += 1
|
|
snapshot_notes.append(f"- {repo['purpose']} README `{path}`: {result}")
|
|
|
|
if args.no_parse_readme:
|
|
continue
|
|
candidates = readme_compose_candidates(clone_dir / path)
|
|
readme_candidates += len(candidates)
|
|
if len(candidates) > 1:
|
|
readme_ambiguous += 1
|
|
if args.apply and len(candidates) == 1:
|
|
dest = purpose_snapshot_dir / NORMALIZED_COMPOSE_NAME
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
dest.write_text(candidates[0]["body"])
|
|
|
|
for path in repo["compose_paths"]:
|
|
result = copy_compose_if_present(clone_dir, path, purpose_snapshot_dir, args.apply)
|
|
if result == "missing":
|
|
compose_missing += 1
|
|
elif result == "invalid":
|
|
compose_missing += 1
|
|
notes.append(f"{repo['purpose']}: Compose `{path}` invalid")
|
|
else:
|
|
compose_found += 1
|
|
snapshot_notes.append(f"- {repo['purpose']} Compose `{path}`: {result}")
|
|
|
|
compose_urls = repo["compose_urls"]
|
|
for index, url in enumerate(compose_urls, start=1):
|
|
if args.apply and (purpose_snapshot_dir / NORMALIZED_COMPOSE_NAME).is_file():
|
|
snapshot_notes.append(
|
|
f"- {repo['purpose']} Compose URL `{url}`: skipped; `{NORMALIZED_COMPOSE_NAME}` already staged"
|
|
)
|
|
continue
|
|
result, body, reason = fetch_compose_url(url)
|
|
if result == "found":
|
|
compose_found += 1
|
|
if args.apply:
|
|
dest = purpose_snapshot_dir / NORMALIZED_COMPOSE_NAME
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
dest.write_text(body)
|
|
result = "copied"
|
|
else:
|
|
compose_missing += 1
|
|
notes.append(f"{repo['purpose']}: compose URL {result}")
|
|
suffix = f" ({reason})" if reason else ""
|
|
snapshot_notes.append(f"- {repo['purpose']} Compose URL `{url}`: {result}{suffix}")
|
|
|
|
if args.apply and snapshot_has_content(purpose_snapshot_dir):
|
|
if repo["purpose"] == "container" and preferred_override_exists(purpose_snapshot_dir):
|
|
notes.append("container override preferred")
|
|
snapshot_notes.append(
|
|
"- container `docker-compose.override.yml`: prefer this when updating `templates/compose.yaml.j2`"
|
|
)
|
|
replace_directory(purpose_snapshot_dir, purpose_dest)
|
|
elif args.apply:
|
|
snapshot_notes.append(f"- {repo['purpose']} snapshot: no files staged; existing purpose directory left unchanged")
|
|
elif repo["purpose"] == "container" and preferred_override_exists(purpose_dest):
|
|
notes.append("container override preferred")
|
|
|
|
if config["local_compose_file"] and not upstream_filter:
|
|
current_step += 1
|
|
render_progress(current_step, total_steps, f"{role}/local")
|
|
result = copy_local_file_if_present(config["local_compose_file"], vendor_dir, args.apply)
|
|
if result in {"missing", "invalid"}:
|
|
compose_missing += 1
|
|
notes.append(f"local compose file {result}")
|
|
else:
|
|
compose_found += 1
|
|
snapshot_notes.append(f"- local Compose file `{config['local_compose_file']}`: {result}")
|
|
|
|
if config.get("status") == "verify":
|
|
notes.append("verify source")
|
|
if readme_ambiguous:
|
|
notes.append("ambiguous README compose")
|
|
|
|
status = "error" if any("clone failed" in note for note in notes) else "partial" if readme_missing or compose_missing else "ok"
|
|
snapshot_notes.append(f"README Compose candidates: {readme_candidates}")
|
|
snapshot_notes.append(f"Ambiguous README files: {readme_ambiguous}")
|
|
snapshot_source(vendor_dir, role, config.get("status", "tracked"), snapshot_notes, args.apply)
|
|
|
|
rows.append(
|
|
[
|
|
role,
|
|
status,
|
|
str(repo_count),
|
|
f"{readme_found}/{readme_found + readme_missing}",
|
|
f"{compose_found}/{compose_found + compose_missing}",
|
|
str(readme_candidates),
|
|
"; ".join(notes),
|
|
]
|
|
)
|
|
|
|
clear_progress()
|
|
print_table(rows)
|
|
|
|
if failures:
|
|
print("\nFailures:", file=sys.stderr)
|
|
for failure in failures:
|
|
print(f"- {failure}", file=sys.stderr)
|
|
return 1
|
|
|
|
if not args.apply:
|
|
print("\nDry run only. Re-run with --apply to copy snapshots into role upstream directories.", file=sys.stderr)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|