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

플로우:
  classify_worker 완료 → enqueue 'markdown' stage (또는 reprocess 스크립트가 force=True 로 enqueue)
  → marker_worker.process()
  → doc_type / handwritten skip
  → 이미 마크다운(file_format=md / .md) = _process_markdown_passthrough (변환 없이 내용 직접 적재)
  → PDF: 확장자 / page_count gauge
  → 소형(≤SPLIT_THRESHOLD_PAGES) = _process_single 통째 1-shot /convert
    대형(>SPLIT_THRESHOLD_PAGES) = _process_split page-range BATCH_PAGES 윈도우 분할
  → 응답 이미지 NAS persist + document_images UPSERT + md_content ref 정규화
  → md_content 저장(대형=head+manifest) / md_status success·partial·failed / 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 asyncio
import base64
import hashlib
import json
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                         # 소형 1-shot 경로 /convert max_pages 안전장치

# LargeDoc split (PR-DocSrv-LargeDoc-Split-Markdown-1, plan brisk-paging-quokka.md):
# >SPLIT_THRESHOLD_PAGES PDF 는 통째 skip 대신 BATCH_PAGES 윈도우로 page-range 분할 변환.
# 각 batch /convert 호출 < MARKER_TIMEOUT 보장 + queue 폭주 회피. 목적 = md_content(Markdown
# 렌더링) 복구. 검색 인덱스는 기존 document_chunks(extracted_text window chunks)가 담당
# (commit 4 marker_section 적재는 이중 적재 회피로 드롭 — 별 PR retrieval 비교 후보).
SPLIT_THRESHOLD_PAGES = int(os.getenv("MARKER_SPLIT_THRESHOLD_PAGES", "120"))
BATCH_PAGES = int(os.getenv("MARKER_BATCH_PAGES", "40"))
MAX_SPLIT_PAGES = int(os.getenv("MARKER_MAX_SPLIT_PAGES", "5000"))     # 초과 = skipped_too_large 안전상태
LARGE_DOC_MD_CONTENT_HEAD_CHARS = int(os.getenv("LARGE_DOC_MD_CONTENT_HEAD_CHARS", "50000"))

# 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 (marker-service). office/hwp 는 C-2 에서 office_md 하이브리드로 분기.
SUPPORTED_EXTENSIONS = {".pdf"}

# C-2: office/hwp → md (OOXML=markitdown / hwp=LibreOffice). 변환기가 지원하는 suffix 집합.
# 레거시 바이너리(.doc/.xls/.ppt)는 markitdown 미지원 → 여기 없음(=PDF-only 게이트에서 skip).
from workers.office_md import SUPPORTED as OFFICE_MD_SUPPORTED  # noqa: E402

# 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


def _is_markdown_doc(doc: Document) -> bool:
    """이미 마크다운 형식인 문서 판정 (file_format=md/markdown 또는 .md/.markdown 확장자)."""
    fmt = (doc.file_format or "").lower()
    if fmt in ("md", "markdown"):
        return True
    if doc.file_path:
        return Path(doc.file_path).suffix.lower() in (".md", ".markdown")
    return False


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

    # ---- (1.7) 이미 마크다운인 문서 = marker 변환 불필요, 내용 직접 적재 (commit 5) ----
    # fileless md(file_path NULL) + .md/.markdown 파일 둘 다 처리. 기존엔 전자=「no file_path」
    # fail, 후자=unsupported extension skip → md_content 미생성이었음.
    if _is_markdown_doc(doc):
        await _process_markdown_passthrough(doc, document_id, session)
        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)
    suffix = Path(container_path).suffix.lower()

    # ---- (3) office/hwp → md (C-2): PDF 외 지원 포맷은 office_md 하이브리드 변환 ----
    if suffix in OFFICE_MD_SUPPORTED:
        await session.execute(
            update(Document).where(Document.id == document_id).values(md_status="processing")
        )
        await session.commit()
        await _process_office(doc, document_id, container_path, session)
        return

    # ---- (3.5) PDF only (그 외 확장자 = skip) ----
    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) page_count gauge + 분기 (LargeDoc split) ----
    page_count = _get_page_count(container_path)

    # >MAX_SPLIT_PAGES = 변환 안전상태(manual_review). silently skip 아님.
    if page_count is not None and page_count > MAX_SPLIT_PAGES:
        logger.info(f"markdown_skip_too_large id={document_id} pages={page_count}")
        await _set_skipped(
            session, document_id,
            f"skipped_too_large: page_count={page_count} exceeds MAX_SPLIT_PAGES={MAX_SPLIT_PAGES}",
        )
        return

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

    # ---- (6) 변환 분기: 소형 1-shot / 대형(>SPLIT_THRESHOLD) page-range 분할 ----
    if page_count is not None and page_count > SPLIT_THRESHOLD_PAGES:
        await _process_split(doc, document_id, container_path, page_count, session)
    else:
        await _process_single(doc, document_id, container_path, session)


