"""Reproducible HTTP smoke for upload -> review -> vector -> retrieval."""

from __future__ import annotations

import hashlib
import json
import os
import sys
import time
import uuid
from pathlib import Path
from typing import Any, cast
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
from urllib.request import Request, urlopen

from app.core.demo_identity import BAILIAN_KNOWLEDGE_BASE_ID, KNOWLEDGE_BASE_ID


class DocumentPipelineSmokeError(RuntimeError):
    """A safe smoke failure without source text, paths, or response bodies."""


def _request(
    base_url: str,
    method: str,
    path: str,
    *,
    body: dict[str, object] | None = None,
    content: bytes | None = None,
    headers: dict[str, str] | None = None,
) -> dict[str, Any]:
    request_headers = {"Accept": "application/json", **(headers or {})}
    payload: bytes | None = None
    if body is not None:
        payload = json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode()
        request_headers["Content-Type"] = "application/json"
    elif content is not None:
        payload = content
        request_headers["Content-Type"] = "application/octet-stream"
    request = Request(  # noqa: S310 - base URL is operator-configured HTTP(S)
        f"{base_url.rstrip('/')}{path}",
        data=payload,
        headers=request_headers,
        method=method,
    )
    try:
        with urlopen(request, timeout=15) as response:  # noqa: S310 - configured local endpoint
            parsed = json.loads(response.read())
    except HTTPError as exc:
        code = "UNKNOWN"
        try:
            problem = json.loads(exc.read())
            if isinstance(problem, dict) and isinstance(problem.get("code"), str):
                code = problem["code"]
        except (OSError, ValueError, TypeError):
            pass
        raise DocumentPipelineSmokeError(f"HTTP {exc.code} ({code}) for {method} {path}") from None
    except (URLError, TimeoutError, OSError, ValueError, TypeError):
        raise DocumentPipelineSmokeError(f"request failed for {method} {path}") from None
    if not isinstance(parsed, dict):
        raise DocumentPipelineSmokeError(f"invalid response for {method} {path}")
    return cast(dict[str, Any], parsed)


def _wait_job(base_url: str, job_id: str, *, timeout_seconds: float) -> dict[str, Any]:
    deadline = time.monotonic() + timeout_seconds
    while time.monotonic() < deadline:
        job = _request(base_url, "GET", f"/api/v1/document-jobs/{job_id}")
        status = job.get("status")
        if status == "SUCCEEDED":
            return job
        if status in {"FAILED", "CANCELLED"}:
            code = job.get("last_error_code")
            safe_code = code if isinstance(code, str) else "UNKNOWN"
            raise DocumentPipelineSmokeError(f"job terminated with {status} ({safe_code})")
        time.sleep(0.25)
    raise DocumentPipelineSmokeError("job polling timed out")


def _wait_document_ready(
    base_url: str,
    document_id: str,
    *,
    timeout_seconds: float,
) -> dict[str, Any]:
    deadline = time.monotonic() + timeout_seconds
    while time.monotonic() < deadline:
        detail = _request(base_url, "GET", f"/api/v1/documents/{document_id}")
        document = detail.get("document")
        if isinstance(document, dict) and document.get("status") == "READY":
            return detail
        if isinstance(document, dict) and document.get("status") in {"FAILED", "REJECTED"}:
            raise DocumentPipelineSmokeError("document reached a non-ready terminal state")
        time.sleep(0.25)
    raise DocumentPipelineSmokeError("document activation polling timed out")


