import requests import json import sys import os import re from datetime import datetime, timedelta import pytz from tqdm import tqdm def unixTime(): dt = datetime.now(tz=pytz.utc) return str(int(dt.timestamp() * 1000)) def aggiusta_fine_trimestre(d): if d.month < 4: return datetime(d.year, 3, 31) elif d.month < 7: return datetime(d.year, 6, 30) elif d.month < 10: return datetime(d.year, 9, 30) else: return datetime(d.year, 12, 31) def divide_in_trimestri(data_iniziale, data_finale): d1 = datetime.strptime(data_iniziale, "%d/%m/%Y") d2 = datetime.strptime(data_finale, "%d/%m/%Y") trimestri = [] while d1 < d2: fine_trimestre = aggiusta_fine_trimestre(d1) if fine_trimestre >= d2: trimestri.append((d1.strftime("%d%m%Y"), d2.strftime("%d%m%Y"))) break else: trimestri.append((d1.strftime("%d%m%Y"), fine_trimestre.strftime("%d%m%Y"))) d1 = fine_trimestre + timedelta(days=1) return trimestri def login_ade(cf, pin, password): s = requests.Session() s.headers.update({'User-Agent': 'Mozilla/5.0'}) payload = { '_58_saveLastPath': 'false', '_58_redirect': '', '_58_doActionAfterLogin': 'false', '_58_login': cf, '_58_pin': pin, '_58_password': password } r = s.post('https://ivaservizi.agenziaentrate.gov.it/portale/home?p_p_id=58&p_p_lifecycle=1&p_p_state=normal&p_p_mode=view&p_p_col_id=column-1&p_p_col_pos=3&p_p_col_count=4&_58_struts_action=%2Flogin%2Flogin', data=payload) if r.status_code != 200: print("Login AdE fallito.") sys.exit(1) liferay = re.findall(r"Liferay.authToken = '.*';", r.text)[0] p_auth = liferay.replace("Liferay.authToken = '","").replace("';", "") return s, p_auth def get_token_headers(s): headers_token = { 'x-xss-protection': '1; mode=block', 'strict-transport-security': 'max-age=16070400; includeSubDomains', 'x-content-type-options': 'nosniff', 'x-frame-options': 'deny' } r = s.get('https://ivaservizi.agenziaentrate.gov.it/cons/cons-services/sc/tokenB2BCookie/get?v='+unixTime(), headers=headers_token) print("Status code:", r.status_code) print("Response text:", r.text) if r.status_code != 200: print("Token B2B non ottenuto.") sys.exit(1) xb2bcookie = r.headers.get('x-b2bcookie') xtoken = r.headers.get('x-token') return xb2bcookie, xtoken def scarica_fatture(parametri, credenziali): cf = credenziali["username"] pin = credenziali["pin"] password = credenziali["password"] cfcliente = parametri["codice_fiscale_condominio"] data_dal = parametri["data_dal"] data_al = parametri["data_al"] trimestri = divide_in_trimestri(data_dal, data_al) print("Trimestri:", trimestri) for data_iniziale_trimestre, data_finale_trimestre in trimestri: print(f"Elaborazione periodo {data_iniziale_trimestre} - {data_finale_trimestre}") s, p_auth = login_ade(cf, pin, password) # Selezione utenza (delega diretta) payload = {'cf_inserito': cfcliente} s.post('https://ivaservizi.agenziaentrate.gov.it/portale/scelta-utenza-lavoro?p_auth='+ p_auth + '&p_p_id=SceltaUtenzaLavoro_WAR_SceltaUtenzaLavoroportlet&p_p_lifecycle=1&p_p_state=normal&p_p_mode=view&p_p_col_id=column-1&p_p_col_count=1&_SceltaUtenzaLavoro_WAR_SceltaUtenzaLavoroportlet_javax.portlet.action=delegaDirettaAction', data=payload) xb2bcookie, xtoken = get_token_headers(s) headers = { 'x-b2bcookie': xb2bcookie, 'x-token': xtoken, 'User-Agent': 'Mozilla/5.0', 'Accept': 'application/json, text/plain, */*' } # Scarica fatture passive ricevute url = f'https://ivaservizi.agenziaentrate.gov.it/cons/cons-services/rs/fe/ricevute/dal/{data_iniziale_trimestre}/al/{data_finale_trimestre}/ricerca/ricezione?v={unixTime()}' r = s.get(url, headers=headers) if r.status_code == 200: ricevute = r.json() with open(f'fe_ricevute_{cfcliente}_{data_iniziale_trimestre}_{data_finale_trimestre}.json', 'w', encoding='utf-8') as f: json.dump(ricevute, f, ensure_ascii=False, indent=2) print(f"Scaricato json fatture passive ricevute per {cfcliente} {data_iniziale_trimestre}-{data_finale_trimestre}") # Scarica XML/P7M delle fatture passive for fattura in ricevute.get("elencoFatture", []): id_sdi = fattura.get("identificativoSdI") if not id_sdi: continue url_xml = f'https://ivaservizi.agenziaentrate.gov.it/cons/cons-services/rs/fe/ricevute/{id_sdi}/file' r_xml = s.get(url_xml, headers=headers) if r_xml.status_code == 200: with open(f'fattura_ricevuta_{id_sdi}.xml.p7m', 'wb') as fx: fx.write(r_xml.content) print(f"Scaricato XML/P7M ricevuta {id_sdi}") else: print(f"Errore download XML/P7M ricevuta {id_sdi}") else: print("Errore nel download delle fatture passive ricevute.") # Scarica fatture emesse url_emesse = f'https://ivaservizi.agenziaentrate.gov.it/cons/cons-services/rs/fe/emesse/dal/{data_iniziale_trimestre}/al/{data_finale_trimestre}?v={unixTime()}' r = s.get(url_emesse, headers=headers) if r.status_code == 200: emesse = r.json() with open(f'fe_emesse_{cfcliente}_{data_iniziale_trimestre}_{data_finale_trimestre}.json', 'w', encoding='utf-8') as f: json.dump(emesse, f, ensure_ascii=False, indent=2) print(f"Scaricato json fatture emesse per {cfcliente} {data_iniziale_trimestre}-{data_finale_trimestre}") # Scarica XML/P7M delle fatture emesse for fattura in emesse.get("elencoFatture", []): id_sdi = fattura.get("identificativoSdI") if not id_sdi: continue url_xml = f'https://ivaservizi.agenziaentrate.gov.it/cons/cons-services/rs/fe/emesse/{id_sdi}/file' r_xml = s.get(url_xml, headers=headers) if r_xml.status_code == 200: with open(f'fattura_emessa_{id_sdi}.xml.p7m', 'wb') as fx: fx.write(r_xml.content) print(f"Scaricato XML/P7M emessa {id_sdi}") else: print(f"Errore download XML/P7M emessa {id_sdi}") else: print("Errore nel download delle fatture emesse.") if __name__ == "__main__": with open("parametri_fatture.json", "r") as f: parametri = json.load(f) with open("credenziali_ade.json", "r") as f: credenziali = json.load(f) scarica_fatture(parametri, credenziali)