#!/usr/bin/env python3 """ Run `astra ingest git --tag ` for each buildinfo ingested into astra.db. Reads buildinfo artifact IDs from astra.db, extracts source name + version, finds the matching git tag from the local clone, and calls astra ingest git. Usage (inside the astra container): python3 scripts/ingest_repos.py --astra-db /data/astra.db --repos-dir /repos python3 scripts/ingest_repos.py --astra-db /data/astra.db --repos-dir /repos --dry-run """ import argparse import os import re import signal import sqlite3 import subprocess import sys signal.signal(signal.SIGPIPE, signal.SIG_DFL) # artifact:buildinfo:deb/@?arch= BUILDINFO_RE = re.compile(r'^artifact:buildinfo:deb/([^@]+)@([^?]+)') _BINNMU_RE = re.compile(r"\+b\d+$") _BPO_RE = re.compile(r"~bpo\d+\+\d+$") def strip_epoch(version: str) -> str: idx = version.find(":") return version[idx + 1:] if idx != -1 else version def strip_binnmu(version: str) -> str: return _BINNMU_RE.sub("", version) def strip_bpo(version: str) -> str: return _BPO_RE.sub("", version) def upstream_from_debian_version(version: str | None) -> str | None: if not version: return None v = strip_epoch(version) if "-" in v: return v.rsplit("-", 1)[0] return v def version_candidates(*versions: str | None) -> list[str]: candidates: list[str] = [] for version in versions: if not version: continue v = strip_epoch(version.strip()) variants = [v, strip_binnmu(v), strip_bpo(v), strip_bpo(strip_binnmu(v))] for var in variants: if var and var not in candidates: candidates.append(var) return candidates def normalize_tag_text(value: str) -> str: v = (value or "").strip() if v.startswith("refs/tags/"): v = v[len("refs/tags/"):] while v.endswith("^{}") or v.endswith("{}"): if v.endswith("^{}"): v = v[:-3] elif v.endswith("{}"): v = v[:-2] v = v.strip() return v def trailing_separator_variants(value: str) -> list[str]: variants: list[str] = [] v = normalize_tag_text(value) if v: variants.append(v) if "/" in v: last = v.rsplit("/", 1)[-1] if last and last not in variants: variants.append(last) for sep in ("+", "~", "_", "-"): if sep in v: base = v.split(sep, 1)[0] if base and base not in variants: variants.append(base) return variants def separator_prefix_match(a: str, b: str) -> bool: if not a or not b: return False if a == b: return True if a.startswith(b): return len(a) == len(b) or a[len(b)] in "+-._~/" if b.startswith(a): return len(b) == len(a) or b[len(a)] in "+-._~/" return False def find_matching_tag(tags: list[str], candidates: list[str]) -> tuple[str, str] | tuple[None, None]: for candidate in candidates: candidate_variants = trailing_separator_variants(candidate) hits: list[str] = [] for tag in tags: tag_variants = trailing_separator_variants(tag) matched = False for cv in candidate_variants: for tv in tag_variants: if cv == tv or cv in tv or tv in cv or separator_prefix_match(cv, tv): matched = True break if matched: break if matched: hits.append(tag) if hits: hits.sort(key=lambda x: (len(x), x)) return hits[0], candidate return None, None def get_remote_tags(repo_path: str) -> list[str]: try: cp = subprocess.run( ["git", "-C", repo_path, "ls-remote", "--tags"], capture_output=True, text=True, timeout=120, ) except subprocess.TimeoutExpired: print(f" [timeout] ls-remote timed out for {repo_path}", file=sys.stderr) return [] if cp.returncode != 0: return [] tags = [] for line in cp.stdout.splitlines(): parts = line.strip().split() if len(parts) != 2: continue ref = parts[1] if ref.endswith("^{}"): continue tag_name = ref.removeprefix("refs/tags/") if tag_name: tags.append(tag_name) return tags def get_remote_url(clone_path: str) -> str | None: result = subprocess.run( ["git", "-C", clone_path, "remote", "get-url", "origin"], capture_output=True, text=True, timeout=10, ) if result.returncode != 0: return None url = result.stdout.strip() if not url: return None # Strip embedded credentials (https://user:token@host -> https://host) from urllib.parse import urlparse, urlunparse parsed = urlparse(url) if parsed.username or parsed.password: url = urlunparse(parsed._replace(netloc=parsed.hostname + (f":{parsed.port}" if parsed.port else ""))) return url or None def query_buildinfos(astra_db: str) -> list[tuple[str, str]]: """Return distinct (source_name, version) from ingested buildinfo artifacts.""" con = sqlite3.connect(astra_db) cur = con.cursor() cur.execute("SELECT DISTINCT astra_id FROM artifacts WHERE kind = 'buildinfo'") rows = cur.fetchall() con.close() results = [] for (astra_id,) in rows: m = BUILDINFO_RE.match(astra_id) if m: results.append((m.group(1), m.group(2))) return results def main(): ap = argparse.ArgumentParser() ap.add_argument("--astra-db", default="/data/astra.db") ap.add_argument("--repos-dir", default="/repos") ap.add_argument("--dry-run", action="store_true") args = ap.parse_args() buildinfos = query_buildinfos(args.astra_db) print(f"[astra] {len(buildinfos)} buildinfo artifacts", file=sys.stderr) ok = fail = no_clone = no_tag = 0 for source_name, version in buildinfos: clone_path = os.path.join(args.repos_dir, source_name) if not os.path.isdir(clone_path): no_clone += 1 continue url = get_remote_url(clone_path) if url is None: no_clone += 1 continue tags = get_remote_tags(clone_path) candidates = version_candidates(version, upstream_from_debian_version(version)) tag, _ = find_matching_tag(tags, candidates) if tag is None: print(f" [no-tag] {source_name} {version}", file=sys.stderr) no_tag += 1 continue cmd = ["astra", "ingest", "git", "--tag", tag, url] if args.dry_run: print(" ".join(cmd)) continue rc = subprocess.run(cmd).returncode if rc == 0: ok += 1 else: print(f" [fail] {source_name} {version} tag={tag}", file=sys.stderr) fail += 1 if not args.dry_run: print(f"Done: {ok} ok, {fail} failed, {no_clone} no clone, {no_tag} no tag matched", file=sys.stderr) if __name__ == "__main__": main()