[Chore] Ajustes CNPJ - #1698
[Chore] Ajustes CNPJ#1698
Conversation
luizavboas
added
test-dev-model
luizavboas
added
the
deploy-flow
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 9
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@pipelines/crawler/rf_cnpj/flows.py`:
- Around line 28-36: Rename _run_rfcnpj to _run_rf_cnpj to match its import in
the dataset flow, and add a Google-style docstring documenting the function’s
purpose, parameters, and return value. Update any local references to the
renamed function while preserving its existing behavior.
In `@pipelines/crawler/rf_cnpj/tasks.py`:
- Around line 30-33: Update the Prefect task decorators in tasks.py, including
both the decorator near the task at the shown location and the one around lines
47-50, to use Prefect 3 keyword arguments: replace max_retries with retries and
retry_delay with retry_delay_seconds while preserving their existing values.
In `@pipelines/crawler/rf_cnpj/utils.py`:
- Around line 322-328: Validate every archive member in the ZipFile handling
block before extraction, ensuring each resolved destination remains within the
extraction directory path; reject unsafe entries such as traversal or absolute
paths, and only call extractall after validation succeeds.
- Around line 356-358: Path.iterdir() already returns full entry paths, so
remove the redundant input_path join in process_csv_estabelecimentos,
process_csv_empresas, process_csv_socios, and process_csv_simples. In
pipelines/crawler/rf_cnpj/utils.py at lines 356-358, 445-447, 505-507, and
565-567, assign each caminho_arquivo_csv directly from nome_arquivo, matching
process_csv_dicionario.
- Around line 111-113: Correct the type annotations on fill_left_zeros: annotate
df as pd.DataFrame instead of datetime.datetime and add an appropriate type hint
for column consistent with its usage. Preserve the existing return annotation
and behavior.
- Around line 90-107: Fix build_paths by preserving the computed Path objects
instead of overwriting input_path and output_path with None, and append table_id
directly to each base path when the corresponding build_input or build_output
flag is enabled. Keep disabled paths represented as None in the returned tuple.
- Around line 867-869: Move os.remove(filepath) into the loop that processes
each matched CSV so every processed file is deleted and empty files collections
do not reference an undefined variable; preserve the existing save/log flow.
Update the containing function’s return annotation from None to the Path type
represented by save_path.
In `@pipelines/datasets/br_rf_cnpj/__init__.py`:
- Line 1: Populate the empty pipelines/datasets/br_rf_cnpj/__init__.py by
importing the flow definitions from the sibling flows.py module, following the
package’s existing import pattern. Keep flow implementation in flows.py and
expose the intended flows through the parent package.
In `@pipelines/datasets/br_rf_cnpj/flows.py`:
- Around line 5-44: Rewrite _rf_cnpj_flow and its generated dataset flows to use
Prefect 0.15.x APIs exclusively: replace the `@flow` decorator and
deploy_schedules assignment with a Flow context manager, Parameter definitions,
task-wrapped execution, and a prefect.schedules.Schedule-based schedule.
Preserve the existing flow names, cron expressions, parameters, defaults, and
_run_rf_cnpj invocation behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: c12a39d0-3ed1-4f8a-8dcf-7e098a9463f2
📒 Files selected for processing (7)
pipelines/crawler/rf_cnpj/__init__.pypipelines/crawler/rf_cnpj/constants.pypipelines/crawler/rf_cnpj/flows.pypipelines/crawler/rf_cnpj/tasks.pypipelines/crawler/rf_cnpj/utils.pypipelines/datasets/br_rf_cnpj/__init__.pypipelines/datasets/br_rf_cnpj/flows.py
luizavboas
added
check-metadata
There was a problem hiding this comment.
Actionable comments posted: 6
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
pipelines/crawler/rf_cnpj/tasks.py (1)
44-60: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winStale
folder_dateentry in docstring.The
Argssection documents afolder_dateparameter that doesn't exist in the signature (onlytables,max_folder_date,max_last_modified_date). As per coding guidelines, "Add type hints and docstrings for Python functions following Google Style".📝 Proposed fix
Args: tables (list): A list of tables to be processed. - folder_date (datetime): Most recent database release extracted from API max_folder_date (datetime): CNPJs max folder date max_last_modified_date (datetime): CNPJs max last modified date🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pipelines/crawler/rf_cnpj/tasks.py` around lines 44 - 60, Update the Args section of main’s docstring to remove the stale folder_date entry and document only tables, max_folder_date, and max_last_modified_date with descriptions matching their actual parameters.Source: Coding guidelines
🧹 Nitpick comments (2)
models/br_rf_cnpj/br_rf_cnpj__empresas.sql (1)
21-21: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winPrefer
NUMERIC/BIGNUMERICoverFLOAT64forcapital_social.Monetary fields cast to
FLOAT64are subject to floating-point precision drift. Since this model is new, switching toNUMERICnow avoids a costlier migration later.♻️ Proposed fix
- safe_cast(capital_social as float64) capital_social, + safe_cast(capital_social as numeric) capital_social,🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@models/br_rf_cnpj/br_rf_cnpj__empresas.sql` at line 21, Update the capital_social expression in the br_rf_cnpj__empresas model to safely cast to NUMERIC or BIGNUMERIC instead of FLOAT64, preserving the existing alias and monetary value handling.pipelines/crawler/rf_cnpj/tasks.py (1)
76-96: 🚀 Performance & Scalability | 🔵 Trivial | 🏗️ Heavy liftSequential
asyncio.run()per segment may forfeit download concurrency.Each of the 10 segments calls
asyncio.run(download_unzip_csv(...))independently, spinning up and tearing down a new event loop per segment rather than awaiting all 10 downloads concurrently in one loop. Ifdownload_unzip_csv/the "concurrent download helpers" from the configuration layer are designed for parallel fetches, this pattern serializes them and slows the pipeline. Please confirmdownload_unzip_csv's intended concurrency model inpipelines/crawler/rf_cnpj/utils.py.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pipelines/crawler/rf_cnpj/tasks.py` around lines 76 - 96, Update the segmented-table flow around download_unzip_csv so all required segment downloads are scheduled and awaited within one shared event loop rather than calling asyncio.run once per segment. Confirm the intended concurrency behavior in download_unzip_csv and reuse the existing concurrent-download helper if applicable, while preserving the arquivos_baixados checks and per-segment processing for Estabelecimentos, Socios, and Empresas.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@models/br_rf_cnpj/br_rf_cnpj__dicionario.sql`:
- Around line 10-12: Update the regexp_replace expression in the valor
transformation to use the pattern ^0+, removing only leading zeros while
preserving the remaining value. Keep the existing validate_null_cols("valor")
input and alias unchanged.
- Around line 56-62: Update the dicionario_not_found invocation for the
estabelecimentos entry by replacing the empty nome_coluna value with
cnae_fiscal_principal, so the CNAE codes 6202100 and 4761000 match
custom_dictionary_coverage lookups.
In `@models/br_rf_cnpj/br_rf_cnpj__estabelecimentos.sql`:
- Line 23: The situacao_cadastral conversion still uses an unsafe inner cast
that can abort on invalid values. Update the situacao_cadastral expression in
the establishments model to use nested safe_cast calls, and apply the same
pattern to the id_pais expression so invalid inputs yield NULL instead of
failing.
In `@models/br_rf_cnpj/br_rf_cnpj__socios.sql`:
- Line 18: Update the casts in the model’s qualificacao, id_pais, and
qualificacao_representante_legal expressions to apply SAFE_CAST directly to the
raw values, removing the inner plain CAST so invalid input yields NULL instead
of failing.
In `@models/br_rf_cnpj/schema.yml`:
- Around line 44-64: Update the br_rf_cnpj__socios schema entry to add the
model-level dbt_utils.unique_combination_of_columns test using the same key
configuration as its sibling models, and add is_row_count_increasing to the data
column’s tests. Preserve the existing dictionary coverage and cnpj_basico
not-null tests.
- Around line 93-95: Update the test configuration for br_rf_cnpj__simples to
remove the invalid __most_recent_date__ scope or replace it with a valid keyword
referencing an existing column, ensuring the not_null test runs successfully.
---
Outside diff comments:
In `@pipelines/crawler/rf_cnpj/tasks.py`:
- Around line 44-60: Update the Args section of main’s docstring to remove the
stale folder_date entry and document only tables, max_folder_date, and
max_last_modified_date with descriptions matching their actual parameters.
---
Nitpick comments:
In `@models/br_rf_cnpj/br_rf_cnpj__empresas.sql`:
- Line 21: Update the capital_social expression in the br_rf_cnpj__empresas
model to safely cast to NUMERIC or BIGNUMERIC instead of FLOAT64, preserving the
existing alias and monetary value handling.
In `@pipelines/crawler/rf_cnpj/tasks.py`:
- Around line 76-96: Update the segmented-table flow around download_unzip_csv
so all required segment downloads are scheduled and awaited within one shared
event loop rather than calling asyncio.run once per segment. Confirm the
intended concurrency behavior in download_unzip_csv and reuse the existing
concurrent-download helper if applicable, while preserving the arquivos_baixados
checks and per-segment processing for Estabelecimentos, Socios, and Empresas.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 4cd167ec-beaa-4e3b-b13e-e30d0fa9eb50
📒 Files selected for processing (8)
models/br_rf_cnpj/br_rf_cnpj__dicionario.sqlmodels/br_rf_cnpj/br_rf_cnpj__empresas.sqlmodels/br_rf_cnpj/br_rf_cnpj__estabelecimentos.sqlmodels/br_rf_cnpj/br_rf_cnpj__simples.sqlmodels/br_rf_cnpj/br_rf_cnpj__socios.sqlmodels/br_rf_cnpj/schema.ymlpipelines/crawler/rf_cnpj/flows.pypipelines/crawler/rf_cnpj/tasks.py
🚧 Files skipped from review as they are similar to previous changes (1)
- pipelines/crawler/rf_cnpj/flows.py
…pipelines into chore/ajustes_cnpj
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
…s into chore/ajustes_cnpj
…cionário, já que não há dataset em prod ainda
…ownload para evitar erro OOMKilled
…cionário, já que não há dataset em prod ainda
…:data_url, tasks.py:get_data_source_max_date, e flows.py
|
Note GitHub couldn't provide a complete incremental comparison for this pull request, so CodeRabbit is performing a full review instead. This review may take a little longer. |
There was a problem hiding this comment.
Actionable comments posted: 13
♻️ Duplicate comments (2)
pipelines/crawler/rf_cnpj/utils.py (2)
527-527: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
open(save_path, "wb")still truncates inside the input-file loop.The header duplication is fixed by
header=(chunk_idx == 0). The truncation is not. Theopen(..., "wb")call stays inside thefor filepath in Path(input_path).iterdir()loop. If the input directory holds more than one matching CSV, each new file discards the output of the previous file. Open the output once before the loop, or append after the first write.
pipelines/crawler/rf_cnpj/utils.py#L527: move the output handle outside the loop inprocess_csv_empresasand track the header flag across files.pipelines/crawler/rf_cnpj/utils.py#L595: apply the same change inprocess_csv_socios.pipelines/crawler/rf_cnpj/utils.py#L665: apply the same change inprocess_csv_simples.Also applies to: 595-595, 665-665
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pipelines/crawler/rf_cnpj/utils.py` at line 527, Update process_csv_empresas, process_csv_socios, and process_csv_simples so each output file is opened once outside the input-file iteration, preventing later CSVs from truncating earlier results. Track whether the header has already been written across all input files, while preserving header output only for the first written chunk.
973-975: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
os.remove(filepath)is still outside the loop.
filepathis the loop variable from line 904. Two defects remain. First, iffilesis empty, line 974 raisesNameError. Second, only the last iterated file is deleted, so unmatched or earlier files stay in the input directory. Move the delete into theif table_name in ...branch.🐛 Proposed fix
header=not save_path.exists(), # Write header only if file doesn't exist ) + os.remove(filepath) log(f"Arquivo {table_name} salvo") - os.remove(filepath) return save_path🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pipelines/crawler/rf_cnpj/utils.py` around lines 973 - 975, Move the os.remove(filepath) call into the loop’s if table_name-inclusion branch, alongside the save and log operations, so each matched file is deleted immediately; ensure no deletion occurs when files is empty or when a file does not match.
🧹 Nitpick comments (6)
pipelines/crawler/rf_cnpj/utils.py (2)
96-99: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueComplete the docstring and add the return annotation.
get_table_fileshas noArgsorReturnssection and no return type. The return islist[tuple[str, str]]. As per coding guidelines, "add Google-Style type hints and docstrings for Python functions".🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pipelines/crawler/rf_cnpj/utils.py` around lines 96 - 99, Update get_table_files with a return annotation of list[tuple[str, str]] and complete its docstring using Google-style Args and Returns sections describing table_name, url_base, and the returned file/link pairs.Source: Coding guidelines
398-399: 🔒 Security & Privacy | 🔵 Trivial | 💤 Low valueConsider validating archive members before extraction.
z.extractall(path)trusts the member names. A crafted entry with../writes outsidepath. The source is the Receita Federal server, so the current risk is low. Add a member-path check to keep the guarantee independent of the remote source.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pipelines/crawler/rf_cnpj/utils.py` around lines 398 - 399, Validate every archive member in the ZipFile extraction block before calling extractall, resolving each member path beneath the destination path and rejecting entries that escape it, including traversal through ../ segments. Preserve extraction for safe members and do not rely on the remote source being trusted.Source: Linters/SAST tools
pipelines/crawler/rf_cnpj/flows.py (1)
133-140: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
folder_dateis neverNoneat this point.Line 49 reassigns
folder_datefromget_data_source_max_date, which always returns a string. The guard at line 133 is therefore always true. Remove it, or move the check to the value returned byget_data_source_max_date.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pipelines/crawler/rf_cnpj/flows.py` around lines 133 - 140, Remove the redundant folder_date is not None guard around commit_source_update_task, since folder_date is reassigned from get_data_source_max_date and is always a string at this point. Keep the existing task arguments and invocation unchanged.pipelines/crawler/rf_cnpj/tasks.py (1)
60-71: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueComplete the docstring.
The
Argssection omitsdownload_chunk_size,download_max_retries,download_max_parallel, anddownload_timeout. TheReturnssection declaresstr, but the annotation isPath. As per coding guidelines, "add Google-Style type hints and docstrings for Python functions".🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pipelines/crawler/rf_cnpj/tasks.py` around lines 60 - 71, Complete the function docstring by documenting download_chunk_size, download_max_retries, download_max_parallel, and download_timeout in the Args section with Google-Style type hints, and update the Returns section to declare Path instead of str to match the function annotation.Source: Coding guidelines
pipelines/crawler/rf_cnpj/constants.py (2)
366-369: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRename these members to UPPER_CASE.
Every other member in
constantsuses UPPER_CASE. These four use lowercase, which breaks the convention of the class.♻️ Proposed rename
- default_chunk_size = 20 * 1024 * 1024 # 20MB - default_max_retries = 32 - default_max_parallel = 16 - default_timeout = 1 * 60 * 1000 # 1 minute + DEFAULT_CHUNK_SIZE = 20 * 1024 * 1024 # 20MB + DEFAULT_MAX_RETRIES = 32 + DEFAULT_MAX_PARALLEL = 16 + DEFAULT_TIMEOUT = 1 * 60 * 1000 # 1 minuteUpdate the references in
pipelines/crawler/rf_cnpj/utils.pyaccordingly.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pipelines/crawler/rf_cnpj/constants.py` around lines 366 - 369, Rename default_chunk_size, default_max_retries, default_max_parallel, and default_timeout to UPPER_CASE names in the constants class, then update all corresponding references in utils.py while preserving their values and behavior.
36-214: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider guarding the shared mutable configuration.
constants.TABLE_CONFIGS.valuereturns the same dict object to every caller. A consumer that mutates a nested entry (for example, appending torelationships) changes the configuration for the whole process. Ruff also reports RUF012 for this member and the other mutable members in the file.If mutation by consumers is a real risk, return copies at the call sites or wrap the value with
types.MappingProxyType. Otherwise, add a Ruff suppression for the Enum pattern so the file stays clean underuv run pre-commit run --all-files.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pipelines/crawler/rf_cnpj/constants.py` around lines 36 - 214, Protect the shared TABLE_CONFIGS value from consumer mutation by returning a deep copy at its access points, including nested relationships and chaves_valores data, so each caller receives independent configuration. Also address Ruff RUF012 for TABLE_CONFIGS and the other mutable Enum members with the narrowest appropriate suppression if the Enum pattern remains.Source: Linters/SAST tools
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@models/br_rf_cnpj/br_rf_cnpj__estabelecimentos.sql`:
- Around line 26-32: Replace the inner CAST(... AS INT64) with SAFE_CAST(... AS
INT64) for numeric-code normalization in
models/br_rf_cnpj/br_rf_cnpj__estabelecimentos.sql at lines 26-32 and 73-81,
covering situacao_cadastral and id_pais in both branches. Apply the same change
in models/br_rf_cnpj/br_rf_cnpj__socios.sql at lines 21-28 and 47-54 for
qualificacao, id_pais, and qualificacao_representante_legal, preserving the
outer casts and NULL behavior for malformed values.
In `@models/br_rf_cnpj/br_rf_cnpj__simples.sql`:
- Around line 5-20: Update the model query around the staging source
`br_rf_cnpj_staging.simples` to preserve pre-migration `data_referencia`
history, either by unioning a legacy source with an exclusive migration boundary
or by documenting that the staging source contains all historical partitions.
Ensure the resulting table does not drop dates before the migration boundary
during rebuilds.
In `@pipelines/crawler/rf_cnpj/constants.py`:
- Line 247: Update the URL constant near URL to remove the hardcoded WebDAV
share token and read the endpoint or token from an environment variable or Vault
instead, using the existing value only as an explicitly documented fallback if
required. Add the necessary os import and preserve the resulting URL format.
In `@pipelines/crawler/rf_cnpj/flows.py`:
- Around line 42-45: Update _run_rf_cnpj to invoke rename_flow_run_dataset_table
through its async task API and await the returned coroutine, removing the
pyrefly unused-coroutine suppression. Ensure the surrounding function supports
async execution so the rename completes before the flow continues.
- Around line 27-41: Add a Google-style docstring to _run_rf_cnpj documenting
its purpose, return type, and every parameter, including the six
download-related and folder_date options. Match the function’s existing type
hints and accurately describe defaults and behavior without changing the
implementation.
In `@pipelines/crawler/rf_cnpj/tasks.py`:
- Around line 49-59: The main task’s retries can append duplicate rows because
its output-writing and cleanup operations are not idempotent. Update main to
clear the output directory before processing begins so every retry starts from a
clean state, or remove the task-level retries and apply retry behavior only to
the download step; preserve the existing processing flow.
- Around line 159-161: Update the non-segmented download call to
download_unzip_csv in the asyncio.run path so it passes the configured
download_chunk_size, download_max_retries, download_max_parallel, and
download_timeout arguments, matching the segmented branch while preserving the
existing URL and input_path arguments.
In `@pipelines/crawler/rf_cnpj/utils.py`:
- Around line 226-241: Update the range-download flow around the HEAD request
and each chunk worker to require server byte-range support before
pre-allocation: validate the HEAD response’s Accept-Ranges header before calling
chunk_range, and validate every chunk response returns HTTP 206 Partial Content
before writing it at its offset. Abort the download with an explicit error when
either check fails, preserving the existing chunked-write behavior only for
valid range responses.
- Line 163: Update the fill_left_zeros function signature to annotate the
documented column-name parameter as column: str, while preserving its existing
behavior and return annotation.
- Around line 397-405: Update the BadZipFile handling in the ZIP extraction flow
to re-raise the exception after logging, instead of swallowing it, so the caller
and Prefect retry mechanism receive the failure and do not process or upload
missing data. Preserve the existing cleanup of save_path.
- Around line 73-93: Update the branch containing the folder_dates lookup so
last_modified_date is always assigned before the final log and return, and stop
swallowing lookup failures. Replace the broad except Exception in the
surrounding function with explicit failure propagation or an intentional
fallback consistent with the function’s contract, preserving the existing
successful date parsing behavior and exposing the original error.
- Around line 718-728: Update the dictionary query construction around
bd.read_sql so the project is passed as a parameter instead of hardcoding
basedosdados-dev. Resolve that project from the flow’s target value, using the
production project when target="prod" and preserving the development project
otherwise, then interpolate the resolved project into the staging table
reference.
In `@pipelines/datasets/br_rf_cnpj/flows.py`:
- Around line 10-29: Add a return annotation to _rf_cnpj_flow and document it
with a Google-Style docstring; also add a Google-Style docstring to the nested
_flow function describing its parameters and return behavior. Preserve the
existing flow configuration and signature semantics.
---
Duplicate comments:
In `@pipelines/crawler/rf_cnpj/utils.py`:
- Line 527: Update process_csv_empresas, process_csv_socios, and
process_csv_simples so each output file is opened once outside the input-file
iteration, preventing later CSVs from truncating earlier results. Track whether
the header has already been written across all input files, while preserving
header output only for the first written chunk.
- Around line 973-975: Move the os.remove(filepath) call into the loop’s if
table_name-inclusion branch, alongside the save and log operations, so each
matched file is deleted immediately; ensure no deletion occurs when files is
empty or when a file does not match.
---
Nitpick comments:
In `@pipelines/crawler/rf_cnpj/constants.py`:
- Around line 366-369: Rename default_chunk_size, default_max_retries,
default_max_parallel, and default_timeout to UPPER_CASE names in the constants
class, then update all corresponding references in utils.py while preserving
their values and behavior.
- Around line 36-214: Protect the shared TABLE_CONFIGS value from consumer
mutation by returning a deep copy at its access points, including nested
relationships and chaves_valores data, so each caller receives independent
configuration. Also address Ruff RUF012 for TABLE_CONFIGS and the other mutable
Enum members with the narrowest appropriate suppression if the Enum pattern
remains.
In `@pipelines/crawler/rf_cnpj/flows.py`:
- Around line 133-140: Remove the redundant folder_date is not None guard around
commit_source_update_task, since folder_date is reassigned from
get_data_source_max_date and is always a string at this point. Keep the existing
task arguments and invocation unchanged.
In `@pipelines/crawler/rf_cnpj/tasks.py`:
- Around line 60-71: Complete the function docstring by documenting
download_chunk_size, download_max_retries, download_max_parallel, and
download_timeout in the Args section with Google-Style type hints, and update
the Returns section to declare Path instead of str to match the function
annotation.
In `@pipelines/crawler/rf_cnpj/utils.py`:
- Around line 96-99: Update get_table_files with a return annotation of
list[tuple[str, str]] and complete its docstring using Google-style Args and
Returns sections describing table_name, url_base, and the returned file/link
pairs.
- Around line 398-399: Validate every archive member in the ZipFile extraction
block before calling extractall, resolving each member path beneath the
destination path and rejecting entries that escape it, including traversal
through ../ segments. Preserve extraction for safe members and do not rely on
the remote source being trusted.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 399240b6-17e5-4460-b9cf-72c56e4f2d9c
📒 Files selected for processing (19)
dbt_project.ymlmacros/custom_get_where_subquery.sqlmodels/br_rf_cnpj/br_rf_cnpj__dicionario.sqlmodels/br_rf_cnpj/br_rf_cnpj__empresas.sqlmodels/br_rf_cnpj/br_rf_cnpj__empresas_legado.sqlmodels/br_rf_cnpj/br_rf_cnpj__estabelecimentos.sqlmodels/br_rf_cnpj/br_rf_cnpj__estabelecimentos_legado.sqlmodels/br_rf_cnpj/br_rf_cnpj__simples.sqlmodels/br_rf_cnpj/br_rf_cnpj__socios._legado.sqlmodels/br_rf_cnpj/br_rf_cnpj__socios.sqlmodels/br_rf_cnpj/schema.ymlpackage-lock.ymlpipelines/crawler/rf_cnpj/__init__.pypipelines/crawler/rf_cnpj/constants.pypipelines/crawler/rf_cnpj/flows.pypipelines/crawler/rf_cnpj/tasks.pypipelines/crawler/rf_cnpj/utils.pypipelines/datasets/br_rf_cnpj/__init__.pypipelines/datasets/br_rf_cnpj/flows.py
🚧 Files skipped from review as they are similar to previous changes (3)
- pipelines/datasets/br_rf_cnpj/init.py
- models/br_rf_cnpj/br_rf_cnpj__dicionario.sql
- models/br_rf_cnpj/schema.yml
| safe_cast(cast(situacao_cadastral as int64) as string) situacao_cadastral, | ||
| safe_cast(data_situacao_cadastral as date) data_situacao_cadastral, | ||
| safe_cast( | ||
| regexp_replace(motivo_situacao_cadastral, '^0', '') as string | ||
| ) motivo_situacao_cadastral, | ||
| safe_cast(nome_cidade_exterior as string) nome_cidade_exterior, | ||
| safe_cast(cast(id_pais as int64) as string) id_pais, |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
# Find all nested SAFE_CAST(CAST(...)) expressions in the CNPJ models.
rg -n -U -P 'safe_cast\s*\(\s*cast\s*\(' models/br_rf_cnpj
# Inspect raw CNPJ transformations for handling of blank numeric-code fields.
rg -n -C 4 --glob '*.py' \
'qualificacao_representante_legal|qualificacao|situacao_cadastral|id_pais' \
pipelines/crawler/rf_cnpjRepository: basedosdados/pipelines
Length of output: 12888
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
echo "== model excerpts =="
sed -n '1,100p' models/br_rf_cnpj/br_rf_cnpj__estabelecimentos.sql
sed -n '1,85p' models/br_rf_cnpj/br_rf_cnpj__socios.sql
echo "== source preparation excerpts around rf_cnpj columns =="
rg -n -C 7 'qualificacao|id_pais|situacao_cadastral|CNPJ|RF_CNPJ|colunas_estabelecimentos|COLUNAS_ESTABELECIMENTOS|COLUNAS_SOCIOS' pipelines/crawler/rf_cnpj -g '*.py'
echo "== deterministic SQL semantics probe if duckdb is available =="
python3 - <<'PY'
from pathlib import Path
p = Path("/tmp/probe_cast.sql")
for val in ("", None, "abc", "1"):
expr = f"SELECT {repr(val) if val else 'NULL'} AS val, CAST({repr(val) if val else 'NULL'} AS INT64) AS safe_cast_of_cast"
txt = p.read_text() if p.exists() else ""
txt += expr + "\n"
p.write_text(txt)
print("sql generated for values: '', NULL, 'abc', '1'\n" + p.read_text())
PYRepository: basedosdados/pipelines
Length of output: 31166
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
echo "== rf_cnpj utils relevant source =="
sed -n '430,510p' pipelines/crawler/rf_cnpj/utils.py
sed -n '780,830p' pipelines/crawler/rf_cnpj/utils.py
echo "== deterministic typecast probe if duckdb is available =="
if command -v duckdb >/dev/null 2>&1; then
cat >/tmp/cast_probe.sql <<'SQL'
select
'NULL' as value, CAST(NULL AS INT64) as cast_result, CAST(NULL AS STRING) as outer_safe_result
union all select 'empty', CAST('' AS INT64), CAST(CAST('' AS INT64) AS STRING)
union all select 'malformed', CAST('abc' AS INT64), CAST(CAST('abc' AS INT64) AS STRING)
union all select 'valid', CAST('1' AS INT64), CAST(CAST('1' AS INT64) AS STRING);
SQL
duckdb /tmp/cast_probe.sql
else
echo "duckdb not available"
fiRepository: basedosdados/pipelines
Length of output: 5146
Make numeric-code normalization fail-safe.
The inner CAST(... AS INT64) must not run before the outer SAFE_CAST, otherwise blank or malformed numeric-code values fail the model instead of returning NULL.
Replace the inner casts with SAFE_CAST(... AS INT64) at every affected site:
models/br_rf_cnpj/br_rf_cnpj__estabelecimentos.sql:situacao_cadastralandid_pais, including the legacy branch.models/br_rf_cnpj/br_rf_cnpj__socios.sql:qualificacao,id_pais, andqualificacao_representante_legal, including the legacy branch.
📍 Affects 2 files
models/br_rf_cnpj/br_rf_cnpj__estabelecimentos.sql#L26-L32(this comment)models/br_rf_cnpj/br_rf_cnpj__estabelecimentos.sql#L73-L81models/br_rf_cnpj/br_rf_cnpj__socios.sql#L21-L28models/br_rf_cnpj/br_rf_cnpj__socios.sql#L47-L54
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@models/br_rf_cnpj/br_rf_cnpj__estabelecimentos.sql` around lines 26 - 32,
Replace the inner CAST(... AS INT64) with SAFE_CAST(... AS INT64) for
numeric-code normalization in models/br_rf_cnpj/br_rf_cnpj__estabelecimentos.sql
at lines 26-32 and 73-81, covering situacao_cadastral and id_pais in both
branches. Apply the same change in models/br_rf_cnpj/br_rf_cnpj__socios.sql at
lines 21-28 and 47-54 for qualificacao, id_pais, and
qualificacao_representante_legal, preserving the outer casts and NULL behavior
for malformed values.
| "TO", | ||
| ] | ||
|
|
||
| URL = "https://arquivos.receitafederal.gov.br/public.php/dav/files/gn672Ad4CF8N6TK/Dados/Cadastros/CNPJ/" |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Move the share token out of the source code.
The URL embeds the WebDAV share token gn672Ad4CF8N6TK. That token grants access to the endpoint, so it is a credential. Receita Federal rotates this share link, and each rotation then requires a code change and a redeploy.
Read the URL (or the token part) from an environment variable or from Vault, with the current value as a documented default only if the team accepts that.
As per coding guidelines: "Never hardcode credentials or secrets; use environment variables or Vault."
🔒 Proposed change
- URL = "https://arquivos.receitafederal.gov.br/public.php/dav/files/gn672Ad4CF8N6TK/Dados/Cadastros/CNPJ/"
+ URL = os.environ["RF_CNPJ_WEBDAV_URL"]Add the import at the top of the file:
import os📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| URL = "https://arquivos.receitafederal.gov.br/public.php/dav/files/gn672Ad4CF8N6TK/Dados/Cadastros/CNPJ/" | |
| URL = os.environ["RF_CNPJ_WEBDAV_URL"] |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pipelines/crawler/rf_cnpj/constants.py` at line 247, Update the URL constant
near URL to remove the hardcoded WebDAV share token and read the endpoint or
token from an environment variable or Vault instead, using the existing value
only as an explicitly documented fallback if required. Add the necessary os
import and preserve the resulting URL format.
Source: Coding guidelines
| @task(retries=3, retry_delay_seconds=30) | ||
| def main( | ||
| tables: list[str], | ||
| folder_date: str, | ||
| last_modified_date: datetime.date, | ||
| chunk_size: int = 100000, | ||
| download_chunk_size: int = 15 * 1024 * 1024, | ||
| download_max_retries: int = 5, | ||
| download_max_parallel: int = 5, | ||
| download_timeout: int = 5 * 60, | ||
| ) -> Path: |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
retries=3 on main is not safe, because the task is not idempotent.
main downloads every file, writes partitioned CSVs, and deletes the input files. The output writers append when the target file already exists. See process_csv_estabelecimentos in pipelines/crawler/rf_cnpj/utils.py lines 488-495, which uses mode = "a" if particao_file_path.exists() else "w". If the task fails midway and Prefect retries it, the retry appends the same rows to the partitions that the first attempt already wrote. The uploaded partition then holds duplicate records.
Clear the output directory at the start of main, or move the retry to the download step only.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pipelines/crawler/rf_cnpj/tasks.py` around lines 49 - 59, The main task’s
retries can append duplicate rows because its output-writing and cleanup
operations are not idempotent. Update main to clear the output directory before
processing begins so every retry starts from a clean state, or remove the
task-level retries and apply retry behavior only to the download step; preserve
the existing processing flow.
| # pyrefly: ignore [bad-argument-type] | ||
| # pyrefly: ignore [unbound-name] | ||
| asyncio.run(download_unzip_csv(url_download, input_path)) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
The non-segmented download ignores the configured download settings.
Line 161 calls download_unzip_csv with only the URL and the path. The download_chunk_size, download_max_retries, download_max_parallel, and download_timeout parameters are dropped, so simples and the dictionary files always use the defaults. The segmented branch at lines 98-109 passes them. Pass the same arguments in both branches.
♻️ Proposed fix
- asyncio.run(download_unzip_csv(url_download, input_path))
+ asyncio.run(
+ download_unzip_csv(
+ url_download,
+ input_path,
+ chunk_size=download_chunk_size,
+ max_retries=download_max_retries,
+ max_parallel=download_max_parallel,
+ timeout=download_timeout,
+ )
+ )📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| # pyrefly: ignore [bad-argument-type] | |
| # pyrefly: ignore [unbound-name] | |
| asyncio.run(download_unzip_csv(url_download, input_path)) | |
| # pyrefly: ignore [bad-argument-type] | |
| # pyrefly: ignore [unbound-name] | |
| asyncio.run( | |
| download_unzip_csv( | |
| url_download, | |
| input_path, | |
| chunk_size=download_chunk_size, | |
| max_retries=download_max_retries, | |
| max_parallel=download_max_parallel, | |
| timeout=download_timeout, | |
| ) | |
| ) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pipelines/crawler/rf_cnpj/tasks.py` around lines 159 - 161, Update the
non-segmented download call to download_unzip_csv in the asyncio.run path so it
passes the configured download_chunk_size, download_max_retries,
download_max_parallel, and download_timeout arguments, matching the segmented
branch while preserving the existing URL and input_path arguments.
| else: | ||
| try: | ||
| index = folder_dates.index( | ||
| next(item for item in folder_dates if folder_date in str(item)) | ||
| ) | ||
| last_modified_date = datetime.datetime.strptime( | ||
| last_modified_dates[index].find("d:getlastmodified").text, | ||
| "%a, %d %b %Y %H:%M:%S GMT", | ||
| ).date() | ||
| except Exception as e: | ||
| log(e) | ||
|
|
||
| log( | ||
| f"A data extraida da API da Receita Federal que será utilizada para comparar com os metadados da BD: {folder_date}" | ||
| ) | ||
| log( | ||
| # pyrefly: ignore [unbound-name] | ||
| f"A data máxima extraida da API da Receita Federal que será utilizada para gerar partições no Storage: {last_modified_date}" | ||
| ) | ||
|
|
||
| return folder_date, last_modified_date |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
last_modified_date can be unbound and the return value is then a crash.
Line 82 catches every exception and only logs it. If the lookup at lines 75-81 fails, last_modified_date is never assigned. Line 90 and line 93 then raise UnboundLocalError, and the original cause is hidden. Ruff also flags the blind except Exception (BLE001).
Fail fast, or assign an explicit fallback before the try block.
🐛 Proposed fix
else:
try:
index = folder_dates.index(
next(item for item in folder_dates if folder_date in str(item))
)
last_modified_date = datetime.datetime.strptime(
last_modified_dates[index].find("d:getlastmodified").text,
"%a, %d %b %Y %H:%M:%S GMT",
).date()
- except Exception as e:
- log(e)
+ except (StopIteration, ValueError, AttributeError) as e:
+ log(f"Falha ao extrair a data de modificação de {folder_date}: {e}")
+ raise📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| else: | |
| try: | |
| index = folder_dates.index( | |
| next(item for item in folder_dates if folder_date in str(item)) | |
| ) | |
| last_modified_date = datetime.datetime.strptime( | |
| last_modified_dates[index].find("d:getlastmodified").text, | |
| "%a, %d %b %Y %H:%M:%S GMT", | |
| ).date() | |
| except Exception as e: | |
| log(e) | |
| log( | |
| f"A data extraida da API da Receita Federal que será utilizada para comparar com os metadados da BD: {folder_date}" | |
| ) | |
| log( | |
| # pyrefly: ignore [unbound-name] | |
| f"A data máxima extraida da API da Receita Federal que será utilizada para gerar partições no Storage: {last_modified_date}" | |
| ) | |
| return folder_date, last_modified_date | |
| else: | |
| try: | |
| index = folder_dates.index( | |
| next(item for item in folder_dates if folder_date in str(item)) | |
| ) | |
| last_modified_date = datetime.datetime.strptime( | |
| last_modified_dates[index].find("d:getlastmodified").text, | |
| "%a, %d %b %Y %H:%M:%S GMT", | |
| ).date() | |
| except (StopIteration, ValueError, AttributeError) as e: | |
| log(f"Falha ao extrair a data de modificação de {folder_date}: {e}") | |
| raise | |
| log( | |
| f"A data extraida da API da Receita Federal que será utilizada para comparar com os metadados da BD: {folder_date}" | |
| ) | |
| log( | |
| # pyrefly: ignore [unbound-name] | |
| f"A data máxima extraida da API da Receita Federal que será utilizada para gerar partições no Storage: {last_modified_date}" | |
| ) | |
| return folder_date, last_modified_date |
🧰 Tools
🪛 Ruff (0.16.1)
[warning] 82-82: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pipelines/crawler/rf_cnpj/utils.py` around lines 73 - 93, Update the branch
containing the folder_dates lookup so last_modified_date is always assigned
before the final log and return, and stop swallowing lookup failures. Replace
the broad except Exception in the surrounding function with explicit failure
propagation or an intentional fallback consistent with the function’s contract,
preserving the existing successful date parsing behavior and exposing the
original error.
Source: Linters/SAST tools
|
|
||
|
|
||
| # ! Adiciona zero a esquerda nas colunas | ||
| def fill_left_zeros(df: pd.DataFrame, column, num_digits: int) -> pd.DataFrame: |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Annotate column.
column has no type hint. The docstring documents it as a column name. Add column: str. As per coding guidelines, "add Google-Style type hints and docstrings for Python functions".
✏️ Proposed fix
-def fill_left_zeros(df: pd.DataFrame, column, num_digits: int) -> pd.DataFrame:
+def fill_left_zeros(
+ df: pd.DataFrame, column: str, num_digits: int
+) -> pd.DataFrame:🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pipelines/crawler/rf_cnpj/utils.py` at line 163, Update the fill_left_zeros
function signature to annotate the documented column-name parameter as column:
str, while preserving its existing behavior and return annotation.
Source: Coding guidelines
| request_head = await client.head(url, timeout=timeout) | ||
| request_head.raise_for_status() | ||
| log(request_head.headers["content-length"]) | ||
| content_length = int(request_head.headers["content-length"]) | ||
| chunk_ranges = chunk_range(content_length, chunk_size) | ||
| total_chunks = len(chunk_ranges) | ||
|
|
||
| log( | ||
| f"Baixando {url} com {content_length} bytes ({content_length / 1e6:.2f} MB). " | ||
| f"Cada chunk terá tamanho de {chunk_size} bytes ({chunk_size / 1e6:.2f} MB). " | ||
| f"Serão feitos {max_parallel} downloads paralelos por vez, com um total de {total_chunks} chunks." | ||
| ) | ||
|
|
||
| # Pre-allocate so each chunk can seek to its own offset independently | ||
| with open(save_path, "wb") as fd: | ||
| fd.truncate(content_length) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Validate that the server honors byte-range requests.
The download pre-allocates the file and writes each chunk at its own offset. This is correct only if the server answers with 206 Partial Content. If the server ignores the Range header and returns 200 with the full body, every worker writes the complete file at its own offset. The result is a silently corrupted ZIP, and the corruption is then masked by the BadZipFile handler at line 401.
Check the status code of each chunk response, and check Accept-Ranges on the HEAD response before you split the download.
🛡️ Proposed guard
response = await client.get(
url, headers=headers, timeout=timeout
)
response.raise_for_status()
+ if response.status_code != 206:
+ raise HTTPError(
+ "Servidor não respondeu 206 para o range "
+ f"{chunk_range[0]}-{chunk_range[1]}"
+ )Also applies to: 342-349
🧰 Tools
🪛 ast-grep (0.45.0)
[warning] 239-239: File path is request-/variable-derived; validate and normalize to prevent path traversal.
Context: open(save_path, "wb")
Note: [CWE-22] Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal').
(open-filename-from-request)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pipelines/crawler/rf_cnpj/utils.py` around lines 226 - 241, Update the
range-download flow around the HEAD request and each chunk worker to require
server byte-range support before pre-allocation: validate the HEAD response’s
Accept-Ranges header before calling chunk_range, and validate every chunk
response returns HTTP 206 Partial Content before writing it at its offset. Abort
the download with an explicit error when either check fails, preserving the
existing chunked-write behavior only for valid range responses.
| query = f"""WITH tmp_split AS( | ||
| SELECT | ||
| split(safe_cast({column} AS STRING),",") AS chave | ||
| FROM `basedosdados-dev.br_rf_cnpj_staging.{table_id}` | ||
| ) | ||
| SELECT DISTINCT chave | ||
| FROM tmp_split, | ||
| UNNEST(chave) AS chave""" | ||
| uniques = bd.read_sql( | ||
| query=query, from_file=True, billing_project_id="basedosdados" | ||
| )["chave"].unique() |
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@pipelines/crawler/rf_cnpj/flows.py`:
- Line 99: Update the folder_date polling configuration in the flow definition
to use the documented month-only format "%Y-%m", matching the value committed by
the polling flow and allowing normal non-forced runs to parse it before main
starts.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 467d6253-28ab-4c2e-89f6-393d1b258545
📒 Files selected for processing (1)
pipelines/crawler/rf_cnpj/flows.py
luizavboas
removed
the
check-metadata
luizavboas
added
the
check-metadata
luizavboas
removed
the
test-dev-model
luizavboas
removed
the
check-metadata
|
Tick the box to add this pull request to the merge queue (same as
|
2 participants
Descrição do PR:
Este PR migra a pipeline do CNPJ da Receita Federal do dataset
br_me_cnpjparabr_rf_cnpj, junto com uma série de ajustes de qualidade de dados e infraestrutura de testes identificados durante a migração.br_me_cnpjestava desatualizado em relação à nova estrutura de organização (br_rf_cnpj) e apresentava problemas de encoding e de granularidade temporal (uso deano/mesnão refletia a real frequência de atualização dos dados, que é por data de referência do arquivo publicado pela Receita Federal).Detalhes Técnicos:
Principais alterações na pipeline/scripts:
br_me_cnpjparabr_rf_cnpj: novo flow empipelines/datasets/br_rf_cnpj/e crawler empipelines/crawler/rf_cnpj/(constants, tasks, utils, flows).br_rf_cnpj__empresas_legado,br_rf_cnpj__estabelecimentos_legado,br_rf_cnpj__socios._legado) para permitir a migração dos dados históricos já existentes nos buckets debasedosdados-devparabasedosdados, via actiontable-approve.br_rf_cnpj__dicionario.sql+ task/util correspondente no crawler) para geração automática da tabela de dicionário a partir dos dicionários manuais e extraídos da fonte.latin1parautf-8empipelines/crawler/rf_cnpj/utils.py(empresas, estabelecimentos, sócios, simples e dicionário), evitando caracteres corrompidos nos dados de saída.custom_get_where_subquery.sql: adição do placeholder__most_recent_date_cnpj__, que filtra os testes incrementais pela data mais recente da colunadata_referencia, seguindo o mesmo padrão já usado para outras fontes (ex.:__most_recent_date_cno__).@laura-l-amaral essa mudança, assim como os sqls das tabelas legado parecem estar gerando erro no
check-metadataMudanças nos dados e no schema:
ano/mespordata_referencia(data de referência da publicação dos dados pela Receita Federal) edata_modificacao(data de última atualização do registro), refletindo com mais precisão a granularidade real de atualização da fonte.br_rf_cnpj__empresas,br_rf_cnpj__estabelecimentos,br_rf_cnpj__socios,br_rf_cnpj__simples) e doschema.ymlpara a nova estrutura de colunas e para consumir as tabelas legado.Impacto no desempenho:
Teste e Validações:
Testado localmente
Testado na Cloud
Caso haja algo relacionado aos testes que vale a pena informar: Testes incrementais dos modelos agora usam
__most_recent_date_cnpj__para filtrar pordata_referencia.Riscos e Mitigações:
Erros de metadados devido à macro e sqls de tabelas legado:
Sem esses arquivos, a action roda sem erros: https://github.com/basedosdados/pipelines/actions/runs/31426064305/job/93583744445
Dependencias:
Summary by CodeRabbit
Summary by CodeRabbit