feat(saints-dictons): table saint_du_jour + API + import standalone 366j
- Nouveau modèle SaintDuJour (mois+jour+saints_json, indépendant de l'année) - Router /api/saints et /api/saints/jour (mois+jour → liste de prénoms) - Script standalone import_webapp_db.py : saints_du_jour.json → saint_du_jour, dictons_du_jour.json → dicton ; modes replace/append, --dry-run, --region - Données JSON 366 jours : saints_du_jour.json + dictons_du_jour.json - Scripts scraping/export calendrier_lunaire/saints_dictons/ Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
307
calendrier_lunaire/saints_dictons/import_webapp_db.py
Normal file
307
calendrier_lunaire/saints_dictons/import_webapp_db.py
Normal file
@@ -0,0 +1,307 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user