diff --git a/scripts/build_summary_annotation_packet.py b/scripts/build_summary_annotation_packet.py new file mode 100644 index 0000000..f6eb16f --- /dev/null +++ b/scripts/build_summary_annotation_packet.py @@ -0,0 +1,164 @@ +#!/usr/bin/env python3 +"""ROUGE 평가용 사람 다중 참조 요약 검수 패킷을 만든다.""" + +from __future__ import annotations + +import argparse +import hashlib +import os +import sys +from collections import defaultdict +from pathlib import Path + +if __package__ in (None, ""): + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from app.engine.summarizer import extractive_summary +from app.engine.training_data import ( + pseudonymous_id, + redact_direct_identifiers, + sanitize_prompt_metadata, +) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("xlsx", type=Path) + parser.add_argument("--out", type=Path, required=True) + parser.add_argument("--text-column", default="에피소드 본문") + parser.add_argument("--title-column", default="에피소드 제목") + parser.add_argument("--group-column", default="id") + parser.add_argument("--limit", type=int, default=300) + parser.add_argument("--min-chars", type=int, default=500) + parser.add_argument("--salt-env", default="DATA_ANONYMIZATION_SALT") + parser.add_argument("--seed", type=int, default=20260820) + return parser.parse_args() + + +def _stable_order(seed: int, key: str) -> str: + return hashlib.sha256(f"{seed}:{key}".encode()).hexdigest() + + +def select_group_diverse(rows: list[dict], limit: int, seed: int) -> list[dict]: + """모든 작성자에서 1건씩 먼저 선택한 후 남은 수를 채운다.""" + grouped: dict[str, list[dict]] = defaultdict(list) + for row in rows: + grouped[row["source_group"]].append(row) + for group_rows in grouped.values(): + group_rows.sort(key=lambda r: _stable_order(seed, r["row_key"])) + + group_names = sorted(grouped, key=lambda g: _stable_order(seed, g)) + chosen: list[dict] = [] + depth = 0 + while len(chosen) < limit: + added = False + for group in group_names: + if depth < len(grouped[group]): + chosen.append(grouped[group][depth]) + added = True + if len(chosen) >= limit: + break + if not added: + break + depth += 1 + return chosen + + +def main() -> int: + args = parse_args() + from openpyxl import Workbook, load_workbook + from openpyxl.styles import Alignment, Font, PatternFill + + salt = os.environ.get(args.salt_env, "") + if not salt: + print(f"{args.salt_env} 환경변수가 필요합니다.", file=sys.stderr) + return 2 + + source = load_workbook(args.xlsx, read_only=True, data_only=True) + sheet = source.worksheets[0] + iterator = sheet.iter_rows(values_only=True) + headers = [str(v).strip() if v is not None else "" for v in next(iterator)] + positions = {name: i for i, name in enumerate(headers)} + missing = [c for c in (args.text_column, args.title_column, args.group_column) if c not in positions] + if missing: + print(f"필수 컬럼 없음: {missing}", file=sys.stderr) + return 2 + + candidates = [] + for rownum, values in enumerate(iterator, start=2): + text = str(values[positions[args.text_column]] or "").strip() + raw_group = str(values[positions[args.group_column]] or "").strip() + if len(text) < args.min_chars or not raw_group: + continue + text = redact_direct_identifiers(text) + title = sanitize_prompt_metadata( + str(values[positions[args.title_column]] or "").strip() + ) + candidates.append({ + "row_key": f"{sheet.title}:{rownum}", + "source_group": pseudonymous_id(raw_group, salt), + "episode_title": title, + "source_text": text, + }) + source.close() + + selected = select_group_diverse(candidates, args.limit, args.seed) + if len(selected) < args.limit: + print(f"후보가 부족합니다: {len(selected)}/{args.limit}", file=sys.stderr) + return 2 + + workbook = Workbook() + ws = workbook.active + ws.title = "annotations" + columns = [ + "annotation_id", "source_group", "episode_title", "source_text", + "baseline_extractive", "reference_summary_1", "reviewer_1", + "reference_summary_2", "reviewer_2", "status", "notes", + ] + ws.append(columns) + for index, row in enumerate(selected, start=1): + baseline = extractive_summary(row["source_text"], ratio=0.3).final + ws.append([ + f"summary-{index:04d}", row["source_group"], row["episode_title"], + row["source_text"], baseline, "", "", "", "", "pending", "", + ]) + + header_fill = PatternFill("solid", fgColor="D9EAF7") + for cell in ws[1]: + cell.font = Font(bold=True) + cell.fill = header_fill + ws.freeze_panes = "A2" + ws.auto_filter.ref = ws.dimensions + widths = {"A": 16, "B": 34, "C": 30, "D": 90, "E": 70, + "F": 70, "G": 18, "H": 70, "I": 18, "J": 14, "K": 30} + for column, width in widths.items(): + ws.column_dimensions[column].width = width + for row in ws.iter_rows(min_row=2): + for cell in row: + cell.alignment = Alignment(vertical="top", wrap_text=True) + + guide = workbook.create_sheet("guidelines") + guidance = [ + ("목적", "ROUGE recall 65 평가를 위한 사람 참조 요약 2종 구축"), + ("독립 작성", "검수자 1·2는 서로의 요약을 보지 않고 작성한다."), + ("길이", "원문 문장 수의 약 25~35%. 필요하면 의미 유지 범위에서 조정한다."), + ("필수", "핵심 인물, 사건, 시간·장소, 원인과 결과를 보존한다."), + ("금지", "원문에 없는 사실·감정·평가를 추가하지 않는다."), + ("문체", "자연스런 한국어 줄글로 작성하고 목록·제목은 쓰지 않는다."), + ("완료", "두 요약과 검수자명을 입력한 후 status를 completed로 변경한다."), + ] + for key, value in guidance: + guide.append([key, value]) + guide.column_dimensions["A"].width = 18 + guide.column_dimensions["B"].width = 100 + for cell in guide[1]: + cell.font = Font(bold=True) + + args.out.parent.mkdir(parents=True, exist_ok=True) + workbook.save(args.out) + print(f"저장: {args.out} / {len(selected)}건 / 작성자 그룹 {len({r['source_group'] for r in selected})}개") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/download_life_writing_drive.py b/scripts/download_life_writing_drive.py new file mode 100644 index 0000000..10cb61f --- /dev/null +++ b/scripts/download_life_writing_drive.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python3 +"""수령한 생활 수기집 목록의 Google Drive 공유 원본을 다운로드한다.""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +from pathlib import Path + +FILE_ID = re.compile(r"drive\.google\.com/file/d/([^/]+)") +FOLDER_ID = re.compile(r"drive\.google\.com/drive/folders/([^/?]+)") + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("manifest", type=Path) + parser.add_argument("--out-dir", type=Path, required=True) + parser.add_argument("--report", type=Path, required=True) + return parser.parse_args() + + +def drive_kind(url: str) -> tuple[str, str] | None: + if match := FILE_ID.search(url): + return "file", match.group(1) + if match := FOLDER_ID.search(url): + return "folder", match.group(1) + return None + + +def main() -> int: + args = parse_args() + try: + import gdown + from openpyxl import load_workbook + except ImportError: + print("gdown과 openpyxl이 필요합니다.", file=sys.stderr) + return 2 + + workbook = load_workbook(args.manifest, read_only=False, data_only=False) + sheet = workbook.worksheets[0] + results = [] + args.out_dir.mkdir(parents=True, exist_ok=True) + + for rownum in range(3, sheet.max_row + 1): + cell = sheet.cell(row=rownum, column=10) + url = cell.hyperlink.target if cell.hyperlink else "" + parsed = drive_kind(url or "") + if not parsed: + continue + kind, drive_id = parsed + target = args.out_dir / f"source_{rownum:03d}" + target.mkdir(parents=True, exist_ok=True) + result = { + "row": rownum, + "kind": kind, + "drive_id": drive_id, + "url": url, + "label": str(cell.value or ""), + "status": "failed", + "files": [], + } + try: + if kind == "folder": + gdown.download_folder( + url=url, output=str(target), quiet=False, remaining_ok=True, + ) + else: + gdown.download(id=drive_id, output=str(target) + "/", quiet=False) + files = sorted(str(p.relative_to(args.out_dir)) for p in target.rglob("*") if p.is_file()) + result["files"] = files + result["status"] = "downloaded" if files else "empty" + except Exception as exc: # noqa: BLE001 + result["error"] = str(exc) + results.append(result) + + report = { + "manifest": args.manifest.name, + "drive_links": len(results), + "downloaded": sum(r["status"] == "downloaded" for r in results), + "failed_or_empty": sum(r["status"] != "downloaded" for r in results), + "results": results, + } + args.report.parent.mkdir(parents=True, exist_ok=True) + args.report.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") + print(json.dumps({k: v for k, v in report.items() if k != "results"}, ensure_ascii=False)) + return 0 if report["downloaded"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_download_life_writing_drive.py b/tests/test_download_life_writing_drive.py new file mode 100644 index 0000000..2a40e22 --- /dev/null +++ b/tests/test_download_life_writing_drive.py @@ -0,0 +1,9 @@ +from scripts.download_life_writing_drive import drive_kind + + +def test_drive_kind_parses_file_and_folder(): + assert drive_kind("https://drive.google.com/file/d/abc123/view") == ("file", "abc123") + assert drive_kind("https://drive.google.com/drive/folders/xyz789?usp=drive_link") == ( + "folder", "xyz789" + ) + assert drive_kind("https://example.com") is None diff --git a/tests/test_summary_annotation_packet.py b/tests/test_summary_annotation_packet.py new file mode 100644 index 0000000..822efcc --- /dev/null +++ b/tests/test_summary_annotation_packet.py @@ -0,0 +1,12 @@ +from scripts.build_summary_annotation_packet import select_group_diverse + + +def test_select_group_diverse_takes_each_group_before_seconds(): + rows = [ + {"source_group": "a", "row_key": "a1"}, + {"source_group": "a", "row_key": "a2"}, + {"source_group": "b", "row_key": "b1"}, + {"source_group": "c", "row_key": "c1"}, + ] + selected = select_group_diverse(rows, limit=3, seed=1) + assert {row["source_group"] for row in selected} == {"a", "b", "c"}