3b54f996ca
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RCcxND1mMWu6Lt2c2PxexN
107 lines
4.1 KiB
Python
107 lines
4.1 KiB
Python
"""Full-catalog batch runner — SKELETON (Fable 5, 2026-07-07).
|
|
|
|
Sonnet: implement the TODO functions, do not redesign the flow. The staging
|
|
logic, resume mechanics and budget enforcement are final; see
|
|
docs/batch-architecture.md for the design and data contracts.
|
|
|
|
Stages (per skillfactor_finalize.md / auftrag_sonnet.md):
|
|
1. packages: deterministic generator for ALL ESCO occupations (no LLM)
|
|
2. evidence: per occupation fetch (JSearch, budget-capped) -> extract
|
|
(Ollama, extract_local.py) -> aggregate (market.md)
|
|
3. depth: tier 1 = top-200 by evidence volume, tier 3 = night batch
|
|
|
|
Every step is idempotent and resumable: progress.json + file existence
|
|
decide what still needs work; an abort at any point loses at most the
|
|
current occupation's in-flight step.
|
|
|
|
Usage:
|
|
python pipeline/batch_run.py --stage packages [--limit N]
|
|
python pipeline/batch_run.py --stage evidence [--slugs a,b] [--limit N]
|
|
python pipeline/batch_run.py --stage tier # (re)compute tiers only
|
|
"""
|
|
import argparse
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
import progress
|
|
from db import connect
|
|
|
|
BASE = os.path.join(os.path.dirname(__file__), "..")
|
|
ADS_PER_OCC_TARGET = 60 # ~6 requests/occupation (1 req ~ 10 ads)
|
|
COUNTRIES = ("us", "gb")
|
|
REQ_RATE_SLEEP = 0.5 # ~2 req/s max (finalize order)
|
|
TOP_TIER_SIZE = 200
|
|
|
|
|
|
def esco_occupations(cur):
|
|
"""All ESCO occupations with slug + altLabels for query building."""
|
|
# TODO(sonnet): return [(slug, preferred_label, [alt_labels...]), ...]
|
|
# from esco_occupation; slug generation must match p2_generate.py
|
|
raise NotImplementedError
|
|
|
|
|
|
def stage_packages(state, limit=0):
|
|
"""Run the existing generator for every occupation without a package.
|
|
TODO(sonnet): refactor p2_generate.py so it can be called per-occupation
|
|
(it currently generates a fixed balanced set of 100). Reuse, don't fork.
|
|
Gitea pushes go through p4_publish.py, throttled (sleep >= 1s/repo,
|
|
resumable by checking repo existence via API before create)."""
|
|
raise NotImplementedError
|
|
|
|
|
|
def stage_evidence(state, slugs=None, limit=0):
|
|
"""Per occupation: fetch (cached) -> extract (Ollama) -> aggregate.
|
|
|
|
fetch: generalize p3a (queries = preferredLabel + top-3 altLabels x
|
|
COUNTRIES, cache data/raw/jobs/<slug>/, budget via
|
|
progress.spend_request BEFORE each HTTP call).
|
|
< 20 usable ads -> synonym fallback, then Adzuna (free), then
|
|
mark evidence="failed", error="thin market" and move on.
|
|
extract: subprocess extract_local.py --ads data/raw/jobs/<slug>_ads.json
|
|
--out data/evidence/<slug>.jsonl (already implemented + tested)
|
|
store: extend p3b: incremental MERGE keyed by (occupation_slug, job_id)
|
|
-- needs ALTER TABLE evidence_job ADD occupation_slug (see
|
|
docs/batch-architecture.md, backfill 'recruitment-consultant')
|
|
aggregate: generalize p3c per occupation (market.md + market-evidence
|
|
sections); QA: run qa_sample.py per batch, stop the run if the
|
|
harness exits 1 twice in a row (systemic problem, not noise).
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
|
|
def stage_tier(state):
|
|
"""Rank occupations for depth staging: tier 1 = TOP_TIER_SIZE by
|
|
(ads collected, crosswalk strength, marketplace demand), rest tier 3.
|
|
Pure bookkeeping in progress.json — no API calls."""
|
|
raise NotImplementedError
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--stage", required=True,
|
|
choices=["packages", "evidence", "tier"])
|
|
ap.add_argument("--slugs", default="")
|
|
ap.add_argument("--limit", type=int, default=0)
|
|
args = ap.parse_args()
|
|
|
|
state = progress.load()
|
|
try:
|
|
if args.stage == "packages":
|
|
stage_packages(state, args.limit)
|
|
elif args.stage == "evidence":
|
|
slugs = [s for s in args.slugs.split(",") if s] or None
|
|
stage_evidence(state, slugs, args.limit)
|
|
else:
|
|
stage_tier(state)
|
|
except progress.BudgetExhausted as exc:
|
|
print(f"STOP: {exc}")
|
|
sys.exit(2)
|
|
finally:
|
|
progress.save(state)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|