#!/usr/bin/env python3
"""한국저작권위원회 판례를 수집해 프로젝트용 후보 JSONL을 만든다.
공식 목록 전체를 먼저 수집한 뒤 제목 기반 후보의 상세 페이지에서 국내 사건번호와
판시 내용을 확인한다. 네트워크 결과는 ``--cache-dir`` 아래에 캐시하므로 중단 후 다시
실행해도 이미 받은 페이지를 재요청하지 않는다.
"""
from __future__ import annotations
import argparse
import html
import json
import re
import subprocess
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from urllib.parse import urljoin
from urllib.parse import urlencode, urlparse
import openpyxl
BASE = "https://www.copyright.or.kr/information-materials/trend/precedents/"
LIST_URL = urljoin(BASE, "list.do?pageIndex={page}")
VIEW_URL = urljoin(BASE, "view.do?brdctsno={board_id}")
USER_AGENT = "Mozilla/5.0 (compatible; O2O-Precedent-Research/1.0)"
CASE_RE = re.compile(
r"(?]*title="상세보기"[^>]*>'
r"(.*?)\s*\s*| (\d{4}-\d{2}-\d{2}) | ",
re.S,
)
# 제목에서 하나라도 발견되면 상세 검토 대상으로 삼는다. 프로젝트 범위는 자서전과
# 출판물의 본문·구조·인용·저작인격권 및 부속 시각/음성 자산이다.
TITLE_TERMS = (
"어문", "문학", "소설", "수필", "시집", "시인", "가사", "도서", "서적",
"출판", "저서", "저술", "기사", "뉴스", "신문", "잡지", "논문", "보고서",
"교재", "교과서", "문제집", "시험문제", "시험지", "수험서", "책자", "학습지",
"편집저작물", "제안서", "설명문", "상세페이지", "사전", "백과", "번역",
"대본", "각본", "시나리오", "스토리", "줄거리", "캐릭터", "편지", "일기",
"이메일", "블로그", "게시글", "SNS", "웹툰", "만화", "문구", "제호", "제목",
"강연", "설교", "폰트", "편집", "요약", "표절", "인용", "공정이용",
"공정 이용", "실질적 유사", "의거", "아이디어", "창작성", "성명표시",
"동일성유지", "공표권", "2차적저작물", "사진", "이미지", "그림", "표지",
"음악", "음원", "노래", "오디오북",
)
WORK_TYPE_TERMS = {
"literary": (
"어문", "문학", "소설", "수필", "시집", "시인", "가사", "도서", "서적",
"출판", "저서", "저술", "기사", "뉴스", "신문", "잡지", "논문", "보고서",
"교재", "교과서", "문제집", "시험문제", "시험지", "수험서", "책자", "학습지",
"편집저작물", "제안서", "설명문", "상세페이지", "사전", "백과", "번역",
"대본", "각본", "시나리오", "스토리", "줄거리", "편지", "일기", "이메일",
"블로그", "게시글", "강연", "설교", "요약",
),
"visual": ("사진", "이미지", "그림", "미술", "캐릭터", "만화", "웹툰", "표지"),
"musical": ("음악", "음원", "노래", "가요", "작곡", "오디오북"),
}
TAG_TERMS = {
"reproduction": (
"복제", "베끼", "표절", "무단 사용", "무단사용", "도용", "저작권 침해",
"저작권침해", "저작재산권 침해", "저작재산권침해", "실질적 유사",
),
"derivative_work": ("2차적저작물", "이차적저작물", "번역", "각색", "변형", "요약"),
"public_transmission": ("공중송신", "전송", "게시", "업로드", "온라인", "인터넷"),
"distribution": ("배포", "판매", "발행", "출판"),
"publication": ("공표", "미공표", "저작인격권"),
"attribution": ("성명표시", "성명 표시", "저작자 표시", "저작인격권"),
"integrity": ("동일성유지", "동일성 유지", "저작인격권"),
"citation_missing": ("인용", "출처표시", "출처 표시"),
"false_authorship": ("자기 저작", "본인 저작", "저작자 아닌", "대필"),
}
CRITERIA_TERMS = (
"실질적 유사성", "의거관계", "의거 관계", "접근 가능성", "보호되는 표현",
"창작적 표현", "창작성", "아이디어와 표현", "공정이용", "공정 이용", "정당한 인용",
"인용 요건", "저작물성", "2차적저작물", "저작인격권", "성명표시권",
"동일성유지권", "공표권",
)
def clean_html(value: str) -> str:
value = re.sub(r"|", " ", value, flags=re.I | re.S)
value = re.sub(r"||", "\n", value, flags=re.I)
value = re.sub(r"<[^>]+>", " ", value)
value = html.unescape(value).replace("\xa0", " ")
lines = [re.sub(r"\s+", " ", line).strip() for line in value.splitlines()]
return "\n".join(line for line in lines if line)
def fetch(url: str, *, retries: int = 2) -> str:
if urlparse(url).hostname != "www.copyright.or.kr":
raise ValueError(f"허용되지 않은 호스트: {url}")
error: Exception | None = None
for attempt in range(retries):
try:
result = subprocess.run(
["curl", "-fsSL", "--max-time", "30", "-A", USER_AGENT, url],
check=True,
capture_output=True,
timeout=40,
)
return result.stdout.decode("utf-8", errors="replace")
except Exception as exc: # pragma: no cover - network behavior
error = exc
time.sleep(2 ** attempt)
raise RuntimeError(f"요청 실패: {url}: {error}")
def cached_fetch(url: str, path: Path) -> str:
if path.exists() and path.stat().st_size > 100:
return path.read_text(encoding="utf-8")
body = fetch(url)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(body, encoding="utf-8")
return body
def parse_list(body: str) -> list[dict[str, str]]:
return [
{
"board_id": board_id,
"title": clean_html(title),
"registered_date": registered_date,
"source_url": VIEW_URL.format(board_id=board_id),
}
for board_id, title, registered_date in LIST_ROW_RE.findall(body)
]
def title_candidate(title: str) -> bool:
folded = title.casefold().replace("파기사유", "").replace("인용상표", "")
if any(term in folded for term in ("상표관련", "인용발명", "특허")) and "제호" not in folded:
return False
return any(term.casefold() in folded for term in TITLE_TERMS)
def extract_detail(
item: dict[str, str], body: str, *, case_ids_override: tuple[str, ...] = ()
) -> list[dict]:
match = re.search(r']*>(.*?)', body, re.I | re.S)
content = clean_html(match.group(1) if match else "")
attachment_names = " ".join(
clean_html(x) for x in re.findall(r'class="attachment"[^>]*>(.*?)', body, re.I | re.S)
)
searchable = f"{item['title']}\n{content}\n{attachment_names}"
attachment_case_ids = find_case_ids(attachment_names)
preamble = re.split(
r"(?:【|○|□)?\s*(?:판시\s*사항|판결\s*요지|사실\s*관계|참조\s*조문|참조\s*판례)",
content,
maxsplit=1,
)[0]
primary_case_ids = attachment_case_ids or find_case_ids(preamble[:1500])
case_ids = list(dict.fromkeys((*case_ids_override, *primary_case_ids)))
if not case_ids:
return []
work_types = [
work_type for work_type, terms in WORK_TYPE_TERMS.items()
if any(term.casefold() in searchable.casefold() for term in terms)
]
legal_tags = [
tag for tag, terms in TAG_TERMS.items()
if any(term.casefold() in searchable.casefold() for term in terms)
]
criteria = [term for term in CRITERIA_TERMS if term in searchable]
# 문서가 판시사항/판결요지를 구분하면 그 부분을 우선 사용한다. 그렇지 않은 경우
# 공식 상세 본문 앞부분을 보존한다. 원문을 임의로 법률 요약하지 않는다.
holding = content or item["title"]
marker = re.search(r"(?:판시\s*사항|판결\s*요지)", content)
if marker:
holding = content[marker.start():]
holding = holding[:1200].strip()
return [
{
"case_id": case_id,
"title": item["title"],
"source_url": item["source_url"],
"work_types": work_types,
"legal_tags": legal_tags,
"criteria": criteria,
"holding_summary": holding,
"outcome": None,
"source_board_id": item["board_id"],
"source_registered_date": item["registered_date"],
"selection_basis": "official_title_keyword_and_verified_domestic_case_id",
}
for case_id in case_ids
]
def write_jsonl(path: Path, rows: list[dict]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
"".join(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n" for row in rows),
encoding="utf-8",
)
def canonical_case_id(value: str) -> str:
compact = value.replace("-", "")
match = re.search(
r"(?:19|20)?\d{2}(?:가단|가합|고단|고정|카단|카합|나|노|누|다|도|두|라|마|재|허|구합)\d+",
compact,
)
return match.group(0) if match else value.strip()
def find_case_ids(value: str) -> list[str]:
pattern = re.compile(
r"(?:19|20)?\d{2}(?:가단|가합|고단|고정|카단|카합|나|노|누|다|도|두|라|마|재|허|구합)\d+"
)
return list(dict.fromkeys(pattern.findall(value.replace("-", ""))))
def load_excel_cases(directory: Path) -> dict[str, dict[str, str]]:
cases: dict[str, dict[str, str]] = {}
for path in sorted(directory.glob("*.xlsx")):
workbook = openpyxl.load_workbook(path, read_only=True, data_only=True)
for sheet in workbook.worksheets:
for row in sheet.iter_rows(min_row=2, values_only=True):
values = (list(row) + [None] * 7)[:7]
if values[0] in (None, "") or values[2] in (None, ""):
continue
case_id = canonical_case_id(str(values[2]))
entry = cases.setdefault(case_id, {
"case_id": case_id,
"title": str(values[1] or "").strip(),
"decision_date": str(values[3] or "").strip(),
"holding": "",
"source_files": [],
})
if values[6] and not entry["holding"]:
entry["holding"] = str(values[6]).strip()
if path.name not in entry["source_files"]:
entry["source_files"].append(path.name)
return cases
def merge_excel_sources(
args: argparse.Namespace, records: list[dict]
) -> tuple[list[dict], dict]:
if args.excel_dir is None:
return records, {}
excel_cases = load_excel_cases(args.excel_dir)
official = {row["case_id"]: row for row in records}
overlap_before = sorted(set(excel_cases) & set(official))
missing = [case_id for case_id in excel_cases if case_id not in official]
def search_case(case_id: str) -> tuple[str, list[dict[str, str]]]:
query = urlencode({"searchText": case_id, "searchTarget": "ALL"})
url = urljoin(BASE, f"list.do?{query}")
path = args.cache_dir / "search" / f"{case_id}.html"
try:
return case_id, parse_list(cached_fetch(url, path))
except RuntimeError:
return case_id, []
found_items: dict[str, dict[str, str]] = {}
with ThreadPoolExecutor(max_workers=args.workers) as pool:
futures = [pool.submit(search_case, case_id) for case_id in missing]
for done, future in enumerate(as_completed(futures), 1):
case_id, items = future.result()
if items:
found_items[case_id] = items[0]
if done % 20 == 0 or done == len(futures):
print(f"엑셀 사건번호 역검색 {done}/{len(futures)}건", flush=True)
def load_found(pair: tuple[str, dict[str, str]]) -> tuple[str, list[dict]]:
case_id, item = pair
body = cached_fetch(item["source_url"], args.cache_dir / "view" / f"{item['board_id']}.html")
return case_id, extract_detail(item, body, case_ids_override=(case_id,))
with ThreadPoolExecutor(max_workers=args.workers) as pool:
futures = [pool.submit(load_found, pair) for pair in found_items.items()]
for future in as_completed(futures):
case_id, extracted = future.result()
selected = next((row for row in extracted if row["case_id"] == case_id), None)
if selected:
official[case_id] = selected
for case_id, row in official.items():
excel = excel_cases.get(case_id)
if not excel:
row["excel_duplicate"] = False
continue
row["excel_duplicate"] = True
row["excel_source_files"] = excel["source_files"]
row["excel_title"] = excel["title"]
row["excel_decision_date"] = excel["decision_date"]
if row["holding_summary"] == row["title"] and excel["holding"]:
row["holding_summary"] = excel["holding"][:1200]
result = sorted(official.values(), key=lambda row: row["case_id"])
audit = {
"official_list_count": 2085,
"official_selected_before_excel_merge_count": len(records),
"excel_row_count": 182,
"excel_unique_case_count": len(excel_cases),
"duplicate_before_reverse_lookup_count": len(overlap_before),
"reverse_lookup_found_count": len(set(official) & set(missing)),
"merged_unique_case_count": len(result),
"excel_without_official_detail_count": len(set(excel_cases) - set(official)),
"excel_without_official_detail_case_ids": sorted(set(excel_cases) - set(official)),
}
return result, audit
def collect(args: argparse.Namespace) -> None:
cache = args.cache_dir
pages: dict[int, list[dict[str, str]]] = {}
def load_page(page: int) -> tuple[int, list[dict[str, str]]]:
body = cached_fetch(LIST_URL.format(page=page), cache / "list" / f"{page:03}.html")
return page, parse_list(body)
with ThreadPoolExecutor(max_workers=args.workers) as pool:
futures = [pool.submit(load_page, page) for page in range(1, args.pages + 1)]
for done, future in enumerate(as_completed(futures), 1):
page, items = future.result()
pages[page] = items
if done % 10 == 0 or done == len(futures):
print(f"목록 {done}/{len(futures)}페이지", flush=True)
index = [item for page in sorted(pages) for item in pages[page]]
candidates = [item for item in index if title_candidate(item["title"])]
write_jsonl(args.index_output, index)
write_jsonl(args.candidate_output, candidates)
print(f"목록 {len(index)}건, 제목 후보 {len(candidates)}건", flush=True)
def load_detail(item: dict[str, str]) -> tuple[dict[str, str], list[dict]]:
path = cache / "view" / f"{item['board_id']}.html"
return item, extract_detail(item, cached_fetch(item["source_url"], path))
records: list[dict] = []
with ThreadPoolExecutor(max_workers=args.workers) as pool:
futures = [pool.submit(load_detail, item) for item in candidates]
for done, future in enumerate(as_completed(futures), 1):
_, extracted = future.result()
records.extend(extracted)
if done % 10 == 0 or done == len(futures):
print(f"상세 {done}/{len(futures)}건, 사건번호 {len(records)}개", flush=True)
# 같은 사건이 여러 게시물에 있으면 최신 등록 게시물을 대표 출처로 사용한다.
records.sort(key=lambda row: (row["source_registered_date"], row["source_board_id"]), reverse=True)
unique: dict[str, dict] = {}
for row in records:
unique.setdefault(row["case_id"], row)
result = sorted(unique.values(), key=lambda row: row["case_id"])
result, audit = merge_excel_sources(args, result)
before_validation_filter = len(result)
result = [
row for row in result
if row["legal_tags"] and (row["work_types"] or row["criteria"])
]
if audit:
audit["official_list_count"] = len(index)
audit["official_title_candidate_count"] = len(candidates)
audit["merged_before_validation_filter_count"] = before_validation_filter
audit["excluded_missing_engine_labels_count"] = before_validation_filter - len(result)
audit["usable_record_count"] = len(result)
audit["usable_excel_duplicate_count"] = sum(
bool(row.get("excel_duplicate")) for row in result
)
write_jsonl(args.output, result)
if args.audit_output and audit:
args.audit_output.parent.mkdir(parents=True, exist_ok=True)
args.audit_output.write_text(
json.dumps(audit, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
print(f"국내 사건번호 중복 제거 결과 {len(result)}건 -> {args.output}", flush=True)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--pages", type=int, default=209)
parser.add_argument("--workers", type=int, default=6)
parser.add_argument("--cache-dir", type=Path, default=Path("/tmp/copyright-precedents-cache"))
parser.add_argument("--index-output", type=Path, default=Path("/tmp/copyright-precedents-index.jsonl"))
parser.add_argument("--candidate-output", type=Path, default=Path("/tmp/copyright-precedents-candidates.jsonl"))
parser.add_argument("--output", type=Path, default=Path("/tmp/copyright-precedents-selected.jsonl"))
parser.add_argument("--excel-dir", type=Path)
parser.add_argument("--audit-output", type=Path, default=Path("/tmp/copyright-precedents-audit.json"))
collect(parser.parse_args())
return 0
if __name__ == "__main__":
raise SystemExit(main())