"""marker_worker — markdown stage 소비. Phase 1B + Phase 1B.5 (ImgAuth).

플로우:
  classify_worker 완료 → enqueue 'markdown' stage (또는 reprocess 스크립트가 force=True 로 enqueue)
  → marker_worker.process()
  → doc_type / 확장자 / page_count 가드 → marker-service POST /convert
  → 응답 이미지 NAS persist + document_images UPSERT + md_content ref 정규화
  → md_content 저장 또는 doc-level failed (404/422) 또는 transient raise (5xx → queue retry)

이미지 저장 위치:  NAS `/documents/extracted_images/{document_id}/{image_key}.{ext}`
md_content ref 형식: `![alt](docimg:img_001)` — image_key 가 sequence 기반 결정적 → idempotent.

plan: ~/.claude/plans/piped-humming-crystal.md
"""
import base64
import hashlib
import logging
import os
import re
from pathlib import Path
from typing import Any

import fitz                              # PyMuPDF
import httpx
from sqlalchemy import delete, desc, select, update
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from models.document import Document
from models.document_image import DocumentImage
from models.queue import ProcessingQueue

logger = logging.getLogger(__name__)

MARKER_ENDPOINT = "http://marker-service:3300/convert"
MARKER_TIMEOUT = 300                    # 큰 PDF 5 분 한도
MAX_PAGES = 200                         # 페이지 hard limit

# Phase 1B.5: 이미지 NAS persist 토글. rollback 시 false → 응답 images 무시 + md_content
# rewrite skip → placeholder card 폴백 자연 유지. 환경변수 미설정 = 기본 활성화.
MARKDOWN_IMAGE_PERSIST = os.getenv("MARKDOWN_IMAGE_PERSIST", "true").lower() in ("1", "true", "yes")
EXTRACTED_IMAGES_ROOT = Path("/documents/extracted_images")

# md_content image ref 정규식. alt + href 캡처. 외부 URL 은 보존 (slug 매칭 안 되는 경우).
_IMAGE_REF_RE = re.compile(r"!\[([^\]]*)\]\(([^)]+)\)")

_FORMAT_TO_MIME = {
    "png": "image/png",
    "jpeg": "image/jpeg",
    "jpg": "image/jpeg",
    "webp": "image/webp",
    "gif": "image/gif",
}

# Phase 1B = PDF only. DOCX 등은 후속 Phase.
SUPPORTED_EXTENSIONS = {".pdf"}

# config.yaml document_types 의 한국어 label 직접 사용 (Pre-flight 결과).
# Round 0 사용자 의도 = 표 중심 발주/계산/명세 도메인.
SKIP_DOC_TYPES = {
    # 한국어 (실제 config label)
    "발주서", "세금계산서", "명세표",
    # 영문 placeholder (현재 config 미포함, 미래 신규 분류 자동 적용)
    "Invoice", "Purchase_Order", "Estimate", "Statement",
}

# Phase 1D 결과 반영 (2026-05-03):
# 애플펜슬 손글씨 필기 PDF 는 사용자 글씨체 + Marker layout/OCR 모델 한계로
# quality fail. Marker 튜닝 영역이 아니라 input 자체가 markdown 부적합. title/path
# 에 명시적 필기 표식이 있으면 enqueue 단계에서 자동 skip.
#
# 보수적 키워드 4개만 (false positive 회피):
#   포함: 필기, 손글씨, handwritten, handwriting
#   제외: 노트 (노트북 매뉴얼 / release notes), scan/스캔 (5127 처럼 정상 변환되는
#         스캔 PDF 가 있어 false positive 위험)
SKIP_HANDWRITTEN_KEYWORDS = ("필기", "손글씨", "handwritten", "handwriting")
HANDWRITTEN_HINT_REGEX = re.compile("|".join(SKIP_HANDWRITTEN_KEYWORDS), re.IGNORECASE)


def _matches_handwritten_hint(title: str | None, file_path: str | None) -> str | None:
    """title 또는 file_path 에 보수적 필기 키워드 매칭 시 매칭 키워드 반환, 아니면 None."""
    blob = " ".join(filter(None, [title or "", file_path or ""]))
    m = HANDWRITTEN_HINT_REGEX.search(blob)
    return m.group(0) if m else None

