"""Phase 6 — Build the static site. `uv run python -m src.build_site` does end-to-end: 1. Runs the analysis pipeline (clean → features → insights → outliers → predictors) 2. Generates syntax-highlighted HTML pages in site/code/ for every src/*.py 3. Generates site/code/index.html (file-tree sidebar) 4. Copies / ensures static assets exist The narrative HTML pages (index, approach, insights, outliers, predictors, agent-review) are hand-written and live alongside site/assets/js/*.js for each page's chart code. They consume the JSON files in site/data/ which are produced upstream. """ import hashlib import html import json import re import shutil import subprocess import sys from pathlib import Path from . import run_pipeline, analysis_insights, analysis_outliers, analysis_predictors from .config import DATA_PROC, ROOT, SITE_DIR, SITE_CODE, SITE_DATA SRC_DIR = ROOT / "src" AUDIT_ARTIFACT_NAMES = ( "outlier_flags_audit.parquet", "outlier_cohort_comparison_audit.parquet", "outlier_iqr_flags_audit.parquet", "outlier_zscore_flags_audit.parquet", "predictors_validation_inpatient.parquet", "predictors_validation_outpatient.parquet", "predictors_validation_inpatient.json", "predictors_validation_outpatient.json", ) # Files to expose in the code browser (skip __init__.py — empty) CODE_FILES = sorted([p for p in SRC_DIR.glob("*.py") if p.name != "__init__.py" and p.name != "explore_raw.py"]) def _highlight_python(src: str) -> str: """HTML-escape the source. Real syntax highlighting is done client-side by highlight.js (loaded from CDN by the page template). Keeping this layer simple avoids a half-baked regex tokenizer that nests spans incorrectly inside its own output. """ return html.escape(src) _PAGE_TEMPLATE = """ {title} — Medicare Hospital Charge/Payment EDA

{file}

{n_lines} lines · {n_bytes:,} bytes · view raw

{body}
""" _INDEX_TEMPLATE = """ Source Code — Medicare Hospital Charge/Payment EDA

Python source

Every analysis step is reproducible. Each file below is runnable via uv run python -m src.<module>; the end-to-end pipeline runs from uv run python -m src.build_site. The narrative pages of this site embed curated snippets — this browser shows the full source.

← back to narrative

