"""Read-only operator dashboard and verified reconciliation evidence."""

from __future__ import annotations

import json
import os
import re
import stat
from collections.abc import Mapping
from html import escape
from pathlib import Path
from typing import Final, cast

from .errors import InvalidConfiguration

_MAX_SNAPSHOT_BYTES: Final = 512 * 1024
_SHA256_PATTERN: Final = re.compile(r"^[a-f0-9]{64}$")
_SNAPSHOT_ENV: Final = "OSM_LEAD_SOURCE_DASHBOARD_SNAPSHOT_PATH"


def _release_snapshot() -> dict[str, object]:
    """Return the latest accepted production evidence bundled with this release."""

    return {
        "kind": "verified-release-evidence",
        "generated_at": "2026-07-17T10:20:22Z",
        "region": "au-act",
        "profile": "luxeillum",
        "profile_version": "0.1.0-draft",
        "report_hash": "bdc4661686bde64905c1c6cce986b4c053f78eeb5a9b88381ae62e45d254e342",
        "source_sha256": "00cdce6d9731661b1d6d650214ce1fdbd0957bf799a8912cc783ba25d67aca19",
        "source_size_bytes": 18_605_640,
        "counts": {
            "active": 980,
            "review": 0,
            "excluded": 601,
            "out_of_scope": 2_716_858,
        },
        "categories": {
            "Restaurant / cafe": 824,
            "Bar / nightlife": 90,
            "Accommodation": 66,
        },
        "coverage": {
            "configured_markets": 1,
            "completed_markets": 1,
            "failed_markets": 0,
            "configured_shards": 1,
            "completed_shards": 1,
            "failed_shards": 0,
        },
        "noco": {
            "comparison_completed": True,
            "exact_identity_matches": 502,
            "unmatched_identities": 478,
            "duplicate_row_excess_detected": False,
        },
        "run": {
            "cache_hit": False,
            "completed_at": "2026-07-17T10:20:22Z",
            "started_at": "2026-07-17T10:20:22Z",
            "status": "completed",
        },
    }


def _bounded_text(value: object, *, field: str, max_length: int = 160) -> str:
    if not isinstance(value, str):
        raise InvalidConfiguration(f"{field} must be a string")
    normalized = value.strip()
    if not normalized or len(normalized) > max_length:
        raise InvalidConfiguration(f"{field} must be nonblank and bounded")
    return normalized


