audit-labs/audit-tools

 1"""Exact-match filter parsing and application."""
 2
 3from __future__ import annotations
 4
 5import pandas as pd
 6
 7from .io import AuditSamplingError
 8
 9
10def parse_filters(filter_values: list[str] | None) -> dict[str, str]:
11    parsed: dict[str, str] = {}
12    for value in filter_values or []:
13        if "=" not in value:
14            raise AuditSamplingError(
15                f"Invalid filter '{value}'. Expected format: Column=Value"
16            )
17        column, expected = value.split("=", 1)
18        column = column.strip()
19        if not column:
20            raise AuditSamplingError(
21                f"Invalid filter '{value}'. Filter column cannot be blank."
22            )
23        parsed[column] = expected.strip()
24    return parsed
25
26
27def apply_filters(
28    population: pd.DataFrame, filters: dict[str, str]
29) -> tuple[pd.DataFrame, pd.DataFrame]:
30    if not filters:
31        return population.copy(), population.iloc[0:0].copy()
32
33    missing = [column for column in filters if column not in population.columns]
34    if missing:
35        raise AuditSamplingError(f"Filter column(s) not found: {', '.join(missing)}")
36
37    keep_mask = pd.Series(True, index=population.index)
38    for column, expected in filters.items():
39        keep_mask &= population[column].astype("string").fillna("") == expected
40
41    excluded = population.loc[~keep_mask].copy()
42    if not excluded.empty:
43        excluded["_exclusion_reason"] = "Filtered out"
44    return population.loc[keep_mask].copy(), excluded