from __future__ import annotations

import hashlib
import io
import uuid
import zipfile
from dataclasses import dataclass, field, replace
from datetime import UTC, datetime, timedelta
from pathlib import Path

import pytest

from app.core.config import Settings
from app.persistence.document_workflows import (
    ArtifactPlan,
    DocumentSource,
    plan_artifact,
)
from app.persistence.job_queue import BackgroundJob, JobLease
from app.services.document_ingestion import (
    ChunkingConfig,
    CloudTextPolicy,
    IngestionArtifact,
)
from app.workers.document_jobs import ParseDocumentHandler, build_document_handlers

NOW = datetime(2026, 7, 13, 8, 0, tzinfo=UTC)
JOB_ID = uuid.UUID("10000000-0000-0000-0000-000000000001")
UPLOAD_ID = uuid.UUID("20000000-0000-0000-0000-000000000002")
DOCUMENT_ID = uuid.UUID("30000000-0000-0000-0000-000000000003")
KB_ID = uuid.UUID("40000000-0000-0000-0000-000000000004")
SCOPE_ID = uuid.UUID("50000000-0000-0000-0000-000000000005")
STORAGE_KEY = uuid.UUID("60000000-0000-0000-0000-000000000006")


def _job(*, token: uuid.UUID | None = None) -> BackgroundJob:
    lease_token = token or uuid.uuid4()
    lease = JobLease(JOB_ID, "worker-documents", lease_token)
    return BackgroundJob(
        id=JOB_ID,
        job_type="PARSE_DOCUMENT",
        required_capability="document_parse",
        resource_type="document",
        resource_id=DOCUMENT_ID,
        idempotency_key=f"parse-document:{DOCUMENT_ID}",
        payload={"upload_id": str(UPLOAD_ID), "document_id": str(DOCUMENT_ID)},
        stage="PENDING",
        progress=0,
        priority=0,
        attempt=1,
        max_attempts=3,
        run_after=NOW,
        lease_until=NOW + timedelta(seconds=60),
        created_at=NOW,
        updated_at=NOW,
        lease=lease,
    )


def _source(content: bytes, *, filename: str, mime_type: str) -> DocumentSource:
    return DocumentSource(
        upload_id=UPLOAD_ID,
        document_id=DOCUMENT_ID,
        knowledge_base_id=KB_ID,
        access_scope_id=SCOPE_ID,
        filename=filename,
        mime_type=mime_type,
        storage_key=STORAGE_KEY,
        byte_size=len(content),
        raw_sha256=hashlib.sha256(content).hexdigest(),
    )


@dataclass
class FakeStorage:
    content: bytes
    calls: list[tuple[uuid.UUID, int, str]] = field(default_factory=list)

    async def read_verified(
        self,
        *,
        storage_key: uuid.UUID,
        expected_size: int,
        expected_sha256: str,
    ) -> bytes:
        self.calls.append((storage_key, expected_size, expected_sha256))
        assert len(self.content) == expected_size
        assert hashlib.sha256(self.content).hexdigest() == expected_sha256
        return self.content


@dataclass
class FakeRepository:
    source: DocumentSource
    plans_by_version: dict[uuid.UUID, ArtifactPlan] = field(default_factory=dict)
    failures: list[tuple[JobLease, str]] = field(default_factory=list)
    load_calls: list[BackgroundJob] = field(default_factory=list)
    persist_calls: int = 0

    def load_source(self, job: BackgroundJob) -> DocumentSource:
        self.load_calls.append(job)
        return self.source

    def record_terminal_parse_failure(
        self,
        *,
        lease: JobLease,
        source: DocumentSource,
        error_code: str,
    ) -> None:
        assert source == self.source
        self.failures.append((lease, error_code))

    def persist_artifact(
        self,
        *,
        lease: JobLease,
        source: DocumentSource,
        artifact: IngestionArtifact,
        cloud_policy: CloudTextPolicy,
        embedding_model: str,
        embedding_dimension: int,
    ) -> ArtifactPlan:
        del lease
        assert source == self.source
        assert embedding_model == "text-embedding-v4"
        assert embedding_dimension == 1024
        plan = plan_artifact(source, artifact, cloud_policy=cloud_policy)
        existing = self.plans_by_version.setdefault(plan.version_id, plan)
        assert existing == plan
        self.persist_calls += 1
        return plan


def _handler(repository: FakeRepository, storage: FakeStorage) -> ParseDocumentHandler:
    return ParseDocumentHandler(
        repository=repository,
        storage=storage,
        max_upload_bytes=1024 * 1024,
        chunking=ChunkingConfig(target_tokens=512, max_tokens=800, overlap_tokens=64),
        cloud_policy=CloudTextPolicy(),
        embedding_model="text-embedding-v4",
        embedding_dimension=1024,
    )


