jardin/calendrier_lunaire/saints_dictons/import_webapp_db.py

#!/usr/bin/env python3
"""
Import des saints et dictons dans la base de données de la webapp Jardin.

Les données sont INDÉPENDANTES DE L'ANNÉE : stockées par (mois, jour) uniquement.

Sources JSON :
  - saints_du_jour.json    → table saint_du_jour
  - dictons_du_jour.json   → table dicton

Usage :
  # Import complet (remplace les données existantes)
  python import_webapp_db.py

  # Import vers une autre DB
  python import_webapp_db.py --db /chemin/vers/jardin.db

  # Mode append (ajoute sans supprimer les existants)
  python import_webapp_db.py --mode append

  # Seulement les saints ou seulement les dictons
  python import_webapp_db.py --only saints
  python import_webapp_db.py --only dictons

  # Tagger les dictons avec une région
  python import_webapp_db.py --region "Auvergne"

  # Prévisualisation sans modification
  python import_webapp_db.py --dry-run
"""
import argparse
import json
import sqlite3
import sys
from datetime import datetime, timezone
from pathlib import Path

# Chemins par défaut relatifs à ce script
_HERE = Path(__file__).parent
_REPO_ROOT = _HERE.parent.parent  # racine du projet jardin/
DEFAULT_DB = _REPO_ROOT / "data" / "jardin.db"
DEFAULT_SAINTS_JSON = _HERE / "saints_du_jour.json"
DEFAULT_DICTONS_JSON = _HERE / "dictons_du_jour.json"


# ─────────────────────────────────────────────
# Schémas SQL attendus par la webapp
# ─────────────────────────────────────────────

SQL_CREATE_SAINT_DU_JOUR = """
CREATE TABLE IF NOT EXISTS saint_du_jour (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    mois        INTEGER NOT NULL,
    jour        INTEGER NOT NULL,
    saints_json TEXT    NOT NULL DEFAULT '[]',
    source_url  TEXT,
    UNIQUE (mois, jour)
)
"""

SQL_CREATE_INDEX_SAINT = """
CREATE INDEX IF NOT EXISTS idx_saint_du_jour_mois_jour
    ON saint_du_jour (mois, jour)
"""

# La table dicton existe déjà dans la webapp (SQLModel).
# Colonnes: id, mois, jour, texte, region
# On s'assure juste qu'elle existe (ne recrée pas si déjà présente).
SQL_CREATE_DICTON = """
CREATE TABLE IF NOT EXISTS dicton (
    id     INTEGER PRIMARY KEY AUTOINCREMENT,
    mois   INTEGER NOT NULL,
    jour   INTEGER,
    texte  TEXT    NOT NULL,
    region TEXT
)
"""

SQL_CREATE_INDEX_DICTON = """
CREATE INDEX IF NOT EXISTS idx_dicton_mois_jour ON dicton (mois, jour)
"""


# ─────────────────────────────────────────────
# Chargement JSON
# ─────────────────────────────────────────────

def load_json(path: Path, label: str) -> list:
    if not path.exists():
        print(f"[ERREUR] Fichier introuvable : {path}", file=sys.stderr)
        sys.exit(1)
    with path.open(encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, list):
        print(f"[ERREUR] {label} : attendu une liste JSON, obtenu {type(data).__name__}", file=sys.stderr)
        sys.exit(1)
    print(f"  {label} : {len(data)} entrées chargées depuis {path.name}")
    return data


# ─────────────────────────────────────────────
# Import saints
# ─────────────────────────────────────────────

def import_saints(conn: sqlite3.Connection, data: list, mode: str, dry_run: bool) -> int:
    """Importe les saints dans saint_du_jour. Retourne le nombre d'entrées traitées."""
    conn.execute(SQL_CREATE_SAINT_DU_JOUR)
    conn.execute(SQL_CREATE_INDEX_SAINT)

    if mode == "replace" and not dry_run:
        conn.execute("DELETE FROM saint_du_jour")
        print("  [replace] saint_du_jour vidée")

    count = 0
    skipped = 0

    for entry in data:
        mois = entry.get("mois")
        jour = entry.get("jour")
        saints = entry.get("saints", [])
        source_url = entry.get("source_url")

        # Validation
        if not isinstance(mois, int) or not isinstance(jour, int):
            continue
        if not (1 <= mois <= 12 and 1 <= jour <= 31):
            continue

        # Normaliser en liste de strings non vides
        if isinstance(saints, str):
            saints = [saints]
        saints = [s.strip() for s in saints if isinstance(s, str) and s.strip()]
        saints_json = json.dumps(saints, ensure_ascii=False)

        if dry_run:
            count += 1
            continue

        if mode == "append":
            existing = conn.execute(
                "SELECT id FROM saint_du_jour WHERE mois=? AND jour=?", (mois, jour)
            ).fetchone()
            if existing:
                skipped += 1
                continue

        conn.execute(
            "INSERT OR REPLACE INTO saint_du_jour (mois, jour, saints_json, source_url) VALUES (?,?,?,?)",
            (mois, jour, saints_json, source_url),
        )
        count += 1

    if skipped:
        print(f"  saints ignorés (mode append, déjà présents) : {skipped}")
    return count