def run_smoke(
    *,
    base_url: str,
    sample_path: Path,
    timeout_seconds: float = 90.0,
    run_id: uuid.UUID | None = None,
    knowledge_base_id: uuid.UUID = KNOWLEDGE_BASE_ID,
) -> dict[str, object]:
    endpoint = urlsplit(base_url)
    if (
        endpoint.scheme not in {"http", "https"}
        or not endpoint.hostname
        or endpoint.username is not None
        or endpoint.password is not None
    ):
        raise DocumentPipelineSmokeError("RAG base URL must be credential-free HTTP(S)")
    try:
        sample_content = sample_path.read_bytes()
    except OSError:
        raise DocumentPipelineSmokeError("synthetic upload sample is unavailable") from None
    if not sample_content or len(sample_content) > 1024 * 1024:
        raise DocumentPipelineSmokeError("synthetic upload sample has an invalid size")
    smoke_run_id = run_id or uuid.uuid4()
    content = sample_content + f"\n\nSynthetic smoke run: {smoke_run_id}\n".encode()
    digest = hashlib.sha256(content).hexdigest()
    key = uuid.uuid5(uuid.NAMESPACE_URL, f"geological-rag-document-smoke:{digest}")
    filename = f"upload_demo-{smoke_run_id.hex[:12]}.md"
    declaration_body: dict[str, object] = {
        "filename": filename,
        "declared_mime_type": "text/markdown",
        "expected_size": len(content),
        "expected_sha256": digest,
    }
    declared = _request(
        base_url,
        "POST",
        "/api/v1/document-uploads",
        headers={"Idempotency-Key": str(key)},
        body=declaration_body,
    )
    if declared.get("replayed") is not False:
        raise DocumentPipelineSmokeError("fresh upload declaration was unexpectedly replayed")
    upload_id = _required_uuid_text(declared, "id")
    _request(
        base_url,
        "PUT",
        f"/api/v1/document-uploads/{upload_id}/content",
        content=content,
    )
    completed = _request(
        base_url,
        "POST",
        f"/api/v1/document-uploads/{upload_id}/complete",
    )
    document = _required_mapping(completed, "document")
    document_id = _required_uuid_text(document, "id")
    parse_job = _required_mapping(completed, "job")
    parse_job_id = _required_uuid_text(parse_job, "id")
    parsed = _wait_job(base_url, parse_job_id, timeout_seconds=timeout_seconds)
    if parsed.get("stage") != "LOCAL_PARSED_PENDING_CLOUD_REVIEW":
        raise DocumentPipelineSmokeError("parse job did not reach the review stage")

    review = _request(
        base_url,
        "GET",
        f"/api/v1/documents/{document_id}/review-bundle?after_ordinal=-1&limit=100",
    )
    version = _required_mapping(review, "version")
    version_id = _required_uuid_text(version, "id")
    review_state = version.get("review_state")
    if review_state == "LOCAL_PARSED_PENDING_CLOUD_REVIEW":
        manifest = _required_hash(version, "outbound_manifest_sha256")
        revision = version.get("review_revision")
        if not isinstance(revision, int) or isinstance(revision, bool) or revision < 0:
            raise DocumentPipelineSmokeError("review revision is invalid")
        decision = _request(
            base_url,
            "POST",
            f"/api/v1/documents/{document_id}/review-decisions",
            body={
                "decision": "APPROVE",
                "reason_code": "SYNTHETIC_REVIEW_APPROVED",
                "expected_revision": revision,
                "outbound_manifest_sha256": manifest,
            },
        )
        embedding_job = _required_mapping(decision, "job")
        embedding_job_id = _required_uuid_text(embedding_job, "id")
        _wait_job(base_url, embedding_job_id, timeout_seconds=timeout_seconds)
    elif review_state != "CLOUD_APPROVED":
        raise DocumentPipelineSmokeError("document version is not eligible for indexing")

    ready = _wait_document_ready(
        base_url,
        document_id,
        timeout_seconds=timeout_seconds,
    )
    ready_document = _required_mapping(ready, "document")
    if ready_document.get("active_version_id") != version_id:
        raise DocumentPipelineSmokeError("ready document did not activate the reviewed version")

    retrieval = _request(
        base_url,
        "POST",
        "/api/v1/retrieval/search",
        body={
            "knowledge_base_id": str(knowledge_base_id),
            "query": "海岳示范区萤石矿需要哪些综合找矿标志？",
            "vector_top_k": 50,
            "rerank_top_n": 10,
        },
    )
    results = retrieval.get("results")
    if not isinstance(results, list):
        raise DocumentPipelineSmokeError("retrieval result is invalid")
    match = next(
        (
            item
            for item in results
            if isinstance(item, dict) and item.get("document_id") == document_id
        ),
        None,
    )
    if match is None:
        raise DocumentPipelineSmokeError("uploaded document was not retrieved")

    replayed = _request(
        base_url,
        "POST",
        "/api/v1/document-uploads",
        headers={"Idempotency-Key": str(key)},
        body=declaration_body,
    )
    if replayed.get("replayed") is not True or _required_uuid_text(replayed, "id") != upload_id:
        raise DocumentPipelineSmokeError("upload declaration replay contract failed")
    _request(
        base_url,
        "PUT",
        f"/api/v1/document-uploads/{upload_id}/content",
        content=content,
    )
    replayed_completion = _request(
        base_url,
        "POST",
        f"/api/v1/document-uploads/{upload_id}/complete",
    )
    replayed_document = _required_mapping(replayed_completion, "document")
    replayed_job = _required_mapping(replayed_completion, "job")
    if _required_uuid_text(replayed_document, "id") != document_id:
        raise DocumentPipelineSmokeError("document identity changed during replay")
    if _required_uuid_text(replayed_job, "id") != parse_job_id:
        raise DocumentPipelineSmokeError("parse job identity changed during replay")
    replayed_ready = _wait_document_ready(
        base_url,
        document_id,
        timeout_seconds=timeout_seconds,
    )
    if _required_mapping(replayed_ready, "document").get("active_version_id") != version_id:
        raise DocumentPipelineSmokeError("active version changed during replay")
    return {
        "status": "ok",
        "run_id": str(smoke_run_id),
        "knowledge_base_id": str(knowledge_base_id),
        "document_id": document_id,
        "document_version_id": version_id,
        "parse_job_id": parse_job_id,
        "document_status": ready_document.get("status"),
        "parse_stage": parsed.get("stage"),
        "retrieval_rank": match.get("rank"),
        "citation_id": match.get("citation_id"),
        "embedding_model": retrieval.get("embedding_model"),
        "rerank_status": retrieval.get("rerank_status"),
        "replay_confirmed": True,
    }


