Ir para o conteúdo

Overview

package wittgenstein_catalog_builder

Wittgenstein Catalog Builder — ETL data-cleaning utilities and catalog inference.

Classes

Functions

  • classificar_regiao — Classify a free-text region label into (regiao, nivel, macro_regiao, uf, rm).

  • clean_date — Normalize a date column to datetime, stripping the time component.

  • clean_numeric — Convert series to numeric, replacing '***', '-', 'X' with NaN.

  • drop_null_columns — Drop columns that are 100% null or have no usable name.

  • normalize_all_columns — Apply snake_case normalization to all column names.

  • normalize_column_name — Convert a column name to snake_case without accents.

  • read_csv_clean — Read a CSV with utf-8-sig encoding and apply basic cleaning.

  • read_excel_sheet — Read an Excel sheet and apply basic cleaning (drop null columns, trim strings).

  • rename_columns — Rename columns using a mapping dict (thin wrapper around DataFrame.rename).

  • save_csv — Save a DataFrame as CSV and print a summary line.

  • trim_strings — Trim whitespace from all string values in object columns.

  • uppercase_text_columns — Uppercase all string values in object columns (canonical casing for chat ingest).

wittgenstein_catalog_builder.CatalogBuilder

mkapi_definition_mkapi class CatalogBuilder(extra_verified: set[str] | None = None)

Infer catalog metadata from DataFrames and merge with existing YAML curations.

Methods

  • infer_granularity — Detect temporal granularity from a date column.

  • get_period — Return (min, max) period strings from a date or year column.

  • get_distinct_values — Return up to max_values sorted distinct non-null string representations.

  • infer_confidence — Return 'verified' for known common columns, 'inferred' otherwise.

  • numeric_sample — Return n evenly-spaced values from sorted distinct uniques (deterministic).

  • date_strings — Return sorted distinct dates as YYYY-MM-DD strings.

  • build_column_meta — Build a ColumnMeta from a single DataFrame column.

  • build_table_entry — Build a complete TableMeta from a DataFrame.

  • build_from_strategy — Extract via strategy then infer — one call instead of loading the DataFrame yourself first (the Excel/CSV path this library originally assumed). **kwargs forwards to build_table_entry (description, column_descriptions, primary_key, ...).

  • load_existing — Load the tables section of an existing catalog.yaml.

  • merge_curations — Overlay manual curations from an existing catalog entry onto a fresh entry.

wittgenstein_catalog_builder.CatalogBuilder.infer_granularity

mkapi_definition_mkapi method CatalogBuilder.infer_granularity(df: pd.DataFrame, date_col: str = 'data_ref') → str

Detect temporal granularity from a date column.

Returns one of: diario / mensal / trimestral / anual / snapshot. Falls back to structural columns (trimestre, ano) when date_col is absent.

wittgenstein_catalog_builder.CatalogBuilder.get_period

mkapi_definition_mkapi method CatalogBuilder.get_period(df: pd.DataFrame, date_col: str = 'data_ref') → tuple[str | None, str | None]

Return (min, max) period strings from a date or year column.

Returns (None, None) when no temporal column is found.

wittgenstein_catalog_builder.CatalogBuilder.get_distinct_values

mkapi_definition_mkapi method CatalogBuilder.get_distinct_values(series: pd.Series, max_values: int = 30) → list[str]

Return up to max_values sorted distinct non-null string representations.

wittgenstein_catalog_builder.CatalogBuilder.infer_confidence

mkapi_definition_mkapi method CatalogBuilder.infer_confidence(col_name: str) → str

Return 'verified' for known common columns, 'inferred' otherwise.

wittgenstein_catalog_builder.CatalogBuilder.numeric_sample

mkapi_definition_mkapi method CatalogBuilder.numeric_sample(series: pd.Series, n: int = 5) → list

Return n evenly-spaced values from sorted distinct uniques (deterministic).

wittgenstein_catalog_builder.CatalogBuilder.date_strings

mkapi_definition_mkapi method CatalogBuilder.date_strings(series: pd.Series) → list[str]

Return sorted distinct dates as YYYY-MM-DD strings.

wittgenstein_catalog_builder.CatalogBuilder.build_column_meta

