Skip to content

Instantly share code, notes, and snippets.

@danilovazb

danilovazb/ReceitaFederal.py

Created January 9, 2015 13:19
Show Gist options
  • Save danilovazb/4311fbca103f2e3f88bf to your computer and use it in GitHub Desktop.
Save danilovazb/4311fbca103f2e3f88bf to your computer and use it in GitHub Desktop.
Download ZIP
Raw
ReceitaFederal.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
##########################################################################
# ██████╗ ██████╗ ███╗ ██╗███████╗██╗ ██╗██╗ ████████╗ █████╗
# ██╔════╝██╔═══██╗████╗ ██║██╔════╝██║ ██║██║ ╚══██╔══╝██╔══██╗
# ██║ ██║ ██║██╔██╗ ██║███████╗██║ ██║██║ ██║ ███████║
# ██║ ██║ ██║██║╚██╗██║╚════██║██║ ██║██║ ██║ ██╔══██║
# ╚██████╗╚██████╔╝██║ ╚████║███████║╚██████╔╝███████╗██║ ██║ ██║
# ╚═════╝ ╚═════╝ ╚═╝ ╚═══╝╚══════╝ ╚═════╝ ╚══════╝╚═╝ ╚═╝ ╚═╝
# ██╗██████╗ ██████╗ ███████╗
# ██║██╔══██╗██╔══██╗██╔════╝
# ██║██████╔╝██████╔╝█████╗
# ██║██╔══██╗██╔═══╝ ██╔══╝
# ██║██║ ██║██║ ██║
# ╚═╝╚═╝ ╚═╝╚═╝ ╚═╝ V. 2.0.2
#
# 15/10/2014
# -------------------------------------------------------------------
# Sistema de consulta em massa de imposto de renda por arquivo.
#
# ~~17/11/2014~~
# - Adicionado o arg para receber o nome do arquivo a ser salvo
#
# ~~13/11/2014~~
# - Novo IF para separar mais um status "utilize o Extrato do IRPF."
#
# ~~15/10/2014~~
# - Adicionado status padronizados
# - Thread ao processar o arquivo
# - Log de erro e cpfs não processados
#
# -------------------------------------------------------------------
#
# Modo de uso:
# ~# python consultaIR.py arquivoCPF.txt
#
# Nome de retorno dos arquivos:
# ~> Sucesso: contribuintes-24102014.csv
# ~> LOG de ERRO: ERRO-contribuintes-24102014.log
# ~> CPFs a REFAZER: Refazer-contribuintes-24102014.log
#
##########################################################################
import threading
import time
import urllib2,sys,json,requests,re
import pycurl
from bs4 import BeautifulSoup
from itertools import islice
from datetime import datetime
# Máximo de conexões/threads simultâneas
MAX_CONEXOES = 50
# Função para imprimir uma linha por vez via lock
print_lock = threading.Lock()
try:
arquivoArgv = sys.argv[1]
nomearquivo = sys.argv[2]
except IndexError:
print "Modo de uso:\n ~# python consultaIR.py arquivoCPF.txt nomeArquivo"
exit()
def mostrar_msg(msg):
print_lock.acquire()
print msg
print_lock.release()
mostrar_msg("\n[+] Processando arquivo com um total de "+str(MAX_CONEXOES)+" threads.\n[+]Aguarde o fim do processamento.")
# Função para cada thread
def consultar_cpf(cpf):
try:
cpfConsulta = cpf.strip()
url = "http://www.receita.fazenda.gov.br/aplicacoes/atrjo/consrest/atual.app/paginas/view/restituicao.asp"
cookie = {'constRest':'1767330705','nova_visita_ano':'1db9c46e-f7f7-ac5b-d09e-0decf4979218','ASPSESSIONIDACASCATR':'CNJJJANCOFLNDBABPCGMJFFF'}
payload = {'senha': '', 'idSom': '', 'valid': '1', 'CPF': cpfConsulta,'exercicio': '2014', 'Submit': 'Consultar'}
headers = {'Referer': url}
response = requests.post(url, cookies=cookie, data=payload, headers=headers)
data = response.text
data = data.split('<hr width="95%">')[0]
soup = BeautifulSoup(data)
contribuinte = {'dados': '','nome':'', 'cpf': '','infoDisp': '','dataDisp': ''}
for txt in soup.select("td font.txtDadosContribuinte"):
contribuinte['dados'] += txt.get_text().encode('utf8').strip() + '|'
for txt in soup.select("font.txtNomeContribuinte"):
contribuinte['nome'] += txt.get_text().encode('utf8').strip() + '|'
for txt in soup.select("td font.txtInfBancoContribuinte"):
contribuinte['infoDisp'] += txt.get_text().encode('utf8').strip() + '|'
for txt in soup.select("td font.txtInfBancoContribuinte2"):
contribuinte['dataDisp'] += txt.get_text().encode('utf8').strip() + '|'
txt_contr = soup.find('label',{'class': 'txtCpfContribuinte'}).get_text().encode('utf8').strip()
cpf = re.findall("\d{3}\.\d{3}\.\d{3}\-\d{2}",txt_contr)[0]
contribuinte['cpf'] = cpf
now = datetime.now()
arquivoContrib = nomearquivo + "-" + str(now.day) + str(now.month) + str(now.year) + ".csv"
arquivoGrava = open(arquivoContrib,'a')
qtd_dados = len(contribuinte['dados'].split('|'))
qtd_dataDisponivel = len(contribuinte['dataDisp'].split('|'))
if 'Creditada' in contribuinte['dados']:
if len(contribuinte['dataDisp'].split('|')) > 1:
arquivoGrava.write("%s|%s|%s|%s|%s|%s|%s|%s|%s" % \
(contribuinte['cpf'].replace('.','').replace('-',''),
contribuinte['nome'].strip().split('|')[0],
contribuinte['dados'].split('|')[int(qtd_dados)-3].strip(),
contribuinte['dados'].split('|')[int(qtd_dados)-2].strip(),
"Creditada",
contribuinte['dataDisp'].split('|')[0].strip(),
contribuinte['dataDisp'].split('|')[1].strip(),
contribuinte['dataDisp'].split('|')[2].strip(),
contribuinte['dataDisp'].split('|')[3].strip()) + "\n")
else:
arquivoGrava.write("%s|%s|%s|%s|%s|%s|%s|%s|%s" % \
(contribuinte['cpf'].replace('.','').replace('-',''),
contribuinte['nome'].strip().split('|')[0],
contribuinte['dados'].split('|')[int(qtd_dados)-3].strip(),
contribuinte['dados'].split('|')[int(qtd_dados)-2].strip(),
"Creditada",
contribuinte['dataDisp'].split('|')[0].strip(),
contribuinte['dataDisp'].strip(),
contribuinte['dataDisp'].strip(),
contribuinte['dataDisp'].strip()) + "\n")
elif 'Sua declaração não consta na base' in contribuinte['dados']:
if len(contribuinte['dataDisp'].split('|')) > 1:
arquivoGrava.write("%s|%s|%s|%s|%s|%s|%s|%s|%s" % \
(contribuinte['cpf'].replace('.','').replace('-',''),
contribuinte['nome'].strip().split('|')[0],
contribuinte['dados'].split('|')[int(qtd_dados)-3].strip(),
contribuinte['dados'].split('|')[int(qtd_dados)-2].strip(),
"Não Declarante",
contribuinte['dataDisp'].split('|')[0].strip(),
contribuinte['dataDisp'].split('|')[1].strip(),
contribuinte['dataDisp'].split('|')[2].strip(),
contribuinte['dataDisp'].split('|')[3].strip()) + "\n")
else:
arquivoGrava.write("%s|%s|%s|%s|%s|%s|%s|%s|%s" % \
(contribuinte['cpf'].replace('.','').replace('-',''),
contribuinte['nome'].strip().split('|')[0],
contribuinte['dados'].split('|')[int(qtd_dados)-3].strip(),
contribuinte['dados'].split('|')[int(qtd_dados)-2].strip(),
"Não Declarante",
contribuinte['dataDisp'].split('|')[0].strip(),
contribuinte['dataDisp'].strip(),
contribuinte['dataDisp'].strip(),
contribuinte['dataDisp'].strip()) + "\n")
elif 'Resultado encontrado: Saldo inexistente' in contribuinte['dados']:
if len(contribuinte['dataDisp'].split('|')) > 1:
arquivoGrava.write("%s|%s|%s|%s|%s|%s|%s|%s|%s" % \
(contribuinte['cpf'].replace('.','').replace('-',''),
contribuinte['nome'].strip().split('|')[0],
contribuinte['dados'].split('|')[int(qtd_dados)-3].strip(),
contribuinte['dados'].split('|')[int(qtd_dados)-2].strip(),
"Sem Saldo",
contribuinte['dataDisp'].split('|')[0].strip(),
contribuinte['dataDisp'].split('|')[1].strip(),
contribuinte['dataDisp'].split('|')[2].strip(),
contribuinte['dataDisp'].split('|')[3].strip()) + "\n")
else:
arquivoGrava.write("%s|%s|%s|%s|%s|%s|%s|%s|%s" % \
(contribuinte['cpf'].replace('.','').replace('-',''),
contribuinte['nome'].strip().split('|')[0],
contribuinte['dados'].split('|')[int(qtd_dados)-3].strip(),
contribuinte['dados'].split('|')[int(qtd_dados)-2].strip(),
"Sem Saldo",
contribuinte['dataDisp'].split('|')[0].strip(),
contribuinte['dataDisp'].strip(),
contribuinte['dataDisp'].strip(),
contribuinte['dataDisp'].strip()) + "\n")
elif 'Resultado encontrado: Imposto a pagar' in contribuinte['dados']:
if len(contribuinte['dataDisp'].split('|')) > 1:
arquivoGrava.write("%s|%s|%s|%s|%s|%s|%s|%s|%s" % \
(contribuinte['cpf'].replace('.','').replace('-',''),
contribuinte['nome'].strip().split('|')[0],
contribuinte['dados'].split('|')[int(qtd_dados)-3].strip(),
contribuinte['dados'].split('|')[int(qtd_dados)-2].strip(),
"Imposto a pagar",
contribuinte['dataDisp'].split('|')[0].strip(),
contribuinte['dataDisp'].split('|')[1].strip(),
contribuinte['dataDisp'].split('|')[2].strip(),
contribuinte['dataDisp'].split('|')[3].strip()) + "\n")
else:
arquivoGrava.write("%s|%s|%s|%s|%s|%s|%s|%s|%s" % \
(contribuinte['cpf'].replace('.','').replace('-',''),
contribuinte['nome'].strip().split('|')[0],
contribuinte['dados'].split('|')[int(qtd_dados)-3].strip(),
contribuinte['dados'].split('|')[int(qtd_dados)-2].strip(),
"Imposto a pagar",
contribuinte['dataDisp'].split('|')[0].strip(),
contribuinte['dataDisp'].strip(),
contribuinte['dataDisp'].strip(),
contribuinte['dataDisp'].strip()) + "\n")
elif 'Situação da Restituição: Aguardando reagendamento pelo contribuinte' in contribuinte['dados']:
if len(contribuinte['dataDisp'].split('|')) > 1:
arquivoGrava.write("%s|%s|%s|%s|%s|%s|%s|%s|%s" % \
(contribuinte['cpf'].replace('.','').replace('-',''),
contribuinte['nome'].strip().split('|')[0],
contribuinte['dados'].split('|')[int(qtd_dados)-3].strip(),
contribuinte['dados'].split('|')[int(qtd_dados)-2].strip(),
"Restituido",
contribuinte['dataDisp'].split('|')[0].strip(),
contribuinte['dataDisp'].split('|')[1].strip(),
contribuinte['dataDisp'].split('|')[2].strip(),
contribuinte['dataDisp'].split('|')[3].strip()) + "\n")
else:
arquivoGrava.write("%s|%s|%s|%s|%s|%s|%s|%s|%s" % \
(contribuinte['cpf'].replace('.','').replace('-',''),
contribuinte['nome'].strip().split('|')[0],
contribuinte['dados'].split('|')[int(qtd_dados)-3].strip(),
contribuinte['dados'].split('|')[int(qtd_dados)-2].strip(),
"Restituido",
contribuinte['dataDisp'].split('|')[0].strip(),
contribuinte['dataDisp'].strip(),
contribuinte['dataDisp'].strip(),
contribuinte['dataDisp'].strip()) + "\n")
elif 'utilize o Extrato do IRPF.' in contribuinte['dados']:
if len(contribuinte['dataDisp'].split('|')) > 1:
arquivoGrava.write("%s|%s|%s|%s|%s|%s|%s|%s|%s" % \
(contribuinte['cpf'].replace('.','').replace('-',''),
contribuinte['nome'].strip().split('|')[0],
contribuinte['dados'].split('|')[int(qtd_dados)-3].strip(),
contribuinte['dados'].split('|')[int(qtd_dados)-2].strip(),
"Declarante",
contribuinte['dataDisp'].split('|')[0].strip(),
contribuinte['dataDisp'].split('|')[1].strip(),
contribuinte['dataDisp'].split('|')[2].strip(),
contribuinte['dataDisp'].split('|')[3].strip()) + "\n")
else:
arquivoGrava.write("%s|%s|%s|%s|%s|%s|%s|%s|%s" % \
(contribuinte['cpf'].replace('.','').replace('-',''),
contribuinte['nome'].strip().split('|')[0],
contribuinte['dados'].split('|')[int(qtd_dados)-3].strip(),
contribuinte['dados'].split('|')[int(qtd_dados)-2].strip(),
"Declarante",
contribuinte['dataDisp'].split('|')[0].strip(),
contribuinte['dataDisp'].strip(),
contribuinte['dataDisp'].strip(),
contribuinte['dataDisp'].strip()) + "\n")
else:
if len(contribuinte['dataDisp'].split('|')) > 1:
arquivoGrava.write("%s|%s|%s|%s|%s|%s|%s|%s|%s" % \
(contribuinte['cpf'].replace('.','').replace('-',''),
contribuinte['nome'].strip().split('|')[0],
contribuinte['dados'].split('|')[int(qtd_dados)-3].strip(),
contribuinte['dados'].split('|')[int(qtd_dados)-2].strip(),
contribuinte['dados'].split('|')[0].split(':')[0],
contribuinte['dataDisp'].split('|')[0].strip(),
contribuinte['dataDisp'].split('|')[1].strip(),
contribuinte['dataDisp'].split('|')[2].strip(),
contribuinte['dataDisp'].split('|')[3].strip()) + "\n")
else:
arquivoGrava.write("%s|%s|%s|%s|%s|%s|%s|%s|%s" % \
(contribuinte['cpf'].replace('.','').replace('-',''),
contribuinte['nome'].strip().split('|')[0],
contribuinte['dados'].split('|')[int(qtd_dados)-3].strip(),
contribuinte['dados'].split('|')[int(qtd_dados)-2].strip(),
contribuinte['dados'].split('|')[0].split(':')[0],
contribuinte['dataDisp'].split('|')[0].strip(),
contribuinte['dataDisp'].strip(),
contribuinte['dataDisp'].strip(),
contribuinte['dataDisp'].strip()) + "\n")
except Exception as ae:
now = datetime.now()
arquivoContrib = "ERRO-" + nomearquivo + "-" + str(now.day) + str(now.month) + str(now.year) + ".log"
arquivoGrava = open(arquivoContrib,'a')
erro = str(ae)
arquivoGrava.write(cpf + "|" + erro + "\n")
arquivoRefazer = "Refazer-" + nomearquivo + "-" + str(now.day) + str(now.month) + str(now.year) + ".log"
arquivoGravaRef = open(arquivoRefazer,'a')
arquivoGravaRef.write(cpf+"\n")
# Thread principal
lista_threads = []
with open(arquivoArgv, 'rb') as arquivo:
for linha in arquivo:
cpf = linha.strip()
while threading.active_count() > MAX_CONEXOES:
time.sleep(1)
thread = threading.Thread(target=consultar_cpf, args=(cpf,))
lista_threads.append(thread)
thread.start()
# Esperando pelas threads abertas terminarem
mostrar_msg("Esperando threads abertas terminarem...")
for thread in lista_threads:
thread.join()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment