docs(ops): update AGENTS.md with Stabili and FE directives; add fast FE XML importer
This commit is contained in:
parent
b660353dde
commit
93de8e9b41
215
scripts/ops/fast_xml_fe_importer.py
Executable file
215
scripts/ops/fast_xml_fe_importer.py
Executable file
|
|
@ -0,0 +1,215 @@
|
|||
#!/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")
|
||||
101
scripts/ops/import_all_fe_stabili.py
Executable file
101
scripts/ops/import_all_fe_stabili.py
Executable file
|
|
@ -0,0 +1,101 @@
|
|||
#!/usr/bin/env python3
|
||||
import os
|
||||
import glob
|
||||
import subprocess
|
||||
import json
|
||||
import pathlib
|
||||
|
||||
repo = pathlib.Path.home() / "netgescon-day0-backup"
|
||||
|
||||
# 1. Fetch stabili with codice_fiscale 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 nel recupero degli stabili dal database.")
|
||||
exit(1)
|
||||
|
||||
stabili = json.loads(out[json_start:json_end])
|
||||
print(f"🤖 Trovati {len(stabili)} stabili con Codice Fiscale nel database:\n")
|
||||
for s in stabili:
|
||||
print(f" • ID {s['id']} [{s['codice_stabile']}] - {s['denominazione']} (CF: {s['codice_fiscale']})")
|
||||
|
||||
# 2. Identify candidate directories containing XML/P7M/ZIP invoice files
|
||||
candidate_dirs = set()
|
||||
|
||||
scan_bases = [
|
||||
"/mnt/gescon-archives/FattureXML",
|
||||
"/mnt/gescon-archives/gescon",
|
||||
"/mnt/gescon-archives/Appo",
|
||||
"/mnt/gescon-archives/scansioni",
|
||||
"/mnt/gescon-archives/Users",
|
||||
"/home/michele/netgescon-day0-backup/storage",
|
||||
"/home/michele/MIki",
|
||||
"/home/michele/netgescon",
|
||||
]
|
||||
|
||||
for base in scan_bases:
|
||||
if os.path.exists(base):
|
||||
for root, dirs, files in os.walk(base):
|
||||
for f in files:
|
||||
ext = f.lower()
|
||||
if ext.endswith(".xml") or ext.endswith(".p7m") or ext.endswith(".zip"):
|
||||
if any(kw in f.lower() or kw in root.lower() for kw in ["xml", "p7m", "fattur", "cassetto", "ade", "0021", "0013", "0002", "0010", "0016", "0018", "0019", "0023", "0024", "0025", "0026"]):
|
||||
candidate_dirs.add(root)
|
||||
|
||||
print(f"\n📂 Trovate {len(candidate_dirs)} directory candidate contenenti file XML/P7M/ZIP di Fatture Elettroniche.")
|
||||
|
||||
# 3. Run fe:cassetto-import-local for each Stabile and candidate directory
|
||||
total_imported = 0
|
||||
|
||||
for s in stabili:
|
||||
sid = s['id']
|
||||
c_code = s['codice_stabile']
|
||||
cf = s['codice_fiscale']
|
||||
print(f"\n🚀 Sincronizzazione FE per Stabile ID {sid} [{c_code}] - {s['denominazione']} (CF: {cf})...")
|
||||
|
||||
for cdir in candidate_dirs:
|
||||
cmd = [
|
||||
"php", "artisan", "fe:cassetto-import-local",
|
||||
str(sid),
|
||||
cdir,
|
||||
"--dal=2019-01-01",
|
||||
"--al=2026-12-31",
|
||||
"--no_skip=1"
|
||||
]
|
||||
res_imp = subprocess.run(cmd, cwd=repo, capture_output=True, text=True)
|
||||
stdout = res_imp.stdout.strip()
|
||||
if "imported=" in stdout and "imported=0" not in stdout:
|
||||
print(f" ✔️ {cdir} -> {stdout}")
|
||||
|
||||
# 4. Final summary count of imported FE invoices
|
||||
tinker_summary = """
|
||||
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_sum = subprocess.run(["php", "artisan", "tinker", "--execute=" + tinker_summary], cwd=repo, capture_output=True, text=True)
|
||||
out_sum = res_sum.stdout.strip()
|
||||
js_start = out_sum.find("{")
|
||||
js_end = out_sum.rfind("}") + 1
|
||||
if js_start != -1 and js_end != 0:
|
||||
data_sum = json.loads(out_sum[js_start:js_end])
|
||||
print(f"\n📊 RIEPILOGO FINALE FATTURE ELETTRONICHE IMPORTATE PER CODICE FISCALE:")
|
||||
print(f" Totale Complessivo FE in Database: {data_sum.get('total_all', 0)}\n")
|
||||
for row in data_sum.get("by_stabile", []):
|
||||
print(f" • Stabile [{row['codice_stabile']}] {row['denominazione']} (CF: {row['codice_fiscale']}): {row['total_fe']} FE")
|
||||
Loading…
Reference in New Issue
Block a user