"""Framework-free ARR portal application with fixed, auditable routes."""

from __future__ import annotations

import base64
import json
import re
import secrets
import threading
import time
from dataclasses import dataclass
from datetime import date, datetime
from http.cookies import SimpleCookie
from pathlib import Path
from typing import Any, Dict, Mapping, Optional
from urllib.parse import parse_qs, urlsplit
from urllib.parse import quote
from zoneinfo import ZoneInfo

from arr_web.auth import LoginAttemptLedger, LoginCredentials
from arr_web.booking_uploads import (
    BookingSourceCoordinator,
    UnavailableBookingSourceCoordinator,
)
from arr_web.contracts import (
    MAX_UPLOAD_BYTES,
    PortalError,
    Response,
    failure,
    success,
    validate_job_id,
    validate_month,
    validate_upload_filename,
    validate_xml_payload,
)
from arr_web.repository import (
    PortalDataError,
    PortalRepository,
    UnavailablePortalRepository,
)
from arr_web.downloads import ArtifactReader, UnavailableArtifactReader
from arr_web.company_jobs import (
    COMPANY_NAMES,
    CompanyReportCoordinator,
    UnavailableCompanyReportCoordinator,
)
from arr_web.services import (
    DailyPriceReviewCoordinator,
    MonthlyCoordinator,
    UnavailableDailyPriceReviewCoordinator,
    UnavailableMonthlyCoordinator,
    UnavailableUploadCoordinator,
    UploadCoordinator,
)


STATIC_ROOT = Path(__file__).resolve().parent / "static"
STATIC_ROUTES = {
    "/": ("index.html", "text/html; charset=utf-8"),
    "/index.html": ("index.html", "text/html; charset=utf-8"),
    "/h5": ("h5.html", "text/html; charset=utf-8"),
    "/h5.html": ("h5.html", "text/html; charset=utf-8"),
    "/assets/styles.css": ("styles.css", "text/css; charset=utf-8"),
    "/assets/i18n.js": ("i18n.js", "text/javascript; charset=utf-8"),
    "/assets/app.js": ("app.js", "text/javascript; charset=utf-8"),
    "/assets/h5.css": ("h5.css", "text/css; charset=utf-8"),
    "/assets/h5.js": ("h5.js", "text/javascript; charset=utf-8"),
    "/login": ("login.html", "text/html; charset=utf-8"),
    "/login.html": ("login.html", "text/html; charset=utf-8"),
    "/assets/login.css": ("login.css", "text/css; charset=utf-8"),
    "/assets/login.js": ("login.js", "text/javascript; charset=utf-8"),
}
LOGIN_STATIC_ROUTES = {
    "/login",
    "/login.html",
    "/assets/login.css",
    "/assets/i18n.js",
    "/assets/login.js",
}
LOGIN_DOCUMENT_ROUTES = {"/login", "/login.html"}
H5_DOCUMENT_ROUTES = {"/h5", "/h5.html"}
PUBLIC_H5_STATIC_ROUTES = H5_DOCUMENT_ROUTES | {
    "/assets/h5.css",
    "/assets/h5.js",
}


_PUBLIC_H5_DASHBOARD_KEYS = (
    "version",
    "month_key",
    "updated_at",
    "min_arrival_date",
    "max_arrival_date",
    "overall",
    "channels",
)


def _paged_success(
    data: list[Dict[str, Any]],
    *,
    total: int,
    limit: int,
    offset: int,
) -> Dict[str, Any]:
    return success(
        data,
        pagination={
            "total": total,
            "limit": limit,
            "offset": offset,
            "has_previous": offset > 0,
            "has_next": offset + len(data) < total,
        },
    )


def _public_h5_dashboard(payload: Mapping[str, Any]) -> Dict[str, Any]:
    return {
        key: payload[key]
        for key in _PUBLIC_H5_DASHBOARD_KEYS
        if key in payload
    }


def _strict_json(raw: bytes) -> Mapping[str, Any]:
    def pairs(values: list[tuple[str, Any]]) -> Dict[str, Any]:
        output: Dict[str, Any] = {}
        for key, value in values:
            if key in output:
                raise ValueError("duplicate key")
            output[key] = value
        return output

    try:
        value = json.loads(raw.decode("utf-8"), object_pairs_hook=pairs)
    except (UnicodeDecodeError, json.JSONDecodeError, ValueError):
        raise PortalError("REQUEST_JSON_INVALID", "请求内容不是有效 JSON") from None
    if not isinstance(value, Mapping):
        raise PortalError("REQUEST_JSON_INVALID", "请求内容不是有效 JSON")
    return value


