feat(cl_res_empresas): onboard Chile Registro de Empresas y Sociedades (#1947)

* feat(cl_res_empresas): add architecture, cleaning code and dbt models

Registro de Empresas y Sociedades (Chile): sociedades constituted under the
Ley 20.659 simplified regime, 2013-05 to 2026-07, 1,609,374 rows across two
tables (sociedad, dicionario).

- Cleaning transform lives in pipelines/datasets/cl_res_empresas/utils.py and
  is imported by the one-shot bootstrap, so the future Prefect refresh shares
  one implementation.
- Comuna names are resolved to current CUT codes against a snapshot of
  br_bd_diretorios_cl; all 349 distinct source spellings map, and an unmapped
  name raises rather than arriving as NULL.
- Region is derived from the CUT code so region and comuna are always mutually
  consistent, which reassigns the 21 Nuble comunas registered before the region
  existed (Ley 21.033) from 08 to 16.
- Staging parquet is all-STRING per house convention.

dbt: 2 models built, 20/20 tests pass in dev.

* feat(cl_res_empresas): add recurring Prefect pipeline

datos.gob.cl republishes the current year's file with a later cut-off date, so
each run is a full rebuild of all years. The whole source is ~243 MB, which makes
a full re-fetch simplest and most robust.

- dump_mode is "append", not "overwrite": "overwrite" calls tb.delete(mode=all),
  which drops the materialized prod table even from the dev half of a run.
  Every partition is rewritten each run, so append has identical semantics here.
- dbt runs every table first, then tests every table. The sociedad model's
  custom_dictionary_coverage test reads the dicionario model, so interleaving
  run/test per table would test against a sibling that does not exist yet.
- Coverage is AllFree: the source is CC BY and fully public.
- Cron 17:17 BRT on days 8/15/22 — a free minute; hour 16 is crowded.

Verified: flow imports, deploy_flows discovers it, and the poll's max period
(2026-07) matches the registered coverage end exactly.

* feat(cl_res_empresas): paywall sociedad behind the BD Pro rolling window

sociedad refreshes monthly, so under the house rule its most recent window goes
to BD Pro and everything older stays free. free_lag is the default 6 months:
free 2013-05..2026-01, pro 2026-02..2026-07 against the current source end.

The dbt model is unchanged — the paywall is enforced by BigQuery Row Access
Policies, which register_table_materialization_task re-issues every run, so the
window rolls forward on its own.

Both the free (is_closed=False) and pro (is_closed=True) Coverages were created
on staging and prod first, with is_closed set on the DateTimeRanges too; without
them assert_coverage_topology hard-fails before anything is written. Verified
against the real backend on both environments, and that the ranges do not
overlap.

dicionario has no date column and takes no coverage spec.

Author: rdahis

dbt_project.yml

@@ -121,6 +121,9 @@ models:
     br_bd_diretorios_cl:
       +materialized: table
       +schema: br_bd_diretorios_cl
+    cl_res_empresas:
+      +materialized: table
+      +schema: cl_res_empresas
     br_bd_diretorios_de:
       +materialized: table
       +schema: br_bd_diretorios_de

models/cl_res_empresas/cl_res_empresas__dicionario.sql

@@ -0,0 +1,9 @@
+{{ config(alias="dicionario", schema="cl_res_empresas", materialized="table") }}
+
+select
+    safe_cast(id_tabela as string) id_tabela,
+    safe_cast(nome_coluna as string) nome_coluna,
+    safe_cast(chave as string) chave,
+    safe_cast(cobertura_temporal as string) cobertura_temporal,
+    safe_cast(valor as string) valor
+from {{ set_datalake_project("cl_res_empresas_staging.dicionario") }} as t

models/cl_res_empresas/cl_res_empresas__sociedad.sql

@@ -0,0 +1,31 @@
+{{
+    config(
+        schema="cl_res_empresas",
+        alias="sociedad",
+        materialized="table",
+        partition_by={
+            "field": "ano",
+            "data_type": "int64",
+            "range": {"start": 2013, "end": 2031, "interval": 1},
+        },
+    )
+}}
+
+
+select
+    safe_cast(ano as int64) ano,
+    safe_cast(mes as int64) mes,
+    safe_cast(id_region_tributaria as string) id_region_tributaria,
+    safe_cast(id_comuna_tributaria as string) id_comuna_tributaria,
+    safe_cast(id_region_social as string) id_region_social,
+    safe_cast(id_comuna_social as string) id_comuna_social,
+    safe_cast(rut as string) rut,
+    safe_cast(id_actuacion as string) id_actuacion,
+    safe_cast(razon_social as string) razon_social,
+    safe_cast(tipo_sociedad as string) tipo_sociedad,
+    safe_cast(tipo_actuacion as string) tipo_actuacion,
+    safe_cast(fecha_actuacion as date) fecha_actuacion,
+    safe_cast(fecha_registro as date) fecha_registro,
+    safe_cast(fecha_aprobacion_sii as date) fecha_aprobacion_sii,
+    safe_cast(capital as float64) capital
+from {{ set_datalake_project("cl_res_empresas_staging.sociedad") }} as t

models/cl_res_empresas/code/architecture/dicionario.csv

@@ -0,0 +1,6 @@
+name,bigquery_type,description,temporal_coverage,covered_by_dictionary,directory_column,measurement_unit,has_sensitive_data,observations,original_name
+id_tabela,STRING,Nombre de la tabla del conjunto a la que corresponde la clave,,no,,,no,,
+nome_coluna,STRING,Nombre de la columna a la que corresponde la clave,,no,,,no,,
+chave,STRING,Valor de la clave tal como aparece almacenado en la columna,,no,,,no,,
+cobertura_temporal,STRING,Cobertura temporal en la que la clave tiene el significado indicado,,no,,,no,,
+valor,STRING,Significado de la clave,,no,,,no,,

models/cl_res_empresas/code/architecture/sociedad.csv

@@ -0,0 +1,16 @@
+name,bigquery_type,description,temporal_coverage,covered_by_dictionary,directory_column,measurement_unit,has_sensitive_data,observations,original_name
+ano,INT64,Año de aprobación de la constitución por el Servicio de Impuestos Internos,2013(1)2026,no,br_bd_diretorios_data_tempo.ano:ano,,no,Columna de partición. Coincide exactamente con el año de fecha_aprobacion_sii en el 100% de los registros,Anio
+mes,INT64,"Mes de aprobación de la constitución por el Servicio de Impuestos Internos, de 1 a 12",,no,br_bd_diretorios_data_tempo.mes:mes,,no,"Derivado del nombre del mes en español publicado por la fuente (Enero..Diciembre). Coincide con el mes de fecha_aprobacion_sii en el 100% de los registros",Mes
+id_region_tributaria,STRING,Código CUT de la región del domicilio tributario de la sociedad,,no,br_bd_diretorios_cl.region:id_region,,no,"Derivado de los dos primeros dígitos de id_comuna_tributaria, no del campo Region Tributaria de la fuente. Para las 21 comunas de Ñuble constituidas antes de la creación de la región (Ley 21.033, septiembre de 2018) la fuente informa la región 08 y aquí se asigna la 16, de modo que id_region_tributaria e id_comuna_tributaria sean siempre consistentes entre sí y con el directorio vigente. Afecta a 6.933 registros",Region Tributaria
+id_comuna_tributaria,STRING,Código CUT de la comuna del domicilio tributario de la sociedad,,no,br_bd_diretorios_cl.comuna:id_comuna,,no,"La fuente publica el nombre de la comuna en mayúsculas y sin acentos; aquí se resuelve al código CUT vigente de 5 dígitos. Los 349 nombres distintos de la fuente resuelven sin excepción",Comuna Tributaria
+id_region_social,STRING,Código CUT de la región del domicilio social de la sociedad,,no,br_bd_diretorios_cl.region:id_region,,no,"Mismo criterio de derivación que id_region_tributaria. Nulo en 5.335 registros en los que la fuente no informa el domicilio social",Region Social
+id_comuna_social,STRING,Código CUT de la comuna del domicilio social de la sociedad,,no,br_bd_diretorios_cl.comuna:id_comuna,,no,"Domicilio social declarado en el estatuto, que difiere del domicilio tributario en el 1,6% de los registros. Nulo en 5.335 registros",Comuna Social
+rut,STRING,"Rol Único Tributario de la sociedad, con dígito verificador",,no,,,no,"Identificador único de la sociedad. Único en los 1.609.374 registros de la serie completa, por lo que identifica la sociedad y no solo la actuación",RUT
+id_actuacion,STRING,Identificador de la actuación en el Registro de Empresas y Sociedades,,no,,,no,"Secuencia global del registro, única en toda la serie. No es correlativa por año: los rangos anuales se traslapan porque la secuencia refleja el inicio del trámite y el archivo se ordena por año de aprobación del SII",ID
+razon_social,STRING,Razón social de la sociedad tal como fue inscrita,,no,,,no,"Publicada por la fuente sin normalizar mayúsculas ni abreviaturas",Razon Social
+tipo_sociedad,STRING,Tipo societario bajo el que se constituye la sociedad,,yes,,,no,"Ocho tipos admitidos por el Régimen Simplificado de la Ley 20.659. SpA aparece desde 2014 y la sociedad anónima cerrada desde 2019, cuando el régimen se amplió",Codigo de sociedad
+tipo_actuacion,STRING,Tipo de actuación registrada,,no,,,no,"Constante CONSTITUCIÓN en toda la serie: la fuente publica únicamente las constituciones, no las modificaciones, disoluciones, migraciones ni anotaciones que el Registro también tramita. Se conserva la columna porque la fuente puede incorporar otros tipos",Tipo de actuacion
+fecha_actuacion,DATE,Fecha de la primera firma de la actuación,,no,,,no,"Menor o igual a fecha_registro en el 100% de los registros",Fecha de actuacion (1era firma)
+fecha_registro,DATE,Fecha de la última firma que completa la inscripción en el Registro,,no,,,no,,Fecha de registro (ultima firma)
+fecha_aprobacion_sii,DATE,Fecha de aprobación de la constitución por el Servicio de Impuestos Internos,,no,,,no,"Define el año y el mes de la serie. Posterior o igual a fecha_registro salvo en 16 registros",Fecha de aprobacion x SII
+capital,FLOAT64,Capital declarado en la constitución de la sociedad,,no,,CLP,no,"Pesos chilenos nominales de la fecha de constitución, sin deflactar. Todos los valores son enteros positivos, entre 1 y 990.000.000.000",Capital

models/cl_res_empresas/code/clean.py

@@ -0,0 +1,56 @@
+"""One-shot onboarding bootstrap for cl_res_empresas.
+
+Downloads every yearly CSV published on datos.gob.cl and writes the partitioned
+parquet output. The cleaning transform itself lives in
+``pipelines/datasets/cl_res_empresas/utils.py`` and is shared with the recurring
+Prefect pipeline.
+
+    python models/cl_res_empresas/code/clean.py --download
+    python models/cl_res_empresas/code/clean.py
+"""
+
+import argparse
+import os
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
+
+from pipelines.datasets.cl_res_empresas.utils import (
+    clean_all,
+    download_all,
+)
+
+DATA_DIR = Path(
+    os.environ.get(
+        "CL_RES_EMPRESAS_DATA",
+        Path.home() / "Downloads" / "cl_res_empresas_data",
+    )
+)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--download", action="store_true", help="fetch the raw CSVs first"
+    )
+    parser.add_argument(
+        "--data-dir", default=str(DATA_DIR), help="scratch directory"
+    )
+    args = parser.parse_args()
+
+    data_dir = Path(args.data_dir)
+    input_dir, output_dir = data_dir / "input", data_dir / "output"
+
+    if args.download:
+        paths = download_all(input_dir)
+        print(f"downloaded {len(paths)} file(s) to {input_dir}")
+
+    counts = clean_all(input_dir, output_dir)
+    for key in sorted(counts):
+        print(f"{key:>18}: {counts[key]:>9,}")
+    print(f"\noutput written to {output_dir}")
+
+
+if __name__ == "__main__":
+    main()