From 0f5ebd25be421b9aa76c66ee4109ea8e34b71c88 Mon Sep 17 00:00:00 2001 From: gilles Date: Sun, 8 Mar 2026 19:04:56 +0100 Subject: [PATCH] =?UTF-8?q?feat(plantes):=20script=20import=20graines=20+?= =?UTF-8?q?=20arbustre=20(JSON=20=E2=86=92=20plant=5Fvariety)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- backend/scripts/import_graines.py | 275 ++++++++++++++++++++++++++++++ 1 file changed, 275 insertions(+) create mode 100644 backend/scripts/import_graines.py diff --git a/backend/scripts/import_graines.py b/backend/scripts/import_graines.py new file mode 100644 index 0000000..2589343 --- /dev/null +++ b/backend/scripts/import_graines.py @@ -0,0 +1,275 @@ +#!/usr/bin/env python3 +""" +Import one-shot : docs/graine/caracteristiques_plantation.json + docs/arbustre/caracteristiques_arbustre.json +Usage: cd /chemin/projet && python3 backend/scripts/import_graines.py +""" +import json +import shutil +import sqlite3 +import uuid +from datetime import datetime, timezone +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent.parent +DB_PATH = ROOT / "data" / "jardin.db" +UPLOADS_DIR = ROOT / "data" / "uploads" +GRAINE_DIR = ROOT / "docs" / "graine" +ARBUSTRE_DIR = ROOT / "docs" / "arbustre" + +ROMAN = {"I": 1, "II": 2, "III": 3, "IV": 4, "V": 5, "VI": 6, + "VII": 7, "VIII": 8, "IX": 9, "X": 10, "XI": 11, "XII": 12} + +# Mapping : mot-clé lowercase → nom_commun BDD +NOM_MAP = [ + ("oignon", "Oignon"), + ("laitue pommee grosse", "Laitue"), + ("laitue attraction", "Laitue"), + ("laitue", "Laitue"), + ("persil", "Persil"), + ("courgette", "Courgette"), + ("pois mangetout", "Pois"), + ("pois a ecosser", "Pois"), + ("pois", "Pois"), + ("tomate cornue", "Tomate"), + ("tomates moneymaker", "Tomate"), + ("tomate", "Tomate"), + ("poireau", "Poireau"), + ("echalion", "Echalote"), + ("courge", "Courge"), + ("chou pomme", "Chou"), + ("chou-fleur", "Chou-fleur"), +] + + +def roman_to_csv(s: str) -> str: + if not s: + return "" + s = s.strip() + # Handle "(selon sachet)" or other parenthetical notes + if "(" in s: + s = s.split("(")[0].strip() + parts = s.split("-") + if len(parts) == 2: + a = ROMAN.get(parts[0].strip(), 0) + b = ROMAN.get(parts[1].strip(), 0) + if a and b: + return ",".join(str(m) for m in range(a, b + 1)) + single = ROMAN.get(s, 0) + return str(single) if single else "" + + +def extract_float(s: str) -> float | None: + if not s: + return None + try: + # Handle "2-3 cm" → take first number + first = s.split()[0].split("-")[0].replace(",", ".") + return float(first) + except Exception: + return None + + +def find_or_create_plant(conn: sqlite3.Connection, nom_commun: str, categorie: str = "potager") -> int: + row = conn.execute( + "SELECT id FROM plant WHERE LOWER(nom_commun) = LOWER(?)", (nom_commun,) + ).fetchone() + if row: + return row[0] + conn.execute( + "INSERT INTO plant (nom_commun, categorie, created_at) VALUES (?, ?, ?)", + (nom_commun, categorie, datetime.now(timezone.utc).isoformat()), + ) + return conn.execute("SELECT last_insert_rowid()").fetchone()[0] + + +def copy_image(src: Path, variety_id: int, conn: sqlite3.Connection) -> None: + if not src.exists(): + print(f" WARNING image absente: {src}") + return + UPLOADS_DIR.mkdir(parents=True, exist_ok=True) + # Use UUID-based filename like the rest of the app + dest_name = f"{uuid.uuid4()}.jpg" + shutil.copy2(src, UPLOADS_DIR / dest_name) + url = f"/uploads/{dest_name}" + conn.execute(""" + INSERT INTO media (entity_type, entity_id, url, created_at) + VALUES ('plant_variety', ?, ?, ?) + """, (variety_id, url, datetime.now(timezone.utc).isoformat())) + + +def normalize(s: str) -> str: + """Normalise string: minuscules, supprime accents simples.""" + import unicodedata + return ''.join(c for c in unicodedata.normalize('NFD', s.lower()) if unicodedata.category(c) != 'Mn') + + +def resolve_nom(full_name: str) -> tuple[str, str]: + """Retourne (nom_commun, variete) depuis le nom complet du sachet.""" + norm = normalize(full_name) + for key, val in NOM_MAP: + norm_key = normalize(key) + if norm.startswith(norm_key): + variete = full_name[len(key):].strip().strip("'\"").title() + return val, variete or full_name + # Fallback : premier mot = nom_commun + parts = full_name.split() + return parts[0].title(), " ".join(parts[1:]).strip() or full_name + + +def import_graines(conn: sqlite3.Connection) -> None: + path = GRAINE_DIR / "caracteristiques_plantation.json" + if not path.exists(): + print(f"WARNING fichier absent: {path}") + return + data = json.loads(path.read_text(encoding="utf-8")) + + key = "plantes" if "plantes" in data else list(data.keys())[0] + entries = data[key] + + for entry in entries: + full_name = entry.get("plante", "") + if not full_name: + continue + nom_commun, variete_name = resolve_nom(full_name) + carac = entry.get("caracteristiques_plantation", {}) + detail = entry.get("detail", {}) + texte = detail.get("texte_integral_visible", {}) if isinstance(detail, dict) else {} + + plant_id = find_or_create_plant(conn, nom_commun) + + # Enrichir plant (ne pas écraser si déjà rempli) + updates: dict = {} + semis = roman_to_csv(carac.get("periode_semis", "")) + recolte = roman_to_csv(carac.get("periode_recolte", "")) + profondeur = extract_float(carac.get("profondeur") or "") + espacement_raw = carac.get("espacement") or "" + espacement = extract_float(espacement_raw) + + if semis: + updates["semis_exterieur_mois"] = semis + if recolte: + updates["recolte_mois"] = recolte + if profondeur: + updates["profondeur_semis_cm"] = profondeur + if espacement: + updates["espacement_cm"] = int(espacement) + if carac.get("exposition"): + updates["besoin_soleil"] = carac["exposition"] + if carac.get("temperature"): + updates["temp_germination"] = carac["temperature"] + if isinstance(texte, dict) and texte.get("arriere"): + updates["astuces_culture"] = texte["arriere"][:1000] + elif isinstance(texte, str) and texte: + updates["astuces_culture"] = texte[:1000] + + if updates: + set_clause = ", ".join(f"{k} = ?" for k in updates) + conn.execute( + f"UPDATE plant SET {set_clause} WHERE id = ?", + (*updates.values(), plant_id), + ) + + # Créer plant_variety + conn.execute( + "INSERT INTO plant_variety (plant_id, variete, created_at) VALUES (?, ?, ?)", + (plant_id, variete_name, datetime.now(timezone.utc).isoformat()), + ) + vid = conn.execute("SELECT last_insert_rowid()").fetchone()[0] + + for img in entry.get("images", []): + copy_image(GRAINE_DIR / img, vid, conn) + + print(f" OK {nom_commun} - {variete_name}") + + +def import_arbustre(conn: sqlite3.Connection) -> None: + path = ARBUSTRE_DIR / "caracteristiques_arbustre.json" + if not path.exists(): + print(f"WARNING fichier absent: {path}") + return + data = json.loads(path.read_text(encoding="utf-8")) + + key = "plantes" if "plantes" in data else list(data.keys())[0] + entries = data[key] + + for entry in entries: + full_name = entry.get("plante", "") + if not full_name: + continue + + nom_latin = entry.get("nom_latin", "") or "" + # Determine nom_commun from the plante field + if "Vitis" in full_name or "Vitis" in nom_latin: + nom_commun = "Vigne" + elif "Ribes nigrum" in full_name or "Ribes nigrum" in nom_latin: + nom_commun = "Cassissier" + elif "Rubus idaeus" in full_name or "Rubus idaeus" in nom_latin: + nom_commun = "Framboisier" + elif "'" in full_name: + nom_commun = full_name.split("'")[0].strip().title() + elif nom_latin: + parts = nom_latin.split() + nom_commun = (parts[0] + " " + parts[1]).title() if len(parts) > 1 else nom_latin.title() + else: + nom_commun = full_name.split()[0].title() + + # variete_name: content inside quotes + if "'" in full_name: + variete_name = full_name.split("'")[1].strip() + else: + variete_name = full_name + + plant_id = find_or_create_plant(conn, nom_commun, "arbuste") + + carac = entry.get("caracteristiques_plantation", {}) + arrosage = carac.get("arrosage") + exposition = carac.get("exposition") + if arrosage: + conn.execute("UPDATE plant SET besoin_eau = ? WHERE id = ?", (arrosage, plant_id)) + if exposition: + conn.execute("UPDATE plant SET besoin_soleil = ? WHERE id = ?", (exposition, plant_id)) + + conn.execute( + "INSERT INTO plant_variety (plant_id, variete, created_at) VALUES (?, ?, ?)", + (plant_id, variete_name, datetime.now(timezone.utc).isoformat()), + ) + vid = conn.execute("SELECT last_insert_rowid()").fetchone()[0] + + for img in entry.get("images", []): + copy_image(ARBUSTRE_DIR / img, vid, conn) + + print(f" OK {nom_commun} - {variete_name}") + + +def run() -> None: + if not DB_PATH.exists(): + print(f"ERREUR : base de données introuvable : {DB_PATH}") + return + + UPLOADS_DIR.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(DB_PATH) + conn.row_factory = sqlite3.Row + + try: + tables = [r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()] + if "plant_variety" not in tables: + print("WARNING Exécutez d'abord migrate_plant_varieties.py") + return + + print("=== Import graines ===") + import_graines(conn) + print("\n=== Import arbustre ===") + import_arbustre(conn) + + conn.commit() + print("\nImport terminé.") + except Exception as e: + conn.rollback() + print(f"ERREUR - rollback : {e}") + raise + finally: + conn.close() + + +if __name__ == "__main__": + run()