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"""Ingestion des données ouvertes officielles d'Élections Québec (DGEQ).67Sources (https://dgeq.org/donnees.html) :8  * GeoJSON des 127 circonscriptions 2026 (carte électorale 2025, sans plans d'eau);9  * liste officielle des circonscriptions 2026;10  * candidatures officielles (avec drapeau « député sortant »);11  * résultats en direct le soir du scrutin (resultats.json — provider prêt).12"""13from __future__ import annotations1415import json16import unicodedata17from datetime import datetime, timezone1819import httpx20from sqlalchemy.orm import Session2122from ..config import DATA_DIR, settings23from .. import models as Mo2425BASE = "https://donnees.electionsquebec.qc.ca"26URL_GEOJSON = f"{BASE}/autres/provincial/circonscriptions_electorales_sans_eau_2026.json"27URL_LISTE = f"{BASE}/autres/provincial/liste_circonscriptions2026.csv"28URL_CANDIDATURES = f"{BASE}/production/provincial/candidatures/candidatures.json"29URL_RESULTATS = f"{BASE}/production/provincial/resultats/resultats.json"3031PARTY_BY_NAME = {32    "parti québécois": "PQ",33    "québec solidaire": "QS",34    "coalition avenir québec": "CAQ",35    "parti conservateur du québec": "PCQ",36    "parti libéral du québec": "PLQ",37}383940def _norm(s: str) -> str:41    """Clé de rapprochement de noms (accents conservés, tirets/apostrophes unifiés)."""42    s = unicodedata.normalize("NFC", str(s)).strip()43    return s.replace("–", "-").replace("—", "-").replace("’", "'").lower()444546def party_code(nom_parti: str | None) -> str:47    if not nom_parti:48        return "AUT"49    n = _norm(nom_parti)50    for key, code in PARTY_BY_NAME.items():51        if key in n:52            return code53    return "AUT"545556def _mark(db: Session, name: str, url: str, status: str) -> None:57    src = db.query(Mo.DataSource).filter_by(url=url).first()58    if src is None:59        src = Mo.DataSource(name=name, url=url, kind="officiel")60        db.add(src)61    src.last_fetch = datetime.now(timezone.utc)62    src.last_status = status[:200]63    db.commit()646566def fetch_geojson(db: Session | None = None) -> dict:67    """Télécharge la carte officielle et l'écrit (propriétés minimales)."""68    r = httpx.get(URL_GEOJSON, timeout=120)69    r.raise_for_status()70    g = r.json()71    slim = {"type": "FeatureCollection",72            "features": [{"type": "Feature",73                          "properties": {"name": f["properties"]["NM_CEP"]},74                          "geometry": f["geometry"]} for f in g["features"]]}75    out = DATA_DIR / "carte_2026.geojson"76    out.write_text(json.dumps(slim, ensure_ascii=False))77    if db:78        _mark(db, "DGEQ — carte électorale 2026 (GeoJSON)", URL_GEOJSON,79              f"ok ({len(slim['features'])} circonscriptions)")80    return {"features": len(slim["features"]), "path": str(out)}818283def validate_names(db: Session) -> dict:84    """Rapproche les noms officiels DGEQ et les circonscriptions en base."""85    f = DATA_DIR / "carte_2026.geojson"86    if not f.exists():87        return {"error": "GeoJSON absent"}88    geo_names = {_norm(x["properties"]["name"]) for x in json.loads(f.read_text())["features"]}89    db_names = {_norm(d.name): d.name for d in db.query(Mo.District).all()}90    missing_geo = sorted(n for n in db_names if n not in geo_names)91    missing_db = sorted(n for n in geo_names if n not in db_names)92    return {"non_trouvées_dans_geo": missing_geo, "non_trouvées_en_base": missing_db}939495def fetch_candidatures(db: Session) -> dict:96    """Candidatures officielles : stockées et utilisées pour raffiner l'incumbency."""97    try:98        r = httpx.get(URL_CANDIDATURES, timeout=60)99        r.raise_for_status()100        data = r.json()101    except Exception as e:102        _mark(db, "DGEQ — candidatures officielles", URL_CANDIDATURES, f"error: {e}")103        raise104    districts = {_norm(d.name): d for d in db.query(Mo.District).all()}105    db.query(Mo.Candidate).delete()106    n, sortants = 0, 0107    for c in data:108        dname = c.get("nom_circonscription")109        if not dname:110            continue111        d = districts.get(_norm(dname))112        if d is None:113            continue114        code = party_code(c.get("nom_parti"))115        is_inc = c.get("depute_sortant") == "O"116        db.add(Mo.Candidate(117            district_id=d.id, party=code,118            name=f"{c.get('prenom_bulletin_vote') or ''} {c.get('nom_bulletin_vote') or ''}".strip(),119            party_full=c.get("nom_parti"), is_incumbent=is_inc,120            source_url=URL_CANDIDATURES))121        n += 1122        if is_inc:123            sortants += 1124            # confirmation officielle que le sortant se représente125            if d.incumbent_party == code:126                d.incumbent_running = True127    db.commit()128    _mark(db, "DGEQ — candidatures officielles", URL_CANDIDATURES,129          f"ok ({n} candidatures, {sortants} sortants)")130    return {"candidatures": n, "sortants_confirmés": sortants}131132133def refresh_dgeq(db: Session) -> dict:134    report = {}135    try:136        report["geojson"] = fetch_geojson(db)137    except Exception as e:138        report["geojson"] = f"échec: {e}"139    try:140        report["candidatures"] = fetch_candidatures(db)141    except Exception as e:142        report["candidatures"] = f"échec: {e}"143    report["validation_noms"] = validate_names(db)144    return report145