gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
57 lines
4.2 KiB
Python
57 lines
4.2 KiB
Python
"""Parse the official SPED EFD-Contribuições tables 4.3.x (Word .doc, converted with antiword) into a versioned NCM-keyed dataset.
|
|
Source: sped.rfb.gov.br/arquivo/download/<id> (current files, fetched directly on 2026-09-21).
|
|
Writes docs/sources/SPED_PISCOFINS_TABLES_V3.json (per table: source id, sha256 of the .doc, version text, entries {code, description, ncm_prefixes, start, end, pis_rate, cofins_rate})."""
|
|
import datetime, hashlib, json, re, sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]; SRC = ROOT / "sources_cache" / "cur"
|
|
TABLES = { # sped item id -> (table, CST it defines for the SAIDA side, meaning)
|
|
1638: ("4.3.10", ["02", "04"], "MONOFASICA_OU_PAUTA"), 1642: ("4.3.12", ["05"], "SUBSTITUICAO_TRIBUTARIA"), 1643: ("4.3.13", ["06"], "ALIQUOTA_ZERO"),
|
|
1646: ("4.3.14", ["07"], "ISENTA"), 1651: ("4.3.15", ["08"], "SEM_INCIDENCIA"), 5786: ("4.3.11", ["03", "04"], "MONOFASICA_AD_REM")}
|
|
NCM = re.compile(r"\d{4}(?:\.\d{2}(?:\.\d{2}(?:\.\d{2})?)?)?")
|
|
DATE = re.compile(r"\d{2}/\d{4}|\d{2}/\d{2}/\d{4}")
|
|
|
|
|
|
def parse(txt):
|
|
rows, cur = [], None
|
|
for line in txt.splitlines():
|
|
if not line.startswith("|"):
|
|
continue
|
|
cells = [c.strip() for c in line.strip().strip("|").split("|")]
|
|
if cells and re.fullmatch(r"\d{3}", cells[0] or ""):
|
|
cur = {"code": cells[0], "cells": [[c] for c in cells[1:]]}; rows.append(cur)
|
|
elif cur is not None and len(cells) > 1 and not cells[0]:
|
|
for i, c in enumerate(cells[1:]):
|
|
if i < len(cur["cells"]) and c:
|
|
cur["cells"][i].append(c)
|
|
return rows
|
|
|
|
|
|
out = {"artifact": "SPED_PISCOFINS_TABLES_V3", "tables": {}}
|
|
for sid, (tab, csts, meaning) in TABLES.items():
|
|
raw = (SRC / f"sped_{sid}.bin").read_bytes(); txt = (SRC / f"sped_{sid}.txt").read_text(encoding="utf-8", errors="replace")
|
|
hdr = (SRC / f"h_{sid}.txt").read_text(errors="replace"); lm = re.search(r"Last-Modified: (.*)", hdr); lm = lm.group(1).strip() if lm else None
|
|
head_all = " ".join(txt.splitlines()[:60])
|
|
ver = re.search(r"Vers[ãa]o (\d+(?:\.\d+)+)", head_all); eff = re.search(r"Atualizad[ao]s? em (\d{2}/\d{2}/\d{4})", head_all)
|
|
head = " ".join(l.strip() for l in txt.splitlines()[:8] if l.strip() and not l.startswith("|"))
|
|
entries = []
|
|
for r in parse(txt):
|
|
cells = [" ".join(c) for c in r["cells"]] or [""]; dcells = ["".join(c) for c in r["cells"]]
|
|
joined_ncm = cells[1] if len(cells) > 1 else ""
|
|
pref = sorted({m.group(0).replace(".", "") for m in NCM.finditer(joined_ncm)})
|
|
dates = [d for c in dcells[2:] for d in DATE.findall(c)]
|
|
rates = re.findall(r"\d+,\d+", " ".join(cells[2:4])) if tab in ("4.3.10", "4.3.12") else []
|
|
entries.append({"code": r["code"], "description": cells[0] if cells else "", "ncm_prefixes": pref, "applies_to_all_ncm": not pref and joined_ncm.strip() in ("-", ""), "start": dates[0] if dates else None, "end": dates[1] if len(dates) > 1 else None, "rates_text": rates})
|
|
out["tables"][tab] = {"sped_item": sid, "source_url": f"http://sped.rfb.gov.br/arquivo/download/{sid}", "download_date": datetime.date.today().isoformat(), "http_last_modified": lm, "version": ver.group(1) if ver else None, "effective_date": eff.group(1) if eff else None, "sha256_doc": hashlib.sha256(raw).hexdigest(), "version_text": head, "saida_cst": csts, "meaning": meaning,
|
|
"fetched_via": "direct official download over http (sped.rfb.gov.br)", "entries": entries, "ncm_prefix_count": len({p for e in entries for p in e["ncm_prefixes"]})}
|
|
print(tab, len(entries), out["tables"][tab]["ncm_prefix_count"], head[:110])
|
|
for sid, name in ((1629, "4.3.3"), (1630, "4.3.4")):
|
|
raw = (SRC / f"sped_{sid}.bin").read_bytes(); txt = (SRC / f"sped_{sid}.txt").read_text(encoding="utf-8", errors="replace")
|
|
cst = {}
|
|
for line in txt.splitlines():
|
|
m = re.match(r"\|(\d{2})\s*\|([^|]+)\|", line)
|
|
if m: cst[m.group(1)] = m.group(2).strip()
|
|
out["tables"][name] = {"sped_item": sid, "sha256_doc": hashlib.sha256(raw).hexdigest(), "cst_domain": cst}
|
|
print(name, sorted(cst))
|
|
(ROOT / "docs" / "sources" / "SPED_PISCOFINS_TABLES_V3.json").write_text(json.dumps(out, ensure_ascii=False, indent=1), encoding="utf-8")
|