"""Sydney beauty index. Python 3.10+, standard library; no network or source writes.

Run python3 analyse.py internally or in the public bundle. Both modes repeat the
same arithmetic. Public mode validates exported inputs, not omitted raw captures.
"""
import csv
import hashlib
import itertools
import json
import re
from collections import Counter
from datetime import datetime
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
from zoneinfo import ZoneInfo

ROOT = Path(__file__).resolve().parent
ENGINES = ['chatgpt', 'google-ai-mode', 'google-ai-overview', 'perplexity']
LABELS = dict(zip(ENGINES, ['ChatGPT', 'Google AI Mode', 'Google AI Overviews', 'Perplexity']))
CATS = ['business website','Google business or map record','directory','booking platform','social','media or editorial','community forum','review platform','government','other or unresolved']

def sha(path): return hashlib.sha256(path.read_bytes()).hexdigest()
def read(path): return json.loads((ROOT/path).read_text())
def write(path, data): (ROOT/path).write_text(json.dumps(data, ensure_ascii=False, indent=2)+'\n')
def norm(name):
    # Preserve the published trades calculator's ASCII tokenisation, accent limit included.
    return ' '.join(re.findall(r'[a-z0-9]+',re.sub(r'\bpty\.?\s+ltd\.?\b','',name,flags=re.I).casefold().replace('’',"'")))
def clean(url):
    u=urlsplit(url)
    assert u.scheme in {'http','https'} and u.hostname and not u.username and not u.password,url
    q=[(k,v) for k,v in parse_qsl(u.query,keep_blank_values=True) if not k.lower().startswith('utm_') and k.lower() not in {'ct-referrer','srsltid','fbclid','gclid'}]
    return urlunsplit((u.scheme,u.netloc,u.path,urlencode(q),u.fragment))
def domain(url): return urlsplit(url).hostname.lower().removeprefix('www.')
def csvfile(path,rows):
    with (ROOT/path).open('w',newline='') as f:
        w=csv.DictWriter(f,fieldnames=list(rows[0]),lineterminator='\n');w.writeheader();w.writerows(rows)
def table(head,rows):
    return '\n'.join(['| '+' | '.join(head)+' |','| '+' | '.join(['---']*len(head))+' |']+['| '+' | '.join(str(v).replace('|',r'\|').replace('\n',' ') for v in row)+' |' for row in rows])
def category(url,config):
    d=domain(url)
    if d=='fresha.com': return 'directory' if '/lp/' in url else 'booking platform'
    if d in {'google.com','google.com.au','maps.google.com','maps.app.goo.gl'}: return 'Google business or map record'
    if d.endswith(('.gov','.gov.au')): return 'government'
    for typ,roots in config['source_roots'].items():
        if any(d==r or d.endswith('.'+r) for r in roots):return typ
    return 'other or unresolved'