def _bounded_count(value: object, *, field: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or not 0 <= value <= 1_000_000_000_000:
        raise InvalidConfiguration(f"{field} must be a bounded nonnegative integer")
    return value


def _count_mapping(value: object, *, field: str, max_items: int) -> dict[str, int]:
    if not isinstance(value, dict) or not 1 <= len(value) <= max_items:
        raise InvalidConfiguration(f"{field} must be a bounded object")
    result: dict[str, int] = {}
    for key, raw_count in value.items():
        name = _bounded_text(key, field=f"{field} key", max_length=80)
        result[name] = _bounded_count(raw_count, field=f"{field}.{name}")
    return result


def _validate_snapshot(value: object) -> dict[str, object]:
    if not isinstance(value, dict):
        raise InvalidConfiguration("dashboard snapshot must be a JSON object")
    expected = {
        "kind",
        "generated_at",
        "region",
        "profile",
        "profile_version",
        "report_hash",
        "source_sha256",
        "source_size_bytes",
        "counts",
        "categories",
        "coverage",
        "noco",
        "run",
    }
    if set(value) != expected:
        raise InvalidConfiguration("dashboard snapshot fields do not match the accepted schema")

    report_hash = _bounded_text(value["report_hash"], field="report_hash", max_length=64)
    source_sha256 = _bounded_text(value["source_sha256"], field="source_sha256", max_length=64)
    if _SHA256_PATTERN.fullmatch(report_hash) is None:
        raise InvalidConfiguration("report_hash must be a lowercase SHA-256")
    if _SHA256_PATTERN.fullmatch(source_sha256) is None:
        raise InvalidConfiguration("source_sha256 must be a lowercase SHA-256")

    coverage = value["coverage"]
    coverage_fields = {
        "configured_markets",
        "completed_markets",
        "failed_markets",
        "configured_shards",
        "completed_shards",
        "failed_shards",
    }
    if not isinstance(coverage, dict) or set(coverage) != coverage_fields:
        raise InvalidConfiguration("coverage fields do not match the accepted schema")
    validated_coverage = {
        name: _bounded_count(coverage[name], field=f"coverage.{name}")
        for name in sorted(coverage_fields)
    }
    if validated_coverage["completed_markets"] > validated_coverage["configured_markets"]:
        raise InvalidConfiguration("completed markets cannot exceed configured markets")
    if validated_coverage["failed_markets"] > validated_coverage["configured_markets"]:
        raise InvalidConfiguration("failed markets cannot exceed configured markets")
    if validated_coverage["completed_shards"] > validated_coverage["configured_shards"]:
        raise InvalidConfiguration("completed shards cannot exceed configured shards")
    if validated_coverage["failed_shards"] > validated_coverage["configured_shards"]:
        raise InvalidConfiguration("failed shards cannot exceed configured shards")

    noco = value["noco"]
    if not isinstance(noco, dict) or set(noco) != {
        "comparison_completed",
        "exact_identity_matches",
        "unmatched_identities",
        "duplicate_row_excess_detected",
    }:
        raise InvalidConfiguration("noco evidence fields do not match the accepted schema")
    comparison_completed = noco["comparison_completed"]
    duplicate_excess = noco["duplicate_row_excess_detected"]
    if not isinstance(comparison_completed, bool) or not isinstance(duplicate_excess, bool):
        raise InvalidConfiguration("noco state flags must be boolean")

    run = value["run"]
    if not isinstance(run, dict) or set(run) != {
        "cache_hit",
        "completed_at",
        "started_at",
        "status",
    }:
        raise InvalidConfiguration("run evidence fields do not match the accepted schema")
    cache_hit = run["cache_hit"]
    if not isinstance(cache_hit, bool):
        raise InvalidConfiguration("run.cache_hit must be boolean")
    run_status = _bounded_text(run["status"], field="run.status", max_length=24)
    if run_status not in {"running", "partial", "completed"}:
        raise InvalidConfiguration("run.status is not an accepted worker state")

    return {
        "kind": _bounded_text(value["kind"], field="kind", max_length=80),
        "generated_at": _bounded_text(value["generated_at"], field="generated_at", max_length=40),
        "region": _bounded_text(value["region"], field="region", max_length=80),
        "profile": _bounded_text(value["profile"], field="profile", max_length=80),
        "profile_version": _bounded_text(
            value["profile_version"],
            field="profile_version",
            max_length=80,
        ),
        "report_hash": report_hash,
        "source_sha256": source_sha256,
        "source_size_bytes": _bounded_count(
            value["source_size_bytes"],
            field="source_size_bytes",
        ),
        "counts": _count_mapping(value["counts"], field="counts", max_items=16),
        "categories": _count_mapping(value["categories"], field="categories", max_items=64),
        "coverage": validated_coverage,
        "noco": {
            "comparison_completed": comparison_completed,
            "exact_identity_matches": _bounded_count(
                noco["exact_identity_matches"],
                field="noco.exact_identity_matches",
            ),
            "unmatched_identities": _bounded_count(
                noco["unmatched_identities"],
                field="noco.unmatched_identities",
            ),
            "duplicate_row_excess_detected": duplicate_excess,
        },
        "run": {
            "cache_hit": cache_hit,
            "completed_at": _bounded_text(
                run["completed_at"],
                field="run.completed_at",
                max_length=40,
            ),
            "started_at": _bounded_text(
                run["started_at"],
                field="run.started_at",
                max_length=40,
            ),
            "status": run_status,
        },
    }


def load_dashboard_snapshot(
    environ: Mapping[str, str] | None = None,
) -> tuple[dict[str, object], str | None]:
    """Load a bounded live snapshot, otherwise use accepted release evidence."""

    source = os.environ if environ is None else environ
    configured_path = source.get(_SNAPSHOT_ENV, "").strip()
    if not configured_path:
        return _release_snapshot(), None

    try:
        path = Path(configured_path)
        before = path.lstat()
        if not stat.S_ISREG(before.st_mode) or path.is_symlink():
            raise InvalidConfiguration("dashboard snapshot must be a regular non-symlink file")
        if not 1 <= before.st_size <= _MAX_SNAPSHOT_BYTES:
            raise InvalidConfiguration("dashboard snapshot size is outside the accepted bounds")
        raw = path.read_bytes()
        after = path.lstat()
        if (
            len(raw) != before.st_size
            or before.st_ino != after.st_ino
            or before.st_mtime_ns != after.st_mtime_ns
        ):
            raise InvalidConfiguration("dashboard snapshot changed during read")
        return _validate_snapshot(json.loads(raw.decode("utf-8"))), None
    except (OSError, UnicodeDecodeError, json.JSONDecodeError, InvalidConfiguration):
        return (
            _release_snapshot(),
            "Live worker evidence is unavailable; showing verified release evidence.",
        )


def _as_mapping(value: object) -> Mapping[str, object]:
    if isinstance(value, Mapping):
        return cast(Mapping[str, object], value)
    return {}


def _as_int(value: object) -> int:
    if isinstance(value, bool) or not isinstance(value, int):
        return 0
    return value


def _as_text(value: object, default: str = "—") -> str:
    if isinstance(value, str) and value:
        return value
    return default


def _number(value: object) -> str:
    return f"{_as_int(value):,}"


def build_dashboard_payload(
    status: Mapping[str, object],
    environ: Mapping[str, str] | None = None,
) -> dict[str, object]:
    """Combine secret-free runtime status with current candidate evidence."""

    snapshot, warning = load_dashboard_snapshot(environ)
    noco = _as_mapping(snapshot.get("noco"))
    coverage = _as_mapping(snapshot.get("coverage"))
    run = _as_mapping(snapshot.get("run"))
    comparison_completed = bool(noco.get("comparison_completed", False))
    configured_markets = _as_int(coverage.get("configured_markets"))
    completed_markets = _as_int(coverage.get("completed_markets"))
    failed_markets = _as_int(coverage.get("failed_markets"))
    market_coverage_complete = (
        configured_markets > 0
        and completed_markets == configured_markets
        and failed_markets == 0
        and run.get("status") == "completed"
    )
    readiness = {
        "deployment": {"ready": True, "label": "Replacement service deployed"},
        "live_worker": {
            "ready": snapshot.get("kind") == "runtime-evidence",
            "label": "Live Geofabrik worker has current evidence",
        },
        "market_coverage": {
            "ready": market_coverage_complete,
            "label": "All configured English-language markets completed",
        },
        "noco_comparison": {
            "ready": comparison_completed,
            "label": (
                "Exact OSM identity comparison complete"
                if comparison_completed
                else "Live Noco identity comparison pending"
            ),
        },
        "sidecar": {
            "ready": bool(status.get("sidecar_configured", False)),
            "label": "Production sidecar configured",
        },
        "legacy_cutover": {
            "ready": bool(status.get("legacy_writer_disabled", False)),
            "label": "Legacy writer disabled",
        },
        "controlled_writer": {
            "ready": bool(status.get("write_runtime_armed", False)),
            "label": "Controlled writer armed",
        },
        "monthly_scheduler": {
            "ready": bool(status.get("scheduler_enabled", False)),
            "label": "Monthly scheduler enabled",
        },
        "stale_lifecycle": {
            "ready": bool(status.get("stale_deletion_enabled", False)),
            "label": "Approved stale deletion enabled",
        },
    }
    next_gate = (
        "Run the live read-only Noco identity and adoption comparison. "
        "Lead creation remains disabled."
        if not comparison_completed
        else (
            "Persist reconciliation evidence and complete the capped insert-only pilot. "
            "Update and deletion remain disabled."
        )
    )
    return {
        "service": dict(status),
        "snapshot": snapshot,
        "snapshot_warning": warning,
        "readiness": readiness,
        "next_gate": next_gate,
    }


def _badge(enabled: bool, on_text: str, off_text: str) -> str:
    text = on_text if enabled else off_text
    tone = "good" if enabled else "muted"
    return f'<span class="badge {tone}">{escape(text)}</span>'


def render_dashboard(payload: Mapping[str, object]) -> bytes:
    """Render a self-contained, dependency-free read-only dashboard."""

    service = _as_mapping(payload.get("service"))
    snapshot = _as_mapping(payload.get("snapshot"))
    counts = _as_mapping(snapshot.get("counts"))
    categories = _as_mapping(snapshot.get("categories"))
    noco = _as_mapping(snapshot.get("noco"))
    run = _as_mapping(snapshot.get("run"))
    coverage = _as_mapping(snapshot.get("coverage"))
    readiness = _as_mapping(payload.get("readiness"))
    comparison_completed = bool(noco.get("comparison_completed", False))

    exact_matches = (
        _number(noco.get("exact_identity_matches")) if comparison_completed else "Pending"
    )
    unmatched = _number(noco.get("unmatched_identities")) if comparison_completed else "Pending"
    category_total = max(sum(_as_int(value) for value in categories.values()), 1)
    category_rows = "".join(
        (
            '<div class="category"><div><strong>'
            f"{escape(str(name))}</strong><span>{_number(value)}</span></div>"
            '<div class="bar"><i style="width:'
            f"{min(100, round((_as_int(value) / category_total) * 100, 1))}%"
            '"></i></div></div>'
        )
        for name, value in categories.items()
    )
    gate_rows = "".join(
        (
            '<li><i class="dot '
            f"{'ready' if bool(_as_mapping(item).get('ready', False)) else 'pending'}"
            '"></i><span>'
            f"{escape(_as_text(_as_mapping(item).get('label')))}</span><strong>"
            f"{'Ready' if bool(_as_mapping(item).get('ready', False)) else 'Pending'}"
            "</strong></li>"
        )
        for item in readiness.values()
    )
    warning = payload.get("snapshot_warning")
    warning_html = (
        f'<div class="notice">{escape(str(warning))}</div>' if isinstance(warning, str) else ""
    )
    identity_badge = (
        _badge(
            not bool(noco.get("duplicate_row_excess_detected", False)),
            "No duplicate excess",
            "Duplicate excess detected",
        )
        if comparison_completed
        else _badge(False, "Complete", "Comparison pending")
    )

    html = f"""<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<meta name="color-scheme" content="dark">
<title>OSM Lead Generator</title>
<style>
:root{{--bg:#07111f;--panel:#102038;--line:#233b5a;--text:#f5f7fb;--muted:#9fb0c5;--green:#72e0bc;--blue:#65adff;--amber:#ffd37a}}
*{{box-sizing:border-box}}body{{margin:0;background:radial-gradient(circle at top right,#123153 0,transparent 36rem),var(--bg);color:var(--text);font:15px/1.5 system-ui,-apple-system,"Segoe UI",sans-serif}}
main{{width:min(1180px,calc(100% - 28px));margin:auto;padding:30px 0 50px}}header{{display:flex;justify-content:space-between;gap:20px;align-items:flex-start;margin-bottom:22px}}
.eyebrow{{color:var(--green);font-size:12px;font-weight:800;letter-spacing:.16em;text-transform:uppercase}}h1{{font-size:clamp(30px,5vw,48px);line-height:1.05;letter-spacing:-.04em;margin:8px 0}}
p{{color:var(--muted)}}.badges{{display:flex;gap:8px;flex-wrap:wrap;justify-content:flex-end}}.badge{{display:inline-block;border:1px solid var(--line);border-radius:999px;padding:7px 10px;font-size:12px;font-weight:800;white-space:nowrap}}.good{{color:var(--green);background:#72e0bc18}}.muted{{color:#d2dce8;background:#9fb0c512}}
.grid{{display:grid;grid-template-columns:repeat(12,1fr);gap:16px}}.panel{{background:linear-gradient(180deg,#11243a,#0d1b2d);border:1px solid var(--line);border-radius:18px;padding:20px;box-shadow:0 18px 55px #0005}}.span12{{grid-column:span 12}}.span7{{grid-column:span 7}}.span5{{grid-column:span 5}}
.metrics{{display:grid;grid-template-columns:repeat(4,1fr);gap:12px;margin-top:18px}}.metric{{background:#071321aa;border:1px solid var(--line);border-radius:14px;padding:17px}}.metric span,.detail span{{display:block;color:var(--muted);font-size:12px;text-transform:uppercase;letter-spacing:.07em}}.metric strong{{display:block;font-size:30px;margin-top:7px}}
h2{{font-size:17px;margin:0}}.hint{{font-size:13px;margin:4px 0 16px}}.category{{margin:14px 0}}.category>div:first-child{{display:flex;justify-content:space-between}}.category span{{color:var(--muted)}}.bar{{height:8px;background:#071321;border-radius:999px;overflow:hidden;margin-top:7px}}.bar i{{display:block;height:100%;background:linear-gradient(90deg,var(--blue),var(--green))}}
.gates{{list-style:none;margin:0;padding:0}}.gates li{{display:grid;grid-template-columns:12px 1fr auto;gap:10px;align-items:center;padding:10px 0;border-bottom:1px solid #233b5abb}}.gates li:last-child{{border:0}}.gates strong{{font-size:12px;color:var(--muted)}}.dot{{width:9px;height:9px;border-radius:50%}}.dot.ready{{background:var(--green)}}.dot.pending{{background:var(--amber)}}
.details{{display:grid;grid-template-columns:repeat(3,1fr);gap:12px}}.detail{{background:#07132199;border-radius:12px;padding:13px}}.detail strong{{display:block;margin-top:4px;word-break:break-word}}.notice{{padding:11px 13px;border:1px solid #ffd37a55;background:#ffd37a18;color:#ffe4aa;border-radius:11px;margin-bottom:16px}}.next{{border-left:3px solid var(--blue);padding-left:14px;color:#e3edf8}}
footer{{color:var(--muted);font-size:12px;margin-top:20px}}a{{color:#91c5ff}}
@media(max-width:860px){{header{{display:block}}.badges{{justify-content:flex-start;margin-top:14px}}.span7,.span5{{grid-column:span 12}}.metrics{{grid-template-columns:repeat(2,1fr)}}}}
@media(max-width:560px){{main{{width:calc(100% - 18px);padding-top:18px}}.panel{{padding:16px}}.metrics,.details{{grid-template-columns:1fr}}}}
</style>
</head>
<body><main>
<header><div><div class="eyebrow">OpenStreetMap Lead Generator</div><h1>Live business candidate generation.</h1>
<p>Geofabrik data is downloaded, filtered and normalized automatically. Noco writes remain disabled until the comparison and controlled pilot gates are complete.</p></div>
<div class="badges">{_badge(True, "Healthy", "Unavailable")}{_badge(bool(service.get("scheduler_enabled", False)), "Monthly worker enabled", "Worker disabled")}{_badge(bool(service.get("read_only", True)), "Read only", "Write mode")}</div></header>
{warning_html}
<div class="grid">
<section class="panel span12"><div class="eyebrow">Latest candidate run · {escape(_as_text(snapshot.get("region")))}</div>
<div class="metrics">
<div class="metric"><span>Active candidates</span><strong>{_number(counts.get("active"))}</strong></div>
<div class="metric"><span>Existing Noco matches</span><strong>{exact_matches}</strong></div>
<div class="metric"><span>Unmatched candidates</span><strong>{unmatched}</strong></div>
<div class="metric"><span>Review items</span><strong>{_number(counts.get("review"))}</strong></div>
</div></section>
<section class="panel span12"><div class="eyebrow">English-language market coverage · {escape(_as_text(run.get("status")))}</div>
<div class="metrics">
<div class="metric"><span>Configured countries</span><strong>{_number(coverage.get("configured_markets"))}</strong></div>
<div class="metric"><span>Completed countries</span><strong>{_number(coverage.get("completed_markets"))}</strong></div>
<div class="metric"><span>Completed shards</span><strong>{_number(coverage.get("completed_shards"))} / {_number(coverage.get("configured_shards"))}</strong></div>
<div class="metric"><span>Failed shards</span><strong>{_number(coverage.get("failed_shards"))}</strong></div>
</div></section>
<section class="panel span7"><h2>Business categories</h2><p class="hint">Normalized active candidates from the latest OpenStreetMap extract.</p>{category_rows}</section>
<section class="panel span5"><h2>Production readiness</h2><p class="hint">Only evidenced stages are marked ready.</p><ul class="gates">{gate_rows}</ul></section>
<section class="panel span7"><h2>Run evidence</h2><p class="hint">Current source and report identity.</p><div class="details">
<div class="detail"><span>Completed</span><strong>{escape(_as_text(run.get("completed_at"), _as_text(snapshot.get("generated_at"))))}</strong></div>
<div class="detail"><span>Profile</span><strong>{escape(_as_text(snapshot.get("profile")))} · {escape(_as_text(snapshot.get("profile_version")))}</strong></div>
<div class="detail"><span>Cache</span><strong>{"Hit" if bool(run.get("cache_hit", False)) else "Downloaded"}</strong></div>
<div class="detail"><span>Report hash</span><strong>{escape(_as_text(snapshot.get("report_hash")))}</strong></div>
<div class="detail"><span>Source SHA-256</span><strong>{escape(_as_text(snapshot.get("source_sha256")))}</strong></div>
<div class="detail"><span>Identity comparison</span><strong>{identity_badge}</strong></div>
</div></section>
<section class="panel span5"><h2>Runtime safeguards</h2><p class="hint">Secret-free container state.</p><div class="details">
<div class="detail"><span>Version</span><strong>{escape(_as_text(service.get("version")))}</strong></div>
<div class="detail"><span>Environment</span><strong>{escape(_as_text(service.get("environment")))}</strong></div>
<div class="detail"><span>Scheduler</span><strong>{"Enabled" if bool(service.get("scheduler_enabled", False)) else "Disabled"}</strong></div>
<div class="detail"><span>Noco writes</span><strong>{"Enabled" if bool(service.get("noco_write_enabled", False)) else "Disabled"}</strong></div>
<div class="detail"><span>Stale deletion</span><strong>{"Enabled" if bool(service.get("stale_deletion_enabled", False)) else "Disabled"}</strong></div>
<div class="detail"><span>Sidecar</span><strong>{"Configured" if bool(service.get("sidecar_configured", False)) else "Not configured"}</strong></div>
</div></section>
<section class="panel span12"><h2>Next production gate</h2><p class="next">{escape(_as_text(payload.get("next_gate")))}</p></section>
</div>
<footer>Read-only APIs: <a href="/status">/status</a> · <a href="/api/dashboard">/api/dashboard</a> · <a href="/healthz">/healthz</a></footer>
</main></body></html>
"""
    return html.encode("utf-8")


__all__ = [
    "build_dashboard_payload",
    "load_dashboard_snapshot",
    "render_dashboard",
]