# ─────────────────────────────────────────────
# Import dictons
# ─────────────────────────────────────────────

def import_dictons(conn: sqlite3.Connection, data: list, mode: str, region: str | None, dry_run: bool) -> int:
    """Importe les dictons dans la table dicton. Retourne le nombre d'entrées insérées."""
    conn.execute(SQL_CREATE_DICTON)
    conn.execute(SQL_CREATE_INDEX_DICTON)

    if mode == "replace" and not dry_run:
        conn.execute("DELETE FROM dicton")
        print("  [replace] dicton vidée")

    count = 0

    for entry in data:
        mois = entry.get("mois")
        jour = entry.get("jour")  # peut être None
        dictons = entry.get("dictons", [])

        # Validation
        if not isinstance(mois, int):
            continue
        if not (1 <= mois <= 12):
            continue
        if jour is not None and not (1 <= jour <= 31):
            continue

        # Normaliser
        if isinstance(dictons, str):
            dictons = [dictons]
        dictons = [d.strip() for d in dictons if isinstance(d, str) and d.strip()]

        for texte in dictons:
            if dry_run:
                count += 1
                continue

            if mode == "append":
                existing = conn.execute(
                    "SELECT id FROM dicton WHERE mois=? AND jour IS ? AND texte=?",
                    (mois, jour, texte),
                ).fetchone()
                if existing:
                    continue

            conn.execute(
                "INSERT INTO dicton (mois, jour, texte, region) VALUES (?,?,?,?)",
                (mois, jour, texte, region),
            )
            count += 1

    return count


# ─────────────────────────────────────────────
# Point d'entrée
# ─────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(
        description="Import saints et dictons dans la BDD webapp Jardin (indépendant de l'année)"
    )
    parser.add_argument("--db", default=str(DEFAULT_DB),
                        help=f"Chemin vers jardin.db (défaut: {DEFAULT_DB})")
    parser.add_argument("--saints-json", default=str(DEFAULT_SAINTS_JSON),
                        help=f"Fichier saints_du_jour.json (défaut: {DEFAULT_SAINTS_JSON.name})")
    parser.add_argument("--dictons-json", default=str(DEFAULT_DICTONS_JSON),
                        help=f"Fichier dictons_du_jour.json (défaut: {DEFAULT_DICTONS_JSON.name})")
    parser.add_argument("--mode", choices=["replace", "append"], default="replace",
                        help="replace = purge + recharge (défaut) | append = ajoute sans doublons")
    parser.add_argument("--only", choices=["saints", "dictons", "both"], default="both",
                        help="Importer uniquement saints, dictons, ou les deux (défaut: both)")
    parser.add_argument("--region", default=None,
                        help="Tag région pour les dictons (ex: 'Auvergne', 'National'). Vide = NULL")
    parser.add_argument("--dry-run", action="store_true",
                        help="Simulation : affiche les comptages sans modifier la BDD")
    args = parser.parse_args()

    db_path = Path(args.db)
    if not db_path.exists():
        print(f"[ERREUR] Base de données introuvable : {db_path}", file=sys.stderr)
        print("  → Vérifiez que la webapp a démarré au moins une fois pour créer la BDD.")
        sys.exit(1)

    region = args.region if args.region else None
    do_saints = args.only in ("saints", "both")
    do_dictons = args.only in ("dictons", "both")

    print(f"\n=== Import saints/dictons → {db_path} ===")
    print(f"  Mode  : {args.mode}")
    print(f"  Scope : {args.only}")
    if region:
        print(f"  Région: {region}")
    if args.dry_run:
        print("  *** DRY-RUN — aucune modification en BDD ***")
    print()

    # Charger les JSON nécessaires
    saints_data = load_json(Path(args.saints_json), "saints_du_jour") if do_saints else []
    dictons_data = load_json(Path(args.dictons_json), "dictons_du_jour") if do_dictons else []
    print()

    # Connexion SQLite
    conn = sqlite3.connect(str(db_path))
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA foreign_keys=OFF")  # pas de cascade gestion manuelle

    saints_count = 0
    dictons_count = 0

    try:
        conn.execute("BEGIN")

        if do_saints:
            saints_count = import_saints(conn, saints_data, args.mode, args.dry_run)
        if do_dictons:
            dictons_count = import_dictons(conn, dictons_data, args.mode, region, args.dry_run)

        if not args.dry_run:
            conn.commit()
        else:
            conn.rollback()

    except Exception as exc:
        conn.rollback()
        print(f"\n[ERREUR] Transaction annulée : {exc}", file=sys.stderr)
        raise
    finally:
        conn.close()

    # Résumé
    print()
    print("=== Résultat ===")
    if do_saints:
        prefix = "[DRY-RUN] " if args.dry_run else ""
        print(f"  {prefix}Saints importés  : {saints_count} jours → table saint_du_jour")
    if do_dictons:
        prefix = "[DRY-RUN] " if args.dry_run else ""
        print(f"  {prefix}Dictons importés : {dictons_count} entrées → table dicton")
    print(f"  Base             : {db_path}")
    if args.dry_run:
        print("\n  → Relancez sans --dry-run pour appliquer les changements.")
    else:
        ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
        print(f"\n  Import terminé le {ts}")


if __name__ == "__main__":
    main()