#!/usr/bin/env python3
"""Rebuild published aggregates: python3 scripts/build-direct-mail-study.py
Refresh from supplied exports (pandas + pyarrow): add --collect /workspace/seo
Default rebuild uses only Python's standard library and the committed raw JSON.
"""
import argparse
import ast
from collections import Counter, defaultdict
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import re
import shutil
import statistics
from urllib.parse import urlencode

ROOT = Path(__file__).resolve().parents[1]
SLUG = 'direct-mail-search-demand-2026'
RAW = ROOT / 'data' / f'{SLUG}.json'
ACCESS = datetime.now(timezone.utc).date().isoformat()
DOMAINS = ['lob.com', 'postalytics.com', 'thanks.io', 'handwrytten.com']
CHANNEL = r'\b(?:direct[ -]mail|post ?cards?|mailers?|eddm|every door direct mail)\b'
# First match wins. Narrow industries precede broader finance and healthcare.
INDUSTRIES = [
    ('mortgage', r'\bmortgage\w*|\brefinanc\w*|\bloan officers?\b'),
    ('insurance', r'\binsur(?:ance|ers?)\b|\bmedicare\b'),
    ('real_estate', r'\breal estate\b|\brealtors?\b|\brealty\b|\bjust (?:listed|sold)\b|\bopen house\b'),
    ('dental', r'\bdent(?:al|ist\w*)\b|\borthodont\w*'),
    ('nonprofit', r'\bnon[ -]?profit\w*|\bcharit\w*|\bfundrais\w*|\bdonors?\b|\bchurch\w*'),
    ('political', r'\bpolitic\w*|\belectoral\b|\belection\w*|\bvoters?\b'),
    ('auto', r'\bautomotive\b|\bauto\b|\bcar (?:dealer\w*|repair|sales)\b|\bdealership\w*'),
    ('home_services', r'\bhome services?\b|\bhvac\b|\broof\w*|\bplumb\w*|\blandscap\w*|\blawn\b|\bpest control\b|\bcontractors?\b|\bcarpet clean\w*|\belectricians?\b|\bsolar\b'),
    ('retail_hospitality', r'\bretail\w*|\brestaurants?\b|\bhotels?\b|\be[ -]?commerce\b|\bsalons?\b|\bgyms?\b|\bfitness\b|\bhealth clubs?\b'),
    ('healthcare', r'\bhealth\w*|\bmedical\b|\bpatient\w*|\bphysician\w*|\bchiropract\w*|\boptometr\w*|\bhospitals?\b|\bpharmac\w*|\bdoctors?\b'),
    ('financial', r'\bfinanc\w*|\bbank\w*|\bcredit unions?\b|\bwealth\b'),
    ('legal', r'\blawyers?\b|\battorneys?\b|\blaw firms?\b|\blegal\b'),
]

def normalize(s):
    return ' '.join(str(s).lower().split())

def industry(k):
    return next((name for name, pattern in INDUSTRIES if re.search(pattern, k)), None)

def candidate(k):
    return bool(re.search(CHANNEL, k)) and industry(k) is not None

def num(v):
    try:
        f = float(v)
        return f if f == f and abs(f) != float('inf') else None
    except (ValueError, TypeError):
        return None

def intent(v):
    labels = sorted({s.strip().title() for s in str(v).split(',') if s.strip().title() in {'Commercial', 'Informational', 'Navigational', 'Transactional'}})
    return ', '.join(labels) or 'Unknown'

def dump(path, obj):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(obj, ensure_ascii=False, indent=2, allow_nan=False) + '\n')

