from __future__ import annotations import re import unicodedata as ud from typing import Any, Dict, List, Tuple import pandas as pd # --------------------------------------------------------------------------- # Regex para linha de VAS # --------------------------------------------------------------------------- RX_VAS_LINE = re.compile( r""" ^\d+\s+ # índice "#" (?P\d{2}/\d{2}/\d{2}) # DD/MM/YY \s+-\s+\d{2}:\d{2}:\d{2}\s+ # " - HH:MM:SS " [A-Z]{2}\sAREA\s\d+\s+ # "SP AREA 11" (?P.+?)\s+ # nome do serviço \d{8,} # número chamado """, re.X, ) # --------------------------------------------------------------------------- class PDFBillingProcessor: """ Pós-processamento dos DataFrames gerados pelo TimBillParser. """ SECTIONS = [ "Fatura Resumo", "Plano", "DANFE-COM", "Descontos", "Itens Eventuais", "Mensalidades Adicionais", "TIM Viagem", "Outros Valores", "Chamadas Rede TIM", "SVA Detalhe Total", "Serviços Bundle Inclusos", "Deduções", "Roaming Internacional", "Cobranças de Terceiros", "Débitos de outras operadoras", ] STRATEGIC_SERVICE_SECTIONS = { "SVA Detalhe Total", "Itens Eventuais", "Serviços Bundle Inclusos", "Mensalidades Adicionais", "Outros Valores", "Cobranças de Terceiros", } SECTION_CLASS_MAP = { "SVA Detalhe Total": "avulso", "Itens Eventuais": "avulso", "Serviços Bundle Inclusos": "bundle", } CLASSE_VERB_MAP = { "avulso": "cancelar", "bundle": "falar sobre", "estrategico": "falar sobre", } STRATEGIC_SERVICES = [ "Apple Music", "Deezer", "Disney", "Fuze", "Forge", "HBO", "Looke", "Max Mensal", "Netflix", "Paramount", "TIM Cloud Gaming", "YouTube", "Globoplay", "Amazon Prime" ] DIGIT_WORDS = { "0": "zero", "1": "um", "2": "dois", "3": "tres", "4": "quatro", "5": "cinco", "6": "seis", "7": "sete", "8": "oito", "9": "nove", } def __init__(self, *, include_danfe: bool = False): self.include_danfe = include_danfe self._strategic_service_patterns = [ re.compile(rf"(?:^|\s){re.escape(self._service_match_key(service))}(?:\s|$)") for service in self.STRATEGIC_SERVICES ] def remove_parentheses(self, text: str) -> str: return re.sub(r'\s*\([^)]*\)', '', str(text)).strip() def remove_plan_recognition_marker(self, text: str) -> str: return re.sub( r"\s*\(\s*\d+\s*/\s*P[ÓO]S\s*/\s*SMP\s*\)", "", str(text), flags=re.I, ).strip() def _format_plan_version_number(self, text: str) -> str: return re.sub(r"(? str: return re.sub(r"\D", "", str(text)) def _vocalize_digits(self, digits: str) -> str: return " ".join(self.DIGIT_WORDS[d] for d in digits if d in self.DIGIT_WORDS) def _collect_msisdns(self, node: Any, found: set) -> None: """Percorre recursivamente o payload coletando todos os msisdn presentes.""" if isinstance(node, dict): for key, value in node.items(): if key == "msisdn" and value is not None: msisdn = self._format_msisdn(value) if msisdn: found.add(msisdn) else: self._collect_msisdns(value, found) elif isinstance(node, list): for item in node: self._collect_msisdns(item, found) def _build_vocalized_msisdn(self, final_output: Dict[str, Any]) -> Dict[str, str]: """Mapeia cada msisdn da fatura para a vocalização dos seus 4 últimos dígitos.""" found: set = set() for key, value in final_output.items(): if re.fullmatch(r"\d+", str(key)): found.add(self._format_msisdn(key)) self._collect_msisdns(value, found) return { msisdn: self._vocalize_digits(msisdn[-4:]) for msisdn in sorted(found) } def _clean_plan_name(self, desc: str, *, format_version_number: bool = True) -> str: text = self.remove_parentheses(self.remove_plan_recognition_marker(desc)) if format_version_number: text = self._format_plan_version_number(text) return text def _discount_match_key(self, desc: str) -> str: text = self._clean_plan_name(desc) text = ud.normalize("NFKD", text) text = "".join(ch for ch in text if not ud.combining(ch)) text = text.casefold() text = re.sub(r"\b\d+/\d+\b", " ", text) text = re.sub(r"\b\d+\b", " ", text) text = re.sub(r"[^a-z]+", " ", text) return re.sub(r"\s+", " ", text).strip() def _service_match_key(self, desc: str) -> str: text = ud.normalize("NFKD", str(desc)) text = "".join(ch for ch in text if not ud.combining(ch)) text = text.casefold() text = re.sub(r"[^a-z0-9]+", " ", text) return re.sub(r"\s+", " ", text).strip() def _is_strategic_service(self, desc: Any) -> bool: desc_key = self._service_match_key(str(desc or "")) return any(pattern.search(desc_key) for pattern in self._strategic_service_patterns) def _is_controle_plan(self, desc: Any) -> bool: desc_key = self._service_match_key(str(desc or "")) return bool(re.search(r"\b(?:controle|ctrl|crtl)\b", desc_key)) def calculate_discount(self, desc_match: str, msisdn: str, descontos_df: pd.DataFrame) -> float: # Filtra por msisdn e verifica se o desc_match está contido no desc descontos_filtrados = descontos_df[ (descontos_df['msisdn'] == msisdn) & (descontos_df['desc'].str.lower().str.contains(desc_match.lower(), na=False)) ] return descontos_filtrados['value'].sum() # ------------------------------------------------------------------ def normalize(self, raw: Dict[str, pd.DataFrame]) -> Dict[str, Any]: all_msisdns = set() for sec, df in raw.items(): if sec != "Fatura Resumo" and "msisdn" in df.columns: all_msisdns.update(df["msisdn"].dropna().astype(str).unique()) final_output: Dict[str, Any] = {} # Processa Fatura Resumo primeiro (independente de MSISDN) if "Fatura Resumo" in raw: fatura_raw = {"Fatura Resumo": raw["Fatura Resumo"]} fatura_dfs = self._clean_dataframes(fatura_raw) fatura_json = self._build_final_json(fatura_dfs) if "Fatura Resumo" in fatura_json: final_output["Fatura Resumo"] = fatura_json["Fatura Resumo"] if self.include_danfe and "DANFE-COM" in raw: danfe_json = self._build_danfe_payload(raw["DANFE-COM"], raw) if danfe_json: final_output["DANFE-COM"] = danfe_json if not all_msisdns: # Se não houver MSISDNs, processa tudo como global dfs = self._clean_dataframes(raw) dfs = self._process_plano(dfs) dfs = self._process_eventuais(dfs) dfs = self._process_vas(dfs) dfs = self._process_descontos(dfs) fallback_json = self._build_final_json(dfs) for k, v in fallback_json.items(): if k not in final_output: final_output[k] = v final_output["vocalized_msisdn"] = self._build_vocalized_msisdn(final_output) return final_output for msisdn in all_msisdns: msisdn_raw = {} for sec, df in raw.items(): if sec == "Fatura Resumo": continue if "msisdn" in df.columns: mask = df["msisdn"].astype(str) == msisdn filtered_df = df[mask].copy() if not filtered_df.empty: msisdn_raw[sec] = filtered_df else: # Seções sem MSISDN podem ser globais? (Ex: Fatura Resumo já tratada) pass if msisdn_raw: dfs = self._clean_dataframes(msisdn_raw) dfs = self._process_plano(dfs) dfs = self._process_eventuais(dfs) dfs = self._process_vas(dfs) dfs = self._process_descontos(dfs) msisdn_json = self._build_final_json(dfs) if msisdn_json: final_output[self._format_msisdn(msisdn)] = msisdn_json final_output["vocalized_msisdn"] = self._build_vocalized_msisdn(final_output) return final_output def _clean_dataframes(self, raw: Dict[str, pd.DataFrame]) -> Dict[str, pd.DataFrame]: # 1) remove value == 0 exceto de fatura resumo e plano (que pode ter o bundle) dfs = { sec: (df if sec in ["Fatura Resumo", "Plano"] else df[df.value != 0.0]).copy() for sec, df in raw.items() if "value" in df.columns } # 2) drop colunas totalmente NaN dfs = { sec: df.dropna(axis=1, how="all") for sec, df in dfs.items() if not df.dropna(axis=1, how="all").empty } # 3) remove sufixo " - -" do campo desc for df in dfs.values(): if "desc" in df.columns: df["desc"] = df["desc"].str.replace(r"\s*- -\s*$", "", regex=True) df["desc"] = df["desc"].str.strip() return dfs def _process_plano(self, dfs: Dict[str, pd.DataFrame]) -> Dict[str, pd.DataFrame]: if "Plano" not in dfs: return dfs plano_df = dfs["Plano"] if 'msisdn' in plano_df.columns: plano_df['msisdn'] = plano_df['msisdn'].astype(str) # Separa somente itens que vieram textualmente como "Incluído". # Planos dependentes podem aparecer como R$ 0,00 e continuam sendo Plano. if "_is_included_value" in plano_df.columns: included_mask = plano_df["_is_included_value"].fillna(False).astype(bool) else: included_mask = plano_df.value == 0.0 bundle_df = plano_df[included_mask].copy() plano_df = plano_df[~included_mask].copy() if not bundle_df.empty: bundle_df["value"] = "Incluído" if "section_total" in bundle_df.columns: bundle_df = bundle_df.drop(columns=["section_total"]) dfs["Serviços Bundle Inclusos"] = bundle_df # Seleciona planos principais mask_plan = plano_df["desc"].str.contains(r"PÓS/SMP", case=False, na=False) if mask_plan.sum() == 0: mask_plan = plano_df["desc"].str.match(r"(?i)^tim", case=False) # Seleciona descontos descontos_df = plano_df[plano_df['desc'].apply(lambda x: 'desc' in str(x).lower() and 'tim' in str(x).lower())] plano_df = plano_df[mask_plan].copy() plano_df["desc"] = plano_df["desc"].apply(self.remove_plan_recognition_marker) dfs["_descontos_temp"] = descontos_df if plano_df.empty: dfs.pop("Plano", None) else: dfs["Plano"] = plano_df return dfs def _process_eventuais(self, dfs: Dict[str, pd.DataFrame]) -> Dict[str, pd.DataFrame]: if "Itens Eventuais" in dfs: ie_df = dfs["Itens Eventuais"] mask_remove = (ie_df["value"] == 0) | (ie_df["desc"].str.contains("Serviços de Valor Adicionado Conteúdo", case=False, na=False)) ie_df = ie_df[~mask_remove].copy() if ie_df.empty: dfs.pop("Itens Eventuais") else: dfs["Itens Eventuais"] = ie_df return dfs def _process_vas(self, dfs: Dict[str, pd.DataFrame]) -> Dict[str, pd.DataFrame]: """Remove de 'Mensalidades Adicionais' os serviços recorrentes que já vêm detalhados (com data de ativação) em 'SVA Detalhe Total', evitando duplicata. A seção 'Serviços de Valor Adicionado Total' do PDF é o detalhamento que reúne tanto os itens eventuais quanto as mensalidades recorrentes; quando o mesmo serviço aparece nas duas seções, mantemos a cópia detalhada do SVA.""" sva_df = dfs.get("SVA Detalhe Total") if sva_df is None or sva_df.empty or "desc" not in sva_df.columns: return dfs detalhados = { (self._format_msisdn(rec.get("msisdn")), self._service_match_key(rec.get("desc"))) for rec in sva_df.to_dict(orient="records") } for section in ("Mensalidades Adicionais", "Itens Eventuais"): df = dfs.get(section) if df is None or df.empty or "desc" not in df.columns: continue mask_dup = df.apply( lambda r: ( self._format_msisdn(r.get("msisdn")), self._service_match_key(r.get("desc")), ) in detalhados, axis=1, ) if mask_dup.any(): df = df[~mask_dup].copy() if df.empty: dfs.pop(section, None) else: dfs[section] = df return dfs def _process_descontos(self, dfs: Dict[str, pd.DataFrame]) -> Dict[str, pd.DataFrame]: if "_descontos_temp" in dfs: descontos_df = dfs.pop("_descontos_temp") if not descontos_df.empty: if "section_total" in descontos_df.columns: descontos_df = descontos_df.drop(columns=["section_total"]) descontos_df['installment'] = descontos_df['desc'].str.extract(r'(\d+/\d+)') descontos_df['desc'] = descontos_df['desc'].apply(self._format_plan_version_number) if "Descontos" in dfs and not dfs["Descontos"].empty: dfs["Descontos"] = pd.concat( [dfs["Descontos"], descontos_df], ignore_index=True, ) else: dfs["Descontos"] = descontos_df return dfs def _clean_record(self, rec: Dict[str, Any]) -> Dict[str, Any]: redundant_values = { "qty": [1.0, 1], "parcel": ["-", None], "franchise": [None, "null", "-"], "consumption": [None, "null"], "period": ["-", None], "value": [None], "emissao": [None, "null", "-"], } cleaned_rec = {} for k, v in rec.items(): if k == "section_total" or k.startswith("_"): continue if k == "msisdn": v = self._format_msisdn(v) if pd.isna(v): v = None elif k == "value" and isinstance(v, (int, float)): v = self._round_money(v) elif k == "days" and isinstance(v, float) and v.is_integer(): v = int(v) if k in redundant_values and v in redundant_values[k]: continue cleaned_rec[k] = v return cleaned_rec def _records_for_section(self, df: pd.DataFrame) -> List[Dict[str, Any]]: return [ self._clean_record(rec) for rec in df.to_dict(orient="records") ] def _split_strategic_records( self, df: pd.DataFrame, section: str, ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: records = self._records_for_section(df) if section not in self.STRATEGIC_SERVICE_SECTIONS: return records, [] regular_records = [] strategic_records = [] for record in records: if "desc" in record and self._is_strategic_service(record["desc"]): record["estrategico"] = True strategic_records.append(record) else: regular_records.append(record) return regular_records, strategic_records def _annotate_class( self, records: List[Dict[str, Any]], section: str, ) -> List[Dict[str, Any]]: section_classe = self.SECTION_CLASS_MAP.get(section) if not section_classe: return records for rec in records: # Itens estratégicos dobrados nesta seção (commit 7fc38a4e) não # herdam a classe da seção (avulso): mantêm classe=estrategico e # verbo "falar sobre". Os demais seguem o default da seção. classe = "estrategico" if rec.get("estrategico") is True else section_classe rec["classe"] = classe verb = self.CLASSE_VERB_MAP.get(classe) if verb: rec["verb"] = verb return records def _round_money(self, value: float) -> float: return round(float(value) + 0.0, 2) def _is_discount_desc(self, desc: str) -> bool: return bool(re.match(r"(?i)^\s*desc(?:onto)?\b", desc)) def _danfe_record(self, rec: Dict[str, Any]) -> Dict[str, Any]: cleaned = {} for k, v in rec.items(): if k in {"section_total", "msisdn", "is_total"}: continue if pd.isna(v): v = None elif k in {"preco_unit", "pis_cofins", "bc_icms", "icms", "value"} and isinstance(v, (int, float)): v = self._round_money(v) elif k == "qty" and isinstance(v, float) and v.is_integer(): v = int(v) cleaned[k] = v return cleaned def _danfe_item_payload(self, rec: Dict[str, Any]) -> Dict[str, Any]: item = self._danfe_record(rec) valor_bruto = self._round_money(item.pop("value", 0.0) or 0.0) item["valor_bruto"] = valor_bruto item["total_descontos"] = 0.0 item["valor_final"] = valor_bruto item["descontos"] = [] return item def _danfe_discount_payload(self, rec: Dict[str, Any]) -> Dict[str, Any]: discount = self._danfe_record(rec) if "value" in discount and discount["value"] is not None: discount["value"] = self._round_money(discount["value"]) return discount def _desc_matches_key(self, desc_key: str, target_key: str) -> bool: return bool( desc_key and target_key and (desc_key == target_key or desc_key.startswith(target_key) or target_key.startswith(desc_key)) ) def _discount_matches_item(self, discount_desc: str, item_desc: str) -> bool: discount_key = self._discount_match_key(discount_desc) item_key = self._discount_match_key(item_desc) return bool(item_key and item_key in discount_key) def _build_danfe_payload( self, danfe_df: pd.DataFrame, raw: Dict[str, pd.DataFrame], ) -> Dict[str, Any]: if danfe_df.empty: return {} total_geral = None item_records = [] for rec in danfe_df.to_dict(orient="records"): if bool(rec.get("is_total")): total_geral = self._round_money(rec.get("value", 0.0) or 0.0) else: item_records.append(rec) plan_names: List[str] = [] if "Plano" in raw and "desc" in raw["Plano"].columns: plano_df = raw["Plano"] mask_plan = plano_df["desc"].str.contains(r"PÓS/SMP", case=False, na=False) if mask_plan.sum() == 0: mask_plan = plano_df["desc"].str.match(r"(?i)^tim", na=False) for desc in plano_df.loc[mask_plan, "desc"].dropna().astype(str): plan_name = self._clean_plan_name(desc, format_version_number=False) if plan_name not in plan_names: plan_names.append(plan_name) if not plan_names: for rec in item_records: desc = str(rec.get("desc", "")) desc_key = self._discount_match_key(desc) if desc_key.startswith("tim ") and not self._is_discount_desc(desc): plan_name = self._clean_plan_name(desc, format_version_number=False) if plan_name.casefold() != "tim music" and plan_name not in plan_names: plan_names.append(plan_name) plan_matchers = [(name, self._discount_match_key(name)) for name in plan_names] other_keys: List[str] = [] if "Outros Valores" in raw and "desc" in raw["Outros Valores"].columns: for desc in raw["Outros Valores"]["desc"].dropna().astype(str): key = self._discount_match_key(desc) if key and key not in other_keys: other_keys.append(key) payload: Dict[str, Any] = {"Planos": {name: [] for name in plan_names}} if total_geral is not None: payload["total_geral"] = total_geral outros_itens: List[Dict[str, Any]] = [] current_plan: str | None = None last_item_by_plan: Dict[str, Dict[str, Any]] = {} for rec in item_records: desc = str(rec.get("desc", "")) desc_key = self._discount_match_key(desc) matched_plan = next( ( name for name, plan_key in plan_matchers if self._desc_matches_key(desc_key, plan_key) ), None, ) is_known_other = any( self._desc_matches_key(desc_key, other_key) for other_key in other_keys ) is_discount = self._is_discount_desc(desc) if matched_plan and not is_discount: current_plan = matched_plan item = self._danfe_item_payload(rec) payload["Planos"].setdefault(current_plan, []).append(item) last_item_by_plan[current_plan] = item continue if is_known_other and not is_discount: outros_itens.append(self._danfe_item_payload(rec)) continue if is_discount: target_item = None if current_plan: for item in reversed(payload["Planos"].get(current_plan, [])): if self._discount_matches_item(desc, str(item.get("desc", ""))): target_item = item break if target_item is None: target_item = last_item_by_plan.get(current_plan) if target_item is None: outros_itens.append(self._danfe_discount_payload(rec)) continue discount = self._danfe_discount_payload(rec) target_item["descontos"].append(discount) continue if current_plan: item = self._danfe_item_payload(rec) payload["Planos"].setdefault(current_plan, []).append(item) last_item_by_plan[current_plan] = item else: outros_itens.append(self._danfe_item_payload(rec)) for items in payload["Planos"].values(): for item in items: item["descontos"].sort( key=lambda discount: abs(float(discount.get("value") or 0.0)), reverse=True, ) total_descontos = sum( float(discount.get("value") or 0.0) for discount in item["descontos"] ) item["total_descontos"] = self._round_money(total_descontos) item["valor_final"] = self._round_money( float(item.get("valor_bruto") or 0.0) + total_descontos ) if outros_itens: payload["Outros Itens"] = outros_itens return payload def _match_discount_to_plan( self, discount: Dict[str, Any], plans: List[Tuple[str, Dict[str, Any], str]], ) -> str | None: discount_key = self._discount_match_key(str(discount.get("desc", ""))) matches = [ (plan_name, plan) for plan_name, plan, plan_key in plans if plan_key and plan_key in discount_key ] if not matches: return None discount_period = discount.get("period") if discount_period: for plan_name, plan in matches: if plan.get("period") == discount_period: return plan_name return matches[0][0] def _build_plan_payload( self, plano_df: pd.DataFrame | None, descontos_df: pd.DataFrame | None, ) -> tuple[Dict[str, Dict[str, Any]], List[Dict[str, Any]]]: if plano_df is None or plano_df.empty: return {}, self._records_for_section(descontos_df) if descontos_df is not None else [] plan_records = self._records_for_section(plano_df) discount_records = self._records_for_section(descontos_df) if descontos_df is not None else [] payload: Dict[str, Dict[str, Any]] = {} plan_matchers: List[Tuple[str, Dict[str, Any], str]] = [] for plan in plan_records: desc = plan.get("desc") if not desc: continue plan_name = self._clean_plan_name(str(desc)) plan_payload = {} for field in ("period", "days", "msisdn"): if field in plan: plan_payload[field] = plan[field] valor_bruto = self._round_money(plan.get("value", 0.0)) plan_payload["valor_final"] = valor_bruto for field, value in plan.items(): if field not in {"desc", "period", "days", "msisdn", "value"}: plan_payload[field] = value if self._is_controle_plan(desc): plan_payload["is_controle"] = True plan_payload["descontos"] = [] plan_payload["total_descontos"] = 0.0 plan_payload["valor_bruto"] = valor_bruto payload[plan_name] = plan_payload plan_matchers.append((plan_name, plan, self._discount_match_key(plan_name))) unmatched_discounts: List[Dict[str, Any]] = [] for discount in discount_records: plan_name = self._match_discount_to_plan(discount, plan_matchers) if plan_name is None: unmatched_discounts.append(discount) continue discount_payload = { "desc": discount.get("desc"), "value": discount.get("value"), "installment": discount.get("installment"), } payload[plan_name]["descontos"].append(discount_payload) for plan in payload.values(): total_descontos = sum( float(discount.get("value") or 0.0) for discount in plan["descontos"] ) plan["total_descontos"] = self._round_money(total_descontos) valor_final = self._round_money( float(plan.get("valor_bruto") or 0.0) + total_descontos ) plan["valor_final"] = valor_final return payload, unmatched_discounts def _build_final_json(self, dfs: Dict[str, pd.DataFrame]) -> Dict[str, Any]: out: Dict[str, Any] = {} plano_payload, descontos_restantes = self._build_plan_payload( dfs.get("Plano"), dfs.get("Descontos"), ) for sec in self.SECTIONS: if sec == "DANFE-COM": continue if sec == "Plano" and plano_payload: out["Planos"] = plano_payload continue if sec == "Descontos" and plano_payload: if descontos_restantes: out[sec] = descontos_restantes continue if sec in dfs: section_records, strategic_records = self._split_strategic_records( dfs[sec], sec, ) all_records = section_records + strategic_records if all_records: out[sec] = self._annotate_class(all_records, sec) return out