Boucle radar → article → extraction → recoupement (sondages de presse automatiques)

- extract_from_radar : les articles de presse détectés (poll-radar) sont
  scrapés, toplines extraites (partis/dates/n, dates presse sans année,
  « conservateurs » nu) → FILE DE RÉVISION (excluded=True, jamais silencieux)
- cross_validate_pending : AUTO-VALIDATION seulement par RECOUPEMENT de 2
  sources indépendantes (mêmes chiffres ±1,5 pp, même terrain ±2 j, ≥4 partis)
  — double provenance consignée, décision journalisée (PipelineLog)
- La presse publie 24-48 h avant Wikipédia : les sondages de campagne entrent
  désormais automatiquement, avec garde-fous
- Automatisation complète : replay + ablation régénérés chaque semaine dans le
  scheduler (plus aucun rapport manuel)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Simon-Pierre Boucher committed 23 days ago (Sep 1, 2026) parent 68a27f8

2 changed files +131 −5

modified backend/app/ingest/pollster_reports.py +117 −5

		`@@ -62,11 +62,11 @@ PARTY_RES = {`
62	62	"PLQ": r"(?:PLQ\|Parti\s+lib[ée]ral\|lib[ée]raux)",
63	63	"PQ": r"(?:PQ\b\|Parti\s+[Qq]u[ée]b[ée]cois)",
64	64	"QS": r"(?:QS\b\|Qu[ée]bec\s+[Ss]olidaire)",
65		− "PCQ": r"(?:PCQ\|Parti\s+conservateur\|conservateurs?\s+du\s+Qu[ée]bec)",
	65	+ "PCQ": r"(?:PCQ\|Parti\s+conservateur\|conservateurs?\b)",
66	66	}
67	67	FIELD_RE = re.compile(
68		− r"(?:du\|from)\s+(\d{1,2})(?:er)?\s+(?:au\|to)\s+(\d{1,2})(?:er)?\s+"
69		− r"([a-zéû]+)\s+(\d{4})", re.I)
	68	+ r"(?:du\|from\|les)\s+(\d{1,2})(?:er)?\s+(?:([a-zéû]+)\s+)?(?:au\|to\|et)\s+"
	69	+ r"(\d{1,2})(?:er)?\s+([a-zéû]+)(?:\s+(\d{4}))?", re.I)
70	70	MONTHS = {"janvier": 1, "février": 2, "mars": 3, "avril": 4, "mai": 5,
71	71	"juin": 6, "juillet": 7, "août": 8, "aout": 8, "septembre": 9,
72	72	"octobre": 10, "novembre": 11, "décembre": 12, "decembre": 12}
		`@@ -91,10 +91,11 @@ def extract_toplines(md: str) -> dict:`
91	91	field_end = None
92	92	m = FIELD_RE.search(md[:6000])
93	93	if m:
94		− month = MONTHS.get(m.group(3).lower())
	94	+ month = MONTHS.get(m.group(4).lower())
	95	+ year = int(m.group(5)) if m.group(5) else settings.election_date.year
95	96	if month:
96	97	try:
97		− field_end = date(int(m.group(4)), month, int(m.group(2)))
	98	+ field_end = date(year, month, int(m.group(3)))
98	99	except ValueError:
99	100	pass
100	101	n = None
		`@@ -222,6 +223,107 @@ def watch_firm(db: Session, firm: str, cfg: dict) -> dict:`
