-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathscript.py
More file actions
76 lines (63 loc) · 2.45 KB
/
Copy pathscript.py
File metadata and controls
76 lines (63 loc) · 2.45 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
import os
import glob
import pandas as pd
import xmltodict
def flatten(d: dict, parent: str = '', sep: str = '_') -> dict:
items = {}
for k, v in d.items():
key = f"{parent}{sep}{k}" if parent else k
if isinstance(v, dict):
items.update(flatten(v, key, sep=sep))
elif isinstance(v, list):
if all(not isinstance(i, dict) for i in v):
items[key] = "|".join(map(str, v))
else:
for idx, elt in enumerate(v, 1):
if isinstance(elt, dict):
items.update(flatten(elt, f"{key}{sep}{idx}", sep=sep))
else:
items[f"{key}{sep}{idx}"] = elt
else:
items[key] = v
return items
def extrair_records(doc: dict, nome: str) -> list[dict]:
def com_arquivo(rec):
rec["__arquivo__"] = nome
return rec
if "nfeProc" in doc:
proc = doc["nfeProc"]
infNFe = proc["NFe"]["infNFe"]
header = flatten(infNFe)
if "protNFe" in proc:
header.update(flatten(proc["protNFe"]["infProt"]))
dets = infNFe.get("det", [])
if isinstance(dets, dict):
dets = [dets]
return [com_arquivo({**header, **flatten(det)}) for det in dets]
if "procEventoNFe" in doc:
proc = doc["procEventoNFe"]
rec = flatten(proc["envEvento"]["evento"]["infEvento"])
rec.update(flatten(proc["retEnvEvento"]["retEvento"]["infEvento"]))
return [com_arquivo(rec)]
return [com_arquivo(flatten(doc))]
def processar(folder: str) -> str:
xml_files = glob.glob(os.path.join(folder, "*.xml"))
print(f"Encontrados {len(xml_files)} arquivos XML em:\n {folder}\n")
records = []
for path in xml_files:
nome = os.path.basename(path)
print("Processando:", nome)
try:
with open(path, "r", encoding="utf-8") as f:
doc = xmltodict.parse(f.read())
records.extend(extrair_records(doc, nome))
except Exception as e:
print(f"Ignorando '{nome}' devido a erro: {e}")
df = pd.DataFrame(records)
output_file = os.path.join(folder, "nfe_data_por_item.xlsx")
df.to_excel(output_file, index=False)
print(f"\nConcluído! {df.shape[0]} linhas × {df.shape[1]} colunas em:\n {output_file}")
return output_file
if __name__ == "__main__":
folder = input("Caminho da pasta com os XMLs: ").strip()
processar(folder)