hyungi_document_server/scripts/backfill_dedup.py

"""기존 file_hash 중복 그룹 backfill — plan ds-s1-backend-1 B-4.

목적:
  A-1 migration 287 로 추가된 duplicate_of / duplicate_count 를 *기존* 중복 그룹에 채운다.
  migration(단일 트랜잭션)과 분리한 별 배치(database.py:29-30 정책 — 대량 UPDATE 를
  startup migration 에 넣지 않는다). 업로드 시점 채움(B-1)은 신규 행만 다루므로 과거는 이 스크립트.

판정:
  - file_hash exact 그룹(OFF-whitelist=law_monitor 제외, deleted 제외, count>1).
    near_duplicate 는 영속화 보류(on-the-fly) — 여기서 다루지 않는다.
  - canonical = 그룹 최古(min id). canonical.duplicate_of=NULL, duplicate_count=group_size-1.
  - 비-canonical 멤버 = duplicate_of=canonical, duplicate_count=0.

안전:
  - 멱등 — 이미 목표값인 행은 UPDATE 안 함(재실행 안전). --dry-run 이 적용될 정확한 set 미리보기.
  - --chunk(기본 500)행/txn 청크 커밋 — 28,941행 단일 트랜잭션 lock 회피.

실행:
  docker compose exec fastapi python /app/scripts/backfill_dedup.py --dry-run
  docker compose exec fastapi python /app/scripts/backfill_dedup.py --apply
  # 변경 전 안전망은 E-3 pre-B-4 pg_dump (별 단계).
"""

import argparse
import asyncio
import os
import sys

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

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

from services.dedup import reconcile_dedup  # 코어 재계산 (야간 잡과 공유)


async def run(*, apply: bool, 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:
            result = await reconcile_dedup(session, apply=apply, chunk_size=chunk_size)

            print(f"=== dedup 그룹 {result['groups']}개 · 관련 문서 {result['docs']}건 ===")
            if result["groups"] == 0:
                print("dedup 그룹 없음(OFF-whitelist 제외 후 count>1 없음) — 종료.")
                return 0

            already = result["docs"] - result["changes"]
            print(f"변경 필요 {result['changes']}건 / 이미 목표값 {already}건 (멱등)")
            if result["changes"] == 0:
                print("모두 목표값 — 적용할 변경 없음.")
                return 0

            # 적용될/된 정확한 UPDATE set 미리보기 (상위 40건)
            print("\n=== UPDATE set (id → duplicate_of / duplicate_count) ===")
            for s in result["sample"]:
                role = "canonical" if s["duplicate_of"] is None else f"dup→{s['duplicate_of']}"
                print(
                    f"  id={s['id']:>7}  duplicate_of={s['duplicate_of']}  "
                    f"duplicate_count={s['duplicate_count']}  [{role}]"
                )
            if result["changes"] > len(result["sample"]):
                print(f"  ... 외 {result['changes'] - len(result['sample'])}건")

            if not apply:
                print(f"\n[dry-run] {result['changes']}건 변경 예정. --apply 로 실제 적용.")
            else:
                print(f"\n[apply] 완료 — {result['applied']}건 갱신.")
            return 0
    finally:
        await engine.dispose()


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--apply", action="store_true", help="실제 적용 (기본 dry-run)")
    parser.add_argument("--dry-run", action="store_true", help="명시적 dry-run (default 동등)")
    parser.add_argument("--chunk", type=int, default=500, help="txn 당 UPDATE 행 수 (기본 500)")
    args = parser.parse_args()
    if args.apply and args.dry_run:
        parser.error("--apply 와 --dry-run 동시 지정 불가")
    return asyncio.run(run(apply=args.apply, chunk_size=args.chunk))


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