diff --git a/models/br_bd_diretorios_brasil/br_bd_diretorios_brasil__empresa.sql b/models/br_bd_diretorios_brasil/br_bd_diretorios_brasil__empresa.sql index ae182dc5b7..2dba660a82 100644 --- a/models/br_bd_diretorios_brasil/br_bd_diretorios_brasil__empresa.sql +++ b/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 diff --git a/models/br_rf_cnpj/br_rf_cnpj__socios._legado.sql b/models/br_rf_cnpj/br_rf_cnpj__socios_legado.sql similarity index 100% rename from models/br_rf_cnpj/br_rf_cnpj__socios._legado.sql rename to models/br_rf_cnpj/br_rf_cnpj__socios_legado.sql diff --git a/models/br_tse_eleicoes/code/python/diagnostics/tier2_overrides.json b/models/br_tse_eleicoes/code/python/diagnostics/tier2_overrides.json index ebe1cb5990..8f4d9e23e6 100644 --- a/models/br_tse_eleicoes/code/python/diagnostics/tier2_overrides.json +++ b/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." ] } -} \ No newline at end of file +} diff --git a/models/us_harvard_cbdb/code/clean.py b/models/us_harvard_cbdb/code/clean.py index c1079244f4..51c71d1b8f 100644 --- a/models/us_harvard_cbdb/code/clean.py +++ b/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("~") diff --git a/models/us_harvard_cbdb/code/gen_architecture.py b/models/us_harvard_cbdb/code/gen_architecture.py index 41ab95835d..9c613ea986 100644 --- a/models/us_harvard_cbdb/code/gen_architecture.py +++ b/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") diff --git a/models/us_harvard_cbdb/code/gen_dbt.py b/models/us_harvard_cbdb/code/gen_dbt.py index 3efe28152c..7b12e073e6 100644 --- a/models/us_harvard_cbdb/code/gen_dbt.py +++ b/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( diff --git a/models/us_harvard_cbdb/code/gen_metadata_payload.py b/models/us_harvard_cbdb/code/gen_metadata_payload.py index 01c309ed02..c070775b4c 100644 --- a/models/us_harvard_cbdb/code/gen_metadata_payload.py +++ b/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"], diff --git a/models/us_harvard_cbdb/code/schema_spec.py b/models/us_harvard_cbdb/code/schema_spec.py index c94233691b..51afbefd81 100644 --- a/models/us_harvard_cbdb/code/schema_spec.py +++ b/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( diff --git a/models/us_harvard_cbdb/code/upload.py b/models/us_harvard_cbdb/code/upload.py index c704facf73..d30a9f0a84 100644 --- a/models/us_harvard_cbdb/code/upload.py +++ b/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 diff --git a/pipelines/crawler/rf_cnpj/flows.py b/pipelines/crawler/rf_cnpj/flows.py index 9afb4809f0..3909762bf9 100644 --- a/pipelines/crawler/rf_cnpj/flows.py +++ b/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", diff --git a/pipelines/crawler/rf_cnpj/tasks.py b/pipelines/crawler/rf_cnpj/tasks.py index 7c93552a8b..e0f1f988b5 100644 --- a/pipelines/crawler/rf_cnpj/tasks.py +++ b/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: """ diff --git a/pipelines/datasets/flows.py b/pipelines/datasets/flows.py new file mode 100644 index 0000000000..601e12b20e --- /dev/null +++ b/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 * * *" +)