Lab: Resolve Harbor Lane's customers

'''Entity resolution: web + POS customers -> golden dim_customer + crosswalk.'''
import re
from collections import defaultdict
from difflib import SequenceMatcher
import duckdb

con = duckdb.connect('harbor.duckdb')

# The match cutoff, in one place so the lab can retune it (try 0.4 and 0.9).
MATCH_THRESHOLD = 0.6

# ---- 1. load + normalize ----------------------------------------------------
def norm_email(e):
    return (e or '').strip().lower()

def norm_phone(p):
    return re.sub(r'\D', '', p or '')[-10:]

def norm_name(n):
    return re.sub(r'\s+', ' ', (n or '').strip().lower())

records = []
for wid, email, first, last, ph, zp in con.execute(
        'SELECT web_id, email, first_name, last_name, phone, zip FROM stg_web_customers').fetchall():
    records.append({'src': 'web', 'id': wid, 'email': norm_email(email),
                    'name': norm_name(f'{first} {last}'), 'phone': norm_phone(ph), 'zip': zp})
for pid, name, email, ph, zp in con.execute(
        'SELECT pos_id, full_name, email, phone, store_zip FROM stg_pos_customers').fetchall():
    records.append({'src': 'pos', 'id': pid, 'email': norm_email(email),
                    'name': norm_name(name), 'phone': norm_phone(ph), 'zip': zp})

# ---- 2. blocking: candidates share email, phone, or (last name, zip) --------
blocks = defaultdict(list)
for i, r in enumerate(records):
    if r['email']:
        blocks[('e', r['email'])].append(i)
    if r['phone']:
        blocks[('p', r['phone'])].append(i)
    if r['name'] and r['zip']:
        blocks[('nz', r['name'].split()[-1], r['zip'])].append(i)

# ---- 3. score within blocks ---------------------------------------------------
def score(a, b):
    s = 0.0
    if a['email'] and a['email'] == b['email']:
        s += 0.6
    if a['phone'] and a['phone'] == b['phone']:
        s += 0.3
    s += 0.3 * SequenceMatcher(None, a['name'], b['name']).ratio()
    return s

# ---- 4. cluster with union-find ------------------------------------------------
parent = list(range(len(records)))
def find(x):
    while parent[x] != x:
        parent[x] = parent[parent[x]]
        x = parent[x]
    return x
def union(a, b):
    parent[find(a)] = find(b)

matched_pairs = 0
for idxs in blocks.values():
    for i in range(len(idxs)):
        for j in range(i + 1, len(idxs)):
            a, b = records[idxs[i]], records[idxs[j]]
            if a['src'] != b['src'] and score(a, b) >= MATCH_THRESHOLD:
                union(idxs[i], idxs[j])
                matched_pairs += 1

clusters = defaultdict(list)
for i in range(len(records)):
    clusters[find(i)].append(records[i])

# ---- 5. survivorship: web wins names/emails; never invent ---------------------
golden = []
for key, members in enumerate(sorted(clusters.values(), key=lambda m: m[0]['id']), start=1):
    web_m = [m for m in members if m['src'] == 'web']
    pos_m = [m for m in members if m['src'] == 'pos']
    best = (web_m or pos_m)[0]
    email = next((m['email'] for m in web_m + pos_m if m['email']), None)
    phone = next((m['phone'] for m in web_m + pos_m if m['phone']), None)
    golden.append((key, email, best['name'], phone, best['zip'],
                   web_m[0]['id'] if web_m else None,
                   pos_m[0]['id'] if pos_m else None))

con.execute('''CREATE OR REPLACE TABLE dim_customer (
    customer_key INTEGER PRIMARY KEY, email TEXT, full_name TEXT,
    phone TEXT, zip TEXT, web_id TEXT, pos_id TEXT)''')
con.executemany('INSERT INTO dim_customer VALUES (?, ?, ?, ?, ?, ?, ?)', golden)

# ---- 6. crosswalk + stamp the fact table ---------------------------------------
con.execute('''CREATE OR REPLACE TABLE customer_xwalk AS
    SELECT web_id AS source_ref, customer_key FROM dim_customer WHERE web_id IS NOT NULL
    UNION ALL
    SELECT pos_id, customer_key FROM dim_customer WHERE pos_id IS NOT NULL''')
con.execute('ALTER TABLE fct_order_lines ADD COLUMN IF NOT EXISTS customer_key INTEGER')
con.execute('''UPDATE fct_order_lines f SET customer_key =
    (SELECT x.customer_key FROM customer_xwalk x WHERE x.source_ref = f.customer_ref)''')

print(f'source records : {len(records)}')
print(f'matched pairs  : {matched_pairs}')
print(f'golden records : {len(golden)}')

-- 'customers' before resolution: source records
SELECT (SELECT count(*) FROM stg_web_customers)
     + (SELECT count(*) FROM stg_pos_customers) AS raw_records;   -- 460

-- after: real humans
SELECT count(*) AS golden_customers FROM dim_customer;            -- ~320

-- repeat-purchase rate, computable correctly for the first time:
SELECT round(avg(CASE WHEN orders > 1 THEN 1.0 ELSE 0 END), 3) AS repeat_rate
FROM (SELECT customer_key, count(DISTINCT order_id) AS orders
      FROM fct_order_lines WHERE status = 'completed'
      GROUP BY 1);