from __future__ import annotations import argparse import json import os import re import time from pathlib import Path from urllib.error import HTTPError, URLError from urllib.parse import quote, urlencode from urllib.request import HTTPCookieProcessor, Request, build_opener, urlopen import mysql.connector ROOT_DIR = Path(__file__).resolve().parents[1] DEFAULT_ENV_PATH = ROOT_DIR / ".env" PLANTAE_TAXON_ID = 47126 OVERRIDE_PATH = ROOT_DIR / "scripts" / "data" / "plant_name_overrides.json" NIBR_BASE_URL = "https://species.nibr.go.kr" NIBR_HEADERS = { "User-Agent": "Mozilla/5.0", "Accept-Language": "ko,en-US;q=0.9,en;q=0.8", "Referer": f"{NIBR_BASE_URL}/", } INFRASPECIFIC_MARKERS = {"subsp.", "subsp", "ssp.", "ssp", "var.", "var", "f.", "f", "forma", "cv.", "cv", "x", "×"} CARE_PROFILES = { "succulent": { "days": 12, "sunlight_ko": "밝은 곳", "tip_ko": "과습에 약하므로 흙이 완전히 마른 뒤 물을 주고, 겨울에는 물주기 간격을 더 길게 잡아주세요.", }, "fern": { "days": 5, "sunlight_ko": "밝은 간접광", "tip_ko": "건조에 약하니 흙이 바싹 마르지 않게 유지하고, 공중 습도도 함께 관리해주세요.", }, "orchid": { "days": 6, "sunlight_ko": "밝은 간접광", "tip_ko": "배지가 거의 마를 때 물을 주고, 뿌리 주변에 물이 오래 고이지 않게 통풍을 확보해주세요.", }, "bromeliad": { "days": 6, "sunlight_ko": "밝은 간접광", "tip_ko": "직사광은 피하고, 중심부나 배지가 오래 탁해지지 않도록 물 상태를 주기적으로 점검해주세요.", }, "carnivorous": { "days": 4, "sunlight_ko": "밝은 곳", "tip_ko": "배지가 마르지 않게 유지하고, 가능하면 정수나 빗물을 사용해 염분 축적을 줄여주세요.", }, "herb": { "days": 4, "sunlight_ko": "햇빛 필요", "tip_ko": "하루 몇 시간 이상 햇빛을 확보하고, 흙이 완전히 마르기 전에 수분을 보충해주세요.", }, "flower": { "days": 5, "sunlight_ko": "밝은 곳", "tip_ko": "개화기에는 수분 소모가 크므로 겉흙이 마르면 바로 물을 주고, 시든 꽃은 정리해주세요.", }, "tree": { "days": 8, "sunlight_ko": "밝은 곳", "tip_ko": "빛과 통풍을 충분히 확보하고, 흙이 어느 정도 마른 뒤 깊게 물을 주는 편이 안정적입니다.", }, "foliage": { "days": 7, "sunlight_ko": "밝은 간접광", "tip_ko": "겉흙이 마르면 물을 주고, 잎이 처지거나 끝이 마르면 빛과 습도를 함께 점검해주세요.", }, } KEYWORDS = { "succulent": [ "cactus", "succulent", "aloe", "agave", "echeveria", "sedum", "crassula", "euphorbia", "haworthia", "lithops", "pachypodium", "adenium", "conophytum", "gasteria", "sansevieria", "kalanchoe", "portulaca", "sempervivum", "tylecodon", "pleiospilos", ], "fern": [ "fern", "adiantum", "platycerium", "asplenium", "nephrolepis", "microsorum", "pteris", "selaginella", "maidenhair", "phlebodium", ], "orchid": [ "orchid", "phalaenopsis", "oncidium", "vanda", "paphiopedilum", "zygopetalum", "miltonia", "dendrobium", "cattleya", "vanilla", "ludisia", "macodes", ], "bromeliad": [ "bromeliad", "guzmania", "vriesea", "aechmea", "neoregelia", "tillandsia", ], "carnivorous": [ "pitcher", "flytrap", "drosera", "carnivorous", "nepenthes", "sarracenia", "dionaea", "utricularia", ], "herb": [ "mint", "basil", "rosemary", "thyme", "sage", "parsley", "cilantro", "coriander", "oregano", "fennel", "dill", "chive", "lavender", "marjoram", ], "flower": [ "rose", "lily", "daisy", "iris", "tulip", "hyacinth", "primrose", "campanula", "delphinium", "lupine", "hellebore", "echinacea", "verbena", "gazania", "nemesia", "scabiosa", "astilbe", "dahlia", "begonia", "violet", "snapdragon", "petunia", "carnation", "chrysanthemum", "orchid", ], "tree": [ "tree", "oak", "maple", "pine", "spruce", "fir", "cedar", "juniper", "willow", "fig", "ficus", "palm", "olive", "lemon", "apple", "pear", "plum", "cherry", "elm", ], "foliage": [ "monstera", "philodendron", "pothos", "syngonium", "scindapsus", "anthurium", "alocasia", "aglaonema", "calathea", "stromanthe", "dieffenbachia", "dracaena", "peperomia", "hoya", "pilea", "tradescantia", "ivy", ], } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Generate NDJSON catalog for many unique plant reminder presets from iNaturalist.") parser.add_argument("--env-file", default=str(DEFAULT_ENV_PATH)) parser.add_argument("--out", default=str(ROOT_DIR / "scripts" / "data" / "plant_reminder_catalog_inat_1000.ndjson")) parser.add_argument("--target", type=int, default=1000) parser.add_argument("--max-pages", type=int, default=20) parser.add_argument("--skip-nibr", action="store_true", help="Skip NIBR exact-match lookup for Korean common names.") return parser.parse_args() 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 fetch_json(url: str, attempt: int = 1) -> dict: req = Request(url, headers={"User-Agent": "Mozilla/5.0", "Connection": "close"}) try: with urlopen(req, timeout=60) as response: return json.loads(response.read().decode("utf-8")) except (HTTPError, URLError, TimeoutError): if attempt >= 5: raise time.sleep(attempt * 2) return fetch_json(url, attempt + 1) _TRANSLATION_CACHE: dict[tuple[str, str, str], str] = {} def translate_text(text: str, target_lang: str, source_lang: str = "auto", attempt: int = 1) -> str: if not text: return "" key = (text, target_lang, source_lang) if key in _TRANSLATION_CACHE: return _TRANSLATION_CACHE[key] url = ( "https://translate.googleapis.com/translate_a/single" f"?client=gtx&sl={quote(source_lang)}&tl={quote(target_lang)}&dt=t&q={quote(text)}" ) req = Request(url, headers={"User-Agent": "Mozilla/5.0", "Connection": "close"}) try: with urlopen(req, timeout=30) as response: data = json.loads(response.read().decode("utf-8")) except (HTTPError, URLError, TimeoutError): if attempt >= 5: raise time.sleep(attempt * 2) return translate_text(text, target_lang, source_lang=source_lang, attempt=attempt + 1) translated = "".join(part[0] for part in data[0] if part and part[0]).strip() _TRANSLATION_CACHE[key] = translated time.sleep(0.08) return translated def batch_translate_texts(texts: list[str], target_lang: str, source_lang: str = "auto", chunk_size: int = 40) -> dict[str, str]: unique_texts = [] seen = set() for text in texts: if text and text not in seen: unique_texts.append(text) seen.add(text) result: dict[str, str] = {} for text in unique_texts: cache_key = (text, target_lang, source_lang) if cache_key in _TRANSLATION_CACHE: result[text] = _TRANSLATION_CACHE[cache_key] pending = [text for text in unique_texts if text not in result] for start in range(0, len(pending), chunk_size): chunk = pending[start:start + chunk_size] joined = "\n".join(chunk) query_parts = [("client", "gtx"), ("sl", source_lang), ("tl", target_lang), ("dt", "t"), ("q", joined)] url = "https://translate.googleapis.com/translate_a/single?" + urlencode(query_parts) req = Request(url, headers={"User-Agent": "Mozilla/5.0", "Connection": "close"}) with urlopen(req, timeout=60) as response: data = json.loads(response.read().decode("utf-8")) translated_lines = "".join(part[0] for part in data[0] if part and part[0]).splitlines() for original, translated in zip(chunk, translated_lines): result[original] = translated _TRANSLATION_CACHE[(original, target_lang, source_lang)] = translated time.sleep(0.3) return result def normalize_name(value: str) -> str: value = re.sub(r"\s+", " ", (value or "").strip()) return value 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 choose_profile(text: str) -> dict: lowered = text.lower() for profile_name, keywords in KEYWORDS.items(): if any(keyword in lowered for keyword in keywords): return CARE_PROFILES[profile_name] return CARE_PROFILES["foliage"] def collect_existing_names() -> set[str]: conn = create_db_connection() cursor = conn.cursor() try: cursor.execute("SELECT type_name FROM plant_reminder_presets") return {row[0] for row in cursor.fetchall()} finally: cursor.close() conn.close() def fetch_taxa_page(page: int, per_page: int = 200) -> list[dict]: url = ( "https://api.inaturalist.org/v1/taxa" f"?taxon_id={PLANTAE_TAXON_ID}&rank=species&photos=true&per_page={per_page}" f"&page={page}&order=desc&order_by=observations_count" ) data = fetch_json(url) return data.get("results", []) def load_name_overrides() -> dict[str, dict[str, str]]: if not OVERRIDE_PATH.exists(): return {} return json.loads(OVERRIDE_PATH.read_text(encoding="utf-8")) def fetch_wikidata_labels(scientific_names: list[str], chunk_size: int = 80) -> dict[str, dict[str, str]]: results: dict[str, dict[str, str]] = {} unique_names = [] seen = set() for name in scientific_names: 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 = row["scientificName"]["value"] candidate = { "ko": row.get("labelKo", {}).get("value", "").strip(), "en": row.get("labelEn", {}).get("value", "").strip(), "ja": row.get("labelJa", {}).get("value", "").strip(), "zh": row.get("labelZh", {}).get("value", "").strip(), } results[scientific_name] = pick_better_label_entry(results.get(scientific_name), candidate) time.sleep(0.2) return results def build_nibr_opener(): opener = build_opener(HTTPCookieProcessor()) opener.addheaders = list(NIBR_HEADERS.items()) opener.open(Request(f"{NIBR_BASE_URL}/index.do", headers=NIBR_HEADERS), timeout=30).read() return opener def is_exact_species_match(query: str, taxon_full_name: str) -> bool: query_tokens = normalize_name(query).split() full_tokens = normalize_name(taxon_full_name).split() if len(query_tokens) < 2 or len(full_tokens) < 2: return False if [token.lower() for token in full_tokens[:2]] != [token.lower() for token in query_tokens[:2]]: return False if len(full_tokens) == 2: return True third_token = full_tokens[2] third = third_token.lower() if third in INFRASPECIFIC_MARKERS: return False return not third_token[:1].islower() def fetch_nibr_korean_names(scientific_names: list[str]) -> dict[str, str]: opener = build_nibr_opener() names: dict[str, str] = {} unique_names = [] seen = set() for name in scientific_names: normalized = normalize_name(name) if normalized and normalized not in seen: unique_names.append(normalized) seen.add(normalized) for index, scientific_name in enumerate(unique_names, start=1): payload = urlencode( { "firstIndex": 0, "lastIndex": 1, "recordCountPerPage": 9, "searchField": scientific_name, } ).encode("utf-8") request = Request( f"{NIBR_BASE_URL}/speciesSearch/searchEngineTotalAjax.do", data=payload, headers={**NIBR_HEADERS, "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}, ) try: with opener.open(request, timeout=30) as response: data = json.loads(response.read().decode("utf-8")) except (HTTPError, URLError, TimeoutError): time.sleep(0.3) continue result_list = json.loads(data.get("resultList") or "[]") for item in result_list: taxon_knm = normalize_name(item.get("TAXON_KNM") or "") if not taxon_knm: continue if is_exact_species_match(scientific_name, item.get("TAXON_FULL_NM", "")): names[scientific_name] = taxon_knm break if index % 100 == 0: print(f"[nibr] {index}/{len(unique_names)} matched={len(names)}", flush=True) time.sleep(0.12) return names def build_raw_candidate(row: dict) -> dict | None: scientific_name = normalize_name(row.get("name") or "") common_name_en = normalize_name(row.get("preferred_common_name") or "") english_name = common_name_en or scientific_name if not scientific_name or not english_name: return None photo = row.get("default_photo") or {} source_image_url = photo.get("large_url") or photo.get("medium_url") or photo.get("square_url") if not source_image_url: return None return { "english_name": english_name, "scientific_name": scientific_name, "source_image_url": source_image_url, "source_taxon_id": row.get("id"), } def main() -> None: args = parse_args() load_env(args.env_file) name_overrides = load_name_overrides() existing_names = collect_existing_names() raw_candidates: list[dict] = [] raw_seen = set() for page in range(1, args.max_pages + 1): rows = fetch_taxa_page(page) if not rows: break for row in rows: candidate = build_raw_candidate(row) if not candidate: continue source_key = (candidate["english_name"], candidate["scientific_name"]) if source_key in raw_seen: continue raw_seen.add(source_key) raw_candidates.append(candidate) if len(raw_candidates) >= args.target * 2: break print(f"[page] {page} raw_candidates={len(raw_candidates)}", flush=True) if len(raw_candidates) >= args.target * 2: break scientific_names = [candidate["scientific_name"] for candidate in raw_candidates] wikidata_labels = fetch_wikidata_labels(scientific_names) nibr_korean_names = {} if args.skip_nibr else fetch_nibr_korean_names(scientific_names) profile_texts = sorted({profile["sunlight_ko"] for profile in CARE_PROFILES.values()}) tip_texts = sorted({profile["tip_ko"] for profile in CARE_PROFILES.values()}) sunlight_en_map = batch_translate_texts(profile_texts, "en", source_lang="ko") sunlight_ja_map = batch_translate_texts(profile_texts, "ja", source_lang="ko") sunlight_zh_map = batch_translate_texts(profile_texts, "zh-CN", source_lang="ko") tip_en_map = batch_translate_texts(tip_texts, "en", source_lang="ko") tip_ja_map = batch_translate_texts(tip_texts, "ja", source_lang="ko") tip_zh_map = batch_translate_texts(tip_texts, "zh-CN", source_lang="ko") accepted: list[dict] = [] candidate_names = set(existing_names) for candidate in raw_candidates: english_name = candidate["english_name"] scientific_name = candidate["scientific_name"] labels = wikidata_labels.get(scientific_name, {}) override = name_overrides.get(scientific_name, {}) ko_name = normalize_name(override.get("ko") or nibr_korean_names.get(scientific_name) or labels.get("ko") or scientific_name) if not ko_name or ko_name in candidate_names: continue profile = choose_profile(f"{english_name} {scientific_name}") candidate_names.add(ko_name) accepted.append({ "type_name": ko_name, "watering_cycle_days": profile["days"], "translations": { "ko": {"type_name": ko_name, "sunlight": profile["sunlight_ko"], "tip": profile["tip_ko"]}, "en": {"type_name": normalize_name(override.get("en") or labels.get("en") or scientific_name), "sunlight": sunlight_en_map[profile["sunlight_ko"]], "tip": tip_en_map[profile["tip_ko"]]}, "ja": {"type_name": normalize_name(override.get("ja") or labels.get("ja") or scientific_name), "sunlight": sunlight_ja_map[profile["sunlight_ko"]], "tip": tip_ja_map[profile["tip_ko"]]}, "zh": {"type_name": normalize_name(override.get("zh") or labels.get("zh") or scientific_name), "sunlight": sunlight_zh_map[profile["sunlight_ko"]], "tip": tip_zh_map[profile["tip_ko"]]}, }, "source_image_url": candidate["source_image_url"], "source_scientific_name": scientific_name, "source_taxon_id": candidate["source_taxon_id"], "is_active": True, }) if len(accepted) >= args.target: break out_path = Path(args.out) out_path.parent.mkdir(parents=True, exist_ok=True) with out_path.open("w", encoding="utf-8") as fp: for record in accepted: fp.write(json.dumps(record, ensure_ascii=False) + "\n") print(json.dumps({ "target": args.target, "generated": len(accepted), "output": str(out_path), }, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()