dc14526Shared challenge tools: schema, validation, metrics, reference selection, comparison site; moved out of the string repositoryJeremy Magland 1"""Loading challenge definitions, solution sources, and submission files."""
2from __future__ import annotations
4import json
5import subprocess
6from dataclasses import dataclass
7from pathlib import Path
11class Source:
12 """One place solutions come from: a registered repository clone, or a local directory."""
14 label: str | None # short name used to namespace submission ids; None for a single local directory
15 dir: Path # directory holding <challenge-id>/<submission-id>.json
16 repo: str | None = None # "<collection>/<repo>"
17 url: str | None = None # web URL of the repository
18 branch: str | None = None
19 root: Path | None = None # clone root, to express files as repository-relative paths
20 description: str | None = None
22 def blob_url(self, path: Path) -> str | None:
23 if self.url is None or self.root is None:
24 return None
25 try:
26 rel = Path(path).resolve().relative_to(self.root.resolve())
27 except ValueError:
28 return None
29 return f"{self.url}/blob/{self.branch or 'main'}/{rel.as_posix()}"
32def load_challenges(challenges_dir: Path) -> dict[str, dict]:
33 out = {}
34 for p in sorted(Path(challenges_dir).glob("*/challenge.json")):
35 with open(p) as f:
36 ch = json.load(f)
37 ch["_dir"] = p.parent
38 out[ch["challenge_id"]] = ch
39 return out
42def read_registry(registry_path: Path) -> list[dict]:
43 with open(registry_path) as f:
44 reg = json.load(f)
45 repos = reg["repos"]
46 for r in repos:
47 r.setdefault("branch", "main")
48 r.setdefault("path", "solutions")
49 r["label"] = r.get("label") or r["repo"].split("/")[-1]
50 labels = [r["label"] for r in repos]
51 if len(set(labels)) != len(labels):
52 raise ValueError("registry labels are not unique; give colliding entries an explicit 'label'")
53 return repos
56def clone_dir(clones: Path, entry: dict) -> Path:
57 return Path(clones) / entry["label"]
60def fetch_registry(registry_path: Path, clones: Path, vault_url: str) -> list[Source]:
61 """Clone or update every registered repository (shallow, one branch). Anonymous, so public repositories only."""
62 sources = []
63 clones.mkdir(parents=True, exist_ok=True)
64 for entry in read_registry(registry_path):
65 dest = clone_dir(clones, entry)
66 remote = f"{vault_url.rstrip('/')}/{entry['repo']}"
67 if (dest / ".git").exists():
68 subprocess.run(["git", "-C", str(dest), "fetch", "--depth", "1", "origin", entry["branch"]], check=True)
69 subprocess.run(["git", "-C", str(dest), "checkout", "-q", "--force", "FETCH_HEAD"], check=True)
70 else:
71 subprocess.run(["git", "clone", "-q", "--depth", "1", "--branch", entry["branch"], remote, str(dest)], check=True)
72 # bring in LFS objects when git-lfs is available; otherwise pointer files stay and are tolerated downstream
73 if subprocess.run(["git", "lfs", "version"], capture_output=True).returncode == 0:
74 lfs = subprocess.run(["git", "-C", str(dest), "lfs", "pull"], capture_output=True, text=True)
75 if lfs.returncode != 0:
76 print(f"warning: git lfs pull failed for {entry['repo']}: {lfs.stderr.strip()[:200]}")
77 sha = subprocess.run(["git", "-C", str(dest), "rev-parse", "HEAD"], check=True, capture_output=True, text=True).stdout.strip()
78 print(f"{entry['repo']}@{entry['branch']} -> {dest} ({sha[:10]})")
79 sources.append(source_from_entry(entry, clones, vault_url))
80 return sources
83def source_from_entry(entry: dict, clones: Path, vault_url: str) -> Source:
84 root = clone_dir(clones, entry)
85 return Source(
86 label=entry["label"],
87 dir=root / entry["path"],
88 repo=entry["repo"],
89 url=f"{vault_url.rstrip('/')}/{entry['repo']}",
90 branch=entry["branch"],
91 root=root,
92 description=entry.get("description"),
93 )
96def registry_sources(registry_path: Path, clones: Path, vault_url: str) -> list[Source]:
97 return [source_from_entry(e, clones, vault_url) for e in read_registry(registry_path)]
100def local_source(solutions_dir: Path) -> Source:
101 return Source(label=None, dir=Path(solutions_dir))
104def submission_key(sub: dict) -> str:
105 label = sub.get("_source").label if sub.get("_source") else None
106 return f"{label}/{sub['submission_id']}" if label else sub["submission_id"]
109def load_submissions(sources: list[Source]) -> list[dict]:
110 """Every submission JSON under <source.dir>/<challenge-id>/*.json, with _source, _path, _dir_challenge, _key set."""
111 out = []
112 for src in sources:
113 if not src.dir.exists():
114 print(f"warning: solutions directory {src.dir} does not exist")
115 continue
116 for p in sorted(src.dir.glob("*/*.json")):
117 with open(p) as f:
118 sub = json.load(f)
119 sub["_source"] = src
120 sub["_path"] = p
121 sub["_dir_challenge"] = p.parent.name
122 sub["_key"] = submission_key(sub)
123 out.append(sub)
124 return out
127def public_fields(sub: dict) -> dict:
128 return {k: v for k, v in sub.items() if not k.startswith("_")}