feat(eixo1): fontes do Cultura em Números por DAG e IPCA 433 no BACEN

dags/data_ingest/bacen/bacen_sgs_ingest_dag.py

@@ -30,7 +30,9 @@
         }
     ]
 
-Sem a Variable, a DAG ingere só o IPCA-Serviços. Todas as séries caem na
+Sem a Variable, a DAG ingere o IPCA-Serviços e o IPCA cheio (SGS 433). O
+433 é o deflator dos scripts do Cultura em Números, Eixo 1 (orçamento das
+três esferas a preços de 2024). Todas as séries caem na
 mesma tabela, identificadas por ``serie``/``codigo_serie`` — assim uma série
 nova não exige tabela nova nem source dbt novo.
 
@@ -56,9 +58,9 @@
 _SCHEMA = "bacen"
 _TABLE = "serie_sgs"
 
-# SGS 10844 — IPCA Serviços (variação % mensal). Só o default: o que vale é
-# a Variable ``bacen_series_sgs``.
-_SERIES_PADRAO = {"ipca_servicos": 10844}
+# SGS 10844 — IPCA Serviços e SGS 433 — IPCA (variação % mensal). Só o
+# default: o que vale é a Variable ``bacen_series_sgs``.
+_SERIES_PADRAO = {"ipca_servicos": 10844, "ipca": 433}
 
 default_args = {
     "owner": "Lucas Bottino",

dags/data_ingest/bsc_pnab/extracao_bbagil_dag.py

@@ -6,7 +6,11 @@
 from airflow.sdk import dag, task
 
 import schemas_minc as schemas
-from agencias_bbagil import gerar_periodos_mensais
+from agencias_bbagil import (
+    gerar_periodos_mensais,
+    periodos_da_conta,
+    periodos_extras_por_programa,
+)
 from cliente_bsc import (
     AsyncBscClient,
     BscRequestError,
@@ -277,7 +281,10 @@ def _consultar_ou_vazio(db: ClientPostgresDB, query: str, descricao: str) -> lis
 
 
 def _combinacoes_extrato_pendentes(
-    db: ClientPostgresDB, entes: list[dict[str, Any]], periodos: list[tuple[str, str]]
+    db: ClientPostgresDB,
+    entes: list[dict[str, Any]],
+    periodos: list[tuple[str, str]],
+    periodos_extras: dict[str, list[tuple[str, str]]] | None = None,
 ) -> list[Any]:
     # A conta (``id_plano_acao_dado_bancario``) faz parte do checkpoint: sem
     # ela, a segunda conta de um plano seria dada como ja feita pela primeira e
@@ -295,7 +302,13 @@ def _combinacoes_extrato_pendentes(
         for id_plano_acao, id_dado_bancario, periodo_inicial, periodo_final in feitas
     }
 
-    combinacoes = [(ente, periodo) for ente in entes for periodo in periodos]
+    combinacoes = [
+        (ente, periodo)
+        for ente in entes
+        for periodo in periodos_da_conta(
+            ente.get("codigo_programa"), periodos, periodos_extras or {}
+        )
+    ]
     return [
         (ente, periodo)
         for ente, periodo in combinacoes
@@ -367,13 +380,23 @@ def _persistir_resultados_extrato(
 def _extrair_extrato(agencias_periodos: dict[str, Any]) -> dict[str, int]:
     entes = agencias_periodos["entes"]
     periodos = [tuple(p) for p in agencias_periodos["periodos"]]
+    # O XCom devolve as tuplas como listas.
+    periodos_extras = {
+        codigo: [tuple(p) for p in lista]
+        for codigo, lista in agencias_periodos.get("periodos_extras", {}).items()
+    }
     db = ClientPostgresDB(get_postgres_conn())
 
-    pendentes = _combinacoes_extrato_pendentes(db, entes, periodos)
+    pendentes = _combinacoes_extrato_pendentes(db, entes, periodos, periodos_extras)
     logging.info(
-        "[extracao_bbagil_dag] Extrato BB Agil: %d/%d combinacoes pendentes",
+        "[extracao_bbagil_dag] Extrato BB Agil: %d/%d combinacoes pendentes "
+        "(%d programas com anos extras)",
         len(pendentes),
-        len(entes) * len(periodos),
+        sum(
+            len(periodos_da_conta(e.get("codigo_programa"), periodos, periodos_extras))
+            for e in entes
+        ),
+        len(periodos_extras),
     )
 
     contagem_total = {"ok": 0, "sem_dados": 0, "erro": 0}
@@ -553,7 +576,11 @@ def extracao_bbagil_dag() -> None:
        acao de ``transferegov.plano_acao_dado_bancario_minc`` (uma entrada por
        conta distinta; planos da LPG tem duas) e gera a lista de periodos
        mensais a extrair. Quem extrai as contas da API e
-       ``api_plano_acao_dado_bancario_dag`` -- esta DAG so consome.
+       ``api_plano_acao_dado_bancario_dag`` -- esta DAG so consome. As contas
+       de um programa listado na Variable
+       ``bsc_pnab_anos_extras_por_programa`` recebem tambem os meses dos anos
+       dele (e como se pede a LAB 1 de 2020-2022 sem pedir esses anos para
+       todas as contas).
     2. ``extrair_extrato_bbagil`` -- para cada conta x periodo pendente,
        chama o extrato via ``AsyncBscClient``. HTTP 400 "sem lancamentos" e
        "conta corrente invalida" sao registrados como dado de negocio, nao
@@ -602,6 +629,7 @@ def carregar_contas_bancarias() -> dict[str, Any]:
         return {
             "entes": _carregar_entes_transferegov(),
             "periodos": gerar_periodos_mensais(),
+            "periodos_extras": periodos_extras_por_programa(),
         }
 
     @task

dags/data_ingest/ibge/ibge_perfil_dag.py

@@ -0,0 +1,175 @@
+"""Ingestão das bases da MUNIC e da ESTADIC do IBGE (Eixo 1 do Cultura em Números).
+
+Fonte estática, publicada uma vez por edição, e por isso segue o caminho de
+seeds do ADR 0012::
+
+    ftp.ibge.gov.br ─▶ raw/ibge/perfil/edicao=<edicao>/<run>.<zip|xls|xlsx>
+                    ─▶ staging/ibge/<edicao>_<aba>/edicao=<edicao>/<run>.parquet
+                    ─▶ Postgres seeds.ibge_<edicao>_<aba>
+
+Uma tabela por aba e edição, porque os códigos de variável (``A2``,
+``MCUL01``) mudam de sentido de uma edição para outra: juntar 2006 e 2014
+na mesma coluna misturaria perguntas diferentes. Cada edição substitui só a
+si mesma (``DELETE`` + ``COPY`` por ``edicao``), então reexecutar não
+duplica. O raw nunca é sobrescrito: cada run grava a própria key, em partes
+se passar do limite do proxy do MinIO.
+
+As edições, as abas esperadas e quantos entes cada aba de dados deve ter
+estão em ``cliente_ibge_perfil.EDICOES``. A conferência contra a origem vai
+para ``seeds.ibge_controle_carga``.
+
+Manual: o IBGE publica edição nova sem data fixa. Dispare com
+``{"edicoes": ["munic_2021"]}`` para uma edição, ou vazio para todas.
+"""
+
+import logging
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from airflow.sdk import Param, dag, get_current_context, task
+
+import cliente_ibge_perfil as ibge
+import datalakehouse
+import schemas_minc as schemas
+import seeds_carga
+from openmetadata.lineage import publicar_linhagem, tabela
+from postgres_helpers import get_postgres_conn
+
+_ENTIDADE_RAW = "perfil"
+
+ASSET_STAGING_IBGE = datalakehouse.asset(
+    f"{datalakehouse.CAMADA_STAGING}/{schemas.FONTE_IBGE}/"
+)
+
+default_args = {
+    "owner": "MinC",
+    "retries": 2,
+    "retry_delay": timedelta(minutes=5),
+}
+
+
+def _run_id() -> str:
+    return str(get_current_context()["dag_run"].run_id)
+
+
+@dag(
+    dag_id="ibge_perfil_dag",
+    schedule=None,
+    start_date=datetime(2026, 1, 1),
+    catchup=False,
+    default_args=default_args,
+    max_active_runs=1,
+    params={
+        "edicoes": Param(
+            [],
+            type="array",
+            items={"type": "string", "enum": sorted(ibge.EDICOES)},
+            description="Edições a carregar. Vazio carrega todas do catálogo.",
+        )
+    },
+    tags=["minc", "ibge", "munic", "estadic", "datalakehouse", "seeds"],
+)
+def ibge_perfil_dag() -> None:
+    """Carrega as bases da MUNIC e da ESTADIC do IBGE no schema seeds,
+    uma tabela por aba e edição."""
+
+    @task
+    def listar_edicoes() -> list[str]:
+        pedidas = get_current_context()["params"]["edicoes"] or sorted(ibge.EDICOES)
+        desconhecidas = set(pedidas) - set(ibge.EDICOES)
+        if desconhecidas:
+            raise ValueError(f"edições fora do catálogo: {sorted(desconhecidas)}")
+        return sorted(pedidas)
+
+    @task(max_active_tis_per_dag=2, execution_timeout=timedelta(minutes=30))
+    def baixar_para_raw(edicao: str) -> dict[str, Any]:
+        """Baixa o arquivo da edição e grava em ``raw/`` como veio."""
+        info = ibge.EDICOES[edicao]
+        conteudo = ibge.baixar(info.url)
+        key = datalakehouse.caminho_particionado(
+            datalakehouse.CAMADA_RAW,
+            schemas.FONTE_IBGE,
+            _ENTIDADE_RAW,
+            {"edicao": edicao},
+            _run_id(),
+            ibge.extensao(info.caminho),
+        )
+        datalakehouse.gravar_bytes_em_partes(conteudo, key)
+        return {"edicao": edicao, "key_raw": key, "arquivo": info.caminho}
+
+    @task(max_active_tis_per_dag=1, execution_timeout=timedelta(hours=1))
+    def raw_para_parquet(item: dict[str, Any]) -> list[dict[str, Any]]:
+        """Uma aba vira um Parquet no staging; confere abas e linhas por ente."""
+        edicao, key_raw = item["edicao"], item["key_raw"]
+        dt_ingest = datetime.now(timezone.utc)
+        nome, planilha = ibge.planilha_do_arquivo(
+            datalakehouse.ler_bytes(key_raw), item["arquivo"]
+        )
+        livro = ibge.abrir_livro(planilha)
+        ibge.conferir_abas(edicao, list(livro.sheet_names))
+
+        arquivos = []
+        for aba in livro.sheet_names:
+            registros = ibge.registros_da_aba(edicao, aba, ibge.linhas_da_aba(livro, aba))
+            entidade = ibge.entidade(edicao, aba)
+            key = datalakehouse.caminho_particionado(
+                datalakehouse.CAMADA_STAGING,
+                schemas.FONTE_IBGE,
+                entidade,
+                {"edicao": edicao},
+                _run_id(),
+                "parquet",
+            )
+            datalakehouse.gravar_parquet(registros, key, dt_ingest=dt_ingest)
+            logging.info(
+                "[ibge_perfil_dag] %s / %s: %d linhas", edicao, aba, len(registros)
+            )
+            arquivos.append(
+                {
+                    "entidade": entidade,
+                    "periodo": edicao,
+                    "tabela": ibge.tabela_seeds(edicao, aba),
+                    "arquivo": f"{nome} / {aba}",
+                    "linhas_origem": len(registros),
+                    "key_raw": key_raw,
+                    "key_staging": key,
+                    "dt_ingest": dt_ingest.isoformat(),
+                }
+            )
+        return arquivos
+
+    @task(
+        max_active_tis_per_dag=1,
+        execution_timeout=timedelta(hours=1),
+        outlets=[ASSET_STAGING_IBGE]
+        + [tabela(schemas.SCHEMA_SEEDS, t) for t in ibge.tabelas_seeds()]
+        + [tabela(schemas.SCHEMA_SEEDS, seeds_carga.tabela_controle(schemas.FONTE_IBGE))],
+    )
+    def parquet_para_seeds(arquivos: list[dict[str, Any]]) -> None:
+        """Substitui a edição em cada tabela de ``seeds`` e registra a conferência."""
+        conn = get_postgres_conn()
+        cargas = []
+        for item in arquivos:
+            df = datalakehouse.ler_parquet(item["key_staging"])
+            carregadas = seeds_carga.substituir_periodo(
+                conn,
+                df,
+                schemas.SCHEMA_SEEDS,
+                item["tabela"],
+                ibge.COLUNA_EDICAO,
+                item["periodo"],
+            )
+            cargas.append(
+                {**item, "linhas_parquet": len(df), "linhas_carregadas": carregadas}
+            )
+        seeds_carga.registrar_carga(
+            conn, schemas.SCHEMA_SEEDS, schemas.FONTE_IBGE, cargas
+        )
+        seeds_carga.conferir_contagens(cargas)
+
+    brutos = baixar_para_raw.expand(edicao=listar_edicoes())
+    convertidos = raw_para_parquet.expand(item=brutos)
+    parquet_para_seeds.expand(arquivos=convertidos) >> publicar_linhagem()
+
+
+ibge_perfil_dag()