From da14137bd9cb6197063b190651630aa01966cdaa Mon Sep 17 00:00:00 2001 From: sylvain Date: Sun, 8 Mar 2026 16:27:20 +0100 Subject: [PATCH] =?UTF-8?q?Refonte=20identification=20des=20s=C3=A9ries=20?= =?UTF-8?q?:=20PDF-first=20en=20deux=20=C3=A9tapes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Avant : le LLM devait simultanément grouper les notes ET les matcher aux titres du site → résultats incohérents, séries perdues si pas de correspondance sur le site. Après (pipeline en 4 étapes) : 1. cluster_notes_into_series : LLM groupe les notes du PDF en séries canoniques, SANS le catalogue du site 2. scrape_catalog : enrichissement optionnel (+ extraction des dates de représentation depuis chaque page événement) 3. match_series_to_catalog : correspondance canonique→site pour enrichir le titre et la description (null si pas de match) 4. Génération ICS pour TOUTES les séries PDF, même sans correspondance site (répétitions seules incluses) Autres changements : - _build_description : inclut les dates du site et l'URL quand dispo - clear_cache : inclut series_clusters.json et series_site_match.json - _parse_json_response : helper robuste pour parser les réponses LLM Co-Authored-By: Claude Sonnet 4.6 --- webapp/app.py | 3 +- webapp/core.py | 316 +++++++++++++++++++++++++++++++++---------------- 2 files changed, 218 insertions(+), 101 deletions(-) diff --git a/webapp/app.py b/webapp/app.py index c7b288a..73e246f 100644 --- a/webapp/app.py +++ b/webapp/app.py @@ -300,7 +300,8 @@ async def cache_status(user: str = Depends(get_current_user)): async def clear_cache(user: str = Depends(get_current_user)): cache_dir = DATA_DIR / "cache" deleted = [] - for name in ["website_catalog.json", "series_mapping.json"]: + for name in ["website_catalog.json", "series_mapping.json", + "series_clusters.json", "series_site_match.json"]: p = cache_dir / name if p.exists(): p.unlink() diff --git a/webapp/core.py b/webapp/core.py index de223ae..b977c0a 100644 --- a/webapp/core.py +++ b/webapp/core.py @@ -1,6 +1,13 @@ """ 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 @@ -35,6 +42,12 @@ CONCERT_KEYWORDS = { '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 ─────────────────────────────────────────────────────────────── @@ -157,9 +170,9 @@ def scrape_catalog(config: dict, cache_dir: Path, with open(cache_file) as f: return json.load(f) - headers = {'User-Agent': 'Mozilla/5.0 (compatible; planning2ics/1.0)'} + headers = {'User-Agent': 'Mozilla/5.0 (compatible; planning2ics/1.0)'} calendar_url = config['site']['calendar_url'] - site_base = config['site']['base_url'] + site_base = config['site']['base_url'] if log: log("Scraping du site web de l'opéra...") @@ -173,9 +186,9 @@ def scrape_catalog(config: dict, cache_dir: Path, 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') + h3 = a.find('h3') cat_tag = a.find('p') - title = h3.get_text(strip=True) if h3 else a.get_text(strip=True) + 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} @@ -196,11 +209,13 @@ def scrape_catalog(config: dict, cache_dir: Path, 'url': info['url'], 'description': _extract_description(page_soup), 'category': info['category'], + 'dates': _extract_dates_from_page(page_soup), } time_module.sleep(0.2) except Exception: catalog[title] = { - 'url': info['url'], 'description': '', 'category': info['category'] + 'url': info['url'], 'description': '', 'category': info['category'], + 'dates': [], } with open(cache_file, 'w') as f: @@ -226,6 +241,24 @@ def _extract_description(soup: BeautifulSoup) -> str: return soup.get_text('\n', strip=True)[:2000] +def _extract_dates_from_page(soup: BeautifulSoup) -> list[str]: + """Extrait les dates de représentation depuis une page événement du site.""" + text = soup.get_text(' ') + months = '|'.join(FRENCH_MONTHS.keys()) + pattern = rf'\b(\d{{1,2}})\s+({months})\s*(?:(\d{{4}}))?\b' + found = [] + for m in re.finditer(pattern, text, re.IGNORECASE): + found.append(m.group(0).strip()) + # Déduplique en conservant l'ordre + seen = set() + unique = [] + for d in found: + if d not in seen: + seen.add(d) + unique.append(d) + return unique[:15] + + # ── LLM ─────────────────────────────────────────────────────────────────────── def _llm_call(prompt: str, ollama_url: str, model: str) -> str: @@ -252,130 +285,190 @@ def _llm_call(prompt: str, ollama_url: str, model: str) -> str: return content -def _apply_parallel_heuristic(note: str, catalog: dict) -> Optional[str]: - m = re.match(r"^\([AB]'?\)\s*:\s*[\"']?(.+?)[\"']?\s*$", note, re.IGNORECASE) - if not m: - return None - inner = m.group(1).strip().lower() - for title in catalog: - if inner in title.lower() or title.lower() in inner: - return title - return m.group(1).strip().strip('"\'') +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 -def cluster_notes_global(unique_notes: set, catalog: dict, config: dict, - cache_dir: Path, log: Callable = None, - force: bool = False) -> dict: - cache_file = cache_dir / "series_mapping.json" +# ── Étape 1 : clustering PDF-first ─────────────────────────────────────────── + +def cluster_notes_into_series(unique_notes: set, config: dict, cache_dir: Path, + log: Callable = None, force: bool = False) -> dict: + """ + Groupe les notes du PDF en séries canoniques via LLM. + 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("Mapping des séries chargé depuis le cache") + log("Clusters de séries chargés depuis le cache") with open(cache_file) as f: return json.load(f) - catalog_titles = sorted(catalog.keys()) - titles_list = '\n'.join(f'- "{t}"' for t in catalog_titles) - notes_list = '\n'.join(f'- {repr(n)}' for n in sorted(unique_notes) if n.strip()) + non_empty = sorted(n for n in unique_notes if n.strip()) + if not non_empty: + return {} - prompt = f"""Tu analyses le planning interne de l'Opéra Orchestre National Montpellier. + # 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('"\'') -Voici les titres OFFICIELS des événements de la saison (depuis le site web) : -{titles_list} + notes_list = '\n'.join(f'- {repr(n)}' for n in non_empty) -Voici toutes les notes du planning interne (certaines sont des variantes de la même série) : + prompt = f"""Tu analyses le planning interne d'un orchestre (Opéra Orchestre National Montpellier). +Voici des notes de programme extraites du PDF de planning interne. + +Ta tâche : regrouper ces notes en SÉRIES (même programme = même série). +Pour chaque note, donne un nom canonique court et propre de la série. + +Règles : +1. Mêmes œuvres/compositeurs avec formulations différentes → MÊME série +2. Préfixes (A), (B), (A'), (B') → SÉRIES PARALLÈLES DIFFÉRENTES (noms distincts) +3. Mentions entre parenthèses (captation, présence de...) ne changent PAS la série +4. Répétitions partielles (Cordes, Vents...) = même série que le Tutti complet +5. Les séries avec seulement des répétitions (pas de concert) sont valides +6. Nom canonique = titre principal de l'œuvre ou des compositeurs (ex: "Pelléas et Mélisande", "Concert Beethoven / Brahms") + +Notes à analyser : {notes_list} -Ta tâche : associer CHAQUE note à UN titre officiel. -Règles IMPORTANTES : -1. Les notes listant les mêmes compositeurs (ordre ou sous-titres différents) → MÊME série -2. Les préfixes "(A) :", "(B) :", "(A') :", "(B') :" → séries PARALLÈLES DIFFÉRENTES - Ex: '(A) : "Magdalena"' → "Magdalena" ; '(B) : "Élémentaire"' → "Élémentaire, mon cher !" -3. Les annotations entre parenthèses (captation, présence de...) ne changent PAS la série -4. Les répétitions partielles (Cordes, Vents...) = même série que le Tutti - Réponds UNIQUEMENT avec un JSON valide, sans texte autour : {{ - "matches": {{ - "note exacte telle quelle": "Titre Officiel du Site", + "clusters": {{ + "note exacte telle quelle": "Nom Canonique de la Série", ... }} }}""" model = config['ollama']['cluster_model'] if log: - log(f"Identification des séries avec l'IA ({model})...") + log(f"Identification des séries depuis le PDF ({model})...") content = _llm_call(prompt, config['ollama']['url'], model) + result = _parse_json_response(content, 'clusters') - json_match = re.search(r'\{[\s\S]*\}', content) - if not json_match: - raise ValueError("Pas de JSON dans la réponse LLM") + # Appliquer les heuristiques pour les notes non assignées + for note, canonical in heuristic.items(): + if note not in result: + result[note] = canonical - raw = json_match.group() - try: - result = json.loads(raw).get('matches', {}) - except json.JSONDecodeError: - result = {} - for m in re.finditer(r'"((?:[^"\\]|\\.)*)"\s*:\s*"((?:[^"\\]|\\.)*)"', raw): - result[m.group(1)] = m.group(2) + # 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 associées à des séries") + log(f"{len(result)} notes → {nb_series} séries identifiées depuis le PDF") return result -def match_notes_to_series(unique_notes: set, catalog: dict, config: dict, - cache_dir: Path, log: Callable = None, - force_series: bool = False) -> dict: - note_to_series = cluster_notes_global( - unique_notes, catalog, config, cache_dir, log, force_series - ) - # Heuristique (A)/(B) pour les non-assignés - for note in unique_notes: - if note not in note_to_series and note.strip(): - r = _apply_parallel_heuristic(note, catalog) - if r: - note_to_series[note] = r +# ── Étape 2 : correspondance séries PDF ↔ site (enrichissement) ────────────── - # Retry local pour les notes restantes - still_missing = [n for n in unique_notes if n.strip() and n not in note_to_series] - if still_missing: +def match_series_to_catalog(canonical_series: set, catalog: dict, config: dict, + cache_dir: Path, log: Callable = None, + force: bool = False) -> dict: + """ + Fait correspondre les noms canoniques (PDF) aux titres officiels du site. + Retourne {nom_canonique: titre_site_ou_None}. + Les séries sans correspondance reçoivent None (pas perdues). + """ + 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(f"Retry pour {len(still_missing)} notes non assignées...") - titles_str = '\n'.join(f'- "{t}"' for t in sorted(catalog.keys())) - notes_str = '\n'.join(f'- {repr(n)}' for n in still_missing) - prompt = ( - f"Associe ces notes à des titres officiels.\n" - f"Titres:\n{titles_str}\nNotes:\n{notes_str}\n" - f'Réponds UNIQUEMENT avec JSON: {{"matches": {{"note": "Titre"}}}}' - ) - content = _llm_call(prompt, config['ollama']['url'], config['ollama']['local_model']) - j = re.search(r'\{[\s\S]*\}', content) - if j: - try: - note_to_series.update(json.loads(j.group()).get('matches', {})) - except Exception: - pass + log("Correspondances séries↔site chargées depuis le cache") + with open(cache_file) as f: + return json.load(f) - return note_to_series + if not canonical_series or not catalog: + return {s: None for s in canonical_series} + + catalog_titles = sorted(catalog.keys()) + titles_list = '\n'.join(f'- "{t}"' for t in catalog_titles) + series_list = '\n'.join(f'- "{s}"' for s in sorted(canonical_series)) + + prompt = f"""Tu dois faire correspondre des noms de séries (extraits d'un planning interne) +aux titres officiels du site web de l'Opéra Orchestre National Montpellier. + +Titres officiels du site : +{titles_list} + +Séries du planning PDF : +{series_list} + +Pour chaque série PDF, donne le titre officiel le plus proche, ou null si aucun ne correspond +(ex: série de répétitions sans concert public). + +Réponds UNIQUEMENT avec un JSON valide : +{{ + "matches": {{ + "Nom série PDF": "Titre officiel du site", + "Série répétitions seules": null + }} +}}""" + + model = config['ollama']['local_model'] + if log: + log("Correspondance séries PDF ↔ site web...") + + content = _llm_call(prompt, config['ollama']['url'], model) + result = _parse_json_response(content, 'matches') + + # S'assurer que toutes les séries ont une entrée (null si absente) + for s in canonical_series: + if s not in result: + result[s] = None + + 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, series_title: str, catalog: dict) -> str: +def _build_description(evt: dict, catalog_entry: Optional[dict]) -> str: lines = [] - if is_public_event(evt['titre']): - desc = catalog.get(series_title, {}).get('description', '') - lines.append(desc[:1500] if desc else f"Programme : {evt['note']}") + + 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']: @@ -384,7 +477,8 @@ def _build_description(evt: dict, series_title: str, catalog: dict) -> str: return '\n'.join(lines) -def _create_ics_bytes(series_title: str, events: list, catalog: dict) -> bytes: +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') @@ -392,7 +486,7 @@ def _create_ics_bytes(series_title: str, events: list, catalog: dict) -> bytes: cal.add('x-wr-timezone', 'Europe/Paris') for evt in sorted(events, key=lambda e: (e['date'], e['start_time'])): - vevent = Event() + vevent = Event() start_dt = datetime.combine(evt['date'], evt['start_time']) vevent.add('dtstart', start_dt) @@ -410,7 +504,7 @@ def _create_ics_bytes(series_title: str, events: list, catalog: dict) -> bytes: vevent.add('summary', f"{evt['titre']} – {series_title}") if evt['lieu']: vevent.add('location', evt['lieu']) - vevent.add('description', _build_description(evt, series_title, catalog)) + vevent.add('description', _build_description(evt, catalog_entry)) vevent.add('uid', str(uuid4()) + '@planning-orchestre') cal.add_component(vevent) @@ -424,10 +518,14 @@ def process_pdfs(pdf_paths: list, config: dict, data_dir: Path, """ 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 + # ── 1. Extraction PDF ────────────────────────────────────────────────────── if log: log(f"Extraction de {len(pdf_paths)} PDF(s)...") all_events = [] @@ -443,30 +541,48 @@ def process_pdfs(pdf_paths: list, config: dict, data_dir: Path, if log: log(f"{len(all_events)} événements extraits au total") - # 2. Catalogue site web - catalog = scrape_catalog(config, cache_dir, log) + if not all_events: + return {} - # 3. Identification des séries + # ── 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_series = match_notes_to_series(unique_notes, catalog, config, cache_dir, log) + note_to_canonical = cluster_notes_into_series(unique_notes, config, cache_dir, log) - # 4. Groupement et génération ICS + # ── 3. Scraping site (pour enrichissement) ───────────────────────────────── + catalog = scrape_catalog(config, cache_dir, log) + + # ── 4. Correspondance canonique → site (optionnel) ───────────────────────── + canonical_series = set(v for v in note_to_canonical.values() if v) + canonical_to_site = match_series_to_catalog(canonical_series, catalog, config, cache_dir, log) + + # ── 5. Groupement des événements par série ───────────────────────────────── series_events: dict[str, list] = {} for evt in all_events: - s = note_to_series.get(evt['note']) - if s: - series_events.setdefault(s, []).append(evt) + 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"Génération de {len(series_events)} fichiers ICS...") + log(f"{len(series_events)} séries à générer...") + # ── 6. Génération ICS ────────────────────────────────────────────────────── result = {} - for series_title, events in series_events.items(): - result[series_title] = { - 'filename': sanitize_filename(series_title) + '.ics', - 'bytes': _create_ics_bytes(series_title, events, catalog), + 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), }