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

# -*- coding: utf-8 -*-
"""e나라 표준인증에서 KS 표준 메타데이터·개정이력 수집 (원문은 DRM 열람 전용이라 미수집)."""

import json, re, time, urllib.parse, urllib.request
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 / "KS"))
SC = Path(str(DATA_DIR))
BASE = "https://www.standard.go.kr/KSCI/standardIntro/getStandardSearchView.do"
UA = {"User-Agent": "Mozilla/5.0"}


def fetch(ks):
    url = f"{BASE}?menuId=919&topMenuId=502&upperMenuId=503&ksNo={urllib.parse.quote(ks)}"
    for k in range(3):
        try:
            with urllib.request.urlopen(urllib.request.Request(url, headers=UA), timeout=45) as r:
                return r.read().decode("utf-8", "replace")
        except Exception as e:
            if k == 2:
                print(f"   ! {ks} :: {str(e)[:60]}")
                return None
            time.sleep(2)


def flatten(h):
    h = re.sub(r"<script.*?</script>", "", h, flags=re.S)
    h = re.sub(r"<style.*?</style>", "", h, flags=re.S)
    h = re.sub(r"<!--.*?-->", "", h, flags=re.S)
    t = re.sub(r"<[^>]+>", "|", h)
    t = t.replace("&nbsp;", " ").replace("&amp;", "&").replace("&lt;", "<").replace("&gt;", ">")
    t = re.sub(r"\s+", " ", t)
    t = re.sub(r"(\|\s*)+", "|", t)
    return t


def field(t, key, stop=("|",)):
    m = re.search(r"\|" + re.escape(key) + r"\|+([^|]*)", t)
    return m.group(1).strip() if m else ""


def parse(ks, h):
    t = flatten(h)
    i = t.find("|기본정보|")
    seg = t[i : i + 4000] if i > 0 else t
    d = {
        "표준번호": field(seg, "표준번호") or ks,
        "표준명": field(seg, "표준명(한글)"),
        "표준명_영문": field(seg, "표준명(영문)"),
        "표준분야": field(seg, "표준분야"),
        "표준구분": field(seg, "표준구분"),
        "제정일": field(seg, "제정일"),
        "최종개정확인일": field(seg, "최종개정확인일"),
        "적용범위": field(seg, "적용범위"),
        "담당부처": field(seg, "담당부처"),
        "표준개발협력기관": field(seg, "표준개발협력기관"),
        "ICS": field(seg, "ICS코드"),
    }
    # 이력사항: 변경일자 / 구분 / 고시번호 반복
    hist = []
    j = t.find("표준 이력사항")
    if j > 0:
        seg2 = t[j : j + 6000]
        for m in re.finditer(
            r"\|변경일자\|([0-9\-]{8,10})\s*\|구분\|([^|]*)\|고시번호\|([^|]*)", seg2
        ):
            hist.append(
                {
                    "일자": m.group(1).strip(),
                    "구분": m.group(2).strip(),
                    "고시번호": m.group(3).strip(),
                }
            )
    d["이력"] = hist
    d["상태"] = (
        "폐지"
        if any(x["구분"] == "폐지" for x in hist)
        else ("현행" if d["표준명"] else "확인필요")
    )
    return d


CODES = json.load(open(SC / "ks_codes.json", encoding="utf-8"))
ROOT.mkdir(parents=True, exist_ok=True)
res = []
for i, ks in enumerate(CODES, 1):
    key = ks.replace(" ", "")
    h = fetch(key)
    if not h:
        res.append({"표준번호": ks, "상태": "조회실패"})
        continue
    d = parse(ks, h)
    d["조회번호"] = key
    res.append(d)
    print(
        f"[{i}/{len(CODES)}] {ks} :: {d['상태']} | {d['표준명'][:34]} | 개정 {d['최종개정확인일']} | 이력 {len(d['이력'])}",
        flush=True,
    )
    time.sleep(0.6)

json.dump(res, open(SC / "ks_meta.json", "w", encoding="utf-8"), ensure_ascii=False, indent=1)