feat(papers): B-3 PR5 — 구매 PDF parent_doi 스탬프 (paper_doi_reconcile 통합)

plan safety-library-b3-1 PR5. Papers_Purchased 수동 드롭 PDF(license.restricted=true)를 서지 holder 에
연결: 본문 DOI 파싱 → paper.parent_doi 링크(child, doi 미보유=인덱스 밖, unique 무충돌).
- doi.py: parse_doi_from_text(본문 전체 DOI 정규식 — PDF 구조 무관).
- paper_doi_reconcile: restricted 분기 — restricted 행은 본문 DOI→parent_doi(child),
  그 외(레거시 arXiv)는 holder 스탬프(PR4). 쿼리에 parent_doi IS NULL 추가(링크분 재처리 회피).
- file_watcher merge-only license 주입 clobber-safe 존중. enqueue 0(콘텐츠 무변경).

단위 29 passed(+parse_doi_from_text). ephemeral PASS: 합성 restricted 행 → parent_doi 링크·
paper.doi 부재·restricted 보존·스키마 수용(insert+rollback). reconcile 멱등(재실행 0 변경).
실 구매 PDF 라이브 검증 = 사용자 첫 논문 구매·드롭 시(로직 검증 완료).

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

app/workers/paper_doi_reconcile.py

@@ -1,14 +1,13 @@
-"""레거시 paper 행 DOI reconcile — B-3 PR4 (plan safety-library-b3-1).
+"""paper DOI reconcile — B-3 PR4(레거시 arXiv) + PR5(구매 PDF) (plan safety-library-b3-1).
 
-paper.doi 없는 paper 행(레거시 194 arXiv/ASME 초록 + 수집기 누락분)을 arXiv DataCite DOI 로 스탬프해
-partial-unique 인덱스에 편입 → 향후 수집기 재유입을 find_paper_holder 가 차단('동일-DOI 재유입 차단만').
+paper.doi/parent_doi 둘 다 없는 paper 행을 두 갈래로 정리:
+- 레거시 arXiv 초록(holder): arXiv id → arxiv_doi(10.48550/arxiv.{id}) 스탬프 → partial-unique
+  인덱스 편입 → 재유입 차단('동일-DOI 재유입 차단만').
+- 구매 PDF(child, license.restricted=true — Papers_Purchased 드롭): 본문 DOI 파싱 → paper.parent_doi
+  링크(서지 holder 와 DOI 공유로 연결). child 는 doi 미보유(인덱스 밖) → unique 무충돌.
 
