dupehell 0.5.1

"""
Refined audit: check date types, phone corruption by GT match_type, and null rates per-entity.
"""
import json, os, re, tempfile, shutil
from datetime import datetime
from dupehell import *
import polars as pl

DOMAINS = ['hospitality', 'travel', 'retail', 'cybersecurity', 'gaming', 'sports', 'social_media']

DATE_PATTERNS = [
    (re.compile(r'^\d{4}-\d{2}-\d{2}$'), '%Y-%m-%d'),
    (re.compile(r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}'), '%Y-%m-%d %H:%M:%S'),
    (re.compile(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}'), '%Y-%m-%dT%H:%M:%S'),
    (re.compile(r'^\d{2}-\d{2}-\d{4}$'), '%d-%m-%Y'),
    (re.compile(r'^\d{2}/\d{2}/\d{4}$'), '%m/%d/%Y'),
    (re.compile(r'^\d{4}/\d{2}/\d{2}$'), '%Y/%m/%d'),
]

for domain in DOMAINS:
    print(f"\n{'='*60}")
    print(f"[{domain}]")
    print(f"{'='*60}")
    
    outdir = tempfile.mkdtemp(prefix=f'rf_{domain}_')
    try:
        result = generate(domain=domain, size=1000, seed=42, output_dir=outdir)
        df = pl.read_parquet(result.dataset)
        gt = pl.read_parquet(result.ground_truth)
    except Exception as e:
        print(f"  ERROR: {e}")
        shutil.rmtree(outdir, ignore_errors=True)
        continue

    n = len(df)
    print(f"  Records: {n}, GT: {len(gt)}")

    # Merge GT match_type into data
    df_with_gt = df.join(gt.select(['record_id', 'match_type', 'master_id']), on='record_id', how='left')

    # 1. Check date column types and values
    date_cols = [c for c in df.columns if any(k in c.lower() for k in ['date', 'birth', 'dob'])]
    with open(os.path.join('schemas', f'{domain}.json')) as f:
        schema = json.load(f)
    for e in schema['entities']:
        for c in e['columns']:
            if c.get('type') in ('date', 'datetime') and c['name'] not in date_cols:
                date_cols.append(c['name'])
    date_cols = list(set(date_cols))

    for dc in date_cols:
        print(f"\n  Date column: {dc}")
        print(f"    Polars dtype: {df[dc].dtype}")
        non_null = df.filter(pl.col(dc).is_not_null())[dc]
        total = len(non_null)
        if total == 0:
            continue
        
        # Check types
        types = set(type(v).__name__ for v in non_null.head(100))
        print(f"    Python types in first 100: {types}")
        
        viol = 0
        samples = []
        for v in non_null:
            if isinstance(v, (datetime,)):
                continue
            v_str = str(v)
            parsed = False
            for pat, fmt in DATE_PATTERNS:
                if pat.match(v_str):
                    try:
                        datetime.strptime(v_str, fmt)
                        parsed = True
                        break
                    except ValueError:
                        continue
            if not parsed:
                viol += 1
                if len(samples) < 10:
                    samples.append(v_str)
        pct = viol / total * 100
        if pct > 0:
            print(f"    FAIL: {viol}/{total} ({pct:.1f}%) unparseable")
            print(f"    Samples (first 10): {samples}")
        else:
            print(f"    PASS: {total} values all valid")

    # 2. Check phone corruption by match_type
    phone_cols = [c for c in df.columns if any(k in c.lower() for k in ['phone', 'telephone', 'contact_number'])]
    for pc in phone_cols:
        print(f"\n  Phone column: {pc}")
        non_null = df_with_gt.filter(pl.col(pc).is_not_null()).select([pc, 'match_type', 'master_id'])
        total = len(non_null)
        if total == 0:
            continue
        # Show all unique patterns for first 30 values
        all_vals = non_null[pc].to_list()
        # Find bad ones
        bad = []
        for v in all_vals:
            v = str(v)
            allowed = set('0123456789 +-()')
            bad_chars = [c for c in v if c not in allowed]
            digits = sum(1 for c in v if c.isdigit())
            if bad_chars or digits < 8:
                bad.append(v)
        
        print(f"    Total non-null: {total}, Bad: {len(bad)} ({len(bad)/total*100:.1f}%)")
        if bad:
            # Check what match_types these belong to
            bad_records = non_null.filter(pl.col(pc).is_in(bad[:50]))
            if len(bad_records) > 0:
                print(f"    Bad phone match_type distribution:")
                mt_counts = bad_records.group_by('match_type').len().sort('len', descending=True)
                for row in mt_counts.to_dicts():
                    print(f"      {row['match_type']}: {row['len']}")
            print(f"    Bad samples: {bad[:10]}")

    # 3. Check cybersecurity.first_seen specifically
    if domain == 'cybersecurity':
        print(f"\n  [SPECIAL] first_seen corrupted values analysis")
        if 'first_seen' in df.columns:
            bad_first_seen = df_with_gt.filter(
                pl.col('first_seen').is_not_null()
            ).with_columns(
                pl.col('first_seen').cast(pl.Utf8).alias('first_seen_str')
            )
            # Show values that don't match date pattern
            corrupted = []
            for row in bad_first_seen.to_dicts():
                v = str(row['first_seen'])
                ok = isinstance(v, datetime)
                if not ok:
                    for pat, fmt in DATE_PATTERNS:
                        if pat.match(v):
                            ok = True
                            break
                if not ok:
                    corrupted.append({
                        'record_id': row.get('record_id'),
                        'mt': row.get('match_type'),
                        'val': v
                    })
            print(f"    Corrupted: {len(corrupted)} values")
            print(f"    First 20: {corrupted[:20]}")

    shutil.rmtree(outdir, ignore_errors=True)