# documents.file_path 는 NAS 상대경로 (예: 'news/SCMP/abc').
# fastapi 와 marker-service 모두 NAS 를 /documents 에 ro 마운트.
CONTAINER_PATH_PREFIX = os.getenv("MARKER_CONTAINER_PATH_PREFIX", "/documents")


def _to_marker_path(file_path: str) -> str:
    """file_path 를 컨테이너 내부 절대경로로 변환."""
    if file_path.startswith("/"):
        return file_path                # 이미 절대경로 (legacy 가능성, 그대로)
    return f"{CONTAINER_PATH_PREFIX}/{file_path}"


def _get_page_count(file_path: str) -> int | None:
    """PyMuPDF 로 페이지 수 사전 검사. 실패 시 None (가드 통과)."""
    try:
        with fitz.open(file_path) as doc:
            return doc.page_count
    except Exception as exc:
        logger.warning(f"[marker] page count read failed path={file_path}: {exc}")
        return None


async def process(document_id: int, session: AsyncSession) -> None:
    """markdown stage 워커 진입점. queue_consumer 가 호출."""
    doc = await session.get(Document, document_id)
    if doc is None:
        logger.warning(f"[marker] document {document_id} not found")
        return

    # ---- (0) force_reprocess flag — reprocess 스크립트가 queue payload 로 전달 ----
    force_reprocess = await _read_force_reprocess(session, document_id)
    if force_reprocess:
        logger.info(f"markdown_force_reprocess id={document_id}")

    # ---- (1) doc_type skip ----
    if doc.document_type in SKIP_DOC_TYPES:
        logger.info(
            f"markdown_skip_table_heavy_doctype id={document_id} doc_type={doc.document_type}"
        )
        await _set_skipped(session, document_id, f"skipped: doc_type={doc.document_type}")
        return

    # ---- (1.5) handwritten hint skip (Phase 1D pilot 결과 반영) ----
    matched_keyword = _matches_handwritten_hint(doc.title, doc.file_path)
    if matched_keyword:
        logger.info(
            f"markdown_skip_handwritten_hint id={document_id} "
            f"keyword={matched_keyword} title={(doc.title or '')[:80]}"
        )
        await _set_skipped(
            session, document_id,
            "skipped: handwritten note (title/path heuristic)",
        )
        return

    # ---- (2) file_path validation ----
    if not doc.file_path:
        await _fail(session, document_id, "no file_path")
        return

    container_path = _to_marker_path(doc.file_path)

    # ---- (3) PDF only ----
    suffix = Path(container_path).suffix.lower()
    if suffix not in SUPPORTED_EXTENSIONS:
        logger.info(f"markdown_skip_unsupported_extension id={document_id} ext={suffix}")
        await _set_skipped(
            session, document_id,
            f"skipped: unsupported extension={suffix} in Phase 1B",
        )
        return

    # ---- (4) MAX_PAGES guard ----
    page_count = _get_page_count(container_path)
    if page_count is not None and page_count > MAX_PAGES:
        logger.info(f"markdown_skip_too_many_pages id={document_id} pages={page_count}")
        await _set_skipped(
            session, document_id,
            f"skipped: page_count={page_count} exceeds MAX_PAGES={MAX_PAGES}",
        )
        return

    # ---- (5) processing 표시 ----
    await session.execute(
        update(Document).where(Document.id == document_id).values(md_status="processing")
    )
    await session.commit()

    # ---- (6) Marker 호출 ----
    try:
        async with httpx.AsyncClient(timeout=MARKER_TIMEOUT) as client:
            resp = await client.post(
                MARKER_ENDPOINT,
                json={"file_path": container_path, "max_pages": MAX_PAGES},
            )
            resp.raise_for_status()
            data = resp.json()
    except httpx.HTTPStatusError as exc:
        # 404 (file_not_found) + 422 (conversion_failed) = non-retryable failed.
        # 5xx 만 transient → re-raise → queue retry.
        if exc.response.status_code in {404, 422}:
            try:
                detail = exc.response.json().get("detail", {})
                code = detail.get("code", "unknown")
                message = detail.get("message", exc.response.text)
            except Exception:
                code, message = "unknown", exc.response.text
            logger.warning(
                f"[marker] non-retryable {exc.response.status_code} {code} id={document_id}: {message[:200]}"
            )
            await _fail(session, document_id, f"{code}: {message[:1000]}")
            return
        logger.exception(f"[marker] http error id={document_id}: {exc}")
        raise
    except httpx.TransportError as exc:
        # transport 계층 오류 — Connect/Read/Write/RemoteProtocol/Timeout 류 전부 transient.
        # marker-service 응답끊김(RemoteProtocolError) / 모델 워밍업 / 큰 PDF I/O 끊김 등에서 발생.
        # queue retry 대상 (max_attempts 까지 자동 재시도).
        logger.warning(
            f"[marker] transient error id={document_id} kind={type(exc).__name__}: {exc}"
        )
        raise
    except Exception as exc:
        logger.exception(f"[marker] unexpected error id={document_id}: {exc}")
        await _fail(session, document_id, str(exc)[:1000])
        return

    # ---- (7) image persist + md_content rewrite (Phase 1B.5) ----
    md_content_raw = data["md_content"]
    images_resp = data.get("images") if MARKDOWN_IMAGE_PERSIST else None

    saved_images: list[dict[str, Any]] = []
    if images_resp:
        try:
            saved_images = _persist_images_to_nas(document_id, images_resp)
        except OSError as exc:
            # NAS 일시 끊김 등 — transient. queue retry 로 복구.
            logger.warning(
                f"[marker] image persist NAS write failed id={document_id}: "
                f"{type(exc).__name__}: {exc}"
            )
            raise

    # md_content 안의 ref 를 stable internal scheme `docimg:img_NNN` 으로 정규화.
    slug_to_key = {img["source_slug"]: img["image_key"] for img in saved_images}
    md_content = _rewrite_image_refs(md_content_raw, slug_to_key)

    # quality 메트릭은 정규화 *후* md_content 기준 (실제 저장본). image_count 도 정확.
    quality = _compute_quality(md_content, doc.extracted_text or "", data["raw_metrics"])
    if data.get("images_truncated"):
        quality.setdefault("warnings", []).append("images_truncated")

    # ---- (8) DB 트랜잭션 — documents UPDATE + document_images UPSERT + 고아 row DELETE ----
    orphan_paths = await _sync_document_images(session, document_id, saved_images, data)

    await session.execute(
        update(Document).where(Document.id == document_id).values(
            md_content=md_content,
            md_status="success",
            md_extraction_engine=data["engine"],
            md_extraction_engine_version=data["engine_version"],
            md_extraction_quality=quality,
            md_content_hash=hashlib.sha256(md_content.encode("utf-8")).hexdigest(),
            md_source_hash=doc.file_hash,
            md_generated_at=_now(),
            md_extraction_error=None,
            md_frontmatter=doc.md_frontmatter or {},
            md_format_version="1.0",
            content_origin="extracted",
        )
    )
    await session.commit()

    # ---- (9) commit 후 고아 NAS 파일 unlink (best-effort, 실패해도 DB 정합 유지) ----
    for orphan_path in orphan_paths:
        try:
            Path(orphan_path).unlink(missing_ok=True)
        except Exception as exc:
            logger.warning(
                f"[marker] orphan image unlink failed id={document_id} path={orphan_path}: "
                f"{type(exc).__name__}: {exc}"
            )

    logger.info(
        f"[marker] success id={document_id} len={len(md_content)} "
        f"images={len(saved_images)} orphans_removed={len(orphan_paths)} "
        f"elapsed_ms={data['elapsed_ms']}"
    )


