"""PDF → data/atlas.json 빌드. .venv/bin/python -m pipeline.build 자동 추출 결과는 모두 review_status='draft' 이며 method 필드로 출처(색인/목차/본문/시안)를 남긴다. """ import hashlib import json import re import sys import time import unicodedata from collections import defaultdict from functools import lru_cache import pymupdf from . import curated from .config import (BOOKS, DATA_DIR, GALAXIES, SECONDARY_CHAPTER_MAP, SECONDARY_KEYWORD_MAP, SECONDARY_SECTION_OVERRIDES) from .index import extract_index from .pdfutil import body_text, label_to_index from .toc import extract_toc HANGUL = re.compile(r"[가-힣]") def log(*a): print(*a, file=sys.stderr, flush=True) def norm_key(s: str) -> str: s = unicodedata.normalize("NFC", s).lower() s = re.sub(r"[\s·\-–_‧∙/'’\.]", "", s) return s def tid(key: str) -> str: return "t_" + hashlib.sha1(key.encode()).hexdigest()[:10] def rid(a: str, b: str, typ: str) -> str: return "r_" + hashlib.sha1(f"{a}|{b}|{typ}".encode()).hexdigest()[:10] def clean_alias(a: str): """동의어 정리: 하이픈 줄바꿈 잇기, 연도·숫자만 있는 값 제거.""" a = re.sub(r"(?<=[A-Za-z])- (?=[a-z])", "", a).strip(" ,;") if not a or re.fullmatch(r"[\d\s.,~\-]+", a) or len(a) < 2: return None return a def split_name(raw: str): """'건강 신념 모형(Health Belief Model; HBM)' → ('건강 신념 모형', ['Health Belief Model', 'HBM'])""" raw = raw.strip() m = re.match(r"^(.+?)\s*[\(（]([^()（）]+)[\)）]?\s*$", raw) if m and m.group(1).strip(): name = m.group(1).strip() inner = m.group(2) aliases = [c for c in (clean_alias(a) for a in re.split(r"[;,]", inner)) if c] return name, aliases return raw, [] # ---------------------------------------------------------------- 단원 매핑 class ChapterIndex: def __init__(self, toc_by_book): self.toc = toc_by_book self.sorted = {b: sorted(n, key=lambda x: (x["page"], x["level"])) for b, n in toc_by_book.items()} self.by_id = {n["id"]: n for nodes in toc_by_book.values() for n in nodes} def locate(self, book_id: str, page: int): """페이지가 속한 가장 깊은 목차 노드.""" best = None for n in self.sorted.get(book_id, []): if n["page"] <= page: if best is None or n["page"] > best["page"] or (n["page"] == best["page"] and n["level"] >= best["level"]): best = n else: break return best def ancestors(self, node): out = [] while node: out.append(node) node = self.by_id.get(node["parent_id"]) return out[::-1] def chapter_section(self, node): chain = self.ancestors(node) if node else [] chapter = next((n for n in chain if n["kind"] == "chapter"), None) section = next((n for n in chain if n["kind"] == "section"), None) return chapter, section, chain def galaxy_for(book, chapter, chain): if book["primary"]: if chapter is None: return None no = int(chapter["number"]) return next((g["id"] for g in GALAXIES if g["chapter_no"] == no), None) titles = " ".join(n["title"] for n in chain[1:]) if chain else "" for kw, g in SECONDARY_KEYWORD_MAP: if kw in titles: return g if chapter is not None: g = SECONDARY_CHAPTER_MAP.get((book["book_id"], int(chapter["number"]))) if g: return g return SECONDARY_CHAPTER_MAP.get((book["book_id"], None)) # ---------------------------------------------------------------- 본문 문장 SENT_SPLIT = re.compile(r"(?<=[다요음함됨임])\.\s+|(?<=\.)\s+(?=[가-힣A-Z①-⑳•\-])|\n") DEF_PATTERNS = [ (re.compile(r"(이)?란\s"), 4), (re.compile(r"(이)?라\s?(함은|한다|하며|하고|부른다)"), 4), (re.compile(r"(말한다|의미한다|뜻한다|정의한다|정의된다|일컫는다|가리킨다|이르는 말)"), 4), (re.compile(r"정의"), 2), ] class Corpus: def __init__(self): self.docs = {} self.label_maps = {} def open(self, book): bid = book["book_id"] if bid not in self.docs: self.docs[bid] = pymupdf.open(book["file"]()) self.label_maps[bid] = label_to_index(self.docs[bid]) return self.docs[bid] def pdf_index(self, book, page: int): self.open(book) return self.label_maps[book["book_id"]].get(str(page)) @lru_cache(maxsize=4096) def sentences(self, book_id: str, pdf_page: int) -> tuple[str, ...]: doc = self.docs[book_id] text = body_text(doc, pdf_page) # 쪽 끝에서 끊긴 문장은 다음 쪽 첫 문장과 잇는다 if text and not re.search(r"[.?!)」』]\s*$", text) and pdf_page + 1 < len(doc): nxt = body_text(doc, pdf_page + 1) m = re.match(r"^(.{1,300}?(?:다|요|음)\.)", nxt) if m: joiner = "" if re.search(r"[가-힣]$", text) and re.match(r"^[가-힣]", nxt) else " " text = text + joiner + m.group(1) text = rejoin_lines(text) out = [] for s in SENT_SPLIT.split(text): s = s.strip(" •-·") if 8 <= len(s) <= 400: out.append(s if s.endswith(".") or not s.endswith("다") else s + ".") return tuple(out) HEADING = re.compile(r"^(\d+[\).]|[①-⑳]|[가-하][\).]|\(\d+\)|[•▪■□○●]|표\s?\d|그림\s?\d)") WORD_END = set("의은는이가을를에로고며서과와도한된인적및등면데때중게") def rejoin_lines(text: str) -> str: """줄 단위로 끊긴 본문을 문장으로 다시 잇는다. 제목·목록 줄은 그대로 둔다.""" out = [] for line in text.split("\n"): line = line.strip() if not line: continue if out: prev = out[-1] ends = re.search(r"[.?!:)」』]$", prev) or len(prev) < 22 or HEADING.match(line) or "=" in prev if not ends: if re.search(r"[가-힣]$", prev) and re.match(r"^[가-힣]", line) and prev[-1] not in WORD_END: out[-1] = prev + line # 단어 중간에서 줄바꿈 else: out[-1] = prev + " " + line continue out.append(line) return "\n".join(out) PARTICLE = r"(?=\s|\(|（|은|는|이|가|란|을|를|의|과|와|에|도|로|으로|,|·|라|$)" def score_sentence(sent: str, name: str, key: str): """정의 문장일수록 높은 점수. 용어가 없으면 None.""" flat = norm_key(sent) if key not in flat or len(flat) - len(key) < 8: # 제목줄 등 너무 짧은 문장 제외 return None pat = re.escape(name).replace(r"\ ", r"\s?") m = re.search(pat + PARTICLE, sent) if not m: return 0 sc = 1 paren = r"\s?(?:\([^)]*\)|（[^）]*）)?\s?" if re.match(r"^\s*" + pat + paren + r"(?:이란|란|은|는|이라 함은|라 함은|이라고 함은)\s", sent): sc += 4 if re.search(r"(말한다|의미한다|뜻한다|정의한다|정의된다|일컫는다|가리킨다|이다|것이다)\.?$", sent): sc += 3 elif re.search(r"(정의|의미|개념|나타내는|지표|방법|과정|상태|활동|체계|이론|모형)", sent): sc += 2 elif re.search(pat + paren + r"(?:이?라고?\s?(?:한다|부른다|일컫는다|정의한다|명명한다))", sent): sc += 6 if re.match(r"^(에|을|를|의|은|는|이|가|와|과|로|으로|하여|하고|되어|및|또는|등)\s", sent): sc -= 6 if 20 <= len(sent) <= 220: sc += 1 if len(sent) > 300: sc -= 3 return sc EN_PAREN = re.compile(r"\(\s*([A-Za-z][A-Za-z0-9 ,;:'&\-/\.]{1,90}?)\s*\)") def english_aliases(sent: str, name: str): """문장에서 '용어(English, ABC)' 꼴을 찾아 영문 표기를 돌려줌.""" flat_name = re.escape(name).replace(r"\ ", r"\s?") m = re.search(flat_name + r"\s?" + EN_PAREN.pattern, sent) if not m: return [] return [c for c in (clean_alias(a) for a in re.split(r"[;,]|:", m.group(1))) if c] # ---------------------------------------------------------------- 메인 def build(): t0 = time.time() corpus = Corpus() toc = {} entries = [] for b in BOOKS: doc = corpus.open(b) toc[b["book_id"]] = extract_toc(doc, b) es = extract_index(doc, b) for e in es: e["book_id"] = b["book_id"] entries.extend(es) log(f"[{b['book_id']}] 목차 {len(toc[b['book_id']])}개, 색인 {len(es)}개") chidx = ChapterIndex(toc) books_by_id = {b["book_id"]: b for b in BOOKS} # ---- 1) 색인 항목 → 용어 통합 terms: dict[str, dict] = {} alias_index: dict[str, str] = {} # 별칭 키 → 용어 키 xrefs = [] def get_term(name, kind): key = norm_key(name) t = terms.get(key) if t is None: t = terms[key] = { "key": key, "id": tid(key), "name": name, "kind": kind, "aliases": [], "sources": [], "index_entries": [], } return t # 한글/인물 먼저, 영문은 뒤에서 별칭 연결을 시도 for e in sorted(entries, key=lambda e: e["section"] == "en"): if e.get("see"): xrefs.append(e) continue name, aliases = split_name(e["term"]) if e["section"] == "en" and not HANGUL.search(name): e["_name"], e["_aliases"] = name, aliases continue kind = "person" if e["section"] == "person" else "concept" t = get_term(name, kind) for a in aliases: if a not in t["aliases"]: t["aliases"].append(a) alias_index.setdefault(norm_key(a), t["key"]) t["index_entries"].append({"book_id": e["book_id"], "raw": e["term"], "pages": e["pages"]}) for p in e["pages"]: t["sources"].append({"book_id": e["book_id"], "page": p}) # ---- 2) 본문에서 영문 표기 찾기 (색인 페이지의 문장) log("본문 문장 분석 중…") for t in terms.values(): for s in t["sources"]: b = books_by_id[s["book_id"]] pi = corpus.pdf_index(b, s["page"]) if pi is None: continue for sent in corpus.sentences(b["book_id"], pi): for a in english_aliases(sent, t["name"]): if a not in t["aliases"] and norm_key(a) != t["key"]: t["aliases"].append(a) alias_index.setdefault(norm_key(a), t["key"]) # ---- 3) 영문 색인 항목: 별칭과 일치하면 합치고, 아니면 독립 용어 en_new = 0 for e in entries: if "_name" not in e: continue k = norm_key(e["_name"]) target = alias_index.get(k) or next((alias_index[norm_key(a)] for a in e["_aliases"] if norm_key(a) in alias_index), None) if target is None and k in terms: target = k if target: t = terms[target] if e["_name"] not in t["aliases"] and norm_key(e["_name"]) != t["key"]: t["aliases"].append(e["_name"]) else: t = get_term(e["_name"], "concept") t["lang"] = "en" en_new += 1 for a in e["_aliases"]: if a not in t["aliases"]: t["aliases"].append(a) t["index_entries"].append({"book_id": e["book_id"], "raw": e["term"], "pages": e["pages"]}) for p in e["pages"]: if {"book_id": e["book_id"], "page": p} not in t["sources"]: t["sources"].append({"book_id": e["book_id"], "page": p}) log(f"영문 독립 용어 {en_new}개") for x in xrefs: src = terms.get(norm_key(split_name(x["see"])[0])) if src and x["term"] not in src["aliases"]: src["aliases"].append(x["term"]) # ---- 4) 출처마다 단원·은하·문장 연결, 정의 후보 선택 log("출처·정의 연결 중…") for t in terms.values(): best_def = None galaxy_votes = defaultdict(float) memberships = {} for s in t["sources"]: b = books_by_id[s["book_id"]] node = chidx.locate(b["book_id"], s["page"]) chapter, section, chain = chidx.chapter_section(node) s["toc_id"] = node["id"] if node else None s["chapter_id"] = chapter["id"] if chapter else None s["section_id"] = section["id"] if section else None s["path"] = " › ".join(n["title"] for n in chain[1:]) if chain else "" g = galaxy_for(b, chapter, chain) s["galaxy_id"] = g if g: galaxy_votes[g] += 3 if b["primary"] else 1 if b["primary"] and section: memberships[section["id"]] = g item = next((n for n in chain if n["kind"] == "item"), None) if item and section["id"] not in t.setdefault("items", {}): t["items"][section["id"]] = item["id"] pi = corpus.pdf_index(b, s["page"]) s["pdf_page"] = pi + 1 if pi is not None else None s["snippet"] = None if pi is None: continue cand = [] for sent in corpus.sentences(b["book_id"], pi): sc = score_sentence(sent, t["name"], t["key"]) if sc is not None: cand.append((sc, sent)) if cand: cand.sort(key=lambda x: (-x[0], len(x[1]))) sc, sent = cand[0] s["snippet"] = sent weight = sc + (1 if b["primary"] else 0) if sc >= 7 and (best_def is None or weight > best_def[0]): best_def = (weight, sent, b["book_id"], s["page"]) t["galaxy_id"] = max(galaxy_votes, key=galaxy_votes.get) if galaxy_votes else None # 주 소속 절: 주 은하에 속한 절 가운데 처음 나온 것 t["section_ids"] = list(dict.fromkeys(memberships)) prim_secs = [sid for sid, g in memberships.items() if g == t["galaxy_id"]] t["section_id"] = prim_secs[0] if prim_secs else None if best_def: t["definition"] = {"text": best_def[1], "book_id": best_def[2], "page": best_def[3], "method": "auto"} else: t["definition"] = None # 보조 교재에만 있는 용어: 2026 절(하위 항목 제목 포함)과 출처 경로의 글자 bigram 유사도로 배정 def bigrams(text): k = norm_key(text) return {k[i:i + 2] for i in range(len(k) - 1)} primary_nodes = [n for b in BOOKS if b["primary"] for n in toc[b["book_id"]]] sec_nodes = [n for n in primary_nodes if n["kind"] == "section"] sec_galaxy, sec_bag, item_bag, sec_items = {}, {}, {}, defaultdict(list) for n in sec_nodes: ch, _, chain = chidx.chapter_section(n) sec_galaxy[n["id"]] = galaxy_for(books_by_id[n["book_id"]], ch, chain) sec_bag[n["id"]] = bigrams(n["title"]) | bigrams(n["title"]) for n in primary_nodes: if n["kind"] in ("item", "subitem"): chain = chidx.ancestors(n) sec = next((c for c in chain if c["kind"] == "section"), None) if sec: sec_bag[sec["id"]] |= bigrams(n["title"]) if n["kind"] == "item": sec_items[sec["id"]].append(n) item_bag[n["id"]] = bigrams(n["title"]) else: it = next((c for c in chain if c["kind"] == "item"), None) if it: item_bag.setdefault(it["id"], set()).update(bigrams(n["title"])) def sim(a, b): return len(a & b) / ((len(a) * len(b)) ** 0.5 or 1) # 보조 교재의 절 → 2026 절 매핑 (제목·상위·하위 항목 제목 전체의 bigram 유사도) sec_map = {} for b in BOOKS: if b["primary"]: continue nodes = toc[b["book_id"]] kids = defaultdict(list) for n in nodes: kids[n["parent_id"]].append(n) def desc_bag(n): out = bigrams(n["title"]) for k in kids[n["id"]]: out |= desc_bag(k) return out for n in nodes: if n["kind"] not in ("section", "chapter"): continue chain = chidx.ancestors(n) ch = next((c for c in chain if c["kind"] == "chapter"), None) mapped = galaxy_for(b, ch, chain) titles = " ".join(c["title"] for c in chain[1:]) forced = next((p for kw, title in SECONDARY_SECTION_OVERRIDES if kw in titles for p in sec_nodes if p["title"].strip() == title), None) if forced: sec_map[n["id"]] = forced["id"] continue bag = desc_bag(n) | {x for c in chain[1:-1] for x in bigrams(c["title"])} sec_map[n["id"]] = max(sec_nodes, key=lambda p: sim(bag, sec_bag[p["id"]]) + (0.08 if sec_galaxy[p["id"]] == mapped else 0))["id"] for t in terms.values(): if t.get("section_id"): continue votes = defaultdict(int) for s in t["sources"]: if books_by_id[s["book_id"]]["primary"]: continue key = s.get("section_id") or s.get("chapter_id") if key in sec_map: votes[sec_map[key]] += 1 if votes: best = max(votes, key=votes.get) else: best = max(sec_nodes, key=lambda n: sim(bigrams(t["name"]), sec_bag[n["id"]]))["id"] t["section_id"] = best t["galaxy_id"] = sec_galaxy[best] t["section_guess"] = True for t in terms.values(): if not t["galaxy_id"]: t["galaxy_id"] = sec_galaxy.get(t["section_id"], "g1") # 절 안의 항목(지도에서 작은 묶음): 원문 위치가 없으면 이름 유사도로 추정 item = t.get("items", {}).get(t["section_id"]) if not item and sec_items.get(t["section_id"]): bag = bigrams(t["name"]) | {b for s in t["sources"] for b in bigrams(s.get("path", "").split(" › ")[-1])} item = max(sec_items[t["section_id"]], key=lambda n: sim(bag, item_bag.get(n["id"], set())))["id"] t["item_id"] = item # ---- 5) 시안 예시 설명 붙이기 def find(name): k = norm_key(name) if k in terms: return terms[k] if k in alias_index: return terms[alias_index[k]] return None for name, desc, example, aliases in curated.WIREFRAME_TERMS: t = find(name) if not t: log(f" (시안 용어 없음) {name}") continue t["easy"] = desc t["example"] = example t["search_terms"] = aliases.split() # 쉬운 뜻·사례 초안 (data/drafts/out_*.json, 교재 문장을 바탕으로 작성한 검토 전 초안) drafts = {} for f in sorted((DATA_DIR / "drafts").glob("out_*.json")): try: drafts.update({k: v for k, v in json.loads(f.read_text()).items() if v and v.get("easy") and v.get("confidence") != "low"}) # 신뢰도 낮은 초안은 제외 except json.JSONDecodeError as e: log(f" 초안 파일 오류 {f.name}: {e}") used = 0 for t in terms.values(): dr = drafts.get(t["id"]) if dr and not t.get("easy"): t["easy"] = dr["easy"].strip() t["example"] = (dr.get("example") or "").strip() or None t["easy_method"] = "draft" used += 1 log(f"쉬운 뜻 초안 {used}개 반영") # ---- 6) 관계 초안 relations = {} def add_rel(a, b, typ, label, method, evidence=None, directed=True): if a is None or b is None or a["id"] == b["id"]: return r = rid(a["id"], b["id"], typ) if r in relations or rid(b["id"], a["id"], typ) in relations and not directed: return relations[r] = { "relation_id": r, "source": a["id"], "target": b["id"], "type": typ, "label": label, "directed": directed, "method": method, "evidence": evidence or [], "review_status": "draft", } for a, b, typ, label in curated.WIREFRAME_RELATIONS: add_rel(find(a), find(b), typ, label, "wireframe") for a, b, label in curated.CONTRAST_PAIRS: add_rel(find(a), find(b), "contrasts_with", label, "curated", directed=False) # 목차 계층: 제목이 용어인 항목 → 상위 항목 (part_of), '과정' 아래 형제는 순서(precedes) for b in BOOKS: nodes = toc[b["book_id"]] by_id = {n["id"]: n for n in nodes} kids = defaultdict(list) for n in nodes: kids[n["parent_id"]].append(n) for n in nodes: parent = by_id.get(n["parent_id"]) tn = find(n["title"]) if parent and tn: tp = find(parent["title"]) if tp: add_rel(tn, tp, "part_of", f"목차에서 ‘{parent['title']}’의 하위 항목", "toc", [{"book_id": b["book_id"], "page": n["page"]}]) for pid, ks in kids.items(): parent = by_id.get(pid) if not parent or "과정" not in parent["title"] and parent["kind"] != "chapter": continue if "과정" not in parent["title"]: continue seq = [(k, find(k["title"])) for k in ks] seq = [(k, t) for k, t in seq if t] for (k1, t1), (k2, t2) in zip(seq, seq[1:]): add_rel(t1, t2, "precedes", "다음 단계", "toc", [{"book_id": b["book_id"], "page": k2["page"]}]) # 정의 문장 속에 등장하는 다른 용어 (가장 긴 것 최대 3개) concept_terms = [t for t in terms.values() if t["kind"] == "concept" and len(t["key"]) >= 2 and HANGUL.search(t["name"])] by_len = sorted(concept_terms, key=lambda t: -len(t["key"])) for t in terms.values(): d = t.get("definition") if not d: continue flat = norm_key(d["text"]) found = [] for o in by_len: if o is t or len(o["key"]) < 3 or o["key"] in t["key"]: continue if o["key"] in flat and not any(o["key"] in f["key"] for f in found): found.append(o) if len(found) >= 3: break for o in found: add_rel(t, o, "related_to", f"정의 문장에 ‘{o['name']}’이(가) 등장", "definition", [{"book_id": d["book_id"], "page": d["page"]}]) # 이름 포함: '건강증진사업' → '건강증진' (같은 은하, 가장 긴 포함어 1개) by_galaxy = defaultdict(list) for t in concept_terms: by_galaxy[t["galaxy_id"]].append(t) for t in concept_terms: best = None for o in by_galaxy[t["galaxy_id"]]: if o is t or len(o["key"]) < 3 or len(o["key"]) >= len(t["key"]): continue if o["key"] in t["key"] and (best is None or len(o["key"]) > len(best["key"])): best = o if best: add_rel(t, best, "related_to", f"‘{best['name']}’에서 갈라져 나온 개념", "name") # ---- 7) 출력 toc_out = [] for b in BOOKS: for n in toc[b["book_id"]]: ch, _, chain = chidx.chapter_section(n) n = dict(n) n["galaxy_id"] = galaxy_for(b, ch, chain) toc_out.append(n) term_out = [] for t in sorted(terms.values(), key=lambda t: t["name"]): term_out.append({ "term_id": t["id"], "name": t["name"], "kind": t["kind"], "lang": t.get("lang", "ko"), "aliases": t["aliases"], "search_terms": t.get("search_terms", []), "galaxy_id": t["galaxy_id"], "section_id": t["section_id"], "section_guess": t.get("section_guess", False), "item_id": t.get("item_id"), "section_ids": t["section_ids"], "definition": t["definition"], "easy": t.get("easy"), "example": t.get("example"), "easy_method": t.get("easy_method") or ("wireframe" if t.get("easy") else None), "sources": t["sources"], "index_entries": t["index_entries"], "review_status": "draft", }) out = { "generated_at": time.strftime("%Y-%m-%dT%H:%M:%S"), "books": [{k: v for k, v in b.items() if k not in ("file", "toc_pages", "index_pages")} | {"filename": b["file"]().split("/")[-1]} for b in BOOKS], "galaxies": GALAXIES, "toc": toc_out, "terms": term_out, "relations": list(relations.values()), } DATA_DIR.mkdir(exist_ok=True) path = DATA_DIR / "atlas.json" path.write_text(json.dumps(out, ensure_ascii=False, indent=1)) by_method = defaultdict(int) for r in relations.values(): by_method[r["method"]] += 1 log(f"용어 {len(term_out)}개 (정의 후보 {sum(1 for t in term_out if t['definition'])}개), " f"관계 {len(relations)}개 {dict(by_method)}, {time.time()-t0:.1f}s → {path}") if __name__ == "__main__": build()