479 lines
18 KiB
Python
479 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""국가법령정보센터 원문 수집기.
|
||
|
||
항목별로 현행본 + 교본시점(2019-12-31 기준) 연혁본을 받아
|
||
<분류>/<명칭>/ 에 XML · Markdown · 별표 PDF · _meta.md · CHANGELOG.md 로 저장.
|
||
"""
|
||
|
||
import json, re, sys, time, urllib.parse, 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))
|
||
SC = Path(str(DATA_DIR))
|
||
BASEURL = "https://www.law.go.kr"
|
||
OC = _os.environ.get("LAW_OC", "umsangdon") # law.go.kr 인증 ID
|
||
교본기준일 = "20191231"
|
||
|
||
|
||
# ───────────────────────── 통신 ─────────────────────────
|
||
def fetch(url, binary=False, tries=3):
|
||
for k in range(tries):
|
||
try:
|
||
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
|
||
with urllib.request.urlopen(req, timeout=40) as r:
|
||
return r.read()
|
||
except Exception as e:
|
||
if k == tries - 1:
|
||
print(f" ! 실패 {url[:90]} :: {str(e)[:60]}")
|
||
return None
|
||
time.sleep(1.5)
|
||
|
||
|
||
def search(target, query, nw=None, display=100):
|
||
url = (
|
||
f"{BASEURL}/DRF/lawSearch.do?OC={OC}&type=XML&target={target}"
|
||
f"&display={display}&query={urllib.parse.quote(query)}"
|
||
)
|
||
if nw:
|
||
url += f"&nw={nw}"
|
||
b = fetch(url)
|
||
try:
|
||
return ET.fromstring(b)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def service(target, key, kind="MST"):
|
||
url = f"{BASEURL}/DRF/lawService.do?OC={OC}&type=XML&target={target}&{kind}={key}"
|
||
b = fetch(url)
|
||
try:
|
||
return ET.fromstring(b), b
|
||
except Exception:
|
||
return None, None
|
||
|
||
|
||
def T(node, *names):
|
||
for n in names:
|
||
el = node.find(n)
|
||
if el is not None and el.text:
|
||
return el.text.strip()
|
||
return ""
|
||
|
||
|
||
def ymd(s):
|
||
s = re.sub(r"\D", "", s or "")
|
||
return f"{s[:4]}.{s[4:6]}.{s[6:8]}" if len(s) == 8 else ""
|
||
|
||
|
||
def safe(s):
|
||
return re.sub(r'[\\/:*?"<>|]', "_", s).strip().rstrip(".")
|
||
|
||
|
||
# ─────────────────── XML → Markdown 변환 ───────────────────
|
||
JOMUN = re.compile(r"^(제\d+조(?:의\d+)?)\s*(?:\(([^)]*)\))?\s*(.*)$", re.S)
|
||
|
||
|
||
def split_jo(text):
|
||
"""조문 텍스트를 (헤딩, 본문)으로 분리."""
|
||
m = JOMUN.match(text.strip())
|
||
if not m:
|
||
return "", text
|
||
num, title, rest = m.group(1), m.group(2), m.group(3)
|
||
head = f"{num}({title})" if title else num
|
||
return head, rest.strip()
|
||
|
||
|
||
def clean(t):
|
||
t = (t or "").replace("\u00a0", " ")
|
||
t = re.sub(r"[ \t]+", " ", t)
|
||
return t.strip()
|
||
|
||
|
||
def fence_ascii_tables(text):
|
||
"""소스의 ASCII 아트표(+---+ / |...|)를 코드펜스로 감싸 원형 보존.
|
||
|
||
셀이 여러 줄로 감겨 md 표 변환이 불안정한 대비표 등에 적용.
|
||
"""
|
||
lines = text.split(chr(10))
|
||
out = []
|
||
i = 0
|
||
infence = False
|
||
|
||
def is_tbl(l):
|
||
s = l.strip()
|
||
return bool(re.match(r"^\+[-=+]{3,}", s)) or (
|
||
s.startswith("|") and ("+" in s or s.count("|") >= 2)
|
||
)
|
||
|
||
while i < len(lines):
|
||
if lines[i].lstrip().startswith("```"):
|
||
infence = not infence
|
||
out.append(lines[i])
|
||
i += 1
|
||
continue
|
||
if not infence and is_tbl(lines[i]):
|
||
j = i
|
||
block = []
|
||
while j < len(lines):
|
||
if is_tbl(lines[j]) or lines[j].strip() == "":
|
||
block.append(lines[j])
|
||
j += 1
|
||
else:
|
||
break
|
||
# 뒤쪽 빈 줄 되돌림
|
||
while block and block[-1].strip() == "":
|
||
block.pop()
|
||
j -= 1
|
||
if any(re.match(r"^\s*\+[-=+]{3,}", b) for b in block):
|
||
out += ["", "```text"] + [b.rstrip() for b in block] + ["```", ""]
|
||
else:
|
||
out += block
|
||
i = j
|
||
else:
|
||
out.append(lines[i])
|
||
i += 1
|
||
return chr(10).join(out)
|
||
|
||
|
||
def law_to_md(root, 분류, 출처url, 실명=""):
|
||
info = root.find("기본정보") or root.find("행정규칙기본정보")
|
||
name = T(info, "법령명_한글", "행정규칙명") or 실명
|
||
out = [f"# {name}", ""]
|
||
meta = [
|
||
("시행일", ymd(T(info, "시행일자"))),
|
||
("공포일", ymd(T(info, "공포일자", "발령일자"))),
|
||
("공포번호", T(info, "공포번호", "발령번호")),
|
||
("종류", T(info, "법종구분", "행정규칙종류")),
|
||
("제개정", T(info, "제개정구분명")),
|
||
("소관", T(info, "소관부처", "소관부처명")),
|
||
]
|
||
out.append("> " + " | ".join(f"{k} {v}" for k, v in meta if v))
|
||
out.append(f"> 출처: {출처url}")
|
||
out.append("")
|
||
|
||
flat = root.findall("조문내용")
|
||
if flat:
|
||
for e in flat:
|
||
body = (e.text or "").replace(" ", " ").rstrip()
|
||
if not body.strip():
|
||
continue
|
||
lines = [l.rstrip() for l in body.split(chr(10))]
|
||
first = lines[0].strip()
|
||
if re.match(r"^제\d+[장절편관]", first):
|
||
out += ["", f"## {first}", ""]
|
||
rest = lines[1:]
|
||
elif re.match(r"^제\d+조", first):
|
||
head, body0 = split_jo(first)
|
||
out += [f"### {head}", ""]
|
||
rest = ([body0] if body0 else []) + lines[1:]
|
||
else:
|
||
rest = lines
|
||
for l in rest:
|
||
if l.strip():
|
||
out.append(l)
|
||
out.append("")
|
||
|
||
조문 = root.find("조문")
|
||
if 조문 is not None:
|
||
for j in 조문.findall("조문단위"):
|
||
내용 = clean(T(j, "조문내용"))
|
||
if T(j, "조문여부") == "전문":
|
||
if 내용:
|
||
out += ["", f"## {내용}", ""]
|
||
continue
|
||
if 내용:
|
||
head, body0 = split_jo(내용)
|
||
if head:
|
||
out += [f"### {head}", ""]
|
||
if body0:
|
||
out += [body0, ""]
|
||
else:
|
||
out += [f"### {내용}", ""]
|
||
for h in j.findall("항"):
|
||
hv = clean(T(h, "항내용"))
|
||
if hv:
|
||
out.append(hv)
|
||
# 목은 호의 자식이 아니라 항의 직속 자식으로 오는 경우가 있다
|
||
# (법제처 XML 관행). 문서 순서대로 순회해야 호-목 대응이 유지된다.
|
||
for kid in h:
|
||
if kid.tag == "호":
|
||
ov = clean(T(kid, "호내용"))
|
||
if ov:
|
||
out.append(f" {ov}")
|
||
for mo in kid.findall("목"):
|
||
mv = clean(T(mo, "목내용")) or clean(
|
||
" ".join(x for x in mo.itertext() if x)
|
||
)
|
||
if mv:
|
||
out.append(f" {mv}")
|
||
elif kid.tag == "목":
|
||
mv = clean(T(kid, "목내용")) or clean(
|
||
" ".join(x for x in kid.itertext() if x)
|
||
)
|
||
if mv:
|
||
out.append(f" {mv}")
|
||
out.append("")
|
||
# 항 없이 호만 있는 조문
|
||
for ho in j.findall("호"):
|
||
ov = clean(T(ho, "호내용"))
|
||
if ov:
|
||
out.append(f" {ov}")
|
||
for mo in ho.findall("목"):
|
||
mv = clean(T(mo, "목내용")) or clean(" ".join(x for x in mo.itertext() if x))
|
||
if mv:
|
||
out.append(f" {mv}")
|
||
if j.findall("호") and not j.findall("항"):
|
||
out.append("")
|
||
|
||
부칙 = root.find("부칙")
|
||
if 부칙 is not None:
|
||
# 법령(law)은 부칙단위 래퍼, 행정규칙(admrul)은 부칙내용이 부칙 직속 —
|
||
# 부칙단위만 처리하면 행정규칙 md 에 부칙이 통째로 빠진다 (2026-08-15 결함 정정).
|
||
units = 부칙.findall("부칙단위") or 부칙.findall("부칙내용")
|
||
if units:
|
||
# 법제처 XML은 부칙을 오래된 순으로 담는다 — 상한을 두면 최신 부칙
|
||
# (연도별 요율 특례 등)이 잘려나가므로 전건 출력한다.
|
||
out += ["", "## 부칙", ""]
|
||
for b in units:
|
||
body = (
|
||
clean(T(b, "부칙내용")) if b.tag == "부칙단위" else clean("".join(b.itertext()))
|
||
)
|
||
# 인용블록: 각 줄 앞에 '> ', 원문 줄바꿈 보존(빈 줄은 '>')
|
||
for ln in body.split("\n"):
|
||
out.append(f"> {ln}" if ln.strip() else ">")
|
||
out.append("")
|
||
|
||
att = root.find("첨부파일")
|
||
if att is not None and len(att):
|
||
names = [x.text.strip() for x in att.findall("첨부파일명") if x.text]
|
||
if names:
|
||
out += ["", "## 첨부파일", ""] + [f"- {n}" for n in names] + [""]
|
||
|
||
별표 = root.find("별표")
|
||
if 별표 is not None and len(별표):
|
||
out += ["", "## 별표·서식 목록", "", "| 구분 | 번호 | 제목 | PDF |", "|---|---|---|---|"]
|
||
for b in 별표.findall("별표단위"):
|
||
n = T(b, "별표번호").lstrip("0") or "-"
|
||
g = T(b, "별표가지번호").lstrip("0")
|
||
title = re.sub(r"\s+", " ", clean(T(b, "별표제목")).replace("\n", " ")).replace(
|
||
"|", "／"
|
||
)
|
||
out.append(f"| {T(b, '별표구분')} | {n}{('의' + g) if g else ''} | {title} | `별표/` |")
|
||
out.append("")
|
||
return fence_ascii_tables("\n".join(out)) + "\n"
|
||
|
||
|
||
# ─────────────────── 별표 PDF 다운로드 ───────────────────
|
||
# limit=None: 상한 없음 — 80건 상한으로 대기환경보전법 시행규칙 별표 158건 중
|
||
# 78건이 누락된 결함 재발 방지 (2026-08-15). 수동 축소 실행 시에만 limit 지정.
|
||
def download_별표(root, folder, limit=None):
|
||
별표 = root.find("별표")
|
||
if 별표 is None or not len(별표):
|
||
return 0
|
||
d = folder / "별표"
|
||
n = 0
|
||
for b in 별표.findall("별표단위"):
|
||
title = clean(T(b, "별표제목"))
|
||
if title.startswith("삭제"):
|
||
continue
|
||
link = T(b, "별표서식PDF파일링크")
|
||
if not link:
|
||
continue
|
||
num = T(b, "별표번호").lstrip("0") or "0"
|
||
g = T(b, "별표가지번호").lstrip("0")
|
||
fn = safe(f"{T(b, '별표구분')}{num}{('의' + g) if g else ''}_{title[:48]}") + ".pdf"
|
||
d.mkdir(parents=True, exist_ok=True)
|
||
p = d / fn
|
||
if p.exists():
|
||
n += 1
|
||
continue
|
||
blob = fetch(BASEURL + link, binary=True)
|
||
if blob and blob[:4] == b"%PDF":
|
||
p.write_bytes(blob)
|
||
n += 1
|
||
time.sleep(0.25)
|
||
if limit is not None and n >= limit:
|
||
break
|
||
return n
|
||
|
||
|
||
# ─────────────────── 항목 1건 수집 ───────────────────
|
||
def collect(item):
|
||
분류폴더, target, 목록명, 조회명 = item["dir"], item["target"], item["name"], item["query"]
|
||
# "folder" = 분류 밖으로 옮긴 원문 폴더 (예: 산림 표준품셈 → 원가계산/산림_표준품셈)
|
||
folder = ROOT / item["folder"] if "folder" in item else ROOT / 분류폴더 / safe(목록명)
|
||
folder.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 1) 현행본
|
||
r = search(target, 조회명)
|
||
if r is None:
|
||
return {"명칭": 목록명, "상태": "검색실패"}
|
||
nodes = r.findall("law") + r.findall("admrul")
|
||
want = item.get("exact", 조회명)
|
||
cur = next((n for n in nodes if T(n, "법령명한글", "행정규칙명") == want), None) or next(
|
||
(n for n in nodes if T(n, "법령명한글", "행정규칙명").startswith(want)), None
|
||
)
|
||
if cur is None:
|
||
return {
|
||
"명칭": 목록명,
|
||
"상태": "현행본 없음",
|
||
"후보": [T(n, "법령명한글", "행정규칙명") for n in nodes[:3]],
|
||
}
|
||
실명 = T(cur, "법령명한글", "행정규칙명")
|
||
cur_key = T(cur, "법령일련번호", "행정규칙일련번호")
|
||
kind = "MST" if target == "law" else "ID"
|
||
cur_id = cur_key if target == "law" else T(cur, "행정규칙ID")
|
||
|
||
# 2) 연혁 목록 → 교본시점 판본
|
||
hist = []
|
||
if target == "law":
|
||
h = search("eflaw", 조회명, display=100)
|
||
if h is not None:
|
||
for n in h.findall("law"):
|
||
if T(n, "법령명한글") != 실명:
|
||
continue
|
||
hist.append(
|
||
{
|
||
"일련": T(n, "법령일련번호"),
|
||
"시행": T(n, "시행일자"),
|
||
"구분": T(n, "현행연혁코드"),
|
||
}
|
||
)
|
||
else:
|
||
h = search(target, 조회명, nw=2, display=100)
|
||
if h is not None:
|
||
for n in h.findall("admrul"):
|
||
hist.append(
|
||
{
|
||
"일련": T(n, "행정규칙일련번호"),
|
||
"시행": T(n, "시행일자"),
|
||
"발령": T(n, "발령일자"),
|
||
"구분": T(n, "현행연혁구분"),
|
||
}
|
||
)
|
||
past = sorted(
|
||
[x for x in hist if x["시행"] and x["시행"] <= 교본기준일], key=lambda x: x["시행"]
|
||
)
|
||
past = past[-1] if past else None
|
||
|
||
saved = []
|
||
versions = [("현행", cur_key, T(cur, "시행일자"))]
|
||
if past and past["일련"] != cur_key:
|
||
versions.append(("교본시점", past["일련"], past["시행"]))
|
||
|
||
root_cur = None
|
||
for tag, key, eff in versions:
|
||
if target == "law":
|
||
root, blob = service("law", key, "MST")
|
||
else:
|
||
root, blob = service("admrul", key, "ID")
|
||
if root is None:
|
||
root, blob = service("admrul", key, "LID")
|
||
if root is None:
|
||
saved.append(f"{tag}:실패")
|
||
continue
|
||
stem = f"{tag}_{eff}"
|
||
(folder / f"{stem}.xml").write_bytes(blob)
|
||
url = f"{BASEURL}/DRF/lawService.do?OC={OC}&type=HTML&target={target}&{'MST' if target == 'law' else 'ID'}={key}"
|
||
(folder / f"{stem}.md").write_text(law_to_md(root, 분류폴더, url, 실명), encoding="utf-8")
|
||
saved.append(f"{tag}:{eff}")
|
||
if tag == "현행":
|
||
root_cur = root
|
||
time.sleep(0.3)
|
||
|
||
n별표 = download_별표(root_cur, folder) if root_cur is not None else 0
|
||
|
||
# 3) _meta.md / CHANGELOG.md
|
||
between = [x for x in hist if x["시행"] and 교본기준일 < x["시행"] <= T(cur, "시행일자")]
|
||
between.sort(key=lambda x: x["시행"], reverse=True)
|
||
meta = [
|
||
f"# {목록명}",
|
||
"",
|
||
f"- 정식명칭: {실명}",
|
||
f"- 목록상 명칭: {목록명}" + (" ← **명칭 상이**" if 실명 != 목록명 else ""),
|
||
f"- 식별자: `{'법령ID' if target == 'law' else '행정규칙ID'}={cur_id}` (불변)",
|
||
f"- 소관: {T(cur, '소관부처명')}",
|
||
f"- 종류: {T(cur, '법령구분명', '행정규칙종류')}",
|
||
f"- 현행 시행일: {ymd(T(cur, '시행일자'))} / 공포 {ymd(T(cur, '공포일자', '발령일자'))} 제{T(cur, '공포번호', '발령번호')}호",
|
||
f"- 교본시점 판본: {ymd(past['시행']) if past else '없음(교본 이후 제정)'}",
|
||
f"- 출처: {BASEURL}/DRF/lawService.do?target={target}&{'MST' if target == 'law' else 'ID'}={cur_key}",
|
||
# 수집일이 없으면 갱신 지연 여부를 파일만으로 판정할 수 없다 (2026-08-14 감사).
|
||
f"- 수집일: {time.strftime('%Y-%m-%d')}",
|
||
f"- 수집 파일: {', '.join(saved)}" + (f" / 별표 PDF {n별표}건" if n별표 else ""),
|
||
"",
|
||
]
|
||
(folder / "_meta.md").write_text("\n".join(meta), encoding="utf-8")
|
||
|
||
ch = [
|
||
f"# {목록명} — 변경이력",
|
||
"",
|
||
f"기준: 교본시점({ymd(교본기준일)}) → 현행({ymd(T(cur, '시행일자'))})",
|
||
"",
|
||
f"**교본 이후 개정 {len(between)}회**",
|
||
"",
|
||
]
|
||
if between:
|
||
ch += ["| 시행일 | 구분 |", "|---|---|"]
|
||
# 상한 없음 — [:40] 절단으로 개정 41회 이상 문서(고용보험법 시행령 42회 등)
|
||
# 이력이 잘리던 결함 재발 방지 (2026-08-15). 헤더 개정횟수와 표 행수 일치 보장.
|
||
for x in between:
|
||
ch.append(f"| {ymd(x['시행'])} | {x.get('구분', '')} |")
|
||
ch.append("")
|
||
if root_cur is not None:
|
||
for tag in ("제개정이유", "개정문"):
|
||
e = root_cur.find(tag)
|
||
if e is not None:
|
||
s = re.sub(r"\s+", " ", "".join(e.itertext())).strip()
|
||
if s:
|
||
ch += [f"## 최근 개정 — {tag}", "", f"> {s}", ""]
|
||
(folder / "CHANGELOG.md").write_text("\n".join(ch), encoding="utf-8")
|
||
|
||
return {
|
||
"명칭": 목록명,
|
||
"실명": 실명,
|
||
"상태": "OK",
|
||
"판본": saved,
|
||
"별표": n별표,
|
||
"개정횟수": len(between),
|
||
}
|
||
|
||
|
||
if __name__ == "__main__":
|
||
items = json.load(open(SC / sys.argv[1], encoding="utf-8"))
|
||
res = []
|
||
for i, it in enumerate(items, 1):
|
||
r = collect(it)
|
||
res.append(r)
|
||
print(
|
||
f"[{i}/{len(items)}] {r['명칭']} :: {r['상태']} {r.get('판본', '')} 별표{r.get('별표', 0)} 개정{r.get('개정횟수', '')}",
|
||
flush=True,
|
||
)
|
||
time.sleep(0.3)
|
||
json.dump(
|
||
res, open(SC / "collect_result.json", "w", encoding="utf-8"), ensure_ascii=False, indent=1
|
||
)
|