#!/usr/bin/env python3
"""Build the standalone MW/M31 profile explorer as an atomic static site."""

from __future__ import annotations

import hashlib
import html
import json
import ctypes
import fcntl
import os
import re
import shutil
import tempfile
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Iterator
from urllib.parse import unquote, urlparse

import pandas as pd
from bs4 import BeautifulSoup
from PIL import Image

ROOT = Path(__file__).resolve().parent
PROJECT_ROOT = ROOT.parent
PAPER_ROOT = PROJECT_ROOT / "paper_apj_v19_cgmsum_draft"
TEMPLATE_ROOT = ROOT / "templates"
PRODUCT_ROOT = ROOT / "products"
FIGURE_ROOT = ROOT / "figures"
SITE_ROOT = ROOT / "cloudflare_mw_m31_profile_explorer" / "site"
CANONICAL_URL = "https://m31cgm-mw-m31-profile-explorer.pages.dev/"

REQUIRED_FIGURES = (
    "mw_latitude_profile_1col.png",
    "mw_latitude_profile_1col.pdf",
    "mw_latitude_profile_2col.png",
    "mw_latitude_profile_2col.pdf",
    "mw_signed_axis_profile_1col.png",
    "mw_signed_axis_profile_1col.pdf",
    "mw_signed_axis_profile_2col.png",
    "mw_signed_axis_profile_2col.pdf",
    "cgmsum_m31_radius_profile_1col.png",
    "cgmsum_m31_radius_profile_1col.pdf",
    "cgmsum_m31_radius_profile_2col.png",
    "cgmsum_m31_radius_profile_2col.pdf",
    "figure3_field_conversion_1col.png",
    "figure3_field_conversion_1col.pdf",
    "figure3_field_conversion_2col.png",
    "figure3_field_conversion_2col.pdf",
)
REQUIRED_PRODUCTS = (
    "mw_latitude_profile.csv",
    "mw_m31_gc_signed_axis_profile.csv",
    "m31_field_profile_decomposition.csv",
    "conditional_m31_templates.csv",
    "explorer_summary.json",
)
PAPER_INPUTS = (
    "m31_cgmsum_v19_primary_measurements_public.csv",
    "m31_cgmsum_locatelli2024_reference_beta0p5_m31_footprint_predictions.csv",
    "m31_cgmsum_conditional_prior_ledger.csv",
    "m31_cgmsum_conditional_prior_audit.json",
    "m31_cgmsum_apec_bandpass_matrix.csv",
    "m31_cgmsum_apec_bandpass_audit.json",
    "audit_apj_v19_cgmsum_bandpass.py",
    "audit_apj_v19_cgmsum_conditional_priors.py",
    "make_apj_v19_cgmsum_conditional_figures.py",
)


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def human_bytes(size: int) -> str:
    if size < 1024:
        return f"{size} B"
    if size < 1024**2:
        return f"{size / 1024:.1f} KB"
    return f"{size / 1024**2:.1f} MB"


def select_column(frame: pd.DataFrame, *candidates: str) -> str:
    for candidate in candidates:
        if candidate in frame.columns:
            return candidate
    raise ValueError(f"none of the required columns exist: {candidates}")