def inputs(frozen,qs):
    internal=(ROOT/'answers').is_dir();rows=[]
    if not internal:
        for p,d in read('input-manifest.json').items():assert sha(ROOT/p)==d,p
        rows=read('answers.json')
    else:
        assert len(list((ROOT/'answers').glob('*.json')))==200,'Unexpected or duplicate answer file'
        for p in sorted((ROOT/'answers').glob('*-Q[0-9][0-9][0-9].json')):
            a=json.loads(p.read_text());rp=ROOT/a['raw_evidence']
            assert rp.resolve().is_relative_to(ROOT.resolve()) and sha(rp)==a['raw_sha256'],p.name
            r=json.loads(rp.read_text());links=r['links'][:]
            assert r['captured_at']==a['captured_at'],p.name
            assert r.get('completion_evidence'),p.name
            for k in ['question','question_id','status']:
                if r.get(k):assert r[k]==a[k],(p.name,k)
            text=('\n\n'.join(c['text'] for c in r.get('place_cards',[]))+'\n\n'+r['text']).strip()
            assert not a.get('answer_text') or a['answer_text'].strip() in text,p.name
            if a.get('supplemental_raw_evidence'):
                sp=ROOT/a['supplemental_raw_evidence'];assert sha(sp)==a['supplemental_raw_sha256'],p.name
                links+=json.loads(sp.read_text()).get('links',[])
            fields=['engine','question_id','question','status','reason','attempted','captured_at','question_set_sha256','answer_text','businesses_named','cited_urls','source_panel_urls','completion_evidence','capture_limitations','aliases','other_names_mentioned','coding_note','model_shown','location_shown','raw_sha256']
            row={k:a[k] for k in fields if k in a}
            row['source_answer_sha256']=sha(p)
            row['captured_links']=[{'scope':l['scope'],'url':l['url'],'text':l.get('text','')} for l in links if l.get('url','').startswith(('https://','http://'))]
            rows.append(row)
    assert len(rows)==200 and {(a['engine'],a['question_id']) for a in rows}=={(e,q) for e in ENGINES for q in qs}
    for a in rows:
        key=a['engine']+'-'+a['question_id'];q=qs[a['question_id']]
        assert a['question']==q['question'] and a['question_set_sha256']==frozen['question_set_sha256'],key
        assert a['status'] in {'answered','blocked','failed','no_overview'},key
        assert datetime.fromisoformat(a['captured_at'].replace('Z','+00:00'))>=datetime.fromisoformat(frozen['frozen_at']),key
        if a['status']=='answered':
            assert a['answer_text'].strip() and a['completion_evidence'],key
            assert all(n in a['answer_text'] for n in a['businesses_named']),key
            assert len({norm(n) for n in a['businesses_named']})==len(a['businesses_named']),key
        else:assert a.get('reason') and not a.get('businesses_named'),key
        a.update({k:q[k] for k in ['category','sector','region']})
    return rows,internal

