#!/usr/bin/env python3 ''' Script de python para realizar consultas de "Mis Aportes" en AFIP de forma automatizada. -> Verificar si está python instalado - En la terminal utilizar el siguiente comando: python --version o python3 --version Si está instalado arroja la versión instalada, si no está instalado descargarlo de la siguiente página web: https://www.python.org/downloads/ -> Instalar pip: python -m ensurepip --upgrade -> Librerías python necesarias a instalar con pip: - sudo pip3 install pyppeteer - sudo pip3 install pyobdc - sudo pip3 install requests - sudo pip3 install asyncio - sudo pip3 install time - sudo pip3 install datetime !! Tener en cuenta que pyppeteer utiliza el navegador Chrome o Chromium (en la instalación de pyppeteer se instala automáticamente uno predeterminado) Igualmente, en la función --> afip_aportes() el método Launch() indicamos el Path ejecutable del Chrome. -----> 'executablePath': 'C:/Program Files/Google/Chrome/Application/chrome.exe' (Windows) -----> 'executablePath': 'chromium' # --> Puede ser Chromium o Chrome o google-chrome (Linux) Por el momento no exige la resolución del Captcha por lo que no se agregó. (Si comienza a fallar y todos los cuils arrojan "Error al ingresar el cuil - {cuil}" quiere decir que está fallando el captcha y habría que agregarlo) Una buena opción es el Anti-Captcha, https://anti-captcha.com/ ''' # ----------------------------------------------------------------------------------------------------------------- import asyncio #import pyodbc import time import requests import datetime import os import sys from pyppeteer import launch cuils_con_errores = [] cuils_ok = 0 # ---------------> url para el proceso del html url_html = 'https://backend.fidelitas.com.ar/api/cuil-aportes-empleado' url_estado = 'https://produccion.fidelitas.com.ar/mis-aportes-error' url_listado = 'https://produccion.fidelitas.com.ar/mis-aportes-procesar' url_procesado = 'https://produccion.fidelitas.com.ar/mis-aportes-procesar-update' # ---------------> Conexión a la base de datos def extrer_cuils_db(cliente): listado_cuils = [] try: response = requests.get(url_listado+'?cliente='+cliente+'&id_taken=99') # response = requests.get(url_listado+'?cliente=5546348&id_taken=1') # AMX response = response.json() for cuil in response: listado_cuils.append(cuil) except Exception as e: print("Error al ejecutar el stored procedure:", e) finally: return listado_cuils # -------------------> Pyppeteer para extraer los html de mis aportes en afip async def afip_aportes(cuil): global cuils_ok browser = await launch({ 'headless': True, # --> Por defecto es True, es decir, no abre la pantalla del Chrome, en False podemos seguir el proceso. #'executablePath': 'C:/Program Files/Google/Chrome/Application/chrome.exe' # --> Path del ejecutable Chrome Windows 'executablePath': '/usr/bin/google-chrome-stable' # --> Path del Chrome en Linux }) try: page = await browser.newPage() await page.goto("https://serviciosweb.afip.gob.ar/TRAMITES_CON_CLAVE_FISCAL/MISAPORTES/app/basica.aspx") # --> Página Mis Aportes de Afip await page.waitForSelector('input[id="ctl00_ContentPlaceHolder2_txtCuil_txtSufijo"]') await page.click('.TextSufijo') # --> Input para ingresar el cuil for n in range(0,12): await page.keyboard.press('ArrowLeft') # --> Para que comience a escribir a la izquierda # await page.type('input[id="ctl00_ContentPlaceHolder2_txtCuil_txtSufijo"]', cuil['CUIL']) # try: await page.click('.Boton') # --> Botón Continuar await page.waitForNavigation() # Si el Cuil ingresado no se encuentra en el sistema o hay un error con el Captcha salta un error y corta el proceso para seguir con el siguiente cuil div_presente = await page.querySelector('.TextoIngresoDatos') is not None if (div_presente): requests.get(url_procesado+'?proceso='+cuil['proceso']+'&id='+str(cuil['id'])) div_text = await page.querySelector("div[id='ctl00_ContentPlaceHolder2_vldSumaryCuil']") text_content = await page.evaluate('(element) => element.innerHTML', div_text) error = text_content.split('
')[0].strip() print(cuil['CUIL']+' - '+error) requests.get(url_estado+'?cuil='+cuil['CUIL']+'&error='+error+'&tipo='+cuil['proceso']+'&p_dias=60') cuils_con_errores.append(cuil['CUIL']) await browser.close() return # Si continua sin problemas, tomamos el html de la página y la enviamos a la url indicada else: div_error = await page.querySelector(".TablaBasicaError") div_error2 = await page.querySelector(".TablaErrorPage") error = '' if (div_error is not None): error = 'NO está registrado en relación de dependencia en los últimos 12 meses' print(error) requests.get(url_procesado+'?proceso='+cuil['proceso']+'&id='+str(cuil['id'])) requests.get(url_estado+'?cuil='+cuil['CUIL']+'&error='+error+'&tipo='+cuil['proceso']+'&p_dias=90') if (div_error2 is not None): error = 'Error en página de ARCA' print(error) if (error != ''): cuils_con_errores.append(cuil['CUIL']) await browser.close() return else: requests.get(url_procesado+'?proceso='+cuil['proceso']+'&id='+str(cuil['id'])) await page.querySelector('.table') html = await page.content() input_element = await page.querySelector('input[id="ctl00_ContentPlaceHolder2_hLst"]') cuit = await page.evaluate('(element) => element.value', input_element) requests.post(url_html, data= {'cuit': cuit[0:11], 'cuil': cuil['CUIL'], 'info': html}) # requests.get(url_procesado+'?proceso='+cuil['proceso']+'&id='+cuil['id']) cuit = cuit[12:] print('1 OK') time.sleep(1) # Si el empleado tiene más de 1 Empleador va a aparecer el botón "Siguiente" hasta que no tenga más, de cada página, toma el html y la envia a la url indicada for n in range(2,13): try: await page.click('input[id="ctl00_ContentPlaceHolder2_btnEmpleSiguiente"]') # Botón "Siguiente" await page.waitForNavigation() html = await page.content() requests.post(url_html, data= {'cuit': cuit[0:11], 'cuil': cuil['CUIL'], 'info': html}) # requests.get(url_procesado+'?proceso='+cuil['proceso']+'&id='+cuil['id']) cuit = cuit[12:] print(f'{n} OK') # except Exception as e: print('Completo') break # # cuils_ok += 1 await browser.close() # except Exception as e: cuils_con_errores.append(cuil['CUIL']) print(f"Error: {e}") finally: await browser.close() except Exception as e: print(e) finally: await browser.close() # ---------------------------------------------------- async def main(cliente): cliente = cliente[1] global cuils_ok fecha_inicio = datetime.datetime.now() if (os.path.isdir('mis_aportes.txt')): with open('mis_aportes.txt', 'a', encoding='UTF-8') as file: file.write('---------------------------------------------\n') file.write(f'start-excecution at {fecha_inicio}\n') else : with open('mis_aportes.txt', 'a', encoding='UTF-8') as file: file.write('---------------------------------------------\n') file.write(f'start-excecution at {fecha_inicio}\n') print(cliente) listado_cuils = extrer_cuils_db(cliente) print(listado_cuils) with open('mis_aportes.txt', 'a', encoding='UTF-8') as file: file.write('DB extraction - OK\n') file.write(f'Cantidad de cuils de BBDD {len(listado_cuils)}\n') n = 1 for cuil in listado_cuils: print(f"{n} de {len(listado_cuils)} - {cuil['CUIL']}") await afip_aportes(cuil) n += 1 fecha_fin = datetime.datetime.now() with open('mis_aportes.txt', 'a', encoding='UTF-8') as file: file.write(f'Cantidad de cuils correctos procesados: {cuils_ok}\n') file.write(f'Cantidad de cuils totales con errores: {len(cuils_con_errores)}\nCuils con errores: {cuils_con_errores}\nFinish-excecution at {fecha_fin}\n++++++++++++++++++++++\n') # if __name__ == "__main__": while True: asyncio.run(main(sys.argv))