def render_field_table(source: Path | None = None) -> str:
    source = source or PRODUCT_ROOT / "m31_field_profile_decomposition.csv"
    frame = pd.read_csv(source)
    columns = {
        "obsid": select_column(frame, "obsid"),
        "side": select_column(frame, "side"),
        "rproj": select_column(frame, "rproj_kpc", "projected_radius_kpc"),
        "rparallel": select_column(
            frame, "signed_axis_r_kpc", "signed_r_parallel_kpc", "m31_to_gc_axis_kpc"
        ),
        "nh": select_column(frame, "nh_hi4pi_1e22_cm-2"),
        "observed": select_column(
            frame,
            "cgmsum_absorbed_0p4_1p25_primary_fluxunit",
            "cgmsum_flux_primary_unit",
            "absorbed_flux_soft_0p4_1p25_primary_fluxunit",
            "absorbed_flux_soft_0p4_1p25_erg_cm-2_s-1_arcmin-2",
        ),
        "error": select_column(
            frame,
            "cgmsum_absorbed_0p4_1p25_staterr_primary_fluxunit",
            "cgmsum_flux_staterr_primary_unit",
            "absorbed_flux_soft_0p4_1p25_staterr_primary_fluxunit",
            "absorbed_flux_soft_0p4_1p25_staterr_erg_cm-2_s-1_arcmin-2",
        ),
        "fixed": select_column(
            frame,
            "mw_reference_fixed_nh_primary_fluxunit",
            "geometry_fixed_nh_primary_fluxunit",
            "reference_response_matched_fixed_nh_primary_fluxunit",
        ),
        "actual": select_column(
            frame,
            "mw_reference_actual_nh_primary_fluxunit",
            "reference_response_matched_absorbed_0p4_1p25_primary_fluxunit",
        ),
    }
    rows = []
    for _, row in frame.sort_values(columns["rproj"]).iterrows():
        observed = float(row[columns["observed"]])
        error = float(row[columns["error"]])
        if abs(observed) < 1.0e-10:
            observed *= 1.0e15
            error *= 1.0e15
        values = (
            html.escape(str(row[columns["obsid"]])),
            html.escape(str(row[columns["side"]])),
            f"{float(row[columns['rproj']]):.2f}",
            f"{float(row[columns['rparallel']]):+.2f}",
            f"{float(row[columns['nh']]):.4f}",
            f"{observed:.3f} ± {error:.3f}",
            f"{float(row[columns['fixed']]):.3f}",
            f"{float(row[columns['actual']]):.3f}",
        )
        rows.append("<tr>" + "".join(f"<td>{value}</td>" for value in values) + "</tr>")
    return (
        '<table class="data"><caption><span data-zh>面亮度：10<sup>−15</sup> erg cm<sup>−2</sup> s<sup>−1</sup> arcmin<sup>−2</sup>；N<sub>H</sub>：10<sup>22</sup> cm<sup>−2</sup>。</span>'
        "<span data-en>Surface brightness: 10<sup>−15</sup> erg cm<sup>−2</sup> s<sup>−1</sup> arcmin<sup>−2</sup>; N<sub>H</sub>: 10<sup>22</sup> cm<sup>−2</sup>.</span></caption><thead><tr>"
        '<th scope="col">OBSID</th><th scope="col"><span data-zh>分区</span><span data-en>Side</span></th><th scope="col">R<sub>proj</sub> (kpc)</th><th scope="col">R<sub>∥</sub> (kpc)</th>'
        '<th scope="col">N<sub>H</sub></th><th scope="col"><span data-zh>观测 CGMsum</span><span data-en>Observed CGMsum</span></th><th scope="col"><span data-zh>MW固定NH</span><span data-en>MW fixed-NH</span></th><th scope="col"><span data-zh>MW实际HI4PI</span><span data-en>MW actual HI4PI</span></th>'
        "</tr></thead><tbody>" + "".join(rows) + "</tbody></table>"
    )