def main():
    config=read('review-config.json');frozen=read('freeze.json')
    assert sha(ROOT/'question-set.json')==frozen['question_set_sha256']
    qset=read('question-set.json');qs={q['id']:q for q in qset['questions']}
    assert len(qs)==50 and all(q['source_ids'] and all(s in qset['sources'] for s in q['source_ids']) for q in qs.values())
    rows,internal=inputs(frozen,qs)
    if internal:write('analysis-inputs.json',rows)
    pilot={(r['engine'],r['question_id']):r for r in read('coding.json')['records']}
    review=[];edge=[];apps=[];citations=[];links=[];coverage=[]
    for a in rows:
        key=a['engine']+'-'+a['question_id'];before=a['businesses_named'][:];c=config['corrections'].get(key,{})
        names=[c.get('replace_names',{}).get(n,n) for n in before]+c.get('add_names',[])
        names.sort(key=lambda n:a.get('answer_text','').find(n))
        assert len({norm(n) for n in names})==len(names) and all(n in (a.get('answer_text') or '') for n in names),key
        a['businesses_named']=names;other=[]
        for n in a.get('other_names_mentioned',[]):other.append(n if isinstance(n,dict) else {'name':n,'note':'Ancillary collector classification; excluded from additional provider counts.'})
        for n,typ in config['ancillary_names'].items():
            if n in (a.get('answer_text') or '') and not any(n in x for x in names) and not any(x['name']==n for x in other):other.append({'name':n,'note':typ+'; not an additional provider.'})
        a['other_names_mentioned']=other
        decisions=[c['decision']] if c.get('decision') else []
        if config['additional_decisions'].get(key):decisions.append(config['additional_decisions'][key])
        if a.get('coding_note'):decisions.append(a['coding_note'])
        decisions += [s for s in a.get('capture_limitations',[]) if any(w in s.lower() for w in ['alias','host','venue','source-only','source-card-only','joint','mismatch','named','provider','landmark','brand','spelling','bundall'])]
        if a.get('aliases'):decisions.append('Same-answer aliases: '+json.dumps(a['aliases'],ensure_ascii=False))
        if other:decisions.append('Ancillary names: '+'; '.join(x['name']+' ('+x['note']+')' for x in other))
        if decisions:edge.append({'answer_key':key,'before':before,'after':names,'decisions':decisions})
        review.append({'answer_key':key,'status':a['status'],'source_answer_sha256':a['source_answer_sha256'],'raw_sha256':a['raw_sha256'],'before':before,'after':names,'other_names_mentioned':other,'decisions':decisions or ['Retain provider/card labels; source-only labels are not extra appearances.']})
        coverage.append({'engine':a['engine'],'question_id':a['question_id'],'status':a['status'],'included':a['status']=='answered','reason':a.get('reason') or '','captured_at':a['captured_at']})
        a['cited_urls']=list(dict.fromkeys(clean(u) for u in a.get('cited_urls',[])))
        a['source_panel_urls']=list(dict.fromkeys(clean(u) for u in a.get('source_panel_urls',[])))
        for l in a['captured_links']:l['url']=clean(l['url'])
        exposed={l['url'] for l in a['captured_links']}
        for u in a['cited_urls']:
            assert u in exposed or (key=='google-ai-mode-Q001' and domain(u)=='google.com'),(key,u)
            links.append({'answer_key':key,'engine':a['engine'],'question_id':a['question_id'],'url':u,'domain':domain(u),'category':category(u,config)})
        if a['status']!='answered':continue
        pnames={n['name']:n for n in pilot.get((a['engine'],a['question_id']),{}).get('names',[])}
        for pos,n in enumerate(names,1):
            known=pnames.get(n,{});matches={k:known.get(k,[])[:] for k in ['own_site_urls','directory_urls','booking_urls']};evidence=[]
            if known.get('evidence'):evidence.append(known['evidence'])
            # Full provider-name matches in saved anchor/title labels only. A panel
            # label may identify an already eligible URL, never add a panel-only URL.
            # Domain spelling, neighbouring chips and link order are not evidence.
            for l in a['captured_links']:
                if l['url'] not in a['cited_urls'] or (' '+norm(n)+' ') not in (' '+norm(l['text'])+' '):continue
                field={'business website':'own_site_urls','directory':'directory_urls','booking platform':'booking_urls'}.get(category(l['url'],config))
                if field:matches[field].append(l['url']);evidence.append('Exposed '+l['scope']+' label: '+l['text'])
            for field in matches:
                matches[field]=sorted({clean(u) for u in matches[field]});assert all(u in a['cited_urls'] for u in matches[field]),(key,n,field)
            uncertain=n in config['identity_uncertain'].get(key,[])
            # A host name inside a practitioner's label does not make the
            # practitioner's site the host's own website (or the reverse).
            for field in matches:
                matches[field]=[u for u in matches[field] if u not in config.get('rejected_matches',{}).get(key,{}).get(n,[])]
            if uncertain:matches['own_site_urls']=[]
            app={'engine':a['engine'],'question_id':a['question_id'],'category':a['category'],'sector':a['sector'],'region':a['region'],'position':pos,'business_name':n,'normalised_name':norm(n),'own_website_match':bool(matches['own_site_urls']),'directory_match':bool(matches['directory_urls']),'booking_match':bool(matches['booking_urls']),'own_website_unresolved':not bool(matches['own_site_urls']),'identity_uncertain':uncertain,'answer_file':'coded-answers.json#'+key}
            apps.append(app);i=a['answer_text'].find(n)
            citations.append({'answer_key':key,'business_name':n,**matches,'identity_uncertain':uncertain,'evidence':evidence,'provider_excerpt':a['answer_text'][max(0,i-80):i+len(n)+200],'own_website_unresolved':app['own_website_unresolved']})
    ok=[a for a in rows if a['status']=='answered'];counts={};freqs={};exact={};sources={};domains={};own={}
    groups={f:{} for f in ['category','sector','region','region_group']}
    for a in rows:a['region_group']=a['region'].split(':')[0]
    for e in ENGINES:
        er=[a for a in rows if a['engine']==e];aa=[a for a in ok if a['engine']==e];ap=[a for a in apps if a['engine']==e];ll=[l for l in links if l['engine']==e];states=Counter(a['status'] for a in er)
        counts[e]={s:states[s] for s in ['answered','failed','blocked','no_overview']}
        counts[e].update({'missing':50-len(er),'no_business_answers':sum(not a['businesses_named'] for a in aa),'named_business_occurrences':len(ap),'unique_normalised_names':len({a['normalised_name'] for a in ap}),'unique_exact_names':len({a['business_name'] for a in ap}),'answers_with_captured_links':sum(bool(a['cited_urls']) for a in aa)})
        freqs[e]=[];exact[e]=[]
        for field,dest in [('normalised_name',freqs[e]),('business_name',exact[e])]:
            for n,num in sorted(Counter(a[field] for a in ap).items(),key=lambda x:(-x[1],x[0])):
                spellings=list(dict.fromkeys(a['business_name'] for a in ap if a[field]==n));dest.append({'name':spellings[0],'key':n,'spellings':spellings,'answer_count':num,'answered_denominator':len(aa)})
        sources[e]={c:len({l['question_id'] for l in ll if l['category']==c}) for c in CATS}
        domains[e]=dict(sorted(Counter(d for a in aa for d in {domain(u) for u in a['cited_urls']}).items(),key=lambda x:(-x[1],x[0])))
        own[e]={k:sum(a[k] for a in ap) for k in ['own_website_match','directory_match','booking_match','own_website_unresolved','identity_uncertain']};own[e]['appearance_denominator']=len(ap)
        for field in groups:
            labels=list(dict.fromkeys(q['region'].split(':')[0] if field=='region_group' else q[field] for q in qs.values()));groups[field][e]={}
            for label in labels:
                ga=[a for a in aa if a[field]==label];ns=[n for a in ga for n in a['businesses_named']]
                groups[field][e][label]={'completed':len(ga),'name_appearances':len(ns),'unique_normalised_names':len({norm(n) for n in ns}),'no_shortlist':sum(not a['businesses_named'] for a in ga)}
    lookup={(a['engine'],a['question_id']):a for a in ok};pairs=[];exact_pairs=[]
    for series,transform in [(pairs,norm),(exact_pairs,lambda n:n)]:
        for x,y in itertools.combinations(ENGINES,2):
            details=[{'question_id':q,'shared':sorted({transform(n) for n in lookup[x,q]['businesses_named']}&{transform(n) for n in lookup[y,q]['businesses_named']})} for q in qs if (x,q) in lookup and (y,q) in lookup]
            series.append({'a':x,'b':y,'paired':len(details),'shared':sum(bool(d['shared']) for d in details),'details':details})
    four=[{'question_id':q,'shared':sorted(set.intersection(*[{norm(n) for n in lookup[e,q]['businesses_named']} for e in ENGINES]))} for q in qs if all((e,q) in lookup for e in ENGINES)]
    dates=sorted({datetime.fromisoformat(a['captured_at'].replace('Z','+00:00')).astimezone(ZoneInfo('Australia/Sydney')).date().isoformat() for a in rows})
    result={'status':'COLLECTION SETTLED / ANALYSIS COMPLETE / UNPUBLISHED','question_set_sha256':frozen['question_set_sha256'],'collection_dates_sydney':dates,'planned':200,'saved':len(rows),'completed':len(ok),'counts':counts,'registered_question_sources':len(qset['sources']),'used_question_sources':len({s for q in qs.values() for s in q['source_ids']}),'business_frequency':freqs,'exact_business_frequency':exact,'source_categories':sources,'domains':domains,'own_site_lower_bounds':own,'groups':groups,'engine_pairs':pairs,'exact_engine_pairs':exact_pairs,'four_engine_paired':len(four),'four_engine_shared':sum(bool(d['shared']) for d in four),'four_engine_details':four,'exclusions':[a for a in coverage if not a['included']],'uncertain_identity_appearances':[a for a in apps if a['identity_uncertain']],'citation_matching_method':'Pilot reviewed associations plus exact full-provider-name exposed anchor/title matches for already eligible URLs; no domain-name inference. Unmatched is unresolved, not absent.','evidence':{'arithmetic':'machine-checkable','coding':'claim-checkable from saved captures','engine_business_claims':'unverified'}}
    write('analysis.json',result);write('coded-answers.json',rows);write('coding-review.json',review);write('citation-review.json',{'method':result['citation_matching_method'],'appearances':citations})
    csvfile('appearances.csv',apps);csvfile('coverage.csv',coverage);csvfile('source-links.csv',links)
    write('validation.json',{'record_count':200,'unique_pairs':200,'frozen_prompt_and_hash':'PASS','statuses_and_completion':'PASS','exact_provider_spans':'PASS','citation_destinations':'PASS','evidence_hash_validation':'original raw/answer files' if internal else 'exported input manifest; original raw evidence omitted','excluded_records':result['exclusions']})
    md=['# Sydney beauty AI search index: results','**Completed analysis, unpublished. Collection: '+', '.join(dates)+' (Australia/Sydney).**','200 saved observations; 199 completed answers. Names are recorded appearances, including host salons and cautionary mentions. They are not ratings or endorsements.','## Completed answers and names',table(['Engine','Completed','Failed','Blocked','No Overview','No shortlist / completed','Name appearances','Normalised names','Exact names'],[[LABELS[e],c['answered'],c['failed'],c['blocked'],c['no_overview'],f"{c['no_business_answers']}/{c['answered']}",c['named_business_occurrences'],c['unique_normalised_names'],c['unique_exact_names']] for e,c in counts.items()]),'## Most-named businesses (all first-place ties)',table(['Engine','Recorded spellings','Answers / completed'],[[LABELS[e],' / '.join(b['spellings']),f"{b['answer_count']}/{b['answered_denominator']}"] for e in ENGINES for b in freqs[e] if b['answer_count']==freqs[e][0]['answer_count']]),'Full exact and normalised frequencies are separate in analysis.json. No fuzzy matching or branch merging.','## Agreement on the same question',table(['Engine pair','Shared-name questions / paired completed answers'],[[LABELS[p['a']]+' + '+LABELS[p['b']],f"{p['shared']}/{p['paired']}"] for p in pairs]),f"All four shared a normalised name on {result['four_engine_shared']}/{len(four)} paired questions. Empty completed answers remain in this denominator.",'## Source-link categories','Each cell counts answers with at least one eligible link of this type. Categories overlap. Google expanded cards count; ChatGPT and Perplexity panel-only destinations do not.',table(['Source type']+[LABELS[e] for e in ENGINES],[[c]+[f"{sources[e][c]}/{counts[e]['answered']}" for e in ENGINES] for c in CATS]),'## Own-site matches: confirmed lower bounds','Only explicit full-name labels and reviewed pilot associations enter this conservative pass. Shortened or unlabelled links remain unresolved even where a person could recognise the domain. Capture formats differ, so do not rank engines by these rates.',table(['Engine','Own website / appearances','Directory / appearances','Booking / appearances','Own site unresolved / appearances'],[[LABELS[e]]+[f"{own[e][k]}/{own[e]['appearance_denominator']}" for k in ['own_website_match','directory_match','booking_match','own_website_unresolved']] for e in ENGINES]),'## Frequently linked domains']
    totals=Counter({d:sum(domains[e].get(d,0) for e in ENGINES) for d in {d for dd in domains.values() for d in dd}})
    md.append(table(['Domain']+[LABELS[e] for e in ENGINES],[[d]+[f"{domains[e].get(d,0)}/{counts[e]['answered']}" for e in ENGINES] for d,_ in sorted(totals.items(),key=lambda x:(-x[1],x[0]))[:15]]))
    for field in groups:
        md += ['## Answers by '+field.replace('_',' '),'Name appearances / completed answers; prompt locations, not verified business service areas.',table([field]+[LABELS[e] for e in ENGINES],[[label]+[f"{groups[field][e][label]['name_appearances']} / {groups[field][e][label]['completed']}" for e in ENGINES] for label in groups[field][ENGINES[0]]])]
    md += ['## Failures and exclusions',table(['Engine','Question','Status','Reason'],[[r['engine'],r['question_id'],r['status'],r['reason']] for r in result['exclusions']]),'No missing records. Coding and location conflicts are in coding-review.md. Original raw files are unchanged.','## Reproduction','Run `python3 analyse.py`. The public bundle repeats arithmetic from sanitized inputs and cannot verify omitted original raw captures. See verification.md for byte comparisons.']
    (ROOT/'results.md').write_text('\n\n'.join(md)+'\n')
    (ROOT/'coding-review.md').write_text('# Coding decisions: Sydney beauty index\n\nAll 200 records have coding-review.json rows. Original answer/raw files are unchanged. Corrections apply in coded-answers.json and the derived tables.\n\n'+table(['Answer','Original names','Reviewed names','Decision'],[[r['answer_key'],'; '.join(r['before']),'; '.join(r['after']),' '.join(r['decisions'])] for r in edge])+'\n')
    print(json.dumps({'completed':len(ok),'counts':counts,'pairs':[{k:v for k,v in p.items() if k!='details'} for p in pairs],'four_way':[result['four_engine_shared'],len(four)]},indent=2))

if __name__=='__main__':main()