hyungi_document_server/app/workers/statute_adapters/kr.py

"""KR 법령 어댑터 — 국가법령정보센터 (law.go.kr DRF) (plan safety-library-1 B-1 PR①).

poll_changes = lawSearch 목록 diff: 워치리스트 행별 정식 법령명 exact 조회 →
MST(법령일련번호) != watermark 이면 ChangeEvent. law_monitor 의 검증된 호출 형태 재사용.

fixture (2026-06-13 라이브 박제, tests/fixtures/statute_kr/):
  - lawsearch_*.xml — 목록 필드: 법령ID(불변)·법령일련번호(MST)·공포일자·시행일자·제개정구분명
  - lawservice_*.xml.gz — 전문 1콜 XML: 조문단위 853(산안기준규칙) + 별표단위 23 전부 포함
    = 스냅샷 의미론 확정(R7-M3 ①: annex 부분 fetch 실패 개념 없음 — 같은 응답에 없는
    별표 = 삭제 간주 가능). 별표번호+별표가지번호 = 구조화 필드(R7-M3 ② — suffix 문자열
    파싱 불요, version_key 합성은 이 필드 기반. PR② fetch_version 소관).
  - 조문 취득 방식 판정(R2-m1): 전문 1콜 + 로컬 파싱 확정 — lawjosub 조 단위 호출이면
    산안기준규칙(853조)은 개정당 호출 폭증. lawjosub fixture 는 보조 박제.

주의: 응답의 '법령상세링크' 필드에 OC 키가 포함됨 — fixture/로그에 raw 응답을 남길 때
새니타이즈 필수 (repo fixture 는 __OC_REDACTED__ 처리됨).
"""

import asyncio
import os
import xml.etree.ElementTree as ET

import httpx

from core.crawl_politeness import CRAWL_UA
from core.utils import setup_logger
from workers.statute_adapters import ChangeEvent

logger = setup_logger("statute_kr")

JURISDICTION = "KR"
SOURCE_API = "law.go.kr"

LAW_SEARCH_URL = "https://www.law.go.kr/DRF/lawSearch.do"
LAW_SERVICE_URL = "https://www.law.go.kr/DRF/lawService.do"

# 같은 도메인 연속 호출 간격 (일 1회 x 26콜 — 보수적)
_POLL_DELAY_S = 1.5


def _oc() -> str:
    oc = os.getenv("LAW_OC", "")
    if not oc:
        raise RuntimeError("LAW_OC 미설정 — statute KR 어댑터 사용 불가")
    return oc


def parse_search_hit(xml_text: str, official_title: str) -> dict | None:
    """lawSearch XML 에서 정식 법령명 exact match 1건 추출 (순수 함수 — fixture 테스트 대상).

    정식명 기준 exact match — 워치리스트 title 이 정식명(가운뎃점 포함)이므로 안전.
    (law_monitor 의 하드코딩 '유해위험작업...'(점 없음)이 영구 미매칭이던 함정의 교훈:
    조회 키는 반드시 레지스트리의 정식명을 쓴다.)
    """
    root = ET.fromstring(xml_text)
    for law in root.findall(".//law"):
        if (law.findtext("법령명한글") or "").strip() != official_title:
            continue
        mst = (law.findtext("법령일련번호") or "").strip()
        if not mst:
            continue
        return {
            "mst": mst,
            "law_id": (law.findtext("법령ID") or "").strip(),
            "promulgation_date": (law.findtext("공포일자") or "").strip() or None,
            "effective_date": (law.findtext("시행일자") or "").strip() or None,
            "revision_type": (law.findtext("제개정구분명") or "").strip() or None,
            "status_code": (law.findtext("현행연혁코드") or "").strip() or None,
        }
    return None


def detect_change(hit: dict | None, act_family_id: str, act_title: str,
                  watermark: str | None) -> ChangeEvent | None:
    """목록 hit + 워터마크 → ChangeEvent (순수 함수 — fixture 테스트 대상).

    - hit 없음 = 감지 불가 (None — 호출측이 fail-loud 로그. 폐지 단정 금지:
      검색 누락/표기 변경 가능성과 구분 불가하므로 repeal 은 제개정구분명 기준만)
    - MST == watermark = 변경 없음
    - 제개정구분명에 '폐지' = repeal, 그 외 = amend
    """
    if hit is None:
        return None
    if watermark and hit["mst"] == watermark:
        return None
    kind = "repeal" if (hit.get("revision_type") or "").find("폐지") >= 0 else "amend"
    return ChangeEvent(
        family_id=act_family_id,
        kind=kind,
        new_version_key=hit["mst"],
        title=act_title,
        promulgation_date=hit.get("promulgation_date"),
        effective_date=hit.get("effective_date"),
        revision_type=hit.get("revision_type"),
    )


async def poll_changes(client: httpx.AsyncClient, watch_rows: list) -> list[ChangeEvent]:
    """워치리스트 행별 lawSearch diff. 행 단위 실패 격리 (한 법령 실패가 나머지를 막지 않음)."""
    oc = _oc()
    events: list[ChangeEvent] = []
    for act in watch_rows:
        try:
            resp = await client.get(
                LAW_SEARCH_URL,
                params={"OC": oc, "target": "law", "type": "XML", "query": act.title},
                headers={"User-Agent": CRAWL_UA},
            )
            resp.raise_for_status()
            hit = parse_search_hit(resp.text, act.title)
            if hit is None:
                # fail-loud: 정식명 미매칭 = 표기 변경/검색 누락 의심 — 침묵 skip 금지
                logger.warning(f"[statute-kr] 목록 미매칭: {act.family_id} {act.title!r}")
            else:
                ev = detect_change(hit, act.family_id, act.title, act.watermark)
                if ev:
                    events.append(ev)
        except Exception as e:
            logger.error(f"[statute-kr] poll 실패 ({act.family_id}): {type(e).__name__}: {e!r}")
        await asyncio.sleep(_POLL_DELAY_S)
    return events