Pipeline

Files/DB -> Ingest -> Column Map -> Auto Fix -> Validate -> Standardize
         -> Matchkeys -> Block -> Score -> Cluster -> Golden -> Output

from goldenmatch.core.ingest import load_file, load_files

lf = load_file("customers.csv")          # Returns LazyFrame
df = load_file("data.parquet").collect()  # Collect to DataFrame
df = load_files([("a.csv", "source_a"), ("b.csv", "source_b")])

gm.auto_map_columns(df_a, df_b)
# [{'col_a': 'full_name', 'col_b': 'first_name', 'score': 0.85, 'method': 'synonym'}, ...]

df, fixes = gm.auto_fix_dataframe(df)
# fixes: [{"column": "phone", "fix": "stripped non-digits", "rows": 142}]

validation:
  auto_fix: true
  rules:
    - column: email
      rule_type: regex
      params: { pattern: "^.+@.+\\..+$" }
      action: flag

standardization:
  rules:
    email: [email]
    first_name: [name_proper, strip]
    phone: [phone]
    zip: [zip5]

df = gm.compute_matchkeys(df, matchkeys)
# Adds __mk_exact_email__, __mk_fuzzy_name__, etc.

blocks = gm.build_blocks(df, blocking_config)
# Returns list of DataFrames, one per block

pairs = gm.find_exact_matches(lf, matchkey)  # lf: LazyFrame, matchkey: MatchkeyConfig

pairs = gm.find_fuzzy_matches(block_df, matchkey, exclude_pairs=set())

clusters = gm.build_clusters(scored_pairs)
# Returns dict[int, dict] with keys: members, size, pair_scores, confidence, bottleneck_pair

gm.add_to_cluster(record_id, matches, clusters)   # Join or merge clusters
gm.unmerge_record(record_id, clusters)             # Remove and re-cluster
gm.unmerge_cluster(cluster_id, clusters)           # Shatter to singletons

golden = gm.build_golden_record(cluster_df, golden_rules)  # cluster_df is one cluster's rows

gm.write_output(golden_df, "results/", "run1", "golden", "csv")  # df, directory, run_name, output_type, fmt

rulebooks = gm.discover_rulebooks()
enhanced_df, low_conf = gm.extract_with_rulebook(df, "title", rulebooks["electronics"])