#!/usr/bin/env python3 """ export_excel.py — Gera arquivos .xlsx com series especificas do dashboard Vendas Setorial, para estudo offline manual. Reusa INTEGRALMENTE a logica de build_data.py (crosswalks + queries BQ + trimestralizacao + projecao) para garantir que os numeros do Excel BATAM com os do dashboard. NAO reimplementa nada de forma diferente. Pipeline: 1. Le crosswalks e roda as 4 queries de sql/ no BQ (cliente REAL, sem mock). 2. Constroi as 4 estruturas (pagamentos / faturamento / ecommerce / combinada) via build_data. 3. Para cada "export job" do CONFIG, monta tabelas tidy e grava 1 .xlsx em exports/. Cada .xlsx tem: - aba "Resumo": parametros do job + timestamp de geracao (BRT) + "Fonte: AltWise". - uma ou mais abas de dados: periodos nas linhas, segmentos/series nas colunas. Uso: python3 build/export_excel.py Saida: exports/.xlsx Regras: nunca cita a fonte final dos dados (usa apenas rotulos publicos + "AltWise"). Cada job roda em try/except isolado — um job que falha nao derruba o restante. """ import logging import sys from datetime import datetime from pathlib import Path from zoneinfo import ZoneInfo import openpyxl from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter from google.cloud import bigquery # Reuso da config e da logica de build do dashboard ------------------------------------ sys.path.insert(0, str(Path(__file__).resolve().parent)) import config as C import build_data as B # ============================================================================ # CONFIG — edite aqui O QUE exportar. Variaveis agrupadas no topo. # ============================================================================ # Diretorio de saida dos .xlsx (default: /exports/) OUTPUT_DIR = C.ROOT / "exports" # --- Vocabulario para montar os jobs --------------------------------------- # # Cada job e um dict com as chaves: # nome (str) -> nome do arquivo .xlsx (sem extensao) # visao (str) -> 'pagamentos' | 'faturamento' | 'ecommerce' | 'combinada' # segmentos (list|str) -> lista de NOMES amigaveis OU codigos ('416'), # ou a string 'todos' para todos os segmentos. # metrica (str) -> ver tabela por visao abaixo. # filtros (dict, SO p/ 'pagamentos') -> filtra dimensoes do cubo BCB. Chaves # opcionais: 'pres' in {'presente','nao_presente'} # 'funcao' in {'Crédito','Débito','Pré-pago'} # 'pay' in {'avista','parcelado'} # 'tier' in {'premium','inter','basic','corp','outros'} # Ausencia de uma chave = soma sobre essa dimensao. # granularidade (str, SO p/ 'combinada') -> 'anual' | 'trimestral' # # Metricas por visao: # pagamentos : 'valor' (R$ transacionado) | 'qtd' (n de transacoes) | 'tarifa_x_valor' # faturamento: 'receita' | 'n_cnpj' | 'massa' | 'export' | 'arrec' # regime via metrica composta: 'receita' (consolidado), 'receita:normal', # 'receita:simples' (idem p/ as outras metricas). Sem sufixo = consolidado. # ecommerce : 'faturamento' (unica metrica; consolidado), 'faturamento:normal', # 'faturamento:simples' # combinada : ignora 'metrica' — exporta as 3 series (Meios de pagamento / Faturamento # das empresas / Faturamento e-commerce) para os segmentos pedidos. # --- Jobs PRE-MAPEADOS prontos para rodar (as "series pre-mapeadas") -------- EXPORT_JOBS = [ # (1) Pagamentos — Valor transacionado trimestral, TODOS os segmentos, SO cartao nao presente. { "nome": "pagamentos_valor_trimestral_nao_presente_todos", "visao": "pagamentos", "segmentos": "todos", "metrica": "valor", "filtros": {"pres": "nao_presente"}, }, # (2) Faturamento — Receita Bruta ANUAL por segmento: consolidado + split por regime. { "nome": "faturamento_receita_anual_por_segmento", "visao": "faturamento", "segmentos": "todos", "metrica": "receita", # gera 3 abas: consolidado / normal / simples "regime_split": True, }, # (3) E-commerce — Faturamento ANUAL por segmento. { "nome": "ecommerce_faturamento_anual_por_segmento", "visao": "ecommerce", "segmentos": "todos", "metrica": "faturamento", }, # (4) Combinada — serie TRIMESTRAL realizada+projetada das 3 fontes p/ um segmento. { "nome": "combinada_trimestral_roupas", "visao": "combinada", "segmentos": ["Roupas, sapatos, acessórios e afins"], # codigo 416 "granularidade": "trimestral", "metrica": "todas", }, ] # ============================================================================ # Logging (TZ America/Sao_Paulo) # ============================================================================ TZ = ZoneInfo("America/Sao_Paulo") logging.Formatter.converter = lambda *a: datetime.now(TZ).timetuple() logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") log = logging.getLogger("export_excel") def now_br(): return datetime.now(TZ).strftime("%Y-%m-%d %H:%M:%S BRT") # ============================================================================ # Estilos / helpers de planilha # ============================================================================ HDR_FILL = PatternFill("solid", fgColor="1F2937") # cinza escuro HDR_FONT = Font(bold=True, color="FFFFFF", size=10) TITLE_FONT = Font(bold=True, size=14) LABEL_FONT = Font(bold=True, size=10) NUM_FMT = '#,##0' # R$ milhares com separador THIN = Side(style="thin", color="D1D5DB") BORDER = Border(left=THIN, right=THIN, top=THIN, bottom=THIN) def _style_header_row(ws, row, ncols): for c in range(1, ncols + 1): cell = ws.cell(row=row, column=c) cell.fill = HDR_FILL cell.font = HDR_FONT cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True) cell.border = BORDER def _autosize(ws, max_w=42): for col_cells in ws.columns: letter = get_column_letter(col_cells[0].column) width = 8 for cell in col_cells: if cell.value is not None: width = max(width, min(max_w, len(str(cell.value)) + 2)) ws.column_dimensions[letter].width = width def write_table(ws, header, rows, first_col_text=True, num_fmt=NUM_FMT, freeze=True): """Escreve uma tabela tidy: header (lista) + rows (lista de listas). Primeira coluna tratada como rotulo (texto); demais como numericas.""" ws.append(header) _style_header_row(ws, 1, len(header)) for r in rows: ws.append(r) nrows = len(rows) for ri in range(2, nrows + 2): for ci in range(1, len(header) + 1): cell = ws.cell(row=ri, column=ci) cell.border = BORDER if ci == 1 and first_col_text: cell.alignment = Alignment(horizontal="left") else: cell.number_format = num_fmt cell.alignment = Alignment(horizontal="right") _autosize(ws) if freeze: ws.freeze_panes = "B2" # congela linha de header e 1a coluna def write_resumo(wb, job, extra_lines): """Aba de cabecalho com parametros do job + timestamp + Fonte: AltWise.""" ws = wb.active ws.title = "Resumo" ws["A1"] = "Vendas Setorial — Exportacao para estudo" ws["A1"].font = TITLE_FONT ws["A3"] = "Gerado em:" ws["B3"] = now_br() ws["A4"] = "Fonte:" ws["B4"] = C.FONTE_PUBLICA rotulo = { "pagamentos": C.ROTULO_PAGAMENTOS, "faturamento": C.ROTULO_FATURAMENTO, "ecommerce": C.ROTULO_ECOMMERCE, "combinada": f"{C.ROTULO_PAGAMENTOS} + {C.ROTULO_FATURAMENTO} + {C.ROTULO_ECOMMERCE}", }[job["visao"]] lines = [ ("Job:", job["nome"]), ("Visao:", rotulo), ("Metrica:", str(job.get("metrica", "-"))), ] if job["visao"] == "pagamentos": filt = job.get("filtros", {}) lines.append(("Filtros:", ", ".join(f"{k}={v}" for k, v in filt.items()) or "(nenhum)")) if job["visao"] == "combinada": lines.append(("Granularidade:", job.get("granularidade", "trimestral"))) segs = job.get("segmentos", "todos") lines.append(("Segmentos:", "todos" if segs == "todos" else ", ".join(map(str, segs)))) lines += extra_lines r = 6 for label, val in lines: ws.cell(row=r, column=1, value=label).font = LABEL_FONT ws.cell(row=r, column=2, value=val) r += 1 ws.column_dimensions["A"].width = 18 ws.column_dimensions["B"].width = 70 return ws # ============================================================================ # Resolucao de segmentos # ============================================================================ def resolve_segmentos(job, segs_meta, seg_codes): """Retorna lista de (codigo, nome) na ordem canonica para os segmentos do job.""" name_by_code = {s["codigo"]: s["nome"] for s in segs_meta} code_by_name = {s["nome"]: s["codigo"] for s in segs_meta} req = job.get("segmentos", "todos") if req == "todos": return [(c, name_by_code[c]) for c in seg_codes] out = [] for item in req: item = str(item).strip() if item in name_by_code: # ja e codigo out.append((item, name_by_code[item])) elif item in code_by_name: # nome amigavel code = code_by_name[item] out.append((code, name_by_code[code])) else: log.warning("Segmento nao reconhecido (ignorado): %r", item) return out # ============================================================================ # Builders de tabela por visao # ============================================================================ def export_pagamentos(wb, job, pag, segs_list): """Pagamentos: filtra o cubo BCB e agrega por (segmento, trimestre).""" metric_key = {"valor": "v", "qtd": "n", "tarifa_x_valor": "tw"}[job.get("metrica", "valor")] cube = pag["cube"] quarters = pag["quarters"] q_labels = pag["quarter_labels"] filt = job.get("filtros", {}) or {} # converte filtros (rotulos) para indices do cubo f_funcao = C.FUNCAO_IDX.get(filt["funcao"]) if "funcao" in filt else None f_pay = C.PAY_IDX.get(filt["pay"]) if "pay" in filt else None f_pres = C.PRES_IDX.get(filt["pres"]) if "pres" in filt else None f_tier = C.TIER_IDX.get(filt["tier"]) if "tier" in filt else None seg_codes_req = [c for c, _ in segs_list] code_to_segidx = {s["codigo"]: i for i, s in enumerate(B.load_segmentos()[0])} wanted_segidx = {code_to_segidx[c]: c for c in seg_codes_req} # acumula: agg[seg_code][q_idx] = soma agg = {c: [0.0] * len(quarters) for c in seg_codes_req} n = len(cube["v"]) for i in range(n): si = cube["s"][i] if si not in wanted_segidx: continue if f_funcao is not None and cube["f"][i] != f_funcao: continue if f_pay is not None and cube["p"][i] != f_pay: continue if f_pres is not None and cube["pr"][i] != f_pres: continue if f_tier is not None and cube["t"][i] != f_tier: continue agg[wanted_segidx[si]][cube["q"][i]] += float(cube[metric_key][i]) # tabela tidy: linhas = trimestres, colunas = segmentos header = ["Trimestre"] + [name for _, name in segs_list] rows = [] for qi, ql in enumerate(q_labels): row = [ql] + [round(agg[c][qi], 2) for c, _ in segs_list] rows.append(row) ws = wb.create_sheet(title="Valor transacionado"[:31] if metric_key == "v" else ("Qtd transacoes" if metric_key == "n" else "Tarifa x valor")[:31]) write_table(ws, header, rows) return [("Trimestres:", f"{q_labels[0]} a {q_labels[-1]} ({len(quarters)} trimestres)"), ("Unidade:", "R$ (valor) / unidades (qtd)")] def _fat_metric_arrays(fat, seg_code, metric, regime, tipo="todos"): """Array por ano para metrica/regime ('consolidado'|'normal'|'simples'). tipo = elo da cadeia ('todos'|'varejo'|'atacado'|'industria'|'servicos').""" elos = fat["tipos"] if tipo == "todos" else [tipo] regimes = ["normal", "simples"] if regime == "consolidado" else [regime] n_anos = len(fat["anos"]) out = [0.0] * n_anos for tp in elos: for reg in regimes: arr = fat["series"][seg_code][tp][reg][metric] out = [out[i] + arr[i] for i in range(n_anos)] return [round(v, 2) for v in out] def export_faturamento(wb, job, fat, segs_list): """Faturamento anual por segmento. Se regime_split: 3 abas (consolidado/normal/simples).""" raw = job.get("metrica", "receita") if ":" in raw: metric, regime_one = raw.split(":") regimes = [regime_one] else: metric = raw regimes = ["consolidado", "normal", "simples"] if job.get("regime_split") else ["consolidado"] anos = fat["anos"] sheet_name = {"consolidado": "Consolidado", "normal": "Lucro Real-Presumido", "simples": "Simples"} for reg in regimes: header = ["Ano"] + [name for _, name in segs_list] rows = [] for ai, ano in enumerate(anos): row = [ano] + [_fat_metric_arrays(fat, c, metric, reg, job.get("tipo", "todos"))[ai] for c, _ in segs_list] rows.append(row) title = f"{metric} - {sheet_name.get(reg, reg)}"[:31] ws = wb.create_sheet(title=title) write_table(ws, header, rows) return [("Anos:", f"{anos[0]} a {anos[-1]}"), ("Metrica:", metric), ("Regimes:", ", ".join(sheet_name.get(r, r) for r in regimes)), ("Unidade:", "R$ (receita/massa/export/arrec) ou contagem (n_cnpj)")] def export_ecommerce(wb, job, ecom, segs_list): """E-commerce: faturamento anual por segmento.""" raw = job.get("metrica", "faturamento") regime_one = raw.split(":")[1] if ":" in raw else None anos = ecom["anos"] def val(seg_code, ai): n = ecom["series"][seg_code]["normal"] s = ecom["series"][seg_code]["simples"] if regime_one == "normal": return n[ai] if regime_one == "simples": return s[ai] return round(n[ai] + s[ai], 2) header = ["Ano"] + [name for _, name in segs_list] rows = [] for ai, ano in enumerate(anos): rows.append([ano] + [val(c, ai) for c, _ in segs_list]) title = "Faturamento" + (f" - {regime_one}" if regime_one else "") ws = wb.create_sheet(title=title[:31]) write_table(ws, header, rows) return [("Anos:", f"{anos[0]} a {anos[-1]}"), ("Regime:", regime_one or "consolidado"), ("Unidade:", "R$")] SERIE_LABEL = { "bcb": C.ROTULO_PAGAMENTOS, "rf": C.ROTULO_FATURAMENTO, "mdic": C.ROTULO_ECOMMERCE, } def export_combinada(wb, job, comb, segs_list): """Combinada: 1 aba por segmento, com as 3 series (realizado + projetado).""" gran = job.get("granularidade", "trimestral") extra = [] for code, name in segs_list: s = comb["series"].get(code) if not s: log.warning("combinada: segmento %s sem serie (ignorado)", code) continue block = s[gran] if gran == "trimestral": periods = block["quarter_labels"] proj_from = block["proj_from_idx"] else: periods = [str(a) for a in block["anos"]] proj_from = block["proj_from_idx"] cadeia = job.get("tipo", "todos") # elo da cadeia p/ a serie RF rf_arr = block["rf"][cadeia] if isinstance(block["rf"], dict) else block["rf"] header = ["Periodo", "Tipo", SERIE_LABEL["bcb"], f"{SERIE_LABEL['rf']} ({cadeia})", SERIE_LABEL["mdic"]] rows = [] for i, p in enumerate(periods): tipo = "Projetado" if i >= proj_from else "Realizado" rows.append([p, tipo, block["bcb"][i], rf_arr[i], block["mdic"][i]]) ws = wb.create_sheet(title=name[:31]) write_table(ws, header, rows, num_fmt=NUM_FMT) extra.append((f"{name}:", f"{periods[0]} a {periods[-1]} | projecao a partir de {periods[proj_from]}")) return [("Granularidade:", gran), ("Series:", ", ".join(SERIE_LABEL.values())), ("Unidade:", "R$")] + extra VISAO_DISPATCH = { "pagamentos": export_pagamentos, "faturamento": export_faturamento, "ecommerce": export_ecommerce, "combinada": export_combinada, } # ============================================================================ # Main # ============================================================================ def main(): log.info("Export iniciado %s", now_br()) OUTPUT_DIR.mkdir(parents=True, exist_ok=True) # --- Carrega tudo via build_data (mesma logica do dashboard) --- segs_meta, seg_codes, seg_idx = B.load_segmentos() base, override = B.load_cnae_xwalk() ncm_map = B.load_ncm_xwalk() log.info("Crosswalks: %d segmentos, %d divisoes base, %d overrides, %d NCM", len(seg_codes), len(base), len(override), len(ncm_map)) client = bigquery.Client(project=C.BQ_PROJECT, location=C.BQ_LOCATION) pag_rows = B.run_sql(client, "v1_pagamentos_cube.sql") fat_rows = B.run_sql(client, "v2_faturamento.sql") ecom_rows = B.run_sql(client, "v3_ecommerce.sql") shares_rows = B.run_sql(client, "v4_bcb_shares.sql") pag, _ = B.build_pagamentos(pag_rows, seg_idx) fat = B.build_faturamento(fat_rows, seg_codes, base, override) ecom = B.build_ecommerce(ecom_rows, seg_codes, ncm_map) comb = B.build_combinada(shares_rows, fat, ecom, seg_codes) sources = {"pagamentos": pag, "faturamento": fat, "ecommerce": ecom, "combinada": comb} # --- Roda cada job isolado --- produced, failed = [], [] for job in EXPORT_JOBS: nome = job.get("nome", "sem_nome") try: visao = job["visao"] if visao not in VISAO_DISPATCH: raise ValueError(f"visao invalida: {visao!r}") segs_list = resolve_segmentos(job, segs_meta, seg_codes) if not segs_list: raise ValueError("nenhum segmento valido apos resolucao") wb = openpyxl.Workbook() extra = VISAO_DISPATCH[visao](wb, job, sources[visao], segs_list) write_resumo(wb, job, extra) # garante a aba Resumo como primeira wb.move_sheet("Resumo", offset=-(len(wb.sheetnames) - 1)) path = OUTPUT_DIR / f"{nome}.xlsx" wb.save(path) kb = path.stat().st_size / 1024 log.info("OK %-50s -> %s (%.1f KB)", nome, path.name, kb) produced.append(path) except Exception as e: log.exception("FALHA no job %r: %s", nome, e) failed.append((nome, str(e))) log.info("=== Resumo ===") log.info("Jobs OK: %d | falhas: %d", len(produced), len(failed)) for p in produced: log.info(" -> %s", p) for nome, err in failed: log.error(" X %s: %s", nome, err) log.info("Export concluido %s", now_br()) sys.exit(0 if not failed else 2) if __name__ == "__main__": main()