213 lines
7.2 KiB
Python
213 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
Extracteur des journées mondiales/internationales.
|
|
Source: https://www.journee-mondiale.com/les-journees-mondiales.htm
|
|
|
|
Fonctionnalités:
|
|
- Cache JSON pour limiter les requêtes
|
|
- Paramètres CLI (base_url, dry-run, ttl cache)
|
|
- Conversion vers format OEDB (what par défaut: culture.arts)
|
|
- Une journée d'événement, positionnée à la prochaine occurrence dans les 365 jours à venir
|
|
- Rapport succès/échecs et impression du GeoJSON en dry-run
|
|
"""
|
|
|
|
import argparse
|
|
import datetime as dt
|
|
import re
|
|
import sys
|
|
from typing import Any, Dict, List, Tuple
|
|
|
|
from bs4 import BeautifulSoup
|
|
from utils_extractor_common import (
|
|
CacheConfig,
|
|
load_cache,
|
|
save_cache,
|
|
oedb_feature,
|
|
post_oedb_features,
|
|
http_get_json,
|
|
)
|
|
|
|
|
|
DEFAULT_CACHE = "extractors_cache/world_days_cache.json"
|
|
OEDB_DEFAULT = "https://api.openeventdatabase.org"
|
|
SOURCE_URL = "https://www.journee-mondiale.com/les-journees-mondiales.htm"
|
|
|
|
|
|
def build_args() -> argparse.Namespace:
|
|
p = argparse.ArgumentParser(description="Extracteur journées mondiales/internationales -> OEDB")
|
|
p.add_argument("--base-url", help="Base URL OEDB", default=OEDB_DEFAULT)
|
|
p.add_argument("--cache", help="Fichier de cache JSON", default=DEFAULT_CACHE)
|
|
p.add_argument("--cache-ttl", help="Durée de vie du cache (sec)", type=int, default=24*3600)
|
|
p.add_argument("--limit", help="Limiter le nombre d'événements à traiter", type=int, default=None)
|
|
# dry-run activé par défaut; passer --no-dry-run pour envoyer
|
|
p.add_argument("--no-dry-run", dest="dry_run", help="Désactive le dry-run (envoie à l'API)", action="store_false")
|
|
p.set_defaults(dry_run=True)
|
|
return p.parse_args()
|
|
|
|
|
|
MONTHS = {
|
|
"janvier": 1, "février": 2, "fevrier": 2, "mars": 3, "avril": 4, "mai": 5, "juin": 6,
|
|
"juillet": 7, "août": 8, "aout": 8, "septembre": 9, "octobre": 10, "novembre": 11, "décembre": 12, "decembre": 12
|
|
}
|
|
|
|
|
|
def parse_days_from_html(html: str) -> List[Tuple[int, int, str, str]]:
|
|
"""Parse les journées depuis le HTML en ciblant les ancres dans id=texte et class=content."""
|
|
days: List[Tuple[int, int, str, str]] = []
|
|
|
|
soup = BeautifulSoup(html, 'html.parser')
|
|
|
|
# Cibler spécifiquement les ancres dans id=texte et class=content
|
|
text_section = soup.find('div', id='texte')
|
|
if not text_section:
|
|
return days
|
|
|
|
content_section = text_section.find('div', class_='content')
|
|
if not content_section:
|
|
return days
|
|
|
|
# Chercher tous les articles (mois) dans cette section
|
|
articles = content_section.find_all('article')
|
|
|
|
for article in articles:
|
|
# Chercher toutes les ancres dans chaque article
|
|
links = article.find_all('a')
|
|
|
|
for link in links:
|
|
# Extraire le texte du lien et l'URL
|
|
text = link.get_text(strip=True)
|
|
url = link.get('href', '')
|
|
if not text:
|
|
continue
|
|
|
|
# Pattern pour capturer: "1er janvier" ou "15 janvier" + "Journée mondiale..."
|
|
pattern = re.compile(r"\b(\d{1,2}|1er)\s+([a-zA-Zéèêëàâîïôöùûüç]+)\s*:\s*(.+)")
|
|
match = pattern.search(text)
|
|
|
|
if match:
|
|
day_str = match.group(1).lower()
|
|
day = 1 if day_str == "1er" else int(re.sub(r"\D", "", day_str))
|
|
month_name = match.group(2).lower()
|
|
month = MONTHS.get(month_name)
|
|
label = match.group(3).strip()
|
|
|
|
if month is not None and label:
|
|
days.append((month, day, label, url))
|
|
|
|
return days
|
|
|
|
|
|
def fetch_sources(cache_cfg: CacheConfig) -> Dict[str, Any]:
|
|
cache = load_cache(cache_cfg)
|
|
if cache:
|
|
return cache
|
|
|
|
# Récupérer la page HTML
|
|
import requests
|
|
r = requests.get(SOURCE_URL, timeout=30)
|
|
r.raise_for_status()
|
|
html = r.text
|
|
|
|
# Parser le HTML pour extraire les journées
|
|
items = parse_days_from_html(html)
|
|
|
|
out: Dict[str, Any] = {"items": items}
|
|
save_cache(cache_cfg, out)
|
|
return out
|
|
|
|
|
|
def create_event_date(month: int, day: int, today: dt.date) -> dt.date:
|
|
"""Crée la date de l'événement pour l'année courante à partir d'aujourd'hui."""
|
|
year = today.year
|
|
try:
|
|
# Essayer de créer la date pour l'année courante
|
|
event_date = dt.date(year, month, day)
|
|
# Si la date est dans le passé, utiliser l'année suivante
|
|
if event_date < today:
|
|
event_date = dt.date(year + 1, month, day)
|
|
return event_date
|
|
except ValueError:
|
|
# Gérer les cas comme le 29 février dans une année non-bissextile
|
|
# Utiliser l'année suivante
|
|
try:
|
|
return dt.date(year + 1, month, day)
|
|
except ValueError:
|
|
# Si toujours impossible, utiliser le 28 février
|
|
return dt.date(year + 1, 2, 28)
|
|
|
|
|
|
def convert_to_oedb(data: Dict[str, Any], limit: int | None = None) -> List[Dict[str, Any]]:
|
|
features: List[Dict[str, Any]] = []
|
|
today = dt.date.today()
|
|
for (month, day, label, url) in data.get("items", []) or []:
|
|
try:
|
|
date_obj = create_event_date(month, day, today)
|
|
except Exception:
|
|
continue
|
|
date_iso = date_obj.isoformat()
|
|
|
|
# Déterminer la zone selon le titre
|
|
label_lower = label.lower()
|
|
if "mondial" in label_lower or "international" in label_lower:
|
|
zone = "world"
|
|
where = "Monde"
|
|
else:
|
|
zone = "france"
|
|
where = "France"
|
|
|
|
# Créer l'événement avec propriété zone
|
|
feature = oedb_feature(
|
|
label=label,
|
|
what="culture.days",
|
|
start=f"{date_iso}T00:00:00Z",
|
|
stop=f"{date_iso}T23:59:59Z",
|
|
description="Journée mondiale/internationale",
|
|
where=where,
|
|
online=True,
|
|
)
|
|
# Ajouter la propriété type requise par l'API OEDB
|
|
feature["properties"]["type"] = "scheduled"
|
|
# Ajouter la propriété zone
|
|
feature["properties"]["zone"] = zone
|
|
# Ajouter l'URL si disponible
|
|
if url:
|
|
feature["properties"]["url"] = url
|
|
|
|
features.append(feature)
|
|
# Appliquer la limite si définie
|
|
if limit and len(features) >= limit:
|
|
break
|
|
return features
|
|
|
|
|
|
def main() -> int:
|
|
args = build_args()
|
|
cache_cfg = CacheConfig(path=args.cache, ttl_seconds=args.cache_ttl)
|
|
|
|
src = fetch_sources(cache_cfg)
|
|
feats = convert_to_oedb(src, args.limit)
|
|
|
|
if args.dry_run:
|
|
# Imprimer le GeoJSON prêt à envoyer
|
|
collection = {"type": "FeatureCollection", "features": feats}
|
|
import json
|
|
print(json.dumps(collection, ensure_ascii=False, indent=2))
|
|
|
|
# Utiliser un cache pour éviter de renvoyer les événements déjà traités
|
|
sent_cache_path = "extractors_cache/world_days_sent.json"
|
|
ok, failed, neterr = post_oedb_features(args.base_url, feats, dry_run=args.dry_run, sent_cache_path=sent_cache_path)
|
|
print(json_report(ok, failed, neterr))
|
|
return 0
|
|
|
|
|
|
def json_report(ok: int, failed: int, neterr: int) -> str:
|
|
import json
|
|
return json.dumps({"success": ok, "failed": failed, "networkErrors": neterr}, indent=2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|
|
|
|
|