#!/usr/bin/env python3
"""Probe public routes for raw 5xx exposure without a maintenance fallback."""

from __future__ import annotations

import argparse
import json
import socket
import sys
import urllib.error
import urllib.request
from datetime import datetime
from pathlib import Path
from typing import Any


SCHEMA_VERSION = "awoooi_public_maintenance_fallback_probe_v1"
DEFAULT_URLS = [
    "https://awoooi.wooo.work/api/v1/health",
    "https://awoooi.wooo.work/",
    "https://stock.wooo.work/api/v1/system/freshness",
    "https://mo.wooo.work/health",
    "https://bitan.wooo.work/",
    "https://www.tsenyang.com/",
]
MAINTENANCE_MARKERS = [
    "AWOOOI 維護模式",
    "系統正在恢復服務",
    "HTTP 502 / 503 / 504 fallback page",
]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Read-only public route verifier for maintenance fallback.",
    )
    parser.add_argument(
        "--url",
        action="append",
        dest="urls",
        help="Public URL to probe. May be passed more than once.",
    )
    parser.add_argument("--timeout", type=float, default=8.0)
    parser.add_argument("--max-bytes", type=int, default=65536)
    parser.add_argument("--generated-at", help="Override generated_at for snapshots.")
    parser.add_argument("--output", type=Path, help="Write JSON report to this path.")
    parser.add_argument(
        "--prometheus-output",
        type=Path,
        help="Optionally write node_exporter textfile metrics.",
    )
    return parser.parse_args()


def fetch_url(url: str, *, timeout: float, max_bytes: int) -> dict[str, Any]:
    request = urllib.request.Request(
        url,
        headers={
            "User-Agent": "awoooi-reboot-public-maintenance-probe/1.0",
            "Accept": "text/html,application/json;q=0.9,*/*;q=0.1",
        },
    )
    status = 0
    headers: dict[str, str] = {}
    body = b""
    error = ""
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            status = int(response.getcode() or 0)
            headers = {key.lower(): value for key, value in response.headers.items()}
            body = response.read(max_bytes)
    except urllib.error.HTTPError as exc:
        status = int(exc.code or 0)
        headers = {key.lower(): value for key, value in exc.headers.items()}
        try:
            body = exc.read(max_bytes)
        except OSError:
            body = b""
    except (urllib.error.URLError, TimeoutError, socket.timeout) as exc:
        error = str(getattr(exc, "reason", exc))

    text = body.decode("utf-8", errors="replace")
    fallback_header = headers.get("x-awoooi-fallback", "")
    maintenance_body_detected = any(marker in text for marker in MAINTENANCE_MARKERS)
    maintenance_fallback_serving = (
        status in {200, 503}
        and fallback_header == "local-maintenance"
        and maintenance_body_detected
    )
    raw_5xx_without_fallback = (
        status >= 500
        and not maintenance_fallback_serving
    )
    route_unreachable_without_external_fallback = status == 0
    if maintenance_fallback_serving:
        route_status = "maintenance_fallback_serving"
    elif raw_5xx_without_fallback:
        route_status = "raw_5xx_without_maintenance_fallback"
    elif route_unreachable_without_external_fallback:
        route_status = "route_unreachable_without_external_fallback"
    elif 200 <= status < 500:
        route_status = "normal_origin_serving"
    else:
        route_status = "unknown"

    return {
        "url": url,
        "http_status": status,
        "status": route_status,
        "fallback_header": fallback_header,
        "maintenance_body_detected": maintenance_body_detected,
        "maintenance_fallback_serving": maintenance_fallback_serving,
        "raw_5xx_without_fallback": raw_5xx_without_fallback,
        "route_unreachable_without_external_fallback": (
            route_unreachable_without_external_fallback
        ),
        "error": error[:300],
    }


def build_report(args: argparse.Namespace) -> dict[str, Any]:
    generated_at = args.generated_at or datetime.now().astimezone().isoformat(
        timespec="seconds"
    )
    urls = args.urls or DEFAULT_URLS
    routes = [
        fetch_url(url, timeout=args.timeout, max_bytes=args.max_bytes) for url in urls
    ]
    raw_5xx_without_fallback = [
        row["url"] for row in routes if row["raw_5xx_without_fallback"]
    ]
    unreachable_without_external_fallback = [
        row["url"]
        for row in routes
        if row["route_unreachable_without_external_fallback"]
    ]
    maintenance_fallback_routes = [
        row["url"] for row in routes if row["maintenance_fallback_serving"]
    ]
    blockers: list[str] = []
    if raw_5xx_without_fallback:
        blockers.append("public_route_raw_5xx_without_maintenance_fallback")
    if unreachable_without_external_fallback:
        blockers.append("public_route_unreachable_without_external_l1_fallback")
    return {
        "schema_version": SCHEMA_VERSION,
        "generated_at": generated_at,
        "runtime_readback_present": True,
        "ready": not blockers,
        "target_url_count": len(urls),
        "route_count": len(routes),
        "raw_5xx_without_fallback_count": len(raw_5xx_without_fallback),
        "route_unreachable_without_external_fallback_count": len(
            unreachable_without_external_fallback
        ),
        "maintenance_fallback_route_count": len(maintenance_fallback_routes),
        "raw_5xx_without_fallback_urls": raw_5xx_without_fallback,
        "route_unreachable_without_external_fallback_urls": (
            unreachable_without_external_fallback
        ),
        "maintenance_fallback_urls": maintenance_fallback_routes,
        "blockers": blockers,
        "routes": routes,
        "forbidden_actions": [
            "nginx_reload_from_probe",
            "dns_cutover_from_probe",
            "service_restart_from_probe",
            "host_reboot_from_probe",
            "secret_value_read",
        ],
    }


def write_prometheus(path: Path, report: dict[str, Any]) -> None:
    ready = 1 if report.get("ready") is True else 0
    raw_5xx = int(report.get("raw_5xx_without_fallback_count") or 0)
    unreachable = int(
        report.get("route_unreachable_without_external_fallback_count") or 0
    )
    text = "\n".join(
        [
            "# HELP awoooi_public_maintenance_fallback_ready Whether public routes avoid raw 5xx exposure.",
            "# TYPE awoooi_public_maintenance_fallback_ready gauge",
            f'awoooi_public_maintenance_fallback_ready{{scope="public_routes"}} {ready}',
            "# HELP awoooi_public_route_raw_5xx_without_maintenance_fallback Raw public 5xx responses without maintenance fallback.",
            "# TYPE awoooi_public_route_raw_5xx_without_maintenance_fallback gauge",
            f'awoooi_public_route_raw_5xx_without_maintenance_fallback{{scope="public_routes"}} {raw_5xx}',
            "# HELP awoooi_public_route_unreachable_without_external_l1_fallback Public routes unreachable without external fallback proof.",
            "# TYPE awoooi_public_route_unreachable_without_external_l1_fallback gauge",
            f'awoooi_public_route_unreachable_without_external_l1_fallback{{scope="public_routes"}} {unreachable}',
            "",
        ]
    )
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text, encoding="utf-8")


def main() -> int:
    args = parse_args()
    report = build_report(args)
    text = json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(text, encoding="utf-8")
    else:
        print(text, end="")
    if args.prometheus_output:
        write_prometheus(args.prometheus_output, report)
    return 0


if __name__ == "__main__":
    sys.exit(main())
