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..aed9d34abe --- /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(lower(ano_modelo) as string) ano_modelo, + safe_cast(lower(ano_fabricacao) as string) 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..290f9b41eb 100644 --- a/models/br_senatran_estatisticas/schema.yml +++ b/models/br_senatran_estatisticas/schema.yml @@ -112,3 +112,273 @@ 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. 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. 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 + 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..35a24b84f5 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,31 +192,99 @@ 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 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"}: + 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. 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)) @@ -296,9 +407,28 @@ def clean_breakdown( "quantidade", ] ) - return final.with_columns( - [pl.col(dim).str.strip().alias(dim) for dim in layout.dimensions] - ), len(descartadas) + final = final.with_columns( + [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 + # 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: 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"} +] 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"