def collect(base):
    import pandas as pd
    base = Path(base)
    sources = []
    def source(path, url, role):
        data = path.read_bytes()
        sid = path.name if path.parent.name == 'analysis' else f'{path.parent.name}/{path.name}'
        sources.append(dict(id=sid, source_file=str(path.relative_to(base)), source_url=url, accessed_at=ACCESS,
                            export_date=None, file_modified_utc=datetime.fromtimestamp(path.stat().st_mtime, timezone.utc).isoformat(),
                            sha256=hashlib.sha256(data).hexdigest(), role=role))
        return sid
    kmt_url = 'https://www.semrush.com/analytics/keywordmagic/?db=us'
    parquet = base / 'analysis/keywords-clean.parquet'
    sid = source(parquet, kmt_url, 'Primary keyword metrics; US rows only')
    clean = pd.read_parquet(parquet)
    rows = []
    for idx, r in clean[clean.db == 'US'].iterrows():
        k = normalize(r.keyword)
        if candidate(k):
            seed = str(r.seeds).split('|')[0]
            rows.append(dict(keyword=k, volume=num(r.volume), kd=num(r.kd), cpc=num(r.cpc), intent=intent(r.intent),
                             source_id=sid, source_row=int(idx), source_url='https://www.semrush.com/analytics/keywordmagic/?' + urlencode({'q':seed, 'db':'us'}), accessed_at=ACCESS))
    gap = base / 'analysis/competitor-gap-keywords.csv'
    sid = source(gap, 'https://www.semrush.com/analytics/organic/positions/?db=us', 'Secondary metrics for keywords absent from cleaned KMT')
    for idx, r in pd.read_csv(gap).iterrows():
        k = normalize(r.keyword)
        if candidate(k):
            rows.append(dict(keyword=k, volume=num(r.volume), kd=num(r.kd), cpc=num(r.cpc), intent=intent(r.intent),
                             source_id=sid, source_row=int(idx)+2, source_url=sources[-1]['source_url'], accessed_at=ACCESS))
    ranking_dates = {}
    for domain in DOMAINS:
        path = base / f'semrush/competitors/{domain}/positions.csv'
        url = 'https://www.semrush.com/analytics/organic/positions/?' + urlencode({'q': domain, 'db':'us'})
        sid = source(path, url, 'Organic positions and tertiary keyword metrics')
        df = pd.read_csv(path)
        ranking_dates[domain] = [df.Timestamp.min(), df.Timestamp.max()]
        for idx, r in df.iterrows():
            k = normalize(r.Keyword)
            if candidate(k) and r['Position Type'] == 'Organic':
                rows.append(dict(keyword=k, volume=num(r['Search Volume']), kd=num(r['Keyword Difficulty']), cpc=num(r.CPC), intent=intent(r['Keyword Intents']),
                                 domain=domain, position=num(r.Position), ranking_url=r.URL, observed_at=r.Timestamp,
                                 source_id=sid, source_row=int(idx)+2, source_url=url, accessed_at=ACCESS))
    # Preserve the upstream noise rules exactly, as data, without executing its script.
    tree = ast.parse((base / 'analysis/scripts/build.py').read_text())
    noise = next(ast.literal_eval(node.value) for node in tree.body if isinstance(node, ast.Assign) and any(isinstance(t, ast.Name) and t.id == 'N' for t in node.targets))
    for name in ['keyword-opportunities.csv', 'stats.json']:
        source(base / 'analysis' / name, kmt_url, 'Context only; never added to keyword totals')
    source(base / 'analysis/scripts/build.py', kmt_url, 'Upstream cleaning rule provenance')
    # Fetch the published offer; fail collection if its price/minimum cannot be verified.
    import subprocess
    offer_url = 'https://fer.marketing/direct-mail-for-agencies'
    offer_html = subprocess.check_output(['curl', '-fsSL', offer_url], text=True)
    evidence = 'Campaigns have a 50-card minimum ($500) at $10 per custom postcard.'
    assert evidence in offer_html, 'Published offer has changed; review before collecting'
    offer = dict(source_url=offer_url, accessed_at=ACCESS,
                 price_usd=10, minimum_postcards=50,
                 evidence=evidence, fetched_html_sha256=hashlib.sha256(offer_html.encode()).hexdigest())
    dump(RAW, dict(slug=SLUG, database='US', accessed_at=ACCESS, sources=sources, offer=offer,
                   export_date_note='Exact download/export dates are absent. File modification dates are not export dates. Competitor Timestamp fields are SERP observation dates.',
                   ranking_observation_ranges=ranking_dates, upstream_stats=json.loads((base/'analysis/stats.json').read_text()),
                   upstream_noise_rules=noise, candidate_observations=rows))