222	223	"en_révision": pending, "déjà_intégrés": integrated}
223	224
224	225
	226	+def extract_from_radar(db: Session, max_articles: int = 6) -> dict:
	227	+ """Ferme la boucle du radar : les ARTICLES DE PRESSE détectés (poll-radar)
	228	+ qui rapportent un sondage absent de la base sont scrapés et leurs toplines
	229	+ extraites → sondage en FILE DE RÉVISION (excluded=True). Pendant la
	230	+ campagne, la presse publie les chiffres 24-48 h avant Wikipédia."""
	231	+ from .firecrawl_watch import scrape_markdown, _enabled
	232	+ if not _enabled():
	233	+ return {"skipped": "Firecrawl requis"}
	234	+ pending = (db.query(Mo.WebSignal)
	235	+ .filter(Mo.WebSignal.kind == "poll-radar",
	236	+ Mo.WebSignal.status == "nouveau")
	237	+ .order_by(Mo.WebSignal.detected_at.desc())
	238	+ .limit(max_articles).all())
	239	+ created, skipped = 0, 0
	240	+ for sig in pending:
	241	+ md = scrape_markdown(sig.url)
	242	+ if not md:
	243	+ continue
	244	+ _archive(sig.pollster or "presse", sig.url, md[:200000])
	245	+ ext = extract_toplines(md)
	246	+ sig.extra = {**(sig.extra or {}),
	247	+ "extraction": {**ext, "field_end": ext["field_end"].isoformat()
	248	+ if ext["field_end"] else None}}
	249	+ firm = sig.pollster or next(
	250	+ (h for h in ("Léger", "Pallas Data", "Synopsis", "Mainstreet",
	251	+ "SEGMA", "Liaison Strategies", "Angus Reid")
	252	+ if h.split()[0].lower() in md[:3000].lower()), None)
	253	+ if not firm or ext["confidence"] < 0.7 or not ext["field_end"]:
	254	+ skipped += 1
	255	+ continue
	256	+ from ..seed import get_or_create
	257	+ pol, _ = get_or_create(db, Mo.Pollster, name=firm)
	258	+ if (db.query(Mo.Poll).filter_by(pollster_id=pol.id,
	259	+ field_end=ext["field_end"]).first()):
	260	+ sig.status = "intégré"
	261	+ continue
	262	+ el = db.query(Mo.Election).filter_by(is_target=True).first()
	263	+ total = sum(ext["shares"].values())
	264	+ poll = Mo.Poll(
	265	+ election_id=el.id, pollster_id=pol.id,
	266	+ field_start=ext["field_end"] - timedelta(days=3),
	267	+ field_end=ext["field_end"], sample_size=ext["sample_size"],
	268	+ mode=PRIMARY_SOURCES.get(firm, {}).get("mode", "unknown"),
	269	+ source_name=f"Article de presse (radar) — extraction auto",
	270	+ source_url=sig.url, excluded=True,
	271	+ exclusion_reason=("révision requise — extrait d'un article de "
	272	+ f"presse (confiance {ext['confidence']})"))
	273	+ for party, v in ext["shares"].items():
	274	+ poll.results.append(Mo.PollResult(
	275	+ party=party, raw_value=v,
	276	+ normalized_value=round(v * 100 / total, 3)))
	277	+ db.add(poll)
	278	+ db.flush()
	279	+ sig.status = "en_révision"
	280	+ sig.extra = {**sig.extra, "poll_id": poll.id}
	281	+ created += 1
	282	+ db.commit()
	283	+ return {"articles_scannés": len(pending), "en_révision": created,
	284	+ "sans_extraction_fiable": skipped}
	285	+
	286	+
	287	+def cross_validate_pending(db: Session, tol_pp: float = 1.5) -> dict:
	288	+ """AUTO-VALIDATION PAR RECOUPEMENT — jamais silencieux, jamais une seule
	289	+ source : un sondage en file de révision est validé (excluded=False)
	290	+ UNIQUEMENT si une DEUXIÈME extraction indépendante (autre URL, même maison,
	291	+ même fin de terrain ±2 j) rapporte les mêmes chiffres à ±tol_pp près.
	292	+ La double provenance et la décision sont journalisées."""
	293	+ pending = (db.query(Mo.Poll)
	294	+ .filter(Mo.Poll.excluded.is_(True),
	295	+ Mo.Poll.exclusion_reason.like("révision requise%")).all())
	296	+ validated = 0
	297	+ for poll in pending:
	298	+ shares = {r.party: r.raw_value for r in poll.results}
	299	+ confirmations = []
	300	+ for sig in db.query(Mo.WebSignal).filter(
	301	+ Mo.WebSignal.kind.in_(("poll-radar", "rapport-primaire"))).all():
	302	+ ext = (sig.extra or {}).get("extraction") or {}
	303	+ if not ext.get("shares") or sig.url == poll.source_url:
	304	+ continue
	305	+ fe = ext.get("field_end")
	306	+ if not fe or abs((date.fromisoformat(fe) - poll.field_end).days) > 2:
	307	+ continue
	308	+ diffs = [abs(ext["shares"].get(p, 0) - v)
	309	+ for p, v in shares.items() if p in ext["shares"]]
	310	+ if len(diffs) >= 4 and max(diffs) <= tol_pp:
	311	+ confirmations.append(sig.url)
	312	+ if confirmations:
	313	+ poll.excluded = False
	314	+ poll.exclusion_reason = None
	315	+ poll.methodology_notes = (
	316	+ "Auto-validé par recoupement de sources indépendantes : "
	317	+ f"{poll.source_url} + {confirmations[0]}")
	318	+ db.add(Mo.PipelineLog(step="auto-validation-sondage", status="ok",
	319	+ message=f"{poll.pollster.name} "
	320	+ f"{poll.field_end} validé par "
	321	+ f"{len(confirmations)} recoupement(s)"))
	322	+ validated += 1
	323	+ db.commit()
	324	+ return {"en_attente": len(pending) - validated, "auto_validés": validated}
	325	+
	326	+
