From 2ea14ca5f610988049a4a6f1e9ad8bb864a7e13c Mon Sep 17 00:00:00 2001 From: Ricardo Dahis Date: Tue, 1 Sep 2026 09:44:26 +1000 Subject: [PATCH 1/5] feat(br_senatran_estatisticas): add the six remaining frota breakdowns MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completa os sete recortes mensais publicados nas páginas anuais do gov.br: cor, potência, restrição, CEP, ano de fabricação/modelo e tipo/espécie/eixos, somando-se a municipio_combustivel. Cada um é uma entrada em LAYOUTS mais um modelo dbt — a estrutura genérica de breakdowns.py já cobria o resto, porque todos têm o mesmo formato. Três correções que só apareceram ao estender: - **Tokens múltiplos por recorte.** O gov.br renomeia os recortes entre anos: o mesmo dado aparece como `ano_de_fabricacao_e_modelo` (2017), `ano_fab_mod` e `ano_fab_modelo` (2021), e como `tipoespecieeixo` (sem separadores, 2017) ou `tipo_especie_eixos` (2021). Um token só perdia 11 dos 12 meses de 2017. - **Casamento por token delimitado, não por substring.** `cor` e `cep` são curtos e casariam dentro de `recorte`, `concept` etc. - **Leitura com dtype=str.** O pandas inferia o CEP como int e destruía os zeros à esquerda (069900 -> 69900). O staging é all-STRING por convenção de qualquer forma; o safe_cast do modelo decide o tipo final. Verificado contra as páginas reais: os sete recortes devolvem 12/12/7 meses em 2017/2021/2026, com duas exceções que são lacunas da fonte, não do casador — dezembro/2021 não é publicado para tipo/espécie/eixos (só 11 arquivos na página) e falta um mês de 2017 para ano fab/modelo. Colunas de dimensão ficam STRING quando são chaves de agrupamento com sentinela 0 = não informado (potência, eixos, CEP); ano_modelo e ano_fabricacao são INT64, como o `ano` particionador. Inclui também suporte a .zip/.rar em read_breakdown: 2013, 2015 e 2016 vêm compactados. Falha de extração levanta UnsupportedArchiveError para o backfill pular o mês em vez de abortar os outros 150. dbt parse limpo, 66 nós resolvem, as 9 flows são encontradas pelo deploy_flows, ruff e pyrefly sem diagnósticos. Crons em horários livres (22h05 a 22h55), sem colisão com os já existentes no repo. --- ...ticas__municipio_ano_fabricacao_modelo.sql | 34 +++ ...r_senatran_estatisticas__municipio_cep.sql | 28 ++ ...r_senatran_estatisticas__municipio_cor.sql | 28 ++ ...atran_estatisticas__municipio_potencia.sql | 30 ++ ...tran_estatisticas__municipio_restricao.sql | 30 ++ ...tisticas__municipio_tipo_especie_eixos.sql | 35 +++ models/br_senatran_estatisticas/schema.yml | 264 ++++++++++++++++++ .../br_senatran_estatisticas/breakdowns.py | 111 +++++++- .../br_senatran_estatisticas/flows.py | 186 ++++++++++++ 9 files changed, 739 insertions(+), 7 deletions(-) create mode 100644 models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_ano_fabricacao_modelo.sql create mode 100644 models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_cep.sql create mode 100644 models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_cor.sql create mode 100644 models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_potencia.sql create mode 100644 models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_restricao.sql create mode 100644 models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_tipo_especie_eixos.sql diff --git a/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_ano_fabricacao_modelo.sql b/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_ano_fabricacao_modelo.sql new file mode 100644 index 0000000000..be2099898c --- /dev/null +++ b/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_ano_fabricacao_modelo.sql @@ -0,0 +1,34 @@ +{{ + config( + alias="municipio_ano_fabricacao_modelo", + schema="br_senatran_estatisticas", + materialized="table", + partition_by={ + "field": "ano", + "data_type": "int64", + "range": { + "start": 2015, + "end": 2031, + "interval": 1, + }, + }, + cluster_by=["mes"], + pre_hook="{% if adapter.get_relation(this.database, this.schema, this.identifier) %}DROP ALL ROW ACCESS POLICIES ON {{ this }}{% else %}SELECT 1{% endif %}", + ) +}} + + +select + safe_cast(ano as int64) ano, + safe_cast(mes as int64) mes, + safe_cast(sigla_uf as string) sigla_uf, + safe_cast(id_municipio as string) id_municipio, + safe_cast(ano_modelo as int64) ano_modelo, + safe_cast(ano_fabricacao as int64) ano_fabricacao, + safe_cast(quantidade as int64) quantidade +from + {{ + set_datalake_project( + "br_senatran_estatisticas_staging.municipio_ano_fabricacao_modelo" + ) + }} as t diff --git a/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_cep.sql b/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_cep.sql new file mode 100644 index 0000000000..7d358962ca --- /dev/null +++ b/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_cep.sql @@ -0,0 +1,28 @@ +{{ + config( + alias="municipio_cep", + schema="br_senatran_estatisticas", + materialized="table", + partition_by={ + "field": "ano", + "data_type": "int64", + "range": { + "start": 2014, + "end": 2031, + "interval": 1, + }, + }, + cluster_by=["mes"], + pre_hook="{% if adapter.get_relation(this.database, this.schema, this.identifier) %}DROP ALL ROW ACCESS POLICIES ON {{ this }}{% else %}SELECT 1{% endif %}", + ) +}} + + +select + safe_cast(ano as int64) ano, + safe_cast(mes as int64) mes, + safe_cast(sigla_uf as string) sigla_uf, + safe_cast(id_municipio as string) id_municipio, + safe_cast(lower(cep) as string) cep, + safe_cast(quantidade as int64) quantidade +from {{ set_datalake_project("br_senatran_estatisticas_staging.municipio_cep") }} as t diff --git a/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_cor.sql b/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_cor.sql new file mode 100644 index 0000000000..8441da0549 --- /dev/null +++ b/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_cor.sql @@ -0,0 +1,28 @@ +{{ + config( + alias="municipio_cor", + schema="br_senatran_estatisticas", + materialized="table", + partition_by={ + "field": "ano", + "data_type": "int64", + "range": { + "start": 2014, + "end": 2031, + "interval": 1, + }, + }, + cluster_by=["mes"], + pre_hook="{% if adapter.get_relation(this.database, this.schema, this.identifier) %}DROP ALL ROW ACCESS POLICIES ON {{ this }}{% else %}SELECT 1{% endif %}", + ) +}} + + +select + safe_cast(ano as int64) ano, + safe_cast(mes as int64) mes, + safe_cast(sigla_uf as string) sigla_uf, + safe_cast(id_municipio as string) id_municipio, + safe_cast(lower(cor) as string) cor, + safe_cast(quantidade as int64) quantidade +from {{ set_datalake_project("br_senatran_estatisticas_staging.municipio_cor") }} as t diff --git a/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_potencia.sql b/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_potencia.sql new file mode 100644 index 0000000000..da89153c6f --- /dev/null +++ b/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_potencia.sql @@ -0,0 +1,30 @@ +{{ + config( + alias="municipio_potencia", + schema="br_senatran_estatisticas", + materialized="table", + partition_by={ + "field": "ano", + "data_type": "int64", + "range": { + "start": 2013, + "end": 2031, + "interval": 1, + }, + }, + cluster_by=["mes"], + pre_hook="{% if adapter.get_relation(this.database, this.schema, this.identifier) %}DROP ALL ROW ACCESS POLICIES ON {{ this }}{% else %}SELECT 1{% endif %}", + ) +}} + + +select + safe_cast(ano as int64) ano, + safe_cast(mes as int64) mes, + safe_cast(sigla_uf as string) sigla_uf, + safe_cast(id_municipio as string) id_municipio, + safe_cast(lower(potencia) as string) potencia, + safe_cast(quantidade as int64) quantidade +from + {{ set_datalake_project("br_senatran_estatisticas_staging.municipio_potencia") }} + as t diff --git a/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_restricao.sql b/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_restricao.sql new file mode 100644 index 0000000000..56f9536a93 --- /dev/null +++ b/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_restricao.sql @@ -0,0 +1,30 @@ +{{ + config( + alias="municipio_restricao", + schema="br_senatran_estatisticas", + materialized="table", + partition_by={ + "field": "ano", + "data_type": "int64", + "range": { + "start": 2013, + "end": 2031, + "interval": 1, + }, + }, + cluster_by=["mes"], + pre_hook="{% if adapter.get_relation(this.database, this.schema, this.identifier) %}DROP ALL ROW ACCESS POLICIES ON {{ this }}{% else %}SELECT 1{% endif %}", + ) +}} + + +select + safe_cast(ano as int64) ano, + safe_cast(mes as int64) mes, + safe_cast(sigla_uf as string) sigla_uf, + safe_cast(id_municipio as string) id_municipio, + safe_cast(lower(restricao) as string) restricao, + safe_cast(quantidade as int64) quantidade +from + {{ set_datalake_project("br_senatran_estatisticas_staging.municipio_restricao") }} + as t diff --git a/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_tipo_especie_eixos.sql b/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_tipo_especie_eixos.sql new file mode 100644 index 0000000000..490faf2a1a --- /dev/null +++ b/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_tipo_especie_eixos.sql @@ -0,0 +1,35 @@ +{{ + config( + alias="municipio_tipo_especie_eixos", + schema="br_senatran_estatisticas", + materialized="table", + partition_by={ + "field": "ano", + "data_type": "int64", + "range": { + "start": 2013, + "end": 2031, + "interval": 1, + }, + }, + cluster_by=["mes"], + pre_hook="{% if adapter.get_relation(this.database, this.schema, this.identifier) %}DROP ALL ROW ACCESS POLICIES ON {{ this }}{% else %}SELECT 1{% endif %}", + ) +}} + + +select + safe_cast(ano as int64) ano, + safe_cast(mes as int64) mes, + safe_cast(sigla_uf as string) sigla_uf, + safe_cast(id_municipio as string) id_municipio, + safe_cast(lower(tipo_veiculo) as string) tipo_veiculo, + safe_cast(lower(especie) as string) especie, + safe_cast(lower(eixos) as string) eixos, + safe_cast(quantidade as int64) quantidade +from + {{ + set_datalake_project( + "br_senatran_estatisticas_staging.municipio_tipo_especie_eixos" + ) + }} as t diff --git a/models/br_senatran_estatisticas/schema.yml b/models/br_senatran_estatisticas/schema.yml index 61dcb83613..9e776e042f 100644 --- a/models/br_senatran_estatisticas/schema.yml +++ b/models/br_senatran_estatisticas/schema.yml @@ -112,3 +112,267 @@ models: description: Tipo de combustível do veículo - name: quantidade description: Quantidade de veículos + - name: br_senatran_estatisticas__municipio_cor + description: > + Frota de veículos por município e cor, com dados mensais a partir de 2014. Fonte: + recorte C das estatísticas de frota da SENATRAN. + tests: + - dbt_utils.unique_combination_of_columns: + combination_of_columns: [ano, mes, id_municipio, cor] + config: + where: __most_recent_year_month__ + - not_null_proportion_multiple_columns: + at_least: 0.05 + config: + where: __most_recent_year_month__ + columns: + - name: ano + description: Ano + tests: [not_null] + - name: mes + description: Mês + tests: [not_null] + - name: sigla_uf + description: Sigla da Unidade da Federação + tests: + - relationships: + to: ref('br_bd_diretorios_brasil__uf') + field: sigla + config: + where: __most_recent_year_month__ + - name: id_municipio + description: ID Município - IBGE 7 Dígitos + tests: + - not_null + - relationships: + to: ref('br_bd_diretorios_brasil__municipio') + field: id_municipio + config: + where: __most_recent_year_month__ + - name: cor + description: Cor do veículo + - name: quantidade + description: Quantidade de veículos + - name: br_senatran_estatisticas__municipio_potencia + description: > + Frota de veículos por município e potência do motor, com dados mensais a partir + de 2013. Fonte: recorte E das estatísticas de frota da SENATRAN. + tests: + - dbt_utils.unique_combination_of_columns: + combination_of_columns: [ano, mes, id_municipio, potencia] + config: + where: __most_recent_year_month__ + - not_null_proportion_multiple_columns: + at_least: 0.05 + config: + where: __most_recent_year_month__ + columns: + - name: ano + description: Ano + tests: [not_null] + - name: mes + description: Mês + tests: [not_null] + - name: sigla_uf + description: Sigla da Unidade da Federação + tests: + - relationships: + to: ref('br_bd_diretorios_brasil__uf') + field: sigla + config: + where: __most_recent_year_month__ + - name: id_municipio + description: ID Município - IBGE 7 Dígitos + tests: + - not_null + - relationships: + to: ref('br_bd_diretorios_brasil__municipio') + field: id_municipio + config: + where: __most_recent_year_month__ + - name: potencia + description: Potência do veículo conforme informada pela fonte; 0 indica não + informado + - name: quantidade + description: Quantidade de veículos + - name: br_senatran_estatisticas__municipio_restricao + description: > + Frota de veículos por município e tipo de restrição, com dados mensais a partir + de 2013. Fonte: recorte H das estatísticas de frota da SENATRAN. + tests: + - dbt_utils.unique_combination_of_columns: + combination_of_columns: [ano, mes, id_municipio, restricao] + config: + where: __most_recent_year_month__ + - not_null_proportion_multiple_columns: + at_least: 0.05 + config: + where: __most_recent_year_month__ + columns: + - name: ano + description: Ano + tests: [not_null] + - name: mes + description: Mês + tests: [not_null] + - name: sigla_uf + description: Sigla da Unidade da Federação + tests: + - relationships: + to: ref('br_bd_diretorios_brasil__uf') + field: sigla + config: + where: __most_recent_year_month__ + - name: id_municipio + description: ID Município - IBGE 7 Dígitos + tests: + - not_null + - relationships: + to: ref('br_bd_diretorios_brasil__municipio') + field: id_municipio + config: + where: __most_recent_year_month__ + - name: restricao + description: Tipo de restrição sobre o veículo + - name: quantidade + description: Quantidade de veículos + - name: br_senatran_estatisticas__municipio_cep + description: > + Frota de veículos por município e CEP, com dados mensais a partir de 2014. Fonte: + recorte A das estatísticas de frota da SENATRAN. + tests: + - dbt_utils.unique_combination_of_columns: + combination_of_columns: [ano, mes, id_municipio, cep] + config: + where: __most_recent_year_month__ + - not_null_proportion_multiple_columns: + at_least: 0.05 + config: + where: __most_recent_year_month__ + columns: + - name: ano + description: Ano + tests: [not_null] + - name: mes + description: Mês + tests: [not_null] + - name: sigla_uf + description: Sigla da Unidade da Federação + tests: + - relationships: + to: ref('br_bd_diretorios_brasil__uf') + field: sigla + config: + where: __most_recent_year_month__ + - name: id_municipio + description: ID Município - IBGE 7 Dígitos + tests: + - not_null + - relationships: + to: ref('br_bd_diretorios_brasil__municipio') + field: id_municipio + config: + where: __most_recent_year_month__ + - name: cep + description: CEP informado no registro do veículo; 0 indica não informado + - name: quantidade + description: Quantidade de veículos + - name: br_senatran_estatisticas__municipio_ano_fabricacao_modelo + description: > + Frota de veículos por município, ano de fabricação e ano do modelo, com dados + mensais a partir de 2015. Fonte: recorte F das estatísticas de frota da SENATRAN. + tests: + - dbt_utils.unique_combination_of_columns: + combination_of_columns: + - ano + - mes + - id_municipio + - ano_modelo + - ano_fabricacao + config: + where: __most_recent_year_month__ + - not_null_proportion_multiple_columns: + at_least: 0.05 + config: + where: __most_recent_year_month__ + columns: + - name: ano + description: Ano + tests: [not_null] + - name: mes + description: Mês + tests: [not_null] + - name: sigla_uf + description: Sigla da Unidade da Federação + tests: + - relationships: + to: ref('br_bd_diretorios_brasil__uf') + field: sigla + config: + where: __most_recent_year_month__ + - name: id_municipio + description: ID Município - IBGE 7 Dígitos + tests: + - not_null + - relationships: + to: ref('br_bd_diretorios_brasil__municipio') + field: id_municipio + config: + where: __most_recent_year_month__ + - name: ano_modelo + description: Ano do modelo do veículo + - name: ano_fabricacao + description: Ano de fabricação do veículo conforme o CRV + - name: quantidade + description: Quantidade de veículos + - name: br_senatran_estatisticas__municipio_tipo_especie_eixos + description: > + Frota de veículos por município, tipo, espécie e número de eixos, com dados + mensais a partir de 2013. Fonte: recorte G das estatísticas de frota da SENATRAN. + tests: + - dbt_utils.unique_combination_of_columns: + combination_of_columns: + - ano + - mes + - id_municipio + - tipo_veiculo + - especie + - eixos + config: + where: __most_recent_year_month__ + - not_null_proportion_multiple_columns: + at_least: 0.05 + config: + where: __most_recent_year_month__ + columns: + - name: ano + description: Ano + tests: [not_null] + - name: mes + description: Mês + tests: [not_null] + - name: sigla_uf + description: Sigla da Unidade da Federação + tests: + - relationships: + to: ref('br_bd_diretorios_brasil__uf') + field: sigla + config: + where: __most_recent_year_month__ + - name: id_municipio + description: ID Município - IBGE 7 Dígitos + tests: + - not_null + - relationships: + to: ref('br_bd_diretorios_brasil__municipio') + field: id_municipio + config: + where: __most_recent_year_month__ + - name: tipo_veiculo + description: Tipo do veículo + - name: especie + description: Espécie do veículo + - name: eixos + description: Número de eixos conforme informado pela fonte; 0 indica não informado + - name: quantidade + description: Quantidade de veículos diff --git a/pipelines/datasets/br_senatran_estatisticas/breakdowns.py b/pipelines/datasets/br_senatran_estatisticas/breakdowns.py index 2728610c40..61403056d3 100644 --- a/pipelines/datasets/br_senatran_estatisticas/breakdowns.py +++ b/pipelines/datasets/br_senatran_estatisticas/breakdowns.py @@ -55,8 +55,11 @@ class Layout: """Um recorte da frota publicado mensalmente.""" table_id: str - #: token que identifica o recorte no nome do arquivo, já normalizado - token: str + #: tokens que identificam o recorte no nome do arquivo, já normalizados. + #: Vários porque o gov.br renomeia os recortes entre anos — p.ex. o mesmo + #: recorte aparece como `ano_de_fabricacao_e_modelo` (2017), + #: `ano_fab_mod` e `ano_fab_modelo` (2021). + tokens: tuple[str, ...] #: nomes finais das colunas de dimensão, na ordem em que aparecem dimensions: tuple[str, ...] #: tokens que, se presentes, desqualificam o arquivo @@ -66,9 +69,44 @@ class Layout: LAYOUTS: dict[str, Layout] = { "municipio_combustivel": Layout( table_id="municipio_combustivel", - token="combustivel", + tokens=("combustivel",), dimensions=("combustivel",), ), + "municipio_cor": Layout( + table_id="municipio_cor", + tokens=("cor",), + dimensions=("cor",), + ), + "municipio_potencia": Layout( + table_id="municipio_potencia", + tokens=("potencia",), + dimensions=("potencia",), + ), + "municipio_restricao": Layout( + table_id="municipio_restricao", + tokens=("restricao",), + dimensions=("restricao",), + ), + "municipio_cep": Layout( + table_id="municipio_cep", + tokens=("cep",), + dimensions=("cep",), + ), + "municipio_ano_fabricacao_modelo": Layout( + table_id="municipio_ano_fabricacao_modelo", + tokens=( + "ano_fab", + "ano_fab_mod", + "ano_fab_modelo", + "ano_de_fabricacao", + ), + dimensions=("ano_modelo", "ano_fabricacao"), + ), + "municipio_tipo_especie_eixos": Layout( + table_id="municipio_tipo_especie_eixos", + tokens=("especie", "tipoespecieeixo", "tipo_especie_eixos"), + dimensions=("tipo_veiculo", "especie", "eixos"), + ), } @@ -121,7 +159,12 @@ def extract_breakdown_links( continue filename = href.rsplit("/", 1)[-1] normalized = normalize(filename) - if layout.token not in normalized: + # Casamento por token delimitado, não por substring: "cor" e "cep" são + # curtos e casariam dentro de "recorte", "concept" etc. + if not any( + re.search(rf"(?:^|_){re.escape(t)}(?:_|$)", normalized) + for t in layout.tokens + ): continue if any(bad in normalized for bad in layout.excludes): continue @@ -149,17 +192,71 @@ def extract_breakdown_links( return resolved +class UnsupportedArchiveError(RuntimeError): + """Arquivo compactado que não conseguimos abrir neste ambiente.""" + + +def _spreadsheet_from_archive(path: Path) -> Path: + """Extrai a planilha de dentro de um .zip/.rar e devolve seu caminho. + + Os recortes de 2013, 2015 e 2016 vêm compactados; de 2017 em diante são + .xlsx direto. O ``rarfile`` depende de um binário externo (unrar/bsdtar) que + existe no worker mas nem sempre localmente — por isso a falha é sinalizada + com :class:`UnsupportedArchiveError`, para o backfill pular o mês em vez de + abortar os outros 150. + """ + from zipfile import ZipFile + + extension = path.suffix.lower().lstrip(".") + if extension == "zip": + opener = ZipFile + elif extension == "rar": + from rarfile import RarFile + + opener = RarFile + else: + raise ValueError(f"Extensão não suportada: {extension}") + + destino = path.parent / f"{path.stem}_extraido" + destino.mkdir(parents=True, exist_ok=True) + try: + with opener(path) as arquivo: # type: ignore[operator] + arquivo.extractall(path=destino) + except Exception as erro: + raise UnsupportedArchiveError( + f"Não foi possível extrair {path.name}: {erro}" + ) from erro + + planilhas = [ + f + for f in sorted(destino.rglob("*")) + if f.suffix.lower() in {".xlsx", ".xls"} + ] + if not planilhas: + raise UnsupportedArchiveError( + f"Nenhuma planilha dentro de {path.name}" + ) + return planilhas[0] + + def read_breakdown(path: str | Path, layout: Layout) -> pl.DataFrame: """Lê um arquivo de recorte e devolve colunas já renomeadas. - O nome da planilha varia (``Layout C``, ``Layout D `` com espaço à direita), - então pegamos a primeira aba que não seja o glossário. + Aceita .xlsx/.xls direto ou compactado em .zip/.rar. O nome da planilha + varia (``Layout C``, ``Layout D `` com espaço à direita), então pegamos a + primeira aba que não seja o glossário. """ + path = Path(path) + if path.suffix.lower() in {".zip", ".rar"}: + path = _spreadsheet_from_archive(path) excel = pd.ExcelFile(path) sheets = [s for s in excel.sheet_names if normalize(str(s)) != "glossario"] if not sheets: raise ValueError(f"Nenhuma aba de dados em {path}") - frame = pd.read_excel(path, sheet_name=sheets[0]) + # dtype=str em tudo: o CEP vem com zeros à esquerda que o pandas destrói ao + # inferir int (069900 -> 69900), e o staging é all-STRING por convenção de + # qualquer forma — o safe_cast do modelo dbt decide o tipo final. + frame = pd.read_excel(path, sheet_name=sheets[0], dtype=str) expected = 2 + len(layout.dimensions) + 1 if frame.shape[1] < expected: diff --git a/pipelines/datasets/br_senatran_estatisticas/flows.py b/pipelines/datasets/br_senatran_estatisticas/flows.py index bef3908d9e..6775bec5eb 100644 --- a/pipelines/datasets/br_senatran_estatisticas/flows.py +++ b/pipelines/datasets/br_senatran_estatisticas/flows.py @@ -378,3 +378,189 @@ def br_senatran_estatisticas__municipio_combustivel( br_senatran_estatisticas__municipio_combustivel.deploy_schedules = [ {"cron": "40 21 10-30 * *", "timezone": "America/Sao_Paulo"} ] + + +@flow( + name="br_senatran_estatisticas__municipio_cor", + log_prints=True, +) +def br_senatran_estatisticas__municipio_cor( + dataset_id: str = "br_senatran_estatisticas", + table_id: str = "municipio_cor", + materialize_after_dump: bool = True, + update_metadata: bool = True, + target: str = "prod", + force_run: bool = False, + backfill_start: str | None = None, +) -> None: + _run_breakdown( + dataset_id=dataset_id, + table_id=table_id, + layout_key="municipio_cor", + materialize_after_dump=materialize_after_dump, + update_metadata=update_metadata, + target=target, + force_run=force_run, + backfill_start=backfill_start, + ) + + +# pyrefly: ignore [missing-attribute] +br_senatran_estatisticas__municipio_cor.deploy_schedules = [ + {"cron": "5 22 10-30 * *", "timezone": "America/Sao_Paulo"} +] + + +@flow( + name="br_senatran_estatisticas__municipio_potencia", + log_prints=True, +) +def br_senatran_estatisticas__municipio_potencia( + dataset_id: str = "br_senatran_estatisticas", + table_id: str = "municipio_potencia", + materialize_after_dump: bool = True, + update_metadata: bool = True, + target: str = "prod", + force_run: bool = False, + backfill_start: str | None = None, +) -> None: + _run_breakdown( + dataset_id=dataset_id, + table_id=table_id, + layout_key="municipio_potencia", + materialize_after_dump=materialize_after_dump, + update_metadata=update_metadata, + target=target, + force_run=force_run, + backfill_start=backfill_start, + ) + + +# pyrefly: ignore [missing-attribute] +br_senatran_estatisticas__municipio_potencia.deploy_schedules = [ + {"cron": "15 22 10-30 * *", "timezone": "America/Sao_Paulo"} +] + + +@flow( + name="br_senatran_estatisticas__municipio_restricao", + log_prints=True, +) +def br_senatran_estatisticas__municipio_restricao( + dataset_id: str = "br_senatran_estatisticas", + table_id: str = "municipio_restricao", + materialize_after_dump: bool = True, + update_metadata: bool = True, + target: str = "prod", + force_run: bool = False, + backfill_start: str | None = None, +) -> None: + _run_breakdown( + dataset_id=dataset_id, + table_id=table_id, + layout_key="municipio_restricao", + materialize_after_dump=materialize_after_dump, + update_metadata=update_metadata, + target=target, + force_run=force_run, + backfill_start=backfill_start, + ) + + +# pyrefly: ignore [missing-attribute] +br_senatran_estatisticas__municipio_restricao.deploy_schedules = [ + {"cron": "25 22 10-30 * *", "timezone": "America/Sao_Paulo"} +] + + +@flow( + name="br_senatran_estatisticas__municipio_cep", + log_prints=True, +) +def br_senatran_estatisticas__municipio_cep( + dataset_id: str = "br_senatran_estatisticas", + table_id: str = "municipio_cep", + materialize_after_dump: bool = True, + update_metadata: bool = True, + target: str = "prod", + force_run: bool = False, + backfill_start: str | None = None, +) -> None: + _run_breakdown( + dataset_id=dataset_id, + table_id=table_id, + layout_key="municipio_cep", + materialize_after_dump=materialize_after_dump, + update_metadata=update_metadata, + target=target, + force_run=force_run, + backfill_start=backfill_start, + ) + + +# pyrefly: ignore [missing-attribute] +br_senatran_estatisticas__municipio_cep.deploy_schedules = [ + {"cron": "35 22 10-30 * *", "timezone": "America/Sao_Paulo"} +] + + +@flow( + name="br_senatran_estatisticas__municipio_ano_fabricacao_modelo", + log_prints=True, +) +def br_senatran_estatisticas__municipio_ano_fabricacao_modelo( + dataset_id: str = "br_senatran_estatisticas", + table_id: str = "municipio_ano_fabricacao_modelo", + materialize_after_dump: bool = True, + update_metadata: bool = True, + target: str = "prod", + force_run: bool = False, + backfill_start: str | None = None, +) -> None: + _run_breakdown( + dataset_id=dataset_id, + table_id=table_id, + layout_key="municipio_ano_fabricacao_modelo", + materialize_after_dump=materialize_after_dump, + update_metadata=update_metadata, + target=target, + force_run=force_run, + backfill_start=backfill_start, + ) + + +# pyrefly: ignore [missing-attribute] +br_senatran_estatisticas__municipio_ano_fabricacao_modelo.deploy_schedules = [ + {"cron": "45 22 10-30 * *", "timezone": "America/Sao_Paulo"} +] + + +@flow( + name="br_senatran_estatisticas__municipio_tipo_especie_eixos", + log_prints=True, +) +def br_senatran_estatisticas__municipio_tipo_especie_eixos( + dataset_id: str = "br_senatran_estatisticas", + table_id: str = "municipio_tipo_especie_eixos", + materialize_after_dump: bool = True, + update_metadata: bool = True, + target: str = "prod", + force_run: bool = False, + backfill_start: str | None = None, +) -> None: + _run_breakdown( + dataset_id=dataset_id, + table_id=table_id, + layout_key="municipio_tipo_especie_eixos", + materialize_after_dump=materialize_after_dump, + update_metadata=update_metadata, + target=target, + force_run=force_run, + backfill_start=backfill_start, + ) + + +# pyrefly: ignore [missing-attribute] +br_senatran_estatisticas__municipio_tipo_especie_eixos.deploy_schedules = [ + {"cron": "55 22 10-30 * *", "timezone": "America/Sao_Paulo"} +] From b54c42a84edb329fba6cba4f0a68994ad5770980 Mon Sep 17 00:00:00 2001 From: Ricardo Dahis Date: Wed, 2 Sep 2026 09:10:46 +1000 Subject: [PATCH 2/5] fix(br_senatran_estatisticas): ano_modelo e ano_fabricacao sao STRING A fonte usa as duas colunas de ano tambem para sentinelas textuais - 'Nao Identificado', 'Nao se Aplica', 'Sem Informacao'. Com safe_cast(... as int64) esses valores viram NULL sem aviso: 36.567 celulas so em 2026-07, 4,5% das linhas em ano_fabricacao. Pior que a perda, isso quebrava o proprio teste de unicidade: as tres sentinelas colapsam num unico NULL, entao (ano, mes, id_municipio, ano_modelo, ano_fabricacao) deixa de ser chave. Medido no mes real: 28 chaves duplicadas com INT64, nenhuma com STRING. STRING segue a convencao da casa para coluna numerica com sentinela, e o proprio PR ja faz isso em eixos (que traz -2 e 99). Mudanca barata agora, quebra de schema depois de publicada. --- ...n_estatisticas__municipio_ano_fabricacao_modelo.sql | 4 ++-- models/br_senatran_estatisticas/schema.yml | 10 ++++++++-- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_ano_fabricacao_modelo.sql b/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_ano_fabricacao_modelo.sql index be2099898c..aed9d34abe 100644 --- a/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_ano_fabricacao_modelo.sql +++ b/models/br_senatran_estatisticas/br_senatran_estatisticas__municipio_ano_fabricacao_modelo.sql @@ -23,8 +23,8 @@ select safe_cast(mes as int64) mes, safe_cast(sigla_uf as string) sigla_uf, safe_cast(id_municipio as string) id_municipio, - safe_cast(ano_modelo as int64) ano_modelo, - safe_cast(ano_fabricacao as int64) ano_fabricacao, + safe_cast(lower(ano_modelo) as string) ano_modelo, + safe_cast(lower(ano_fabricacao) as string) ano_fabricacao, safe_cast(quantidade as int64) quantidade from {{ diff --git a/models/br_senatran_estatisticas/schema.yml b/models/br_senatran_estatisticas/schema.yml index 9e776e042f..290f9b41eb 100644 --- a/models/br_senatran_estatisticas/schema.yml +++ b/models/br_senatran_estatisticas/schema.yml @@ -320,9 +320,15 @@ models: config: where: __most_recent_year_month__ - name: ano_modelo - description: Ano do modelo do veículo + description: > + Ano do modelo do veículo. A fonte usa esta coluna também para os + sentinelas "não identificado", "não se aplica" e "sem informação", + então ela é STRING — como INT64 esses valores viravam NULL + silenciosamente - name: ano_fabricacao - description: Ano de fabricação do veículo conforme o CRV + description: > + Ano de fabricação do veículo conforme o CRV. STRING pelo mesmo + motivo de ano_modelo: 4,5% das linhas trazem sentinelas textuais - name: quantidade description: Quantidade de veículos - name: br_senatran_estatisticas__municipio_tipo_especie_eixos From a7adab7338964294630e482602d0f86f6ef19b50 Mon Sep 17 00:00:00 2001 From: Ricardo Dahis Date: Wed, 2 Sep 2026 09:31:28 +1000 Subject: [PATCH 3/5] fix(br_senatran_estatisticas): le todas as abas de dados, nao so a primeira Quando um recorte passa de 999.999 linhas a fonte continua numa segunda aba. O arquivo de potencia de julho/2026 tem 'Layout E' e 'Continuacao_Layout E'; read_breakdown pegava so a primeira e descartava a segunda em silencio. Como o arquivo e ordenado por UF, o que se perdia era a cauda: Sergipe, Tocantins e Sao Paulo a partir de Lencois Paulista - 554 municipios, 120.670 linhas, sem erro nenhum. Medido em 2026-07: antes 999.999 linhas brutas, 5.017 municipios depois 1.120.669 linhas brutas, 5.571 municipios O recorte e renomeado por posicao antes do concat: o cabecalho muda de grafia entre meses ('Municipio' vs 'MUNICIPIO') e pd.concat alinha por nome, entao juntar as abas cruas produziria colunas extras cheias de NaN. --- .../br_senatran_estatisticas/breakdowns.py | 46 ++++++++++++------- 1 file changed, 30 insertions(+), 16 deletions(-) diff --git a/pipelines/datasets/br_senatran_estatisticas/breakdowns.py b/pipelines/datasets/br_senatran_estatisticas/breakdowns.py index 61403056d3..6e0b65fe59 100644 --- a/pipelines/datasets/br_senatran_estatisticas/breakdowns.py +++ b/pipelines/datasets/br_senatran_estatisticas/breakdowns.py @@ -243,8 +243,15 @@ def read_breakdown(path: str | Path, layout: Layout) -> pl.DataFrame: """Lê um arquivo de recorte e devolve colunas já renomeadas. Aceita .xlsx/.xls direto ou compactado em .zip/.rar. O nome da planilha - varia (``Layout C``, ``Layout D `` com espaço à direita), então pegamos a - primeira aba que não seja o glossário. + varia (``Layout C``, ``Layout D `` com espaço à direita), então usamos toda + aba que não seja o glossário. + + **Todas as abas, não só a primeira.** Quando o recorte passa de 999.999 + linhas a fonte continua numa segunda aba — ``Layout E`` mais + ``Continuação_Layout E`` no recorte de potência. Ler só a primeira perde a + cauda do arquivo, que é ordenado por UF: em 2026-07 isso deixava de fora + Sergipe, Tocantins e São Paulo a partir de Lençóis Paulista, 554 municípios + ao todo, sem erro nenhum. """ path = Path(path) if path.suffix.lower() in {".zip", ".rar"}: @@ -256,21 +263,28 @@ def read_breakdown(path: str | Path, layout: Layout) -> pl.DataFrame: # dtype=str em tudo: o CEP vem com zeros à esquerda que o pandas destrói ao # inferir int (069900 -> 69900), e o staging é all-STRING por convenção de # qualquer forma — o safe_cast do modelo dbt decide o tipo final. - frame = pd.read_excel(path, sheet_name=sheets[0], dtype=str) - expected = 2 + len(layout.dimensions) + 1 - if frame.shape[1] < expected: - raise ValueError( - f"{path}: esperava >= {expected} colunas para o recorte " - f"{layout.table_id}, encontrou {frame.shape[1]}" - ) - frame = frame.iloc[:, :expected] - frame.columns = [ - "nome_uf", - "nome_denatran", - *layout.dimensions, - "quantidade", - ] + colunas = ["nome_uf", "nome_denatran", *layout.dimensions, "quantidade"] + + partes = [] + for aba in sheets: + parte = pd.read_excel(path, sheet_name=aba, dtype=str) + if parte.shape[1] < expected: + raise ValueError( + f"{path} [{aba}]: esperava >= {expected} colunas para o " + f"recorte {layout.table_id}, encontrou {parte.shape[1]}" + ) + # Recortar e renomear por posição *antes* de juntar: o cabeçalho muda + # de grafia entre meses (`Município` vs `MUNICIPIO`), e `pd.concat` + # alinha por nome — juntar as abas cruas produziria colunas extras + # cheias de NaN em vez de empilhar as linhas. + parte = parte.iloc[:, :expected] + parte.columns = colunas + partes.append(parte) + + frame = ( + partes[0] if len(partes) == 1 else pd.concat(partes, ignore_index=True) + ) return pl.from_pandas(frame.astype(str)) From f7fe0edf2be22e4645f2e202afe8cc9bfb4f5b16 Mon Sep 17 00:00:00 2001 From: Ricardo Dahis Date: Wed, 2 Sep 2026 09:37:45 +1000 Subject: [PATCH 4/5] fix(br_senatran_estatisticas): soma as chaves que o strip torna iguais clean_breakdown apara as colunas de dimensao no final, e a fonte emite variantes so de espaco do mesmo rotulo: '0' e '0 ' no recorte de CEP, 'GASOLINA' e 'GASOLINA ' no de combustivel. Depois do strip as duas viram a mesma chave, entao (ano, mes, id_municipio, ) deixa de ser unica e dbt_utils.unique_combination_of_columns reprova a tabela inteira. Nao era hipotetico: 38 chaves repetidas em municipio_cep 2026-07, e 7 na municipio_combustivel ja carregada em dev - sempre a linha cheia mais uma de quantidade 1. Somar preserva o total. Verificado em 2026-07, nos seis recortes: nenhuma chave duplicada, e cinco deles somam exatamente 132.323.803 veiculos cada, a frota nacional do mes. restricao soma diferente por construcao, ja que um veiculo pode ter mais de uma restricao. --- .../br_senatran_estatisticas/breakdowns.py | 23 +++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/pipelines/datasets/br_senatran_estatisticas/breakdowns.py b/pipelines/datasets/br_senatran_estatisticas/breakdowns.py index 6e0b65fe59..065390ab7c 100644 --- a/pipelines/datasets/br_senatran_estatisticas/breakdowns.py +++ b/pipelines/datasets/br_senatran_estatisticas/breakdowns.py @@ -407,9 +407,28 @@ def clean_breakdown( "quantidade", ] ) - return final.with_columns( + final = final.with_columns( [pl.col(dim).str.strip().alias(dim) for dim in layout.dimensions] - ), len(descartadas) + ) + + # Somar depois de aparar. A fonte emite variantes so de espaco do mesmo + # rotulo -- `'0'` e `'0 '` no recorte de CEP, `'GASOLINA'` e + # `'GASOLINA '` no de combustivel -- que viram a mesma chave assim que o + # `strip` roda, e ai `(ano, mes, id_municipio, )` deixa de ser + # unica e o teste do dbt reprova a tabela inteira. Medido: 38 chaves + # repetidas em municipio_cep 2026-07 e 7 em municipio_combustivel, sempre + # a linha cheia mais uma de quantidade 1. Somar preserva o total. + chave = ["ano", "mes", "sigla_uf", "id_municipio", *layout.dimensions] + final = final.group_by(chave, maintain_order=True).agg( + # `sum` devolve 0 quando o grupo e todo nulo; um quantidade ausente + # tem que continuar ausente, nao virar zero. + pl.when(pl.col("quantidade").is_null().all()) + .then(None) + .otherwise(pl.col("quantidade").sum()) + .alias("quantidade") + ) + + return final, len(descartadas) def reference_date(year: int, month: int) -> datetime.date: From ff993547336973949c3569336274219f116bbb02 Mon Sep 17 00:00:00 2001 From: Ricardo Dahis Date: Wed, 2 Sep 2026 09:38:49 +1000 Subject: [PATCH 5/5] test(br_senatran_estatisticas): fixa a deduplicacao do clean_breakdown Seis testes sobre o caso real: variantes de espaco somadas, municipios distintos preservados, quantidade ausente que continua ausente em vez de virar zero, e a dimensao aparada. De passagem, .str.strip() -> .str.strip_chars(): o nome antigo esta depreciado no polars e sai num upgrade. --- .../br_senatran_estatisticas/breakdowns.py | 2 +- .../tests/__init__.py | 0 .../tests/test_clean_breakdown_dedup.py | 103 ++++++++++++++++++ 3 files changed, 104 insertions(+), 1 deletion(-) create mode 100644 pipelines/datasets/br_senatran_estatisticas/tests/__init__.py create mode 100644 pipelines/datasets/br_senatran_estatisticas/tests/test_clean_breakdown_dedup.py diff --git a/pipelines/datasets/br_senatran_estatisticas/breakdowns.py b/pipelines/datasets/br_senatran_estatisticas/breakdowns.py index 065390ab7c..35a24b84f5 100644 --- a/pipelines/datasets/br_senatran_estatisticas/breakdowns.py +++ b/pipelines/datasets/br_senatran_estatisticas/breakdowns.py @@ -408,7 +408,7 @@ def clean_breakdown( ] ) final = final.with_columns( - [pl.col(dim).str.strip().alias(dim) for dim in layout.dimensions] + [pl.col(dim).str.strip_chars().alias(dim) for dim in layout.dimensions] ) # Somar depois de aparar. A fonte emite variantes so de espaco do mesmo diff --git a/pipelines/datasets/br_senatran_estatisticas/tests/__init__.py b/pipelines/datasets/br_senatran_estatisticas/tests/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/pipelines/datasets/br_senatran_estatisticas/tests/test_clean_breakdown_dedup.py b/pipelines/datasets/br_senatran_estatisticas/tests/test_clean_breakdown_dedup.py new file mode 100644 index 0000000000..9ad8ce85b9 --- /dev/null +++ b/pipelines/datasets/br_senatran_estatisticas/tests/test_clean_breakdown_dedup.py @@ -0,0 +1,103 @@ +"""A fonte repete a chave com variantes de espaço; `clean_breakdown` tem que somar. + +Sem isso `dbt_utils.unique_combination_of_columns` reprova a tabela inteira — +foi o que aconteceria com `municipio_cep` (38 chaves) e `municipio_combustivel` +(7 chaves). +""" + +import polars as pl +import pytest + +from pipelines.datasets.br_senatran_estatisticas.breakdowns import ( + LAYOUTS, + clean_breakdown, +) + +IBGE = pl.DataFrame( + { + "nome": ["Goiânia", "Anápolis"], + "id_municipio": ["5208707", "5201108"], + "sigla_uf": ["GO", "GO"], + } +) + + +def _bruto(linhas: list[tuple[str, str, str, str]]) -> pl.DataFrame: + return pl.DataFrame( + { + "nome_uf": [x[0] for x in linhas], + "nome_denatran": [x[1] for x in linhas], + "combustivel": [x[2] for x in linhas], + "quantidade": [x[3] for x in linhas], + } + ) + + +LAYOUT = LAYOUTS["municipio_combustivel"] + + +def test_soma_variantes_de_espaco_da_mesma_dimensao(): + """O caso real: 'GASOLINA' e 'GASOLINA ' viram a mesma chave após o strip.""" + final, _ = clean_breakdown( + _bruto( + [ + ("GOIAS", "GOIANIA", "GASOLINA", "9377"), + ("GOIAS", "GOIANIA", "GASOLINA ", "1"), + ("GOIAS", "GOIANIA", "ALCOOL", "42"), + ] + ), + LAYOUT, + 2026, + 7, + IBGE, + ) + + chave = ["ano", "mes", "id_municipio", "combustivel"] + assert final.group_by(chave).len().filter(pl.col("len") > 1).height == 0 + + gasolina = final.filter(pl.col("combustivel") == "GASOLINA") + assert gasolina.height == 1 + assert gasolina.get_column("quantidade").item() == 9378 # 9377 + 1 + # o total do município tem que sobreviver à soma + assert final.get_column("quantidade").sum() == 9420 + + +def test_nao_junta_municipios_diferentes(): + final, _ = clean_breakdown( + _bruto( + [ + ("GOIAS", "GOIANIA", "GASOLINA", "10"), + ("GOIAS", "ANAPOLIS", "GASOLINA", "20"), + ] + ), + LAYOUT, + 2026, + 7, + IBGE, + ) + assert final.height == 2 + assert set(final.get_column("id_municipio").to_list()) == { + "5208707", + "5201108", + } + + +def test_quantidade_ausente_continua_ausente(): + """`sum` devolveria 0 para um grupo todo nulo; NULL não é zero veículos.""" + final, _ = clean_breakdown( + _bruto([("GOIAS", "GOIANIA", "GASOLINA", "sem informação")]), + LAYOUT, + 2026, + 7, + IBGE, + ) + assert final.height == 1 + assert final.get_column("quantidade").item() is None + + +@pytest.mark.parametrize("dim", ["0", "0 ", " 0"]) +def test_dimensao_e_aparada(dim): + final, _ = clean_breakdown( + _bruto([("GOIAS", "GOIANIA", dim, "5")]), LAYOUT, 2026, 7, IBGE + ) + assert final.get_column("combustivel").item() == "0"