def build():
    raw = json.loads(RAW.read_text())
    groups = defaultdict(list)
    for r in raw['candidate_observations']:
        groups[normalize(r['keyword'])].append(r)
    exclusions = Counter()
    records = []
    def priority(r):
        if r['source_id'] == 'keywords-clean.parquet': return (0, '', '')
        if r['source_id'] == 'competitor-gap-keywords.csv': return (1, '', '')
        # Most recently observed Organic row, then domain and position break ties.
        return (2, ''.join(chr(127-ord(c)) for c in r['observed_at']), r['domain'], r['position'])
    for k, observations in sorted(groups.items()):
        reason = next((name for name, pattern in raw['upstream_noise_rules'].items() if re.search(pattern, k, re.I)), None)
        if not reason and re.search(r'\b(?:irs|990|e[ -]?postcard|postcard filing|landscape size|ecrwss|capital one|is direct mail legal)\b', k):
            reason = 'study_off_topic_or_brand'
        if not reason and any(int(y) < 2026 for y in re.findall(r'\b20\d{2}\b', k)):
            reason = 'historical_year'
        if not reason and re.search(r'^(?:how|what|which|why|when|where|does|do|is|are|can)\b|\b(?:statistics?|stats|response rates?|conversion rates?|effectiveness|roi|benchmarks?)\b', k):
            reason = 'question_or_statistics_scope'
        if not reason and re.search(r'\b(?:eddm retail|every door direct mail retail|retail direct mail)\b', k) and not re.search(r'\b(?:stores?|retailers?|brands?|marketing|campaigns?)\b', k):
            reason = 'postal_product_not_retail_industry'
        if not reason and re.search(r'\b(?:compass|merrill|fieldedge|toolssports|postcard realty|automotive direct mail pros)\b', k):
            reason = 'named_brand_or_malformed'
        if reason:
            exclusions[reason] += 1
            continue
        selected = sorted(observations, key=priority)[0]
        if selected['volume'] is None or selected['volume'] <= 0:
            exclusions['nonpositive_or_missing_volume'] += 1
            continue
        ranks = []
        for domain in DOMAINS:
            matches = [r for r in observations if r.get('domain') == domain and r.get('position') is not None and 1 <= r['position'] <= 100]
            if matches:
                best = sorted(matches, key=lambda r: (r['position'], r['ranking_url']))[0]
                ranks.append({key:best[key] for key in ['domain','position','ranking_url','observed_at','source_id','source_row','source_url','accessed_at']})
        records.append(dict(keyword=k, industry=industry(k), volume=selected['volume'], kd=selected['kd'], cpc=selected['cpc'], intent=selected['intent'],
                            metrics_source={key:selected[key] for key in ['source_id','source_row','source_url','accessed_at']}, rankings=ranks))
    def stats(rs):
        kds = [r['kd'] for r in rs if r['kd'] is not None]
        cpcs = [r['cpc'] for r in rs if r['cpc'] is not None]
        return dict(keywords=len(rs), volume=int(sum(r['volume'] for r in rs)), median_kd=statistics.median(kds) if kds else None,
                    kd_known=len(kds), median_cpc=statistics.median(cpcs) if cpcs else None, cpc_known=len(cpcs),
                    positive_cpc=sum(c > 0 for c in cpcs), low_kd_volume=int(sum(r['volume'] for r in rs if r['kd'] is not None and r['kd'] <= 15)),
                    any_top10_keywords=sum(any(x['position'] <= 10 for x in r['rankings']) for r in rs),
                    no_top10_volume=int(sum(r['volume'] for r in rs if not any(x['position'] <= 10 for x in r['rankings']))))
    industries = []
    for name, _ in INDUSTRIES:
        rs = [r for r in records if r['industry'] == name]
        industries.append(dict(industry=name, **stats(rs), competitors={d:dict(top10=sum(any(x['domain']==d and x['position']<=10 for x in r['rankings']) for r in rs), top100=sum(any(x['domain']==d for x in r['rankings']) for r in rs)) for d in DOMAINS}))
    industries.sort(key=lambda r: (-r['volume'], r['industry']))
    for r in industries:
        r['demand_share_pct'] = 100*r['volume']/sum(x['volume'] for x in industries)
        r['low_kd_volume_share_pct'] = 100*r['low_kd_volume']/r['volume'] if r['volume'] else None
        r['top10_keyword_share_pct'] = 100*r['any_top10_keywords']/r['keywords'] if r['keywords'] else None
    eligible = [r for r in records if len(r['keyword'].split()) >= 4 and r['volume'] >= 20 and r['kd'] is not None and r['kd'] <= 15 and r['cpc'] is not None and r['cpc'] > 0 and 'Navigational' not in r['intent']]
    eligible.sort(key=lambda r: (-r['volume'], r['kd'], -r['cpc'], r['keyword']))
    # One phrase per industry before additional phrases, keeping the shortlist useful across niches.
    shortlist, seen = [], set()
    for r in eligible:
        if r['industry'] not in seen:
            shortlist.append(r); seen.add(r['industry'])
    for r in eligible:
        if len(shortlist) >= 12: break
        if r not in shortlist: shortlist.append(r)
    summary = dict(slug=SLUG, accessed_at=raw['accessed_at'], database=raw['database'], candidate_observations=len(raw['candidate_observations']),
                   upstream_us_keywords=raw['upstream_stats']['clean_by_db']['US'],
                   candidate_keywords=len(groups), exclusions=dict(exclusions), totals=stats(records), industries=industries,
                   intents=[dict(intent=i, **stats([r for r in records if r['intent']==i])) for i in sorted({r['intent'] for r in records})],
                   shortlist_eligible=len(eligible), shortlist=shortlist,
                   shortlist_pool=stats(eligible), shortlist_gap=stats([r for r in eligible if not any(x['position']<=10 for x in r['rankings'])]),
                   metric_source_counts=dict(Counter(r['metrics_source']['source_id'] for r in records)),
                   ranking_observation_ranges=raw['ranking_observation_ranges'], sources=raw['sources'], offer=raw['offer'])
    assert sum(r['volume'] for r in industries) == summary['totals']['volume']
    assert sum(r['keywords'] for r in industries) == len(records)
    assert sum(r['keywords'] for r in summary['intents']) == len(records)
    assert len(records) + sum(exclusions.values()) == len(groups)
    assert all(r['industry'] and r['volume'] > 0 and (r['kd'] is None or 0 <= r['kd'] <= 100) for r in records)
    assert len({r['keyword'] for r in records}) == len(records)
    dump(ROOT/'data'/f'{SLUG}-summary.json', summary)
    dest = ROOT/'public/studies'/SLUG
    dest.mkdir(parents=True, exist_ok=True)
    shutil.copyfile(RAW, dest/'data.json')
    dump(dest/'keywords.json', records)
    dump(dest/'summary.json', summary)
    shutil.copyfile(Path(__file__), dest/'build.py')
    print(json.dumps({k:summary[k] for k in ['candidate_observations','candidate_keywords','exclusions','totals','metric_source_counts']}, indent=2))
    print('\nIndustries:', [(r['industry'],r['keywords'],r['volume'],r['median_kd'],r['median_cpc']) for r in industries])
    print('\nShortlist:', [(r['keyword'],r['volume'],r['kd'],r['cpc']) for r in shortlist])

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--collect', metavar='SEO_ROOT')
    args = parser.parse_args()
    if args.collect: collect(args.collect)
    build()