def source_inventory() -> dict[str, Path]:
    inventory: dict[str, Path] = {}
    for name in REQUIRED_PRODUCTS:
        inventory[f"assets/data/{name}"] = PRODUCT_ROOT / name
    for extra in sorted(PRODUCT_ROOT.glob("*.csv")) + sorted(
        PRODUCT_ROOT.glob("*.json")
    ):
        inventory.setdefault(f"assets/data/{extra.name}", extra)
    for name in REQUIRED_FIGURES:
        inventory[f"assets/figures/{name}"] = FIGURE_ROOT / name
    for extra in sorted(FIGURE_ROOT.glob("*.png")) + sorted(FIGURE_ROOT.glob("*.pdf")):
        inventory.setdefault(f"assets/figures/{extra.name}", extra)
    inventory["assets/figures/draft_figure3_contribution_plane.png"] = (
        PAPER_ROOT / "m31_cgmsum_mw_m31_constraint_plane_2col.png"
    )
    for column in ("1col", "2col"):
        for suffix in ("png", "pdf"):
            name = f"draft_figure3_contribution_plane_loglog_{column}.{suffix}"
            inventory[f"assets/figures/{name}"] = (
                PAPER_ROOT
                / f"m31_cgmsum_mw_m31_constraint_plane_loglog_{column}.{suffix}"
            )
    for name in PAPER_INPUTS:
        inventory[f"assets/source/{name}"] = PAPER_ROOT / name
    for name in (
        "generate_explorer_products.py",
        "build_explorer_site.py",
        "verify_production.py",
        "test_generate_explorer_products.py",
        "test_build_explorer_site.py",
        "test_verify_production.py",
        "pytest.ini",
        "README.md",
        "SITE_SPEC.md",
    ):
        path = ROOT / name
        if path.exists():
            inventory[f"assets/source/{name}"] = path
    missing = [str(path) for path in inventory.values() if not path.is_file()]
    if missing:
        raise FileNotFoundError(f"required explorer assets are missing: {missing}")
    return inventory


def snapshot_inputs(
    snapshot_root: Path,
) -> tuple[dict[str, Path], Path, dict[Path, str]]:
    live_inventory = source_inventory()
    live_templates = tuple(
        TEMPLATE_ROOT / name for name in ("index.html", "styles.css", "app.js")
    )
    live_paths = set(live_inventory.values()) | set(live_templates)
    expected_hashes = {path.resolve(): sha256(path) for path in live_paths}

    snapshot_inventory: dict[str, Path] = {}
    inventory_root = snapshot_root / "inventory"
    for destination, source in live_inventory.items():
        output = inventory_root / destination
        output.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(source, output)
        if sha256(output) != expected_hashes[source.resolve()]:
            raise RuntimeError(f"input changed while snapshotting: {source}")
        snapshot_inventory[destination] = output

    snapshot_templates = snapshot_root / "templates"
    snapshot_templates.mkdir(parents=True, exist_ok=True)
    for source in live_templates:
        output = snapshot_templates / source.name
        shutil.copy2(source, output)
        if sha256(output) != expected_hashes[source.resolve()]:
            raise RuntimeError(f"template changed while snapshotting: {source}")
    verify_source_hashes(expected_hashes)
    return snapshot_inventory, snapshot_templates, expected_hashes


def verify_source_hashes(expected_hashes: dict[Path, str]) -> None:
    changed = [
        str(path)
        for path, expected in expected_hashes.items()
        if not path.is_file() or sha256(path) != expected
    ]
    if changed:
        raise RuntimeError(f"build inputs changed after snapshot: {changed}")


def render_downloads(inventory: dict[str, Path]) -> str:
    rows = []
    for destination, source in sorted(inventory.items()):
        label = destination.removeprefix("assets/")
        rows.append(
            f'<a class="download" href="{html.escape(destination)}">'
            f"<span>{html.escape(label)}</span><span>{human_bytes(source.stat().st_size)}</span></a>"
        )
    return "".join(rows)


def remove_generated_tree(path: Path) -> None:
    if path.is_symlink() or path.is_file():
        path.unlink()
    elif path.exists():
        shutil.rmtree(path)


