Compare commits
12 Commits
d78ac420d8
...
main
Author | SHA1 | Date | |
---|---|---|---|
2908f94b4a
|
|||
645951823a
|
|||
47cfdf705a
|
|||
600aa8d5c1
|
|||
205f8a5c84
|
|||
313b161a06
|
|||
47641924df
|
|||
9543df0d8d
|
|||
b514a61d72
|
|||
22ebabda99
|
|||
b4d6885631
|
|||
0b0508958c
|
8
main.py
8
main.py
@ -14,9 +14,12 @@ def parse_args():
|
||||
parser = argparse.ArgumentParser(description="Gestion d'élections")
|
||||
parser.add_argument('--type', '-t', type=str, help="Type d'élection",
|
||||
choices=["présidentielle", "législatives", "sénatoriales", "européennes",
|
||||
"régionales", "départementales", "municipales"], default="européennes")
|
||||
"régionales", "départementales", "municipales"], default="législatives")
|
||||
parser.add_argument('--year', '-y', type=int, help="Année de l'élection",
|
||||
choices=[2022, 2024], default=2024)
|
||||
parser.add_argument('--round', '-r', type=int,
|
||||
help="Tour de l'élection à importer (0 pour tous, valeur par défaut)",
|
||||
choices=[0, 1, 2], default=0)
|
||||
parser.add_argument('action', help="Action à réaliser",
|
||||
choices=["import_candidats", "import_geographie", "import_resultats",
|
||||
"import_resultats_web", "export_resultats"])
|
||||
@ -30,6 +33,7 @@ def main():
|
||||
load_dotenv()
|
||||
|
||||
args = parse_args()
|
||||
data_round = args.round
|
||||
debug = args.debug
|
||||
verbose = args.verbose
|
||||
|
||||
@ -58,7 +62,7 @@ def main():
|
||||
case "import_resultats":
|
||||
if verbose:
|
||||
print(f"Import des résultats pour les élections {args.type} {args.year}")
|
||||
election_module.import_resultats.run(engine, verbose)
|
||||
election_module.import_resultats.run(engine, data_round, verbose)
|
||||
case "import_resultats_web":
|
||||
if verbose:
|
||||
print(f"Import des résultats depuis le web pour les élections {args.type} {args.year}")
|
||||
|
@ -1,8 +1,9 @@
|
||||
import {createBrowserRouter, RouterProvider} from "react-router-dom"
|
||||
|
||||
import './App.css'
|
||||
import ElectionsLegislatives2022 from "./ElectionsLegislatives2022"
|
||||
import ElectionsLegislatives2022 from './ElectionsLegislatives2022'
|
||||
import ElectionsEuropeennes2024 from './ElectionsEuropeennes2024'
|
||||
import ElectionsLegislatives2024 from './ElectionsLegislatives2024'
|
||||
|
||||
function App() {
|
||||
const router = createBrowserRouter([
|
||||
@ -26,14 +27,14 @@ function App() {
|
||||
path: "/elections/europeennes/2024/:typeResultats/:zoneId/",
|
||||
element: <ElectionsEuropeennes2024 />
|
||||
},
|
||||
// {
|
||||
// path: "/elections/legislatives/2024/:typeResultats/",
|
||||
// element: <ElectionsLegislatives2024 />,
|
||||
// },
|
||||
// {
|
||||
// path: "/elections/legislatives/2024/:typeResultats/:zoneId/",
|
||||
// element: <ElectionsLegislatives2024 />
|
||||
// },
|
||||
{
|
||||
path: "/elections/legislatives/2024/:typeResultats/",
|
||||
element: <ElectionsLegislatives2024 />,
|
||||
},
|
||||
{
|
||||
path: "/elections/legislatives/2024/:typeResultats/:zoneId/",
|
||||
element: <ElectionsLegislatives2024 />
|
||||
},
|
||||
])
|
||||
|
||||
return <>
|
||||
|
148
nupes-elections-front/src/ElectionsLegislatives2024.js
Normal file
148
nupes-elections-front/src/ElectionsLegislatives2024.js
Normal file
@ -0,0 +1,148 @@
|
||||
import {useParams} from "react-router-dom"
|
||||
import {AppBar, Container, Toolbar} from "@mui/material"
|
||||
import * as Highcharts from 'highcharts'
|
||||
import highchartsItem from 'highcharts/modules/item-series'
|
||||
import {useEffect, useMemo, useState} from "react"
|
||||
import {
|
||||
SelectionAffichage,
|
||||
TableauParticipation,
|
||||
CarteResultats,
|
||||
HistogrammeVoix, GroupementParBloc, SelectionTour
|
||||
} from "./includes/composants_elections"
|
||||
import {getNomZone, regrouperVoix} from "./utils"
|
||||
import 'leaflet/dist/leaflet.css'
|
||||
import {
|
||||
TableauResultatsCandidatsLegislatives,
|
||||
TableauResultatsNuancesLegislatives
|
||||
} from "./includes/composants_elections_legislatives"
|
||||
|
||||
|
||||
highchartsItem(Highcharts)
|
||||
|
||||
export default function ElectionsLegislatives2024() {
|
||||
const {typeResultats, zoneId} = useParams()
|
||||
|
||||
const [grouperParBloc, setGrouperParBloc] = useState(false)
|
||||
const [tour, setTour] = useState(1)
|
||||
const [blocs, setBlocs] = useState([])
|
||||
const [candidats, setCandidats] = useState([])
|
||||
const [nuances, setNuances] = useState([])
|
||||
const [resultats, setResultats] = useState([])
|
||||
const [typeSousZone, setTypeSousZone] = useState("region")
|
||||
|
||||
useEffect(() => {
|
||||
fetch("/data/resultats/legislatives/2024/blocs.json").then(response => response.json())
|
||||
.then(data => setBlocs(data))
|
||||
fetch("/data/resultats/legislatives/2024/nuances.json").then(response => response.json())
|
||||
.then(data => setNuances(data))
|
||||
|
||||
if (typeResultats === "france") {
|
||||
fetch("/data/resultats/legislatives/2024/france.json").then(response => response.json())
|
||||
.then(data => setResultats(data))
|
||||
}
|
||||
else {
|
||||
fetch(`/data/resultats/legislatives/2024/${typeResultats}/${zoneId}.json`)
|
||||
.then(response => response.json())
|
||||
.then(data => setResultats(data))
|
||||
}
|
||||
}, [typeResultats, zoneId])
|
||||
|
||||
const zoneInfo = useMemo(() => resultats?.zone ?? {}, [resultats])
|
||||
const nomZone = useMemo(() => getNomZone(typeResultats, zoneInfo), [typeResultats, zoneInfo])
|
||||
|
||||
useEffect(() => {
|
||||
let circonscription = ""
|
||||
|
||||
if (typeResultats === "circonscription")
|
||||
circonscription = zoneId
|
||||
else if (typeResultats === "bureau_vote")
|
||||
circonscription = zoneInfo?.circonscription ?? ""
|
||||
else {
|
||||
setCandidats(nuances)
|
||||
}
|
||||
|
||||
if (!circonscription)
|
||||
return
|
||||
|
||||
fetch(`/data/resultats/legislatives/2024/candidats/${circonscription}.json`)
|
||||
.then(response => response.json())
|
||||
.then(data => setCandidats(data))
|
||||
}, [typeResultats, zoneId, zoneInfo, nuances])
|
||||
|
||||
const donnees = useMemo(() => {
|
||||
if (tour === 1)
|
||||
return resultats?.tour1 ?? {}
|
||||
else if (tour === 2)
|
||||
return resultats?.tour2 ?? {}
|
||||
else
|
||||
return {}
|
||||
}, [resultats, tour])
|
||||
|
||||
const dejaGroupeParNuance = typeResultats !== "circonscription" && typeResultats !== "bureau_vote"
|
||||
const [voixParBloc, voixParNuance] = regrouperVoix(donnees.voix, candidats, blocs, nuances,
|
||||
dejaGroupeParNuance)
|
||||
|
||||
const candidatKey = typeSousZone === "circonscription" || typeSousZone === "bureau_vote" ? "numero" : "code"
|
||||
|
||||
const [categoriesHistogramme, valeursHistogramme, couleursHistogramme] = useMemo(() => {
|
||||
if (grouperParBloc) {
|
||||
const categories = {}
|
||||
const couleurs = {}
|
||||
for (let bloc of blocs) {
|
||||
categories[bloc.nom] = bloc.nom
|
||||
couleurs[bloc.nom] = bloc.couleur
|
||||
}
|
||||
return [categories, voixParBloc, couleurs]
|
||||
}
|
||||
else {
|
||||
if (typeResultats === "circonscription" || typeResultats === "bureau_vote") {
|
||||
// On affiche les noms des candidat⋅es
|
||||
const categories = {}
|
||||
const couleurs = {}
|
||||
for (let candidat of candidats) {
|
||||
categories[candidat.numero] = `${candidat.prenom} ${candidat.nom} (${candidat.nuance})`
|
||||
couleurs[candidat.numero] = nuances.filter(nuance => nuance.code === candidat.nuance)[0]?.couleur
|
||||
}
|
||||
return [categories, donnees.voix, couleurs]
|
||||
}
|
||||
else {
|
||||
// On affiche les nuances
|
||||
const categories = {}
|
||||
const couleurs = {}
|
||||
for (let nuance of nuances) {
|
||||
categories[nuance.code] = nuance.nom
|
||||
couleurs[nuance.code] = nuance.couleur
|
||||
}
|
||||
return [categories, voixParNuance, couleurs]
|
||||
|
||||
}
|
||||
}
|
||||
}, [typeResultats, candidats, nuances, blocs, donnees, voixParNuance, voixParBloc, grouperParBloc])
|
||||
|
||||
const tableauResultats = useMemo(() => {
|
||||
if (typeResultats === "circonscription" || typeResultats === "bureau_vote")
|
||||
return <TableauResultatsCandidatsLegislatives candidats={candidats} blocs={blocs} nuances={nuances} donnees_t1={resultats.tour1} donnees_t2={resultats.tour2} />
|
||||
else
|
||||
return <TableauResultatsNuancesLegislatives blocs={blocs} nuances={nuances} donnees={donnees} />
|
||||
}, [typeResultats, candidats, blocs, nuances, donnees])
|
||||
|
||||
return <>
|
||||
<AppBar position="sticky">
|
||||
<Toolbar>
|
||||
<GroupementParBloc grouperParBloc={grouperParBloc} setGrouperParBloc={setGrouperParBloc} />
|
||||
<SelectionAffichage typeResultats={typeResultats} typeSousZone={typeSousZone} setTypeSousZone={setTypeSousZone} />
|
||||
<SelectionTour tour={tour} setTour={setTour} />
|
||||
</Toolbar>
|
||||
</AppBar>
|
||||
<Container>
|
||||
<HistogrammeVoix titre={`Résultats des élections législatives 2024 : ${nomZone}`}
|
||||
nomCategories={categoriesHistogramme} valeurParCategorie={valeursHistogramme}
|
||||
totalExprimes={donnees.exprimes} couleurParCategorie={couleursHistogramme} />
|
||||
{tableauResultats}
|
||||
<TableauParticipation donnees_t1={resultats.tour1 ?? {}} donnees_t2={resultats.tour2 ?? {}} />
|
||||
<CarteResultats typeElection={"legislatives"} anneeElection={"2024"} typeResultats={typeResultats} zoneInfo={zoneInfo}
|
||||
typeSousZone={typeSousZone} candidats={candidats} blocs={blocs} nuances={nuances} tour={tour}
|
||||
grouperParBloc={grouperParBloc} candidatKey={candidatKey} />
|
||||
</Container>
|
||||
</>
|
||||
}
|
@ -118,7 +118,7 @@ export function CompositionHemicycle({titre, blocs, nuances, siegesParBloc, sieg
|
||||
*/
|
||||
export function TableauParticipation({donnees_t1, donnees_t2 = null}) {
|
||||
const headerRow = useMemo(() => {
|
||||
if (donnees_t2) {
|
||||
if (donnees_t2 && donnees_t2.inscrits) {
|
||||
return <TableRow>
|
||||
<TableCell></TableCell>
|
||||
<TableCell>Nombre tour 1</TableCell>
|
||||
@ -140,7 +140,7 @@ export function TableauParticipation({donnees_t1, donnees_t2 = null}) {
|
||||
}, [donnees_t2])
|
||||
|
||||
const bodyRows = useMemo(() => {
|
||||
if (donnees_t2) {
|
||||
if (donnees_t2 && donnees_t2.inscrits) {
|
||||
return <>
|
||||
<TableRow key={"Inscrit⋅es"}>
|
||||
<TableCell>Inscrit⋅es</TableCell>
|
||||
|
@ -12,6 +12,18 @@ export function TableauResultatsCandidatsLegislatives({blocs, candidats, nuances
|
||||
const voixCandidats_t2 = donnees_t2?.voix ?? {}
|
||||
const candidatsTriees = trierCandidats(candidats, voixCandidats_t1)
|
||||
|
||||
const cellules_t1 = <>
|
||||
<TableCell>Voix tour 1</TableCell>
|
||||
<TableCell>% Inscrit⋅es tour 1</TableCell>
|
||||
<TableCell>% Exprimé⋅es tour 1</TableCell>
|
||||
</>
|
||||
|
||||
const cellules_t2 = Math.max(...Object.values(voixCandidats_t2)) === 0 ? <></> : <>
|
||||
<TableCell>Voix tour 2</TableCell>
|
||||
<TableCell>% Inscrit⋅es tour 2</TableCell>
|
||||
<TableCell>% Exprimé⋅es tour 2</TableCell>
|
||||
</>
|
||||
|
||||
return <>
|
||||
<TableContainer component={Paper}>
|
||||
<Table sx={{ minWidth: 650 }} aria-label="simple table">
|
||||
@ -21,12 +33,8 @@ export function TableauResultatsCandidatsLegislatives({blocs, candidats, nuances
|
||||
<TableCell>Candidat</TableCell>
|
||||
<TableCell colSpan={2}>Nuance</TableCell>
|
||||
<TableCell colSpan={2}>Bloc</TableCell>
|
||||
<TableCell>Voix tour 1</TableCell>
|
||||
<TableCell>% Inscrit⋅es tour 1</TableCell>
|
||||
<TableCell>% Exprimé⋅es tour 1</TableCell>
|
||||
<TableCell>Voix tour 2</TableCell>
|
||||
<TableCell>% Inscrit⋅es tour 2</TableCell>
|
||||
<TableCell>% Exprimé⋅es tour 2</TableCell>
|
||||
{cellules_t1}
|
||||
{cellules_t2}
|
||||
</TableRow>
|
||||
</TableHead>
|
||||
<TableBody>
|
||||
|
@ -417,7 +417,11 @@ def calculer_resultats_france(engine: Engine, verbose: bool = False) -> None:
|
||||
session.commit()
|
||||
|
||||
|
||||
def run(engine: Engine, verbose: bool = False) -> None:
|
||||
def run(engine: Engine, data_round: int = 0, verbose: bool = False) -> None:
|
||||
if data_round != 0 and data_round != 1:
|
||||
# Un seul tour à l'élection européenne
|
||||
return
|
||||
|
||||
importer_resultats_region(engine, verbose)
|
||||
importer_resultats_departement(engine, verbose)
|
||||
calculer_resultats_france(engine, verbose)
|
||||
|
@ -151,7 +151,7 @@ def exporter_resultats_regions(engine: Engine, verbose: bool = False) -> None:
|
||||
if not resultats_region:
|
||||
resultats_france = session.execute(select(ResultatsFranceLegislatives2022)).scalar_one()
|
||||
resultats_region = ResultatsRegionLegislatives2022(region_id=region.code_insee,
|
||||
resultats_france_id=resultats_france.id)
|
||||
resultats_france_id=resultats_france.id)
|
||||
session.add(resultats_region)
|
||||
|
||||
resultats_dict = {
|
||||
@ -366,7 +366,7 @@ def exporter_resultats_communes(engine: Engine, verbose: bool = False) -> None:
|
||||
resultats_departement = session.execute(select(ResultatsDepartementLegislatives2022)
|
||||
.filter_by(dpt_id=commune.departement_code)).scalar_one()
|
||||
resultats_commune = ResultatsCommuneLegislatives2022(commune_id=commune.code_insee,
|
||||
resultats_dpt_id=resultats_departement.id)
|
||||
resultats_dpt_id=resultats_departement.id)
|
||||
session.add(resultats_commune)
|
||||
|
||||
resultats_dict = {
|
||||
@ -436,7 +436,7 @@ def exporter_resultats_bureaux_vote(engine: Engine, verbose: bool = False) -> No
|
||||
resultats_commune = session.execute(select(ResultatsCommuneLegislatives2022)
|
||||
.filter_by(commune_id=bureau_vote.commune_code)).scalar_one()
|
||||
resultats_bureau_vote = ResultatsBureauVoteLegislatives2022(bv_id=bureau_vote.id,
|
||||
resultats_commune_id=resultats_commune.id)
|
||||
resultats_commune_id=resultats_commune.id)
|
||||
session.add(resultats_bureau_vote)
|
||||
|
||||
resultats_dict = {
|
||||
|
@ -9,7 +9,7 @@ from nupes.models import BureauVote
|
||||
from nupes.models.legislatives2022 import *
|
||||
|
||||
|
||||
def importer_resultats_bv(engine: Engine, verbose: bool = False) -> None:
|
||||
def importer_resultats_bv(engine: Engine, data_round: int = 0, verbose: bool = False) -> None:
|
||||
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/a1f73b85-8194-44f4-a2b7-c343edb47d32"),
|
||||
(2, "https://www.data.gouv.fr/fr/datasets/r/96ffddda-59b4-41b8-a6a3-dfe1adb7fa36")]
|
||||
|
||||
@ -20,8 +20,11 @@ def importer_resultats_bv(engine: Engine, verbose: bool = False) -> None:
|
||||
for bv in session.execute(select(BureauVote)).scalars().all()}
|
||||
|
||||
for tour, file_url in tours:
|
||||
file = get_file("https://www.data.gouv.fr/fr/datasets/r/a1f73b85-8194-44f4-a2b7-c343edb47d32",
|
||||
"resultats-legislatives-2022-t1-par-bureau-de-vote.csv")
|
||||
if data_round > 0 and tour != data_round:
|
||||
# On saute ce tour
|
||||
continue
|
||||
|
||||
file = get_file(file_url, f"resultats-legislatives-2022-t{tour}-par-bureau-de-vote.csv")
|
||||
|
||||
with file.open('r', encoding="ISO-8859-1") as f:
|
||||
reader = csv.reader(f, delimiter=';')
|
||||
@ -167,32 +170,40 @@ def importer_resultats_bv(engine: Engine, verbose: bool = False) -> None:
|
||||
session.commit()
|
||||
|
||||
|
||||
def importer_resultats_commune(engine: Engine, verbose: bool = False) -> None:
|
||||
def importer_resultats_commune(engine: Engine, data_round: int = 0, verbose: bool = False) -> None:
|
||||
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/a9a82bcc-304e-491f-a4f0-c06575113745"),
|
||||
(2, "https://www.data.gouv.fr/fr/datasets/r/e79b2e51-0841-4266-b626-57cb271c7a35")]
|
||||
|
||||
with Session(engine) as session:
|
||||
for resultats_commune in session.execute(select(ResultatsCommuneLegislatives2022)).scalars().all():
|
||||
resultats_commune.inscrits_t1 = 0
|
||||
resultats_commune.votants_t1 = 0
|
||||
resultats_commune.abstentions_t1 = 0
|
||||
resultats_commune.exprimes_t1 = 0
|
||||
resultats_commune.blancs_t1 = 0
|
||||
resultats_commune.nuls_t1 = 0
|
||||
resultats_commune.inscrits_t2 = 0
|
||||
resultats_commune.votants_t2 = 0
|
||||
resultats_commune.abstentions_t2 = 0
|
||||
resultats_commune.exprimes_t2 = 0
|
||||
resultats_commune.blancs_t2 = 0
|
||||
resultats_commune.nuls_t2 = 0
|
||||
if data_round == 0 or data_round == 1:
|
||||
resultats_commune.inscrits_t1 = 0
|
||||
resultats_commune.votants_t1 = 0
|
||||
resultats_commune.abstentions_t1 = 0
|
||||
resultats_commune.exprimes_t1 = 0
|
||||
resultats_commune.blancs_t1 = 0
|
||||
resultats_commune.nuls_t1 = 0
|
||||
if data_round == 0 or data_round == 2:
|
||||
resultats_commune.inscrits_t2 = 0
|
||||
resultats_commune.votants_t2 = 0
|
||||
resultats_commune.abstentions_t2 = 0
|
||||
resultats_commune.exprimes_t2 = 0
|
||||
resultats_commune.blancs_t2 = 0
|
||||
resultats_commune.nuls_t2 = 0
|
||||
|
||||
for voix_nuance in resultats_commune.voix:
|
||||
voix_nuance.voix_t1 = 0
|
||||
voix_nuance.voix_t2 = 0
|
||||
if data_round == 0 or data_round == 1:
|
||||
voix_nuance.voix_t1 = 0
|
||||
if data_round == 0 or data_round == 2:
|
||||
voix_nuance.voix_t2 = 0
|
||||
|
||||
session.commit()
|
||||
|
||||
for tour, file_url in tours:
|
||||
if data_round > 0 and tour != data_round:
|
||||
# On saute ce tour
|
||||
continue
|
||||
|
||||
file = get_file(file_url,
|
||||
f"resultats-legislatives-2022-t{tour}-par-commune.csv")
|
||||
|
||||
@ -311,11 +322,15 @@ def importer_resultats_commune(engine: Engine, verbose: bool = False) -> None:
|
||||
session.commit()
|
||||
|
||||
|
||||
def importer_resultats_circo(engine: Engine, verbose: bool = False) -> None:
|
||||
def importer_resultats_circo(engine: Engine, data_round: int = 0, verbose: bool = False) -> None:
|
||||
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/114b13e8-7ff9-437f-9ec8-7a29258a80e3"),
|
||||
(2, "https://www.data.gouv.fr/fr/datasets/r/436b9eaa-64e1-46bd-ad1d-9638ebbd6caf")]
|
||||
|
||||
for tour, file_url in tours:
|
||||
if data_round > 0 and tour != data_round:
|
||||
# On saute ce tour
|
||||
continue
|
||||
|
||||
file = get_file(file_url, f"resultats-legislatives-2022-t{tour}-par-circonscription.csv")
|
||||
|
||||
with file.open('r', encoding="ISO-8859-1") as f:
|
||||
@ -429,11 +444,15 @@ def importer_resultats_circo(engine: Engine, verbose: bool = False) -> None:
|
||||
session.commit()
|
||||
|
||||
|
||||
def importer_resultats_departement(engine: Engine, verbose: bool = False) -> None:
|
||||
def importer_resultats_departement(engine: Engine, data_round: int = 0, verbose: bool = False) -> None:
|
||||
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/4acba699-e512-4dc5-87c7-ba74db773633"),
|
||||
(2, "https://www.data.gouv.fr/fr/datasets/r/bc76bd72-cf65-45b2-b651-7716dc3c5520")]
|
||||
|
||||
for tour, file_url in tours:
|
||||
if data_round > 0 and tour != data_round:
|
||||
# On saute ce tour
|
||||
continue
|
||||
|
||||
file = get_file(file_url, f"resultats-legislatives-2022-t{tour}-par-departement.csv")
|
||||
|
||||
with file.open('r', encoding="ISO-8859-1") as f:
|
||||
@ -532,11 +551,15 @@ def importer_resultats_departement(engine: Engine, verbose: bool = False) -> Non
|
||||
session.commit()
|
||||
|
||||
|
||||
def importer_resultats_region(engine: Engine, verbose: bool = False) -> None:
|
||||
def importer_resultats_region(engine: Engine, data_round: int = 0, verbose: bool = False) -> None:
|
||||
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/f79576c9-9ddc-43f9-8d99-f0856550054e"),
|
||||
(2, "https://www.data.gouv.fr/fr/datasets/r/fdba421a-d4e9-4f38-9753-982a7c065911")]
|
||||
|
||||
for tour, file_url in tours:
|
||||
if data_round > 0 and tour != data_round:
|
||||
# On saute ce tour
|
||||
continue
|
||||
|
||||
file = get_file(file_url, f"resultats-legislatives-2022-t{tour}-par-region.csv")
|
||||
|
||||
with file.open('r', encoding="ISO-8859-1") as f:
|
||||
@ -601,10 +624,15 @@ def importer_resultats_region(engine: Engine, verbose: bool = False) -> None:
|
||||
session.commit()
|
||||
|
||||
|
||||
def importer_resultats_france(engine: Engine, verbose: bool = False) -> None:
|
||||
def importer_resultats_france(engine: Engine, data_round: int = 0, verbose: bool = False) -> None:
|
||||
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/cbe72f2c-2c56-4251-89cb-213de8bfbea0"),
|
||||
(2, "https://www.data.gouv.fr/fr/datasets/r/6639bb39-13ec-49dd-ab4d-1f8227b86c01")]
|
||||
|
||||
for tour, file_url in tours:
|
||||
if data_round > 0 and tour != data_round:
|
||||
# On saute ce tour
|
||||
continue
|
||||
|
||||
file = get_file(file_url, f"resultats-legislatives-2022-t{tour}-france-entiere.csv")
|
||||
|
||||
with file.open('r', encoding="ISO-8859-1") as f:
|
||||
@ -659,10 +687,10 @@ def importer_resultats_france(engine: Engine, verbose: bool = False) -> None:
|
||||
session.commit()
|
||||
|
||||
|
||||
def run(engine: Engine, verbose: bool = False) -> None:
|
||||
importer_resultats_france(engine, verbose)
|
||||
importer_resultats_region(engine, verbose)
|
||||
importer_resultats_departement(engine, verbose)
|
||||
importer_resultats_circo(engine, verbose)
|
||||
importer_resultats_commune(engine, verbose)
|
||||
importer_resultats_bv(engine, verbose)
|
||||
def run(engine: Engine, data_round: int = 0, verbose: bool = False) -> None:
|
||||
importer_resultats_france(engine, data_round, verbose)
|
||||
importer_resultats_region(engine, data_round, verbose)
|
||||
importer_resultats_departement(engine, data_round, verbose)
|
||||
importer_resultats_circo(engine, data_round, verbose)
|
||||
importer_resultats_commune(engine, data_round, verbose)
|
||||
importer_resultats_bv(engine, data_round, verbose)
|
||||
|
@ -1 +1 @@
|
||||
from . import import_candidats
|
||||
from . import export_resultats, import_candidats, import_resultats
|
||||
|
494
nupes/scripts/legislatives2024/export_resultats.py
Normal file
494
nupes/scripts/legislatives2024/export_resultats.py
Normal file
@ -0,0 +1,494 @@
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import Engine, select
|
||||
from sqlalchemy.orm import Session
|
||||
from tqdm import tqdm
|
||||
|
||||
from nupes.models.geographie import BureauVote
|
||||
from nupes.models.legislatives2024 import *
|
||||
|
||||
DATA_DIR = Path(__file__).parent.parent.parent.parent / 'data'
|
||||
|
||||
|
||||
def exporter_nuances(engine: Engine, verbose: bool = False) -> None:
|
||||
with Session(engine) as session:
|
||||
blocs = session.execute(select(BlocLegislatives2024)).scalars().all()
|
||||
blocs_json = []
|
||||
|
||||
for bloc in blocs:
|
||||
bloc_json = {'id': bloc.id, 'nom': bloc.nom, 'couleur': bloc.couleur}
|
||||
blocs_json.append(bloc_json)
|
||||
|
||||
file = DATA_DIR / "resultats" / "legislatives" / "2024" / "blocs.json"
|
||||
if not file.parent.is_dir():
|
||||
file.parent.mkdir(parents=True)
|
||||
|
||||
with file.open('w') as f:
|
||||
json.dump(blocs_json, f)
|
||||
|
||||
nuances = session.execute(select(NuanceLegislatives2024)).scalars().all()
|
||||
nuances_json = []
|
||||
|
||||
for nuance in nuances:
|
||||
nuance_json = {
|
||||
'code': nuance.code,
|
||||
'nom': nuance.nom,
|
||||
'couleur': nuance.couleur,
|
||||
"bloc": nuance.bloc.nom,
|
||||
}
|
||||
nuances_json.append(nuance_json)
|
||||
|
||||
file = DATA_DIR / "resultats" / "legislatives" / "2024" / "nuances.json"
|
||||
if not file.parent.is_dir():
|
||||
file.parent.mkdir(parents=True)
|
||||
|
||||
with file.open('w') as f:
|
||||
json.dump(nuances_json, f)
|
||||
|
||||
|
||||
def exporter_candidats(engine: Engine, verbose: bool = False) -> None:
|
||||
with Session(engine) as session:
|
||||
circonscriptions = session.execute(select(Circonscription)).scalars().all()
|
||||
for circonscription in tqdm(circonscriptions, desc="Candidat⋅es", disable=not verbose):
|
||||
candidats = session.execute(select(CandidatLegislatives2024)
|
||||
.filter_by(circonscription_id=circonscription.id)).scalars().all()
|
||||
candidats_json = []
|
||||
|
||||
for candidat in candidats:
|
||||
candidat_json = {
|
||||
'numero': candidat.numero,
|
||||
'nuance': candidat.nuance_id,
|
||||
'bloc': candidat.nuance.bloc.nom,
|
||||
'nom': candidat.nom,
|
||||
'prenom': candidat.prenom,
|
||||
'nom_suppleance': candidat.nom_suppleance,
|
||||
'prenom_suppleance': candidat.prenom_suppleance,
|
||||
}
|
||||
candidats_json.append(candidat_json)
|
||||
|
||||
file = DATA_DIR / "resultats" / "legislatives" / "2024" / "candidats" / f"{circonscription.id}.json"
|
||||
if not file.parent.is_dir():
|
||||
file.parent.mkdir(parents=True)
|
||||
|
||||
with file.open('w') as f:
|
||||
json.dump(candidats_json, f)
|
||||
|
||||
|
||||
def exporter_resultats_france(engine: Engine, verbose: bool = False) -> None:
|
||||
with Session(engine) as session:
|
||||
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one_or_none()
|
||||
if not resultats_france:
|
||||
resultats_france = ResultatsFranceLegislatives2024()
|
||||
session.add(resultats_france)
|
||||
|
||||
resultats_dict = {
|
||||
'zone': {
|
||||
'type': "france",
|
||||
'regions': [reg.code_insee for reg in session.execute(select(Region)).scalars().all()],
|
||||
'departements': [dpt.code_insee for dpt in session.execute(select(Departement)).scalars().all()],
|
||||
'circonscriptions': [circo.id for circo in session.execute(select(Circonscription)).scalars().all()],
|
||||
'geometry': {},
|
||||
},
|
||||
'tour1': {
|
||||
'inscrits': resultats_france.inscrits_t1,
|
||||
'votants': resultats_france.votants_t1,
|
||||
'abstentions': resultats_france.abstentions_t1,
|
||||
'exprimes': resultats_france.exprimes_t1,
|
||||
'blancs': resultats_france.blancs_t1,
|
||||
'nuls': resultats_france.nuls_t1,
|
||||
},
|
||||
'tour2': {
|
||||
'inscrits': resultats_france.inscrits_t2,
|
||||
'votants': resultats_france.votants_t2,
|
||||
'abstentions': resultats_france.abstentions_t2,
|
||||
'exprimes': resultats_france.exprimes_t2,
|
||||
'blancs': resultats_france.blancs_t2,
|
||||
'nuls': resultats_france.nuls_t2,
|
||||
},
|
||||
}
|
||||
|
||||
resultats_t1, resultats_t2 = {}, {}
|
||||
resultats_dict['tour1']['voix'] = resultats_t1
|
||||
resultats_dict['tour2']['voix'] = resultats_t2
|
||||
for voix_nuance in resultats_france.voix:
|
||||
resultats_t1[voix_nuance.nuance_id] = voix_nuance.voix_t1
|
||||
resultats_t2[voix_nuance.nuance_id] = voix_nuance.voix_t2
|
||||
|
||||
file = DATA_DIR / "resultats" / "legislatives" / "2024" / "france.json"
|
||||
if not file.parent.is_dir():
|
||||
file.parent.mkdir(parents=True)
|
||||
|
||||
with file.open('w') as f:
|
||||
json.dump(resultats_dict, f)
|
||||
|
||||
session.commit()
|
||||
|
||||
|
||||
def exporter_resultats_regions(engine: Engine, verbose: bool = False) -> None:
|
||||
with Session(engine) as session:
|
||||
regions = session.execute(select(Region)).scalars().all()
|
||||
regions_json = []
|
||||
|
||||
regions_iterator = tqdm(regions, desc="Régions") if verbose else regions
|
||||
for region in regions_iterator:
|
||||
region_json = {
|
||||
'type': "region",
|
||||
'code_insee': region.code_insee,
|
||||
'nom': region.libelle,
|
||||
'departements': [dpt.code_insee for dpt in region.departements],
|
||||
'circonscriptions': [
|
||||
circo.id for circo in session.execute(
|
||||
select(Circonscription).join(Departement).filter_by(region_code=region.code_insee))
|
||||
.scalars().all()
|
||||
],
|
||||
'geometry': region.geometry,
|
||||
}
|
||||
regions_json.append(region_json)
|
||||
|
||||
resultats_region = session.execute(select(ResultatsRegionLegislatives2024)
|
||||
.filter_by(region_id=region.code_insee)).scalar_one_or_none()
|
||||
if not resultats_region:
|
||||
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one()
|
||||
resultats_region = ResultatsRegionLegislatives2024(region_id=region.code_insee,
|
||||
resultats_france_id=resultats_france.id)
|
||||
session.add(resultats_region)
|
||||
|
||||
resultats_dict = {
|
||||
'zone': region_json,
|
||||
'tour1': {
|
||||
'inscrits': resultats_region.inscrits_t1,
|
||||
'votants': resultats_region.votants_t1,
|
||||
'abstentions': resultats_region.abstentions_t1,
|
||||
'exprimes': resultats_region.exprimes_t1,
|
||||
'blancs': resultats_region.blancs_t1,
|
||||
'nuls': resultats_region.nuls_t1,
|
||||
},
|
||||
'tour2': {
|
||||
'inscrits': resultats_region.inscrits_t2,
|
||||
'votants': resultats_region.votants_t2,
|
||||
'abstentions': resultats_region.abstentions_t2,
|
||||
'exprimes': resultats_region.exprimes_t2,
|
||||
'blancs': resultats_region.blancs_t2,
|
||||
'nuls': resultats_region.nuls_t2,
|
||||
},
|
||||
}
|
||||
|
||||
resultats_t1, resultats_t2 = {}, {}
|
||||
resultats_dict['tour1']['voix'] = resultats_t1
|
||||
resultats_dict['tour2']['voix'] = resultats_t2
|
||||
for voix_nuance in resultats_region.voix:
|
||||
resultats_t1[voix_nuance.nuance_id] = voix_nuance.voix_t1
|
||||
resultats_t2[voix_nuance.nuance_id] = voix_nuance.voix_t2
|
||||
|
||||
file = DATA_DIR / "resultats" / "legislatives" / "2024" / "region" / f"{region.code_insee}.json"
|
||||
if not file.parent.is_dir():
|
||||
file.parent.mkdir(parents=True)
|
||||
|
||||
with file.open('w') as f:
|
||||
json.dump(resultats_dict, f)
|
||||
|
||||
session.commit()
|
||||
|
||||
regions_file = DATA_DIR / "resultats" / "legislatives" / "2024" / "region" / "regions.json"
|
||||
if not regions_file.parent.is_dir():
|
||||
regions_file.parent.mkdir(parents=True)
|
||||
|
||||
with regions_file.open('w') as f:
|
||||
json.dump(regions_json, f)
|
||||
|
||||
|
||||
def exporter_resultats_departements(engine: Engine, verbose: bool = False) -> None:
|
||||
with Session(engine) as session:
|
||||
departements = session.execute(select(Departement)).scalars().all()
|
||||
departements_json = []
|
||||
|
||||
iterator = tqdm(departements, desc="Départements") if verbose else departements
|
||||
for departement in iterator:
|
||||
departement_json = {
|
||||
'type': "departement",
|
||||
'code_insee': departement.code_insee,
|
||||
'nom': departement.libelle,
|
||||
'region': departement.region_code,
|
||||
'circonscriptions': [circo.id for circo in departement.circonscriptions],
|
||||
'communes': [commune.code_insee for commune in departement.communes],
|
||||
'geometry': departement.geometry,
|
||||
}
|
||||
departements_json.append(departement_json)
|
||||
|
||||
resultats_departement = session.execute(
|
||||
select(ResultatsDepartementLegislatives2024)
|
||||
.filter_by(dpt_id=departement.code_insee)).scalar_one_or_none()
|
||||
if not resultats_departement:
|
||||
resultats_region = session.execute(select(ResultatsRegionLegislatives2024)
|
||||
.filter_by(region_id=departement.region_code)).scalar_one()
|
||||
resultats_departement = ResultatsDepartementLegislatives2024(dpt_id=departement.code_insee,
|
||||
resultats_region_id=resultats_region.id)
|
||||
session.add(resultats_departement)
|
||||
|
||||
resultats_dict = {
|
||||
'zone': departement_json,
|
||||
'tour1': {
|
||||
'inscrits': resultats_departement.inscrits_t1,
|
||||
'votants': resultats_departement.votants_t1,
|
||||
'abstentions': resultats_departement.abstentions_t1,
|
||||
'exprimes': resultats_departement.exprimes_t1,
|
||||
'blancs': resultats_departement.blancs_t1,
|
||||
'nuls': resultats_departement.nuls_t1,
|
||||
},
|
||||
'tour2': {
|
||||
'inscrits': resultats_departement.inscrits_t2,
|
||||
'votants': resultats_departement.votants_t2,
|
||||
'abstentions': resultats_departement.abstentions_t2,
|
||||
'exprimes': resultats_departement.exprimes_t2,
|
||||
'blancs': resultats_departement.blancs_t2,
|
||||
'nuls': resultats_departement.nuls_t2,
|
||||
},
|
||||
}
|
||||
|
||||
resultats_t1, resultats_t2 = {}, {}
|
||||
resultats_dict['tour1']['voix'] = resultats_t1
|
||||
resultats_dict['tour2']['voix'] = resultats_t2
|
||||
for voix_nuance in resultats_departement.voix:
|
||||
resultats_t1[voix_nuance.nuance_id] = voix_nuance.voix_t1
|
||||
resultats_t2[voix_nuance.nuance_id] = voix_nuance.voix_t2
|
||||
|
||||
file = DATA_DIR / "resultats" / "legislatives" / "2024" / "departement" / f"{departement.code_insee}.json"
|
||||
if not file.parent.is_dir():
|
||||
file.parent.mkdir(parents=True)
|
||||
|
||||
with file.open('w') as f:
|
||||
json.dump(resultats_dict, f)
|
||||
|
||||
session.commit()
|
||||
|
||||
departements_file = DATA_DIR / "resultats" / "legislatives" / "2024" / "departement" / "departements.json"
|
||||
if not departements_file.parent.is_dir():
|
||||
departements_file.parent.mkdir(parents=True)
|
||||
|
||||
with departements_file.open('w') as f:
|
||||
json.dump(departements_json, f)
|
||||
|
||||
|
||||
def exporter_resultats_circonscriptions(engine: Engine, verbose: bool = False) -> None:
|
||||
with Session(engine) as session:
|
||||
circonscriptions = session.execute(select(Circonscription)).scalars().all()
|
||||
circonscriptions_json = []
|
||||
|
||||
iterator = tqdm(circonscriptions, desc="Circonscriptions") if verbose else circonscriptions
|
||||
for circonscription in iterator:
|
||||
circonscription_json = {
|
||||
'type': "circonscription",
|
||||
'id': circonscription.id,
|
||||
'departement': circonscription.departement_code,
|
||||
'numero': circonscription.numero,
|
||||
'bureaux_vote': [bv.id for bv in circonscription.bureaux_vote],
|
||||
'geometry': circonscription.geometry,
|
||||
}
|
||||
circonscriptions_json.append(circonscription_json)
|
||||
|
||||
resultats_circonscription = session.execute(
|
||||
select(ResultatsCirconscriptionLegislatives2024)
|
||||
.filter_by(circo_id=circonscription.id)).scalar_one_or_none()
|
||||
if not resultats_circonscription:
|
||||
resultats_departement = session.execute(select(ResultatsDepartementLegislatives2024)
|
||||
.filter_by(dpt_id=circonscription.departement_code)) \
|
||||
.scalar_one()
|
||||
resultats_circonscription = ResultatsCirconscriptionLegislatives2024(
|
||||
circo_id=circonscription.id, resultats_departement_id=resultats_departement.id)
|
||||
session.add(resultats_circonscription)
|
||||
|
||||
resultats_dict = {
|
||||
'zone': circonscription_json,
|
||||
'tour1': {
|
||||
'inscrits': resultats_circonscription.inscrits_t1,
|
||||
'votants': resultats_circonscription.votants_t1,
|
||||
'abstentions': resultats_circonscription.abstentions_t1,
|
||||
'exprimes': resultats_circonscription.exprimes_t1,
|
||||
'blancs': resultats_circonscription.blancs_t1,
|
||||
'nuls': resultats_circonscription.nuls_t1,
|
||||
},
|
||||
'tour2': {
|
||||
'inscrits': resultats_circonscription.inscrits_t2,
|
||||
'votants': resultats_circonscription.votants_t2,
|
||||
'abstentions': resultats_circonscription.abstentions_t2,
|
||||
'exprimes': resultats_circonscription.exprimes_t2,
|
||||
'blancs': resultats_circonscription.blancs_t2,
|
||||
'nuls': resultats_circonscription.nuls_t2,
|
||||
},
|
||||
}
|
||||
|
||||
resultats_t1, resultats_t2 = {}, {}
|
||||
resultats_dict['tour1']['voix'] = resultats_t1
|
||||
resultats_dict['tour2']['voix'] = resultats_t2
|
||||
for voix_candidat in resultats_circonscription.voix:
|
||||
resultats_t1[voix_candidat.candidat.numero] = voix_candidat.voix_t1
|
||||
resultats_t2[voix_candidat.candidat.numero] = voix_candidat.voix_t2
|
||||
|
||||
file = DATA_DIR / "resultats" / "legislatives" / "2024" / "circonscription" / f"{circonscription.id}.json"
|
||||
if not file.parent.is_dir():
|
||||
file.parent.mkdir(parents=True)
|
||||
|
||||
with file.open('w') as f:
|
||||
json.dump(resultats_dict, f)
|
||||
|
||||
session.commit()
|
||||
|
||||
circonscriptions_file = (DATA_DIR / "resultats" / "legislatives" / "2024"
|
||||
/ "circonscription" / "circonscriptions.json")
|
||||
if not circonscriptions_file.parent.is_dir():
|
||||
circonscriptions_file.parent.mkdir(parents=True)
|
||||
|
||||
with circonscriptions_file.open('w') as f:
|
||||
json.dump(circonscriptions_json, f)
|
||||
|
||||
|
||||
def exporter_resultats_communes(engine: Engine, verbose: bool = False) -> None:
|
||||
with Session(engine) as session:
|
||||
communes = session.execute(select(Commune)).scalars().all()
|
||||
communes_json = []
|
||||
|
||||
iterator = tqdm(communes, desc="Communes") if verbose else communes
|
||||
for commune in iterator:
|
||||
commune_json = {
|
||||
'type': "commune",
|
||||
'code_insee': commune.code_insee,
|
||||
'nom': commune.libelle,
|
||||
'departement': commune.departement_code,
|
||||
'bureaux_vote': [bv.id for bv in commune.bureaux_vote],
|
||||
'geometry': commune.geometry,
|
||||
}
|
||||
communes_json.append(commune_json)
|
||||
|
||||
resultats_commune = session.execute(
|
||||
select(ResultatsCommuneLegislatives2024).filter_by(commune_id=commune.code_insee)).scalar_one_or_none()
|
||||
if not resultats_commune:
|
||||
resultats_departement = session.execute(select(ResultatsDepartementLegislatives2024)
|
||||
.filter_by(dpt_id=commune.departement_code)).scalar_one()
|
||||
resultats_commune = ResultatsCommuneLegislatives2024(commune_id=commune.code_insee,
|
||||
resultats_dpt_id=resultats_departement.id)
|
||||
session.add(resultats_commune)
|
||||
|
||||
resultats_dict = {
|
||||
'zone': commune_json,
|
||||
'tour1': {
|
||||
'inscrits': resultats_commune.inscrits_t1,
|
||||
'votants': resultats_commune.votants_t1,
|
||||
'abstentions': resultats_commune.abstentions_t1,
|
||||
'exprimes': resultats_commune.exprimes_t1,
|
||||
'blancs': resultats_commune.blancs_t1,
|
||||
'nuls': resultats_commune.nuls_t1,
|
||||
},
|
||||
'tour2': {
|
||||
'inscrits': resultats_commune.inscrits_t2,
|
||||
'votants': resultats_commune.votants_t2,
|
||||
'abstentions': resultats_commune.abstentions_t2,
|
||||
'exprimes': resultats_commune.exprimes_t2,
|
||||
'blancs': resultats_commune.blancs_t2,
|
||||
'nuls': resultats_commune.nuls_t2,
|
||||
},
|
||||
}
|
||||
|
||||
resultats_t1, resultats_t2 = {}, {}
|
||||
resultats_dict['tour1']['voix'] = resultats_t1
|
||||
resultats_dict['tour2']['voix'] = resultats_t2
|
||||
for voix_nuance in resultats_commune.voix:
|
||||
resultats_t1[voix_nuance.nuance_id] = voix_nuance.voix_t1
|
||||
resultats_t2[voix_nuance.nuance_id] = voix_nuance.voix_t2
|
||||
|
||||
file = DATA_DIR / "resultats" / "legislatives" / "2024" / "commune" / f"{commune.code_insee}.json"
|
||||
if not file.parent.is_dir():
|
||||
file.parent.mkdir(parents=True)
|
||||
|
||||
with file.open('w') as f:
|
||||
json.dump(resultats_dict, f)
|
||||
|
||||
session.commit()
|
||||
|
||||
communes_file = DATA_DIR / "resultats" / "legislatives" / "2024" / "commune" / "communes.json"
|
||||
if not communes_file.parent.is_dir():
|
||||
communes_file.parent.mkdir(parents=True)
|
||||
|
||||
with communes_file.open('w') as f:
|
||||
json.dump(communes_json, f)
|
||||
|
||||
|
||||
def exporter_resultats_bureaux_vote(engine: Engine, verbose: bool = False) -> None:
|
||||
with Session(engine) as session:
|
||||
bureaux_vote = session.execute(select(BureauVote)).scalars().all()
|
||||
bureaux_vote_json = []
|
||||
|
||||
iterator = tqdm(bureaux_vote, desc="Bureaux de vote") if verbose else bureaux_vote
|
||||
for bureau_vote in iterator:
|
||||
bureau_vote_json = {
|
||||
'type': "bureau_vote",
|
||||
'id': bureau_vote.id,
|
||||
'libelle': bureau_vote.libelle,
|
||||
'commune': bureau_vote.commune_code,
|
||||
'circonscription': bureau_vote.circo_code,
|
||||
'geometry': bureau_vote.geometry,
|
||||
}
|
||||
bureaux_vote_json.append(bureau_vote_json)
|
||||
|
||||
resultats_bureau_vote = session.execute(
|
||||
select(ResultatsBureauVoteLegislatives2024).filter_by(bv_id=bureau_vote.id)).scalar_one_or_none()
|
||||
if not resultats_bureau_vote:
|
||||
resultats_commune = session.execute(select(ResultatsCommuneLegislatives2024)
|
||||
.filter_by(commune_id=bureau_vote.commune_code)).scalar_one()
|
||||
resultats_bureau_vote = ResultatsBureauVoteLegislatives2024(bv_id=bureau_vote.id,
|
||||
resultats_commune_id=resultats_commune.id)
|
||||
session.add(resultats_bureau_vote)
|
||||
|
||||
resultats_dict = {
|
||||
'zone': bureau_vote_json,
|
||||
'tour1': {
|
||||
'inscrits': resultats_bureau_vote.inscrits_t1,
|
||||
'votants': resultats_bureau_vote.votants_t1,
|
||||
'abstentions': resultats_bureau_vote.abstentions_t1,
|
||||
'exprimes': resultats_bureau_vote.exprimes_t1,
|
||||
'blancs': resultats_bureau_vote.blancs_t1,
|
||||
'nuls': resultats_bureau_vote.nuls_t1,
|
||||
},
|
||||
'tour2': {
|
||||
'inscrits': resultats_bureau_vote.inscrits_t2,
|
||||
'votants': resultats_bureau_vote.votants_t2,
|
||||
'abstentions': resultats_bureau_vote.abstentions_t2,
|
||||
'exprimes': resultats_bureau_vote.exprimes_t2,
|
||||
'blancs': resultats_bureau_vote.blancs_t2,
|
||||
'nuls': resultats_bureau_vote.nuls_t2,
|
||||
},
|
||||
}
|
||||
|
||||
resultats_t1, resultats_t2 = {}, {}
|
||||
resultats_dict['tour1']['voix'] = resultats_t1
|
||||
resultats_dict['tour2']['voix'] = resultats_t2
|
||||
for voix_candidat in resultats_bureau_vote.voix:
|
||||
resultats_t1[voix_candidat.candidat.numero] = voix_candidat.voix_t1
|
||||
resultats_t2[voix_candidat.candidat.numero] = voix_candidat.voix_t2
|
||||
|
||||
file = DATA_DIR / "resultats" / "legislatives" / "2024" / "bureau_vote" / f"{bureau_vote.id}.json"
|
||||
if not file.parent.is_dir():
|
||||
file.parent.mkdir(parents=True)
|
||||
|
||||
with file.open('w') as f:
|
||||
json.dump(resultats_dict, f)
|
||||
|
||||
session.commit()
|
||||
|
||||
bureaux_vote_file = DATA_DIR / "resultats" / "legislatives" / "2024" / "bureau_vote" / "bureaux_vote.json"
|
||||
if not bureaux_vote_file.parent.is_dir():
|
||||
bureaux_vote_file.parent.mkdir(parents=True)
|
||||
|
||||
with bureaux_vote_file.open('w') as f:
|
||||
json.dump(bureaux_vote_json, f)
|
||||
|
||||
|
||||
def run(engine: Engine, verbose: bool = False) -> None:
|
||||
exporter_nuances(engine, verbose)
|
||||
exporter_candidats(engine, verbose)
|
||||
exporter_resultats_france(engine, verbose)
|
||||
exporter_resultats_regions(engine, verbose)
|
||||
exporter_resultats_departements(engine, verbose)
|
||||
exporter_resultats_circonscriptions(engine, verbose)
|
||||
exporter_resultats_communes(engine, verbose)
|
||||
exporter_resultats_bureaux_vote(engine, verbose)
|
@ -71,12 +71,12 @@ def creer_nuances(engine: Engine, verbose: bool = False) -> None:
|
||||
|
||||
|
||||
def importer_candidats(engine: Engine, verbose: bool = False) -> None:
|
||||
DATASET_URL = "https://www.data.gouv.fr/fr/datasets/r/e02a62a1-9f4b-49d6-bb32-05dc4da8c581"
|
||||
DATASET_URL = "https://www.data.gouv.fr/fr/datasets/r/cd5b39e6-5feb-4864-b274-3431ed81442c"
|
||||
file = get_file(DATASET_URL, "candidats_legislatives_2024.csv")
|
||||
|
||||
with file.open('r') as f:
|
||||
with Session(engine) as session:
|
||||
reader = csv.DictReader(f)
|
||||
reader = csv.DictReader(f, delimiter=";")
|
||||
for line in tqdm(reader, desc="Candidat⋅es", disable=not verbose):
|
||||
line: dict[str, str]
|
||||
numero_panneau = int(line.get("Numéro de panneau"))
|
||||
|
644
nupes/scripts/legislatives2024/import_resultats.py
Normal file
644
nupes/scripts/legislatives2024/import_resultats.py
Normal file
@ -0,0 +1,644 @@
|
||||
import csv
|
||||
|
||||
from sqlalchemy import Engine, select
|
||||
from sqlalchemy.orm import Session
|
||||
from tqdm import tqdm
|
||||
|
||||
from nupes.cache import get_file
|
||||
from nupes.models import BureauVote
|
||||
from nupes.models.legislatives2024 import *
|
||||
|
||||
|
||||
def importer_resultats_bv(engine: Engine, data_round: int = 0, verbose: bool = False) -> None:
|
||||
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/6813fb28-7ec0-42ff-a528-2bc3d82d7dcd"),
|
||||
(2, "https://www.data.gouv.fr/fr/datasets/r/ca974f04-cfd9-4da8-8554-4a868a09c6c2")]
|
||||
|
||||
with Session(engine) as session:
|
||||
candidats = {(candidat.circonscription_id, candidat.numero): candidat
|
||||
for candidat in session.execute(select(CandidatLegislatives2024)).scalars().all()}
|
||||
bv_ids = {(bv.commune_code, bv.code_bureau): bv
|
||||
for bv in session.execute(select(BureauVote)).scalars().all()}
|
||||
|
||||
for tour, file_url in tours:
|
||||
if data_round > 0 and tour != data_round:
|
||||
# On saute ce tour
|
||||
continue
|
||||
|
||||
file = get_file(file_url, f"resultats-legislatives-2024-t{tour}-par-bureau-de-vote.csv")
|
||||
|
||||
with file.open('r') as f:
|
||||
reader = csv.DictReader(f, delimiter=';')
|
||||
with Session(engine) as session:
|
||||
for row in tqdm(reader, desc=f"Bureau de vote tour {tour}", disable=not verbose):
|
||||
dpt_code = row['Code département']
|
||||
|
||||
match dpt_code:
|
||||
case "ZX":
|
||||
dpt_code = "977"
|
||||
|
||||
if dpt_code == "987" or dpt_code == "988":
|
||||
# Les communes de Polynésie française ne sont pas importées
|
||||
continue
|
||||
|
||||
com_code = row['Code commune']
|
||||
|
||||
if com_code == "98601":
|
||||
# 3 royaumes à Wallis-et-Futuna, mais un seul résultat de commune
|
||||
com_code = "98611"
|
||||
|
||||
if com_code.startswith("987"): # Les communes de Polynésie française ne sont pas importées
|
||||
continue
|
||||
|
||||
bv_code = row['Code BV']
|
||||
|
||||
if (com_code, bv_code) not in bv_ids:
|
||||
print(f"Bureau de vote {com_code}_{bv_code} non trouvé")
|
||||
continue
|
||||
bv = bv_ids[(com_code, bv_code)]
|
||||
bv_id = bv.id
|
||||
circo_id = bv.circo_code
|
||||
|
||||
resultats_bv = session.execute(select(ResultatsBureauVoteLegislatives2024)
|
||||
.filter_by(bv_id=bv_id)).scalar_one_or_none()
|
||||
if not resultats_bv:
|
||||
resultats_commune = session.execute(select(ResultatsCommuneLegislatives2024)
|
||||
.filter_by(commune_id=com_code)).scalar_one_or_none()
|
||||
if not resultats_commune:
|
||||
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024)
|
||||
.filter_by(dpt_id=dpt_code)) \
|
||||
.scalar_one_or_none()
|
||||
if not resultats_dpt:
|
||||
dpt = session.execute(select(Departement).filter_by(code_insee=dpt_code)) \
|
||||
.scalar_one()
|
||||
resultats_reg = session.execute(
|
||||
select(ResultatsRegionLegislatives2024).filter_by(region_id=dpt.region_code)) \
|
||||
.scalar_one_or_none()
|
||||
if not resultats_reg:
|
||||
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \
|
||||
.scalar_one_or_none()
|
||||
if not resultats_france:
|
||||
session.add(ResultatsFranceLegislatives2024())
|
||||
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \
|
||||
.scalar_one()
|
||||
|
||||
resultats_reg = ResultatsRegionLegislatives2024(
|
||||
region_id=str(dpt.region_code), resultats_france_id=resultats_france.id)
|
||||
session.add(resultats_reg)
|
||||
resultats_reg = session.execute(select(ResultatsRegionLegislatives2024)
|
||||
.filter_by(region_id=dpt.region_code)).scalar_one()
|
||||
|
||||
resultats_dpt = ResultatsDepartementLegislatives2024(
|
||||
dpt_id=dpt_code, resultats_region_id=resultats_reg.id)
|
||||
session.add(resultats_dpt)
|
||||
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024)
|
||||
.filter_by(dpt_id=dpt_code)).scalar_one()
|
||||
|
||||
resultats_commune = ResultatsCommuneLegislatives2024(commune_id=com_code,
|
||||
resultats_dpt_id=resultats_dpt.id)
|
||||
session.add(resultats_commune)
|
||||
resultats_commune = session.execute(select(ResultatsCommuneLegislatives2024)
|
||||
.filter_by(commune_id=com_code)).scalar_one()
|
||||
|
||||
resultats_circo = session.execute(select(ResultatsCirconscriptionLegislatives2024)
|
||||
.filter_by(circo_id=circo_id)).scalar_one_or_none()
|
||||
|
||||
resultats_bv = ResultatsBureauVoteLegislatives2024(bv_id=bv_id,
|
||||
resultats_commune_id=resultats_commune.id,
|
||||
resultats_circo_id=resultats_circo.id)
|
||||
session.add(resultats_bv)
|
||||
resultats_bv = session.execute(select(ResultatsBureauVoteLegislatives2024)
|
||||
.filter_by(bv_id=bv_id)).scalar_one()
|
||||
|
||||
if tour == 1:
|
||||
resultats_bv.inscrits_t1 = int(row['Inscrits'])
|
||||
resultats_bv.votants_t1 = int(row['Votants'])
|
||||
resultats_bv.abstentions_t1 = int(row['Abstentions'])
|
||||
resultats_bv.exprimes_t1 = int(row['Exprimés'])
|
||||
resultats_bv.blancs_t1 = int(row['Blancs'])
|
||||
resultats_bv.nuls_t1 = int(row['Nuls'])
|
||||
elif tour == 2:
|
||||
resultats_bv.inscrits_t2 = int(row['Inscrits'])
|
||||
resultats_bv.votants_t2 = int(row['Votants'])
|
||||
resultats_bv.abstentions_t2 = int(row['Abstentions'])
|
||||
resultats_bv.exprimes_t2 = int(row['Exprimés'])
|
||||
resultats_bv.blancs_t2 = int(row['Blancs'])
|
||||
resultats_bv.nuls_t2 = int(row['Nuls'])
|
||||
|
||||
for i in range(1, 23):
|
||||
if f'Nuance candidat {i}' not in row:
|
||||
# On a épuisé toutes les nuances
|
||||
break
|
||||
|
||||
numero = row[f'Numéro de panneau {i}']
|
||||
if not numero:
|
||||
continue
|
||||
|
||||
voix = int(row[f'Voix {i}'])
|
||||
|
||||
candidat = candidats[(circo_id, int(numero))]
|
||||
|
||||
voix_candidat_bv = session.execute(select(VoixBureauVoteLegislatives2024)
|
||||
.filter_by(resultats_bureau_vote_id=resultats_bv.id,
|
||||
candidat_id=candidat.id)) \
|
||||
.scalar_one_or_none()
|
||||
if not voix_candidat_bv:
|
||||
voix_candidat_bv = VoixBureauVoteLegislatives2024(
|
||||
resultats_bureau_vote_id=resultats_bv.id, candidat_id=candidat.id)
|
||||
session.add(voix_candidat_bv)
|
||||
|
||||
if tour == 1:
|
||||
voix_candidat_bv.voix_t1 = voix
|
||||
elif tour == 2:
|
||||
voix_candidat_bv.voix_t2 = voix
|
||||
|
||||
session.commit()
|
||||
|
||||
|
||||
def importer_resultats_commune(engine: Engine, data_round: int = 0, verbose: bool = False) -> None:
|
||||
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/bd32fcd3-53df-47ac-bf1d-8d8003fe23a1"),
|
||||
(2, "https://www.data.gouv.fr/fr/datasets/r/5a8088fd-8168-402a-9f40-c48daab88cd1")]
|
||||
|
||||
with Session(engine) as session:
|
||||
for resultats_commune in session.execute(select(ResultatsCommuneLegislatives2024)).scalars().all():
|
||||
if data_round == 0 or data_round == 1:
|
||||
resultats_commune.inscrits_t1 = 0
|
||||
resultats_commune.votants_t1 = 0
|
||||
resultats_commune.abstentions_t1 = 0
|
||||
resultats_commune.exprimes_t1 = 0
|
||||
resultats_commune.blancs_t1 = 0
|
||||
resultats_commune.nuls_t1 = 0
|
||||
if data_round == 0 or data_round == 2:
|
||||
resultats_commune.inscrits_t2 = 0
|
||||
resultats_commune.votants_t2 = 0
|
||||
resultats_commune.abstentions_t2 = 0
|
||||
resultats_commune.exprimes_t2 = 0
|
||||
resultats_commune.blancs_t2 = 0
|
||||
resultats_commune.nuls_t2 = 0
|
||||
|
||||
for voix_nuance in resultats_commune.voix:
|
||||
if data_round == 0 or data_round == 1:
|
||||
voix_nuance.voix_t1 = 0
|
||||
if data_round == 0 or data_round == 2:
|
||||
voix_nuance.voix_t2 = 0
|
||||
|
||||
session.commit()
|
||||
|
||||
com_codes = {commune.code_insee for commune in session.execute(select(Commune)).scalars().all()}
|
||||
|
||||
for tour, file_url in tours:
|
||||
if data_round > 0 and tour != data_round:
|
||||
# On saute ce tour
|
||||
continue
|
||||
|
||||
file = get_file(file_url,
|
||||
f"resultats-legislatives-2024-t{tour}-par-commune.csv")
|
||||
|
||||
with file.open('r') as f:
|
||||
reader = csv.DictReader(f, delimiter=';')
|
||||
with Session(engine) as session:
|
||||
for row in tqdm(reader, desc=f"Commune tour {tour}", disable=not verbose):
|
||||
dpt_code = row['Code département']
|
||||
|
||||
match dpt_code:
|
||||
case "ZX":
|
||||
dpt_code = "977"
|
||||
|
||||
if dpt_code == "987" or dpt_code == "988":
|
||||
# Les communes de Polynésie française ne sont pas importées
|
||||
continue
|
||||
|
||||
com_code = row['Code commune']
|
||||
|
||||
if com_code == "98601":
|
||||
# 3 royaumes à Wallis-et-Futuna, mais un seul résultat de commune
|
||||
com_code = "98611"
|
||||
|
||||
if com_code not in com_codes:
|
||||
print(f"Commune {com_code} non trouvée")
|
||||
continue
|
||||
|
||||
resultats_commune = session.execute(select(ResultatsCommuneLegislatives2024)
|
||||
.filter_by(commune_id=com_code)).scalar_one_or_none()
|
||||
if not resultats_commune:
|
||||
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024)
|
||||
.filter_by(dpt_id=dpt_code)).scalar_one_or_none()
|
||||
if not resultats_dpt:
|
||||
dpt = session.execute(select(Departement)
|
||||
.filter_by(code_insee=f"{dpt_code.zfill(2)}")).scalar_one()
|
||||
resultats_reg = session.execute(select(ResultatsRegionLegislatives2024)
|
||||
.filter_by(region_id=dpt.region_code)).scalar_one_or_none()
|
||||
if not resultats_reg:
|
||||
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \
|
||||
.scalar_one_or_none()
|
||||
if not resultats_france:
|
||||
session.add(ResultatsFranceLegislatives2024())
|
||||
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \
|
||||
.scalar_one()
|
||||
|
||||
resultats_reg = ResultatsRegionLegislatives2024(region_id=str(dpt.region_code),
|
||||
resultats_france_id=resultats_france.id)
|
||||
session.add(resultats_reg)
|
||||
resultats_reg = session.execute(select(ResultatsRegionLegislatives2024)
|
||||
.filter_by(region_id=dpt.region_code)).scalar_one()
|
||||
|
||||
resultats_dpt = ResultatsDepartementLegislatives2024(dpt_id=dpt_code,
|
||||
resultats_region_id=resultats_reg.id)
|
||||
session.add(resultats_dpt)
|
||||
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024)
|
||||
.filter_by(dpt_id=dpt_code)).scalar_one()
|
||||
|
||||
resultats_commune = ResultatsCommuneLegislatives2024(commune_id=com_code,
|
||||
resultats_dpt_id=resultats_dpt.id)
|
||||
session.add(resultats_commune)
|
||||
resultats_commune = session.execute(select(ResultatsCommuneLegislatives2024)
|
||||
.filter_by(commune_id=com_code)).scalar_one()
|
||||
|
||||
if tour == 1:
|
||||
resultats_commune.inscrits_t1 += int(row['Inscrits'])
|
||||
resultats_commune.votants_t1 += int(row['Votants'])
|
||||
resultats_commune.abstentions_t1 += int(row['Abstentions'])
|
||||
resultats_commune.exprimes_t1 += int(row['Exprimés'])
|
||||
resultats_commune.blancs_t1 += int(row['Blancs'])
|
||||
resultats_commune.nuls_t1 += int(row['Nuls'])
|
||||
elif tour == 2:
|
||||
resultats_commune.inscrits_t2 += int(row['Inscrits'])
|
||||
resultats_commune.votants_t2 += int(row['Votants'])
|
||||
resultats_commune.abstentions_t2 += int(row['Abstentions'])
|
||||
resultats_commune.exprimes_t2 += int(row['Exprimés'])
|
||||
resultats_commune.blancs_t2 += int(row['Blancs'])
|
||||
resultats_commune.nuls_t2 += int(row['Nuls'])
|
||||
|
||||
for i in range(1, 23):
|
||||
if f'Nuance candidat {i}' not in row:
|
||||
# On a épuisé toutes les nuances
|
||||
break
|
||||
|
||||
nuance_code = row[f'Nuance candidat {i}']
|
||||
if not nuance_code:
|
||||
continue
|
||||
|
||||
voix = int(row[f'Voix {i}'])
|
||||
|
||||
voix_nuance_commune = session.execute(select(VoixCommuneLegislatives2024)
|
||||
.filter_by(resultats_commune_id=resultats_commune.id,
|
||||
nuance_id=nuance_code)) \
|
||||
.scalar_one_or_none()
|
||||
if not voix_nuance_commune:
|
||||
voix_nuance_commune = VoixCommuneLegislatives2024(resultats_commune_id=resultats_commune.id,
|
||||
nuance_id=nuance_code,
|
||||
voix_t1=0, voix_t2=0)
|
||||
session.add(voix_nuance_commune)
|
||||
|
||||
if tour == 1:
|
||||
voix_nuance_commune.voix_t1 += voix
|
||||
elif tour == 2:
|
||||
voix_nuance_commune.voix_t2 += voix
|
||||
|
||||
session.commit()
|
||||
|
||||
|
||||
def importer_resultats_circo(engine: Engine, data_round: int = 0, verbose: bool = False) -> None:
|
||||
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/5163f2e3-1362-4c35-89a0-1934bb74f2d9"),
|
||||
(2, "https://www.data.gouv.fr/fr/datasets/r/41ed46cd-77c2-4ecc-b8eb-374aa953ca39")]
|
||||
|
||||
for tour, file_url in tours:
|
||||
if data_round > 0 and tour != data_round:
|
||||
# On saute ce tour
|
||||
continue
|
||||
|
||||
file = get_file(file_url, f"resultats-legislatives-2024-t{tour}-par-circonscription.csv")
|
||||
|
||||
with file.open('r') as f:
|
||||
reader = csv.DictReader(f, delimiter=';')
|
||||
with Session(engine) as session:
|
||||
for row in tqdm(reader, desc=f"Circonscription tour {tour}", disable=not verbose):
|
||||
dpt_code = row['Code département']
|
||||
circo_code = row['Code circonscription législative'][-2:]
|
||||
|
||||
match dpt_code:
|
||||
case "ZX":
|
||||
dpt_code = "977"
|
||||
|
||||
circo_id = f"{dpt_code.zfill(2)}-{circo_code.zfill(2)}"
|
||||
|
||||
resultats_circo = session.execute(select(ResultatsCirconscriptionLegislatives2024)
|
||||
.filter_by(circo_id=circo_id)).scalar_one_or_none()
|
||||
if not resultats_circo:
|
||||
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024)
|
||||
.filter_by(dpt_id=dpt_code)).scalar_one_or_none()
|
||||
if not resultats_dpt:
|
||||
dpt = session.execute(select(Departement).filter_by(code_insee=f"{dpt_code.zfill(2)}")) \
|
||||
.scalar_one()
|
||||
resultats_reg = session.execute(select(ResultatsRegionLegislatives2024)
|
||||
.filter_by(region_id=dpt.region_code)) \
|
||||
.scalar_one_or_none()
|
||||
if not resultats_reg:
|
||||
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \
|
||||
.scalar_one_or_none()
|
||||
if not resultats_france:
|
||||
session.add(ResultatsFranceLegislatives2024())
|
||||
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one()
|
||||
|
||||
resultats_reg = ResultatsRegionLegislatives2024(region_id=str(dpt.region_code),
|
||||
resultats_france_id=resultats_france.id)
|
||||
session.add(resultats_reg)
|
||||
resultats_reg = session.execute(select(ResultatsRegionLegislatives2024)
|
||||
.filter_by(region_id=dpt.region_code)).scalar_one()
|
||||
|
||||
resultats_dpt = ResultatsDepartementLegislatives2024(dpt_id=dpt_code,
|
||||
resultats_region_id=resultats_reg.id)
|
||||
session.add(resultats_dpt)
|
||||
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024)
|
||||
.filter_by(dpt_id=dpt_code)).scalar_one()
|
||||
|
||||
resultats_circo = ResultatsCirconscriptionLegislatives2024(
|
||||
circo_id=circo_id, resultats_departement_id=resultats_dpt.id)
|
||||
session.add(resultats_circo)
|
||||
resultats_circo = session.execute(select(ResultatsCirconscriptionLegislatives2024)
|
||||
.filter_by(circo_id=circo_id)).scalar_one()
|
||||
|
||||
if tour == 1:
|
||||
resultats_circo.inscrits_t1 = int(row['Inscrits'])
|
||||
resultats_circo.votants_t1 = int(row['Votants'])
|
||||
resultats_circo.abstentions_t1 = int(row['Abstentions'])
|
||||
resultats_circo.exprimes_t1 = int(row['Exprimés'])
|
||||
resultats_circo.blancs_t1 = int(row['Blancs'])
|
||||
resultats_circo.nuls_t1 = int(row['Nuls'])
|
||||
elif tour == 2:
|
||||
resultats_circo.inscrits_t2 = int(row['Inscrits'])
|
||||
resultats_circo.votants_t2 = int(row['Votants'])
|
||||
resultats_circo.abstentions_t2 = int(row['Abstentions'])
|
||||
resultats_circo.exprimes_t2 = int(row['Exprimés'])
|
||||
resultats_circo.blancs_t2 = int(row['Blancs'])
|
||||
resultats_circo.nuls_t2 = int(row['Nuls'])
|
||||
|
||||
for i in range(1, 23):
|
||||
if f'Nuance candidat {i}' not in row:
|
||||
# On a épuisé toutes les nuances
|
||||
break
|
||||
|
||||
numero = row[f'Numéro de panneau {i}']
|
||||
if not numero:
|
||||
continue
|
||||
|
||||
voix = int(row[f'Voix {i}'])
|
||||
|
||||
candidat = session.execute(select(CandidatLegislatives2024)
|
||||
.filter_by(circonscription_id=circo_id, numero=numero)) \
|
||||
.scalar_one_or_none()
|
||||
|
||||
voix_candidat_circo = session.execute(select(VoixCirconscriptionLegislatives2024)
|
||||
.filter_by(resultats_circonscription_id=resultats_circo.id,
|
||||
candidat_id=candidat.id)) \
|
||||
.scalar_one_or_none()
|
||||
if not voix_candidat_circo:
|
||||
voix_candidat_circo = VoixCirconscriptionLegislatives2024(
|
||||
resultats_circonscription_id=resultats_circo.id, candidat_id=candidat.id)
|
||||
session.add(voix_candidat_circo)
|
||||
|
||||
if tour == 1:
|
||||
voix_candidat_circo.voix_t1 = voix
|
||||
elif tour == 2:
|
||||
voix_candidat_circo.voix_t2 = voix
|
||||
|
||||
session.commit()
|
||||
|
||||
|
||||
def importer_resultats_departement(engine: Engine, data_round: int = 0, verbose: bool = False) -> None:
|
||||
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/78c708c5-5bc5-438d-8379-f432beae3f2b"),
|
||||
(2, "https://www.data.gouv.fr/fr/datasets/r/8d4a6927-c96f-4cf5-b757-ea745eca26bd")]
|
||||
|
||||
for tour, file_url in tours:
|
||||
if data_round > 0 and tour != data_round:
|
||||
# On saute ce tour
|
||||
continue
|
||||
|
||||
file = get_file(file_url, f"resultats-legislatives-2024-t{tour}-par-departement.csv")
|
||||
|
||||
with file.open('r') as f:
|
||||
reader = csv.DictReader(f, delimiter=';')
|
||||
with Session(engine) as session:
|
||||
for row in tqdm(reader, desc=f"Département tour {tour}", disable=not verbose):
|
||||
dpt_code = row['Code département']
|
||||
|
||||
match dpt_code:
|
||||
case "ZX":
|
||||
dpt_code = "977"
|
||||
|
||||
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024)
|
||||
.filter_by(dpt_id=dpt_code)).scalar_one_or_none()
|
||||
if not resultats_dpt:
|
||||
dpt = session.execute(select(Departement)
|
||||
.filter_by(code_insee=f"{dpt_code.zfill(2)}")) \
|
||||
.scalar_one()
|
||||
resultats_reg = session.execute(select(ResultatsRegionLegislatives2024)
|
||||
.filter_by(region_id=dpt.region_code)) \
|
||||
.scalar_one_or_none()
|
||||
if not resultats_reg:
|
||||
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \
|
||||
.scalar_one_or_none()
|
||||
if not resultats_france:
|
||||
session.add(ResultatsFranceLegislatives2024())
|
||||
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one()
|
||||
|
||||
resultats_reg = ResultatsRegionLegislatives2024(region_id=str(dpt.region_code),
|
||||
resultats_france_id=resultats_france.id)
|
||||
session.add(resultats_reg)
|
||||
resultats_reg = session.execute(select(ResultatsRegionLegislatives2024)
|
||||
.filter_by(region_id=dpt.region_code)).scalar_one()
|
||||
|
||||
resultats_dpt = ResultatsDepartementLegislatives2024(dpt_id=dpt_code,
|
||||
resultats_region_id=resultats_reg.id)
|
||||
session.add(resultats_dpt)
|
||||
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024)
|
||||
.filter_by(dpt_id=dpt_code)).scalar_one()
|
||||
|
||||
if tour == 1:
|
||||
resultats_dpt.inscrits_t1 = int(row['Inscrits'])
|
||||
resultats_dpt.votants_t1 = int(row['Votants'])
|
||||
resultats_dpt.abstentions_t1 = int(row['Abstentions'])
|
||||
resultats_dpt.exprimes_t1 = int(row['Exprimés'])
|
||||
resultats_dpt.blancs_t1 = int(row['Blancs'])
|
||||
resultats_dpt.nuls_t1 = int(row['Nuls'])
|
||||
elif tour == 2:
|
||||
resultats_dpt.inscrits_t2 = int(row['Inscrits'])
|
||||
resultats_dpt.votants_t2 = int(row['Votants'])
|
||||
resultats_dpt.abstentions_t2 = int(row['Abstentions'])
|
||||
resultats_dpt.exprimes_t2 = int(row['Exprimés'])
|
||||
resultats_dpt.blancs_t2 = int(row['Blancs'])
|
||||
resultats_dpt.nuls_t2 = int(row['Nuls'])
|
||||
|
||||
for i in range(1, 23):
|
||||
if f'Nuance candidat {i}' not in row:
|
||||
# On a épuisé toutes les nuances
|
||||
break
|
||||
|
||||
nuance_code = row[f'Nuance candidat {i}']
|
||||
if not nuance_code:
|
||||
continue
|
||||
|
||||
voix = int(row[f'Voix {i}'])
|
||||
|
||||
voix_nuance_dpt = session.execute(select(VoixDepartementLegislatives2024)
|
||||
.filter_by(resultats_departement_id=resultats_dpt.id,
|
||||
nuance_id=nuance_code)) \
|
||||
.scalar_one_or_none()
|
||||
if not voix_nuance_dpt:
|
||||
voix_nuance_dpt = VoixDepartementLegislatives2024(resultats_departement_id=resultats_dpt.id,
|
||||
nuance_id=nuance_code)
|
||||
session.add(voix_nuance_dpt)
|
||||
|
||||
if tour == 1:
|
||||
voix_nuance_dpt.voix_t1 = voix
|
||||
elif tour == 2:
|
||||
voix_nuance_dpt.voix_t2 = voix
|
||||
|
||||
session.commit()
|
||||
|
||||
|
||||
def importer_resultats_region(engine: Engine, data_round: int = 0, verbose: bool = False) -> None:
|
||||
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/f69ffab7-fe37-494e-ad6d-a7cfc80ddc1f"),
|
||||
(2, "https://www.data.gouv.fr/fr/datasets/r/ac4e272a-3ce2-4f20-941e-fb0ded444a5d")]
|
||||
|
||||
for tour, file_url in tours:
|
||||
if data_round > 0 and tour != data_round:
|
||||
# On saute ce tour
|
||||
continue
|
||||
|
||||
file = get_file(file_url, f"resultats-legislatives-2024-t{tour}-par-region.csv")
|
||||
|
||||
with file.open('r') as f:
|
||||
reader = csv.DictReader(f, delimiter=';')
|
||||
|
||||
with Session(engine) as session:
|
||||
for row in tqdm(reader, desc=f"Région tour {tour}", disable=not verbose):
|
||||
reg_code = row['Code région']
|
||||
|
||||
resultats_reg = session.execute(select(ResultatsRegionLegislatives2024)
|
||||
.filter_by(region_id=reg_code)).scalar_one_or_none()
|
||||
if not resultats_reg:
|
||||
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one_or_none()
|
||||
if not resultats_france:
|
||||
session.add(ResultatsFranceLegislatives2024())
|
||||
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one()
|
||||
|
||||
resultats_reg = ResultatsRegionLegislatives2024(region_id=reg_code,
|
||||
resultats_france_id=resultats_france.id)
|
||||
session.add(resultats_reg)
|
||||
session.commit()
|
||||
resultats_reg = session.execute(select(ResultatsRegionLegislatives2024)
|
||||
.filter_by(region_id=reg_code)).scalar_one()
|
||||
|
||||
if tour == 1:
|
||||
resultats_reg.inscrits_t1 = int(row['Inscrits'])
|
||||
resultats_reg.votants_t1 = int(row['Votants'])
|
||||
resultats_reg.abstentions_t1 = int(row['Abstentions'])
|
||||
resultats_reg.exprimes_t1 = int(row['Exprimés'])
|
||||
resultats_reg.blancs_t1 = int(row['Blancs'])
|
||||
resultats_reg.nuls_t1 = int(row['Nuls'])
|
||||
elif tour == 2:
|
||||
resultats_reg.inscrits_t2 = int(row['Inscrits'])
|
||||
resultats_reg.votants_t2 = int(row['Votants'])
|
||||
resultats_reg.abstentions_t2 = int(row['Abstentions'])
|
||||
resultats_reg.exprimes_t2 = int(row['Exprimés'])
|
||||
resultats_reg.blancs_t2 = int(row['Blancs'])
|
||||
resultats_reg.nuls_t2 = int(row['Nuls'])
|
||||
|
||||
for i in range(1, 23):
|
||||
if f'Nuance candidat {i}' not in row:
|
||||
# On a épuisé toutes les nuances
|
||||
break
|
||||
|
||||
nuance_code = row[f'Nuance candidat {i}']
|
||||
if not nuance_code:
|
||||
continue
|
||||
|
||||
voix = int(row[f'Voix {i}'])
|
||||
|
||||
voix_nuance_reg = session.execute(select(VoixRegionLegislatives2024)
|
||||
.filter_by(resultats_region_id=resultats_reg.id,
|
||||
nuance_id=nuance_code)) \
|
||||
.scalar_one_or_none()
|
||||
if not voix_nuance_reg:
|
||||
voix_nuance_reg = VoixRegionLegislatives2024(resultats_region_id=resultats_reg.id,
|
||||
nuance_id=nuance_code)
|
||||
session.add(voix_nuance_reg)
|
||||
|
||||
if tour == 1:
|
||||
voix_nuance_reg.voix_t1 = voix
|
||||
elif tour == 2:
|
||||
voix_nuance_reg.voix_t2 = voix
|
||||
|
||||
session.commit()
|
||||
|
||||
|
||||
def importer_resultats_france(engine: Engine, data_round: int = 0, verbose: bool = False) -> None:
|
||||
# Erreur avec les données France Entière du 1er tour, qui sera j'espère corrigée un jour
|
||||
tours = [# (1, "https://www.data.gouv.fr/fr/datasets/r/386fd5ac-e7f1-4e0f-8929-12d2c5391081"),
|
||||
(2, "https://www.data.gouv.fr/fr/datasets/r/f64d2781-4ffa-414b-8f6d-e9d849f2f5e7")]
|
||||
for tour, file_url in tours:
|
||||
if data_round > 0 and tour != data_round:
|
||||
# On saute ce tour
|
||||
continue
|
||||
|
||||
file = get_file(file_url, f"resultats-legislatives-2024-t{tour}-france-entiere.csv")
|
||||
|
||||
with file.open('r') as f:
|
||||
reader = csv.DictReader(f, delimiter=';')
|
||||
|
||||
with Session(engine) as session:
|
||||
for row in tqdm(reader, desc=f"France Entière tour {tour}", disable=not verbose):
|
||||
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one_or_none()
|
||||
|
||||
if not resultats_france:
|
||||
session.add(ResultatsFranceLegislatives2024())
|
||||
session.commit()
|
||||
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one()
|
||||
|
||||
if tour == 1:
|
||||
resultats_france.inscrits_t1 = int(row['Inscrits'])
|
||||
resultats_france.votants_t1 = int(row['Votants'])
|
||||
resultats_france.abstentions_t1 = int(row['Abstentions'])
|
||||
resultats_france.exprimes_t1 = int(row['Exprimés'])
|
||||
resultats_france.blancs_t1 = int(row['Blancs'])
|
||||
resultats_france.nuls_t1 = int(row['Nuls'])
|
||||
elif tour == 2:
|
||||
resultats_france.inscrits_t2 = int(row['Inscrits'])
|
||||
resultats_france.votants_t2 = int(row['Votants'])
|
||||
resultats_france.abstentions_t2 = int(row['Abstentions'])
|
||||
resultats_france.exprimes_t2 = int(row['Exprimés'])
|
||||
resultats_france.blancs_t2 = int(row['Blancs'])
|
||||
resultats_france.nuls_t2 = int(row['Nuls'])
|
||||
|
||||
for i in range(1, 23):
|
||||
if f'Nuance candidat {i}' not in row:
|
||||
# On a épuisé toutes les nuances
|
||||
break
|
||||
|
||||
nuance_code = row[f'Nuance candidat {i}']
|
||||
voix = int(row[f'Voix {i}'])
|
||||
|
||||
voix_nuance_france = session.execute(select(VoixFranceLegislatives2024)
|
||||
.filter_by(resultats_france_id=resultats_france.id,
|
||||
nuance_id=nuance_code)) \
|
||||
.scalar_one_or_none()
|
||||
if not voix_nuance_france:
|
||||
voix_nuance_france = VoixFranceLegislatives2024(resultats_france_id=resultats_france.id,
|
||||
nuance_id=nuance_code)
|
||||
session.add(voix_nuance_france)
|
||||
|
||||
if tour == 1:
|
||||
voix_nuance_france.voix_t1 = voix
|
||||
elif tour == 2:
|
||||
voix_nuance_france.voix_t2 = voix
|
||||
|
||||
session.commit()
|
||||
|
||||
|
||||
def run(engine: Engine, data_round: int = 1, verbose: bool = False) -> None:
|
||||
importer_resultats_france(engine, data_round, verbose)
|
||||
importer_resultats_region(engine, data_round, verbose)
|
||||
importer_resultats_departement(engine, data_round, verbose)
|
||||
importer_resultats_circo(engine, data_round, verbose)
|
||||
importer_resultats_commune(engine, data_round, verbose)
|
||||
importer_resultats_bv(engine, data_round, verbose)
|
Reference in New Issue
Block a user