Source code for bradata.tse.candidatos
from bradata.connection import Connection
from bradata.utils import _must_contain, _treat_inputs, _unzip, _set_download_directory
from bradata.tse.utils_tse import unzip_tse, aggregate_tse, download_headers
import time
import os
[docs]class Candidatos:
"""
Download, organize and pre-process candidatos data from TSE
http://www.tse.jus.br/eleicoes/estatisticas/repositorio-de-dados-eleitorais
"""
[docs] def download(self, type=None, year=None):
"""
Download a certain type of data from a year in the Candidatos option
You can also get several years or types, just pass a list
Types can be:
- candidatos
- bens
- legendas
- vagas
This method covers the following years: 2016, 2014
So, to download candidatos data from 2014, just put download(type='candidatos', ano=2015)
Args:
type: str or list with the type of the data
year: str or int or list with a year
Returns: Saves data to a local data file as ../bradata/tse/[state]/candidatos_[type]_[year].csv
"""
_must_contain({'type': type, 'year': year}, ['type', 'year']) # raises error if keywords do not exist
if not isinstance(type, list):
type = [type]
if not isinstance(year, list):
year = [year]
conn = Connection()
download_headers()
for t in type:
if t == 'candidatos':
base_url = "http://agencia.tse.jus.br/estatistica/sead/odsele/consulta_cand/consulta_cand_"
elif t == 'bens':
base_url = "http://agencia.tse.jus.br/estatistica/sead/odsele/bem_candidato/bem_candidato_"
elif t == 'legendas':
base_url = "http://agencia.tse.jus.br/estatistica/sead/odsele/consulta_legendas/consulta_legendas_"
elif t == 'vagas':
base_url = "http://agencia.tse.jus.br/estatistica/sead/odsele/consulta_vagas/consulta_vagas_"
else:
print(t)
raise Exception('Type should be candidatos, bens, legendas or vagas')
print('Type: ', t)
for y in year:
print('Year: ', y)
url = base_url + _treat_inputs(y) + '.zip' # treat_inputs turn int into str, raises error if diff type
print('Downloading\n')
time0 = time.time()
result = conn.perform_request(url, binary=True)
if result['status'] == 'ok':
result = result['content']
else:
print('File was not dowloaded')
continue
print('Download Time: ', time.time() - time0)
print('Unzip')
unzip_tse(result, _set_download_directory() + '/tse/' + t)
print('Aggregate')
aggregate_tse(_set_download_directory() + '/tse/', t, y)
print('Done')
print('Finished')