feat(plantes): script import graines + arbustre (JSON → plant_variety)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
275
backend/scripts/import_graines.py
Normal file
275
backend/scripts/import_graines.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user