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

# -*- coding: utf-8 -*-
"""국가건설기준센터(KCSC) OpenApi 수집기.

CodeList로 전체 코드를 받고, 대상 KCS 코드의 CodeViewer 본문을 받아
표준시방서/<명칭>/KCS/<코드>_<이름>.md 로 저장한다.
API Key는 keyfile(_kcsc_key.txt)에서 읽는다.
"""

import json, re, sys, time, html, 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 / "표준시방서"))
SC = Path(str(DATA_DIR))
KEY = _load_key("KCSC_KEY")  # .secrets.local.md 또는 환경변수 KCSC_KEY
BASE = "https://kcsc.re.kr/OpenApi"
UA = {"User-Agent": "Mozilla/5.0"}


def api(path):
    url = f"{BASE}/{path}{'&' if '?' in path else '?'}key={KEY}"
    for k in range(3):
        try:
            with urllib.request.urlopen(urllib.request.Request(url, headers=UA), timeout=90) as r:
                return json.loads(r.read().decode("utf-8", "replace"))
        except Exception as e:
            if k == 2:
                print(f"   ! {path} :: {str(e)[:70]}")
                return None
            time.sleep(2)


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


# ── HTML → Markdown ──
def cell_text(td):
    t = re.sub(r"<br\s*/?>", " ", td)
    t = re.sub(r"<sup>(.*?)</sup>", r"^\1", t, flags=re.S)
    t = re.sub(r"<[^>]+>", "", t)
    t = html.unescape(t)
    return re.sub(r"\s+", " ", t).strip()


def table_to_md(tbl):
    cap = ""
    mcap = re.search(r"<caption[^>]*>(.*?)</caption>", tbl, re.S)
    if mcap:
        cap = cell_text(mcap.group(1))
    rows = []
    for tr in re.findall(r"<tr[^>]*>(.*?)</tr>", tbl, re.S):
        cells = [cell_text(td) for td in re.findall(r"<t[dh][^>]*>(.*?)</t[dh]>", tr, re.S)]
        if cells:
            rows.append(cells)
    if not rows:
        return cap
    w = max(len(r) for r in rows)
    rows = [r + [""] * (w - len(r)) for r in rows]
    esc = lambda c: c.replace("|", "\\|")
    out = []
    if cap:
        out.append(f"**{cap}**")
        out.append("")
    out.append("| " + " | ".join(esc(c) for c in rows[0]) + " |")
    out.append("|" + "|".join(["---"] * w) + "|")
    for r in rows[1:]:
        out.append("| " + " | ".join(esc(c) for c in r) + " |")
    return "\n".join(out)


def content_to_md(c):
    if not c:
        return ""
    c = c.strip()
    if "<table" in c:
        parts = []
        pos = 0
        for m in re.finditer(r"<table.*?</table>", c, re.S):
            pre = c[pos : m.start()]
            pt = re.sub(r"<[^>]+>", "", pre)
            pt = html.unescape(re.sub(r"\s+", " ", pt)).strip()
            if pt:
                parts.append(pt)
            parts.append(table_to_md(m.group(0)))
            pos = m.end()
        tail = re.sub(r"<[^>]+>", "", c[pos:])
        tail = html.unescape(re.sub(r"\s+", " ", tail)).strip()
        if tail:
            parts.append(tail)
        return "\n\n".join(parts)
    if "<img" in c:
        c = re.sub(r"<img[^>]*>", "〔그림〕", c)
    t = re.sub(r"<br\s*/?>", "\n", c)
    t = re.sub(r"<sup>(.*?)</sup>", r"^\1", t, flags=re.S)
    t = re.sub(r"<[^>]+>", "", t)
    return html.unescape(t).strip()


def viewer_to_md(doc):
    out = [f"# KCS {doc['code']} {doc['name']}", ""]
    out.append(
        f"> 버전 {doc.get('version', '')} | 수정 {(doc.get('updateDate') or '')[:10]} | fullCode {doc.get('fullCode', '')}"
    )
    out.append(f"> 출처: https://kcsc.re.kr/OpenApi/CodeViewer/{doc['codeType']}/{doc['code']}")
    out.append("")
    last_head = ""
    for it in doc.get("list", []):
        lvl = it.get("level", 1) or 1
        title = (it.get("title") or "").strip()
        label = (it.get("label") or "").strip()
        body = content_to_md(it.get("contents") or "")
        if label == "본문":
            if body and body != "내용 없음":
                out += [body, ""]
            continue
        norm_t = re.sub(r"\s+", "", title)
        # 직전 heading과 같은 title이 다시 나오면(상세 항목) heading 반복 생략
        if norm_t == last_head:
            plain = re.sub(r"\s+", "", body)
            if body and plain and plain != norm_t:
                out += [body, ""]
            continue
        h = "#" * min(lvl + 1, 6)
        out += [f"{h} {title}", ""]
        last_head = norm_t
        plain = re.sub(r"\s+", "", body)
        if body and plain and plain != norm_t:
            out += [body, ""]
    md, blank = [], 0
    for l in out:
        if l.strip() == "":
            blank += 1
            if blank > 1:
                continue
        else:
            blank = 0
        md.append(l)
    return "\n".join(md).strip() + "\n"


TARGETS = {
    "콘크리트 표준시방서 (KCS 14 20 00)": ["1420"],
    "도로공사 표준시방서 (KCS 44 00 00)": ["44"],
    "토목공사 표준시방서 (현 KCS 10 00 00 공통공사)": ["10"],
    "건설공사 비탈면 표준시방서 (KCS 11 70 00)": ["117", "114030"],
}


def run():
    codelist = api("CodeList")
    if not codelist:
        print("CodeList 실패")
        return
    (SC / "kcsc_codelist.json").write_text(
        json.dumps(codelist, ensure_ascii=False), encoding="utf-8"
    )
    kcs = [x for x in codelist if x["codeType"] == "KCS"]
    print(f"CodeList {len(codelist)}건 (KCS {len(kcs)})")

    summary = []
    for folder_name, prefixes in TARGETS.items():
        codes = sorted(
            {x["code"]: x for x in kcs if any(x["code"].startswith(p) for p in prefixes)}.items()
        )
        folder = ROOT / safe(folder_name) / "KCS"
        folder.mkdir(parents=True, exist_ok=True)
        idx = [
            f"# {folder_name} — KCS 코드 목록",
            "",
            f"> 국가건설기준센터 OpenApi 수집. 총 {len(codes)}개 코드.",
            "",
            "| 코드 | 이름 | 버전 | md |",
            "|---|---|---|---|",
        ]
        ok = 0
        for code, meta in codes:
            doc = api(f"CodeViewer/KCS/{code}")
            if not doc:
                idx.append(f"| KCS {code} | {meta['name']} | | 실패 |")
                continue
            d = doc[0]
            fn = safe(f"{code}_{d['name']}") + ".md"
            (folder / fn).write_text(viewer_to_md(d), encoding="utf-8")
            idx.append(f"| KCS {code} | {d['name']} | {d.get('version', '')} | [{fn}](<{fn}>) |")
            ok += 1
            print(f"  KCS {code} {d['name'][:30]} ({len(d.get('list', []))}절)", flush=True)
            time.sleep(0.4)
        (folder / "_목록.md").write_text("\n".join(idx) + "\n", encoding="utf-8")
        summary.append((folder_name, len(codes), ok))
        print(f"[{folder_name}] {ok}/{len(codes)}", flush=True)

    print("\n=== 요약 ===")
    for n, t, o in summary:
        print(f"  {o}/{t}  {n}")


if __name__ == "__main__":
    run()