Spaces:
Sleeping
Sleeping
File size: 3,222 Bytes
df898a6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | """
Run history — saves and loads completed pipeline runs as JSON.
Each run is stored at data/output/run_history/run_YYYY-MM-DD_HH-MM-SS.json.
Keeps the full job list so results can be restored in the UI without rerunning.
"""
import json
import logging
from pathlib import Path
from datetime import datetime
log = logging.getLogger("run_history")
HISTORY_DIR = Path("data/output/run_history")
def save_run(jobs: list, run_meta: dict) -> str:
"""
Save a completed run to history.
run_meta should include: run_id, excel_path, platforms, roles.
Returns the history file path, or '' on failure.
"""
HISTORY_DIR.mkdir(parents=True, exist_ok=True)
run_id = run_meta.get("run_id") or datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
high = sum(1 for j in jobs if j.get("relevance_score", 0) >= 8)
med = sum(1 for j in jobs if 6 <= j.get("relevance_score", 0) <= 7)
resumes = sum(1 for j in jobs if j.get("resume_path"))
pdfs = sum(1 for j in jobs if j.get("resume_pdf_path"))
ats_before = [j["ats_score_before"] for j in jobs
if j.get("ats_score_before") not in (None, "", 0)]
ats_after = [j["ats_score_after"] for j in jobs
if j.get("ats_score_after") not in (None, "", 0)]
record = {
"run_id": run_id,
"date": datetime.now().strftime("%Y-%m-%d %H:%M"),
"total_jobs": len(jobs),
"high_priority": high,
"med_priority": med,
"resumes": resumes,
"pdfs": pdfs,
"avg_ats_before": int(sum(ats_before) / len(ats_before)) if ats_before else 0,
"avg_ats_after": int(sum(ats_after) / len(ats_after)) if ats_after else 0,
"platforms": list(set(j.get("platform", "") for j in jobs if j.get("platform"))),
"roles": run_meta.get("roles", []),
"excel_path": run_meta.get("excel_path", ""),
"jobs": jobs,
}
path = HISTORY_DIR / f"run_{run_id}.json"
try:
with open(path, "w", encoding="utf-8") as f:
json.dump(record, f, ensure_ascii=False, default=str)
log.info(f"Run saved to history: {path}")
return str(path)
except Exception as e:
log.error(f"Failed to save run history: {e}")
return ""
def list_runs() -> list:
"""
List all saved runs, newest first.
Returns list of summary dicts (no jobs key — kept light for UI listing).
"""
if not HISTORY_DIR.exists():
return []
runs = []
for p in sorted(HISTORY_DIR.glob("run_*.json"), reverse=True):
try:
with open(p, "r", encoding="utf-8") as f:
data = json.load(f)
summary = {k: v for k, v in data.items() if k != "jobs"}
summary["_path"] = str(p)
runs.append(summary)
except Exception:
pass
return runs
def load_run(path: str) -> dict:
"""Load a full run (including jobs list) from a history JSON file."""
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
log.error(f"Failed to load run {path}: {e}")
return {}
|