# app/crawlers/kepco_org_parser.py from __future__ import annotations import io, re, json, unicodedata from typing import List import requests from pdfminer.high_level import extract_text_to_fp from pdfminer.layout import LAParams UA = "GovBot/1.0 (+https://example.invalid)" HEADERS = {"User-Agent": UA, "Accept": "*/*", "Referer": "https://alio.go.kr/"} # ------------------------ HTTP & PDF ------------------------ def _download(url: str, timeout: float = 20.0) -> requests.Response: r = requests.get(url, headers=HEADERS, timeout=timeout) r.raise_for_status() return r def _pdf_to_text(pdf_bytes: bytes) -> str: buf_out = io.StringIO() laparams = LAParams( line_margin=0.08, word_margin=0.05, char_margin=2.0, boxes_flow=0.5, detect_vertical=False, all_texts=True, ) extract_text_to_fp(io.BytesIO(pdf_bytes), buf_out, laparams=laparams) return buf_out.getvalue() # ------------------------ Normalization ------------------------ _ZW = "".join(chr(c) for c in [0x200B, 0x200C, 0x200D, 0xFEFF]) def _normalize(s: str) -> str: if not s: return "" s = s.replace("\r", "\n").replace("\xa0", " ") s = s.translate({ord(c): None for c in _ZW}) s = s.replace("–", "-").replace("—", "-").replace("（", "(").replace("）", ")") s = unicodedata.normalize("NFKC", s) s = re.sub(r"[ \t]+", " ", s) return s.strip() def _prep_lines(text: str) -> List[str]: text = _normalize(text) # 원문이 너무 많은 빈줄을 포함하므로 제거 lines = [_normalize(x) for x in text.split("\n")] return [ln for ln in lines if ln] # ------------------------ 라벨/토큰 ------------------------ ALL_LABELS = { "직위","성명","직책","성별","임기","(시작일)","(종료일)","시작일","종료일", "주요경력","선임절차","선임절차규정","당연직여부","임명권자", } ROW_START_SYNONYMS = {"상임기관장","기관장"} CHANGE_TOKENS = {"변경 전","변경전","변경 후","변경후","변경사유"} LABEL_FLOW = {"선임절차","선임절차규정","당연직여부","임명권자"} LABEL_BLOCK = {"주요경력"} LABEL_INLINE = {"직위","성명","직책","성별","임기","(시작일)","(종료일)"} TITLE_HINT_SUFFIX = ("사장","원장","이사장","대표이사","부사장","본부장") def _looks_like_title(tok: str | None) -> bool: if not tok: return False tok = tok.strip() return any(tok.endswith(s) for s in TITLE_HINT_SUFFIX) def _is_label(t: str) -> bool: return ( t in ALL_LABELS or t in LABEL_FLOW or t in LABEL_BLOCK or t in ROW_START_SYNONYMS or t in CHANGE_TOKENS ) # ------------------------ ALIO pdf.json → PDF URL ------------------------ def _fetch_pdf_url_from_alio_json(endpoint_url: str, timeout: float = 20.0) -> str: """ https://alio.go.kr/download/pdf.json?... 응답에서 PDF URL을 추출. 키가 들쭉날쭉하므로 여러 후보 + 백업 정규식으로 탐색. """ r = _download(endpoint_url, timeout=timeout) try: j = r.json() except Exception: j = {} for k in ("fileUrl","pdfUrl","url","file_path","fileurl","pdf"): v = j.get(k) if isinstance(v, str) and v.lower().endswith(".pdf"): return v # 백업: 본문 문자열에서 .pdf 첫 URL m = re.search(r"https?://[^\s\"']+?\.pdf", r.text) return m.group(0) if m else "" # ------------------------ 텍스트 선형화(사람이 읽기 쉬운 한 줄 구성) ------------------------ def _linearize_text_for_export(raw_text: str) -> str: """ 규칙: - 라벨 체인(직위/성명/직책/성별/임기/(시작일)/(종료일))은 가능한 한 줄로 이어 붙임 예) '직위 상임기관장 성명 김동철' / '임기 (시작일) 2023... (종료일) 2026...' - '주요경력'은 헤더 1줄 + 항목들 각 1줄 유지 - '선임절차/선임절차규정/당연직여부/임명권자'는 '라벨 + 값'을 한 줄로 - '변경 전/변경 후/변경사유' 블록은 직전 '직위' 값을 추정해 요약 한 줄: 예) '상임이사 서근배 공석 의원면직(5.19)' """ L = _prep_lines(raw_text) n = len(L) def guess_position_before(idx: int) -> str: # 직전 25개 토큰 안에서 '이사'류나 기관장류를 찾아 직위 후보로 사용 for k in range(idx - 1, max(-1, idx - 25), -1): t = L[k] if not t or _is_label(t): continue if t.endswith("이사") or t in ROW_START_SYNONYMS: return t return "" out: list[str] = [] i = 0 last_pos_value = "" # 최근에 본 '직위'의 값 while i < n: t = L[i] # 1) 변경 전/후/사유 묶음 if t in CHANGE_TOKENS: pos = guess_position_before(i) or last_pos_value j = i names: list[str] = [] reason = "" while j < n: tt = L[j] if tt in {"변경 전","변경전"}: j += 1 # '성명' 라벨 스킵 후 값 수집 if j < n and L[j] == "성명": j += 1 # 다음 라벨 전까지 이름 하나 while j < n and not _is_label(L[j]): # 직후에 '상임기관장' 등의 라벨+직책이 이어질 수 있어 필터 if _looks_like_title(L[j]) or L[j] in ROW_START_SYNONYMS: j += 1 continue names.append(L[j]); j += 1 break continue if tt in {"변경 후","변경후"}: j += 1 if j < n and L[j] == "성명": j += 1 while j < n and not _is_label(L[j]): if _looks_like_title(L[j]) or L[j] in ROW_START_SYNONYMS: j += 1 continue names.append(L[j]); j += 1 break continue if tt == "변경사유": j += 1 reason_buf = [] while j < n and not _is_label(L[j]): reason_buf.append(L[j]); j += 1 reason = " ".join(reason_buf).strip() break # 다른 라벨을 만나면 블록 종료 if _is_label(tt) and tt not in {"성명"}: break j += 1 if pos and names: body = " ".join(names[:2]) # 최대 2개 out.append(" ".join(x for x in (pos, body, reason) if x)) i = max(j, i + 1) continue # 2) 주요경력 블록 if t in LABEL_BLOCK: out.append(t) i += 1 while i < n and not _is_label(L[i]): out.append(L[i]); i += 1 continue # 3) 선임절차/규정/당연직/임명권자: 한 줄 if t in LABEL_FLOW: j = i + 1 buf = [t] while j < n and not _is_label(L[j]): buf.append(L[j]); j += 1 out.append(" ".join(buf)) i = j continue # 4) 라벨 체인(직위/성명/직책/성별/임기/(시작일)/(종료일) 및 기관장 라벨)을 한 줄로 if t in LABEL_INLINE or t in ROW_START_SYNONYMS: j = i buf = [] while j < n: tt = L[j] if tt in LABEL_BLOCK or tt in LABEL_FLOW or tt in CHANGE_TOKENS: break if tt in LABEL_INLINE or tt in ROW_START_SYNONYMS: buf.append(tt); j += 1 # 뒤에 따라오는 값들(다음 라벨 전까지)을 이어 붙임 while j < n and not _is_label(L[j]): if tt == "직위": last_pos_value = L[j] buf.append(L[j]); j += 1 continue break if buf: out.append(" ".join(buf)) i = j continue # 5) 그 외(헤더/기관명/표제 등)는 그대로 out.append(t) i += 1 # 후처리: 공백 정리 out = [re.sub(r"[ \t]+", " ", x).strip() for x in out if x and x.strip()] return "\n".join(out) # ------------------------ Public API (1단계 전용) ------------------------ def endpoint_to_text_json(alio_pdf_json_url: str) -> dict: """ 1) alio pdf.json → PDF URL 추출 2) PDF 다운로드 → 텍스트 추출 3) 기대 레이아उ트로 선형화 4) {"text": "..."} 반환 """ pdf_url = _fetch_pdf_url_from_alio_json(alio_pdf_json_url) if not pdf_url: raise RuntimeError("PDF URL을 alio json에서 찾지 못했습니다.") pdf = _download(pdf_url).content text = _pdf_to_text(pdf) linear = _linearize_text_for_export(text) return {"text": linear}