def _required_mapping(value: dict[str, Any], key: str) -> dict[str, Any]:
    item = value.get(key)
    if not isinstance(item, dict):
        raise DocumentPipelineSmokeError(f"response field is invalid: {key}")
    return cast(dict[str, Any], item)


def _required_uuid_text(value: dict[str, Any], key: str) -> str:
    item = value.get(key)
    if not isinstance(item, str):
        raise DocumentPipelineSmokeError(f"response field is invalid: {key}")
    try:
        parsed = uuid.UUID(item)
    except ValueError:
        raise DocumentPipelineSmokeError(f"response field is invalid: {key}") from None
    if str(parsed) != item:
        raise DocumentPipelineSmokeError(f"response field is invalid: {key}")
    return item


def _required_hash(value: dict[str, Any], key: str) -> str:
    item = value.get(key)
    if (
        not isinstance(item, str)
        or len(item) != 64
        or any(character not in "0123456789abcdef" for character in item)
    ):
        raise DocumentPipelineSmokeError(f"response field is invalid: {key}")
    return item


def main() -> None:
    base_url = os.getenv("RAG_BASE_URL", "http://127.0.0.1:8000")
    sample_path = Path(os.getenv("RAG_UPLOAD_SAMPLE", "data/samples/public/upload_demo.md"))
    namespace_mode = os.getenv("DOCUMENT_NAMESPACE_MODE", "fake").strip().lower()
    if namespace_mode == "fake":
        knowledge_base_id = KNOWLEDGE_BASE_ID
    elif namespace_mode == "bailian":
        knowledge_base_id = BAILIAN_KNOWLEDGE_BASE_ID
    else:
        sys.stdout.write(
            json.dumps(
                {"status": "failed", "error": "document namespace mode is invalid"},
                sort_keys=True,
            )
            + "\n"
        )
        raise SystemExit(1)
    try:
        result = run_smoke(
            base_url=base_url,
            sample_path=sample_path,
            knowledge_base_id=knowledge_base_id,
        )
    except DocumentPipelineSmokeError as exc:
        sys.stdout.write(json.dumps({"status": "failed", "error": str(exc)}, sort_keys=True) + "\n")
        raise SystemExit(1) from None
    sys.stdout.write(json.dumps(result, sort_keys=True) + "\n")


if __name__ == "__main__":
    main()