from __future__ import annotations import json import os import sys import time from pathlib import Path from urllib.parse import quote from urllib.request import Request, urlopen import mysql.connector ROOT_DIR = Path(__file__).resolve().parents[1] DEFAULT_ENV_PATH = ROOT_DIR / ".env" DEFAULT_OVERRIDE_PATH = ROOT_DIR / "scripts" / "data" / "plant_name_overrides.json" def load_env(env_path: str) -> None: path = Path(env_path) if not path.exists(): return for line in path.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) os.environ.setdefault(key.strip(), value.strip()) def create_db_connection(): return mysql.connector.connect( host=os.environ["DB_HOST"], port=int(os.environ.get("DB_PORT", "3306")), user=os.environ["DB_USER"], password=os.environ["DB_PASSWORD"], database=os.environ["DB_NAME"], charset="utf8mb4", ) def normalize(value: str) -> str: return " ".join((value or "").strip().split()) def fetch_json(url: str) -> dict: req = Request(url, headers={"User-Agent": "Mozilla/5.0", "Connection": "close"}) with urlopen(req, timeout=120) as response: return json.loads(response.read().decode("utf-8")) def pick_better_label_entry(current: dict[str, str] | None, candidate: dict[str, str]) -> dict[str, str]: if current is None: return candidate def score(entry: dict[str, str]) -> tuple[int, int, int, int]: non_empty = sum(1 for key in ("ko", "en", "ja", "zh") if entry.get(key)) return ( 1 if entry.get("ko") else 0, 1 if entry.get("ja") else 0, 1 if entry.get("zh") else 0, non_empty, ) if score(candidate) > score(current): return candidate return current def fetch_wikidata_labels(scientific_names: list[str], chunk_size: int = 80) -> dict[str, dict[str, str]]: labels: dict[str, dict[str, str]] = {} unique_names = [] seen = set() for name in scientific_names: name = normalize(name) if name and name not in seen: unique_names.append(name) seen.add(name) for start in range(0, len(unique_names), chunk_size): chunk = unique_names[start:start + chunk_size] values = " ".join(f'"{name.replace(chr(34), r"\\\"")}"' for name in chunk) query = f""" SELECT ?scientificName ?labelKo ?labelEn ?labelJa ?labelZh WHERE {{ VALUES ?scientificName {{ {values} }} ?item wdt:P225 ?scientificName . OPTIONAL {{ ?item rdfs:label ?labelKo FILTER(LANG(?labelKo) = "ko") }} OPTIONAL {{ ?item rdfs:label ?labelEn FILTER(LANG(?labelEn) = "en") }} OPTIONAL {{ ?item rdfs:label ?labelJa FILTER(LANG(?labelJa) = "ja") }} OPTIONAL {{ ?item rdfs:label ?labelZh FILTER(LANG(?labelZh) = "zh") }} }} """ url = "https://query.wikidata.org/sparql?format=json&query=" + quote(query) data = fetch_json(url) for row in data.get("results", {}).get("bindings", []): scientific_name = normalize(row["scientificName"]["value"]) candidate = { "ko": normalize(row.get("labelKo", {}).get("value", "")), "en": normalize(row.get("labelEn", {}).get("value", "")), "ja": normalize(row.get("labelJa", {}).get("value", "")), "zh": normalize(row.get("labelZh", {}).get("value", "")), } labels[scientific_name] = pick_better_label_entry(labels.get(scientific_name), candidate) time.sleep(0.2) return labels def find_target_rows(rows_by_name: dict[str, list[dict]], scientific_name: str, target_ko: str) -> tuple[dict | None, dict | None]: source_row = (rows_by_name.get(scientific_name) or [None])[0] target_row = None for row in rows_by_name.get(target_ko) or []: if not source_row or row["id"] != source_row["id"]: target_row = row break return source_row, target_row def apply_update(cursor, row_id: int, *, type_name: str, type_name_ko: str, type_name_en: str, type_name_ja: str, type_name_zh: str, is_active: int) -> None: cursor.execute( """ UPDATE plant_reminder_presets SET type_name = %s, type_name_ko = %s, type_name_en = %s, type_name_ja = %s, type_name_zh = %s, is_active = %s, updated_at = NOW() WHERE id = %s """, (type_name, type_name_ko, type_name_en, type_name_ja, type_name_zh, is_active, row_id), ) def main() -> None: try: sys.stdout.reconfigure(encoding="utf-8") except Exception: pass load_env(str(DEFAULT_ENV_PATH)) overrides = json.loads(DEFAULT_OVERRIDE_PATH.read_text(encoding="utf-8")) wikidata_labels = fetch_wikidata_labels(list(overrides.keys())) conn = create_db_connection() cursor = conn.cursor(dictionary=True) updated = 0 localized_only = 0 renamed = 0 deactivated_duplicates = 0 skipped_missing = 0 changed = [] try: cursor.execute("SELECT id, type_name, type_name_ko, type_name_en, type_name_ja, type_name_zh, is_active FROM plant_reminder_presets") rows = cursor.fetchall() rows_by_name: dict[str, list[dict]] = {} for row in rows: rows_by_name.setdefault(row["type_name"], []).append(dict(row)) for scientific_name, payload in overrides.items(): scientific_name = normalize(scientific_name) labels = wikidata_labels.get(scientific_name, {}) target_ko = normalize(payload.get("ko") or labels.get("ko") or scientific_name) target_en = normalize(payload.get("en") or labels.get("en") or scientific_name) target_ja = normalize(payload.get("ja") or labels.get("ja") or scientific_name) target_zh = normalize(payload.get("zh") or labels.get("zh") or scientific_name) source_row, target_row = find_target_rows(rows_by_name, scientific_name, target_ko) if source_row and target_row: # Generic/common-name row already exists. Canonicalize that row and hide the duplicate scientific row. apply_update( cursor, target_row["id"], type_name=target_row["type_name"], type_name_ko=target_ko, type_name_en=target_en, type_name_ja=target_ja, type_name_zh=target_zh, is_active=1, ) target_row.update( { "type_name_ko": target_ko, "type_name_en": target_en, "type_name_ja": target_ja, "type_name_zh": target_zh, "is_active": 1, } ) if source_row["is_active"] != 0: apply_update( cursor, source_row["id"], type_name=source_row["type_name"], type_name_ko=target_ko, type_name_en=target_en, type_name_ja=target_ja, type_name_zh=target_zh, is_active=0, ) source_row.update( { "type_name_ko": target_ko, "type_name_en": target_en, "type_name_ja": target_ja, "type_name_zh": target_zh, "is_active": 0, } ) deactivated_duplicates += 1 updated += 1 localized_only += 1 changed.append( { "scientific_name": scientific_name, "type_name": target_row["type_name"], "target_ko": target_ko, "duplicate_deactivated": source_row["type_name"], } ) continue if source_row: next_type_name = target_ko if target_ko != scientific_name and rows_by_name.get(target_ko): next_type_name = scientific_name old_type_name = source_row["type_name"] apply_update( cursor, source_row["id"], type_name=next_type_name, type_name_ko=target_ko, type_name_en=target_en, type_name_ja=target_ja, type_name_zh=target_zh, is_active=source_row["is_active"], ) source_row.update( { "type_name": next_type_name, "type_name_ko": target_ko, "type_name_en": target_en, "type_name_ja": target_ja, "type_name_zh": target_zh, } ) if old_type_name != next_type_name: rows_by_name.get(old_type_name, []).remove(source_row) if not rows_by_name.get(old_type_name): rows_by_name.pop(old_type_name, None) rows_by_name.setdefault(next_type_name, []).append(source_row) renamed += 1 else: localized_only += 1 updated += 1 changed.append( { "scientific_name": scientific_name, "type_name": next_type_name, "target_ko": target_ko, } ) continue if target_row: apply_update( cursor, target_row["id"], type_name=target_row["type_name"], type_name_ko=target_ko, type_name_en=target_en, type_name_ja=target_ja, type_name_zh=target_zh, is_active=target_row["is_active"], ) target_row.update( { "type_name_ko": target_ko, "type_name_en": target_en, "type_name_ja": target_ja, "type_name_zh": target_zh, } ) updated += 1 localized_only += 1 changed.append( { "scientific_name": scientific_name, "type_name": target_row["type_name"], "target_ko": target_ko, } ) continue skipped_missing += 1 conn.commit() print( json.dumps( { "updated": updated, "renamed": renamed, "localized_only": localized_only, "deactivated_duplicates": deactivated_duplicates, "skipped_missing": skipped_missing, "changed": changed[:150], }, ensure_ascii=False, indent=2, ) ) finally: cursor.close() conn.close() if __name__ == "__main__": main()