def render_site(
    target: Path,
    *,
    inventory: dict[str, Path] | None = None,
    template_root: Path | None = None,
    builder_sha256: str | None = None,
) -> None:
    inventory = inventory or source_inventory()
    template_root = template_root or TEMPLATE_ROOT
    remove_generated_tree(target)
    target.mkdir(parents=True)
    for destination, source in inventory.items():
        output = target / destination
        output.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(source, output)
    shutil.copy2(template_root / "styles.css", target / "assets/styles.css")
    shutil.copy2(template_root / "app.js", target / "assets/app.js")

    build_utc = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
    page = (template_root / "index.html").read_text(encoding="utf-8")
    page = page.replace(
        "{{FIELD_TABLE}}",
        render_field_table(
            inventory["assets/data/m31_field_profile_decomposition.csv"]
        ),
    )
    page = page.replace("{{DOWNLOADS}}", render_downloads(inventory))
    page = page.replace("{{BUILD_TIMESTAMP}}", build_utc)
    if "{{" in page or "}}" in page:
        raise ValueError("unresolved template token in index.html")
    (target / "index.html").write_text(page, encoding="utf-8")
    english_page = (
        page.replace(
            '<html lang="zh-CN" data-lang="zh">', '<html lang="en" data-lang="en">', 1
        )
        .replace(
            "<title>MW参考模型如何进入M31 Figure 3 · Rui Huang</title>",
            "<title>How the MW reference enters M31 Figure 3 · Rui Huang</title>",
            1,
        )
        .replace(
            'data-lang-button="zh" aria-pressed="true"',
            'data-lang-button="zh" aria-pressed="false"',
            1,
        )
        .replace(
            'data-lang-button="en" aria-pressed="false"',
            'data-lang-button="en" aria-pressed="true"',
            1,
        )
    )
    (target / "index-en.html").write_text(english_page, encoding="utf-8")
    (target / "_headers").write_text(
        "/*\n"
        "  Content-Security-Policy: default-src 'self'; img-src 'self' data:; "
        "style-src 'self'; script-src 'self'; object-src 'none'; base-uri 'self'; "
        "frame-ancestors 'none'; form-action 'none'\n"
        "  X-Content-Type-Options: nosniff\n"
        "  Referrer-Policy: strict-origin-when-cross-origin\n"
        "  Permissions-Policy: camera=(), microphone=(), geolocation=(), payment=()\n"
        "  Cross-Origin-Opener-Policy: same-origin\n"
        "/assets/*\n"
        "  Cache-Control: public, max-age=3600, must-revalidate\n",
        encoding="utf-8",
    )

    manifest_entries = []
    deployment_config_entries = []
    for path in sorted(item for item in target.rglob("*") if item.is_file()):
        relative = path.relative_to(target).as_posix()
        if relative == "report_manifest.json":
            continue
        if path.name in {"_headers", "_redirects"}:
            deployment_config_entries.append(
                {"path": relative, "bytes": path.stat().st_size, "sha256": sha256(path)}
            )
            continue
        manifest_entries.append(
            {"path": relative, "bytes": path.stat().st_size, "sha256": sha256(path)}
        )
    manifest = {
        "canonical_url": CANONICAL_URL,
        "build_utc": build_utc,
        "generator": "build_explorer_site.py",
        "builder_sha256": builder_sha256 or sha256(Path(__file__).resolve()),
        "file_count": len(manifest_entries),
        "total_bytes": sum(item["bytes"] for item in manifest_entries),
        "files": manifest_entries,
        "deployment_config": deployment_config_entries,
    }
    (target / "report_manifest.json").write_text(
        json.dumps(manifest, indent=2) + "\n", encoding="utf-8"
    )


def validate_document_references(
    target: Path, documents: tuple[Path, ...]
) -> list[str]:
    broken: list[str] = []
    target_resolved = target.resolve()
    soup_cache: dict[Path, BeautifulSoup] = {}
    for document in documents:
        soup = BeautifulSoup(document.read_text(encoding="utf-8"), "html.parser")
        soup_cache[document.resolve()] = soup
        for element in soup.find_all(src=True) + soup.find_all(href=True):
            attribute = "src" if element.has_attr("src") else "href"
            reference = str(element[attribute])
            parsed = urlparse(reference)
            if parsed.scheme or reference.startswith(("data:", "mailto:")):
                continue
            raw_path = unquote(parsed.path)
            if not raw_path:
                destination = document
            elif raw_path.startswith("/"):
                destination = target / raw_path.lstrip("/")
            else:
                destination = document.parent / raw_path
            resolved = destination.resolve()
            if (
                not resolved.is_relative_to(target_resolved)
                or not destination.is_file()
            ):
                broken.append(f"{document.name}: {reference}")
                continue
            if parsed.fragment:
                destination_soup = soup_cache.get(resolved)
                if destination_soup is None:
                    destination_soup = BeautifulSoup(
                        destination.read_text(encoding="utf-8"), "html.parser"
                    )
                    soup_cache[resolved] = destination_soup
                if destination_soup.find(id=unquote(parsed.fragment)) is None:
                    broken.append(f"{document.name}: missing fragment {reference}")
    return broken