225	327	def run_primary_watch(db: Session, max_firms: int \| None = None) -> dict:
226	328	"""Cycle des connecteurs primaires — chaque firme isolée."""
227	329	report = {}
		`@@ -233,4 +335,14 @@ def run_primary_watch(db: Session, max_firms: int \| None = None) -> dict:`
233	335	except Exception as e:
234	336	db.rollback()
235	337	report[firm] = f"échec: {e}"
	338	+ try:
	339	+ report["articles_radar"] = extract_from_radar(db)
	340	+ except Exception as e:
	341	+ db.rollback()
	342	+ report["articles_radar"] = f"échec: {e}"
	343	+ try:
	344	+ report["recoupement"] = cross_validate_pending(db)
	345	+ except Exception as e:
	346	+ db.rollback()
	347	+ report["recoupement"] = f"échec: {e}"
236	348	return report

modified backend/app/pipeline.py +14 −0

		`@@ -390,6 +390,20 @@ def run_pipeline(full_refresh: bool = True) -> dict:`
390	390	from .ingest.social_account_discovery import run_discovery
391	391	report["dgeq_canonique"] = run_canonical_checks(db)
392	392	report["découverte_comptes"] = run_discovery(db)
	393	+ # validation hebdomadaire AUTOMATIQUE : replay + ablation
	394	+ # régénérés sans intervention (rapports publics à jour)
	395	+ try:
	396	+ from .modeling.validation.historical_replay import run_replay
	397	+ run_replay()
	398	+ _log(db, "replay-hebdo", "ok", "rapport régénéré")
	399	+ except Exception:
	400	+ _log(db, "replay-hebdo", "error", traceback.format_exc())
	401	+ try:
	402	+ from .modeling.validation.ablation import run_ablation
	403	+ run_ablation(db, n_sims=6000)
	404	+ _log(db, "ablation-hebdo", "ok", "rapport régénéré")
	405	+ except Exception:
	406	+ _log(db, "ablation-hebdo", "error", traceback.format_exc())
393	407	db.add(Mo.Indicator(name="canonical_checks_last",
394	408	as_of=date.today(), value=1.0,
395	409	method="scheduler"))
396	410