async def _process_single(
    doc: Document, document_id: int, container_path: str, session: AsyncSession
) -> None:
    """소형 PDF(≤ SPLIT_THRESHOLD_PAGES) 통째 1-shot 변환 (Phase 1B/1B.5 기존 경로)."""
    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 _process_markdown_passthrough(
    doc: Document, document_id: int, session: AsyncSession
) -> None:
    """마크다운 문서 — 변환 없이 파일 내용(없으면 extracted_text)을 md_content 로 직접 적재.

    내용이 비면 skipped. engine='passthrough' 로 marker 변환본과 구분.
    """
    content = ""
    if doc.file_path:
        path = _to_marker_path(doc.file_path)
        try:
            content = Path(path).read_text(encoding="utf-8", errors="replace")
        except OSError as exc:
            logger.warning(
                f"[marker] md passthrough file read failed id={document_id} "
                f"path={path}: {type(exc).__name__}: {exc}"
            )
            content = ""
    if not content.strip():
        content = doc.extracted_text or ""        # fileless md = extracted_text 사용
    content = content.strip()

    if not content:
        await _set_skipped(session, document_id, "skipped: markdown doc with no content")
        return

    quality = _compute_quality(content, doc.extracted_text or "", {"page_count": None})
    await session.execute(
        update(Document).where(Document.id == document_id).values(
            md_content=content,
            md_status="success",
            md_extraction_engine="passthrough",
            md_extraction_engine_version=None,
            md_extraction_quality=quality,
            md_content_hash=hashlib.sha256(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()
    logger.info(
        f"[marker] md passthrough success id={document_id} len={len(content)} "
        f"src={'file' if doc.file_path else 'extracted_text'}"
    )


async def _process_office(
    doc: Document, document_id: int, container_path: str, session: AsyncSession
) -> None:
    """office/hwp → md (C-2). C-5 상태머신 postcondition 의 office arm.

    office_md.convert_office_to_md 는 이진 계약: 성공=비공백 md 반환 / 실패·빈출력·타임아웃·
    의존성부재=OfficeMdError raise. 따라서:
      - 성공 → md_status='success' (+ 비공백 md). 불변식 md_status ∈ {success,partial} ⟹ md 非공백 유지.
      - 실패/예외 → _fail (md_status='failed', ¬success·¬skipped). silent 'success+빈md' 절대 없음.
    partial arm 은 PDF split 전용 — office 는 이진이라 여기 없음. 'completed' 는 A-3 직렬화 전용(워커 미사용).
    quality 는 content-type-aware: office=scored(_compute_quality). 동기 변환은 to_thread 로 event loop 비차단.
    """
    from workers.office_md import (
        OfficeMdError,
        convert_hwp_to_md_and_images,
        convert_office_to_md,
    )

    suffix = Path(container_path).suffix.lower()
    if suffix == ".hwp":
        engine = "pyhwp"  # HWP5 binary: libhwplo 못 읽어 pyhwp 로 교체(2026-06-09)
    elif suffix == ".hwpx":
        engine = "libreoffice_hwp"  # HWPX 는 pyhwp 미지원 → LibreOffice 폴백
    else:
        engine = "markitdown"

    hwp_images: list[dict[str, Any]] = []
    try:
        # 동기 subprocess/markitdown — 스레드로 빼서 이벤트 루프 비차단.
        if suffix == ".hwp":
            md_content, hwp_images = await asyncio.to_thread(
                convert_hwp_to_md_and_images, container_path
            )
        else:
            md_content = await asyncio.to_thread(convert_office_to_md, container_path)
    except OfficeMdError as exc:
        logger.warning(f"[marker] office md 변환 실패 id={document_id} engine={engine}: {exc}")
        await _fail(session, document_id, f"office_md: {str(exc)[:990]}", engine=engine)
        return
    except Exception as exc:  # 예기치 못한 예외도 failed (success+빈md 절대 금지)
        logger.exception(f"[marker] office md unexpected error id={document_id}: {exc}")
        await _fail(session, document_id, f"office_md_unexpected: {str(exc)[:980]}", engine=engine)
        return

    # ---- 이미지 NAS persist (.hwp 전용) ----
    # hwp5html 은 bindata raster 를 추출하나 본문 xhtml 에 <img> 앵커가 없어(orphan, --css/--html
    # 동일) 인라인 위치 복원 불가 → marker(PDF) 의 _persist_images_to_nas 로 NAS 영속 후 md 말미
    # 갤러리로 부착(docimg: ref = 뷰어 해석). OLE 수식/도형은 앵커도 raster 도 아니라 제외.
    # docx/xlsx/pptx/hwpx 는 이미지 미처리(기존 동작 유지).
    saved_images: list[dict[str, Any]] = []
    orphan_paths: list[str] = []
    if suffix == ".hwp" and MARKDOWN_IMAGE_PERSIST:
        if hwp_images:
            images_resp = [
                {
                    "bytes_b64": base64.b64encode(im["data"]).decode("ascii"),
                    "format": im.get("format") or "png",
                    "slug": "",
                    "width": None,
                    "height": None,
                }
                for im in hwp_images
            ]
            try:
                saved_images = _persist_images_to_nas(document_id, images_resp)
            except OSError as exc:
                # NAS 일시 끊김 등 — transient. queue retry 로 복구.
                logger.warning(
                    f"[marker] hwp image persist NAS write failed id={document_id}: "
                    f"{type(exc).__name__}: {exc}"
                )
                raise
            if saved_images:
                gallery = "\n\n## 첨부 이미지\n\n" + "\n\n".join(
                    f"![](docimg:{img['image_key']})" for img in saved_images
                )
                md_content = md_content + gallery
        # 재변환 시 현재 saved_images 기준으로 과거 document_images row/NAS 파일 정리.
        orphan_paths = await _sync_document_images(
            session, document_id, saved_images, {"engine": engine}
        )

    # 성공 — 계약상 md_content 는 비공백(빈출력은 raise). quality scored.
    quality = _compute_quality(md_content, doc.extracted_text or "", {"page_count": None})
    if saved_images:
        quality.setdefault("warnings", []).append(f"hwp_images_appended:{len(saved_images)}")

    await session.execute(
        update(Document).where(Document.id == document_id).values(
            md_content=md_content,
            md_status="success",
            md_extraction_engine=engine,
            md_extraction_engine_version=None,
            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()

    # 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] office success id={document_id} engine={engine} "
        f"len={len(md_content)} images={len(saved_images)}"
    )


async def _process_split(
    doc: Document,
    document_id: int,
    container_path: str,
    page_count: int,
    session: AsyncSession,
) -> None:
    """대형 PDF page-range 분할 변환.

    BATCH_PAGES 윈도우로 순차 /convert (각 호출 < MARKER_TIMEOUT). batch 단위로
    이미지 persist + ref rewrite 후 stitch. md_content = head+manifest (full blob 미저장).
    검색 인덱스는 기존 document_chunks(extracted_text window)가 담당 (marker_section 적재
    = commit 4 드롭). 결과: 전부 성공=success / 일부=partial / 전부 실패=failed.

    invariant: page numbering = 1-based inclusive (batch1: 1..BATCH_PAGES, ...).
    marker slug(`_page_0_*`) 는 batch 마다 재시작 → batch 별 rewrite 후 stitch (충돌 회피).
    """
    n_batches = (page_count + BATCH_PAGES - 1) // BATCH_PAGES
    succeeded: list[dict[str, Any]] = []   # {start_page, end_page, md}
    failed: list[dict[str, Any]] = []      # {start_page, end_page, error}
    all_saved: list[dict[str, Any]] = []
    engine: str | None = None
    engine_version: str | None = None
    images_truncated = False

    async with httpx.AsyncClient(timeout=MARKER_TIMEOUT) as client:
        for b in range(n_batches):
            start_page = b * BATCH_PAGES + 1
            end_page = min((b + 1) * BATCH_PAGES, page_count)
            try:
                resp = await client.post(
                    MARKER_ENDPOINT,
                    json={
                        "file_path": container_path,
                        "start_page": start_page,
                        "end_page": end_page,
                    },
                )
                resp.raise_for_status()
                data = resp.json()
            except httpx.HTTPStatusError as exc:
                # 404 = 파일 자체 부재 → 전체 doc fatal (batch 의미 없음).
                if exc.response.status_code == 404:
                    await _fail(
                        session, document_id,
                        f"file_not_found during split: {container_path}",
                    )
                    return
                msg = _http_error_message(exc)
                logger.warning(
                    f"[marker] split batch failed id={document_id} "
                    f"pages={start_page}-{end_page} status={exc.response.status_code}: {msg[:200]}"
                )
                failed.append({"start_page": start_page, "end_page": end_page, "error": msg[:500]})
                continue
            except Exception as exc:
                # transient(5xx/transport/timeout) 포함 — batch 실패로 기록(성공 batch 보존).
                # 복구는 re-enqueue(force_reprocess)로 전체 재시도.
                logger.warning(
                    f"[marker] split batch error id={document_id} "
                    f"pages={start_page}-{end_page} kind={type(exc).__name__}: {exc}"
                )
                failed.append({
                    "start_page": start_page, "end_page": end_page,
                    "error": f"{type(exc).__name__}: {exc}"[:500],
                })
                continue

            engine = data.get("engine") or engine
            engine_version = data.get("engine_version") or engine_version
            if data.get("images_truncated"):
                images_truncated = True

            md_raw = data["md_content"]
            images_resp = data.get("images") if MARKDOWN_IMAGE_PERSIST else None
            saved_batch: list[dict[str, Any]] = []
            if images_resp:
                try:
                    saved_batch = _persist_images_to_nas(
                        document_id, images_resp, seq_offset=len(all_saved)
                    )
                except OSError as exc:
                    logger.warning(
                        f"[marker] split image persist failed id={document_id} "
                        f"pages={start_page}-{end_page}: {type(exc).__name__}: {exc}"
                    )
                    failed.append({
                        "start_page": start_page, "end_page": end_page,
                        "error": f"image_persist: {type(exc).__name__}: {exc}"[:500],
                    })
                    continue

            slug_to_key = {img["source_slug"]: img["image_key"] for img in saved_batch}
            md_batch = _rewrite_image_refs(md_raw, slug_to_key)
            all_saved.extend(saved_batch)
            succeeded.append({"start_page": start_page, "end_page": end_page, "md": md_batch})

    manifest = _split_manifest(page_count, n_batches, succeeded, failed)

    # ---- 결과 판정 ----
    if not succeeded:
        await _fail(session, document_id, json.dumps(manifest, ensure_ascii=False)[:4000])
        logger.warning(f"[marker] split failed id={document_id} all {n_batches} batches failed")
        return

    md_status = "success" if not failed else "partial"
    stitched = "\n\n".join(b["md"] for b in succeeded)
    md_content = _build_large_md_content(stitched[:LARGE_DOC_MD_CONTENT_HEAD_CHARS], manifest)

    quality = _compute_quality(stitched, doc.extracted_text or "", {"page_count": page_count})
    quality.setdefault("metrics", {})["split"] = manifest
    if images_truncated:
        quality.setdefault("warnings", []).append("images_truncated")

    orphan_paths = await _sync_document_images(
        session, document_id, all_saved,
        {"engine": engine or "marker", "engine_version": engine_version},
    )

    await session.execute(
        update(Document).where(Document.id == document_id).values(
            md_content=md_content,
            md_status=md_status,
            md_extraction_engine=engine or "marker",
            md_extraction_engine_version=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 if md_status == "success"
                else json.dumps(manifest, ensure_ascii=False)[:4000]
            ),
            md_frontmatter=doc.md_frontmatter or {},
            md_format_version="1.0",
            content_origin="extracted",
        )
    )
    await session.commit()

    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] split done id={document_id} status={md_status} pages={page_count} "
        f"batches={n_batches} ok={len(succeeded)} failed={len(failed)} "
        f"images={len(all_saved)} md_head_len={len(md_content)}"
    )


def _http_error_message(exc: httpx.HTTPStatusError) -> str:
    """marker-service 에러 응답에서 code: message 추출 (JSON 실패 시 raw text)."""
    try:
        detail = exc.response.json().get("detail", {})
        return f'{detail.get("code", "unknown")}: {detail.get("message", exc.response.text)}'
    except Exception:
        return exc.response.text


def _split_manifest(
    total_pages: int, n_batches: int,
    succeeded: list[dict[str, Any]], failed: list[dict[str, Any]],
) -> dict[str, Any]:
    """split 결과 manifest — md_extraction_error/quality.metrics.split 공통."""
    return {
        "mode": "split",
        "total_pages": total_pages,
        "batch_pages": BATCH_PAGES,
        "batch_count": n_batches,
        "succeeded_batches": len(succeeded),
        "failed_batches": [
            {"start_page": f["start_page"], "end_page": f["end_page"], "error": f["error"]}
            for f in failed
        ],
        "search_index": "document_chunks(legacy/extracted_text window chunks)",
    }


def _build_large_md_content(head: str, manifest: dict[str, Any]) -> str:
    """대형 split 문서의 md_content = manifest 헤더 + head preview (full blob 미저장)."""
    failed_n = len(manifest["failed_batches"])
    badge = "partial" if failed_n else "success"
    header = (
        f"<!-- LARGE_DOC_SPLIT mode=split total_pages={manifest['total_pages']} "
        f"batch_pages={manifest['batch_pages']} batches={manifest['batch_count']} "
        f"failed={failed_n} -->\n"
        f"> 📄 **대형 문서 분할 변환** — 총 {manifest['total_pages']}p · "
        f"{manifest['batch_count']} batch (batch_pages={manifest['batch_pages']}) · 상태 **{badge}**\n"
        f"> 검색 인덱스 = 기존 `document_chunks` long_pdf/extracted_text window chunks. "
        f"아래는 Markdown 렌더링 preview.\n\n"
    )
    return header + head


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]], seq_offset: int = 0
) -> list[dict[str, Any]]:
    """marker 응답 이미지 list 를 NAS 에 저장하고 메타 dict 리스트 반환.

    image_key 는 sequence 기반 결정적 (`img_001` → `img_NNN`, marker 출력 순서 = 안정적).
    같은 doc 재변환 시 같은 key 가 같은 path 에 overwrite → idempotent.

    seq_offset: split 변환에서 batch 간 key 연속성 유지용 (batch1 0개 → batch2 offset=N).
    """
    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=seq_offset + 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, *, engine: str = "marker"
) -> None:
    """doc-level failed (재시도 무의미). engine = 실패한 변환 엔진(office=markitdown/libreoffice_hwp)."""
    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=engine,
            md_generated_at=_now(),
            content_origin="extracted",
        )
    )
    await session.commit()


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