|
|
| from __future__ import annotations |
|
|
| import hashlib |
| from datetime import date |
| from decimal import Decimal |
| from pathlib import Path |
| from typing import Any |
|
|
|
|
| def _fmt_data(d: date | None) -> str: |
| if d is None: |
| return "" |
| return d.strftime("%d%m%Y") |
|
|
|
|
| def _fmt_decimal(v: Decimal | None, casas: int = 2) -> str: |
| if v is None or v == 0: |
| return "" |
| return f"{v:.{casas}f}".replace(".", ",") |
|
|
|
|
| def _fmt_str(v: Any) -> str: |
| if v is None: |
| return "" |
| return str(v).strip() |
|
|
|
|
| class RegistroSPED: |
| |
|
|
| def __init__(self, tipo: str): |
| self.tipo = tipo |
| self._campos: list[str] = [] |
|
|
| def add(self, valor: Any, tipo_campo: str = "str", casas: int = 2) -> "RegistroSPED": |
| if valor is None or valor == "": |
| self._campos.append("") |
| elif tipo_campo == "data": |
| self._campos.append(_fmt_data(valor)) |
| elif tipo_campo == "decimal": |
| self._campos.append(_fmt_decimal(Decimal(str(valor)), casas)) |
| elif tipo_campo == "int": |
| self._campos.append(str(int(valor))) |
| else: |
| self._campos.append(str(valor).strip()) |
| return self |
|
|
| def to_line(self) -> str: |
| campos = [self.tipo] + self._campos |
| return "|" + "|".join(campos) + "|\n" |
|
|
|
|
| class ArquivoSPED: |
| |
|
|
| def __init__(self, nome_arquivo: str): |
| self.nome_arquivo = nome_arquivo |
| self._linhas: list[str] = [] |
| self._contadores: dict[str, int] = {} |
|
|
| def _add_linha(self, linha: str) -> None: |
| self._linhas.append(linha) |
| tipo = linha.split("|")[1] if "|" in linha else "" |
| self._contadores[tipo] = self._contadores.get(tipo, 0) + 1 |
|
|
| def adicionar_registro(self, reg: RegistroSPED) -> None: |
| self._add_linha(reg.to_line()) |
|
|
| def adicionar_linha_raw(self, linha: str) -> None: |
| if not linha.endswith("\n"): |
| linha += "\n" |
| self._add_linha(linha) |
|
|
| def total_linhas(self) -> int: |
| return len(self._linhas) |
|
|
| def conteudo(self) -> str: |
| return "".join(self._linhas) |
|
|
| def salvar(self, diretorio: str | Path = ".") -> Path: |
| caminho = Path(diretorio) / self.nome_arquivo |
| caminho.parent.mkdir(parents=True, exist_ok=True) |
| caminho.write_text(self.conteudo(), encoding="utf-8") |
| return caminho |
|
|
| def hash_md5(self) -> str: |
| return hashlib.md5(self.conteudo().encode("utf-8")).hexdigest() |
|
|
|
|
| def criar_registro(tipo: str, *campos) -> str: |
| |
| partes = [tipo] |
| for campo in campos: |
| if campo is None: |
| partes.append("") |
| elif isinstance(campo, date): |
| partes.append(_fmt_data(campo)) |
| elif isinstance(campo, Decimal): |
| partes.append(_fmt_decimal(campo)) |
| elif isinstance(campo, float): |
| partes.append(_fmt_decimal(Decimal(str(campo)))) |
| else: |
| partes.append(str(campo).strip()) |
| return "|" + "|".join(partes) + "|\n" |
|
|
