spb/qc-election

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"""Élections Québec (DGEQ) — SOURCE CANONIQUE (ADDENDUM §2, TIER 1).67Complète l'ingestion DGEQ existante (carte, candidatures — ingest/dgeq.py) :89  * **partis autorisés** : liste officielle scrapeée (Firecrawl) et archivée10    (`data/partis_autorises.json`) — la plateforme ne fige jamais la liste;11  * **vérification des chefs** : le chef affiché par la source officielle prime12    sur le registre; toute divergence lève une alerte (jamais de remplacement13    silencieux);14  * **électeurs par circonscription** : conservés depuis les propriétés du15    GeoJSON officiel lorsque publiés (amorce du feature store de population).1617Quand Élections Québec contredit Wikipédia, Élections Québec a raison.18"""19from __future__ import annotations2021import json22import logging23import re24from datetime import date2526from sqlalchemy.orm import Session2728from ..config import DATA_DIR, settings29from .. import models as Mo3031log = logging.getLogger("elections-quebec")3233URL_PARTIS = ("https://www.electionsquebec.qc.ca/partis-et-autres-entites-"34              "politiques/partis-politiques/")35OUT_PARTIS = DATA_DIR / "partis_autorises.json"3637# Rangée du tableau DGEQ des partis PROVINCIAUX :38# | [Nom](…/fiche-du-parti/ID/ID/) | n candidats | Chef | date d'autorisation |39_ROW_RE = re.compile(40    r"\|\s*\[([^\]]+)\]\((https://www\.electionsquebec\.qc\.ca/[^)]*"41    r"fiche-du-parti/[^)]*)\)\s*\|\s*(\d+)\s*\|\s*([^|]+?)\s*\|\s*([^|]+?)\s*\|")424344def refresh_authorized_parties(db: Session) -> dict:45    """Scrape le tableau officiel des partis provinciaux autorisés46    (nom, fiche, nombre de candidats, chef, date d'autorisation)."""47    from .firecrawl_watch import scrape_markdown48    md = scrape_markdown(URL_PARTIS)49    if not md:50        return {"skipped": "page DGEQ inaccessible"}51    parties = []52    for m in _ROW_RE.finditer(md):53        name, url, n_cand, leader, authorized = m.groups()54        if "fiche-du-parti-municipal" in url:55            continue56        parties.append({"name": name.strip(), "fiche_url": url,57                        "candidates": int(n_cand),58                        "leader": leader.strip(),59                        "authorized_since": authorized.strip()})60    if len(parties) < 5:61        return {"skipped": f"extraction trop pauvre ({len(parties)} partis)"}62    OUT_PARTIS.write_text(json.dumps(63        {"as_of": date.today().isoformat(), "source_url": URL_PARTIS,64         "parties": parties}, ensure_ascii=False, indent=1))65    # vérification des chefs des partis principaux vs registre interne66    alerts = 067    for p in db.query(Mo.Party).all():68        if not p.leader or p.code == "AUT":69            continue70        match = next((x for x in parties71                      if p.name.lower() in x["name"].lower()72                      or x["name"].lower() in p.name.lower()), None)73        if match and match["leader"]:74            reg_last = p.leader.split()[-1].lower()75            if reg_last not in match["leader"].lower():76                url = f"{URL_PARTIS}#chef-{p.code}-{date.today().isoformat()}"77                if not db.query(Mo.WebSignal).filter_by(url=url,78                                                        kind="dérive-chef").first():79                    db.add(Mo.WebSignal(80                        kind="dérive-chef", status="pending_review", url=url,81                        title=(f"DGEQ indique « {match['leader']} » pour "82                               f"{p.name} — registre : « {p.leader} »"),83                        extra={"party": p.code, "dgeq_leader": match["leader"],84                               "registry_leader": p.leader}))85                    alerts += 186    db.commit()87    return {"partis_autorisés": len(parties), "alertes_chef": alerts}888990def refresh_electors(db: Session) -> dict:91    """Électeurs inscrits par circonscription depuis le GeoJSON officiel brut92    (propriétés NB_ELECT*/ELECTEURS si publiées) — amorce du feature store."""93    import httpx94    from .dgeq import URL_GEOJSON95    try:96        r = httpx.get(URL_GEOJSON, timeout=120)97        r.raise_for_status()98        feats = r.json().get("features") or []99    except Exception as e:100        return {"skipped": f"GeoJSON inaccessible: {e}"}101    stored = 0102    for f in feats:103        props = f.get("properties") or {}104        name = props.get("NM_CEP") or props.get("name")105        electors = next((props[k] for k in props106                         if "ELECT" in str(k).upper() and107                         isinstance(props[k], (int, float))), None)108        if not name or electors is None:109            continue110        d = db.query(Mo.District).filter(Mo.District.name == name).first()111        if d is not None:112            notes = json.loads(d.notes) if (d.notes or "").startswith("{") else {}113            notes["electors"] = int(electors)114            d.notes = json.dumps(notes, ensure_ascii=False)115            stored += 1116    db.commit()117    return {"circonscriptions_avec_électeurs": stored}118119120def run_canonical_checks(db: Session) -> dict:121    report = {}122    for name, fn in [("partis_autorisés", refresh_authorized_parties),123                     ("électeurs", refresh_electors)]:124        try:125            report[name] = fn(db)126        except Exception as e:127            db.rollback()128            report[name] = f"échec: {e}"129    return report130