Aislo/resources/knowledge/original/_pipeline/get_attach.py

# -*- coding: utf-8 -*-
"""고시·훈령 본문이 껍데기인 경우 실제 내용이 담긴 첨부파일/별표 원본(HWP)을 내려받는다."""

import re, time, urllib.request
import xml.etree.ElementTree as ET
from pathlib import Path
import os as _os
from pathlib import Path as _P

# 이 스크립트는 original/_pipeline/ 에 위치. 코퍼스 루트 = 상위 폴더.
ROOT_DIR = _P(__file__).resolve().parent.parent  # ...\original
DATA_DIR = _P(__file__).resolve().parent / "data"  # 생성 데이터(JSON)


# API 키: 커밋 금지 파일(law/.secrets.local.md) 또는 환경변수에서 읽는다.
def _load_key(name):
    v = _os.environ.get(name)
    if v:
        return v.strip()
    sec = ROOT_DIR.parent / ".secrets.local.md"
    if sec.exists():
        import re as _re

        for pat in (r"KCSC[\s\S]*?`([A-Za-z0-9]{20,})`", r"인증키[:\s]*`?([A-Za-z0-9]{30,})`?"):
            m = _re.search(pat, sec.read_text(encoding="utf-8"))
            if m:
                return m.group(1)
    return ""


ROOT = Path(str(ROOT_DIR))
UA = {"User-Agent": "Mozilla/5.0"}


def safe(s):
    return re.sub(r'[\\/:*?"<>|\n\r]', "_", s).strip().rstrip(".")


def get(url, tries=3):
    url = url.strip().replace("http://law.go.kr", "https://www.law.go.kr")
    if url.startswith("/"):
        url = "https://www.law.go.kr" + url
    for k in range(tries):
        try:
            with urllib.request.urlopen(urllib.request.Request(url, headers=UA), timeout=90) as r:
                return r.read()
        except Exception as e:
            if k == tries - 1:
                print(f"    ! {url[-40:]} :: {str(e)[:60]}")
                return None
            time.sleep(2)


tot_att = tot_hwp = 0
for xml in sorted(ROOT.rglob("현행_*.xml")):
    folder = xml.parent
    root = ET.parse(xml).getroot()

    # ── 1) 첨부파일 ──
    att = root.find("첨부파일")
    if att is not None and len(att):
        names = [e.text.strip() for e in att.findall("첨부파일명") if e.text]
        links = [e.text.strip() for e in att.findall("첨부파일링크") if e.text]
        d = folder / "첨부"
        for nm, lk in zip(names, links):
            p = d / safe(nm)
            if p.exists() and p.stat().st_size > 1000:
                tot_att += 1
                continue
            blob = get(lk)
            if not blob or len(blob) < 1000:
                continue
            d.mkdir(parents=True, exist_ok=True)
            p.write_bytes(blob)
            tot_att += 1
            print(f"  첨부 {len(blob) // 1024:6d}KB  {folder.name[:34]} / {nm[:44]}", flush=True)
            time.sleep(0.3)

    # ── 2) 별표 PDF가 안내문뿐인 경우 HWP 원본 확보 ──
    byl = root.find("별표")
    if byl is None:
        continue
    for b in byl.findall("별표단위"):
        title = (b.findtext("별표제목") or "").strip()
        if title.startswith("삭제"):
            continue
        num = (b.findtext("별표번호") or "0").lstrip("0") or "0"
        g = (b.findtext("별표가지번호") or "").lstrip("0")
        stem = safe(f"{b.findtext('별표구분')}{num}{('의' + g) if g else ''}_{title[:48]}")
        pdf = folder / "별표" / f"{stem}.pdf"
        if not pdf.exists():
            continue
        try:
            import pymupdf

            t = "".join(pg.get_text() for pg in pymupdf.open(pdf))
        except Exception:
            continue
        if len(re.sub(r"\s", "", t)) >= 120 and "자세한 내용은" not in t:
            continue  # 정상 PDF
        hlk = b.findtext("별표서식파일링크")
        hnm = b.findtext("별표HWP파일명") or f"{stem}.hwp"
        if not hlk:
            continue
        ext = Path(hnm.strip()).suffix or ".hwp"
        hp = folder / "별표" / f"{stem}{ext}"
        if hp.exists() and hp.stat().st_size > 1000:
            tot_hwp += 1
            continue
        blob = get(hlk)
        if not blob or len(blob) < 1000:
            continue
        hp.write_bytes(blob)
        tot_hwp += 1
        print(f"  HWP  {len(blob) // 1024:6d}KB  {folder.name[:34]} / {stem[:44]}", flush=True)
        time.sleep(0.3)

print(f"\n첨부파일 {tot_att}건 / 안내문 별표의 HWP 원본 {tot_hwp}건")