from __future__ import annotations import argparse import csv import json import re import sys import time from pathlib import Path from urllib.parse import urlencode from urllib.request import Request, urlopen from bs4 import BeautifulSoup ROOT_DIR = Path(__file__).resolve().parents[1] DEFAULT_SOURCE_CSV = ROOT_DIR / "scripts" / "data" / "latin_fallback_presets.csv" DEFAULT_OVERRIDE_PATH = ROOT_DIR / "scripts" / "data" / "plant_name_overrides.json" KPNI_URL = "https://www.nature.go.kr/kpni/stndasrch/dtl/selectNtnStndaPlantList2.do" HEADERS = { "User-Agent": "Mozilla/5.0", "Accept-Language": "ko,en-US;q=0.9,en;q=0.8", "Content-Type": "application/x-www-form-urlencoded", } INFRASPECIFIC_MARKERS = {"subsp.", "subsp", "ssp.", "ssp", "var.", "var", "f.", "f", "forma", "cv.", "cv", "x", "×"} def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Fill plant name overrides with Korean names from KPNI exact species matches.") parser.add_argument("--source-csv", default=str(DEFAULT_SOURCE_CSV)) parser.add_argument("--out", default=str(DEFAULT_OVERRIDE_PATH)) parser.add_argument("--delay", type=float, default=0.08) parser.add_argument("--limit", type=int, default=0) return parser.parse_args() def normalize(value: str) -> str: return " ".join((value or "").strip().split()) def strip_authors(full_name: str) -> str: tokens = normalize(full_name).split() if len(tokens) < 2: return normalize(full_name) kept: list[str] = [] for index, token in enumerate(tokens): if index < 2: kept.append(token) continue lowered = token.lower() if lowered in INFRASPECIFIC_MARKERS: kept.append(token) continue if token.startswith("(") or re.match(r"^[A-Z]", token) or token.endswith("."): break kept.append(token) return " ".join(kept) def load_source_rows(path: Path) -> list[dict]: with path.open("r", encoding="utf-8-sig", newline="") as fp: return list(csv.DictReader(fp)) def load_overrides(path: Path) -> dict[str, dict[str, str]]: if not path.exists(): return {} return json.loads(path.read_text(encoding="utf-8")) def save_overrides(path: Path, payload: dict[str, dict[str, str]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) ordered = {name: payload[name] for name in sorted(payload)} path.write_text(json.dumps(ordered, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def should_skip_existing(scientific_name: str, existing: dict[str, str]) -> bool: if not existing: return False current_ko = normalize(existing.get("ko", "")) source = normalize(existing.get("source", "")) if not current_ko or current_ko == scientific_name: return False # Keep manual overrides. Entries from automated sources can be upgraded. if not source: return True return False def fetch_kpni_rows(scientific_name: str) -> list[dict[str, str]]: payload = urlencode({"plantInfoSearchWrd": scientific_name}).encode("utf-8") request = Request(KPNI_URL, data=payload, headers=HEADERS) with urlopen(request, timeout=60) as response: html = response.read().decode("utf-8", "ignore") soup = BeautifulSoup(html, "html.parser") tbody = soup.find("tbody") if not tbody: return [] rows: list[dict[str, str]] = [] for tr in tbody.find_all("tr"): cells = [normalize(td.get_text(" ", strip=True)) for td in tr.find_all("td")] if len(cells) < 5: continue rows.append( { "kind": cells[0], "status": cells[1], "scientific_name": cells[2], "korean_name": cells[3], "updated_at": cells[4], } ) return rows def pick_best_match(query: str, rows: list[dict[str, str]]) -> dict[str, str] | None: query = normalize(query).lower() candidates: list[dict[str, str]] = [] for row in rows: if not row.get("korean_name"): continue stripped = strip_authors(row.get("scientific_name", "")).lower() if stripped != query: continue candidates.append(row) if not candidates: return None def score(row: dict[str, str]) -> tuple[int, int, int]: status = row.get("status", "") kind = row.get("kind", "") scientific_name = row.get("scientific_name", "") return ( 1 if status == "정명" else 0, 1 if kind == "자생식물" else 0, -len(scientific_name), ) candidates.sort(key=score, reverse=True) return candidates[0] def main() -> None: try: sys.stdout.reconfigure(encoding="utf-8") except Exception: pass args = parse_args() source_path = Path(args.source_csv) out_path = Path(args.out) source_rows = load_source_rows(source_path) if args.limit > 0: source_rows = source_rows[: args.limit] overrides = load_overrides(out_path) updated = 0 already_present = 0 no_match = 0 sample = [] for index, row in enumerate(source_rows, start=1): scientific_name = normalize(row.get("type_name", "")) if not scientific_name: continue existing = overrides.get(scientific_name, {}) if should_skip_existing(scientific_name, existing): already_present += 1 continue rows = fetch_kpni_rows(scientific_name) match = pick_best_match(scientific_name, rows) if not match: no_match += 1 else: overrides[scientific_name] = { **existing, "ko": normalize(match["korean_name"]), "source": "KPNI", "source_kind": normalize(match["kind"]), "source_status": normalize(match["status"]), "source_scientific_name": normalize(match["scientific_name"]), "source_updated_at": normalize(match["updated_at"]), } updated += 1 if len(sample) < 30: sample.append( { "scientific_name": scientific_name, "ko": overrides[scientific_name]["ko"], "source_kind": overrides[scientific_name]["source_kind"], "source_status": overrides[scientific_name]["source_status"], } ) save_overrides(out_path, overrides) if index % 100 == 0: print(json.dumps({"progress": index, "updated": updated, "already_present": already_present, "no_match": no_match}, ensure_ascii=False)) if args.delay > 0: time.sleep(args.delay) save_overrides(out_path, overrides) print( json.dumps( { "source_csv": str(source_path), "out": str(out_path), "processed": len(source_rows), "updated": updated, "already_present": already_present, "no_match": no_match, "sample": sample, }, ensure_ascii=False, indent=2, ) ) if __name__ == "__main__": main()