""" Plant reminder preset bulk seeder for very large catalogs. What it does - Reads NDJSON line by line, so very large files are supported. - Inserts plant presets into `plant_reminder_presets`. - Skips duplicates by `type_name` before insert. - Uploads source images to NCP Object Storage and stores CDN URLs. - Fills 4-language fields: ko / en / ja / zh. - Can also backfill the multi-language columns if they are missing. Recommended usage 1) Prepare an NDJSON file. One JSON object per line. 2) Run: python scripts/bulk_seed_plant_reminder_presets.py --file scripts/data/plant_reminder_catalog_specials.ndjson Optional - --limit 100000 - --start-sort-order 1000 - --skip-image-upload - --dry-run Expected NDJSON shape { "type_name": "피쉬본 선인장", "watering_cycle_days": 10, "translations": { "ko": {"type_name": "피쉬본 선인장", "sunlight": "밝은 간접광", "tip": "겉흙이 마르면 물을 주세요."}, "en": {"type_name": "Fishbone Cactus", "sunlight": "Bright indirect light", "tip": "Water when the topsoil dries."}, "ja": {"type_name": "フィッシュボーンカクタス", "sunlight": "明るい間接光", "tip": "土の表面が乾いたら水を与えてください。"}, "zh": {"type_name": "鱼骨仙人掌", "sunlight": "明亮散射光", "tip": "表土变干后浇水。"} }, "source_image_url": "https://....jpg", "is_active": true } """ from __future__ import annotations import argparse import io import json import os import time import uuid from pathlib import Path from typing import Dict, Iterable, Tuple from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen import boto3 from botocore.exceptions import ClientError from botocore.config import Config import mysql.connector from PIL import Image, ImageOps ROOT_DIR = Path(__file__).resolve().parents[1] DEFAULT_ENV_PATH = ROOT_DIR / ".env" NCP_ACCESS_KEY = "ncp_iam_BPASKR5aqRFd5UVurpnn" NCP_SECRET_KEY = "ncp_iam_BPKSKR33fSB43WgsitoAeirg4QGRNecX5p" NCP_BUCKET_NAME = "babynote" NCP_ENDPOINT = "https://kr.object.ncloudstorage.com" CDN_DOMAIN = "https://uscppbpkqffl28953595.gcdn.ntruss.com" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Bulk seed plant reminder presets from NDJSON.") parser.add_argument("--file", required=True, help="Path to NDJSON catalog") parser.add_argument("--limit", type=int, default=100000, help="Max number of new rows to insert") parser.add_argument("--start-sort-order", type=int, default=None, help="Override initial sort_order") parser.add_argument("--skip-image-upload", action="store_true", help="Use source_image_url directly without NCP upload") parser.add_argument( "--fallback-on-upload-failure", action="store_true", help="If NCP upload fails, fall back to source_image_url instead of aborting", ) parser.add_argument("--dry-run", action="store_true", help="Parse and validate only, do not upload or insert") parser.add_argument("--env-file", default=str(DEFAULT_ENV_PATH), help="Path to .env file") 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(): if not line or line.strip().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 create_s3_client(): session = boto3.session.Session() # Match admin.js (aws-sdk) settings: SigV2 + path-style addressing. # This avoids subtle 403s on some S3-compatible endpoints. cfg = Config(signature_version="s3", s3={"addressing_style": "path"}) return session.client( "s3", region_name="kr-standard", endpoint_url=NCP_ENDPOINT, aws_access_key_id=NCP_ACCESS_KEY, aws_secret_access_key=NCP_SECRET_KEY, config=cfg, ) def ensure_multilang_schema(cursor) -> None: statements = [ "ALTER TABLE plant_reminder_presets ADD COLUMN type_name_ko VARCHAR(100) NULL AFTER type_name", "ALTER TABLE plant_reminder_presets ADD COLUMN type_name_en VARCHAR(100) NULL AFTER type_name_ko", "ALTER TABLE plant_reminder_presets ADD COLUMN type_name_ja VARCHAR(100) NULL AFTER type_name_en", "ALTER TABLE plant_reminder_presets ADD COLUMN type_name_zh VARCHAR(100) NULL AFTER type_name_ja", "ALTER TABLE plant_reminder_presets ADD COLUMN sunlight_ko VARCHAR(255) NULL AFTER sunlight", "ALTER TABLE plant_reminder_presets ADD COLUMN sunlight_en VARCHAR(255) NULL AFTER sunlight_ko", "ALTER TABLE plant_reminder_presets ADD COLUMN sunlight_ja VARCHAR(255) NULL AFTER sunlight_en", "ALTER TABLE plant_reminder_presets ADD COLUMN sunlight_zh VARCHAR(255) NULL AFTER sunlight_ja", "ALTER TABLE plant_reminder_presets ADD COLUMN tip_ko TEXT NULL AFTER tip", "ALTER TABLE plant_reminder_presets ADD COLUMN tip_en TEXT NULL AFTER tip_ko", "ALTER TABLE plant_reminder_presets ADD COLUMN tip_ja TEXT NULL AFTER tip_en", "ALTER TABLE plant_reminder_presets ADD COLUMN tip_zh TEXT NULL AFTER tip_ja", "ALTER TABLE plant_reminder_presets ADD UNIQUE KEY uniq_plant_reminder_presets_type_name (type_name)", ] for sql in statements: try: cursor.execute(sql) print(f"[altered] {sql}") except mysql.connector.Error as exc: if exc.errno not in {1060, 1061}: raise cursor.execute( """ UPDATE plant_reminder_presets SET type_name_ko = COALESCE(NULLIF(type_name_ko, ''), type_name), type_name_en = COALESCE(NULLIF(type_name_en, ''), type_name), type_name_ja = COALESCE(NULLIF(type_name_ja, ''), type_name), type_name_zh = COALESCE(NULLIF(type_name_zh, ''), type_name), sunlight_ko = COALESCE(NULLIF(sunlight_ko, ''), sunlight, ''), sunlight_en = COALESCE(NULLIF(sunlight_en, ''), sunlight, ''), sunlight_ja = COALESCE(NULLIF(sunlight_ja, ''), sunlight, ''), sunlight_zh = COALESCE(NULLIF(sunlight_zh, ''), sunlight, ''), tip_ko = COALESCE(NULLIF(tip_ko, ''), tip, ''), tip_en = COALESCE(NULLIF(tip_en, ''), tip, ''), tip_ja = COALESCE(NULLIF(tip_ja, ''), tip, ''), tip_zh = COALESCE(NULLIF(tip_zh, ''), tip, '') """ ) def get_max_sort_order(cursor) -> int: cursor.execute("SELECT COALESCE(MAX(sort_order), 0) FROM plant_reminder_presets") row = cursor.fetchone() return int(row[0] or 0) def preset_exists(cursor, type_name: str) -> bool: cursor.execute("SELECT id FROM plant_reminder_presets WHERE type_name = %s LIMIT 1", (type_name,)) return cursor.fetchone() is not None def iter_ndjson(path: str) -> Iterable[Dict]: with open(path, "r", encoding="utf-8") as fp: for line_number, line in enumerate(fp, start=1): line = line.strip() if not line: continue try: yield json.loads(line) except json.JSONDecodeError as exc: raise ValueError(f"NDJSON parse error at line {line_number}: {exc}") from exc def normalize_record(record: Dict, sort_order: int) -> Dict: translations = record.get("translations") or {} ko = translations.get("ko") or {} en = translations.get("en") or {} ja = translations.get("ja") or {} zh = translations.get("zh") or {} type_name_ko = (ko.get("type_name") or record.get("type_name") or "").strip() if not type_name_ko: raise ValueError("type_name is required") sunlight_ko = (ko.get("sunlight") or record.get("sunlight") or "").strip() tip_ko = (ko.get("tip") or record.get("tip") or "").strip() return { "type_name": type_name_ko, "type_name_ko": type_name_ko, "type_name_en": (en.get("type_name") or record.get("type_name_en") or type_name_ko).strip(), "type_name_ja": (ja.get("type_name") or record.get("type_name_ja") or type_name_ko).strip(), "type_name_zh": (zh.get("type_name") or record.get("type_name_zh") or type_name_ko).strip(), "watering_cycle_days": int(record.get("watering_cycle_days") or 7), "sunlight": sunlight_ko, "sunlight_ko": sunlight_ko, "sunlight_en": (en.get("sunlight") or record.get("sunlight_en") or sunlight_ko).strip(), "sunlight_ja": (ja.get("sunlight") or record.get("sunlight_ja") or sunlight_ko).strip(), "sunlight_zh": (zh.get("sunlight") or record.get("sunlight_zh") or sunlight_ko).strip(), "tip": tip_ko, "tip_ko": tip_ko, "tip_en": (en.get("tip") or record.get("tip_en") or tip_ko).strip(), "tip_ja": (ja.get("tip") or record.get("tip_ja") or tip_ko).strip(), "tip_zh": (zh.get("tip") or record.get("tip_zh") or tip_ko).strip(), "source_image_url": (record.get("source_image_url") or "").strip(), "image_url": (record.get("image_url") or "").strip(), "image_path": (record.get("image_path") or "").strip(), "is_active": bool(record.get("is_active", True)), "sort_order": sort_order, } def insert_preset(cursor, preset: Dict) -> None: cursor.execute( """ INSERT INTO plant_reminder_presets ( type_name, type_name_ko, type_name_en, type_name_ja, type_name_zh, watering_cycle_days, sunlight, sunlight_ko, sunlight_en, sunlight_ja, sunlight_zh, tip, tip_ko, tip_en, tip_ja, tip_zh, image_url, image_path, is_active, sort_order ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ) """, ( preset["type_name"], preset["type_name_ko"], preset["type_name_en"], preset["type_name_ja"], preset["type_name_zh"], preset["watering_cycle_days"], preset["sunlight"], preset["sunlight_ko"], preset["sunlight_en"], preset["sunlight_ja"], preset["sunlight_zh"], preset["tip"], preset["tip_ko"], preset["tip_en"], preset["tip_ja"], preset["tip_zh"], preset["image_url"], preset["image_path"], 1 if preset["is_active"] else 0, preset["sort_order"], ), ) def upload_source_image_to_ncp(s3_client, source_image_url: str) -> Tuple[str, str]: image_bytes = fetch_bytes(source_image_url) optimized = optimize_to_webp(image_bytes) image_path = f"plant-reminder/presets/{uuid.uuid4()}.webp" s3_client.put_object( Bucket=NCP_BUCKET_NAME, Key=image_path, Body=optimized, ContentType="image/webp", ACL="private", ) return f"{CDN_DOMAIN}/{image_path}", image_path def fetch_bytes(url: str, attempt: int = 1) -> bytes: req = Request( url, headers={ "User-Agent": "Codex Plant Python Seeder/1.0", "Connection": "close", }, ) try: with urlopen(req, timeout=30) as response: return response.read() except HTTPError as exc: if attempt >= 5 or exc.code not in {429, 500, 502, 503, 504}: raise retry_after = int(exc.headers.get("Retry-After", "0") or "0") wait_seconds = max(retry_after, attempt * 2) print(f"[retry] {url} -> {exc.code}, waiting {wait_seconds}s") time.sleep(wait_seconds) return fetch_bytes(url, attempt + 1) except URLError: if attempt >= 5: raise wait_seconds = attempt * 2 print(f"[retry] {url} -> network error, waiting {wait_seconds}s") time.sleep(wait_seconds) return fetch_bytes(url, attempt + 1) def optimize_to_webp(image_bytes: bytes) -> bytes: image = Image.open(io.BytesIO(image_bytes)) image = ImageOps.exif_transpose(image).convert("RGB") image.thumbnail((512, 512)) canvas = Image.new("RGB", (512, 512), (255, 255, 255)) left = (512 - image.width) // 2 top = (512 - image.height) // 2 canvas.paste(image, (left, top)) output = io.BytesIO() canvas.save(output, format="WEBP", quality=88, method=6) return output.getvalue() def main() -> None: args = parse_args() load_env(args.env_file) file_path = Path(args.file) if not file_path.exists(): raise FileNotFoundError(f"Catalog not found: {file_path}") conn = create_db_connection() cursor = conn.cursor() s3_client = create_s3_client() try: ensure_multilang_schema(cursor) conn.commit() sort_order = args.start_sort_order or (get_max_sort_order(cursor) + 1) inserted = 0 skipped = 0 for record in iter_ndjson(str(file_path)): if inserted >= args.limit: break preset = normalize_record(record, sort_order) if preset_exists(cursor, preset["type_name"]): skipped += 1 if skipped % 100 == 0: print(f"[skip-duplicate] skipped={skipped} latest={preset['type_name']}") continue if not preset["image_url"] or not preset["image_path"]: if args.skip_image_upload: preset["image_url"] = preset["source_image_url"] preset["image_path"] = "" else: if not preset["source_image_url"]: print(f"[skip-no-image] {preset['type_name']}") skipped += 1 continue try: preset["image_url"], preset["image_path"] = upload_source_image_to_ncp( s3_client, preset["source_image_url"] ) except ClientError as exc: if args.fallback_on_upload_failure: code = ( ((exc.response or {}).get("Error") or {}).get("Code") if hasattr(exc, "response") else None ) print( f"[upload-failed:fallback] {preset['type_name']} code={code} url={preset['source_image_url']}" ) preset["image_url"] = preset["source_image_url"] preset["image_path"] = "" else: raise if args.dry_run: print(f"[dry-run] {preset['type_name']}") else: insert_preset(cursor, preset) conn.commit() inserted += 1 sort_order += 1 if inserted % 50 == 0: print(f"[progress] inserted={inserted} skipped={skipped} next_sort_order={sort_order}") print(json.dumps({ "file": str(file_path), "inserted": inserted, "skipped": skipped, "next_sort_order": sort_order, }, ensure_ascii=False, indent=2)) finally: cursor.close() conn.close() if __name__ == "__main__": main()