""" def _sidebar(active_filename: str) -> str: items = [] for p in CODE_FILES: cls = "active" if p.name == active_filename else "" items.append( f'
  • {p.name}
    • ' ) return "\n ".join(items) def generate_code_page(src_path: Path): src = src_path.read_text() body = _highlight_python(src) # Save both the highlighted HTML page and a raw text file (for "view raw") raw_path = SITE_CODE / f"{src_path.name}.txt" raw_path.write_text(src) html_path = SITE_CODE / f"{src_path.name}.html" html_path.write_text(_PAGE_TEMPLATE.format( title=f"src/{src_path.name}", file=src_path.name, n_lines=src.count("\n") + 1, n_bytes=len(src), sidebar=_sidebar(src_path.name), body=body, )) return html_path, raw_path def generate_code_index() -> Path: items = [] # Group: pipeline (config/load/clean/features/run_pipeline), analysis # (insights/outliers/predictors), build (build_site) for p in CODE_FILES: n_bytes = p.stat().st_size items.append( f'
  • {p.name} ' f'{n_bytes:,} bytes
    • ' ) html = _INDEX_TEMPLATE.format(files="\n ".join(items)) path = SITE_CODE / "index.html" path.write_text(html) return path def build_code_browser(): SITE_CODE.mkdir(parents=True, exist_ok=True) for f in CODE_FILES: html_path, _ = generate_code_page(f) generate_code_index() print(f" code browser: {len(CODE_FILES)} files written to site/code/") def _sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def _file_record(path: Path, relative_to: Path = ROOT) -> dict: if not path.exists(): return {"exists": False, "path": str(path.relative_to(relative_to))} return { "exists": True, "path": str(path.relative_to(relative_to)), "bytes": int(path.stat().st_size), "sha256": _sha256(path), } def _run_contract_tests() -> dict: """Run the repository's standard-library contract tests during a build.""" command = [sys.executable, "-m", "unittest", "discover", "-s", "tests", "-q"] result = subprocess.run( command, cwd=ROOT, capture_output=True, text=True, check=False, ) if result.returncode: raise RuntimeError( "contract tests failed:\n" f"{result.stdout}\n{result.stderr}" ) return { "status": "passed", "command": "uv run python -m unittest discover -s tests -q", "stdout": result.stdout.strip(), } def _copy_portable_audit_artifacts() -> dict: """Copy complete audit tables into the deployable static site.""" target_dir = SITE_DATA / "audit" target_dir.mkdir(parents=True, exist_ok=True) records = {} for name in AUDIT_ARTIFACT_NAMES: source = DATA_PROC / name target = target_dir / name if source.exists(): shutil.copy2(source, target) records[name] = _file_record(target) return records def build_audit_manifest(test_result: dict | None = None) -> Path: """Write a compact provenance and artifact-integrity manifest.""" meta_path = SITE_DATA / "meta_summary.json" meta = json.loads(meta_path.read_text()) source_paths = [ ROOT / "data" / "raw" / "inpatient_2023.csv", ROOT / "data" / "raw" / "outpatient_2023.csv", ] source_files = {} for path in source_paths: record = _file_record(path) if record["exists"]: record["rows"] = ( meta["inpatient"]["rows"] if path.name.startswith("inpatient") else meta["outpatient_full"]["rows"] ) source_files[record["path"]] = record artifacts = {} for path in sorted(SITE_DATA.glob("*.json")): if path.name == "audit_manifest.json": continue record = _file_record(path) try: payload = json.loads(path.read_text()) except (OSError, json.JSONDecodeError): payload = None if isinstance(payload, dict): metadata = payload.get("metadata") if isinstance(metadata, dict): for key in ( "preview", "truncated", "preview_limit_per_dataset", "full_row_counts", "preview_row_counts", ): if key in metadata: record[key] = metadata[key] artifacts[path.name] = record processed_artifacts = { name: _file_record(DATA_PROC / name) for name in AUDIT_ARTIFACT_NAMES } code_files = { str(path.relative_to(ROOT)): _file_record(path) for path in sorted(SRC_DIR.glob("*.py")) } test_files = { str(path.relative_to(ROOT)): _file_record(path) for path in sorted((ROOT / "tests").glob("*.py")) } dependency_files = { str(path.relative_to(ROOT)): _file_record(path) for path in (ROOT / "pyproject.toml", ROOT / "uv.lock") if path.exists() } provenance = { **source_files, **code_files, **test_files, **dependency_files, } source_hash = hashlib.sha256( json.dumps( {path: record.get("sha256", "missing") for path, record in sorted(provenance.items())}, sort_keys=True, ).encode() ).hexdigest() portable_audit_artifacts = _copy_portable_audit_artifacts() manifest = { "schema_version": "round2-v1", "run_id": source_hash, "periods": meta.get("periods", {}), "source_files": source_files, "code_files": code_files, "test_files": test_files, "dependency_files": dependency_files, "mapping_validation": meta.get("mapping_validation", {}), "denominators": { "outpatient": meta.get("outpatient_denominators", {}), "provider_overlap_full": meta.get("provider_overlap_full", {}), "provider_overlap_cost_observed": meta.get( "provider_overlap_cost_observed", {} ), }, "artifacts": artifacts, "processed_audit_artifacts": processed_artifacts, "portable_audit_artifacts": portable_audit_artifacts, "checks": { "source_files_present": all( record["exists"] for record in source_files.values() ), "json_artifacts_present": all( record["exists"] for record in artifacts.values() ), "outlier_preview_marked": bool( artifacts.get("outlier_flags.json", {}).get("truncated") ), "complete_audit_artifacts_present": all( record["exists"] for record in portable_audit_artifacts.values() ), }, "contract_tests": test_result or {"status": "not_run"}, } path = SITE_DATA / "audit_manifest.json" path.write_text(json.dumps(manifest, indent=2) + "\n") print(" audit_manifest.json saved.") return path def run_all_analyses(): """Ensures all site/data/*.json are present before bundling.""" print("[build_site] running pipeline ...") run_pipeline.run_pipeline() print("[build_site] insights ...") analysis_insights.run() print("[build_site] outliers ...") analysis_outliers.run() print("[build_site] predictors ...") analysis_predictors.run() # Sanity: confirm all JSONs exist expected = [ "meta_summary.json", "cv_tables.json", "five_rankings.json", "regional_choropleths.json", "billing_payment_correlation.json", "cross_dataset_variators.json", "outlier_cohort_comparison.json", "outlier_method_agreement.json", "outlier_concentration.json", "cross_dataset_outliers.json", "outlier_profile.json", "outlier_flags.json", "predictors.json", "mapping_validation.json", "denominator_audit.json", "model_validation_manifest.json", "maryland_sensitivity.json", ] missing = [f for f in expected if not (SITE_DATA / f).exists()] if missing: raise RuntimeError(f"missing site/data files: {missing}") _copy_portable_audit_artifacts() test_result = _run_contract_tests() build_audit_manifest(test_result) def build_all(): SITE_DIR.mkdir(parents=True, exist_ok=True) (SITE_DIR / "assets" / "css").mkdir(parents=True, exist_ok=True) (SITE_DIR / "assets" / "js").mkdir(parents=True, exist_ok=True) (SITE_DIR / "data").mkdir(parents=True, exist_ok=True) (SITE_DIR / "code").mkdir(parents=True, exist_ok=True) # 1. Re-run analysis (idempotent — overwrites site/data/*.json) run_all_analyses() # 2. Build the code browser print("[build_site] generating code browser ...") build_code_browser() print("[build_site] DONE.") print(f" site root: {SITE_DIR}") print(f" deploy: scp -r {SITE_DIR}/* user@vps:/var/www/medicare-cost/") if __name__ == "__main__": # pragma: no cover build_all()