feat(seed): expand corpus to 1178 papers + download-only/unseed/granular reset
PRIMARY — corpus breadth (505->1178 papers, 18->60 specs, all URLs HEAD-verified): - AQA (enumerated): Maths, English Lang/Lit, Geography, Computer Science, Business, Psychology, MFL (French/Spanish/German), GCSE + A-level, on top of round-1 sciences. - Edexcel + OCR (confirmed direct URLs via research): Maths, English, Geography, History, Business, Computer Science, GCSE + A-level. - generate_corpus_manifest.py: _subj/_mfl AQA builders, Edexcel/OCR spec+URL tables, derived exam_code (_mk_exam_code) matching the locked convention, concurrent re-verify. Verified on dev .94: eb_specifications=60, eb_exams=1178, QP=469, doc_type all 'pdf', seed idempotent (uploaded=673 new, skipped=505), failed=0. SECONDARY: - --download-only + persistent bucket-shaped local store (manifests/_corpus_store/, gitignored): download-once, seed-many, offline-repeatable; --store-dir/--no-store. (_store_path/_item_bytes/ download_corpus). Verified: store populated, seed reads offline (download_cached). - --unseed [--board/--spec]: inverse loader — storage objects (Storage API; protect_delete blocks raw SQL), first-sweep seed templates, eb_exams, eb_specifications. Verified reversible on .94. - Granular admin reset: POST /admin/reset?scope=all|exam-corpus|timetable. reset_environment.reset(scope) adds EXAM_CORPUS_TABLES (10) + cc.examboards storage cleanup + TIMETABLE_TABLES (13); 'all' now also clears the exam subsystem the legacy reset missed. No schema migration required. Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
CC Worker
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
5750413f43
commit
cdc105ae54
@@ -60,6 +60,13 @@ DOC_ROLES = {"QP", "MS", "INSERT", "ER", "SPECIMEN", "GRADE_BOUNDARIES", "DATA_S
|
||||
TIERS = {"H", "F", None}
|
||||
# Default working dir for cached url: downloads (override with --cache-dir / EXAM_CORPUS_CACHE).
|
||||
DEFAULT_CACHE_DIR = os.getenv("EXAM_CORPUS_CACHE", "/tmp/exam-corpus-cache")
|
||||
# Persistent, mountable local store laid out exactly like the bucket (download once, seed many,
|
||||
# offline-repeatable). Override with --store-dir / EXAM_CORPUS_STORE. Distinct from --cache-dir,
|
||||
# which is a throwaway url hash-cache.
|
||||
DEFAULT_STORE_DIR = os.getenv(
|
||||
"EXAM_CORPUS_STORE",
|
||||
os.path.join(os.path.dirname(os.path.abspath(__file__)), "manifests", "_corpus_store"),
|
||||
)
|
||||
|
||||
|
||||
# ─────────────────────────────── canonical storage paths ───────────────────────────────
|
||||
@@ -95,12 +102,24 @@ class LoadReport:
|
||||
user_copies: int = 0
|
||||
swept: int = 0
|
||||
sweep_failed: int = 0
|
||||
downloaded: int = 0
|
||||
download_cached: int = 0
|
||||
unseed_objects: int = 0
|
||||
unseed_exams: int = 0
|
||||
unseed_specs: int = 0
|
||||
unseed_templates: int = 0
|
||||
errors: List[str] = field(default_factory=list)
|
||||
|
||||
def as_dict(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"specs_upserted": self.specs_upserted,
|
||||
"papers_upserted": self.papers_upserted,
|
||||
"downloaded": self.downloaded,
|
||||
"download_cached": self.download_cached,
|
||||
"unseed_objects": self.unseed_objects,
|
||||
"unseed_exams": self.unseed_exams,
|
||||
"unseed_specs": self.unseed_specs,
|
||||
"unseed_templates": self.unseed_templates,
|
||||
"files_uploaded": self.files_uploaded,
|
||||
"files_skipped": self.files_skipped,
|
||||
"files_failed": self.files_failed,
|
||||
@@ -181,6 +200,70 @@ def _resolve_source_bytes(source: str, *, cache_dir: str) -> bytes:
|
||||
return fh.read()
|
||||
|
||||
|
||||
# ─────────────────────── persistent local store (download-once, seed-many) ───────────────────────
|
||||
def _store_path(store_dir: str, storage_loc: str) -> str:
|
||||
"""Local path mirroring the bucket layout (so the store is directly mountable as the corpus):
|
||||
storage_loc 'cc.examboards/aqa/physics/8463/1h/2022-jun/qp.pdf'
|
||||
-> {store_dir}/aqa/physics/8463/1h/2022-jun/qp.pdf
|
||||
"""
|
||||
_, _, path = storage_loc.partition("/")
|
||||
return os.path.join(store_dir, path)
|
||||
|
||||
def _item_bytes(source: str, storage_loc: str, *, store_dir: Optional[str], cache_dir: str,
|
||||
populate: bool = True, rep: Optional[LoadReport] = None) -> bytes:
|
||||
"""Resolve bytes for an item, preferring the persistent local store when present.
|
||||
|
||||
If store_dir holds the file → read it (offline). Otherwise resolve the source (local|url:) and,
|
||||
when populate=True, write it into the store at its canonical path for future offline runs.
|
||||
"""
|
||||
if store_dir:
|
||||
sp = _store_path(store_dir, storage_loc)
|
||||
if os.path.exists(sp) and os.path.getsize(sp) > 0:
|
||||
if rep is not None:
|
||||
rep.download_cached += 1
|
||||
with open(sp, "rb") as fh:
|
||||
return fh.read()
|
||||
data = _resolve_source_bytes(source, cache_dir=cache_dir)
|
||||
if store_dir and populate:
|
||||
sp = _store_path(store_dir, storage_loc)
|
||||
os.makedirs(os.path.dirname(sp), exist_ok=True)
|
||||
tmp = sp + ".part"
|
||||
with open(tmp, "wb") as fh:
|
||||
fh.write(data)
|
||||
os.replace(tmp, sp)
|
||||
if rep is not None:
|
||||
rep.downloaded += 1
|
||||
return data
|
||||
|
||||
def download_corpus(m: Dict[str, Any], *, store_dir: str, board_filter: Optional[str],
|
||||
spec_filter: Optional[str], cache_dir: str, rep: LoadReport) -> None:
|
||||
"""--download-only: populate the persistent local store from the manifest. No DB/bucket writes.
|
||||
A later run with the same --store-dir (e.g. mounted into the container) seeds offline from it."""
|
||||
for board in m.get("boards", []):
|
||||
if board_filter and board.get("exam_board_code") != board_filter:
|
||||
continue
|
||||
for spec in board.get("specifications", []):
|
||||
if spec_filter and spec.get("spec_code") != spec_filter:
|
||||
continue
|
||||
sf = spec.get("spec_file")
|
||||
if sf and sf.get("source"):
|
||||
sloc = spec_storage_loc(board["exam_board_code"], spec.get("subject_code", ""),
|
||||
spec.get("award_code", ""), spec.get("spec_ver", ""))
|
||||
try:
|
||||
_item_bytes(sf["source"], sloc, store_dir=store_dir, cache_dir=cache_dir, rep=rep)
|
||||
except Exception as exc:
|
||||
rep.errors.append(f"download spec {spec.get('spec_code')}: {exc}")
|
||||
for p in spec.get("papers", []):
|
||||
ploc = paper_storage_loc(board["exam_board_code"], spec.get("subject_code", ""),
|
||||
spec.get("award_code", ""), p["paper_code"], p["session"], p["doc_type"])
|
||||
try:
|
||||
_item_bytes(p["file"]["source"], ploc, store_dir=store_dir, cache_dir=cache_dir, rep=rep)
|
||||
except Exception as exc:
|
||||
rep.errors.append(f"download {p.get('exam_code')}: {exc}")
|
||||
logger.info(f"download-only done: downloaded={rep.downloaded} already_in_store={rep.download_cached} "
|
||||
f"errors={len(rep.errors)} store={store_dir}")
|
||||
|
||||
|
||||
# ─────────────────────────────── storage upload (skip-if-exists + sha256) ───────────────────────────────
|
||||
def _split_loc(storage_loc: str) -> Tuple[str, str]:
|
||||
bucket, _, path = storage_loc.partition("/")
|
||||
@@ -491,10 +574,88 @@ def first_sweep(client: SupabaseServiceRoleClient, storage: StorageAdmin,
|
||||
rep.swept += 1
|
||||
|
||||
|
||||
# ─────────────────────────────── unseed (inverse of the loader) ───────────────────────────────
|
||||
def _chunks(seq: List[Any], n: int = 100):
|
||||
for i in range(0, len(seq), n):
|
||||
yield seq[i:i + n]
|
||||
|
||||
def unseed(client: SupabaseServiceRoleClient, storage: StorageAdmin, *,
|
||||
board_filter: Optional[str], spec_filter: Optional[str],
|
||||
drop_specs: bool = True, drop_seed_templates: bool = True, rep: LoadReport) -> None:
|
||||
"""Inverse of the loader: remove the seeded public corpus, scoped by --board/--spec (or all).
|
||||
|
||||
Deletes (in FK-safe order): cc.examboards storage objects (via the Storage API, since the
|
||||
protect_delete trigger blocks direct SQL deletes), first-sweep exam_templates created by the
|
||||
seed (title '... (auto-map seed)', cascades children), eb_exams rows, then eb_specifications.
|
||||
"""
|
||||
sb = client.supabase
|
||||
q = sb.table("eb_specifications").select("spec_code, storage_loc, exam_board_code")
|
||||
if board_filter:
|
||||
q = q.eq("exam_board_code", board_filter)
|
||||
if spec_filter:
|
||||
q = q.eq("spec_code", spec_filter)
|
||||
specs = getattr(q.execute(), "data", None) or []
|
||||
spec_codes = [s["spec_code"] for s in specs]
|
||||
if not spec_codes:
|
||||
logger.info("[unseed] no matching specifications; nothing to do")
|
||||
return
|
||||
|
||||
exams: List[Dict[str, Any]] = []
|
||||
for chunk in _chunks(spec_codes):
|
||||
res = sb.table("eb_exams").select("id, exam_code, storage_loc").in_("spec_code", chunk).execute()
|
||||
exams.extend(getattr(res, "data", None) or [])
|
||||
|
||||
# 1) Storage objects (Storage API; batch-remove per bucket). Specs may carry a spec PDF too.
|
||||
by_bucket: Dict[str, List[str]] = {}
|
||||
for row in exams + specs:
|
||||
loc = row.get("storage_loc")
|
||||
if not loc or "/" not in loc:
|
||||
continue
|
||||
bkt, _, path = loc.partition("/")
|
||||
by_bucket.setdefault(bkt, []).append(path)
|
||||
for bkt, paths in by_bucket.items():
|
||||
for chunk in _chunks(paths, 100):
|
||||
try:
|
||||
storage.client.supabase.storage.from_(bkt).remove(chunk)
|
||||
rep.unseed_objects += len(chunk)
|
||||
except Exception as exc:
|
||||
logger.warning(f"[unseed] storage remove failed ({bkt}, {len(chunk)} objs): {exc}")
|
||||
|
||||
# 2) First-sweep templates created by the seed (cascades questions/regions/boundaries/layout).
|
||||
if drop_seed_templates and exams:
|
||||
exam_codes = [e["exam_code"] for e in exams if e.get("exam_code")]
|
||||
for chunk in _chunks(exam_codes, 100):
|
||||
try:
|
||||
res = sb.table("exam_templates").delete(count="exact") \
|
||||
.in_("exam_code", chunk).like("title", "%(auto-map seed)%").execute()
|
||||
rep.unseed_templates += getattr(res, "count", None) or len(getattr(res, "data", []) or [])
|
||||
except Exception as exc:
|
||||
logger.warning(f"[unseed] template delete failed: {exc}")
|
||||
|
||||
# 3) Catalogue rows: eb_exams (by id), then eb_specifications (by spec_code).
|
||||
exam_ids = [e["id"] for e in exams]
|
||||
for chunk in _chunks(exam_ids, 100):
|
||||
try:
|
||||
sb.table("eb_exams").delete().in_("id", chunk).execute()
|
||||
rep.unseed_exams += len(chunk)
|
||||
except Exception as exc:
|
||||
logger.warning(f"[unseed] eb_exams delete failed: {exc}")
|
||||
if drop_specs:
|
||||
for chunk in _chunks(spec_codes, 100):
|
||||
try:
|
||||
sb.table("eb_specifications").delete().in_("spec_code", chunk).execute()
|
||||
rep.unseed_specs += len(chunk)
|
||||
except Exception as exc:
|
||||
logger.warning(f"[unseed] eb_specifications delete failed: {exc}")
|
||||
|
||||
logger.info(f"unseed done: storage_objects={rep.unseed_objects} templates={rep.unseed_templates} "
|
||||
f"exams={rep.unseed_exams} specs={rep.unseed_specs}")
|
||||
|
||||
|
||||
# ─────────────────────────────── orchestration ───────────────────────────────
|
||||
def load(manifest_path: str, *, dry_run: bool, force: bool, board_filter: Optional[str],
|
||||
spec_filter: Optional[str], user_subset: bool, do_first_sweep: bool,
|
||||
cache_dir: str = DEFAULT_CACHE_DIR) -> LoadReport:
|
||||
cache_dir: str = DEFAULT_CACHE_DIR, store_dir: Optional[str] = None) -> LoadReport:
|
||||
with open(manifest_path) as f:
|
||||
m = yaml.safe_load(f)
|
||||
rep = LoadReport()
|
||||
@@ -526,7 +687,9 @@ def load(manifest_path: str, *, dry_run: bool, force: bool, board_filter: Option
|
||||
spec.get("award_code", ""), spec.get("spec_ver", ""))
|
||||
if not dry_run:
|
||||
try:
|
||||
spec_sha = upload_file(storage, sloc, _resolve_source_bytes(sf["source"], cache_dir=cache_dir),
|
||||
spec_sha = upload_file(storage, sloc,
|
||||
_item_bytes(sf["source"], sloc, store_dir=store_dir,
|
||||
cache_dir=cache_dir, rep=rep),
|
||||
force=force, rep=rep)
|
||||
except Exception as exc:
|
||||
logger.error(f"[spec-file] {spec.get('spec_code')}: {exc}")
|
||||
@@ -543,7 +706,9 @@ def load(manifest_path: str, *, dry_run: bool, force: bool, board_filter: Option
|
||||
continue
|
||||
psha = None
|
||||
try:
|
||||
psha = upload_file(storage, ploc, _resolve_source_bytes(p["file"]["source"], cache_dir=cache_dir),
|
||||
psha = upload_file(storage, ploc,
|
||||
_item_bytes(p["file"]["source"], ploc, store_dir=store_dir,
|
||||
cache_dir=cache_dir, rep=rep),
|
||||
force=force, rep=rep)
|
||||
except Exception as exc:
|
||||
logger.error(f"[paper-file] {p.get('exam_code')}: {exc}")
|
||||
@@ -563,19 +728,49 @@ def load(manifest_path: str, *, dry_run: bool, force: bool, board_filter: Option
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser(description="Seed the public exam-paper corpus from a manifest.")
|
||||
ap.add_argument("--manifest", required=True)
|
||||
ap = argparse.ArgumentParser(description="Seed (or unseed) the public exam-paper corpus from a manifest.")
|
||||
ap.add_argument("--manifest", help="corpus manifest (required except for --unseed)")
|
||||
ap.add_argument("--dry-run", action="store_true", help="validate + report, no writes")
|
||||
ap.add_argument("--force", action="store_true", help="re-upload/overwrite existing storage objects")
|
||||
ap.add_argument("--board", default=None, help="only this exam_board_code")
|
||||
ap.add_argument("--spec", default=None, help="only this spec_code")
|
||||
ap.add_argument("--user-subset", action="store_true", help="also seed a user-side test subset")
|
||||
ap.add_argument("--first-sweep", action="store_true", help="run docling/auto-map first pass on seeded papers")
|
||||
ap.add_argument("--cache-dir", default=DEFAULT_CACHE_DIR, help="cache dir for url: downloads")
|
||||
ap.add_argument("--cache-dir", default=DEFAULT_CACHE_DIR, help="throwaway url-hash cache dir")
|
||||
ap.add_argument("--store-dir", default=DEFAULT_STORE_DIR,
|
||||
help="persistent, bucket-shaped local store (download-once, seed-many)")
|
||||
ap.add_argument("--no-store", action="store_true",
|
||||
help="ignore the local store; always fetch from source (don't read/populate the store)")
|
||||
ap.add_argument("--download-only", action="store_true",
|
||||
help="populate the local store from the manifest; no DB/bucket writes")
|
||||
ap.add_argument("--unseed", action="store_true",
|
||||
help="INVERSE: remove seeded eb_*/storage/first-sweep templates (scoped by --board/--spec)")
|
||||
a = ap.parse_args()
|
||||
rep = load(a.manifest, dry_run=a.dry_run, force=a.force, board_filter=a.board, spec_filter=a.spec,
|
||||
user_subset=a.user_subset, do_first_sweep=a.first_sweep, cache_dir=a.cache_dir)
|
||||
store_dir = None if a.no_store else a.store_dir
|
||||
import json
|
||||
|
||||
if a.unseed:
|
||||
rep = LoadReport()
|
||||
unseed(SupabaseServiceRoleClient(), StorageAdmin(),
|
||||
board_filter=a.board, spec_filter=a.spec, rep=rep)
|
||||
print(json.dumps(rep.as_dict(), indent=2))
|
||||
return
|
||||
|
||||
if not a.manifest:
|
||||
ap.error("--manifest is required unless --unseed is given")
|
||||
|
||||
if a.download_only:
|
||||
with open(a.manifest) as f:
|
||||
m = yaml.safe_load(f)
|
||||
rep = LoadReport()
|
||||
download_corpus(m, store_dir=(a.store_dir), board_filter=a.board, spec_filter=a.spec,
|
||||
cache_dir=a.cache_dir, rep=rep)
|
||||
print(json.dumps(rep.as_dict(), indent=2))
|
||||
return
|
||||
|
||||
rep = load(a.manifest, dry_run=a.dry_run, force=a.force, board_filter=a.board, spec_filter=a.spec,
|
||||
user_subset=a.user_subset, do_first_sweep=a.first_sweep, cache_dir=a.cache_dir,
|
||||
store_dir=store_dir)
|
||||
print(json.dumps(rep.as_dict(), indent=2))
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user