-- KEYLESS·결정적: arXiv id(paper.arxiv_id 또는 extracted_text 파싱) → arxiv_doi(10.48550/arxiv.{id},
-  OpenAlex canonical 실측 일치). OpenAlex 호출 불요. arXiv id 없는 ASME RSS 행은 skip.
-- ★dedup_reconcile(file_hash 캐시 재계산·무IO)와 **별 worker** — 적대리뷰 B·C major: 외부키/네트워크
-  실패모드를 캐시 무결성 잡에 결합하지 않음(여긴 keyless 라 네트워크도 없음, 순수 in-DB 메타 갱신).
-- 이미 같은 DOI holder 존재 = 선재 중복 → 스탬프 대신 parent_doi 마킹(unique 위반 회피).
-- 콘텐츠(extracted_text) 무변경, 메타만 갱신 → 어떤 stage 도 enqueue 안 함(summarize/embed/chunk 0 자명).
+- KEYLESS·결정적(OpenAlex 호출 0)·in-DB·enqueue 0(콘텐츠 무변경). dedup_reconcile(file_hash 캐시)와
+  별 worker(적대리뷰 B·C major). 선재 DOI holder 존재 시 arXiv 행도 parent_doi 마킹(unique 위반 회피).
 """
 
 import asyncio
@@ -18,21 +17,37 @@ from sqlalchemy import select
 from core.database import async_session
 from core.utils import setup_logger
 from models.document import Document
-from services.papers.doi import arxiv_doi, parse_arxiv_id, with_paper_doi, with_parent_doi
+from services.papers.doi import (
+    arxiv_doi,
+    parse_arxiv_id,
+    parse_doi_from_text,
+    with_paper_doi,
+    with_parent_doi,
+)
 from services.papers.holder import find_paper_holder
 
 logger = setup_logger("paper_doi_reconcile")
 
 _DOI_TEXT = Document.extract_meta[("paper", "doi")].astext
+_PARENT_DOI_TEXT = Document.extract_meta[("paper", "parent_doi")].astext
+
+
+def _is_restricted(meta: dict) -> bool:
+    return (meta.get("license") or {}).get("restricted") in (True, "true")
 
 
 async def run(limit: int = 0) -> None:
-    """paper.doi 없는 paper 행을 arXiv DOI 로 스탬프(멱등). limit=0 = 전건."""
+    """paper.doi/parent_doi 없는 paper 행 reconcile(멱등). limit=0 = 전건."""
     stamped = marked_dup = skipped_no_arxiv = 0
+    linked_purchased = skipped_purchased_no_doi = 0
     async with async_session() as session:
         q = (
             select(Document)
-            .where(Document.material_type == "paper", _DOI_TEXT.is_(None))
+            .where(
+                Document.material_type == "paper",
+                _DOI_TEXT.is_(None),
+                _PARENT_DOI_TEXT.is_(None),
+            )
             .order_by(Document.id)
         )
         if limit:
@@ -42,6 +57,18 @@ async def run(limit: int = 0) -> None:
         for row in rows:
             meta = dict(row.extract_meta or {})
             paper = dict(meta.get("paper") or {})
+
+            # PR5: 구매 PDF(restricted) = child → 본문 DOI 파싱 → parent_doi 링크
+            if _is_restricted(meta):
+                doi = parse_doi_from_text(row.extracted_text)
+                if not doi:
+                    skipped_purchased_no_doi += 1
+                    continue
+                row.extract_meta = with_parent_doi(meta, doi)
+                linked_purchased += 1
+                continue
+
+            # PR4: 레거시 arXiv 초록(holder) = arXiv DataCite DOI 스탬프
             arxiv_id = paper.get("arxiv_id") or parse_arxiv_id(row.extracted_text)
             doi = arxiv_doi(arxiv_id)
             if not doi:
@@ -51,26 +78,25 @@ async def run(limit: int = 0) -> None:
             meta["paper"] = paper
             holder = await find_paper_holder(session, doi)
             if holder is not None and holder.id != row.id:
-                # 선재 중복(다른 행이 이미 이 DOI holder) → 자식 마킹(인덱스 밖, unique 위반 회피)
-                row.extract_meta = with_parent_doi(meta, doi)
+                row.extract_meta = with_parent_doi(meta, doi)  # 선재 중복 → child 마킹
                 marked_dup += 1
             else:
-                # 스탬프 → 이 행이 holder, partial-unique 인덱스 진입 (재유입 차단 성립)
-                row.extract_meta = with_paper_doi(meta, doi)
+                row.extract_meta = with_paper_doi(meta, doi)   # holder 스탬프, 인덱스 진입
                 stamped += 1
-            # 콘텐츠 무변경 → enqueue 없음 (summarize/embed/chunk 0)
+            # 콘텐츠 무변경 → enqueue 없음(summarize/embed/chunk 0)
         await session.commit()
 
     logger.info(
-        f"[paper_doi_reconcile] paper.doi 없는 {len(rows)}행 → 스탬프 {stamped} · "
-        f"선재중복 마킹 {marked_dup} · arXiv id 없음 skip {skipped_no_arxiv}"
+        f"[paper_doi_reconcile] {len(rows)}행 → arXiv 스탬프 {stamped} · 선재중복 {marked_dup} · "
+        f"arXiv id 없음 skip {skipped_no_arxiv} / 구매PDF parent_doi 링크 {linked_purchased} · "
+        f"구매PDF DOI 없음 skip {skipped_purchased_no_doi}"
     )
 
 
 if __name__ == "__main__":
     import argparse
 
-    parser = argparse.ArgumentParser(description="레거시 paper DOI reconcile (arXiv DataCite, keyless)")
+    parser = argparse.ArgumentParser(description="paper DOI reconcile (arXiv 레거시 + 구매 PDF, keyless)")
     parser.add_argument("--limit", type=int, default=0, help="처리 상한(0=전건)")
     args = parser.parse_args()
     asyncio.run(run(limit=args.limit))