spb/qc-election
Public
Python 66.6%
HTML 24.8%
CSS 4.9%
JavaScript 3.6%
1# QC Élection Forecast — Plateforme de prévision électorale du Québec 20262# Auteur : Simon-Pierre Boucher3# Contact : contact@spboucher.ai4# https://www.qc-election.com5"""Connecteurs PRIMAIRES des maisons de sondage (directive maître).67Pour Léger, Pallas Data, Synopsis, Liaison Strategies, Angus Reid, SEGMA et8Mainstreet : surveillance de leurs pages de publications publiques →910 1. détection d'un nouveau rapport (lien inédit, mots-clés Québec/intentions);11 2. archivage LÉGAL du document public (markdown Firecrawl, PDF inclus) dans12 `data/raw/reports/` avec provenance complète;13 3. extraction par RÈGLES des toplines nationales (partis + %, dates de14 terrain, n) avec NIVEAU DE CONFIANCE;15 4. **file de révision** : un sondage extrait est créé `excluded=True`16 (« révision requise ») — il n'entre JAMAIS silencieusement dans le modèle;17 l'admin valide via PATCH /api/admin/poll/{id} (excluded=false);18 5. réconciliation : si Wikipédia a déjà le sondage (maison + fin de terrain),19 la détection est marquée « intégré » sans doublon.2021Chaque connecteur est tolérant : une page murée = statut de source consigné,22jamais un pipeline cassé. Qc125/338Canada ne servent jamais de source primaire.23"""24from __future__ import annotations2526import logging27import re28from datetime import date, datetime, timedelta, timezone2930from sqlalchemy.orm import Session3132from ..config import DATA_DIR, settings33from .. import models as Mo3435log = logging.getLogger("pollster-reports")3637REPORTS_DIR = DATA_DIR / "raw" / "reports"3839# Pages de publications publiques (vérifiées best-effort; un 404 est consigné).40PRIMARY_SOURCES: dict[str, dict] = {41 "Léger": {"watch": "https://leger360.com/fr/publications/",42 "mode": "web"},43 "Pallas Data": {"watch": "https://pallasdata.com/",44 "mode": "ivr"},45 "Synopsis": {"watch": "https://www.synopsisrecherche.com/",46 "mode": "web"},47 "Liaison Strategies": {"watch": "https://liaisonstrategies.ca/",48 "mode": "ivr"},49 "Angus Reid": {"watch": "https://angusreid.org/?s=quebec",50 "mode": "web"},51 "SEGMA": {"watch": "https://www.segma.ca/",52 "mode": "phone"},53 "Mainstreet": {"watch": "https://www.mainstreetresearch.ca/polls/",54 "mode": "ivr"},55}5657KEYWORDS = re.compile(r"qu[ée]bec|intention|sondage|provincial|caq|pq\b|élection",58 re.I)5960PARTY_RES = {61 "CAQ": r"(?:CAQ|Coalition\s+avenir)",62 "PLQ": r"(?:PLQ|Parti\s+lib[ée]ral|lib[ée]raux)",63 "PQ": r"(?:PQ\b|Parti\s+[Qq]u[ée]b[ée]cois)",64 "QS": r"(?:QS\b|Qu[ée]bec\s+[Ss]olidaire)",65 "PCQ": r"(?:PCQ|Parti\s+conservateur|conservateurs?\b)",66}67FIELD_RE = re.compile(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)70MONTHS = {"janvier": 1, "février": 2, "mars": 3, "avril": 4, "mai": 5,71 "juin": 6, "juillet": 7, "août": 8, "aout": 8, "septembre": 9,72 "octobre": 10, "novembre": 11, "décembre": 12, "decembre": 12}73N_RE = re.compile(r"(?:n\s*=\s*|aupr[èe]s\s+de\s+|échantillon\s+de\s+)"74 r"([\d\s ,]{3,6})", re.I)757677REGIONAL_RE = re.compile(78 r"grande?\s+région|région\s+de\s+(qu[ée]bec|montr[ée]al)|couronne|\b450\b|"79 r"île\s+de\s+montr[ée]al|chez\s+les\s+(francophones|anglophones|jeunes|"80 r"femmes|hommes|a[îi]n[ée]s)|région\s+métropolitaine", re.I)81NATIONAL_RE = re.compile(82 r"échelle\s+nationale|ensemble\s+du\s+qu[ée]bec|à\s+travers\s+le\s+qu[ée]bec|"83 r"intentions\s+de\s+vote|province", re.I)848586def extract_toplines(md: str) -> dict:87 """Extraction PAR RÈGLES des intentions NATIONALES d'un rapport ou article.8889 Scopée par PHRASE : une phrase contenant un marqueur régional/démographique90 (« grande région de Québec », « chez les jeunes »…) n'alimente JAMAIS les91 toplines nationales — les articles mêlent souvent les deux. Les phrases à92 ancre nationale (« à l'échelle nationale ») sont lues en priorité."""93 reasons, shares = [], {}94 text = md[:20000]95 sentences = re.split(r"(?<=[.!?])\s+|\n{2,}", text)96 natl = [s for s in sentences if NATIONAL_RE.search(s)97 and not REGIONAL_RE.search(s)]98 neutral = [s for s in sentences if not NATIONAL_RE.search(s)99 and not REGIONAL_RE.search(s)]100 for pool in (natl, neutral):101 for s in pool:102 for party, pat in PARTY_RES.items():103 if party in shares:104 continue105 m = re.search(pat + r"[^%\n]{0,120}?(\d{1,2})\s*%", s)106 if m:107 v = float(m.group(1))108 if 1 <= v <= 60:109 shares[party] = v110 if len([p for p in shares if p != "AUT"]) >= 4:111 break112 if not shares and REGIONAL_RE.search(text):113 reasons.append("contenu possiblement régional/démographique seulement")114 total = sum(shares.values())115 field_end = None116 m = FIELD_RE.search(md[:6000])117 if m:118 month = MONTHS.get(m.group(4).lower())119 year = int(m.group(5)) if m.group(5) else settings.election_date.year120 if month:121 try:122 field_end = date(year, month, int(m.group(3)))123 except ValueError:124 pass125 field_end_estimated = False126 if field_end is None and re.search(127 r"fin de semaine|cette semaine|derniers jours|week-?end", md[:6000], re.I):128 # dates exactes absentes mais terrain très récent affirmé par l'article :129 # estimation ÉTIQUETÉE (sert au recoupement ±2 j, jamais présentée130 # comme une date officielle — la fiche reste en révision)131 field_end = date.today() - timedelta(days=2)132 field_end_estimated = True133 n = None134 mn = N_RE.search(md[:6000])135 if mn:136 digits = re.sub(r"[^\d]", "", mn.group(1))137 if digits and 100 <= int(digits) <= 100000:138 n = int(digits)139 conf = 0.0140 if len(shares) >= 4:141 conf += 0.4142 else:143 reasons.append(f"seulement {len(shares)} partis extraits")144 if 75 <= total <= 105:145 conf += 0.3146 elif shares:147 reasons.append(f"somme implausible ({total:.0f})")148 if field_end and not field_end_estimated:149 conf += 0.2150 elif field_end_estimated:151 conf += 0.1152 reasons.append("fin de terrain ESTIMÉE (article : terrain très récent)")153 else:154 reasons.append("dates de terrain introuvables")155 if n:156 conf += 0.1157 return {"shares": shares, "field_end": field_end, "sample_size": n,158 "field_end_estimated": field_end_estimated,159 "confidence": round(conf, 2), "reasons": reasons}160161162def _archive(firm: str, url: str, md: str) -> None:163 try:164 import hashlib165 REPORTS_DIR.mkdir(parents=True, exist_ok=True)166 h = hashlib.sha1(url.encode()).hexdigest()[:12]167 safe = re.sub(r"[^a-z0-9]+", "-", firm.lower())168 (REPORTS_DIR / f"{date.today().isoformat()}_{safe}_{h}.md").write_text(169 f"<!-- {url} — archivé {datetime.now(timezone.utc).isoformat()} -->\n{md}")170 except Exception:171 pass172173174def _source_status(db: Session, firm: str, url: str, status: str) -> None:175 src = db.query(Mo.DataSource).filter_by(name=f"Primaire — {firm}").first()176 if src is None:177 src = Mo.DataSource(name=f"Primaire — {firm}", url=url, kind="poll-primary")178 db.add(src)179 src.last_fetch = datetime.now(timezone.utc)180 src.last_status = status[:200]181182183def watch_firm(db: Session, firm: str, cfg: dict) -> dict:184 """Un connecteur : page de publications → nouveaux liens → extraction."""185 from .firecrawl_watch import _enabled, _post, scrape_markdown186 if not _enabled():187 return {"skipped": "Firecrawl requis"}188 data = _post("/scrape", {"url": cfg["watch"], "formats": ["links", "markdown"]},189 timeout=90.0)190 d = (data or {}).get("data") or {}191 links, page_md = d.get("links") or [], d.get("markdown") or ""192 if not links and not page_md:193 _source_status(db, firm, cfg["watch"], "inaccessible")194 return {"statut": "page inaccessible"}195 _source_status(db, firm, cfg["watch"], "ok")196 # candidats : liens de la même maison contenant les mots-clés197 host = cfg["watch"].split("/")[2].replace("www.", "")198 cands = []199 for l in links:200 if host in l and KEYWORDS.search(l) and len(l) < 220 \201 and not any(x in l for x in ("#", "mailto:", "?share", "/tag/",202 "/category/", "/page/")):203 cands.append(l.split("?utm")[0])204 cands = list(dict.fromkeys(cands))[:6]205 new, pending, integrated = 0, 0, 0206 for url in cands:207 if db.query(Mo.WebSignal).filter_by(url=url, kind="rapport-primaire").first():208 continue209 md = scrape_markdown(url)210 if not md or not KEYWORDS.search(md[:4000]):211 continue212 _archive(firm, url, md[:250000])213 ext = extract_toplines(md)214 sig = Mo.WebSignal(kind="rapport-primaire", url=url,215 title=f"{firm} — rapport détecté",216 pollster=firm, status="nouveau",217 snippet=md[:400],218 extra={"extraction": {**ext,219 "field_end": ext["field_end"].isoformat()220 if ext["field_end"] else None}})221 db.add(sig)222 new += 1223 # réconciliation Wikipédia / création en file de révision224 if ext["field_end"] and ext["confidence"] >= 0.7:225 pollster = db.query(Mo.Pollster).filter_by(name=firm).first()226 exists = (pollster and db.query(Mo.Poll)227 .filter_by(pollster_id=pollster.id,228 field_end=ext["field_end"]).first())229 if exists:230 sig.status = "intégré"231 sig.extra = {**sig.extra, "poll_id": exists.id}232 integrated += 1233 else:234 from ..seed import get_or_create235 el = db.query(Mo.Election).filter_by(is_target=True).first()236 pol, _ = get_or_create(db, Mo.Pollster, name=firm)237 total = sum(ext["shares"].values())238 poll = Mo.Poll(239 election_id=el.id, pollster_id=pol.id,240 field_start=ext["field_end"] - timedelta(days=3),241 field_end=ext["field_end"],242 sample_size=ext["sample_size"],243 mode=cfg.get("mode", "unknown"),244 source_name=f"Rapport primaire {firm} (extraction auto)",245 source_url=url, excluded=True,246 exclusion_reason=("révision requise — extraction "247 f"automatique (confiance {ext['confidence']})"))248 for party, v in ext["shares"].items():249 poll.results.append(Mo.PollResult(250 party=party, raw_value=v,251 normalized_value=round(v * 100 / total, 3)))252 db.add(poll)253 db.flush()254 sig.status = "en_révision"255 sig.extra = {**sig.extra, "poll_id": poll.id}256 pending += 1257 db.commit()258 return {"liens": len(cands), "nouveaux": new,259 "en_révision": pending, "déjà_intégrés": integrated}260261262def extract_from_radar(db: Session, max_articles: int = 6) -> dict:263 """Ferme la boucle du radar : les ARTICLES DE PRESSE détectés (poll-radar)264 qui rapportent un sondage absent de la base sont scrapés et leurs toplines265 extraites → sondage en FILE DE RÉVISION (excluded=True). Pendant la266 campagne, la presse publie les chiffres 24-48 h avant Wikipédia."""267 from .firecrawl_watch import scrape_markdown, _enabled268 if not _enabled():269 return {"skipped": "Firecrawl requis"}270 pending = (db.query(Mo.WebSignal)271 .filter(Mo.WebSignal.kind == "poll-radar",272 Mo.WebSignal.status == "nouveau")273 .order_by(Mo.WebSignal.detected_at.desc())274 .limit(max_articles).all())275 created, skipped = 0, 0276 for sig in pending:277 md = scrape_markdown(sig.url)278 if not md:279 continue280 _archive(sig.pollster or "presse", sig.url, md[:200000])281 ext = extract_toplines(md)282 sig.extra = {**(sig.extra or {}),283 "extraction": {**ext, "field_end": ext["field_end"].isoformat()284 if ext["field_end"] else None}}285 firm = sig.pollster or next(286 (h for h in ("Léger", "Pallas Data", "Synopsis", "Mainstreet",287 "SEGMA", "Liaison Strategies", "Angus Reid")288 if h.split()[0].lower() in md[:3000].lower()), None)289 if not firm or ext["confidence"] < 0.7 or not ext["field_end"]:290 skipped += 1291 continue292 from ..seed import get_or_create293 pol, _ = get_or_create(db, Mo.Pollster, name=firm)294 if (db.query(Mo.Poll).filter_by(pollster_id=pol.id,295 field_end=ext["field_end"]).first()):296 sig.status = "intégré"297 continue298 el = db.query(Mo.Election).filter_by(is_target=True).first()299 total = sum(ext["shares"].values())300 poll = Mo.Poll(301 election_id=el.id, pollster_id=pol.id,302 field_start=ext["field_end"] - timedelta(days=3),303 field_end=ext["field_end"], sample_size=ext["sample_size"],304 mode=PRIMARY_SOURCES.get(firm, {}).get("mode", "unknown"),305 source_name=f"Article de presse (radar) — extraction auto",306 source_url=sig.url, excluded=True,307 exclusion_reason=("révision requise — extrait d'un article de "308 f"presse (confiance {ext['confidence']})"))309 for party, v in ext["shares"].items():310 poll.results.append(Mo.PollResult(311 party=party, raw_value=v,312 normalized_value=round(v * 100 / total, 3)))313 db.add(poll)314 db.flush()315 sig.status = "en_révision"316 sig.extra = {**sig.extra, "poll_id": poll.id}317 created += 1318 db.commit()319 return {"articles_scannés": len(pending), "en_révision": created,320 "sans_extraction_fiable": skipped}321322323def cross_validate_pending(db: Session, tol_pp: float = 1.5) -> dict:324 """AUTO-VALIDATION PAR RECOUPEMENT — jamais silencieux, jamais une seule325 source : un sondage en file de révision est validé (excluded=False)326 UNIQUEMENT si une DEUXIÈME extraction indépendante (autre URL, même maison,327 même fin de terrain ±2 j) rapporte les mêmes chiffres à ±tol_pp près.328 La double provenance et la décision sont journalisées."""329 pending = (db.query(Mo.Poll)330 .filter(Mo.Poll.excluded.is_(True),331 Mo.Poll.exclusion_reason.like("révision requise%")).all())332 validated = 0333 for poll in pending:334 shares = {r.party: r.raw_value for r in poll.results}335 confirmations = []336 for sig in db.query(Mo.WebSignal).filter(337 Mo.WebSignal.kind.in_(("poll-radar", "rapport-primaire"))).all():338 ext = (sig.extra or {}).get("extraction") or {}339 if not ext.get("shares") or sig.url == poll.source_url:340 continue341 fe = ext.get("field_end")342 if not fe or abs((date.fromisoformat(fe) - poll.field_end).days) > 2:343 continue344 diffs = [abs(ext["shares"].get(p, 0) - v)345 for p, v in shares.items() if p in ext["shares"]]346 if len(diffs) >= 4 and max(diffs) <= tol_pp:347 confirmations.append(sig.url)348 if confirmations:349 poll.excluded = False350 poll.exclusion_reason = None351 poll.methodology_notes = (352 "Auto-validé par recoupement de sources indépendantes : "353 f"{poll.source_url} + {confirmations[0]}")354 db.add(Mo.PipelineLog(step="auto-validation-sondage", status="ok",355 message=f"{poll.pollster.name} "356 f"{poll.field_end} validé par "357 f"{len(confirmations)} recoupement(s)"))358 validated += 1359 db.commit()360 return {"en_attente": len(pending) - validated, "auto_validés": validated}361362363def run_primary_watch(db: Session, max_firms: int | None = None) -> dict:364 """Cycle des connecteurs primaires — chaque firme isolée."""365 report = {}366 for i, (firm, cfg) in enumerate(PRIMARY_SOURCES.items()):367 if max_firms is not None and i >= max_firms:368 break369 try:370 report[firm] = watch_firm(db, firm, cfg)371 except Exception as e:372 db.rollback()373 report[firm] = f"échec: {e}"374 try:375 report["articles_radar"] = extract_from_radar(db)376 except Exception as e:377 db.rollback()378 report["articles_radar"] = f"échec: {e}"379 try:380 report["recoupement"] = cross_validate_pending(db)381 except Exception as e:382 db.rollback()383 report["recoupement"] = f"échec: {e}"384 return report385