401 lines
18 KiB
Python
401 lines
18 KiB
Python
#!/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"(?<![0-9A-Za-z가-힣])(?:19|20)?\d{2}"
|
|
r"(?:가단|가합|고단|고정|카단|카합|나|노|누|다|도|두|라|마|재|허|구합)\d+"
|
|
)
|
|
LIST_ROW_RE = re.compile(
|
|
r'<a\s+href="view\.do\?brdctsno=(\d+)[^"]*"[^>]*title="상세보기"[^>]*>'
|
|
r"(.*?)</a>\s*</td>\s*<td>(\d{4}-\d{2}-\d{2})</td>",
|
|
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"<script\b.*?</script>|<style\b.*?</style>", " ", value, flags=re.I | re.S)
|
|
value = re.sub(r"<br\s*/?>|</p>|</li>", "\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'<div\s+class="view_contents"[^>]*>(.*?)</div>', 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"[^>]*>(.*?)</a>', 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())
|