216 lines
10 KiB
Python
Executable File
216 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import os
|
|
import glob
|
|
import xml.etree.ElementTree as ET
|
|
import subprocess
|
|
import json
|
|
import hashlib
|
|
import pathlib
|
|
|
|
repo = pathlib.Path.home() / "netgescon-day0-backup"
|
|
|
|
# 1. Fetch stabili CF mapping from DB
|
|
tinker_cmd = """
|
|
use Illuminate\\Support\\Facades\\DB;
|
|
$stabili = DB::table("stabili")->whereNotNull("codice_fiscale")->where("codice_fiscale", "!=", "")->get(["id", "codice_stabile", "denominazione", "codice_fiscale"]);
|
|
echo json_encode($stabili);
|
|
"""
|
|
|
|
res = subprocess.run(["php", "artisan", "tinker", "--execute=" + tinker_cmd], cwd=repo, capture_output=True, text=True)
|
|
out = res.stdout.strip()
|
|
json_start = out.find("[")
|
|
json_end = out.rfind("]") + 1
|
|
|
|
if json_start == -1 or json_end == 0:
|
|
print("Errore recupero stabili.")
|
|
exit(1)
|
|
|
|
stabili = json.loads(out[json_start:json_end])
|
|
cf_to_stabile = {s["codice_fiscale"].strip().upper(): s for s in stabili}
|
|
|
|
print(f"🤖 Trovati {len(cf_to_stabile)} Stabili con Codice Fiscale per il matching FE:\n")
|
|
for cf, s in cf_to_stabile.items():
|
|
print(f" • CF {cf} -> ID {s['id']} [{s['codice_stabile']}] - {s['denominazione']}")
|
|
|
|
# 2. Scan XML files
|
|
scan_dirs = [
|
|
"/mnt/gescon-archives",
|
|
"/home/michele/MIki",
|
|
"/home/michele/netgescon-day0-backup/storage",
|
|
"/home/michele/netgescon",
|
|
]
|
|
|
|
to_insert = []
|
|
seen_hashes = set()
|
|
|
|
print("\n🔍 Scansione massiva in corso...")
|
|
|
|
for bdir in scan_dirs:
|
|
if os.path.exists(bdir):
|
|
for root, dirs, files in os.walk(bdir):
|
|
for f in files:
|
|
if f.lower().endswith(".xml"):
|
|
fpath = os.path.join(root, f)
|
|
try:
|
|
with open(fpath, "r", encoding="utf-8", errors="ignore") as fh:
|
|
content = fh.read()
|
|
|
|
xml_hash = hashlib.sha256(content.encode("utf-8")).hexdigest()
|
|
if xml_hash in seen_hashes:
|
|
continue
|
|
seen_hashes.add(xml_hash)
|
|
|
|
tree = ET.fromstring(content)
|
|
|
|
# Extract CessionarioCommittente CodiceFiscale / IdCodice
|
|
cf_dest = None
|
|
den_dest = ""
|
|
num_fattura = ""
|
|
data_fattura = ""
|
|
totale = "0.00"
|
|
imponibile = "0.00"
|
|
iva = "0.00"
|
|
forn_den = ""
|
|
forn_cf = ""
|
|
forn_piva = ""
|
|
iban = ""
|
|
modalita_pag = ""
|
|
desc = ""
|
|
|
|
for elem in tree.iter():
|
|
tag = elem.tag.split("}")[-1]
|
|
|
|
if tag == "CessionarioCommittente":
|
|
for child in elem.iter():
|
|
ctag = child.tag.split("}")[-1]
|
|
if ctag in ["CodiceFiscale", "IdCodice"] and child.text:
|
|
cf_dest = child.text.strip().upper()
|
|
elif ctag in ["Denominazione", "Nome"] and child.text:
|
|
den_dest = child.text.strip()
|
|
|
|
elif tag == "CedentePrestatore":
|
|
for child in elem.iter():
|
|
ctag = child.tag.split("}")[-1]
|
|
if ctag == "CodiceFiscale" and child.text:
|
|
forn_cf = child.text.strip().upper()
|
|
elif ctag == "IdCodice" and child.text:
|
|
forn_piva = child.text.strip()
|
|
elif ctag in ["Denominazione", "Nome"] and child.text:
|
|
forn_den = child.text.strip()
|
|
|
|
elif tag == "DatiGeneraliDocumento":
|
|
for child in elem.iter():
|
|
ctag = child.tag.split("}")[-1]
|
|
if ctag == "Numero" and child.text:
|
|
num_fattura = child.text.strip()
|
|
elif ctag == "Data" and child.text:
|
|
data_fattura = child.text.strip()
|
|
elif ctag == "ImportoTotaleDocumento" and child.text:
|
|
totale = child.text.strip()
|
|
|
|
elif tag == "DatiRiepilogo":
|
|
for child in elem.iter():
|
|
ctag = child.tag.split("}")[-1]
|
|
if ctag == "ImponibileImporto" and child.text:
|
|
imponibile = child.text.strip()
|
|
elif ctag == "Imposta" and child.text:
|
|
iva = child.text.strip()
|
|
|
|
elif tag == "DettaglioLinee" and not desc:
|
|
for child in elem.iter():
|
|
ctag = child.tag.split("}")[-1]
|
|
if ctag == "Descrizione" and child.text:
|
|
desc = child.text.strip()
|
|
|
|
elif tag == "DettaglioPagamento":
|
|
for child in elem.iter():
|
|
ctag = child.tag.split("}")[-1]
|
|
if ctag == "IBAN" and child.text:
|
|
iban = child.text.strip()
|
|
elif ctag == "ModalitaPagamento" and child.text:
|
|
modalita_pag = child.text.strip()
|
|
|
|
if cf_dest and cf_dest in cf_to_stabile:
|
|
stabile = cf_to_stabile[cf_dest]
|
|
to_insert.append({
|
|
"stabile_id": stabile["id"],
|
|
"numero_fattura": num_fattura or f"FT-{len(to_insert)+1}",
|
|
"data_fattura": data_fattura or "2021-01-01",
|
|
"fornitore_denominazione": forn_den or "Fornitore Sconosciuto",
|
|
"fornitore_cf": forn_cf,
|
|
"fornitore_piva": forn_piva,
|
|
"imponibile": imponibile or "0.00",
|
|
"iva": iva or "0.00",
|
|
"totale": totale or "0.00",
|
|
"pagamento_iban": iban,
|
|
"pagamento_modalita": modalita_pag,
|
|
"consumo_raw": desc,
|
|
"destinatario_cf": cf_dest,
|
|
"destinatario_denominazione": den_dest,
|
|
"nome_file_xml": f,
|
|
"xml_hash": xml_hash,
|
|
"xml_path": fpath,
|
|
"xml_content": content,
|
|
"stato": "ricevuta",
|
|
"created_at": "NOW()",
|
|
"updated_at": "NOW()"
|
|
})
|
|
except Exception:
|
|
pass
|
|
|
|
print(f"✔️ Scansione completata: Trovate {len(to_insert)} Fatture Elettroniche valide abbinate per Codice Fiscale.")
|
|
|
|
# 3. Bulk insert into DB via Tinker
|
|
if to_insert:
|
|
# Save payload to json file
|
|
payload_file = repo / "storage/app/fe_import_payload.json"
|
|
with open(payload_file, "w", encoding="utf-8") as pf:
|
|
json.dump(to_insert, pf)
|
|
|
|
tinker_bulk = """
|
|
use Illuminate\\Support\\Facades\\DB;
|
|
$file = storage_path("app/fe_import_payload.json");
|
|
$items = json_decode(file_get_contents($file), true);
|
|
|
|
$inserted = 0;
|
|
foreach ($items as $item) {
|
|
$exists = DB::table("fatture_elettroniche")->where("xml_hash", $item["xml_hash"])->first();
|
|
if (!$exists) {
|
|
$item["created_at"] = now();
|
|
$item["updated_at"] = now();
|
|
DB::table("fatture_elettroniche")->insert($item);
|
|
$inserted++;
|
|
}
|
|
}
|
|
|
|
echo json_encode(["inserted" => $inserted, "total_db" => DB::table("fatture_elettroniche")->count()]);
|
|
"""
|
|
|
|
res_b = subprocess.run(["php", "artisan", "tinker", "--execute=" + tinker_bulk], cwd=repo, capture_output=True, text=True)
|
|
out_b = res_b.stdout.strip()
|
|
print("RES_BULK:", out_b[out_b.find("{"):out_b.rfind("}")+1])
|
|
|
|
# 4. Show summary per Stabile
|
|
tinker_final = """
|
|
use Illuminate\\Support\\Facades\\DB;
|
|
$summary = DB::table("fatture_elettroniche")
|
|
->join("stabili", "stabili.id", "=", "fatture_elettroniche.stabile_id")
|
|
->select("stabili.codice_stabile", "stabili.denominazione", "stabili.codice_fiscale", DB::raw("count(fatture_elettroniche.id) as total_fe"))
|
|
->groupBy("stabili.id", "stabili.codice_stabile", "stabili.denominazione", "stabili.codice_fiscale")
|
|
->get();
|
|
|
|
$total = DB::table("fatture_elettroniche")->count();
|
|
echo json_encode(["total_all" => $total, "by_stabile" => $summary]);
|
|
"""
|
|
|
|
res_f = subprocess.run(["php", "artisan", "tinker", "--execute=" + tinker_final], cwd=repo, capture_output=True, text=True)
|
|
out_f = res_f.stdout.strip()
|
|
js_f1 = out_f.find("{")
|
|
js_f2 = out_f.rfind("}") + 1
|
|
if js_f1 != -1 and js_f2 != 0:
|
|
data_f = json.loads(out_f[js_f1:js_f2])
|
|
print(f"\n📊 RIEPILOGO FINALE FATTURE ELETTRONICHE IMPORTATE IN DATABASE:")
|
|
print(f" Totale Complessivo FE in Database: {data_f.get('total_all', 0)}\n")
|
|
for row in data_f.get("by_stabile", []):
|
|
print(f" • Stabile [{row['codice_stabile']}] {row['denominazione']} (CF: {row['codice_fiscale']}): {row['total_fe']} FE")
|