async def _read_force_reprocess(session: AsyncSession, document_id: int) -> bool:
    """현재 markdown stage queue 행의 payload.force_reprocess 조회. 없으면 False."""
    row = await session.scalar(
        select(ProcessingQueue)
        .where(
            ProcessingQueue.document_id == document_id,
            ProcessingQueue.stage == "markdown",
            ProcessingQueue.status == "processing",
        )
        .order_by(desc(ProcessingQueue.id))
        .limit(1)
    )
    if not row or not row.payload:
        return False
    return bool(row.payload.get("force_reprocess"))


def _persist_images_to_nas(
    document_id: int, images_resp: list[dict[str, Any]]
) -> list[dict[str, Any]]:
    """marker 응답 이미지 list 를 NAS 에 저장하고 메타 dict 리스트 반환.

    image_key 는 sequence 기반 결정적 (`img_001` → `img_NNN`, marker 출력 순서 = 안정적).
    같은 doc 재변환 시 같은 key 가 같은 path 에 overwrite → idempotent.
    """
    img_root = EXTRACTED_IMAGES_ROOT / str(document_id)
    img_root.mkdir(parents=True, exist_ok=True)

    saved: list[dict[str, Any]] = []
    for seq, img in enumerate(images_resp, start=1):
        try:
            raw_bytes = base64.b64decode(img["bytes_b64"])
        except Exception as exc:
            logger.warning(
                f"[marker] image base64 decode failed id={document_id} "
                f"seq={seq} slug={img.get('slug')}: {exc}"
            )
            continue

        fmt = (img.get("format") or "png").lower()
        ext = "jpeg" if fmt == "jpg" else fmt
        image_key = f"img_{seq:03d}"
        filename = f"{image_key}.{ext}"
        rel_path = f"extracted_images/{document_id}/{filename}"
        abs_path = img_root / filename

        # NAS write — 실패 시 OSError raise (transient retry).
        abs_path.write_bytes(raw_bytes)

        saved.append({
            "image_key": image_key,
            "source_slug": img.get("slug") or "",
            "relative_path": rel_path,
            "file_path": str(abs_path),
            "mime_type": _FORMAT_TO_MIME.get(ext, "application/octet-stream"),
            "file_size": len(raw_bytes),
            "content_hash": hashlib.sha256(raw_bytes).hexdigest(),
            "width": img.get("width"),
            "height": img.get("height"),
        })
    return saved


