154 lines
6.9 KiB
Python
154 lines
6.9 KiB
Python
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) |