@pytest.mark.asyncio
async def test_ready_document_is_verified_parsed_and_idempotent_across_new_leases() -> None:
    content = "# 地质概况\n\n虚构铜矿资料用于入库验证。".encode()
    source = _source(content, filename="synthetic.md", mime_type="text/markdown")
    repository = FakeRepository(source)
    storage = FakeStorage(content)
    handler = _handler(repository, storage)

    await handler(_job(token=uuid.UUID("70000000-0000-0000-0000-000000000007")))
    await handler(_job(token=uuid.UUID("80000000-0000-0000-0000-000000000008")))

    assert repository.failures == []
    assert repository.persist_calls == 2
    assert len(repository.plans_by_version) == 1
    plan = next(iter(repository.plans_by_version.values()))
    assert plan.document_status == "LOCAL_PARSED_PENDING_CLOUD_REVIEW"
    assert plan.review_state == "LOCAL_PARSED_PENDING_CLOUD_REVIEW"
    assert len(plan.pages) == 1
    assert len(plan.blocks) == 2
    assert len(plan.chunks) == 1
    chunk = plan.chunks[0]
    assert chunk.embedding_text == chunk.embedding_prefix + chunk.cloud_text
    assert chunk.metadata["source_anchor"]
    assert len(storage.calls) == 2


@pytest.mark.asyncio
async def test_deterministic_parser_rejection_is_recorded_without_retry_exception() -> None:
    content = b"\x81\x82\x83"
    source = _source(content, filename="invalid.txt", mime_type="text/plain")
    repository = FakeRepository(source)

    await _handler(repository, FakeStorage(content))(_job())

    assert repository.persist_calls == 0
    assert len(repository.failures) == 1
    assert repository.failures[0][1] == "INVALID_TEXT_ENCODING"


@pytest.mark.asyncio
async def test_pdf_creates_ocr_required_plan_without_text_or_map_claims() -> None:
    content = b"%PDF-1.7\nsynthetic"
    source = _source(content, filename="synthetic.pdf", mime_type="application/pdf")
    repository = FakeRepository(source)

    await _handler(repository, FakeStorage(content))(_job())

    plan = next(iter(repository.plans_by_version.values()))
    assert plan.document_status == "LOCAL_OCR_REQUIRED"
    assert plan.job_stage == "OCR_REQUIRED"
    assert plan.pages == ()
    assert plan.blocks == ()
    assert plan.chunks == ()
    assert plan.version_error_code == "PDF_PARSER_UNAVAILABLE"


def _docx(*, unsafe_entry: str | None = None) -> bytes:
    content_types = b"""<Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types">
      <Override PartName="/word/document.xml"
       ContentType="application/vnd.openxmlformats-officedocument.wordprocessingml.document.main+xml"/>
    </Types>"""
    document = b"""<w:document xmlns:w="http://schemas.openxmlformats.org/wordprocessingml/2006/main">
      <w:body><w:p><w:r><w:t>DOCX evidence</w:t></w:r></w:p><w:sectPr/></w:body>
    </w:document>"""
    output = io.BytesIO()
    with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED) as package:
        package.writestr("[Content_Types].xml", content_types)
        package.writestr("word/document.xml", document)
        if unsafe_entry is not None:
            package.writestr(unsafe_entry, b"must never be extracted")
    return output.getvalue()


@pytest.mark.asyncio
async def test_docx_preserves_explicit_unknown_physical_page() -> None:
    content = _docx()
    source = _source(
        content,
        filename="synthetic.docx",
        mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
    )
    repository = FakeRepository(source)

    await _handler(repository, FakeStorage(content))(_job())

    plan = next(iter(repository.plans_by_version.values()))
    assert plan.pages[0].page_number is None
    assert plan.blocks[0].page_start is None
    assert plan.chunks[0].page_start is None


@pytest.mark.asyncio
async def test_malicious_docx_is_rejected_without_persistence_or_retry() -> None:
    content = _docx(unsafe_entry="../outside.xml")
    source = _source(
        content,
        filename="malicious.docx",
        mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
    )
    repository = FakeRepository(source)
    job = _job()

    await _handler(repository, FakeStorage(content))(job)

    assert repository.persist_calls == 0
    assert repository.failures == [(job.lease, "DOCX_PATH_TRAVERSAL")]


def test_factory_exposes_only_local_parse_handler_without_model_client(tmp_path: Path) -> None:
    handlers = build_document_handlers(Settings(upload_root=tmp_path))

    assert set(handlers) == {"PARSE_DOCUMENT"}
    handler = handlers["PARSE_DOCUMENT"]
    assert isinstance(handler, ParseDocumentHandler)
    assert not hasattr(handler, "model_client")
    assert handler.embedding_model == "text-embedding-v4"
    assert handler.embedding_dimension == 1024


def test_plan_ids_are_stable_but_namespaced_by_knowledge_base_and_version() -> None:
    from app.services.document_ingestion import ingest_document

    content = b"stable synthetic evidence"
    first_source = _source(content, filename="stable.txt", mime_type="text/plain")
    second_source = replace(first_source, knowledge_base_id=uuid.uuid4())
    artifact = ingest_document(
        filename=first_source.filename,
        declared_mime_type=first_source.mime_type,
        content=content,
    )

    first = plan_artifact(first_source, artifact, cloud_policy=CloudTextPolicy())
    repeated = plan_artifact(first_source, artifact, cloud_policy=CloudTextPolicy())
    other_kb = plan_artifact(second_source, artifact, cloud_policy=CloudTextPolicy())

    assert first == repeated
    assert first.version_id != other_kb.version_id
    assert {item.id for item in first.pages}.isdisjoint({item.id for item in other_kb.pages})
    assert {item.id for item in first.blocks}.isdisjoint({item.id for item in other_kb.blocks})
    assert {item.id for item in first.chunks}.isdisjoint({item.id for item in other_kb.chunks})