def _rewrite_image_refs(md_text: str, slug_to_key: dict[str, str]) -> str:
    """md_content 안의 `![alt](slug)` ref 를 `![alt](docimg:image_key)` 로 정규화.

    - slug_to_key 에 없는 href 는 원본 유지 (외부 URL / 경로 변형 등)
    - alt 그대로 보존
    - 매칭 = href 가 (a) 정확히 slug 와 같음 OR (b) basename 이 slug 와 같음
      (marker 가 `_page_0_Picture_3.jpeg` 또는 `subdir/_page_0_Picture_3.jpeg` 어느 쪽도 emit 가능)
    """
    if not slug_to_key:
        return md_text

    def _replace(match: re.Match) -> str:
        alt, href = match.group(1), match.group(2)
        # 정확 매치 우선
        if href in slug_to_key:
            return f"![{alt}](docimg:{slug_to_key[href]})"
        # basename 매치 fallback
        basename = href.rsplit("/", 1)[-1]
        if basename in slug_to_key:
            return f"![{alt}](docimg:{slug_to_key[basename]})"
        return match.group(0)

    return _IMAGE_REF_RE.sub(_replace, md_text)


async def _sync_document_images(
    session: AsyncSession,
    document_id: int,
    saved_images: list[dict[str, Any]],
    response_data: dict[str, Any],
) -> list[str]:
    """document_images 동기화 — 신규 keys UPSERT + 고아 row DELETE.

    반환: commit 후 unlink 해야 할 NAS 파일 경로 리스트.
    """
    # 기존 row 조회 (file_path 보존 — 고아 파일 unlink 용)
    existing = (await session.execute(
        select(DocumentImage.image_key, DocumentImage.file_path)
        .where(DocumentImage.document_id == document_id)
    )).all()
    existing_map = {key: path for key, path in existing}

    new_keys = {img["image_key"] for img in saved_images}
    orphan_keys = set(existing_map.keys()) - new_keys
    orphan_paths = [existing_map[k] for k in orphan_keys]

    # UPSERT — image_key 가 같으면 file_path/hash/dimensions 등을 갱신.
    for img in saved_images:
        stmt = (
            pg_insert(DocumentImage)
            .values(
                document_id=document_id,
                image_key=img["image_key"],
                relative_path=img["relative_path"],
                file_path=img["file_path"],
                mime_type=img["mime_type"],
                file_size=img["file_size"],
                content_hash=img["content_hash"],
                width=img.get("width"),
                height=img.get("height"),
                page_index=img.get("page_index"),
                alt_text=img.get("alt_text"),
                source_slug=img.get("source_slug"),
                extraction_engine=response_data.get("engine") or "marker",
                extraction_engine_version=response_data.get("engine_version"),
            )
            .on_conflict_do_update(
                index_elements=["document_id", "image_key"],
                set_={
                    "relative_path": img["relative_path"],
                    "file_path": img["file_path"],
                    "mime_type": img["mime_type"],
                    "file_size": img["file_size"],
                    "content_hash": img["content_hash"],
                    "width": img.get("width"),
                    "height": img.get("height"),
                    "page_index": img.get("page_index"),
                    "alt_text": img.get("alt_text"),
                    "source_slug": img.get("source_slug"),
                    "extraction_engine": response_data.get("engine") or "marker",
                    "extraction_engine_version": response_data.get("engine_version"),
                },
            )
        )
        await session.execute(stmt)

    if orphan_keys:
        await session.execute(
            delete(DocumentImage).where(
                DocumentImage.document_id == document_id,
                DocumentImage.image_key.in_(orphan_keys),
            )
        )

    return orphan_paths


