data/br_rf_cnpj__socios_legado

models/br_bd_diretorios_brasil/br_bd_diretorios_brasil__empresa.sql

@@ -10,7 +10,8 @@
 --
 with
     max_bdpro_date as (
-        select max(data) as max_date from `basedosdados.br_me_cnpj.estabelecimentos`
+        select max(data_referencia) as max_date
+        from `basedosdados.br_rf_cnpj.estabelecimentos`
     ),
 
     estabelecimento as (
@@ -152,12 +153,12 @@ with
             concat(ddd_2, " ", telefone_2) as telefone_2,
             concat(ddd_fax, " ", fax) as fax,
             email,
-        from `basedosdados.br_me_cnpj.estabelecimentos` a
+        from `basedosdados.br_rf_cnpj.estabelecimentos` a
         inner join
-            `basedosdados.br_me_cnpj.dicionario` b
+            `basedosdados.br_rf_cnpj.dicionario` b
             on a.identificador_matriz_filial = b.chave
         inner join
-            `basedosdados.br_me_cnpj.dicionario` t
+            `basedosdados.br_rf_cnpj.dicionario` t
             on a.identificador_matriz_filial = t.chave
         left join
             `basedosdados-dev.br_bd_diretorios_brasil_staging.bairro_code_iso3` g
@@ -166,7 +167,7 @@ with
             `basedosdados-dev.br_bd_diretorios_mundo_staging.pais_code` f
             on a.id_pais = f.co_pais
         where
-            a.data = (select max_date from max_bdpro_date)
+            a.data_referencia = (select max_date from max_bdpro_date)
             and b.nome_coluna = 'identificador_matriz_filial'
             and t.nome_coluna = 'situacao_cadastral'
     ),
@@ -178,14 +179,16 @@ with
             ente_federativo,
             capital_social,
             b.valor as porte,
-            a.data
-        from `basedosdados.br_me_cnpj.empresas` a
-        inner join `basedosdados.br_me_cnpj.dicionario` b on a.porte = b.chave
-        where b.nome_coluna = 'porte' and a.data = (select max_date from max_bdpro_date)
+            a.data_referencia
+        from `basedosdados.br_rf_cnpj.empresas` a
+        inner join `basedosdados.br_rf_cnpj.dicionario` b on a.porte = b.chave
+        where
+            b.nome_coluna = 'porte'
+            and a.data_referencia = (select max_date from max_bdpro_date)
     ),
     simples as (
         select distinct cnpj_basico, opcao_simples, opcao_mei
-        from `basedosdados.br_me_cnpj.simples`
+        from `basedosdados.br_rf_cnpj.simples`
     )
 
 select

models/br_tse_eleicoes/code/python/diagnostics/tier2_overrides.json

@@ -92,4 +92,4 @@
    "1950 and 1958 leiame layouts verified identical; 1954 files share the historical republication format."
   ]
  }
-}
+}

models/us_harvard_cbdb/code/clean.py

@@ -14,6 +14,8 @@
 import pandas as pd
 import pyarrow as pa
 import pyarrow.parquet as pq
+
+# pyrefly: ignore [missing-import]
 from schema_spec import DICT_SOURCES, TABLE_ORDER, TABLES
 
 HOME = os.path.expanduser("~")

models/us_harvard_cbdb/code/gen_architecture.py

@@ -11,6 +11,7 @@
 import csv
 import os
 
+# pyrefly: ignore [missing-import]
 from schema_spec import TABLE_ORDER, TABLES
 
 OUT = os.path.join(os.path.dirname(__file__), "architecture")

models/us_harvard_cbdb/code/gen_dbt.py

@@ -6,6 +6,7 @@
 
 import os
 
