feat(br_ms_sim): pipeline sob demanda, com os dados preliminares do DATASUS

models/br_ms_sim/br_ms_sim__microdados.sql

@@ -1,5 +1,3 @@
--- Microdados SIM: carga 2020-2024; particao end: 2024 (ver README).
--- Staging reprocessado com cleaning.py (circunstancia_obito, estado_civil).
 {{
     config(
         alias="microdados",
@@ -8,7 +6,7 @@
         partition_by={
             "field": "ano",
             "data_type": "int64",
-            "range": {"start": 1996, "end": 2024, "interval": 1},
+            "range": {"start": 1996, "end": 2031, "interval": 1},
         },
         cluster_by="sigla_uf",
     )
@@ -105,5 +103,6 @@ select
     safe_cast(tipo_nivel_investigador as string) tipo_nivel_investigador,
     safe_cast(numero_dias_informacao as int64) numero_dias_informacao,
     safe_cast(fontes_informacao as string) fontes_informacao,
-    safe_cast(alt_causa as string) alt_causa
+    safe_cast(alt_causa as string) alt_causa,
+    safe_cast(coalesce(dado_preliminar, '0') as string) dado_preliminar
 from {{ set_datalake_project("br_ms_sim_staging.microdados") }} as t

models/br_ms_sim/schema.yml

@@ -213,6 +213,10 @@ models:
         description: Fontes Informação
       - name: alt_causa
         description: Alt. Causa
+      - name: dado_preliminar
+        description: >-
+          Indica se o registro vem da versão preliminar da fonte, ainda sujeita
+          a revisão (1), ou da versão definitiva (0)
   - name: br_ms_sim__dicionario
     description: Dicionário para tradução dos códigos do conjunto br_ms_sim. Para
       taduzir códigos compartilhados entre instituições, como id_municipio, buscar

pipelines/datasets/br_ms_sim/README.md

@@ -0,0 +1,102 @@
+# br_ms_sim — pipeline
+
+Carga dos microdados de óbitos não fetais (CID-10) do SIM/DATASUS, do FTP até a
+materialização em `basedosdados.br_ms_sim.microdados`.
+
+Contexto da base, investigações de qualidade e decisões de tratamento estão em
+[`models/br_ms_sim/README.md`](../../../models/br_ms_sim/README.md).
+
+## Sem schedule, por quê
+
+O DATASUS republica o SIM duas vezes por ano, sem data fixa. O flow é deployado
+sem `deploy_schedules`: o deployment existe, aceita execução avulsa e não dispara
+sozinho. Para armar depois, basta acrescentar a lista de crons em `flows.py`.
+
+## Definitivo e preliminar
+
+A fonte serve o mesmo ano em dois diretórios:
+
+| Diretório | Conteúdo |
+|---|---|
+| `SIM/CID10/DORES/` | definitivo, publicado cerca de um ano após o fechamento |
+| `SIM/PRELIM/DORES/` | preliminar, ainda sujeito a revisão |
+
+`resolve_year_source` dá precedência ao definitivo. Reprocessar um ano depois do
+fechamento troca o dado e muda `dado_preliminar` de `1` para `0` — não há passo
+manual para a virada, só rodar o flow com aquele ano.
+
+## Parâmetros
+
+| Parâmetro | Padrão | Efeito |
+|---|---|---|
+| `ano` | vazio | Vazio pega o ano mais recente da fonte. Preenchido é backfill: o flow pula o poll e não mexe no metadado da fonte |
+| `materialize_after_dump` | `True` | Sobe para prod e materializa lá |
+| `update_metadata` | `True` | Registra a cobertura materializada |
+| `force_run` | `False` | Materializa mesmo sem novidade na fonte |
+
+Execução de teste no pool de dev, sem tocar em produção:
+
+```json
+{"materialize_after_dump": false, "update_metadata": false, "force_run": true}
+```
+
+Os padrões escrevem em **produção**, mesmo saindo do pool de teste.
+
+## Qual ano entra
+
+Sem `ano`, o flow carrega o ano mais recente que existe na fonte, que é o ano em
+curso: o diretório preliminar publica o ano corrente antes de ele fechar.
+
+O poll compara esse ano com o fim da cobertura da tabela. Depois que a cobertura
+alcança o ano, as execuções seguintes encerram sem carregar nada, e as revisões
+que o DATASUS publicar naquele ano preliminar não entram. Para trazê-las, executar
+com `ano` preenchido ou com `force_run`.
+
+## Formato da staging
+
+A staging é CSV desde a carga original. O particionado sai em
+`ano=<ano>/sigla_uf=<UF>/microdados.csv` e sobe com `dump_mode="append"`: os
+caminhos são fixos, então reenviar um ano sobrescreve aquele ano e preserva o
+resto da série. `overwrite` apagaria o prefixo inteiro, com ele 1996 em diante.
+
+Trocar para parquet exigiria recriar a tabela externa e, com ela, recarregar
+toda a série.
+
+## Carga manual
+
+`utils.py` não importa Prefect, então a transformação roda fora do flow:
+
+```python
+from pipelines.datasets.br_ms_sim import utils
+
+source = utils.resolve_year_source(2025)
+utils.download_table("microdados", 2025, source)
+utils.clean_table("microdados", 2025, source)
+```
+
+## Série histórica
+
+Os CSVs de 1996–2024 na staging foram gravados pela limpeza anterior, que
+nulificava o código `6` de `local_ocorrencia` (aldeia indígena) antes do upload.
+O dado não chegou à staging, então `full-refresh` não o recupera — só
+reprocessar os `.dbc`. É o que faz `reprocess_br_ms_sim.py`, na raiz do repo.
+
+Enquanto o histórico não é reprocessado, ele também não traz `dado_preliminar`,
+e a coluna volta nula nesses anos. O modelo aplica `coalesce(dado_preliminar,
+'0')`: tudo anterior à primeira execução do pipeline veio do diretório
+definitivo. Depois do reprocessamento o `coalesce` deixa de encontrar nulos e
+continua correto.
+
+## Pontos de atenção
+
+- `dado_preliminar` é coluna nova no modelo, e em `dump_mode="append"` quem
+  amplia o schema da tabela externa é o `_sync_staging_schema`
+  (`pipelines/utils/tasks.py`). Esse caminho nunca tinha sido exercitado e
+  falhava com `403 bigquery.tables.update`, porque abria o cliente do BigQuery
+  sem credencial; corrigido nesta mesma branch para usar o cliente da lib.
+- O dicionário do conjunto vem de uma staging própria, alimentada por
+  `models/br_ms_sim/code/update_dicionario.py`. As linhas de `dado_preliminar`
+  precisam ser acrescentadas por lá.
+- Os scripts em `models/br_ms_sim/code/microdados/` são a carga anterior e não
+  conhecem `dado_preliminar`: gravam CSV com uma coluna a menos do que o modelo
+  espera. Usar o flow.