jardin/calendrier_lunaire/saints_dictons/import_saints_dictons_db.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path


def _ensure_schema(conn: sqlite3.Connection) -> None:
    conn.execute(
        """
        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,
            source_url TEXT,
            updated_at TEXT NOT NULL,
            UNIQUE(mois, jour)
        )
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS dicton (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            mois INTEGER NOT NULL,
            jour INTEGER,
            texte TEXT NOT NULL,
            region TEXT
        )
        """
    )
    conn.execute("CREATE INDEX IF NOT EXISTS idx_dicton_mois_jour ON dicton(mois, jour)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_saint_du_jour_mois_jour ON saint_du_jour(mois, jour)")


def _load_json_rows(path: Path, required_key: str) -> list[dict]:
    raw = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(raw, list):
        raise ValueError(f"{path}: format JSON attendu = liste d'objets")
    rows: list[dict] = []
    for item in raw:
        if not isinstance(item, dict):
            continue
        if required_key not in item:
            continue
        rows.append(item)
    return rows


def _import_saints(conn: sqlite3.Connection, saints_rows: list[dict], mode: str) -> int:
    now = datetime.now(timezone.utc).isoformat()
    inserted = 0

    if mode == "replace":
        conn.execute("DELETE FROM saint_du_jour")

    for row in saints_rows:
        mois = int(row["mois"])
        jour = int(row["jour"])
        saints = row.get("saints") or []
        if not isinstance(saints, list):
            saints = [str(saints)]
        saints = [str(x).strip() for x in saints if str(x).strip()]
        if not saints:
            continue
        saints_json = json.dumps(saints, ensure_ascii=False)
        source_url = row.get("source_url")

        conn.execute(
            """
            INSERT INTO saint_du_jour (mois, jour, saints_json, source_url, updated_at)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(mois, jour)
            DO UPDATE SET
                saints_json = excluded.saints_json,
                source_url = excluded.source_url,
                updated_at = excluded.updated_at
            """,
            (mois, jour, saints_json, source_url, now),
        )
        inserted += 1
    return inserted


def _import_dictons(conn: sqlite3.Connection, dicton_rows: list[dict], mode: str, region: str | None) -> int:
    inserted = 0

    if mode == "replace":
        if region:
            conn.execute("DELETE FROM dicton WHERE region = ?", (region,))
        else:
            conn.execute("DELETE FROM dicton")

    for row in dicton_rows:
        mois = int(row["mois"])
        jour = int(row["jour"])
        dictons = row.get("dictons") or []
        if not isinstance(dictons, list):
            dictons = [str(dictons)]
        dictons = [str(x).strip() for x in dictons if str(x).strip()]
        if not dictons:
            continue

        for texte in dictons:
            if mode == "append":
                exists = conn.execute(
                    """
                    SELECT 1
                    FROM dicton
                    WHERE mois = ? AND jour = ? AND texte = ? AND COALESCE(region, '') = COALESCE(?, '')
                    LIMIT 1
                    """,
                    (mois, jour, texte, region),
                ).fetchone()
                if exists:
                    continue
            conn.execute(
                "INSERT INTO dicton (mois, jour, texte, region) VALUES (?, ?, ?, ?)",
                (mois, jour, texte, region),
            )
            inserted += 1
    return inserted


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Importe saints du jour + dictons dans SQLite (hors webapp)."
    )
    parser.add_argument(
        "--db",
        default="data/jardin.db",
        help="Chemin SQLite cible",
    )
    parser.add_argument(
        "--saints-json",
        default="calendrier_lunaire/saints_dictons/saints_du_jour.json",
        help="JSON saints_du_jour",
    )
    parser.add_argument(
        "--dictons-json",
        default="calendrier_lunaire/saints_dictons/dictons_du_jour.json",
        help="JSON dictons_du_jour",
    )
    parser.add_argument(
        "--mode",
        choices=["replace", "append"],
        default="replace",
        help="replace: purge puis recharge ; append: ajoute sans doublons",
    )
    parser.add_argument(
        "--region",
        default="National",
        help="Région stockée dans table dicton (vide pour NULL)",
    )
    args = parser.parse_args()

    db_path = Path(args.db)
    saints_path = Path(args.saints_json)
    dictons_path = Path(args.dictons_json)
    region = args.region.strip() or None

    saints_rows = _load_json_rows(saints_path, "saints")
    dicton_rows = _load_json_rows(dictons_path, "dictons")

    db_path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(db_path)
    try:
        _ensure_schema(conn)
        conn.execute("BEGIN")
        saints_count = _import_saints(conn, saints_rows, args.mode)
        dictons_count = _import_dictons(conn, dicton_rows, args.mode, region)
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

    print(f"Import terminé ({args.mode})")
    print(f"  Saints importés: {saints_count}")
    print(f"  Dictons importés: {dictons_count} (region={region!r})")
    print(f"  Base: {db_path}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())