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 sha = subprocess.run(["git", "-C", str(dest), "rev-parse", "HEAD"], check=True, capture_output=True, text=True).stdout.strip()
73 print(f"{entry['repo']}@{entry['branch']} -> {dest} ({sha[:10]})")
74 sources.append(source_from_entry(entry, clones, vault_url))
75 return sources
78def source_from_entry(entry: dict, clones: Path, vault_url: str) -> Source:
79 root = clone_dir(clones, entry)
80 return Source(
81 label=entry["label"],
82 dir=root / entry["path"],
83 repo=entry["repo"],
84 url=f"{vault_url.rstrip('/')}/{entry['repo']}",
85 branch=entry["branch"],
86 root=root,
87 description=entry.get("description"),
88 )
91def registry_sources(registry_path: Path, clones: Path, vault_url: str) -> list[Source]:
92 return [source_from_entry(e, clones, vault_url) for e in read_registry(registry_path)]
95def local_source(solutions_dir: Path) -> Source:
96 return Source(label=None, dir=Path(solutions_dir))
99def submission_key(sub: dict) -> str:
100 label = sub.get("_source").label if sub.get("_source") else None
101 return f"{label}/{sub['submission_id']}" if label else sub["submission_id"]
104def load_submissions(sources: list[Source]) -> list[dict]:
105 """Every submission JSON under <source.dir>/<challenge-id>/*.json, with _source, _path, _dir_challenge, _key set."""
106 out = []
107 for src in sources:
108 if not src.dir.exists():
109 print(f"warning: solutions directory {src.dir} does not exist")
110 continue
111 for p in sorted(src.dir.glob("*/*.json")):
112 with open(p) as f:
113 sub = json.load(f)
114 sub["_source"] = src
115 sub["_path"] = p
116 sub["_dir_challenge"] = p.parent.name
117 sub["_key"] = submission_key(sub)
118 out.append(sub)
119 return out
122def public_fields(sub: dict) -> dict:
123 return {k: v for k, v in sub.items() if not k.startswith("_")}