def _compute_quality(md: str, raw_text: str, raw_metrics: dict[str, Any]) -> dict[str, Any]:
    """1B 휴리스틱 quality. 임계 판정 미적용 (Phase 1D 후행)."""
    heading_lines = re.findall(r"^(#{1,6})\s", md, flags=re.MULTILINE)
    table_rows = re.findall(r"^\|.*\|\s*$", md, flags=re.MULTILINE)
    image_refs = re.findall(r"!\[[^\]]*\]\([^)]+\)", md)
    image_with_alt = re.findall(r"!\[[^\]]+\]\([^)]+\)", md)

    metrics = {
        "source_page_count": raw_metrics.get("page_count"),
        "markdown_heading_count": len(heading_lines),
        "heading_jump_count": _count_heading_jumps(heading_lines),
        "markdown_table_row_count": len(table_rows),
        "markdown_image_count": len(image_refs),
        "image_alt_text_ratio": (len(image_with_alt) / len(image_refs)) if image_refs else None,
        "text_length_ratio": (len(md) / len(raw_text)) if raw_text else None,
    }
    warnings = []
    if metrics["heading_jump_count"] > 0:
        warnings.append("heading_hierarchy_jump")
    if metrics["image_alt_text_ratio"] is not None and metrics["image_alt_text_ratio"] < 0.5:
        warnings.append("low_image_alt_text_ratio")
    return {
        "score": None,                  # Phase 1D 에서 가중합 결정
        "metrics": metrics,
        "warnings": warnings,
    }


def _count_heading_jumps(headings: list[str]) -> int:
    levels = [len(h) for h in headings]
    return sum(1 for i in range(1, len(levels)) if levels[i] - levels[i - 1] > 1)


async def _set_skipped(session: AsyncSession, document_id: int, reason: str) -> None:
    """skip 시 md_content/hash NULL 클리어 (이전 success 잔존 방지)."""
    await session.execute(
        update(Document).where(Document.id == document_id).values(
            md_status="skipped",
            md_content=None,
            md_content_hash=None,
            md_extraction_engine="marker",
            md_extraction_error=reason,
            md_generated_at=_now(),
            content_origin="extracted",
        )
    )
    await session.commit()


async def _fail(session: AsyncSession, document_id: int, error: str) -> None:
    """doc-level failed (재시도 무의미)."""
    await session.execute(
        update(Document).where(Document.id == document_id).values(
            md_status="failed",
            md_content=None,
            md_content_hash=None,
            md_extraction_error=error,
            md_extraction_engine="marker",
            md_generated_at=_now(),
            content_origin="extracted",
        )
    )
    await session.commit()


def _now():
    from datetime import datetime, timezone
    return datetime.now(timezone.utc)
