Commit 5f25a6ff authored by Emmanuel Raviart's avatar Emmanuel Raviart

New script to harvest nosdeputes.fr & nossenateurs.fr using their APIs.

parent 022dc3f5
#! /usr/bin/env python3
# Eliasse-Daemon -- Récupère en permanence les amendements depuis Eliasse.
# By: Emmanuel Raviart <emmanuel@raviart.com>
#
# Copyright (C) 2018 Paula Forteza & Emmanuel Raviart
# https://framagit.org/parlement-ouvert/prendlaloi
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Attend sur Eliasse les actualités sur les amendements et met à jour les fichiers JSON correspondants"""
import argparse
import json
import logging
import os
import shutil
import sys
import tempfile
import time
import requests
app_name = os.path.splitext(os.path.basename(__file__))[0]
legislature = 15
log = logging.getLogger(app_name)
target_dir = None
def main():
parser = argparse.ArgumentParser()
parser.add_argument('-v', '--verbose', action='store_true',
default=False, help='increase output verbosity')
parser.add_argument(
'target_dir',
help='path of target directory containing JSON files',
)
args = parser.parse_args()
global target_dir
target_dir = args.target_dir
logging.basicConfig(
level=logging.INFO if args.verbose else logging.WARNING, stream=sys.stdout)
for type_parlementaire in ('depute', 'senateur'):
response = requests.get(
'https://www.nos{0}s.fr/{0}s/json'.format(type_parlementaire))
response.raise_for_status()
data = response.json()
assert sorted(data.keys()) == ['{}s'.format(type_parlementaire)]
parlementaires = data['{}s'.format(type_parlementaire)]
assert all(
sorted(parlementaire.keys()) == [type_parlementaire]
for parlementaire in parlementaires
)
parlementaires = [
parlementaire[type_parlementaire]
for parlementaire in parlementaires
]
write_json(parlementaires, os.path.join(target_dir, 'nos{}s'.format(
type_parlementaire), '{}s.json'.format(type_parlementaire)))
for parlementaire in parlementaires:
response = requests.get(
'https://www.nos{}s.fr/{}/json'.format(type_parlementaire, parlementaire['slug']))
response.raise_for_status()
data = response.json()
assert sorted(data.keys()) == [type_parlementaire]
parlementaire_complet = data[type_parlementaire]
write_json(parlementaire_complet, os.path.join(target_dir, 'nos{}s'.format(
type_parlementaire), '{}s'.format(type_parlementaire), '{}.json'.format(parlementaire['slug'])))
response = requests.get(
'https://www.nos{0}s.fr/{0}/photo/{1}'.format(type_parlementaire, parlementaire['slug']))
response.raise_for_status()
write_bytes(response.content, os.path.join(target_dir, 'nos{}s'.format(
type_parlementaire), '{}s'.format(type_parlementaire), '{}.png'.format(parlementaire['slug'])))
for type_parlementaire in ('depute', 'senateur'):
slugs_organismes = []
for type_organisme, slug_organisme in (
('groupe', 'groupes-parlementaires'),
('parlementaire', 'organes-parlementaires'),
('extra', 'organes-extra-parlementaires'),
('groupes', 'groupes-etudes-et-amities'),
):
response = requests.get(
'https://www.nos{}s.fr/organismes/{}/json'.format(type_parlementaire, type_organisme))
response.raise_for_status()
data = response.json()
assert sorted(data.keys()) == ['organismes']
organismes = data['organismes']
assert all(
sorted(organisme.keys()) == ['organisme']
for organisme in organismes
)
organismes = [
organisme['organisme']
for organisme in organismes
]
for organisme in organismes:
slugs_organismes.append(organisme['slug'])
write_json(organismes, os.path.join(
target_dir, 'nos{}s'.format(type_parlementaire), '{}.json'.format(slug_organisme)))
slugs_parlementaires_par_slugs_organisme = {}
for slug_organisme in slugs_organismes:
response = requests.get(
'https://www.nos{}s.fr/organisme/{}/json?includePast=true'.format(type_parlementaire, slug_organisme))
response.raise_for_status()
data = response.json()
parlementaires = data['{}s'.format(type_parlementaire)]
assert all(
sorted(parlementaire.keys()) == [type_parlementaire]
for parlementaire in parlementaires
)
parlementaires = [
parlementaire[type_parlementaire]
for parlementaire in parlementaires
]
for parlementaire in parlementaires:
slugs_parlementaires_par_slugs_organisme.setdefault(
slug_organisme, []).append(parlementaire['slug'])
write_json(slugs_parlementaires_par_slugs_organisme, os.path.join(
target_dir, 'nos{}s'.format(type_parlementaire), 'membres.json'))
return 0
def write_bytes(data, file_path):
dir = os.path.dirname(file_path)
os.makedirs(dir, exist_ok=True)
if os.path.exists(file_path):
with open(file_path, 'rb') as data_file:
old_data = data_file.read()
else:
old_data = None
if data != old_data:
_, temp_file_path = tempfile.mkstemp(
suffix='.data', prefix='nosparlementaires-', text=False)
with open(temp_file_path, 'wb') as data_file:
data_file.write(data)
shutil.move(temp_file_path, file_path)
return True
return False
def write_json(data, file_path):
dir = os.path.dirname(file_path)
os.makedirs(dir, exist_ok=True)
new_text = json.dumps(data, ensure_ascii=False, indent=2, sort_keys=True)
if os.path.exists(file_path):
with open(file_path, 'r', encoding='utf-8') as json_file:
old_text = json_file.read()
else:
old_text = None
if new_text != old_text:
_, temp_file_path = tempfile.mkstemp(
suffix='.json', prefix='nosparlementaires-', text=True)
with open(temp_file_path, 'w', encoding='utf-8') as json_file:
json_file.write(new_text)
shutil.move(temp_file_path, file_path)
return True
return False
if __name__ == '__main__':
sys.exit(main())
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment