실습 3 — 정제하고 조각 나누기

import json, re

docs = [json.loads(line) for line in open("out_docs_raw.jsonl", encoding="utf-8")]


def fix_pdf_text(text):
    """쪽 번호(- 1 -)가 있는 글만 PDF 에서 뽑은 글로 보고, 쪽 머리말과 쪽 번호를 빼고 끊긴 줄을 잇는다"""
    lines = text.split("\n")
    if not any(re.fullmatch(r"- \d+ -", l.strip()) for l in lines):
        return text                                     # PDF 에서 뽑은 글이 아니면 그대로 둔다
    header = lines[0]                                   # 쪽마다 되풀이되는 머리말
    lines = [l for l in lines if l != header and not re.fullmatch(r"- \d+ -", l.strip())]
    out = []
    for l in lines:
        if out and out[-1] and l and not out[-1].endswith("."):
            out[-1] += l                                # 앞 줄이 문장 끝(.)이 아니면 끊긴 줄로 보고 잇는다
        else:
            out.append(l)
    return "\n".join(out)


expense = next(d for d in docs if d["경로"] == "finance/경비정산기준.txt")
print("--- 고치기 전(앞 8줄) ---")
print("\n".join(expense["글"].split("\n")[:8]))
for d in docs:
    d["글"] = fix_pdf_text(d["글"])
print("--- 고친 뒤(앞 3줄) ---")
print("\n".join(expense["글"].strip().split("\n")[:3]))

PHONE = re.compile(r"01[016789]-\d{3,4}-\d{4}")            # 휴대폰 번호 꼴
EMAIL = re.compile(r"[\w.-]+@[\w-]+\.[\w.]+")               # 메일 주소 꼴

masked = 0
for d in docs:
    text, n1 = PHONE.subn("[전화번호]", d["글"])             # subn: 바꾼 글과 바꾼 횟수를 함께 준다
    text, n2 = EMAIL.subn("[메일]", text)
    text = re.sub(r"[ \t]+", " ", text)                      # 겹친 빈칸을 하나로
    text = re.sub(r"\n{3,}", "\n\n", text).strip()           # 빈 줄이 셋 이상이면 둘로
    d["글"], masked = text, masked + n1 + n2
    if n1 + n2:
        print(d["경로"], "에서", n1 + n2, "개를 가렸습니다")
security = next(d for d in docs if d["경로"] == "it/보안수칙.md")
print(security["글"].split("## 담당자")[1].strip())

# 이름이 짧은 것부터 본다. 사본 표시(" (1)")가 없는 원래 이름이 남게 된다
seen, unique = {}, []
for d in sorted(docs, key=lambda d: len(d["경로"])):
    if d["해시"] in seen:
        print("똑같은 파일:", d["경로"], "=", seen[d["해시"]])
        continue
    seen[d["해시"]] = d["경로"]
    unique.append(d)

# 제목이 같은 문서는 버전이 가장 높은 것만 남긴다
latest = {}
for d in unique:
    if d["제목"] not in latest or d["버전"] > latest[d["제목"]]["버전"]:
        latest[d["제목"]] = d
for d in unique:
    if latest[d["제목"]] is not d:
        new = latest[d["제목"]]
        print("옛 판:", d["경로"], f"(판 {d['버전']}) → 남길 판:", new["경로"], f"(판 {new['버전']})")
        # 두 판에서 "반차" 절의 글을 견준다
        for x in (d, new):
            print("   ", x["작성일"], x["글"].split("## 반차")[1].split("##")[0].split())
clean = sorted((d for d in unique if latest[d["제목"]] is d), key=lambda d: d["경로"])
print(len(docs), "개 →", len(clean), "개")

def split_doc(d):
    """'## 제목', '제N조', 'N. 제목', 표의 행을 경계로 조각을 나누고, 조각마다 라벨을 붙인다"""
    pieces, head, body = [], "", []
    for line in d["글"].split("\n"):
        if line.startswith("# ") or line.startswith("["):       # 문서 제목 줄은 라벨에 있으므로 뺀다
            continue
        is_head = line.startswith("## ") or re.match(r"\d+\. ", line)            # 절 제목 줄
        is_item = re.match(r"제\d+조", line) or (d["경로"].endswith(".csv") and line.strip())   # 조항, 표의 행
        if is_head or is_item:
            if "".join(body).strip():
                pieces.append((head, body))
            head, body = (line.lstrip("# "), []) if is_head else ("", [line])
        else:
            body.append(line)
    if "".join(body).strip():
        pieces.append((head, body))
    chunks = []
    for n, (h, b) in enumerate(pieces):
        text = "\n".join(l for l in b if l.strip())             # 조각 안의 빈 줄은 뺀다
        chunks.append({"id": f"{d['경로']}#{n}", "글": f"[{d['제목']}] {h}".strip() + "\n" + text,
                       **{k: d[k] for k in ("경로", "제목", "부서", "등급", "버전", "작성일")}})
    return chunks


chunks = [c for d in clean for c in split_doc(d)]
with open("out_docs_clean.jsonl", "w", encoding="utf-8") as fp:
    fp.writelines(json.dumps(d, ensure_ascii=False) + "\n" for d in clean)
with open("out_chunks.jsonl", "w", encoding="utf-8") as fp:
    fp.writelines(json.dumps(c, ensure_ascii=False) + "\n" for c in chunks)

print("조각", len(chunks), "개")
import pandas as pd
# 결과: 문서마다 조각 수와 등급
print(pd.DataFrame(chunks).groupby(["제목", "등급"]).size().rename("조각 수").reset_index().to_string(index=False))
for c in chunks[:2] + [c for c in chunks if "불량률" in c["제목"]][:1]:
    print(repr(c["글"]))