Spaces:
Sleeping
Sleeping
| """ | |
| Practical, honest normalization — not a textbook-complete 5NF engine (see the plan's | |
| "feasibility note": BCNF/4NF/5NF require detecting multi-valued/join dependencies, which has | |
| no safe general algorithm for arbitrary real-world data). What this module actually does: | |
| - 1NF: handled at classification time (backend/databases.py splits delimiter-separated values | |
| only in columns classified as "tag" — the one column type where a multi-value cell is | |
| expected by design, not a guess). | |
| - 2NF/3NF: real, mechanical functional-dependency discovery (a determinant column whose every | |
| distinct value maps to exactly one value of another column, and which isn't itself a | |
| candidate key) — automatically extracted into a natural-keyed lookup table. | |
| - Beyond 3NF: only ever *suggested* (a plain-language reason attached), never auto-applied — | |
| a composite pair of columns that together determine a third is flagged, not restructured. | |
| """ | |
| from __future__ import annotations | |
| from dataclasses import dataclass, field | |
| class NormalizedTable: | |
| name: str | |
| columns: list[str] | |
| rows: list[dict] | |
| class NormalizationResult: | |
| root: NormalizedTable | |
| lookups: list[NormalizedTable] = field(default_factory=list) | |
| suggestions: list[str] = field(default_factory=list) | |
| def _distinct_count(rows: list[dict], col: str) -> int: | |
| return len({row.get(col) for row in rows}) | |
| def discover_functional_dependencies( | |
| rows: list[dict], headers: list[str], exclude_determinants: set[str] = frozenset(), | |
| ) -> dict[str, set[str]]: | |
| """{determinant: {columns it functionally determines}}. A column only qualifies as a | |
| determinant if it has fewer distinct values than there are rows (otherwise it's a | |
| candidate key, not a useful grouping column), more than one distinct value (otherwise | |
| it's constant and splitting on it is meaningless), and isn't in `exclude_determinants` — | |
| a per-row identity column (an email address) can look like a "determinant" purely because | |
| duplicate rows collapse its distinct count, which isn't a real grouping dimension; it's | |
| still fine for such a column to be a *dependent* of something else.""" | |
| n = len(rows) | |
| deps: dict[str, set[str]] = {} | |
| if n < 2: | |
| return deps | |
| for a in headers: | |
| if a in exclude_determinants: | |
| continue | |
| distinct_a = _distinct_count(rows, a) | |
| if distinct_a >= n or distinct_a <= 1: | |
| continue | |
| for b in headers: | |
| if a == b: | |
| continue | |
| mapping: dict[object, object] = {} | |
| consistent = True | |
| for row in rows: | |
| va, vb = row.get(a), row.get(b) | |
| if va in mapping and mapping[va] != vb: | |
| consistent = False | |
| break | |
| mapping[va] = vb | |
| if consistent: | |
| deps.setdefault(a, set()).add(b) | |
| return deps | |
| def normalize( | |
| rows: list[dict], headers: list[str], root_name: str = "root", exclude_determinants: set[str] = frozenset(), | |
| auto_split: bool = True, | |
| ) -> NormalizationResult: | |
| """Splits out one lookup table per determinant column (first-match-wins: once a column's | |
| dependents are extracted, it isn't reconsidered as a dependent of a later determinant, to | |
| avoid conflicting/overlapping splits), then scans what's left for a composite-determinant | |
| signal worth flagging but not acting on. `exclude_determinants` — see | |
| discover_functional_dependencies — keeps identity-like columns (email) from being treated | |
| as a grouping dimension. | |
| auto_split=False keeps every column in the one table and reports nothing. Uploads use that: | |
| "A determines B" found in a few hundred rows is usually coincidence (names are nearly unique), | |
| and moving real columns out of the sheet because of it hid people's data from them.""" | |
| if auto_split is False: | |
| return NormalizationResult( | |
| root=NormalizedTable(name=root_name, columns=list(headers), rows=[{c: r.get(c) for c in headers} for r in rows]), | |
| ) | |
| deps = discover_functional_dependencies(rows, headers, exclude_determinants) | |
| extracted_as_dependent: set[str] = set() | |
| used_as_determinant: set[str] = set() | |
| lookups: list[NormalizedTable] = [] | |
| for determinant, dependents in deps.items(): | |
| # A column already folded into another table as a dependent can't turn around and | |
| # become a determinant itself — and a two-way dependency (A<->B, common whenever both | |
| # sides happen to have the same distinct-value count, e.g. company <-> company_city) | |
| # would otherwise get extracted TWICE, once from each direction, stripping both | |
| # columns out of root instead of just one. | |
| if determinant in extracted_as_dependent: | |
| continue | |
| dependents = { | |
| d for d in dependents | |
| if d not in extracted_as_dependent and d not in used_as_determinant and d != determinant | |
| } | |
| if not dependents: | |
| continue | |
| lookup_cols = [determinant, *sorted(dependents)] | |
| seen_keys: set[object] = set() | |
| lookup_rows = [] | |
| for row in rows: | |
| key = row.get(determinant) | |
| if key in seen_keys: | |
| continue | |
| seen_keys.add(key) | |
| lookup_rows.append({c: row.get(c) for c in lookup_cols}) | |
| lookups.append(NormalizedTable(name=f"{determinant}_lookup", columns=lookup_cols, rows=lookup_rows)) | |
| used_as_determinant.add(determinant) | |
| extracted_as_dependent.update(dependents) | |
| root_columns = [h for h in headers if h not in extracted_as_dependent] | |
| root_rows = [{c: row.get(c) for c in root_columns} for row in rows] | |
| suggestions = _composite_dependency_suggestions(rows, root_columns, exclude_determinants) | |
| return NormalizationResult( | |
| root=NormalizedTable(name=root_name, columns=root_columns, rows=root_rows), | |
| lookups=lookups, | |
| suggestions=suggestions, | |
| ) | |
| def _composite_dependency_suggestions( | |
| rows: list[dict], headers: list[str], exclude_determinants: set[str] = frozenset(), | |
| ) -> list[str]: | |
| """A bounded (O(columns^2)) scan for pairs (a, b) that together determine a third column | |
| c, where neither a nor b alone does — a BCNF/4NF-flavored signal. Reported as plain- | |
| language text for a human to accept, never auto-applied.""" | |
| n = len(rows) | |
| if n < 3: | |
| return [] | |
| single_deps = discover_functional_dependencies(rows, headers, exclude_determinants) | |
| suggestions: list[str] = [] | |
| candidate_headers = [h for h in headers if h not in exclude_determinants] | |
| for i, a in enumerate(candidate_headers): | |
| for b in candidate_headers[i + 1 :]: | |
| pair_values = {(row.get(a), row.get(b)) for row in rows} | |
| if len(pair_values) >= n: | |
| continue # (a, b) together are already unique per row — not a useful grouping | |
| for c in headers: | |
| if c in (a, b): | |
| continue | |
| if c in single_deps.get(a, set()) or c in single_deps.get(b, set()): | |
| continue # already explained by a single-column dependency | |
| mapping: dict[tuple, object] = {} | |
| consistent = True | |
| for row in rows: | |
| key = (row.get(a), row.get(b)) | |
| vc = row.get(c) | |
| if key in mapping and mapping[key] != vc: | |
| consistent = False | |
| break | |
| mapping[key] = vc | |
| if consistent and len(mapping) < n: | |
| suggestions.append( | |
| f"Columns '{a}' and '{b}' together always determine '{c}' — " | |
| f"consider extracting a lookup table (not applied automatically)." | |
| ) | |
| return suggestions | |