def served_asset_paths(target: Path) -> set[str]:
    return {
        path.relative_to(target).as_posix()
        for path in target.rglob("*")
        if path.is_file()
        and path.name not in {"report_manifest.json", "_headers", "_redirects"}
    }


def validate_site(target: Path) -> None:
    index = target / "index.html"
    page = index.read_text(encoding="utf-8")
    required_phrases = (
        "观测视线总和",
        "observed line-of-sight sum",
        "条件模板",
        "conditional template",
        "固定NH的几何曲线",
        "fixed-NH geometry curve",
        "模型张力残差",
        "model-tension residual",
        "n<sub>h</sub>²+n<sub>d</sub>²",
        "No real MOS RMF/ARF",
        "0.949876",
        "1.431–1.668",
        "1.560",
        "1.602",
    )
    missing_phrases = [phrase for phrase in required_phrases if phrase not in page]
    if missing_phrases:
        raise ValueError(f"required scientific wording missing: {missing_phrases}")
    if "{{" in page or "}}" in page:
        raise ValueError("unresolved template token remains")

    documents = (index, target / "index-en.html")
    forbidden_patterns = (
        re.compile(
            r"\bthe 14 observed points are an? M31 (?:surface-brightness )?profile\b",
            re.I,
        ),
        re.compile(r"\bthe 14 observed points are an? M31 contribution\b", re.I),
        re.compile(r"\bforeground-subtracted M31 detection\b", re.I),
        re.compile(r"14个观测点是(?:一个)?M31(?:面亮度)?(?:径向)?profile", re.I),
    )
    violations = []
    for document in documents:
        soup = BeautifulSoup(document.read_text(encoding="utf-8"), "html.parser")
        for node in soup.find_all(string=True):
            text = " ".join(str(node).split())
            if any(
                pattern.search(text) for pattern in forbidden_patterns
            ) and not node.find_parent(class_="warning"):
                violations.append(f"{document.name}: {text}")
    if violations:
        raise ValueError(f"forbidden scientific wording outside warning: {violations}")

    broken = validate_document_references(target, documents)
    if broken:
        raise FileNotFoundError(f"broken local links: {sorted(broken)}")

    for image in target.rglob("*.png"):
        with Image.open(image) as opened:
            if opened.width < 300 or opened.height < 200:
                raise ValueError(
                    f"invalid scientific image dimensions: {image} {opened.size}"
                )

    manifest_path = target / "report_manifest.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    if manifest["canonical_url"] != CANONICAL_URL:
        raise ValueError("manifest canonical URL mismatch")
    entries = manifest.get("files", [])
    paths = [item.get("path") for item in entries]
    if any(not isinstance(path, str) for path in paths) or len(paths) != len(
        set(paths)
    ):
        raise ValueError("manifest paths must be unique strings")
    config_entries = manifest.get("deployment_config", [])
    config_paths = [item.get("path") for item in config_entries]
    if any(not isinstance(path, str) for path in config_paths) or len(
        config_paths
    ) != len(set(config_paths)):
        raise ValueError("deployment config paths must be unique strings")
    if set(paths) & set(config_paths):
        raise ValueError("served assets and deployment config must be disjoint")
    target_resolved = target.resolve()
    listed = set(paths)
    actual = served_asset_paths(target)
    if listed != actual:
        raise ValueError(
            f"manifest inventory mismatch: missing={sorted(actual - listed)}, extra={sorted(listed - actual)}"
        )
    actual_config = {
        path.relative_to(target).as_posix()
        for path in target.rglob("*")
        if path.is_file() and path.name in {"_headers", "_redirects"}
    }
    if set(config_paths) != actual_config:
        raise ValueError(
            "deployment config inventory mismatch: "
            f"missing={sorted(actual_config - set(config_paths))}, "
            f"extra={sorted(set(config_paths) - actual_config)}"
        )
    if manifest.get("file_count") != len(entries):
        raise ValueError("manifest file_count mismatch")
    if manifest.get("total_bytes") != sum(item["bytes"] for item in entries):
        raise ValueError("manifest total_bytes mismatch")
    for item in [*entries, *config_entries]:
        pure = PurePosixPath(item["path"])
        if pure.is_absolute() or ".." in pure.parts:
            raise ValueError(f"unsafe manifest path: {item['path']}")
        path = target / item["path"]
        if not path.resolve().is_relative_to(target_resolved):
            raise ValueError(f"manifest path escapes target: {item['path']}")
        if (
            not path.is_file()
            or path.stat().st_size != item["bytes"]
            or sha256(path) != item["sha256"]
        ):
            raise ValueError(f"manifest mismatch: {item['path']}")
    deployed_builder = target / "assets/source/build_explorer_site.py"
    if manifest.get("builder_sha256") != sha256(deployed_builder):
        raise ValueError("manifest builder snapshot hash mismatch")


