hyungi_document_server/app/workers/office_md.py

"""office/hwp → Markdown 하이브리드 변환기 (plan ds-s1-backend-1, C-1 PoC).

★ PoC 상태 — marker_worker 에 아직 연결하지 않음(그건 C-2). 본 모듈은 변환 *계약*과
   PoC 하니스(scripts/poc_office_md.py)가 호출하는 순수 함수만 제공한다.

전략 (하이브리드):
  - OOXML(.docx/.xlsx/.pptx) → markitdown        ← 신규 의존성(pip install markitdown). lazy import.
  - .hwp/.hwpx               → LibreOffice(headless) → HTML → markdownify  ← markdownify 기존 의존성.
    (LibreOffice 가 hwp import 필터 보유. .hwpx 는 .hwp 와 다른 필터·버전 의존 → E-1: prod LibreOffice
     버전핀 안전컨텍스트에서 PoC 실행. 표 fidelity 가 진짜 리스크 — 하니스가 측정.)

실패 계약 (C-5 postcondition 의 backend 절반):
  변환 실패·빈 출력·타임아웃·의존성 부재 → OfficeMdError 를 raise 한다.
  **success + 빈 md 를 절대 반환하지 않는다** — 호출부(C-2 marker_worker)가 이를 잡아
  md_status='failed'(¬success·¬skipped) 로 라우팅한다. 불변식: md_status ∈ {success,partial} ⟹ md_content 非공백.
"""

from __future__ import annotations

import os
import shutil
import subprocess
import tempfile
from pathlib import Path

OOXML_FORMATS = {".docx", ".xlsx", ".pptx"}
HWP_FORMATS = {".hwp", ".hwpx"}
SUPPORTED = OOXML_FORMATS | HWP_FORMATS

# 빈 출력 판정 임계 — 공백 제거 후 이 미만이면 '실패(빈 변환)'로 본다.
_MIN_BODY_CHARS = 16

_SOFFICE_BIN = os.environ.get("LIBREOFFICE_BIN", "soffice")


class OfficeMdError(Exception):
    """office/hwp → md 변환 실패 신호. 호출부는 md_status='failed' 로 라우팅."""


def convert_office_to_md(path: str | Path, *, timeout: int = 90) -> str:
    """office/hwp 파일을 Markdown 문자열로 변환. 실패/빈출력 시 OfficeMdError raise."""
    p = Path(path)
    suffix = p.suffix.lower()
    if suffix not in SUPPORTED:
        raise OfficeMdError(f"unsupported suffix for office_md: {suffix!r}")
    if not p.exists():
        raise OfficeMdError(f"file not found: {p}")

    if suffix in OOXML_FORMATS:
        md = _via_markitdown(p)
    else:  # .hwp / .hwpx
        md = _via_libreoffice_html(p, timeout=timeout)

    md = (md or "").strip()
    if len(md) < _MIN_BODY_CHARS:
        raise OfficeMdError(f"empty/too-short conversion ({len(md)} chars) for {p.name}")
    return md


def _via_markitdown(path: Path) -> str:
    try:
        from markitdown import MarkItDown  # lazy — 신규 의존성
    except ImportError as e:  # noqa: BLE001
        raise OfficeMdError(
            "markitdown 미설치 (OOXML 변환에 필요) — `pip install markitdown`. "
            "C-1 PoC 는 prod worker 이미지/버전핀 컨텍스트에서 실행(E-1)."
        ) from e
    try:
        result = MarkItDown().convert(str(path))
    except Exception as e:  # noqa: BLE001 — 어떤 변환 예외든 failed 로 라우팅
        raise OfficeMdError(f"markitdown 변환 실패: {path.name}: {e}") from e
    return getattr(result, "text_content", "") or ""


def _via_libreoffice_html(path: Path, *, timeout: int) -> str:
    """LibreOffice headless 로 HTML 변환 후 markdownify. hwp/hwpx 용."""
    try:
        from markdownify import markdownify  # 기존 의존성
    except ImportError as e:  # noqa: BLE001
        raise OfficeMdError("markdownify 미설치(기존 의존성이어야 함)") from e

    with tempfile.TemporaryDirectory(prefix="office_md_") as tmp:
        tmpdir = Path(tmp)
        # soffice 동시 실행 시 user profile 락 충돌 회피 — 호출별 격리 프로필.
        profile = tmpdir / "lo_profile"
        cmd = [
            _SOFFICE_BIN,
            "--headless",
            "--nologo",
            "--nofirststartwizard",
            f"-env:UserInstallation=file://{profile}",
            "--convert-to",
            "html",
            "--outdir",
            str(tmpdir),
            str(path),
        ]
        try:
            proc = subprocess.run(
                cmd, capture_output=True, text=True, timeout=timeout, check=False
            )
        except FileNotFoundError as e:
            raise OfficeMdError(
                f"LibreOffice 바이너리 부재({_SOFFICE_BIN}) — LIBREOFFICE_BIN 설정 또는 설치 필요"
            ) from e
        except subprocess.TimeoutExpired as e:
            raise OfficeMdError(f"LibreOffice 변환 타임아웃({timeout}s): {path.name}") from e

        html_path = tmpdir / f"{path.stem}.html"
        if proc.returncode != 0 or not html_path.exists():
            raise OfficeMdError(
                f"LibreOffice html 변환 실패: {path.name} (rc={proc.returncode}): "
                f"{(proc.stderr or proc.stdout or '').strip()[:300]}"
            )
        html = html_path.read_text(encoding="utf-8", errors="replace")
        # 표 보존 위해 markdownify 가 table 을 GFM 으로 — heading_style ATX.
        return markdownify(html, heading_style="ATX", strip=["span", "font"])


def table_fidelity(md: str) -> dict:
    """E-1 표 fidelity 의 crude 지표 — GFM 표 행/구분행 카운트 (정밀 평가 아님, 회귀 신호)."""
    lines = md.splitlines()
    pipe_rows = sum(1 for ln in lines if ln.strip().startswith("|") and ln.strip().endswith("|"))
    sep_rows = sum(
        1 for ln in lines
        if ln.strip().startswith("|") and set(ln.strip()) <= set("|-: ")
    )
    return {
        "chars": len(md),
        "lines": len(lines),
        "table_pipe_rows": pipe_rows,
        "table_separator_rows": sep_rows,  # 표 개수의 근사
        "has_heading": any(ln.lstrip().startswith("#") for ln in lines),
    }