mkapi_definition_mkapi method CatalogBuilder.build_column_meta(col: str, series: pd.Series, description: str = '', confidence: str | None = None) → ColumnMeta

Build a ColumnMeta from a single DataFrame column.

wittgenstein_catalog_builder.CatalogBuilder.build_table_entry

mkapi_definition_mkapi method CatalogBuilder.build_table_entry(name: str, df: pd.DataFrame, description: str = '', column_descriptions: dict[str, str] | None = None, column_confidence: dict[str, str] | None = None, table_type: str = 'fact', primary_key: str | None = None, foreign_keys: list[dict] | None = None) → TableMeta

Build a complete TableMeta from a DataFrame.

Parameters

  • name : str — Table name (used for labelling only).

  • df : pd.DataFrame — Source DataFrame.

  • description : str — Human-readable table description.

  • column_descriptions : dict[str, str] | None — {col_name: description} override map.

  • column_confidence : dict[str, str] | None — {col_name: confidence} override map.

  • table_type : str — "fact" or "dimension".

  • primary_key : str | None — Primary key column name.

  • foreign_keys : list[dict] | None — List of dicts with keys column and references.

wittgenstein_catalog_builder.CatalogBuilder.build_from_strategy

mkapi_definition_mkapi method CatalogBuilder.build_from_strategy(strategy: SourceStrategy, name: str, **kwargs: Any) → TableMeta

Extract via strategy then infer — one call instead of loading the DataFrame yourself first (the Excel/CSV path this library originally assumed). **kwargs forwards to build_table_entry (description, column_descriptions, primary_key, ...).

wittgenstein_catalog_builder.CatalogBuilder.load_existing

mkapi_definition_mkapi method CatalogBuilder.load_existing(path: Path) → dict[str, dict]

Load the tables section of an existing catalog.yaml.

Returns an empty dict if the file is absent or unreadable.

wittgenstein_catalog_builder.CatalogBuilder.merge_curations

mkapi_definition_mkapi method CatalogBuilder.merge_curations(entry: TableMeta, existing: dict) → TableMeta

Overlay manual curations from an existing catalog entry onto a fresh entry.

Preserved from existing

  • Table description (when not blank)
  • Column description (when not blank)
  • Column confidence (when explicitly set)

Always recomputed from data

  • rows, granularity, period, type, is_categorical, unique_count, values, values_sample, range

wittgenstein_catalog_builder.classificar_regiao

mkapi_definition_mkapi classificar_regiao(r: str) → pd.Series

Classify a free-text region label into (regiao, nivel, macro_regiao, uf, rm).

Resolution order: macro_regiao → UF → RM with '(XX)' suffix → RM nominal → 'outro'. Returns a pd.Series so DataFrame.apply expands it into 5 columns automatically.

wittgenstein_catalog_builder.ColumnMeta

mkapi_definition_mkapi class ColumnMeta()

Bases : BaseModel

wittgenstein_catalog_builder.ForeignKey

mkapi_definition_mkapi class ForeignKey()

Bases : BaseModel

wittgenstein_catalog_builder.TableMeta

mkapi_definition_mkapi class TableMeta()

Bases : BaseModel

wittgenstein_catalog_builder.ExcelCsvSourceStrategy

mkapi_definition_mkapi class ExcelCsvSourceStrategy(path: Path, *, sheet_name: str | None = None)

Bases : SourceStrategy

Wraps the existing Excel/CSV read+clean helpers — zero behavior change.