@contextmanager
def build_lock(parent: Path) -> Iterator[None]:
    parent.mkdir(parents=True, exist_ok=True)
    lock_path = parent / ".site-build.lock"
    with lock_path.open("a+", encoding="utf-8") as handle:
        fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
        try:
            yield
        finally:
            fcntl.flock(handle.fileno(), fcntl.LOCK_UN)


def atomic_exchange(left: Path, right: Path) -> None:
    renameat2 = getattr(ctypes.CDLL(None, use_errno=True), "renameat2", None)
    if renameat2 is None:
        raise RuntimeError("atomic directory exchange requires Linux renameat2")
    renameat2.argtypes = [
        ctypes.c_int,
        ctypes.c_char_p,
        ctypes.c_int,
        ctypes.c_char_p,
        ctypes.c_uint,
    ]
    renameat2.restype = ctypes.c_int
    result = renameat2(
        -100,
        os.fsencode(left),
        -100,
        os.fsencode(right),
        2,
    )
    if result != 0:
        error = ctypes.get_errno()
        raise OSError(error, os.strerror(error), f"{left} <-> {right}")


def promote_staging_site(staging: Path, final: Path) -> None:
    if final.is_symlink() or (final.exists() and not final.is_dir()):
        raise ValueError(f"refusing to replace non-directory site root: {final}")
    if final.exists():
        atomic_exchange(staging, final)
    else:
        os.replace(staging, final)


def build(site_root: Path = SITE_ROOT) -> None:
    parent = site_root.parent
    with build_lock(parent):
        workspace = Path(tempfile.mkdtemp(prefix=".site-build-", dir=parent))
        snapshot = workspace / "snapshot"
        staging = workspace / "staging"
        try:
            inventory, templates, expected_hashes = snapshot_inputs(snapshot)
            builder_hash = expected_hashes[Path(__file__).resolve()]
            render_site(
                staging,
                inventory=inventory,
                template_root=templates,
                builder_sha256=builder_hash,
            )
            validate_site(staging)
            verify_source_hashes(expected_hashes)
            promote_staging_site(staging, site_root)
        finally:
            remove_generated_tree(workspace)
    manifest = json.loads(
        (site_root / "report_manifest.json").read_text(encoding="utf-8")
    )
    print(
        f"Built {site_root} with {manifest['file_count']} manifest files "
        f"({manifest['total_bytes']:,} bytes)"
    )


if __name__ == "__main__":
    build()