+# pyrefly: ignore [missing-import]
 from schema_spec import TABLE_ORDER, TABLES
 
 MODELS = os.path.abspath(

models/us_harvard_cbdb/code/gen_metadata_payload.py

@@ -3,6 +3,7 @@
 import json
 import os
 
+# pyrefly: ignore [missing-import]
 from schema_spec import TABLE_ORDER, TABLES
 
 OUT = os.path.join(os.path.dirname(__file__), "metadata_payload")
@@ -25,7 +26,9 @@
 
 def main():
     index = {}
+    # pyrefly: ignore [unknown-name]
     for name in TABLE_ORDER:
+        # pyrefly: ignore [unknown-name]
         spec = TABLES[name]
         cols = []
         for c in spec["columns"]:
@@ -45,6 +48,7 @@ def main():
         with open(path, "w", encoding="utf-8") as f:
             json.dump(cols, f, ensure_ascii=False)
         index[name] = dict(
+            # pyrefly: ignore [unknown-name]
             source=TABLES[name]["source"],
             name_pt=spec["name_pt"],
             name_en=spec["name_en"],

models/us_harvard_cbdb/code/schema_spec.py

@@ -610,6 +610,7 @@ def col_year(src, name, pt, en, es):
     desc_pt="Vocabulário de cargos oficiais do governo imperial chinês, com nomes em pinyin, chinês e tradução em inglês, por dinastia.",
     desc_en="Vocabulary of official offices of the Chinese imperial government, with pinyin, Chinese, and English names, by dynasty.",
     desc_es="Vocabulario de cargos oficiales del gobierno imperial chino, con nombres en pinyin, chino y traducción al inglés, por dinastía.",
+    # pyrefly: ignore [bad-assignment]
     ol=None,
     columns=[
         C(
@@ -691,6 +692,7 @@ def col_year(src, name, pt, en, es):
     desc_pt="Gazetteer de lugares históricos referidos no CBDB, com nomes em pinyin e chinês, tipo administrativo e coordenadas geográficas.",
     desc_en="Gazetteer of historical places referenced in CBDB, with pinyin and Chinese names, administrative type, and geographic coordinates.",
     desc_es="Gaceteer de lugares históricos referidos en el CBDB, con nombres en pinyin y chino, tipo administrativo y coordenadas geográficas.",
+    # pyrefly: ignore [bad-assignment]
     ol=None,
     columns=[
         C(
@@ -798,6 +800,7 @@ def col_year(src, name, pt, en, es):
     desc_pt="Vocabulário de tipos de relação de parentesco, com o termo em chinês e inglês e a distância genealógica em gerações e passos.",
     desc_en="Vocabulary of kinship-relation types, with the Chinese and English term and the genealogical distance in generations and steps.",
     desc_es="Vocabulario de tipos de relación de parentesco, con el término en chino e inglés y la distancia genealógica en generaciones y pasos.",
+    # pyrefly: ignore [bad-assignment]
     ol=None,
     columns=[
         C(
@@ -911,6 +914,7 @@ def col_year(src, name, pt, en, es):
     desc_pt="Vocabulário de tipos de conexão social não familiar, com descrição em inglês e chinês e o tipo de papel.",
     desc_en="Vocabulary of non-kin social-connection types, with English and Chinese description and role type.",
     desc_es="Vocabulario de tipos de conexión social no familiar, con descripción en inglés y chino y el tipo de rol.",
+    # pyrefly: ignore [bad-assignment]
     ol=None,
     columns=[
         C(
@@ -969,13 +973,16 @@ def col_year(src, name, pt, en, es):
 # ------------------------------------------------------------ dicionario -----
 # Standard BD dictionary. Populated by clean.py from the small CBDB code tables.
 TABLES["dicionario"] = dict(
+    # pyrefly: ignore [bad-assignment]
     source=None,
     name_en="Dictionary",
     name_pt="Dicionário",
     name_es="Diccionario",
     desc_pt="Dicionário que traduz os valores codificados das colunas cobertas por dicionário nas demais tabelas do conjunto.",
+    # pyrefly: ignore [bad-assignment]
     desc_en="Dictionary translating the coded values of dictionary-covered columns in the other tables of this dataset.",
     desc_es="Diccionario que traduce los valores codificados de las columnas cubiertas por diccionario en las demás tablas del conjunto.",
+    # pyrefly: ignore [bad-assignment]
     ol=None,
     columns=[
         C(

models/us_harvard_cbdb/code/upload.py

@@ -14,6 +14,8 @@
 
 import basedosdados as bd
 from google.cloud import storage as _gcs
+
+# pyrefly: ignore [missing-import]
 from schema_spec import TABLE_ORDER
 
 BILLING = "basedosdados-dev"
@@ -28,6 +30,7 @@ def _bucket(self, name, user_project=BILLING):
     return _orig_bucket(self, name, user_project=user_project)
 
 
+# pyrefly: ignore [bad-assignment]
 _gcs.Client.bucket = _bucket
 
 

pipelines/crawler/rf_cnpj/flows.py

@@ -150,7 +150,6 @@ def _run_rf_cnpj(
 
     if update_metadata:
         if table_id == "simples" or table_id == "dicionario":
-            # historical_database=False (sem coluna de data confiável) → NonHistorical
             register_table_materialization_task(
                 dataset_id=dataset_id,
                 table_id=table_id,
@@ -159,14 +158,12 @@ def _run_rf_cnpj(
                 bq_project="basedosdados",
             )
         else:
-            # data-only DATE column (partition col in the dbt model) — not
-            # ano/mes, which don't exist in empresas/estabelecimentos/socios
             register_table_materialization_task(
                 dataset_id=dataset_id,
                 table_id=table_id,
                 coverage=PartBdpro(
                     date_column=DateOnly(col="data_referencia"),
-                    date_format=DateFormat.YEAR_MD,
+                    date_format=DateFormat.YEAR_MONTH,
                 ),
                 env="prod",
                 bq_project="basedosdados",
@@ -178,10 +175,10 @@ def _run_rf_cnpj(
                 table_id=table_id,
                 source_max_date=folder_date,
                 env="prod",
-                date_format="%Y-%m",
+                date_format=DateFormat.YEAR_MONTH,
             )
 
-    # estabelecimentos: atualiza diretório de empresas + download p/ GCS
+    # estabelecimentos: atualiza diretório de empresas
     if table_id == "estabelecimentos":
         run_dbt(
             dataset_id="br_bd_diretorios_brasil",
@@ -200,7 +197,7 @@ def _run_rf_cnpj(
                 table_id="empresa",
                 coverage=AllBdpro(
                     date_column=DateOnly(col="data_referencia"),
-                    date_format=DateFormat.YEAR_MD,
+                    date_format=DateFormat.YEAR_MONTH,
                 ),
                 env="prod",
                 bq_project="basedosdados",

pipelines/crawler/rf_cnpj/tasks.py

@@ -54,7 +54,7 @@ def main(
     chunk_size: int = 100000,
     download_chunk_size: int = 15 * 1024 * 1024,
     download_max_retries: int = 5,
-    download_max_parallel: int = 5,
+    download_max_parallel: int = 15,
     download_timeout: int = 5 * 60,
 ) -> Path:
     """

pipelines/datasets/flows.py

@@ -0,0 +1,57 @@
+"""
+Flows for br_rf_cnpj — Prefect 3.
+"""
+
+from prefect import flow
+
+from pipelines.crawler.rf_cnpj.flows import _run_rf_cnpj
+
+
+def _rf_cnpj_flow(table_id: str, cron: str):
+    @flow(
+        name=f"br_rf_cnpj__{table_id}",
+        log_prints=True,
+    )
+    def _flow(
+        dataset_id: str = "br_rf_cnpj",
+        table_id: str = table_id,
+        materialize_after_dump: bool = True,
+        dbt_alias: bool = True,
+        update_metadata: bool = True,
+        target: str = "prod",
+        force_run: bool = False,
+        chunk_size: int = 100000,
+        folder_date: str | None = None,
+        download_chunk_size: int = 15 * 1024 * 1024,
+        download_max_retries: int = 5,
+        download_max_parallel: int = 15,
+        download_timeout: int = 5 * 60,
+    ) -> None:
+        _run_rf_cnpj(
+            dataset_id=dataset_id,
+            table_id=table_id,
+            materialize_after_dump=materialize_after_dump,
+            dbt_alias=dbt_alias,
+            update_metadata=update_metadata,
+            target=target,
+            force_run=force_run,
+            chunk_size=chunk_size,
+            folder_date=folder_date,
+            download_chunk_size=download_chunk_size,
+            download_max_retries=download_max_retries,
+            download_max_parallel=download_max_parallel,
+            download_timeout=download_timeout,
+        )
+
+    # pyrefly: ignore [missing-attribute]
+    _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}]
+    return _flow
+
+
+br_rf_cnpj__dicionario = _rf_cnpj_flow(table_id="dicionario", cron="0 5 * * *")
+br_rf_cnpj__empresas = _rf_cnpj_flow(table_id="empresas", cron="0 6 * * *")
+br_rf_cnpj__socios = _rf_cnpj_flow(table_id="socios", cron="0 7 * * *")
+br_rf_cnpj__simples = _rf_cnpj_flow(table_id="simples", cron="0 8 * * *")
+br_rf_cnpj__estabelecimentos = _rf_cnpj_flow(
+    table_id="estabelecimentos", cron="0 9 * * *"
+)