534 lines
22 KiB
Python
534 lines
22 KiB
Python
from __future__ import annotations
|
||
|
||
import io
|
||
import re
|
||
import unicodedata as ud
|
||
from pathlib import Path
|
||
from typing import Dict, Union, Any
|
||
|
||
import pandas as pd
|
||
import pdfplumber
|
||
|
||
###############################################################################
|
||
# Normalização #
|
||
###############################################################################
|
||
_DASH_CHARS = "\u2010\u2011\u2012\u2013\u2014\u2212"
|
||
_NBSP_CHARS = "\u00A0\u202F\u2007"
|
||
_dash_trans = str.maketrans({c: "-" for c in _DASH_CHARS})
|
||
_nbsp_trans = str.maketrans({c: " " for c in _NBSP_CHARS})
|
||
|
||
|
||
def _normalize_line(s: str) -> str:
|
||
s = s.translate(_dash_trans).translate(_nbsp_trans)
|
||
s = ud.normalize("NFKC", s)
|
||
return re.sub(r"\s{2,}", " ", s.strip())
|
||
|
||
|
||
def _parse_money(value: str) -> float | None:
|
||
value = value.strip()
|
||
if value == "-":
|
||
return None
|
||
return float(value.replace(".", "").replace(",", "."))
|
||
|
||
###############################################################################
|
||
# Regex – Cabeçalhos #
|
||
###############################################################################
|
||
RX_MSISDN_HEADER = re.compile(r"Vantagens que seu plano oferece:\s*(?P<msisdn>\d{2}\s\d{5}-\d{4})", re.I)
|
||
RX_MSISDN_SEU_NUM = re.compile(r"SEU\s+NÚMERO\s+TIM\s+(?P<msisdn>\d{2}\s\d{5}-\d{4})", re.I)
|
||
RX_MSISDN_DETALHE = re.compile(r"Detalhamento de Serviços\s+N[°º]\s*(?P<msisdn>\d{2}\s\d{5}-\d{4})", re.I)
|
||
|
||
SECTION_HEADERS: Dict[str, str] = {
|
||
"IGNORE Ilimitados": r"^Detalhamento de Serviços Ilimitados",
|
||
"IGNORE Detalhamento": r"^Detalhamento de Serviç[io]s\b",
|
||
"Fatura Resumo": r"^FATURA\s+RESUMO",
|
||
"DANFE-COM": r"^DANFE-COM\b",
|
||
"Plano": r"^Plano\b",
|
||
"Mensalidades Adicionais": r"^MENSALIDADES\s+ADICIONAIS",
|
||
"Itens Eventuais": r"^ITENS\s+EVENTUAIS",
|
||
"TIM Viagem": r"^TIM\s+VIAGEM",
|
||
"SVA Detalhe Total": r"^Serviços\s+de\s+Valor\s+Adicionado\s+Total",
|
||
"Desconto Franquia": r"^Desconto\(s\)\s+Franquia",
|
||
"Desconto SVA": r"^Desconto\(s\)\s+Serviç[io]s",
|
||
"Franquia": r"^Franquia\s*\(s\)",
|
||
"SVA": r"^Serviços\s+de\s+valor\s+adicionado\(SVA\)",
|
||
"Chamadas Rede TIM": r"^CHAMADAS\s+DENTRO\s+DA\s+REDE\s+TIM",
|
||
"Chamadas Fora Rede TIM": r"^CHAMADAS\s+FORA\s+DA\s+REDE\s+TIM",
|
||
"Outros Valores": r"^OUTROS\s+VALORES",
|
||
"Deduções": r"^DEDUÇÕES",
|
||
"Roaming Internacional": r"^ROAMING\s+INTERNACIONAL",
|
||
"Cobranças de Terceiros": r"^COBRANÇAS\s+DE\s+TERCEIROS",
|
||
"Débitos de outras operadoras": r"^DÉBITOS\s+DE\s+OUTRAS\s+OPERADORAS"
|
||
}
|
||
SECTION_REGEX = {k: re.compile(v, re.I) for k, v in SECTION_HEADERS.items()}
|
||
RX_SECTION_TOTAL = re.compile(r"(?:R\$\s*)?(-?[\d\.]+,\d{2})")
|
||
|
||
# Patterns que encerram a seção atual sem iniciar uma nova
|
||
RX_SECTION_BREAK = re.compile(
|
||
r"^(Nota Fiscal de Servi|SEUS\s+DADOS|ITEM\s+QTDE\s+ICMS|TOTAL\s+TIM|"
|
||
r"Ficou\s+com\s+dúvidas|Bancos\s+Conveniados|Reservado\s+ao\s+fisco|"
|
||
r"Tipo:\s+N\s+-\s+Normal)",
|
||
re.I,
|
||
)
|
||
|
||
###############################################################################
|
||
# Regex – Linhas #
|
||
###############################################################################
|
||
RX_PERIOD = r"(?P<period>(?:\d{2}/\d{2}\s+a\s+\d{2}/\d{2}|-))"
|
||
RX_DAYS = r"(?P<days>(?:\d+|-))"
|
||
RX_VALUE = r"(?P<value>-?\d+,\d{2}|Incluído)"
|
||
RX_PARCEL = r"(?P<parcel>(?:\d+/\d+|-))"
|
||
|
||
# --- Plano / Desconto / Chamadas (layout padrão: QTY DESC PARCELA PERIOD DIAS VALOR) ---
|
||
RX_SIMPLE = re.compile(rf"^\s*(?P<qty>\d+)\s+(?P<desc>.+?)\s+{RX_PARCEL}\s+{RX_PERIOD}\s+{RX_DAYS}\s+{RX_VALUE}\s*$")
|
||
RX_CONSUMPTION = re.compile(rf"^\s*(?P<qty>\d+)\s+(?P<desc>.+?)\s+{RX_PARCEL}\s+(?P<franchise>Ilimitado|-)\s+(?P<consumption>\d{{1,3}}m\d{{2}}s)\s+{RX_PERIOD}\s+{RX_DAYS}\s+{RX_VALUE}\s*$")
|
||
RX_SUBTOTAL = re.compile(r"^Subtotal\s+(?P<value>-?\d+,\d{2})\s*$", re.I)
|
||
RX_DETAIL = re.compile(r"^\d+\s+.+?\s+(-?\d+,\d{2})\s*$")
|
||
# Fatura Resumo
|
||
RX_RESUMO = re.compile(r"^\s*(?P<desc>.+?)\s+R\$\s*(?P<value>-?\d+,\d{2})\s*$")
|
||
RX_TOTAL_GERAL = re.compile(r"^Total\s+geral\s+R\$\s*(?P<value>-?\d+,\d{2})", re.I)
|
||
RX_FATURA_METADATA = re.compile(
|
||
r"FATURA\s+PER[ÍI]ODO\s+EMISS[ÃA]O\s+POSTAGEM\s+"
|
||
r"(?P<fatura>\S+)\s+"
|
||
r"(?P<period>\d{2}/\d{2}\s+a\s+\d{2}/\d{2})\s+"
|
||
r"(?P<emissao>\d{2}/\d{2}/\d{4})\s+"
|
||
r"(?P<postagem>\d{2}/\d{2}/\d{4})",
|
||
re.I,
|
||
)
|
||
RX_DANFE_TOTAL = re.compile(r"^Total\s+geral\s+R\$\s*(?P<value>-?\d+(?:\.\d{3})*,\d{2})", re.I)
|
||
RX_DANFE_ROW = re.compile(
|
||
r"^(?P<desc>.+?)\s+"
|
||
r"(?P<unit>[A-Z]{2})\s+"
|
||
r"(?P<qty>\d+(?:,\d+)?)\s+"
|
||
r"(?P<preco_unit>-?\d+(?:\.\d{3})*,\d{2})\s+"
|
||
r"(?P<pis_cofins>-?\d+(?:\.\d{3})*,\d{2}|-)\s+"
|
||
r"(?P<bc_icms>-?\d+(?:\.\d{3})*,\d{2}|-)\s+"
|
||
r"(?P<aliq_icms>\d+(?:,\d+)?%|-)\s+"
|
||
r"(?P<icms>-?\d+(?:\.\d{3})*,\d{2}|-)\s+"
|
||
r"(?P<value>-?\d+(?:\.\d{3})*,\d{2})$"
|
||
)
|
||
|
||
# --- Itens Eventuais (layout: QTY DESC PARCELA FRANQUIA CONSUMO PERIODO DIAS VALOR) ---
|
||
# Com consumo real (ex: 19,77GB) ou consumo zerado ("0")
|
||
RX_EVENTUAIS = re.compile(
|
||
r"^\s*(?P<qty>\d+)\s+(?P<desc>.+?)\s+-\s+-\s+"
|
||
r"(?P<consumption>(?:\S+(?:GB|MB|KB)|0))\s+-\s+-\s+"
|
||
r"(?P<value>-?\d+,\d{2})\s*$"
|
||
)
|
||
# Sem consumo (todos "-")
|
||
RX_EVENTUAIS_NO_CONS = re.compile(
|
||
r"^\s*(?P<qty>\d+)\s+(?P<desc>.+?)"
|
||
r"(?:\s+-){5}\s+"
|
||
r"(?P<value>-?\d+,\d{2})\s*$"
|
||
)
|
||
|
||
# --- SVA Detalhe Total (layout: # DATA HORA ORIGEM DESC NUMERO TIPO PACOTE - - VALOR) ---
|
||
# Exemplo: 1 09/04/25 - 02:54:18 RJ AREA 21 TIM Saude Mensal 00700001003511 N FP - - 14,99
|
||
RX_SVA_DETAIL = re.compile(
|
||
r"^\s*(?P<seq>\d+)\s+"
|
||
r"(?P<date>\d{2}/\d{2}/\d{2})\s+-\s+(?P<time>\d{2}:\d{2}:\d{2})\s+"
|
||
r"\S+\s+AREA\s+\d{2}\s+" # origem (ex: RJ AREA 21)
|
||
r"(?P<desc>.+?)\s+"
|
||
r"(?P<number>\d{10,})\s+"
|
||
r"[A-Z/]+\s+[A-Z]+\s+"
|
||
r"-\s+-\s+"
|
||
r"(?P<value>-?\d+,\d{2})\s*$"
|
||
)
|
||
# Linha de total SVA (ex: "3 - 2 2 29,98")
|
||
RX_SVA_SUMMARY = re.compile(
|
||
r"^\s*(?P<seq>\d+)\s+-\s+\d+\s+\d+\s+(?P<value>-?\d+,\d{2})\s*$"
|
||
)
|
||
|
||
# --- Informações Complementares (layout: QTY DESC PARCELA PERÍODO DIAS VALOR) ---
|
||
RX_INFO_COMPL = re.compile(
|
||
r"^\s*(?P<qty>\d+)\s+(?P<desc>.+?)\s+"
|
||
r"(?P<parcel>(?:\d+/\d+|-))\s+"
|
||
r"(?P<period>\d{2}/\d{2}\s+a\s+\d{2}/\d{2})\s+"
|
||
r"(?P<days>\d+)\s+"
|
||
r"(?P<value>-?\d+,\d{2})\s*$"
|
||
)
|
||
# --- Chamadas resumo (layout: QTY DESC PARCELA FRANQUIA CONSUMO PERIODO DIAS VALOR) ---
|
||
RX_CHAMADAS = re.compile(
|
||
r"^\s*(?P<qty>\d+)\s+(?P<desc>.+?)\s+-\s+-\s+"
|
||
r"(?P<consumption>\d{1,3}m\d{2}s)\s+-\s+-\s+"
|
||
r"(?P<value>-?\d+,\d{2})\s*$"
|
||
)
|
||
|
||
# --- Mensalidades Adicionais (layout: QTY DESC PARCELA FRANQUIA CONSUMO PERIODO DIAS VALOR) ---
|
||
# Ex.: "1 Apple Music SVA Mes - - 0 - - 21,40" (consumo "0" e sem período)
|
||
# "1 Internet 30GB - - - 25/04 a 24/05 30 0,00"
|
||
RX_MENSALIDADE = re.compile(
|
||
rf"^\s*(?P<qty>\d+)\s+(?P<desc>.+?)\s+{RX_PARCEL}\s+"
|
||
rf"(?P<franchise>Ilimitado|-)\s+(?P<consumption>\S+)\s+"
|
||
rf"{RX_PERIOD}\s+{RX_DAYS}\s+{RX_VALUE}\s*$"
|
||
)
|
||
|
||
###############################################################################
|
||
# Parser #
|
||
###############################################################################
|
||
class TimBillParser:
|
||
"""Extrai e estrutura faturas TIM."""
|
||
|
||
def __init__(self, *, x_tolerance: float = 1.5, y_tolerance: float = 3.0):
|
||
self.x_tol = x_tolerance
|
||
self.y_tol = y_tolerance
|
||
self.data: Dict[str, pd.DataFrame] = {}
|
||
|
||
# ---------------- API pública ----------------
|
||
def parse_pdf(self, pdf_bytes: io.BytesIO) -> Dict[str, pd.DataFrame]:
|
||
|
||
# 1. Extrai texto bruto de todas as páginas
|
||
with pdfplumber.open(pdf_bytes) as pdf:
|
||
raw_text = "\n".join(
|
||
page.extract_text(x_tolerance=self.x_tol,
|
||
y_tolerance=self.y_tol) or ""
|
||
for page in pdf.pages
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# 2. Captura o bloco FATURA RESUMO inteiro
|
||
bloco_pat = re.compile(
|
||
r"FATURA\s+RESUMO(?P<body>.*?)Total\s+geral\s+R\$\s*(?P<total>-?[\d\.,]+)",
|
||
re.S | re.I,
|
||
)
|
||
resumo_items = []
|
||
resumo_metadata = self._extract_fatura_metadata(raw_text)
|
||
if resumo_metadata.get("period"):
|
||
resumo_items.append(
|
||
dict(qty=None, desc="PERÍODO", parcel=None, period=resumo_metadata["period"],
|
||
days=None, value=None, franchise=None, consumption=None, msisdn=None,
|
||
emissao=None, section_total=None)
|
||
)
|
||
if resumo_metadata.get("emissao"):
|
||
resumo_items.append(
|
||
dict(qty=None, desc="EMISSÃO", parcel=None, period=None,
|
||
days=None, value=None, franchise=None, consumption=None, msisdn=None,
|
||
emissao=resumo_metadata["emissao"], section_total=None)
|
||
)
|
||
m_resumo = bloco_pat.search(raw_text)
|
||
if m_resumo:
|
||
corpo = m_resumo.group("body")
|
||
total_val = float(m_resumo.group("total").replace(".", "").replace(",", "."))
|
||
for ln in corpo.splitlines():
|
||
ln = _normalize_line(ln)
|
||
if not ln:
|
||
continue
|
||
m_ln = re.match(r"(.+?)\s+R\$\s*([-\d\.,]+)", ln)
|
||
if m_ln:
|
||
resumo_items.append(
|
||
dict(qty=None, desc=m_ln.group(1), parcel=None, period=None,
|
||
days=None,
|
||
value=float(m_ln.group(2).replace(".", "").replace(",", ".")),
|
||
franchise=None, consumption=None, msisdn=None,
|
||
section_total=None)
|
||
)
|
||
resumo_items.append(
|
||
dict(qty=None, desc="Total geral", parcel=None, period=None,
|
||
days=None, value=total_val, franchise=None, consumption=None,
|
||
msisdn=None, section_total=None)
|
||
)
|
||
# 3. Remove o bloco para que não polua a etapa linha-a-linha
|
||
raw_text = raw_text.replace(m_resumo.group(0), "")
|
||
|
||
# ------------------------------------------------------------------
|
||
# 4. Processa o restante normalmente
|
||
dfs = self._parse_text(raw_text)
|
||
if resumo_items:
|
||
dfs["Fatura Resumo"] = pd.DataFrame(resumo_items)
|
||
|
||
self.data = dfs
|
||
return dfs
|
||
|
||
# ---------------- interno -------------------
|
||
def _extract_fatura_metadata(self, text: str) -> dict[str, str]:
|
||
normalized_lines = [_normalize_line(line) for line in text.splitlines()]
|
||
normalized_text = " ".join(line for line in normalized_lines if line)
|
||
if m := RX_FATURA_METADATA.search(normalized_text):
|
||
return {
|
||
"period": m.group("period"),
|
||
"emissao": m.group("emissao"),
|
||
}
|
||
return {}
|
||
|
||
def _parse_text(self, text: str) -> Dict[str, pd.DataFrame]:
|
||
"""Quebra o texto extraído em DataFrames por seção."""
|
||
buf: Dict[str, list] = {k: [] for k in SECTION_HEADERS if not k.startswith("IGNORE")}
|
||
current_sec = current_msisdn = None
|
||
current_total: float | None = None
|
||
last_sva_detail_item: dict | None = None
|
||
|
||
for raw in text.splitlines():
|
||
line = _normalize_line(raw)
|
||
if not line:
|
||
continue
|
||
|
||
# cabeçalho de MSISDN (3 variantes)
|
||
for rx_ms in (RX_MSISDN_HEADER, RX_MSISDN_SEU_NUM, RX_MSISDN_DETALHE):
|
||
if (m := rx_ms.search(line)):
|
||
current_msisdn = m.group("msisdn").replace(" ", "")
|
||
break
|
||
else:
|
||
m = None
|
||
if m:
|
||
continue
|
||
|
||
# section break — encerra seção atual sem iniciar outra
|
||
if RX_SECTION_BREAK.match(line):
|
||
current_sec = None
|
||
continue
|
||
|
||
# detecta novo cabeçalho (a menos que estejamos em Fatura Resumo)
|
||
if current_sec != "Fatura Resumo":
|
||
sec = next((s for s, rx in SECTION_REGEX.items() if rx.match(line)), None)
|
||
else:
|
||
sec = None
|
||
|
||
if (sec in ("IGNORE Ilimitados", "IGNORE Detalhamento")):
|
||
current_sec = None
|
||
continue
|
||
if sec is not None:
|
||
current_sec = sec
|
||
last_sva_detail_item = None
|
||
mt = RX_SECTION_TOTAL.search(line)
|
||
current_total = float(mt.group(1).replace(".", "").replace(",", ".")) if mt else None
|
||
continue
|
||
|
||
if not current_sec:
|
||
if line[0].isdigit():
|
||
item = self._parse_sva_detail(line)
|
||
if item:
|
||
current_sec = "SVA Detalhe Total"
|
||
current_total = None
|
||
item["msisdn"] = current_msisdn
|
||
item["section_total"] = None
|
||
buf.setdefault(current_sec, []).append(item)
|
||
last_sva_detail_item = item
|
||
continue
|
||
|
||
# pula cabeçalhos de página
|
||
if re.match(r"^Página\s+\d+\s+de\s+\d+", line, re.I):
|
||
continue
|
||
|
||
if current_sec == "DANFE-COM":
|
||
item = self._parse_danfe(line)
|
||
if item:
|
||
item["msisdn"] = None
|
||
item["section_total"] = current_total
|
||
buf.setdefault(current_sec, []).append(item)
|
||
if item.get("is_total"):
|
||
current_sec = None
|
||
continue
|
||
|
||
if current_sec == "SVA Detalhe Total":
|
||
if self._is_sva_header_line(line):
|
||
continue
|
||
if RX_SVA_SUMMARY.match(line):
|
||
last_sva_detail_item = None
|
||
continue
|
||
if line[0].isdigit():
|
||
item = self._parse_sva_detail(line)
|
||
if item:
|
||
item["msisdn"] = current_msisdn
|
||
item["section_total"] = current_total
|
||
buf.setdefault(current_sec, []).append(item)
|
||
last_sva_detail_item = item
|
||
continue
|
||
if last_sva_detail_item:
|
||
last_sva_detail_item["desc"] = (
|
||
f"{last_sva_detail_item['desc']} {line}"
|
||
).strip()
|
||
continue
|
||
|
||
# -------- linhas padrão --------
|
||
if not (line[0].isdigit() or line.lower().startswith("subtotal")):
|
||
continue
|
||
|
||
item = self._parse_item_line(line, current_sec)
|
||
if item:
|
||
item["msisdn"] = current_msisdn
|
||
item["section_total"] = current_total
|
||
buf.setdefault(current_sec, []).append(item)
|
||
|
||
return {s: pd.DataFrame(lst) for s, lst in buf.items() if lst}
|
||
|
||
# ------------------------------------------------------------------
|
||
def _parse_item_line(self, line: str, section: str = "") -> dict | None:
|
||
"""Parseia uma linha de item de acordo com a seção."""
|
||
|
||
# Ignora subtotal
|
||
if RX_SUBTOTAL.match(line):
|
||
return None
|
||
|
||
# ── Mensalidades Adicionais ──
|
||
if section == "Mensalidades Adicionais":
|
||
return self._parse_mensalidades(line)
|
||
|
||
# ── Itens Eventuais ──
|
||
if section == "Itens Eventuais":
|
||
return self._parse_eventuais(line)
|
||
|
||
# ── SVA Detalhe Total ──
|
||
if section == "SVA Detalhe Total":
|
||
return self._parse_sva_detail(line)
|
||
|
||
# ── DANFE-COM ──
|
||
if section == "DANFE-COM":
|
||
return self._parse_danfe(line)
|
||
|
||
# ── Chamadas Rede TIM / Chamadas Fora Rede TIM ──
|
||
if section.startswith("Chamadas"):
|
||
return self._parse_chamadas(line)
|
||
|
||
# ── Plano / Franquia / SVA / Desconto Franquia / Desconto SVA (layout padrão) ──
|
||
return self._parse_standard(line)
|
||
|
||
# ------------------------------------------------------------------
|
||
def _parse_danfe(self, line: str) -> dict | None:
|
||
"""DANFE-COM: linhas da tabela de itens e total geral."""
|
||
if line.upper().startswith("ITENS "):
|
||
return None
|
||
|
||
if (m := RX_DANFE_TOTAL.match(line)):
|
||
return {
|
||
"desc": "Total geral",
|
||
"unit": None,
|
||
"qty": None,
|
||
"preco_unit": None,
|
||
"pis_cofins": None,
|
||
"bc_icms": None,
|
||
"aliq_icms": None,
|
||
"icms": None,
|
||
"value": _parse_money(m.group("value")),
|
||
"is_total": True,
|
||
}
|
||
|
||
if (m := RX_DANFE_ROW.match(line)):
|
||
d = m.groupdict()
|
||
return {
|
||
"desc": d["desc"].strip(),
|
||
"unit": d["unit"],
|
||
"qty": float(d["qty"].replace(",", ".")),
|
||
"preco_unit": _parse_money(d["preco_unit"]),
|
||
"pis_cofins": _parse_money(d["pis_cofins"]),
|
||
"bc_icms": _parse_money(d["bc_icms"]),
|
||
"aliq_icms": None if d["aliq_icms"] == "-" else d["aliq_icms"],
|
||
"icms": _parse_money(d["icms"]),
|
||
"value": _parse_money(d["value"]),
|
||
"is_total": False,
|
||
}
|
||
|
||
return None
|
||
|
||
# ------------------------------------------------------------------
|
||
def _parse_standard(self, line: str) -> dict | None:
|
||
"""Regex padrão: QTY DESC PARCELA [FRANCHISE CONSUMPTION] PERIOD DAYS VALUE."""
|
||
for rx in (RX_CONSUMPTION, RX_SIMPLE):
|
||
if (m := rx.match(line)):
|
||
d = m.groupdict()
|
||
|
||
# --- corrige parcel dentro de desc (caso d["parcel"] == "-") ----
|
||
if d["parcel"] == "-":
|
||
tail = re.search(r"\b(\d+/\d+)$", d["desc"])
|
||
if tail:
|
||
d["parcel"] = tail.group(1)
|
||
d["desc"] = d["desc"][: tail.start()].rstrip(" -")
|
||
# ----------------------------------------------------------------
|
||
|
||
# limpa trailing dashes e valores de franquia residuais do desc
|
||
d["desc"] = re.sub(r"(\s+-\s+\d+(?:GB|MB|KB))(?:\s+-)*\s*$|(?:\s+-)+\s*$", "", d["desc"]).strip()
|
||
|
||
value = d["value"]
|
||
d["qty"] = float(d["qty"])
|
||
d["days"] = None if d.get("days") in (None, "-") else int(d["days"])
|
||
d["_is_included_value"] = value == "Incluído"
|
||
d["value"] = 0.0 if d["_is_included_value"] else float(value.replace(",", "."))
|
||
return d
|
||
return None
|
||
|
||
# ------------------------------------------------------------------
|
||
def _parse_eventuais(self, line: str) -> dict | None:
|
||
"""Itens Eventuais: QTY DESC PARCELA FRANQUIA CONSUMO PERIODO DIAS VALUE."""
|
||
# Tenta com consumo real (ex: 19,77GB)
|
||
if (m := RX_EVENTUAIS.match(line)):
|
||
return {
|
||
"qty": float(m.group("qty")),
|
||
"desc": m.group("desc").strip(),
|
||
"parcel": None,
|
||
"period": None,
|
||
"days": None,
|
||
"value": float(m.group("value").replace(",", ".")),
|
||
"franchise": None,
|
||
"consumption": None if m.group("consumption") == "0" else m.group("consumption"),
|
||
}
|
||
# Tenta sem consumo (todos "-")
|
||
if (m := RX_EVENTUAIS_NO_CONS.match(line)):
|
||
return {
|
||
"qty": float(m.group("qty")),
|
||
"desc": m.group("desc").strip(),
|
||
"parcel": None,
|
||
"period": None,
|
||
"days": None,
|
||
"value": float(m.group("value").replace(",", ".")),
|
||
"franchise": None,
|
||
"consumption": None,
|
||
}
|
||
return None
|
||
|
||
# ------------------------------------------------------------------
|
||
def _parse_mensalidades(self, line: str) -> dict | None:
|
||
"""Mensalidades Adicionais: QTY DESC PARCELA FRANQUIA CONSUMO PERIODO DIAS VALOR."""
|
||
if (m := RX_MENSALIDADE.match(line)):
|
||
d = m.groupdict()
|
||
d["qty"] = float(d["qty"])
|
||
d["desc"] = d["desc"].strip()
|
||
d["parcel"] = None if d["parcel"] == "-" else d["parcel"]
|
||
d["franchise"] = None if d["franchise"] == "-" else d["franchise"]
|
||
d["consumption"] = None if d["consumption"] in ("-", "0") else d["consumption"]
|
||
d["period"] = None if d["period"] == "-" else d["period"]
|
||
d["days"] = None if d["days"] == "-" else int(d["days"])
|
||
value = d["value"]
|
||
d["_is_included_value"] = value == "Incluído"
|
||
d["value"] = 0.0 if d["_is_included_value"] else float(value.replace(",", "."))
|
||
return d
|
||
# Layout de 6 colunas (sem franquia/consumo) → regex padrão
|
||
return self._parse_standard(line)
|
||
|
||
# ------------------------------------------------------------------
|
||
def _is_sva_header_line(self, line: str) -> bool:
|
||
return bool(
|
||
re.match(r"^(DURAÇÃO/VOLUME|#\s+DATA\s*/\s*HORA)\b", line, re.I)
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
def _parse_sva_detail(self, line: str) -> dict | None:
|
||
"""SVA Detalhe Total: linhas com data/hora e linha de totalização."""
|
||
# Linha de detalhe com data/hora
|
||
if (m := RX_SVA_DETAIL.match(line)):
|
||
return {
|
||
"qty": 1.0,
|
||
"desc": m.group("desc").strip(),
|
||
"parcel": None,
|
||
"period": m.group("date"),
|
||
"days": None,
|
||
"value": float(m.group("value").replace(",", ".")),
|
||
"franchise": None,
|
||
"consumption": None,
|
||
}
|
||
return None
|
||
|
||
# ------------------------------------------------------------------
|
||
def _parse_chamadas(self, line: str) -> dict | None:
|
||
"""Chamadas Rede TIM / Fora Rede: QTY DESC - - CONSUMPTION - - VALUE."""
|
||
if (m := RX_CHAMADAS.match(line)):
|
||
return {
|
||
"qty": float(m.group("qty")),
|
||
"desc": m.group("desc").strip(),
|
||
"parcel": None,
|
||
"period": None,
|
||
"days": None,
|
||
"value": float(m.group("value").replace(",", ".")),
|
||
"franchise": None,
|
||
"consumption": m.group("consumption"),
|
||
}
|
||
# Fallback para o padrão (algumas chamadas usam formato padrão)
|
||
return self._parse_standard(line)
|