feat(markdown): hwp raster 이미지 NAS 영속 + library backfill 스크립트

pyhwp(hwp5html) 가 bindata/ 로 추출하는 raster 이미지를 NAS 에 영속한다. 기존엔
변환 tempdir 와 함께 폐기돼 경고 없이 silent 유실(도식·수식)이었다(적대 리뷰 MEDIUM).

- office_md.py: _run_hwp5html 으로 hwp5html 1회 실행 → (markdown, raster_images).
  convert_hwp_to_md_and_images() 신규 = marker_worker 이미지 경로용. hwp5html 은 이미지를
  본문 xhtml 에 <img> 앵커하지 않아(--css/--html 동일) 인라인 위치 복원 불가 → 호출부가
  말미 갤러리로 부착. OLE 수식/도형은 앵커도 raster 도 아니라 영속 제외.
- marker_worker._process_office: .hwp raster 를 marker(PDF)의 _persist_images_to_nas 로
  NAS 영속 + document_images UPSERT(_sync_document_images, 재변환 orphan 정리) + md 말미
  ## 첨부 이미지 docimg: 갤러리 + quality.warnings hwp_images_appended. docx/xlsx/pptx/
  hwpx 는 이미지 미처리(기존 동작 유지).
- scripts/backfill_hwp_library.py: 지정 PKM 폴더 .hwp 를 content-hash dedup(Inbox 중복 +
  _1/카피본 사본 흡수) 후 category=library 일회성 ingest.

검증(E2E): Knowledge/Engineering 18개 → dedup 후 신규 5개(산업안전기사 3~7과목) ingest,
5/5 success. 제4과목 raster 3장 → NAS extracted_images/35778/img_001~003.jpeg 실재 +
document_images 3 row(engine=pyhwp) + md 갤러리 docimg ref. 이미지 없는 문서는 갤러리
미생성. 텍스트/표 경로 회귀 0(기존 4건 재변환 success).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

scripts/backfill_hwp_library.py

@@ -0,0 +1,110 @@
+"""HWP(library) 백필 — 지정 PKM 폴더의 .hwp 를 content-hash dedup 후 일회성 ingest.
+
+산업안전기사 등 외부 학습자료(category='library')를 코퍼스에 편입한다. file_watcher 의
+PKM 트랙 로직을 재사용하되 dedup 을 file_path 가 아닌 **file_hash** 기준으로 해서
+(a) Inbox 중복 (b) `_1`/`카피본` 사본을 1건으로 수렴시킨다(file_watcher 는 path dedup 이라
+동일내용 다른경로를 중복 ingest 함). 이후 파이프라인:
+  extract(텍스트) → classify → embed/chunk(검색) → markdown(.hwp=pyhwp hwp5html + raster NAS 영속)
+
+실행 (GPU 서버):
+  # dry-run (기본) — 무엇이 ingest/skip 될지만 출력
+  docker exec hyungi_document_server-fastapi-1 \
+    python /app/scripts/backfill_hwp_library.py --subdir Knowledge/Engineering
+
+  # 실제 ingest
+  docker exec hyungi_document_server-fastapi-1 \
+    python /app/scripts/backfill_hwp_library.py --subdir Knowledge/Engineering --commit
+"""
+
+import argparse
+import asyncio
+import sys
+from pathlib import Path
+
+from sqlalchemy import select
+
+from core.config import settings
+from core.database import async_session
+from core.utils import file_hash
+from models.document import Document
+from models.queue import enqueue_stage
+
+
+async def run(subdir: str, commit: bool) -> int:
+    nas_root = Path(settings.nas_mount_path)
+    scan_root = nas_root / "PKM" / subdir
+    if not scan_root.exists():
+        print(f"[backfill] scan_root 부재: {scan_root}", file=sys.stderr)
+        return 2
+
+    files = sorted(
+        p for p in scan_root.rglob("*") if p.is_file() and p.suffix.lower() == ".hwp"
+    )
+    print(f"[backfill] {scan_root} 하위 .hwp {len(files)}개 발견 / mode={'COMMIT' if commit else 'DRY-RUN'}")
+
+    ingested = skipped_existing = skipped_batchdup = 0
+    seen_hashes: set[str] = set()
+
+    async with async_session() as session:
+        for fp in files:
+            rel_path = str(fp.relative_to(nas_root))
+            fhash = file_hash(fp)
+
+            if fhash in seen_hashes:
+                print(f"  SKIP(batch-dup)    {rel_path}")
+                skipped_batchdup += 1
+                continue
+            seen_hashes.add(fhash)
+
+            # content-hash dedup (path 무관) — Inbox 중복 + _1/카피본 사본 흡수
+            existing = (
+                await session.execute(
+                    select(Document.id).where(Document.file_hash == fhash).limit(1)
+                )
+            ).first()
+            if existing:
+                print(f"  SKIP(exists id={existing[0]})  {rel_path}")
+                skipped_existing += 1
+                continue
+
+            ingested += 1
+            if not commit:
+                print(f"  INGEST(dry)        {rel_path}")
+                continue
+
+            doc = Document(
+                file_path=rel_path,
+                file_hash=fhash,
+                file_format="hwp",
+                file_size=fp.stat().st_size,
+                file_type="immutable",
+                title=fp.stem,
+                source_channel="drive_sync",
+                category="library",
+                needs_conversion=False,
+            )
+            session.add(doc)
+            await session.flush()
+            await enqueue_stage(session, doc.id, "extract")
+            print(f"  INGEST id={doc.id}      {rel_path}")
+
+        if commit:
+            await session.commit()
+
+    print(
+        f"[backfill] done — ingest={ingested} "
+        f"skip_existing={skipped_existing} skip_batchdup={skipped_batchdup}"
+    )
+    return 0
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--subdir", default="Knowledge/Engineering", help="PKM 하위 스캔 폴더")
+    ap.add_argument("--commit", action="store_true", help="실제 ingest (없으면 dry-run)")
+    args = ap.parse_args()
+    return asyncio.run(run(args.subdir, args.commit))
+
+
+if __name__ == "__main__":
+    sys.exit(main())