def _decode_filename_header(
    encoded: str,
    *,
    code: str,
    safe_message: str,
) -> str:
    try:
        padding = "=" * (-len(encoded) % 4)
        return base64.urlsafe_b64decode(encoded + padding).decode("utf-8")
    except (ValueError, UnicodeDecodeError):
        raise PortalError(code, safe_message) from None


@dataclass(frozen=True)
class RuntimeHealth:
    database_ready: bool
    upload_ready: bool
    monthly_ready: bool
    download_ready: bool = False
    company_reports_ready: bool = False
    company_source_upload_ready: bool = False


@dataclass(frozen=True)
class _SessionRecord:
    username: str
    csrf: str
    expires_at: float


class SessionLedger:
    """Small authenticated-session ledger for the single-process Web runtime."""

    def __init__(self, lifetime_seconds: int = 8 * 60 * 60) -> None:
        self._lifetime = lifetime_seconds
        self._lock = threading.Lock()
        self._sessions: Dict[str, _SessionRecord] = {}

    @property
    def lifetime_seconds(self) -> int:
        return self._lifetime

    def issue(self, username: str) -> tuple[str, str]:
        session_id = secrets.token_urlsafe(24)
        csrf = secrets.token_urlsafe(24)
        now = time.time()
        with self._lock:
            self._sessions[session_id] = _SessionRecord(
                username=username,
                csrf=csrf,
                expires_at=now + self._lifetime,
            )
            if len(self._sessions) > 2048:
                self._sessions = {
                    key: value
                    for key, value in self._sessions.items()
                    if value.expires_at > now
                }
                if len(self._sessions) > 2048:
                    oldest = sorted(
                        self._sessions,
                        key=lambda key: self._sessions[key].expires_at,
                    )
                    for key in oldest[: len(self._sessions) - 2048]:
                        self._sessions.pop(key, None)
        return session_id, csrf

    def current(self, session_id: Optional[str]) -> Optional[tuple[str, str]]:
        if not session_id:
            return None
        now = time.time()
        with self._lock:
            value = self._sessions.get(session_id)
            if value is None or value.expires_at <= now:
                self._sessions.pop(session_id, None)
                return None
            return value.username, value.csrf

    def verify(self, session_id: Optional[str], csrf: Optional[str]) -> bool:
        if not session_id or not csrf:
            return False
        current = self.current(session_id)
        return current is not None and secrets.compare_digest(current[1], csrf)

    def revoke(self, session_id: Optional[str]) -> None:
        if not session_id:
            return
        with self._lock:
            self._sessions.pop(session_id, None)


