"""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(HWP5 binary) → pyhwp hwp5html → HTML → markdownify ← pyhwp+six 의존성. (2026-06-09: LibreOffice 번들 libhwplo 필터가 실제 한컴 HWP5 파일을 못 읽어 rc=0 + 'source file could not be loaded' 로 전건 실패 → 순수 Python HWP5 전용 변환기 pyhwp 로 교체.) - .hwpx → LibreOffice(headless) → HTML → markdownify ← markdownify 기존 의존성. (HWPX(zip)는 pyhwp 미지원 → LibreOffice 폴백 유지. 현재 코퍼스는 전부 HWP5 binary.) 실패 계약 (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 re 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 # extract_worker.py 가 이미 `libreoffice` 바이너리로 office 텍스트 추출에 성공(컨테이너 검증된 # 이름) → 기본값 정합. soffice 만 있는 환경은 LIBREOFFICE_BIN 으로 override. _SOFFICE_BIN = os.environ.get("LIBREOFFICE_BIN", "libreoffice") # pyhwp 콘솔 스크립트(pip install pyhwp 시 PATH 등록). HWP5 binary(.hwp) 전용. _HWP5HTML_BIN = os.environ.get("HWP5HTML_BIN", "hwp5html") # hwp5html 이 bindata/ 로 추출하는 첨부물 중 NAS 영속 대상 raster 확장자. # (OLE 수식/도형은 index.xhtml 에 앵커가 없어 위치 복원 불가 → 영속 제외.) _RASTER_EXTS = {"jpg", "jpeg", "png", "gif", "bmp"} 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) elif suffix == ".hwp": md = _via_pyhwp_html(p, timeout=timeout) else: # .hwpx (pyhwp 미지원 → LibreOffice 폴백) 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 _run_hwp5html(path: Path, *, timeout: int) -> tuple[str, list[dict]]: """HWP5 binary(.hwp) → (markdown, raster_images). hwp5html 1회 실행 = md + 이미지 동시 추출. LibreOffice 번들 libhwplo 필터가 실제 한컴 HWP5 파일을 못 읽어(rc=0 + 'source file could not be loaded') 전건 실패 → 순수 Python HWP5 전용 변환기 pyhwp(CLI hwp5html)로 교체. `_via_libreoffice_html` 와 동일한 실패 계약(rc≠0 또는 출력 부재 → OfficeMdError raise). raster_images = [{'data': bytes, 'format': 'jpeg'|'png'|...}] — bindata/ 의 래스터만. hwp5html 은 이미지를 본문 xhtml 에 로 앵커하지 않으므로(bindata orphan, --css/--html 동일) 인라인 위치는 복원 불가 → 호출부가 NAS 영속 후 말미 갤러리로 부착한다. """ try: from markdownify import markdownify # 기존 의존성 except ImportError as e: # noqa: BLE001 raise OfficeMdError("markdownify 미설치(기존 의존성이어야 함)") from e with tempfile.TemporaryDirectory(prefix="office_md_hwp_") as tmp: outdir = Path(tmp) # hwp5html --output /index.xhtml + styles.css + bindata/ cmd = [_HWP5HTML_BIN, "--output", str(outdir), str(path)] try: proc = subprocess.run( cmd, capture_output=True, text=True, timeout=timeout, check=False ) except FileNotFoundError as e: raise OfficeMdError( f"pyhwp(hwp5html) 바이너리 부재({_HWP5HTML_BIN}) — `pip install pyhwp six` 필요" ) from e except subprocess.TimeoutExpired as e: raise OfficeMdError(f"pyhwp 변환 타임아웃({timeout}s): {path.name}") from e index_path = outdir / "index.xhtml" if proc.returncode != 0 or not index_path.exists(): raise OfficeMdError( f"pyhwp html 변환 실패: {path.name} (rc={proc.returncode}): " f"{(proc.stderr or proc.stdout or '').strip()[:300]}" ) html = index_path.read_text(encoding="utf-8", errors="replace") # hwp5html 의 xhtml 은 최상단 선언을 가짐(LibreOffice 의 .html 경로엔 없음). # markdownify 의 html.parser 가 이를 PI 텍스트('xml version="1.0" encoding="utf-8"?')로 # 본문에 흘려 (1) md 최상단 잡음·검색/청크 오염, (2) 빈 body 셸일 때 그 ~34자가 # _MIN_BODY_CHARS(16) 빈출력 게이트를 무력화(빈 변환의 false-success) → markdownify 전에 제거. html = re.sub(r"^\s*<\?xml[^>]*\?>\s*", "", html) # 표 보존 위해 markdownify 가 table 을 GFM 으로 — heading_style ATX (libreoffice 경로와 동일). md = markdownify(html, heading_style="ATX", strip=["span", "font"]) images: list[dict] = [] bindata = outdir / "bindata" if bindata.is_dir(): for f in sorted(bindata.iterdir()): ext = f.suffix.lower().lstrip(".") if ext in _RASTER_EXTS: images.append({ "data": f.read_bytes(), "format": "jpeg" if ext == "jpg" else ext, }) return md, images def _via_pyhwp_html(path: Path, *, timeout: int) -> str: """HWP5 binary(.hwp) → markdown (이미지 제외). convert_office_to_md 단일 텍스트 경로용.""" md, _images = _run_hwp5html(path, timeout=timeout) return md def convert_hwp_to_md_and_images( path: str | Path, *, timeout: int = 90 ) -> tuple[str, list[dict]]: """HWP5(.hwp) → (markdown, raster_images). marker_worker 이미지 영속 경로 전용. 실패/빈출력 계약은 convert_office_to_md 와 동일(OfficeMdError raise / 빈 md 절대 반환 금지). raster_images 원소 = {'data': bytes, 'format': str}; 비어있을 수 있음(이미지 없는 문서). """ p = Path(path) if p.suffix.lower() != ".hwp": raise OfficeMdError(f"convert_hwp_to_md_and_images: .hwp 전용, got {p.suffix!r}") if not p.exists(): raise OfficeMdError(f"file not found: {p}") md, images = _run_hwp5html(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, images def _via_libreoffice_html(path: Path, *, timeout: int) -> str: """LibreOffice headless 로 HTML 변환 후 markdownify. hwpx 용(.hwp 는 pyhwp).""" 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), }