An existing ETL keeps calling CatalogBuilder exactly as it does today (it loads its own DataFrames via etl/loaders/*.py and never touches this class) — this strategy exists for a NEW caller that wants build_from_strategy to also do the Excel/CSV loading step.

Methods

wittgenstein_catalog_builder.ExcelCsvSourceStrategy.extract

mkapi_definition_mkapi method ExcelCsvSourceStrategy.extract() → pd.DataFrame

wittgenstein_catalog_builder.SapHanaSourceStrategy

mkapi_definition_mkapi class SapHanaSourceStrategy(*, host: str, port: int, user: str, password: str, schema: str, package: str, view: str, encrypt: bool = True)

Bases : SourceStrategy

Reads one purpose-built SAP HANA calculation view via hdbcli.

Optional dependency (the sap extra): hdbcli is imported lazily so installing this library for the Excel/CSV path (the original use) never requires it.

Attributes

  • qualified_view_name : str — The double-quoted HANA identifier for this view.

Methods

wittgenstein_catalog_builder.SapHanaSourceStrategy.qualified_view_name

mkapi_definition_mkapi property SapHanaSourceStrategy.qualified_view_name: str

The double-quoted HANA identifier for this view.

Both segments need quoting: the schema (_SYS_BIC, by SAP convention) and <package>/<view> because of the literal / a calculation view's technical name carries.

wittgenstein_catalog_builder.SapHanaSourceStrategy.extract

mkapi_definition_mkapi method SapHanaSourceStrategy.extract() → pd.DataFrame

wittgenstein_catalog_builder.SapHanaSourceStrategy.extract_chunks

mkapi_definition_mkapi method SapHanaSourceStrategy.extract_chunks(chunk_size: int) → Iterator[pd.DataFrame]

wittgenstein_catalog_builder.SourceStrategy

mkapi_definition_mkapi class SourceStrategy()

Bases : ABC

A named tabular source CatalogBuilder can extract + infer from.

Methods

  • extract — Load the full source as one DataFrame.

  • extract_chunks — Load the source in row-chunks of at most chunk_size.

wittgenstein_catalog_builder.SourceStrategy.extract

mkapi_definition_mkapi method SourceStrategy.extract() → pd.DataFrame

Load the full source as one DataFrame.

wittgenstein_catalog_builder.SourceStrategy.extract_chunks

mkapi_definition_mkapi method SourceStrategy.extract_chunks(chunk_size: int) → Iterator[pd.DataFrame]

Load the source in row-chunks of at most chunk_size.

The default splits the result of extract() in memory — fine for sources small enough to already fit in memory (Excel/CSV). A source with its own server-side pagination (e.g. a large SAP view) should override this instead of loading everything up front.

wittgenstein_catalog_builder.clean_date

mkapi_definition_mkapi clean_date(series: pd.Series) → pd.Series

Normalize a date column to datetime, stripping the time component.

wittgenstein_catalog_builder.clean_numeric

mkapi_definition_mkapi clean_numeric(series: pd.Series) → pd.Series

Convert series to numeric, replacing '***', '-', 'X' with NaN.

wittgenstein_catalog_builder.drop_null_columns

mkapi_definition_mkapi drop_null_columns(df: pd.DataFrame) → pd.DataFrame

Drop columns that are 100% null or have no usable name.

wittgenstein_catalog_builder.normalize_all_columns

mkapi_definition_mkapi normalize_all_columns(df: pd.DataFrame) → pd.DataFrame

Apply snake_case normalization to all column names.

wittgenstein_catalog_builder.normalize_column_name

mkapi_definition_mkapi normalize_column_name(name: str) → str

Convert a column name to snake_case without accents.

wittgenstein_catalog_builder.read_csv_clean

mkapi_definition_mkapi read_csv_clean(path: Path, **kwargs) → pd.DataFrame

Read a CSV with utf-8-sig encoding and apply basic cleaning.

wittgenstein_catalog_builder.read_excel_sheet

mkapi_definition_mkapi read_excel_sheet(path: Path, sheet_name: str, **kwargs) → pd.DataFrame

Read an Excel sheet and apply basic cleaning (drop null columns, trim strings).

wittgenstein_catalog_builder.rename_columns

mkapi_definition_mkapi rename_columns(df: pd.DataFrame, mapping: dict) → pd.DataFrame

Rename columns using a mapping dict (thin wrapper around DataFrame.rename).

wittgenstein_catalog_builder.save_csv

mkapi_definition_mkapi save_csv(df: pd.DataFrame, output_dir: Path, table_name: str) → Path

Save a DataFrame as CSV and print a summary line.

wittgenstein_catalog_builder.trim_strings

mkapi_definition_mkapi trim_strings(df: pd.DataFrame) → pd.DataFrame

Trim whitespace from all string values in object columns.

wittgenstein_catalog_builder.uppercase_text_columns

mkapi_definition_mkapi uppercase_text_columns(df: pd.DataFrame) → pd.DataFrame

Uppercase all string values in object columns (canonical casing for chat ingest).