hyungi_document_server/scripts/backfill_nonpdf_markdown.py

"""과거 office/hwp pending 문서 markdown stage 백필 — plan ds-s1-backend-1 C-4.

신규 ingest 는 classify→markdown 전이(queue_consumer.py:142)로 자동 도달하므로 이 스크립트는
*과거* office/hwp 행만 다룬다. C-2 가 office_md 변환을 붙이기 전까지 markdown stage 에서
skip 된 행들을 다시 큐에 넣어 md_content 를 생성한다.

대상 (WHERE):
  - file_format IN (office_md 지원 실값: docx, xlsx, pptx, hwp, hwpx)
    ★ 제외 축 = file_format. INCLUDE 필터가 article(file_format='article')을 구조적으로 배제
      → P0-3 가드(md 없는 article 이 completed 도달 금지, correctness-critical). source_channel 절 불필요.
    ★ 레거시 바이너리(.doc/.xls/.ppt)는 markitdown 미지원 → 기본 목록 제외(넣어도 marker 가 skip).
  - md_status = 'pending'  (이미 success/failed/skipped 는 건드리지 않음)
  - extracted_text IS NOT NULL  (폴백 존재 모집단)
      C-5 failed-postcondition 상속: 변환 실패는 md_status='failed' 로 시끄럽게 남는다(앱이
      '변환 실패' 표시). extracted_text NULL office(폴백 없음)는 배제 — 실패 시 더 시끄러운
      별 집합이라 phase2 재검토(C-4 배제 honest).

스케줄:
  ★ C-2 라이브 office ingestion 과 백필 창 비중첩 — markdown 워커는 BATCH=1 직렬이라
    야간 단발로 돌려 라이브 office 업로드 stall 회피(plan C-2 reflection).

실행:
  docker compose exec fastapi python /app/scripts/backfill_nonpdf_markdown.py --dry-run
  docker compose exec fastapi python /app/scripts/backfill_nonpdf_markdown.py --apply
  docker compose exec fastapi python /app/scripts/backfill_nonpdf_markdown.py --apply --limit 50
"""

import argparse
import asyncio
import json
import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app"))

from sqlalchemy import bindparam, text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

# office_md 가 실제 변환하는 file_format(확장자 소문자, 점 없음). 단일 source.
DEFAULT_FORMATS = ("docx", "xlsx", "pptx", "hwp", "hwpx")

CANDIDATES_SQL = text(
    """
    SELECT id, file_format, title, file_path
    FROM documents
    WHERE deleted_at IS NULL
      AND md_status = 'pending'
      AND extracted_text IS NOT NULL
      AND file_format IN :formats
    ORDER BY id
    """
).bindparams(bindparam("formats", expanding=True))

# 활성 markdown 큐 행이 없는 doc 만 통과 (UNIQUE 부분 인덱스). 충돌 = silent skip.
ENQUEUE_SQL = text(
    """
    INSERT INTO processing_queue (document_id, stage, status, payload)
    VALUES (:doc_id, 'markdown', 'pending', CAST(:payload AS jsonb))
    ON CONFLICT DO NOTHING
    """
)


def _chunks(seq, size):
    for i in range(0, len(seq), size):
        yield seq[i : i + size]


async def run(*, apply: bool, formats: tuple[str, ...], limit: int | None, chunk_size: int) -> int:
    database_url = os.getenv(
        "DATABASE_URL", "postgresql+asyncpg://pkm:pkm@localhost:5432/pkm"
    )
    engine = create_async_engine(database_url)
    session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

    try:
        async with session_factory() as session:
            rows = (
                await session.execute(CANDIDATES_SQL, {"formats": list(formats)})
            ).all()
            if limit:
                rows = rows[:limit]

            print(f"=== office/hwp pending 후보 = {len(rows)}건 (formats={','.join(formats)}) ===")
            if not rows:
                print("후보 없음 — 종료.")
                return 0

            by_fmt: dict[str, int] = {}
            for r in rows:
                by_fmt[r.file_format] = by_fmt.get(r.file_format, 0) + 1
            print("포맷별:", ", ".join(f"{k}={v}" for k, v in sorted(by_fmt.items())))
            for r in rows[:20]:
                print(f"  id={r.id:>7}  {r.file_format:<5}  {(r.title or '')[:70]}")
            if len(rows) > 20:
                print(f"  ... 외 {len(rows) - 20}건")

            if not apply:
                print(f"\n[dry-run] {len(rows)}건 markdown 큐 enqueue 예정. --apply 로 실제 적용.")
                print("        (적용 전 C-2 라이브 office ingestion 과 비중첩 야간창 확인.)")
                return 0

            payload = json.dumps(
                {"force_reprocess": True, "reason": "c4_nonpdf_markdown_backfill"}
            )
            inserted = 0
            processed = 0
            for batch in _chunks(rows, chunk_size):
                for r in batch:
                    result = await session.execute(
                        ENQUEUE_SQL, {"doc_id": r.id, "payload": payload}
                    )
                    if result.rowcount > 0:
                        inserted += 1
                await session.commit()
                processed += len(batch)
                print(f"[apply] {processed}/{len(rows)} 처리 (enqueue 누적 {inserted})")

            print(f"\n[apply] 완료 — {inserted}/{len(rows)} 신규 markdown 큐 추가.")
            print("        (skip = 이미 활성 markdown 큐 행이 있는 문서)")
            return 0
    finally:
        await engine.dispose()


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--apply", action="store_true", help="실제 enqueue (기본 dry-run)")
    parser.add_argument("--dry-run", action="store_true", help="명시적 dry-run (default 동등)")
    parser.add_argument(
        "--formats", type=str, default=",".join(DEFAULT_FORMATS),
        help=f"쉼표 구분 file_format (기본 {','.join(DEFAULT_FORMATS)})",
    )
    parser.add_argument("--limit", type=int, default=None, help="후보 상한(샘플 검증용)")
    parser.add_argument("--chunk", type=int, default=200, help="enqueue txn 청크 크기")
    args = parser.parse_args()
    if args.apply and args.dry_run:
        parser.error("--apply 와 --dry-run 동시 지정 불가")
    formats = tuple(f.strip().lower() for f in args.formats.split(",") if f.strip())
    return asyncio.run(
        run(apply=args.apply, formats=formats, limit=args.limit, chunk_size=args.chunk)
    )


if __name__ == "__main__":
    sys.exit(main())