sylvain
325eff5ede
- _get_performance_text_blocks : cible les éléments CSS spécifiques au site (section-spectacle-dates-date) plutôt que le texte brut de la page, évitant la contamination par le calendrier de navigation du site - match_series_to_catalog : remplace le matching LLM (trop imprécis avec les titres poétiques) par un recoupement de dates entre PDF et site - cluster_notes_into_series : passe les événements complets (avec dates) au lieu des notes seules → le LLM identifie correctement les répétitions partielles (ex: STRAUSS/PREVIN = même série que BEETHOVEN/STRAUSS/PREVIN) Résultat : Beethoven/Strauss/Previn→"Là où bat le cœur", Chostakovitch/Salonen/Prokofiev→"Virtuosité et destin", etc. Scraping réduit de 143 à 9 requêtes HTTP pour mars+avril. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
707 lines
28 KiB
Python
707 lines
28 KiB
Python
"""
|
||
core.py - Logique métier pour planning2ics web app.
|
||
Adapté de planning2ics.py pour usage web (config injectable, callback de progression).
|
||
|
||
Pipeline :
|
||
1. Extraction PDF → événements bruts
|
||
2. Clustering PDF-first → notes groupées en séries canoniques (LLM, sans site)
|
||
3. Scraping site → catalogue officiel (enrichissement optionnel)
|
||
4. Match canonique→site → correspondance pour titre officiel + description (LLM)
|
||
5. Génération ICS → une série = un fichier, même sans correspondance site
|
||
"""
|
||
|
||
import re
|
||
import json
|
||
import time as time_module
|
||
from pathlib import Path
|
||
from datetime import datetime, date, time, timedelta
|
||
from typing import Callable, Optional
|
||
|
||
import pdfplumber
|
||
import requests
|
||
from bs4 import BeautifulSoup
|
||
from icalendar import Calendar, Event
|
||
from uuid import uuid4
|
||
|
||
MONTH_MAP = {
|
||
"JANV": 1, "JAN": 1, "JANVIER": 1,
|
||
"FEV": 2, "FEVR": 2, "FEVRIER": 2,
|
||
"MARS": 3, "MAR": 3,
|
||
"AVRIL": 4, "AVR": 4,
|
||
"MAI": 5, "JUIN": 6,
|
||
"JUIL": 7, "JUILLET": 7,
|
||
"AOUT": 8, "AOÛT": 8,
|
||
"SEPT": 9, "SEP": 9, "SEPTEMBRE": 9,
|
||
"OCT": 10, "OCTOBRE": 10,
|
||
"NOV": 11, "NOVEMBRE": 11,
|
||
"DEC": 12, "DÉC": 12, "DECEMBRE": 12, "DÉCEMBRE": 12,
|
||
}
|
||
|
||
CONCERT_KEYWORDS = {
|
||
'concert', 'représentation', 'générale publique',
|
||
'raccord', 'italienne', 'scène orch'
|
||
}
|
||
|
||
FRENCH_MONTHS = {
|
||
'janvier': 1, 'février': 2, 'mars': 3, 'avril': 4,
|
||
'mai': 5, 'juin': 6, 'juillet': 7, 'août': 8,
|
||
'septembre': 9, 'octobre': 10, 'novembre': 11, 'décembre': 12,
|
||
}
|
||
|
||
|
||
# ── Utilitaires ───────────────────────────────────────────────────────────────
|
||
|
||
def normalize_note(note: str) -> str:
|
||
return re.sub(r'\s+', ' ', note).strip()
|
||
|
||
def is_public_event(titre: str) -> bool:
|
||
t = titre.lower()
|
||
return any(k in t for k in CONCERT_KEYWORDS)
|
||
|
||
def sanitize_filename(name: str) -> str:
|
||
clean = re.sub(r'[^\w\s\-éèêàùûîôç]', '', name, flags=re.UNICODE)
|
||
return clean.strip().replace(' ', '_')[:80] or 'SERIE_INCONNUE'
|
||
|
||
def extract_year_month_from_filename(filename: str):
|
||
year_match = re.search(r'(\d{4})', filename)
|
||
year = int(year_match.group(1)) if year_match else 2026
|
||
stem = Path(filename).stem.upper()
|
||
main_month = 1
|
||
for key, val in MONTH_MAP.items():
|
||
if key in stem:
|
||
main_month = val
|
||
break
|
||
return year, main_month
|
||
|
||
def parse_date(date_str: str, main_year: int, main_month: int) -> Optional[date]:
|
||
try:
|
||
day, month = map(int, date_str.strip().split('/'))
|
||
if month > main_month + 3:
|
||
year = main_year - 1
|
||
elif month < main_month - 3:
|
||
year = main_year + 1
|
||
else:
|
||
year = main_year
|
||
return date(year, month, day)
|
||
except Exception:
|
||
return None
|
||
|
||
def parse_time(s: str) -> Optional[time]:
|
||
m = re.match(r'(\d{1,2}):(\d{2})', s.strip())
|
||
return time(int(m.group(1)), int(m.group(2))) if m else None
|
||
|
||
def parse_horaires(s: str):
|
||
s = s.strip()
|
||
m = re.match(r'(\d{1,2}:\d{2})\s*[-–]\s*(\d{1,2}:\d{2})', s)
|
||
if m:
|
||
return parse_time(m.group(1)), parse_time(m.group(2))
|
||
m = re.match(r'(\d{1,2}:\d{2})', s)
|
||
if m:
|
||
return parse_time(m.group(1)), None
|
||
return None, None
|
||
|
||
|
||
# ── Extraction PDF ────────────────────────────────────────────────────────────
|
||
|
||
def extract_events_from_pdf(pdf_path: Path) -> list:
|
||
events = []
|
||
main_year, main_month = extract_year_month_from_filename(pdf_path.name)
|
||
current_date = None
|
||
|
||
with pdfplumber.open(pdf_path) as pdf:
|
||
for page in pdf.pages:
|
||
for table in (page.extract_tables() or []):
|
||
for row in table:
|
||
if not row:
|
||
continue
|
||
cells = [str(c).strip() if c else '' for c in row]
|
||
if cells[0].lower() == 'jour' or len(cells) < 5:
|
||
continue
|
||
|
||
date_str = cells[1]
|
||
horaires = cells[2]
|
||
titre = cells[3]
|
||
lieu = cells[4]
|
||
note = cells[5] if len(cells) > 5 else ''
|
||
dec = cells[6] if len(cells) > 6 else ''
|
||
voy = cells[7] if len(cells) > 7 else ''
|
||
|
||
if date_str and re.match(r'\d{1,2}/\d{2}', date_str):
|
||
parsed = parse_date(date_str, main_year, main_month)
|
||
if parsed:
|
||
current_date = parsed
|
||
|
||
if not current_date:
|
||
continue
|
||
if 'repos' in horaires.lower():
|
||
continue
|
||
if not re.search(r'\d{1,2}:\d{2}', horaires):
|
||
continue
|
||
|
||
start_time, end_time = parse_horaires(horaires)
|
||
if not start_time:
|
||
continue
|
||
|
||
events.append({
|
||
'date': current_date,
|
||
'horaires': horaires,
|
||
'start_time': start_time,
|
||
'end_time': end_time,
|
||
'titre': titre,
|
||
'lieu': lieu,
|
||
'note': normalize_note(note),
|
||
'dec': dec,
|
||
'voy': voy,
|
||
'source_file': pdf_path.name,
|
||
})
|
||
return events
|
||
|
||
|
||
# ── Scraping site web ─────────────────────────────────────────────────────────
|
||
|
||
def scrape_catalog(config: dict, cache_dir: Path,
|
||
log: Callable = None, force: bool = False,
|
||
target_dates: set = None) -> dict:
|
||
"""
|
||
Scrape les événements du site en se limitant aux dates de concerts du PDF.
|
||
|
||
target_dates : set de date objects issus du PDF (concerts/représentations).
|
||
Seules les pages site dont au moins une date correspond sont conservées.
|
||
Si target_dates est None ou vide, toutes les pages sont récupérées (mode exhaustif).
|
||
"""
|
||
cache_file = cache_dir / "website_catalog.json"
|
||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
if not force and cache_file.exists():
|
||
if log:
|
||
log("Catalogue site web chargé depuis le cache")
|
||
with open(cache_file) as f:
|
||
return json.load(f)
|
||
|
||
headers = {'User-Agent': 'Mozilla/5.0 (compatible; planning2ics/1.0)'}
|
||
calendar_url = config['site']['calendar_url']
|
||
site_base = config['site']['base_url']
|
||
|
||
# Année de référence : la plus représentée dans les dates PDF
|
||
if target_dates:
|
||
years = [d.year for d in target_dates]
|
||
reference_year = max(set(years), key=years.count)
|
||
else:
|
||
reference_year = datetime.now().year
|
||
|
||
if log:
|
||
if target_dates:
|
||
log(f"Scraping ciblé : {len(target_dates)} dates de concerts à croiser avec le site...")
|
||
else:
|
||
log("Scraping du site web de l'opéra (mode exhaustif)...")
|
||
|
||
resp = requests.get(calendar_url, headers=headers, timeout=30)
|
||
resp.raise_for_status()
|
||
soup = BeautifulSoup(resp.text, 'html.parser')
|
||
|
||
event_links = {}
|
||
for a in soup.find_all('a', href=True):
|
||
href = a['href']
|
||
if '/evenements/' in href and href.rstrip('/') != f'{site_base}/evenements':
|
||
full_url = href if href.startswith('http') else site_base + href
|
||
h3 = a.find('h3')
|
||
cat_tag = a.find('p')
|
||
title = h3.get_text(strip=True) if h3 else a.get_text(strip=True)
|
||
category = cat_tag.get_text(strip=True) if cat_tag else ''
|
||
if title and len(title) > 3:
|
||
event_links[title] = {'url': full_url, 'category': category}
|
||
|
||
total = len(event_links)
|
||
if log:
|
||
log(f"{total} événements trouvés sur le site, filtrage par dates...")
|
||
|
||
catalog = {}
|
||
fetched = 0
|
||
skipped = 0
|
||
|
||
for title, info in event_links.items():
|
||
try:
|
||
r = requests.get(info['url'], headers=headers, timeout=20)
|
||
r.raise_for_status()
|
||
page_soup = BeautifulSoup(r.text, 'html.parser')
|
||
page_dates = _parse_french_dates_from_page(page_soup, reference_year)
|
||
|
||
# Filtrage par correspondance de dates
|
||
if target_dates and not (page_dates & target_dates):
|
||
skipped += 1
|
||
time_module.sleep(0.1)
|
||
continue
|
||
|
||
catalog[title] = {
|
||
'url': info['url'],
|
||
'description': _extract_description(page_soup),
|
||
'category': info['category'],
|
||
'dates': _extract_dates_from_page(page_soup),
|
||
}
|
||
fetched += 1
|
||
time_module.sleep(0.2)
|
||
|
||
except Exception:
|
||
pass # On ignore silencieusement les pages inaccessibles
|
||
|
||
with open(cache_file, 'w') as f:
|
||
json.dump(catalog, f, ensure_ascii=False, indent=2)
|
||
|
||
if log:
|
||
log(f"Site : {fetched} événements retenus, {skipped} ignorés (dates hors planning)")
|
||
return catalog
|
||
|
||
|
||
def _extract_description(soup: BeautifulSoup) -> str:
|
||
for selector in ['div.wp-block-group', 'div.entry-content', 'article', 'main']:
|
||
container = soup.select_one(selector)
|
||
if container:
|
||
for tag in container.find_all(['nav', 'header', 'footer', 'button', 'form']):
|
||
tag.decompose()
|
||
lines = [
|
||
l.strip() for l in container.get_text('\n', strip=True).splitlines()
|
||
if l.strip() and len(l.strip()) > 15
|
||
][:40]
|
||
if lines:
|
||
return '\n'.join(lines)
|
||
return soup.get_text('\n', strip=True)[:2000]
|
||
|
||
|
||
def _get_performance_text_blocks(soup: BeautifulSoup) -> list[str]:
|
||
"""
|
||
Retourne les blocs de texte contenant les vraies dates de représentation.
|
||
Cible d'abord les éléments CSS spécifiques au site de l'opéra,
|
||
puis utilise le texte complet en fallback.
|
||
"""
|
||
# Balises ciblées sur le site opera-orchestre-montpellier.fr
|
||
for css_class in ['section-spectacle-dates-date', 'section-spectacle-dates',
|
||
'event-dates', 'dates-seances', 'programme-dates']:
|
||
elements = soup.find_all(class_=css_class)
|
||
if elements:
|
||
return [el.get_text(' ', strip=True) for el in elements]
|
||
|
||
# Fallback : balises <time>
|
||
times = soup.find_all('time')
|
||
if times:
|
||
return [t.get('datetime', '') + ' ' + t.get_text(' ', strip=True) for t in times]
|
||
|
||
# Fallback final : texte complet (moins précis)
|
||
return [soup.get_text(' ')]
|
||
|
||
|
||
def _extract_dates_from_page(soup: BeautifulSoup) -> list[str]:
|
||
"""Extrait les dates de représentation depuis une page événement du site."""
|
||
months = '|'.join(FRENCH_MONTHS.keys())
|
||
pattern = rf'\b(\d{{1,2}})\s+({months})\s*(?:(\d{{4}}))?\b'
|
||
blocks = _get_performance_text_blocks(soup)
|
||
seen, unique = set(), []
|
||
for block in blocks:
|
||
for m in re.finditer(pattern, block, re.IGNORECASE):
|
||
d = m.group(0).strip()
|
||
if d not in seen:
|
||
seen.add(d)
|
||
unique.append(d)
|
||
return unique[:15]
|
||
|
||
|
||
def _parse_french_dates_from_page(soup: BeautifulSoup, reference_year: int) -> set:
|
||
"""
|
||
Extrait et convertit les dates de représentation en objets date.
|
||
Utilisé pour filtrer les pages site par correspondance avec les dates PDF.
|
||
"""
|
||
months = '|'.join(FRENCH_MONTHS.keys())
|
||
pattern = rf'\b(\d{{1,2}})\s+({months})(?:\s+(\d{{4}}))?\b'
|
||
blocks = _get_performance_text_blocks(soup)
|
||
result = set()
|
||
for block in blocks:
|
||
for m in re.finditer(pattern, block, re.IGNORECASE):
|
||
day = int(m.group(1))
|
||
month_name = m.group(2).lower()
|
||
year = int(m.group(3)) if m.group(3) else reference_year
|
||
month = FRENCH_MONTHS.get(month_name)
|
||
if not month:
|
||
continue
|
||
try:
|
||
result.add(date(year, month, day))
|
||
except ValueError:
|
||
pass
|
||
return result
|
||
|
||
|
||
# ── LLM ───────────────────────────────────────────────────────────────────────
|
||
|
||
def _llm_call(prompt: str, ollama_url: str, model: str) -> str:
|
||
resp = requests.post(
|
||
f"{ollama_url}/api/chat",
|
||
json={
|
||
"model": model,
|
||
"messages": [{"role": "user", "content": prompt}],
|
||
"stream": True,
|
||
"options": {"temperature": 0.05, "num_predict": 16384},
|
||
"think": False,
|
||
},
|
||
stream=True,
|
||
timeout=600,
|
||
)
|
||
resp.raise_for_status()
|
||
content = ""
|
||
for line in resp.iter_lines():
|
||
if line:
|
||
chunk = json.loads(line)
|
||
content += chunk.get('message', {}).get('content', '')
|
||
if chunk.get('done'):
|
||
break
|
||
return content
|
||
|
||
|
||
def _parse_json_response(content: str, key: str) -> dict:
|
||
"""Extrait un dict depuis une réponse LLM JSON (robuste aux erreurs de parsing)."""
|
||
json_match = re.search(r'\{[\s\S]*\}', content)
|
||
if not json_match:
|
||
return {}
|
||
raw = json_match.group()
|
||
try:
|
||
return json.loads(raw).get(key, {})
|
||
except json.JSONDecodeError:
|
||
result = {}
|
||
for m in re.finditer(r'"((?:[^"\\]|\\.)*)"\s*:\s*(?:"((?:[^"\\]|\\.)*)"|null)', raw):
|
||
result[m.group(1)] = m.group(2) if m.group(2) is not None else None
|
||
return result
|
||
|
||
|
||
# ── Étape 1 : clustering PDF-first ───────────────────────────────────────────
|
||
|
||
def cluster_notes_into_series(all_events: list, config: dict, cache_dir: Path,
|
||
log: Callable = None, force: bool = False) -> dict:
|
||
"""
|
||
Groupe les notes du PDF en séries canoniques via LLM.
|
||
Utilise les dates des événements pour identifier les répétitions partielles.
|
||
Ne nécessite PAS le catalogue du site.
|
||
Retourne {note: nom_canonique_série}.
|
||
"""
|
||
cache_file = cache_dir / "series_clusters.json"
|
||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
if not force and cache_file.exists():
|
||
if log:
|
||
log("Clusters de séries chargés depuis le cache")
|
||
with open(cache_file) as f:
|
||
return json.load(f)
|
||
|
||
# Construire un index note → dates et types d'événements
|
||
note_context: dict[str, dict] = {}
|
||
for e in all_events:
|
||
n = e['note']
|
||
if not n.strip():
|
||
continue
|
||
if n not in note_context:
|
||
note_context[n] = {'dates': [], 'titres': []}
|
||
note_context[n]['dates'].append(e['date'].isoformat())
|
||
if e['titre'] not in note_context[n]['titres']:
|
||
note_context[n]['titres'].append(e['titre'])
|
||
|
||
non_empty = sorted(note_context.keys())
|
||
if not non_empty:
|
||
return {}
|
||
|
||
# Heuristique (A)/(B) : pré-traitement avant LLM
|
||
heuristic = {}
|
||
for note in non_empty:
|
||
m = re.match(r"^\(([AB]'?)\)\s*:\s*[\"']?(.+?)[\"']?\s*$", note, re.IGNORECASE)
|
||
if m:
|
||
heuristic[note] = m.group(2).strip().strip('"\'')
|
||
|
||
# Construire la liste avec contexte de dates
|
||
lines = []
|
||
for n in non_empty:
|
||
ctx = note_context[n]
|
||
dates = sorted(set(ctx['dates']))
|
||
titres = ', '.join(ctx['titres'])
|
||
date_range = f"{dates[0]} → {dates[-1]}" if len(dates) > 1 else dates[0]
|
||
lines.append(f'- {repr(n)}\n dates: {date_range} | types: {titres}')
|
||
notes_block = '\n'.join(lines)
|
||
|
||
prompt = f"""Tu analyses le planning interne d'un orchestre (Opéra Orchestre National Montpellier).
|
||
Voici des notes de programme extraites du PDF, avec leurs dates et types d'événements.
|
||
|
||
Ta tâche : regrouper ces notes en SÉRIES (même programme de concert = même série).
|
||
Pour chaque note, donne un nom canonique court et propre de la série.
|
||
|
||
Règles IMPORTANTES :
|
||
1. Mêmes compositeurs avec formulations ou ordres différents → MÊME série
|
||
2. Une note avec seulement CERTAINS compositeurs d'un programme plus complet,
|
||
dont les dates se chevauchent avec ce programme → c'est une répétition PARTIELLE
|
||
de la MÊME série (ex: "STRAUSS / PREVIN" est une répétition partielle de
|
||
"BEETHOVEN / STRAUSS / PREVIN" si leurs dates sont proches)
|
||
3. Préfixes (A), (B), (A'), (B') → SÉRIES PARALLÈLES DIFFÉRENTES (noms distincts)
|
||
4. Mentions entre parenthèses (captation, présence de..., ordre différent...) → MÊME série
|
||
5. Les séries avec seulement des répétitions (pas de concert public) sont valides
|
||
6. Nom canonique = programme COMPLET de la série (inclure tous les compositeurs du concert)
|
||
|
||
Notes à analyser (avec contexte de dates) :
|
||
{notes_block}
|
||
|
||
Réponds UNIQUEMENT avec un JSON valide, sans texte autour :
|
||
{{
|
||
"clusters": {{
|
||
"note exacte telle quelle": "Nom Canonique de la Série",
|
||
...
|
||
}}
|
||
}}"""
|
||
|
||
model = config['ollama']['cluster_model']
|
||
if log:
|
||
log(f"Identification des séries depuis le PDF ({model})...")
|
||
|
||
content = _llm_call(prompt, config['ollama']['url'], model)
|
||
result = _parse_json_response(content, 'clusters')
|
||
|
||
# Appliquer les heuristiques pour les notes non assignées
|
||
for note, canonical in heuristic.items():
|
||
if note not in result:
|
||
result[note] = canonical
|
||
|
||
# Fallback : note non assignée → elle-même (tronquée)
|
||
for note in non_empty:
|
||
if note not in result or not result[note]:
|
||
result[note] = note[:80]
|
||
|
||
with open(cache_file, 'w') as f:
|
||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||
|
||
nb_series = len(set(v for v in result.values() if v))
|
||
if log:
|
||
log(f"{len(result)} notes → {nb_series} séries identifiées depuis le PDF")
|
||
return result
|
||
|
||
|
||
# ── Étape 2 : correspondance séries PDF ↔ site par dates ─────────────────────
|
||
|
||
def _parse_catalog_dates(date_strings: list, reference_year: int) -> set:
|
||
"""Parse les dates texte du catalogue site en objets date."""
|
||
months = '|'.join(FRENCH_MONTHS.keys())
|
||
pattern = rf'(\d{{1,2}})\s+({months})(?:\s+(\d{{4}}))?'
|
||
result = set()
|
||
for s in date_strings:
|
||
m = re.search(pattern, s, re.IGNORECASE)
|
||
if m:
|
||
day = int(m.group(1))
|
||
month = FRENCH_MONTHS.get(m.group(2).lower())
|
||
year = int(m.group(3)) if m.group(3) else reference_year
|
||
if month:
|
||
try:
|
||
result.add(date(year, month, day))
|
||
except ValueError:
|
||
pass
|
||
return result
|
||
|
||
|
||
def match_series_to_catalog(series_concert_dates: dict, catalog: dict,
|
||
cache_dir: Path, log: Callable = None,
|
||
force: bool = False) -> dict:
|
||
"""
|
||
Fait correspondre les séries PDF aux événements du site par recoupement de dates.
|
||
series_concert_dates : {nom_canonique: set[date]} — dates de concerts du PDF par série.
|
||
Retourne {nom_canonique: titre_site_ou_None}.
|
||
"""
|
||
cache_file = cache_dir / "series_site_match.json"
|
||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
if not force and cache_file.exists():
|
||
if log:
|
||
log("Correspondances séries↔site chargées depuis le cache")
|
||
with open(cache_file) as f:
|
||
return json.load(f)
|
||
|
||
if not series_concert_dates or not catalog:
|
||
return {s: None for s in series_concert_dates}
|
||
|
||
# Année de référence : la plus fréquente dans toutes les dates
|
||
all_dates = [d for dates in series_concert_dates.values() for d in dates]
|
||
reference_year = max(set(d.year for d in all_dates), key=lambda y: sum(1 for d in all_dates if d.year == y)) if all_dates else datetime.now().year
|
||
|
||
# Pré-calculer les dates parsées pour chaque événement du site
|
||
site_dates: dict[str, set] = {}
|
||
for title, info in catalog.items():
|
||
site_dates[title] = _parse_catalog_dates(info.get('dates', []), reference_year)
|
||
|
||
result = {}
|
||
if log:
|
||
log("Correspondance séries PDF ↔ site par recoupement de dates...")
|
||
|
||
for canonical, pdf_concert_dates in series_concert_dates.items():
|
||
if not pdf_concert_dates:
|
||
result[canonical] = None
|
||
continue
|
||
|
||
best_title = None
|
||
best_overlap = 0
|
||
for site_title, s_dates in site_dates.items():
|
||
overlap = len(pdf_concert_dates & s_dates)
|
||
if overlap > best_overlap:
|
||
best_overlap = overlap
|
||
best_title = site_title
|
||
|
||
result[canonical] = best_title if best_overlap > 0 else None
|
||
if log:
|
||
status = f"→ {best_title!r} ({best_overlap} date(s) communes)" if best_title else "→ aucune correspondance site"
|
||
log(f" {canonical!r} {status}")
|
||
|
||
with open(cache_file, 'w') as f:
|
||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||
|
||
matched = sum(1 for v in result.values() if v)
|
||
if log:
|
||
log(f"{matched}/{len(result)} séries PDF trouvées sur le site")
|
||
return result
|
||
|
||
|
||
# ── Génération ICS ────────────────────────────────────────────────────────────
|
||
|
||
def _build_description(evt: dict, catalog_entry: Optional[dict]) -> str:
|
||
lines = []
|
||
|
||
if is_public_event(evt['titre']) and catalog_entry:
|
||
desc = catalog_entry.get('description', '')
|
||
if desc:
|
||
lines.append(desc[:1500])
|
||
site_dates = catalog_entry.get('dates', [])
|
||
if site_dates:
|
||
lines.append(f"Dates sur le site : {', '.join(site_dates)}")
|
||
url = catalog_entry.get('url', '')
|
||
if url:
|
||
lines.append(f"Site : {url}")
|
||
else:
|
||
if evt['note']:
|
||
lines.append(f"Œuvres : {evt['note']}")
|
||
lines.append(f"Type : {evt['titre']}")
|
||
|
||
if evt['dec']:
|
||
lines.append(f"Durée déclarée : {evt['dec']}")
|
||
if evt['voy']:
|
||
lines.append(f"Déplacement : {evt['voy']}h de trajet")
|
||
lines.append(f"Source : {evt['source_file']}")
|
||
return '\n'.join(lines)
|
||
|
||
|
||
def _create_ics_bytes(series_title: str, events: list,
|
||
catalog_entry: Optional[dict]) -> bytes:
|
||
cal = Calendar()
|
||
cal.add('prodid', '-//Opéra Orchestre National Montpellier//planning2ics//FR')
|
||
cal.add('version', '2.0')
|
||
cal.add('x-wr-calname', series_title)
|
||
cal.add('x-wr-timezone', 'Europe/Paris')
|
||
|
||
for evt in sorted(events, key=lambda e: (e['date'], e['start_time'])):
|
||
vevent = Event()
|
||
start_dt = datetime.combine(evt['date'], evt['start_time'])
|
||
vevent.add('dtstart', start_dt)
|
||
|
||
if evt['end_time']:
|
||
end_dt = datetime.combine(evt['date'], evt['end_time'])
|
||
else:
|
||
dec_m = re.match(r'(\d{1,2}):(\d{2})', evt['dec'])
|
||
duration = (
|
||
timedelta(hours=int(dec_m.group(1)), minutes=int(dec_m.group(2)))
|
||
if dec_m else timedelta(hours=2)
|
||
)
|
||
end_dt = start_dt + duration
|
||
|
||
vevent.add('dtend', end_dt)
|
||
vevent.add('summary', f"{evt['titre']} – {series_title}")
|
||
if evt['lieu']:
|
||
vevent.add('location', evt['lieu'])
|
||
vevent.add('description', _build_description(evt, catalog_entry))
|
||
vevent.add('uid', str(uuid4()) + '@planning-orchestre')
|
||
cal.add_component(vevent)
|
||
|
||
return cal.to_ical()
|
||
|
||
|
||
# ── Point d'entrée principal ──────────────────────────────────────────────────
|
||
|
||
def process_pdfs(pdf_paths: list, config: dict, data_dir: Path,
|
||
log: Callable = None) -> dict:
|
||
"""
|
||
Traite une liste de PDFs.
|
||
Retourne {series_title: {filename, bytes, event_count}}.
|
||
|
||
Pipeline :
|
||
PDF → extraction → clustering PDF-first → scraping site → match optionnel → ICS
|
||
Toutes les séries du PDF sont générées, même sans correspondance site.
|
||
"""
|
||
cache_dir = data_dir / "cache"
|
||
|
||
# ── 1. Extraction PDF ──────────────────────────────────────────────────────
|
||
if log:
|
||
log(f"Extraction de {len(pdf_paths)} PDF(s)...")
|
||
all_events = []
|
||
seen = set()
|
||
for i, pdf_path in enumerate(pdf_paths):
|
||
if log:
|
||
log(f"Extraction {i+1}/{len(pdf_paths)} : {pdf_path.name}")
|
||
for evt in extract_events_from_pdf(pdf_path):
|
||
key = (evt['date'], evt['start_time'], evt['titre'], evt['note'])
|
||
if key not in seen:
|
||
seen.add(key)
|
||
all_events.append(evt)
|
||
if log:
|
||
log(f"{len(all_events)} événements extraits au total")
|
||
|
||
if not all_events:
|
||
return {}
|
||
|
||
# ── 2. Clustering PDF-first (sans site) ────────────────────────────────────
|
||
unique_notes = {e['note'] for e in all_events}
|
||
if log:
|
||
log(f"{len(unique_notes)} notes uniques à analyser...")
|
||
note_to_canonical = cluster_notes_into_series(all_events, config, cache_dir, log)
|
||
|
||
# ── 3. Scraping site ciblé sur les dates de concerts du PDF ───────────────
|
||
concert_dates = {e['date'] for e in all_events if is_public_event(e['titre'])}
|
||
if log and concert_dates:
|
||
log(f"{len(concert_dates)} dates de concerts identifiées dans le PDF")
|
||
catalog = scrape_catalog(config, cache_dir, log, target_dates=concert_dates or None)
|
||
|
||
# ── 4. Correspondance canonique → site par dates ───────────────────────────
|
||
# Dates de concerts du PDF par série canonique
|
||
series_concert_dates: dict[str, set] = {}
|
||
for evt in all_events:
|
||
canonical = note_to_canonical.get(evt['note'])
|
||
if canonical and is_public_event(evt['titre']):
|
||
series_concert_dates.setdefault(canonical, set()).add(evt['date'])
|
||
canonical_to_site = match_series_to_catalog(series_concert_dates, catalog, cache_dir, log)
|
||
|
||
# ── 5. Groupement des événements par série ─────────────────────────────────
|
||
series_events: dict[str, list] = {}
|
||
for evt in all_events:
|
||
canonical = note_to_canonical.get(evt['note'])
|
||
if not canonical:
|
||
# Note vide : regrouper par type de titre
|
||
canonical = evt['titre'] or 'SANS PROGRAMME'
|
||
|
||
# Titre final = titre site si trouvé, sinon nom canonique PDF
|
||
site_title = canonical_to_site.get(canonical)
|
||
series_key = site_title if site_title else canonical
|
||
|
||
evt['_canonical'] = canonical
|
||
evt['_site_title'] = site_title
|
||
series_events.setdefault(series_key, []).append(evt)
|
||
|
||
if log:
|
||
log(f"{len(series_events)} séries à générer...")
|
||
|
||
# ── 6. Génération ICS ──────────────────────────────────────────────────────
|
||
result = {}
|
||
for series_key, events in series_events.items():
|
||
catalog_entry = catalog.get(series_key) # None si pas de correspondance site
|
||
result[series_key] = {
|
||
'filename': sanitize_filename(series_key) + '.ics',
|
||
'bytes': _create_ics_bytes(series_key, events, catalog_entry),
|
||
'event_count': len(events),
|
||
}
|
||
|
||
if log:
|
||
log(f"Terminé : {len(result)} séries générées")
|
||
return result
|