301 lines
10 KiB
Python
301 lines
10 KiB
Python
#!/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 unicodedata
|
|
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
|
|
# Vérifier si cette image existe déjà dans media pour cette variété
|
|
existing_m = conn.execute(
|
|
"SELECT id FROM media WHERE entity_type = 'plant_variety' AND entity_id = ? AND url LIKE ?",
|
|
(variety_id, f"%{src.stem}%")
|
|
).fetchone()
|
|
if existing_m:
|
|
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."""
|
|
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),
|
|
)
|
|
|
|
# Vérifier si cette variété existe déjà pour cette plante
|
|
existing_v = conn.execute(
|
|
"SELECT id FROM plant_variety WHERE plant_id = ? AND LOWER(variete) = LOWER(?)",
|
|
(plant_id, variete_name)
|
|
).fetchone()
|
|
if existing_v:
|
|
print(f" ↷ {nom_commun} — {variete_name} (déjà importé)")
|
|
continue
|
|
|
|
# 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))
|
|
|
|
# Vérifier si cette variété existe déjà pour cette plante
|
|
existing_v = conn.execute(
|
|
"SELECT id FROM plant_variety WHERE plant_id = ? AND LOWER(variete) = LOWER(?)",
|
|
(plant_id, variete_name)
|
|
).fetchone()
|
|
if existing_v:
|
|
print(f" ↷ {nom_commun} — {variete_name} (déjà importé)")
|
|
continue
|
|
|
|
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()
|