Files
Aislo/resources/knowledge/original/_pipeline/collect_law.py
T
eomsangdonandClaude Fable 5 81cd7e23c3 feat(knowledge): 도메인 지식저장소 메인 통합 + resources 그룹 체계 재편
knowledge (구 Aislo-law 독립 저장소 → resources/knowledge 이관, 저장소 폐지):
- 법령·행정규칙·표준시방서·교본 원문 + 기술문서 55건 + 실무 분석·종합비교
- 루트 지침 체계: README(지도)·00_운영지침·01_수집지침·02_분석지침·
  03_미결_및_확인사항(교본 충돌 리스트 포함)·04_참조_법령기준_목록
- 기술문서 55건 원문 전수 검증 완료 (사방 설계홍수량 법정 기준 등 반영)
- 정리: CAD·오피스 잔재 142건, 중복 zip 7건(413MB), 빈 폴더 30개 제거

resources 그룹 재편 (이름순 그룹핑):
- app_branding(구 prog_icon.jpg)·app_policies(구 legal)·
  data_global_contours(구 grobal_contours)·data_rainfall_idf_cache(구 wamis_contours)·
  template_2dDrawing(구 dwg_analysis/templete — 오타 교정, 상수·경로 동기화)
- dwg_analysis(분석 완료 1.8GB)·templates(빈 폴더)·templete_calc_cost.xlsx 삭제
- 참조 코드 5파일 경로 수정 + 프론트 재빌드 (구 경로 잔존 0)
- .gitignore: resources 추적 전환, national_contours.gpkg(22GB) 영구 제외
- .env: knowledge 수집용 API 정보 주석 통합 (KCSC·법령센터·조달청 제비율)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-12 19:14:32 +09:00

384 lines
16 KiB
Python
Raw Blame History

This file contains invisible Unicode characters
This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- 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)
for ho in h.findall("호"):
ov = clean(T(ho, "호내용"))
if ov:
out.append(f" {ov}")
for mo in ho.findall("목"):
mv = clean(" ".join(x for x in (mo.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}")
if j.findall("호") and not j.findall("항"):
out.append("")
부칙 = root.find("부칙")
if 부칙 is not None:
units = 부칙.findall("부칙단위")
if units:
out += ["", "## 부칙", ""]
for b in units[:20]:
body = clean(T(b, "부칙내용"))
# 인용블록: 각 줄 앞에 '> ', 원문 줄바꿈 보존(빈 줄은 '>')
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 다운로드 ───────────────────
def download_별표(root, folder, limit=80):
별표 = 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 n >= limit:
break
return n
# ─────────────────── 항목 1건 수집 ───────────────────
def collect(item):
분류폴더, target, 목록명, 조회명 = item["dir"], item["target"], item["name"], item["query"]
folder = 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}",
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 += ["| 시행일 | 구분 |", "|---|---|"]
for x in between[:40]:
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)