class PortalApplication:
    def __init__(
        self,
        repository: Optional[PortalRepository] = None,
        upload: Optional[UploadCoordinator] = None,
        price_reviews: Optional[DailyPriceReviewCoordinator] = None,
        monthly: Optional[MonthlyCoordinator] = None,
        company_reports: Optional[CompanyReportCoordinator] = None,
        booking_sources: Optional[BookingSourceCoordinator] = None,
        artifact_reader: Optional[ArtifactReader] = None,
        *,
        health: Optional[RuntimeHealth] = None,
        static_root: Path = STATIC_ROOT,
        sessions: Optional[SessionLedger] = None,
        credentials: Optional[LoginCredentials] = None,
        login_attempts: Optional[LoginAttemptLedger] = None,
        secure_cookies: bool = False,
    ) -> None:
        self._repository = repository or UnavailablePortalRepository()
        self._upload = upload or UnavailableUploadCoordinator()
        self._price_reviews = price_reviews or UnavailableDailyPriceReviewCoordinator()
        self._monthly = monthly or UnavailableMonthlyCoordinator()
        self._company_reports = (
            company_reports or UnavailableCompanyReportCoordinator()
        )
        self._booking_sources = (
            booking_sources or UnavailableBookingSourceCoordinator()
        )
        self._artifact_reader = artifact_reader or UnavailableArtifactReader()
        self._health = health or RuntimeHealth(False, False, False)
        self._static_root = static_root.resolve()
        self._sessions = sessions or SessionLedger()
        self._credentials = credentials
        self._login_attempts = login_attempts or LoginAttemptLedger()
        self._secure_cookies = secure_cookies

    def handle(
        self,
        method: str,
        target: str,
        headers: Mapping[str, str],
        body: bytes = b"",
        client_id: str = "direct",
    ) -> Response:
        normalized_headers = {key.lower(): value for key, value in headers.items()}
        route = urlsplit(target)
        try:
            session_id = self._session_id(normalized_headers)
            current_session = self._sessions.current(session_id)
            if method == "GET" and route.path == "/healthz":
                return self._readiness()
            if method == "GET" and route.path in LOGIN_STATIC_ROUTES:
                if route.path in LOGIN_DOCUMENT_ROUTES and current_session is not None:
                    return self._redirect(self._safe_next(route.query))
                return self._static(route.path)
            if method == "POST" and route.path == "/api/login":
                return self._login(body, normalized_headers, client_id)
            if method == "GET" and route.path in PUBLIC_H5_STATIC_ROUTES:
                return self._static(route.path)
            if method == "GET" and route.path == "/api/public/h5/months":
                return Response.json(
                    200,
                    success(
                        [
                            {
                                "month_key": item.get("month_key"),
                                "updated_at": item.get("updated_at"),
                                "max_arrival_date": item.get("max_arrival_date"),
                            }
                            for item in self._repository.list_months()
                        ]
                    ),
                )
            if method == "GET" and route.path == "/api/public/h5/analytics":
                query = parse_qs(route.query, keep_blank_values=True)
                month = validate_month(
                    self._one(query, "month", self._current_month())
                )
                return Response.json(
                    200,
                    success(_public_h5_dashboard(self._dashboard_payload(month))),
                )
            if current_session is None:
                return self._authentication_required(method, route.path)
            if method == "GET" and route.path in STATIC_ROUTES:
                return self._static(route.path)
            if method == "GET" and route.path == "/api/session":
                return Response.json(
                    200,
                    success(
                        {
                            "username": current_session[0],
                            "csrf_token": current_session[1],
                            "max_upload_bytes": MAX_UPLOAD_BYTES,
                            "retention": "month_minimum",
                        }
                    ),
                )
            if method == "POST" and route.path == "/api/logout":
                self._require_csrf(normalized_headers)
                self._sessions.revoke(session_id)
                return Response.json(
                    200,
                    success({"logged_out": True}),
                    {"Set-Cookie": self._session_cookie("", max_age=0)},
                )
            if method == "GET" and route.path == "/api/health":
                return Response.json(
                    200,
                    success(
                        {
                            "database_ready": self._health.database_ready,
                            "processing_ready": self._health.upload_ready,
                            "monthly_ready": self._health.monthly_ready,
                            "download_ready": self._health.download_ready,
                            "company_reports_ready": self._health.company_reports_ready,
                            "company_source_upload_ready": self._health.company_source_upload_ready,
                            "business_time_zone": "Asia/Bangkok",
                            "checked_at": datetime.now(ZoneInfo("Asia/Bangkok")).isoformat(),
                        }
                    ),
                )
            if method == "GET" and route.path == "/api/jobs":
                query = parse_qs(route.query, keep_blank_values=True)
                month = validate_month(self._one(query, "month", self._current_month()))
                limit = self._integer(query, "limit", 50, 1, 200)
                offset = self._integer(query, "offset", 0, 0, 10_000_000)
                jobs, total = self._repository.list_jobs(month, limit, offset)
                return Response.json(
                    200,
                    _paged_success(
                        jobs,
                        total=total,
                        limit=limit,
                        offset=offset,
                    ),
                )
            job_parts = route.path.strip("/").split("/")
            if (
                method == "GET"
                and len(job_parts) == 4
                and job_parts[:2] == ["api", "jobs"]
                and job_parts[3] == "trace"
            ):
                job_id = validate_job_id(job_parts[2])
                return Response.json(
                    200,
                    success(self._repository.get_job_trace(job_id)),
                )
            if (
                method == "GET"
                and len(job_parts) == 4
                and job_parts[:2] == ["api", "jobs"]
                and job_parts[3] == "review"
            ):
                job_id = validate_job_id(job_parts[2])
                query = parse_qs(route.query, keep_blank_values=True)
                limit = self._integer(query, "limit", 50, 1, 200)
                offset = self._integer(query, "offset", 0, 0, 10_000_000)
                return Response.json(
                    200,
                    success(self._price_reviews.get_price_review(job_id, limit, offset)),
                )
            if method == "GET" and route.path == "/api/monthly-runs":
                query = parse_qs(route.query, keep_blank_values=True)
                month = validate_month(self._one(query, "month", self._current_month()))
                limit = self._integer(query, "limit", 50, 1, 200)
                offset = self._integer(query, "offset", 0, 0, 10_000_000)
                runs, total = self._repository.list_monthly_runs(
                    month,
                    limit,
                    offset,
                )
                return Response.json(
                    200,
                    _paged_success(
                        runs,
                        total=total,
                        limit=limit,
                        offset=offset,
                    ),
                )
            if method == "GET" and route.path == "/api/history-months":
                combined: Dict[str, Dict[str, Any]] = {}
                for item in self._repository.list_history_month_counts():
                    month_key = str(item.get("month_key", ""))
                    validate_month(month_key)
                    combined[month_key] = {
                        "month_key": month_key,
                        "daily_count": int(item.get("daily_count", 0) or 0),
                        "monthly_count": int(item.get("monthly_count", 0) or 0),
                        "company_count": 0,
                    }
                for item in self._company_reports.list_month_counts():
                    month_key = str(item.get("month_key", ""))
                    validate_month(month_key)
                    row = combined.setdefault(
                        month_key,
                        {
                            "month_key": month_key,
                            "daily_count": 0,
                            "monthly_count": 0,
                            "company_count": 0,
                        },
                    )
                    row["company_count"] = int(item.get("company_count", 0) or 0)
                return Response.json(
                    200,
                    success([combined[key] for key in sorted(combined, reverse=True)]),
                )
            if method == "GET" and route.path == "/api/months":
                return Response.json(200, success(self._repository.list_months()))
            if method == "GET" and route.path == "/api/h5/months":
                return Response.json(
                    200,
                    {
                        "months": [
                            {
                                "month_key": item.get("month_key"),
                                "updated_at": item.get("updated_at"),
                                "max_arrival_date": item.get("max_arrival_date"),
                            }
                            for item in self._repository.list_months()
                        ]
                    },
                )
            if method == "GET" and route.path == "/api/analytics":
                query = parse_qs(route.query, keep_blank_values=True)
                month = validate_month(self._one(query, "month", self._current_month()))
                return Response.json(200, success(self._repository.read_dashboard(month)))
            analytics_parts = route.path.strip("/").split("/")
            if (
                method == "GET"
                and len(analytics_parts) == 4
                and analytics_parts[:2] == ["api", "monthly"]
                and analytics_parts[3] == "analytics"
            ):
                try:
                    month = validate_month(analytics_parts[2])
                except PortalError as error:
                    raise PortalError(
                        "MONTH_KEY_INVALID",
                        error.safe_message,
                        422,
                    ) from None
                return Response.json(200, self._dashboard_payload(month))
            if method == "GET" and route.path == "/api/channel-detail":
                query = parse_qs(route.query, keep_blank_values=True)
                month = validate_month(self._one(query, "month", self._current_month()))
                worksheet = self._one(query, "worksheet", "")
                if not worksheet or len(worksheet) > 31:
                    raise PortalError("CHANNEL_INVALID", "渠道名称无效")
                limit = self._integer(query, "limit", 100, 1, 500)
                offset = self._integer(query, "offset", 0, 0, 10_000_000)
                return Response.json(
                    200,
                    success(
                        self._repository.read_channel_detail(
                            month,
                            worksheet,
                            limit,
                            offset,
                        )
                    ),
                )
            if method == "GET" and route.path == "/api/company-reports/jobs":
                query = parse_qs(route.query, keep_blank_values=True)
                limit = self._integer(query, "limit", 50, 1, 200)
                offset = self._integer(query, "offset", 0, 0, 10_000_000)
                report_month = None
                if "month" in query:
                    report_month = validate_month(self._one(query, "month", ""))
                jobs, total = self._company_reports.list_jobs(
                    report_month,
                    limit,
                    offset,
                )
                return Response.json(
                    200,
                    _paged_success(
                        jobs,
                        total=total,
                        limit=limit,
                        offset=offset,
                    ),
                )
            if method == "GET" and route.path == "/api/company-reports/source":
                return Response.json(200, success(self._booking_sources.current()))
            if method == "GET" and route.path == "/api/company-reports/source/draft":
                query = parse_qs(route.query, keep_blank_values=True)
                limit = self._integer(query, "limit", 50, 1, 200)
                offset = self._integer(query, "offset", 0, 0, 10_000_000)
                return Response.json(
                    200,
                    success(self._booking_sources.draft(limit, offset)),
                )
            company_parts = route.path.strip("/").split("/")
            if (
                method == "GET"
                and len(company_parts) == 6
                and company_parts[:3] == ["api", "company-reports", "jobs"]
                and company_parts[4] == "downloads"
            ):
                job_id = company_parts[3]
                company = company_parts[5]
                if company not in COMPANY_NAMES:
                    raise PortalError(
                        "COMPANY_REPORT_COMPANY_INVALID", "公司名称无效"
                    )
                return self._download(
                    self._company_reports.resolve_download(job_id, company)
                )
            if (
                method == "GET"
                and len(company_parts) == 4
                and company_parts[:3] == ["api", "company-reports", "jobs"]
            ):
                return Response.json(
                    200,
                    success(self._company_reports.get_job(company_parts[3])),
                )
            if method == "GET" and route.path == "/api/download/daily":
                query = parse_qs(route.query, keep_blank_values=True)
                job_id = self._one(query, "job_id", "")
                if not job_id:
                    raise PortalError("DOWNLOAD_REQUEST_INVALID", "下载请求无效")
                return self._download(self._repository.resolve_daily_download(job_id))
            if method == "GET" and route.path == "/api/download/monthly":
                query = parse_qs(route.query, keep_blank_values=True)
                report_id = self._integer(query, "report_id", 0, 1, 9_223_372_036_854_775_807)
                return self._download(self._repository.resolve_monthly_download(report_id))
            if method == "POST" and route.path == "/api/jobs":
                self._require_csrf(normalized_headers)
                filename = _decode_filename_header(
                    normalized_headers.get("x-arr-filename-b64", ""),
                    code="UPLOAD_FILENAME_INVALID",
                    safe_message="请选择 XML 文件",
                )
                filename = validate_upload_filename(filename)
                validate_xml_payload(body)
                return Response.json(202, success(self._upload.submit(filename, body)))
            if (
                method == "PATCH"
                and len(job_parts) == 6
                and job_parts[:2] == ["api", "jobs"]
                and job_parts[3:5] == ["review", "items"]
            ):
                self._require_csrf(normalized_headers)
                job_id = validate_job_id(job_parts[2])
                try:
                    item_id = int(job_parts[5])
                except ValueError:
                    raise PortalError("REVIEW_ITEM_NOT_FOUND", "待人工处理项不存在", 404) from None
                if item_id < 1 or len(body) > 8192:
                    raise PortalError("REVIEW_REQUEST_INVALID", "复核请求字段无效")
                payload = _strict_json(body)
                if set(payload) != {"case_id", "revision", "real_price"}:
                    raise PortalError("REVIEW_REQUEST_INVALID", "复核请求字段无效")
                case_id = payload.get("case_id")
                revision = payload.get("revision")
                real_price = payload.get("real_price")
                if (
                    not self._valid_review_case_id(case_id)
                    or not isinstance(revision, int)
                    or isinstance(revision, bool)
                    or revision < 0
                    or not isinstance(real_price, str)
                    or re.fullmatch(r"(?:0|[1-9][0-9]{0,15})", real_price) is None
                ):
                    raise PortalError("REVIEW_REQUEST_INVALID", "复核请求字段无效")
                return Response.json(
                    200,
                    success(
                        self._price_reviews.update_price_review_item(
                            job_id,
                            item_id,
                            case_id,
                            revision,
                            real_price,
                            current_session[0],
                        )
                    ),
                )
            if (
                method in {"POST"}
                and len(job_parts) == 5
                and job_parts[:2] == ["api", "jobs"]
                and job_parts[3] == "review"
                and job_parts[4] in {"finalize", "cancel"}
            ):
                self._require_csrf(normalized_headers)
                job_id = validate_job_id(job_parts[2])
                if len(body) > 8192:
                    raise PortalError("REVIEW_REQUEST_INVALID", "复核请求字段无效")
                payload = _strict_json(body)
                if set(payload) != {"case_id", "revision"}:
                    raise PortalError("REVIEW_REQUEST_INVALID", "复核请求字段无效")
                case_id = payload.get("case_id")
                revision = payload.get("revision")
                if (
                    not self._valid_review_case_id(case_id)
                    or not isinstance(revision, int)
                    or isinstance(revision, bool)
                    or revision < 0
                ):
                    raise PortalError("REVIEW_REQUEST_INVALID", "复核请求字段无效")
                if job_parts[4] == "finalize":
                    result = self._price_reviews.finalize_price_review(
                        job_id, case_id, revision, current_session[0]
                    )
                else:
                    result = self._price_reviews.cancel_price_review(
                        job_id, case_id, revision, current_session[0]
                    )
                return Response.json(200, success(result))
            if method == "POST" and route.path == "/api/company-reports/source":
                self._require_csrf(normalized_headers)
                filename = _decode_filename_header(
                    normalized_headers.get("x-arr-filename-b64", ""),
                    code="BOOKING_EXCEL_FILENAME_INVALID",
                    safe_message="请选择 .xlsx 格式的 Excel 文件",
                )
                return Response.json(
                    200,
                    success(self._booking_sources.submit(filename, body)),
                )
            if method == "DELETE" and route.path == "/api/company-reports/source/draft/items":
                self._require_csrf(normalized_headers)
                if len(body) > 8192:
                    raise PortalError("REQUEST_TOO_LARGE", "请求内容过大", 413)
                payload = _strict_json(body)
                if set(payload) != {"draft_id", "item_ids"}:
                    raise PortalError(
                        "BOOKING_EXCEL_REVIEW_REQUEST_INVALID",
                        "批量删除记录请求无效",
                    )
                return Response.json(
                    200,
                    success(
                        self._booking_sources.delete_items(
                            payload.get("draft_id"),
                            payload.get("item_ids"),
                        )
                    ),
                )
            if (
                method in {"PATCH", "DELETE"}
                and len(company_parts) == 6
                and company_parts[:5]
                == ["api", "company-reports", "source", "draft", "items"]
            ):
                self._require_csrf(normalized_headers)
                try:
                    item_id = int(company_parts[5])
                except ValueError:
                    raise PortalError(
                        "BOOKING_EXCEL_REVIEW_ITEM_INVALID",
                        "待确认记录编号无效",
                    ) from None
                if item_id <= 0:
                    raise PortalError(
                        "BOOKING_EXCEL_REVIEW_ITEM_INVALID",
                        "待确认记录编号无效",
                    )
                if len(body) > 8192:
                    raise PortalError("REQUEST_TOO_LARGE", "请求内容过大", 413)
                payload = _strict_json(body)
                if method == "PATCH":
                    if set(payload) != {"draft_id", "room_type", "quantity"}:
                        raise PortalError(
                            "BOOKING_EXCEL_REVIEW_REQUEST_INVALID",
                            "人工确认字段无效",
                        )
                    return Response.json(
                        200,
                        success(
                            self._booking_sources.update_item(
                                payload.get("draft_id"),
                                item_id,
                                payload.get("room_type"),
                                payload.get("quantity"),
                            )
                        ),
                    )
                if set(payload) != {"draft_id"}:
                    raise PortalError(
                        "BOOKING_EXCEL_REVIEW_REQUEST_INVALID",
                        "删除记录请求无效",
                    )
                return Response.json(
                    200,
                    success(
                        self._booking_sources.delete_item(
                            payload.get("draft_id"),
                            item_id,
                        )
                    ),
                )
            if method == "DELETE" and route.path == "/api/company-reports/source/draft":
                self._require_csrf(normalized_headers)
                if len(body) > 8192:
                    raise PortalError("REQUEST_TOO_LARGE", "请求内容过大", 413)
                payload = _strict_json(body)
                if set(payload) != {"draft_id"}:
                    raise PortalError(
                        "BOOKING_EXCEL_REVIEW_REQUEST_INVALID",
                        "放弃提取请求无效",
                    )
                return Response.json(
                    200,
                    success(self._booking_sources.discard(payload.get("draft_id"))),
                )
            if method == "POST" and route.path == "/api/company-reports/source/draft/activate":
                self._require_csrf(normalized_headers)
                if len(body) > 8192:
                    raise PortalError("REQUEST_TOO_LARGE", "请求内容过大", 413)
                payload = _strict_json(body)
                if set(payload) != {"draft_id"}:
                    raise PortalError(
                        "BOOKING_EXCEL_REVIEW_REQUEST_INVALID",
                        "确认启用请求无效",
                    )
                return Response.json(
                    200,
                    success(self._booking_sources.activate(payload.get("draft_id"))),
                )
            if method == "POST" and route.path == "/api/monthly-runs":
                self._require_csrf(normalized_headers)
                if len(body) > 8192:
                    raise PortalError("REQUEST_TOO_LARGE", "请求内容过大", 413)
                payload = _strict_json(body)
                if set(payload) != {"month", "as_of_date"}:
                    raise PortalError("MONTHLY_REQUEST_INVALID", "月报请求字段无效")
                month = validate_month(str(payload.get("month", "")))
                try:
                    as_of = date.fromisoformat(str(payload.get("as_of_date", "")))
                except ValueError:
                    raise PortalError("MONTHLY_REQUEST_INVALID", "月报截止日期无效") from None
                return Response.json(202, success(self._monthly.generate(month, as_of)))
            if method == "POST" and route.path == "/api/company-reports/jobs":
                self._require_csrf(normalized_headers)
                if self._booking_sources.draft(1, 0) is not None:
                    raise PortalError(
                        "BOOKING_EXCEL_REVIEW_OPEN",
                        "请先完成或放弃当前 Excel 提取草稿",
                        409,
                    )
                current_source = self._booking_sources.current()
                if current_source is None:
                    raise PortalError(
                        "BOOKING_EXCEL_SOURCE_REQUIRED",
                        "请先上传并校验 Excel 报表",
                        409,
                    )
                if len(body) > 8192:
                    raise PortalError("REQUEST_TOO_LARGE", "请求内容过大", 413)
                payload = _strict_json(body)
                if set(payload) != {"report_month", "period"}:
                    raise PortalError(
                        "COMPANY_REPORT_REQUEST_INVALID",
                        "渠道明细请求字段无效",
                    )
                report_month = payload.get("report_month")
                period = payload.get("period")
                if not isinstance(report_month, str) or not isinstance(period, str):
                    raise PortalError(
                        "COMPANY_REPORT_REQUEST_INVALID",
                        "渠道明细请求字段无效",
                    )
                return Response.json(
                    202,
                    success(
                        self._company_reports.create(
                            report_month,
                            period,
                            source=current_source,
                        )
                    ),
                )
            raise PortalError("ROUTE_NOT_FOUND", "页面或接口不存在", 404)
        except PortalDataError as error:
            status = 503
            if error.code in {"DOWNLOAD_NOT_FOUND", "JOB_NOT_FOUND"}:
                status = 404
            elif error.code == "DOWNLOAD_REQUEST_INVALID":
                status = 400
            return Response.json(
                status,
                failure(PortalError(error.code, error.safe_message, status)),
            )
        except PortalError as error:
            return Response.json(error.status, failure(error))
        except Exception:
            return Response.json(
                500,
                failure(PortalError("INTERNAL_ERROR", "服务暂时无法完成请求", 500)),
            )

    def _readiness(self) -> Response:
        ready = self._health.database_ready and self._health.upload_ready
        return Response(
            200 if ready else 503,
            b"ready\n" if ready else b"unavailable\n",
            "text/plain; charset=utf-8",
            {"Cache-Control": "no-store"},
        )

    def _dashboard_payload(self, month: str) -> Dict[str, Any]:
        try:
            return self._repository.read_dashboard(month)
        except PortalDataError as error:
            if error.code == "ANALYTICS_MONTH_NOT_FOUND":
                raise PortalError(
                    "MONTHLY_NOT_FOUND",
                    "找不到该月份的正式月报。",
                    404,
                ) from None
            raise

    def _login(
        self,
        body: bytes,
        headers: Mapping[str, str],
        client_id: str,
    ) -> Response:
        media_type = headers.get("content-type", "").split(";", 1)[0].strip().lower()
        if media_type != "application/json":
            raise PortalError("LOGIN_REQUEST_INVALID", "登录请求格式无效", 415)
        if len(body) > 8192:
            raise PortalError("REQUEST_TOO_LARGE", "请求内容过大", 413)
        retry_after = self._login_attempts.retry_after(client_id)
        if retry_after:
            return self._login_rate_limited(retry_after)
        payload = _strict_json(body)
        if set(payload) != {"username", "password"}:
            raise PortalError("LOGIN_REQUEST_INVALID", "登录请求字段无效")
        username = payload.get("username")
        password = payload.get("password")
        if self._credentials is None or not self._credentials.verify(
            username,
            password,
        ):
            retry_after = self._login_attempts.record_failure(client_id)
            if retry_after:
                return self._login_rate_limited(retry_after)
            raise PortalError("LOGIN_FAILED", "用户名或密码不正确", 401)
        assert isinstance(username, str)
        self._login_attempts.record_success(client_id)
        self._sessions.revoke(self._session_id(headers))
        session_id, csrf = self._sessions.issue(username)
        return Response.json(
            200,
            success(
                {
                    "username": username,
                    "csrf_token": csrf,
                    "expires_in": self._sessions.lifetime_seconds,
                }
            ),
            {"Set-Cookie": self._session_cookie(session_id)},
        )

    @staticmethod
    def _login_rate_limited(retry_after: int) -> Response:
        return Response.json(
            429,
            failure(
                PortalError(
                    "LOGIN_RATE_LIMITED",
                    "登录尝试过于频繁，请稍后再试",
                    429,
                )
            ),
            {"Retry-After": str(retry_after)},
        )

    def _authentication_required(self, method: str, route_path: str) -> Response:
        if method == "GET" and not route_path.startswith("/api/"):
            next_path = "/h5" if route_path in H5_DOCUMENT_ROUTES else "/"
            return self._redirect(f"/login?next={quote(next_path, safe='')}")
        return Response.json(
            401,
            failure(PortalError("AUTH_REQUIRED", "请先登录", 401)),
        )

    @staticmethod
    def _redirect(location: str) -> Response:
        return Response(
            303,
            b"",
            "text/plain; charset=utf-8",
            {"Cache-Control": "no-store", "Location": location},
        )

    @staticmethod
    def _safe_next(query: str) -> str:
        values = parse_qs(query, keep_blank_values=True).get("next", [])
        if len(values) == 1 and values[0] in {"/", "/h5"}:
            return values[0]
        return "/"

    @staticmethod
    def _session_id(headers: Mapping[str, str]) -> Optional[str]:
        cookie = SimpleCookie()
        try:
            cookie.load(headers.get("cookie", ""))
        except Exception:
            return None
        item = cookie.get("arr_session")
        return item.value if item is not None else None

    def _session_cookie(self, session_id: str, *, max_age: Optional[int] = None) -> str:
        age = self._sessions.lifetime_seconds if max_age is None else max_age
        cookie = (
            f"arr_session={session_id}; Path=/; HttpOnly; "
            f"SameSite=Strict; Max-Age={age}"
        )
        if age == 0:
            cookie += "; Expires=Thu, 01 Jan 1970 00:00:00 GMT"
        if self._secure_cookies:
            cookie += "; Secure"
        return cookie

    def _static(self, route: str) -> Response:
        filename, mime_type = STATIC_ROUTES[route]
        path = (self._static_root / filename).resolve()
        try:
            path.relative_to(self._static_root)
            body = path.read_bytes()
        except (ValueError, OSError):
            raise PortalError("STATIC_NOT_FOUND", "页面资源不存在", 404) from None
        return Response(200, body, mime_type, {"Cache-Control": "no-cache"})

    def _download(self, descriptor: Any) -> Response:
        body = self._artifact_reader.read(descriptor)
        encoded = quote(descriptor.original_filename, safe="")
        return Response(
            200,
            body,
            descriptor.mime_type,
            {
                "Cache-Control": "private, no-store",
                "Content-Disposition": (
                    'attachment; filename="arr-report.xlsx"; '
                    f"filename*=UTF-8''{encoded}"
                ),
                "X-Artifact-SHA256": descriptor.sha256,
            },
        )

    def _require_csrf(self, headers: Mapping[str, str]) -> None:
        session_id = self._session_id(headers)
        if not self._sessions.verify(session_id, headers.get("x-arr-csrf")):
            raise PortalError("SESSION_INVALID", "页面会话已失效", 403)

    @staticmethod
    def _current_month() -> str:
        return datetime.now(ZoneInfo("Asia/Bangkok")).strftime("%Y-%m")

    @staticmethod
    def _one(query: Mapping[str, list[str]], key: str, default: str) -> str:
        values = query.get(key)
        if values is None:
            return default
        if len(values) != 1:
            raise PortalError("QUERY_INVALID", "查询参数无效")
        return values[0]

    @classmethod
    def _integer(
        cls,
        query: Mapping[str, list[str]],
        key: str,
        default: int,
        minimum: int,
        maximum: int,
    ) -> int:
        raw = cls._one(query, key, str(default))
        try:
            value = int(raw)
        except ValueError:
            raise PortalError("QUERY_INVALID", "查询参数无效") from None
        if not minimum <= value <= maximum:
            raise PortalError("QUERY_INVALID", "查询参数无效")
        return value

    @staticmethod
    def _valid_review_case_id(value: Any) -> bool:
        return (
            isinstance(value, str)
            and len(value) == 44
            and value.startswith("dailyreview-")
            and all(character in "0123456789abcdef" for character in value[12:])
        )
