84 lines
3.4 KiB
Python
Executable File
84 lines
3.4 KiB
Python
Executable File
"""Copyright (c) 2020 Sdj Geek
|
|
|
|
Voir le fichier LICENSE
|
|
------------------------------------------------------------------------------------------------------------------------
|
|
|
|
Ce programme permet d'insérer les fichiers fournis par l'INSEE dans
|
|
une base de donnée SQLite.
|
|
|
|
Les fichiers fournis par l'INSEE sont disponibles en open data à
|
|
[cette
|
|
adresse](https://www.data.gouv.fr/fr/datasets/fichier-des-personnes-decedees/#_). Leur
|
|
formatage n'est pas très optimisé pour une utilisation dans un
|
|
programme. L'objet de ce programme est de transformer ces fichiers en
|
|
une base de donnée SDLite.
|
|
|
|
------------------------------------------------------------------------------------------------------------------------
|
|
|
|
Auteur : SDJ GeeK <sdjgeek@protonmail.com>
|
|
Date : 7 juin 2020
|
|
|
|
"""
|
|
|
|
import argparse
|
|
import glob
|
|
import os
|
|
import peewee
|
|
|
|
from bdd_insee import BddInsee
|
|
|
|
|
|
def parse_insee_file(data_file_path):
|
|
"""Parse un fichier de l'INSEE
|
|
|
|
:param data_file_path: chemin vers le fichier à parser
|
|
:returns: liste de dictionnaires contenant les informations à insérer
|
|
|
|
"""
|
|
data_list = []
|
|
with open(data_file_path, 'r', encoding='latin1') as data_file:
|
|
for line in data_file.readlines():
|
|
try:
|
|
last_name, first_name = line[0:80].strip()[:-1].split("*")
|
|
data_list.append({'last_name': last_name.upper(),
|
|
'first_name': first_name.upper(),
|
|
'is_woman': (line[80] == "2"),
|
|
'annee_naissance': int(line[81:85]),
|
|
'mois_naissance': int(line[85:87]),
|
|
'jour_naissance': int(line[87:89]),
|
|
'code_lieu_naissance': line[89:94].upper(),
|
|
'commune_naissance': line[94: 124].strip().upper(),
|
|
'pays_naissance': line[124: 154].strip().upper(),
|
|
'annee_deces': int(line[154:158]),
|
|
'mois_deces': int(line[158:160]),
|
|
'jour_deces': int(line[160:162]),
|
|
'code_lieu_deces': line[162:167].upper(),
|
|
'numero_act_deces': line[167:].strip().upper()})
|
|
except ValueError:
|
|
print("Erreur pour parser ligne :")
|
|
print(line)
|
|
continue
|
|
return data_list
|
|
|
|
|
|
def completer_bdd_insee(chemin_base_donnees, chemin_repertoire_donnees):
|
|
"""Ajouter des données dans la base
|
|
|
|
:param chemin_base_donnees: chemin vers le fichier SQLite
|
|
:param chemin_repertoire_donnees: chemin vers le répertoire contenant les fichiers de l'INSEE
|
|
|
|
"""
|
|
base_insee = BddInsee(chemin_base_donnees)
|
|
for file_path in sorted(glob.glob(os.path.join(chemin_repertoire_donnees, "deces-*.txt"))):
|
|
print(f"Import du fichier {file_path}")
|
|
data_list = parse_insee_file(file_path)
|
|
base_insee.import_data_list(data_list)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("bdd_path", type=str, help="Chemin vers le fichier de base de donnée")
|
|
parser.add_argument("input_dir_path", type=str, help="Chemin vers le répertoire contenant les fichiers de l'INSEE")
|
|
args = parser.parse_args()
|
|
completer_bdd_insee(args.bdd_path, args.input_dir_path)
|