"""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 = """<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<link rel="icon" href="data:,">
<title>{title} — Medicare Hospital Charge/Payment EDA</title>
<link rel="stylesheet" href="/assets/css/style.css">
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/highlight.js@11.10.0/styles/atom-one-dark.min.css">
<script src="https://cdn.jsdelivr.net/npm/highlight.js@11.10.0/lib/highlight.min.js"></script>
</head>
<body>
<nav class="topnav">
<a href="/index.html" class="brand">Medicare Hospital Charge/Payment Patterns · CMS 2023 EDA</a>
<div class="nav-links">
<a href="/index.html">Overview</a>
<a href="/approach.html">Approach</a>
<a href="/insights.html">Insights</a>
<a href="/outliers.html">Outliers</a>
<a href="/predictors.html">Predictors</a>
<a href="/agent-review.html">Agent Review</a>
<a href="/code/">Code</a>
</div>
</nav>
<main class="code-page">
<aside class="sidebar">
<h2>Source files</h2>
<ul>
{sidebar}
</ul>
<p class="sidebar-foot"><a href="/">← back to narrative</a></p>
</aside>
<article class="code-article">
<header class="code-header">
<h1>{file}</h1>
<p class="code-meta">{n_lines} lines · {n_bytes:,} bytes · <a href="/code/{file}.txt">view raw</a></p>
</header>
<pre class="code-block"><code class="language-python hljs">{body}</code></pre>
</article>
</main>
<script>hljs.highlightAll();</script>
</body>
</html>
"""
_INDEX_TEMPLATE = """<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<link rel="icon" href="data:,">
<title>Source Code — Medicare Hospital Charge/Payment EDA</title>
<link rel="stylesheet" href="/assets/css/style.css">
</head>
<body>
<nav class="topnav">
<a href="/index.html" class="brand">Medicare Hospital Charge/Payment Patterns · CMS 2023 EDA</a>
<div class="nav-links">
<a href="/index.html">Overview</a>
<a href="/approach.html">Approach</a>
<a href="/insights.html">Insights</a>
<a href="/outliers.html">Outliers</a>
<a href="/predictors.html">Predictors</a>
<a href="/agent-review.html">Agent Review</a>
<a href="/code/" class="active">Code</a>
</div>
</nav>
<main class="code-index">
<h1>Python source</h1>
<p class="lead">Every analysis step is reproducible. Each file below is runnable
via <code>uv run python -m src.<module></code>; the end-to-end pipeline
runs from <code>uv run python -m src.build_site</code>. The narrative pages of
this site embed curated snippets — this browser shows the full source.</p>
<ul class="file-list">
{files}
</ul>
<p class="foot"><a href="/">← back to narrative</a></p>
</main>
</body>
</html>
"""
def _sidebar(active_filename: str) -> str:
items = []
for p in CODE_FILES:
cls = "active" if p.name == active_filename else ""
items.append(
f'<li><a class="{cls}" href="/code/{p.name}.html">{p.name}</a></li>'
)
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'<li><a href="/code/{p.name}.html"><strong>{p.name}</strong></a> '
f'<span class="size">{n_bytes:,} bytes</span></li>'
)
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()