"""Extraction and statistics, second implementation (2 September 2026, after gate 2). Reads the frozen files only: discovery rows
(steps 1 and 2, with SERP file links from freeze.py), the step 3 probe manifest (probe2.py) and the capture manifest (hashed by
freeze.py). For every profile: count and rating from the page's own JSON-LD aggregateRating (Trustpilot: businessUnit JSON), the
canonical URL parsed order-independently, the listing name from the JSON-LD product object, the SHA-256 of the saved page, and the
sub-ratings the page displays as numbers (Capterra: one decimal; Software Advice: two decimals; GetApp shows sub-ratings as star
glyphs only, recorded as label-only; G2 and Trustpilot display no comparable labels). Eligibility for the signature test requires
both a count and a one-decimal rating. Statistics follow SCOPE.md (T_p, D_p, R_p, median over comparable products, pooled share,
coverage, sub-10 sensitivity that never changes F) plus the corroboration check required by gate 2: within each exact matching
group, labels shared by two or more members after case and whitespace normalisation; corroborated only when every shared numeric
value matches at one decimal (a two-decimal value is rounded half up; a second decimal of exactly 5 makes that label a tie and the
group unassessable); discordant when any shared value differs; unassessable when fewer than two shared numeric labels exist.
Usage: python3 scripts/extract2.py cohort1 2026-09-02"""
import json,re,statistics,os,sys,hashlib
from collections import defaultdict
from decimal import Decimal, ROUND_HALF_UP
cohort,date=sys.argv[1],sys.argv[2]
d=json.load(open(f'data/profiles-{cohort}-{date}.json')); base=d['rows'] if isinstance(d,dict) else d
probe=json.load(open(f'data/profiles-probe-{cohort}-{date}.json')) if os.path.exists(f'data/profiles-probe-{cohort}-{date}.json') else []
cap=json.load(open(f'data/capture-{cohort}-{date}.json')); cap=cap['rows'] if isinstance(cap,dict) else cap
capidx={(r['product'],r['site']):r for r in cap}
supp=json.load(open(f'data/supplement-{cohort}-{date}.json')) if os.path.exists(f'data/supplement-{cohort}-{date}.json') else []
SITES=('Capterra','GetApp','Software Advice','G2','Trustpilot')
def canonical(h):
    for tag in re.findall(r'<link[^>]+>',h[:400000]):
        if re.search(r'rel=["\']canonical["\']',tag):
            m=re.search(r'href=["\']([^"\']+)["\']',tag)
            if m: return m.group(1)
    return None
def jsonld_product(h):
    for blk in re.findall(r'<script[^>]*type="application/ld\+json"[^>]*>(.*?)</script>',h,re.S):
        try: o=json.loads(blk)
        except Exception: continue
        st=[o]
        while st:
            x=st.pop()
            if isinstance(x,dict):
                if x.get('@type') in ('SoftwareApplication','Product') and 'aggregateRating' in x: return x
                st+=[v for v in x.values() if isinstance(v,(dict,list))]
            elif isinstance(x,list): st+=x
    return None
def fetch_error(h):
    """Classify a saved body: ('page',None,None) for an HTML page; ('fetch-layer error response',<HTTP status>,<error type>) for a
    proxy error JSON such as {"StatusCode":502,"ErrorType":"ROTATION_FAILED",...}; ('empty body',None,None); ('non-HTML body',None,None)."""
    s=h.strip()
    if not s: return 'empty body',None,None
    if s.startswith('{'):
        try:
            j=json.loads(s)
            if isinstance(j,dict) and 'StatusCode' in j and ('ErrorType' in j or 'ErrorCode' in j):
                return 'fetch-layer error response',int(j['StatusCode']),str(j.get('ErrorType') or j.get('ErrorCode'))
        except Exception: pass
    if not re.search(r'(?i)<html|<!doctype|<head|<body',s[:5000]): return 'non-HTML body',None,None
    return 'page',None,None
HR_FAMILY=('hr','human resources','recruit','pre-employment','preemployment','assessment','talent','interview','reference check','diversity','employee')
GENERIC_SEGMENTS=('all-software','product')
def listing_category(site,url,h):
    """Category of a Gartner Digital Markets listing (amendment 6, 2 September 2026): Capterra = the breadcrumb category in the
    page's BreadcrumbList JSON-LD (second element); GetApp and Software Advice = the first URL path segment. Returns (category, ok)."""
    if site=='Capterra':
        cat=None
        for blk in re.findall(r'<script[^>]*type="application/ld\+json"[^>]*>(.*?)</script>',h,re.S):
            try: o=json.loads(blk)
            except Exception: continue
            st=[o]
            while st and cat is None:
                x=st.pop()
                if isinstance(x,dict):
                    if x.get('@type')=='BreadcrumbList':
                        els=x.get('itemListElement',[])
                        if len(els)>=2:
                            e=els[1]; cat=(e.get('item') or {}).get('name') if isinstance(e.get('item'),dict) else e.get('name')
                    st+=[v for v in x.values() if isinstance(v,(dict,list))]
                elif isinstance(x,list): st+=x
        if not cat: return None,False
        return cat,any(k in cat.lower() for k in HR_FAMILY)
    if site in ('GetApp','Software Advice'):
        parts=[p for p in url.split('/')[3:] if p]; seg=parts[0] if parts else None
        if not seg: return None,False
        return seg,(seg in GENERIC_SEGMENTS or any(k in seg.lower().replace('-',' ') for k in HR_FAMILY))
    return None,True
def one_dec(v):
    return float(Decimal(str(v)).quantize(Decimal('0.1'),rounding=ROUND_HALF_UP))
rows=[]
for r in base:
    if r['site'] not in SITES: continue
    c=capidx.get((r['product'],r['site']),{}); pr=next((p for p in probe if p['product']==r['product'] and p['site']==r['site']),None)
    if r['found'] and c.get('saved') and os.path.exists(c['saved']):
        x={'product':r['product'],'site':r['site'],'protocol_step':r.get('protocol_step'),'discovery_query':r.get('query'),'serp_file':r.get('serp_file_used'),'url_requested':r['profile_url'],'saved':c['saved'],'fetched_utc':c.get('fetched_utc')}
    elif pr and pr.get('found') and os.path.exists(pr['saved']):
        x={'product':r['product'],'site':r['site'],'protocol_step':'3 pattern probe','url_requested':pr['url'],'saved':pr['saved'],'fetched_utc':pr.get('fetched_utc'),'probe_attempts':len(pr.get('attempts',[]))}
    else:
        note=(pr or {}).get('note')
        x={'product':r['product'],'site':r['site'],'protocol_step':None,'saved':None,'status':('not found in steps 1 to 3' if not note else f'not found in steps 1 and 2; step 3: {note}'),'probe_attempts':len((pr or {}).get('attempts',[]))}
        rows.append(x); continue
    fb=open(x['saved'],'rb').read(); h=fb.decode('utf-8','ignore'); x['sha256']=hashlib.sha256(fb).hexdigest(); x['bytes']=len(fb)
    # gate 6 correction, 2 September 2026: a saved body that is a fetch-layer error response (proxy JSON with StatusCode and
    # ErrorType), empty, or not HTML is a capture failure, never a profile observation; it keeps the discovery result and the
    # saved file's hash, carries the HTTP status and error type, and is labelled 'capture unresolved'
    x['capture_status'],x['http_status'],x['error_type']=fetch_error(h)
    if x['capture_status']!='page':
        x.update({'canonical':None,'url':x['url_requested'],'subratings':{},'count':None,'rating_raw':None,'listing_name':None,'bins_counts_1to5':None,'source_field':None,'rating_display':None,'listing_key':None,'eligible':False,
                  'status':f"profile found in discovery; capture unresolved ({x['http_status'] if x['http_status'] else x['capture_status']})"})
        rows.append(x); continue
    x['canonical']=canonical(h); x['url']=x['canonical'] or x['url_requested']
    # amendment 6, 2 September 2026 (after the live gate): a Gartner Digital Markets listing outside the HR category family is a
    # different product or product line; the row keeps the saved page and its hash, is labelled as rejected and counts as not found
    x['listing_category'],x['category_ok']=listing_category(x['site'],x['url'],h)
    if not x['category_ok']:
        x.update({'subratings':{},'count':None,'rating_raw':None,'listing_name':None,'bins_counts_1to5':None,'source_field':None,'rating_display':None,'listing_key':None,'eligible':False,
                  'status':f"listing rejected: category '{x['listing_category']}' outside the HR family; treated as not found in steps 1 to 3"})
        rows.append(x); continue
    x['subratings']={}
    if x['site']=='Trustpilot':
        m=re.search(r'"numberOfReviews":(\d+),"numberOfReviewsLast12Months":\d+,"trustScore":([\d.]+)',h); b=re.search(r'"ratings":\{"total":(\d+),"one":(\d+),"two":(\d+),"three":(\d+),"four":(\d+),"five":(\d+)\}',h)
        nm=re.search(r'"displayName":"([^"]+)"',h)
        x['count']=int(m.group(1)) if m else None; x['rating_raw']=float(m.group(2)) if m else None; x['listing_name']=nm.group(1).strip() if nm else None
        x['bins_counts_1to5']=[int(b.group(i)) for i in range(2,7)] if b else None; x['source_field']='businessUnit.numberOfReviews / trustScore / ratings'
    else:
        o=jsonld_product(h); ag=(o or {}).get('aggregateRating',{}) if o else {}
        cnt=ag.get('reviewCount') or ag.get('ratingCount'); rv=ag.get('ratingValue')
        if cnt is None:
            m=re.search(r'"aggregateRating":\{[^}]*\}',h); j=m.group(0) if m else ''
            mc=re.search(r'"(?:reviewCount|ratingCount)":"?(\d+)"?',j); mr=re.search(r'"ratingValue":"?([\d.]+)"?',j)
            cnt=mc.group(1) if mc else None; rv=mr.group(1) if mr else None
        x['count']=int(str(cnt).replace(',','')) if cnt is not None else None
        try: x['rating_raw']=float(rv) if rv is not None else None
        except Exception: x['rating_raw']=None
        x['listing_name']=(o or {}).get('name') if o else None
        if not x['listing_name']:
            for blk in re.findall(r'<script[^>]*type="application/ld\+json"[^>]*>(.*?)</script>',h,re.S):
                i=blk.find('"aggregateRating"')
                if i>=0:
                    names=re.findall(r'"name":\s*"([^"]{1,120})"',blk[:i])
                    if names: x['listing_name']=names[-1].strip(); break
        x['source_field']='JSON-LD aggregateRating'
        if x['site'] in ('Capterra','Software Advice'):
            # overall sub-ratings live on the product page; use the captured page when it is the product page, else the supplement
            sup=next((s for s in supp if s['product']==x['product'] and s['site']==x['site'] and s.get('saved')),None)
            # read the captured page and, where one exists, the supplement page (the counterpart product or reviews page); labels merge, first value wins
            src=h+('\n<!-- SUPPLEMENT -->\n'+open(sup['saved'],encoding='utf-8',errors='ignore').read() if sup else '')
            x['subratings_source']='captured page'+(f" + {sup['saved']}" if sup else '')
            if x['site']=='Capterra':
                # three Capterra templates seen on 2 September 2026: (a) product page with tabular-nums values, (b) older product page with
                # semibold label spans followed by the value text node, (c) reviews-page header with sr2r3oj value spans
                for lab,val in re.findall(r'<span>(Ease of Use|Customer Service|Features|Value for Money)</span><span class="font-medium tabular-nums">([\d.]+)</span>',src):
                    x['subratings'].setdefault(lab,val)
                for lab in dict.fromkeys(re.findall(r'<span class="text-typo-20 font-semibold text-neutral-95">(Ease of use|Ease of Use|Customer Service|Customer service|Features|Value for money|Value for Money)</span>',src)):
                    m=re.search(r'<span class="text-typo-20 font-semibold text-neutral-95">'+re.escape(lab)+r'</span>',src); seg=re.sub(r'<svg.*?</svg>','',src[m.end():m.end()+2500])
                    v=re.search(r'>\s*(\d\.\d)\s*<',seg)
                    if v: x['subratings'].setdefault(lab,v.group(1))
                for lab,val in re.findall(r'<span>(Ease of use|Ease of Use|Customer Service|Customer service|Features|Value for Money|Value for money)</span><div[^>]*>(?:(?!</div>).){0,400}?<span class="e1xzmg0z sr2r3oj">([\d.]+)\s*</span>',src,re.S):
                    x['subratings'].setdefault(lab,val)
            else:
                for val,lab in re.findall(r'<span class="text-sm font-bold text-grey-91">([\d.]+)</span><span class="ml-1 text-sm font-normal text-grey-91">(Ease of use|Value for money|Customer support|Functionality)</span>',src):
                    x['subratings'].setdefault(lab,val)
        elif x['site']=='GetApp':
            for lab in re.findall(r'data-testid="Rating-CriteriaItem"><span class="[^"]*">(Value for money|Ease of use|Features|Customer support)</span>',h):
                x['subratings'].setdefault(lab,'stars only')
    x['rating_display']=None if x.get('rating_raw') is None else one_dec(x['rating_raw'])
    x['listing_key']=re.sub(r'[^a-z0-9]+','',x['listing_name'].lower()) if x.get('listing_name') else None
    x['eligible']=bool(x.get('count')) and x.get('rating_display') is not None
    x['status']='captured' if x['eligible'] else ('profile found, 0 reviews displayed' if x.get('count')==0 else 'profile found, no displayed count or rating parsed')
    rows.append(x)
json.dump(rows,open(f'data/signatures-{cohort}-{date}.json','w'),indent=1)
def same_listing(a,b):
    ka,kb=a.get('listing_key') or '',b.get('listing_key') or ''
    return bool(ka and kb) and (ka==kb or ka in kb or kb in ka)
def corroborate(group):
    # labels normalised by case and whitespace only; numeric values only
    vals=defaultdict(dict)
    for r in group:
        for lab,v in (r.get('subratings') or {}).items():
            try: vals[re.sub(r'\s+',' ',lab.strip().lower())][r['site']]=Decimal(str(v))
            except Exception: pass
    shared={lab:sv for lab,sv in vals.items() if len(sv)>=2}
    if len(shared)<2: return 'unassessable', {lab:{s:str(v) for s,v in sv.items()} for lab,sv in shared.items()}, 'fewer than two shared numeric labels'
    verdict='corroborated'; notes=[]
    for lab,sv in shared.items():
        disp=set()
        for s,v in sv.items():
            if v.as_tuple().exponent<=-2 and str(v)[-1]=='5': return 'unassessable', {lab:{s:str(v) for s,v in sv.items()} for lab,sv in shared.items()}, f'rounding tie on {lab}'
            disp.add(v.quantize(Decimal('0.1'),rounding=ROUND_HALF_UP))
        if len(disp)>1: verdict='discordant'; notes.append(lab)
    return verdict, {lab:{s:str(v) for s,v in sv.items()} for lab,sv in shared.items()}, ('differs on '+', '.join(notes) if notes else 'all shared values match at one decimal')
products=list(dict.fromkeys(r['product'] for r in rows)); N=len(products)
def stats(minc):
    prods=[]
    for p in products:
        found=[r for r in rows if r['product']==p and r.get('saved') and r.get('category_ok',True)]
        el=[r for r in found if r.get('eligible') and r['count']>=minc]
        T=sum(r['count'] for r in el); g=defaultdict(list)
        for r in el: g[(r['count'],r['rating_display'])].append(r)
        groups=[]
        for v in g.values():
            if len(v)<2: continue
            clusters=[]
            for r in v:
                for c in clusters:
                    if all(same_listing(r,q) for q in c): c.append(r); break
                else: clusters.append([r])
            groups+=[c for c in clusters if len(c)>=2]
        D=sum(sum(r['count'] for r in v)-max(r['count'] for r in v) for v in groups)
        cor=[corroborate(v) for v in groups]
        prods.append({'product':p,'found':len(found),'eligible':len(el),'T':T,'D':D,'R':(D/T if T else None),'groups':[[r['site'] for r in v] for v in groups],'corroboration':[c[0] for c in cor],'corroboration_detail':[c for c in cor],'distinct_signatures':len(g),'listing_names':sorted({r.get('listing_name') or '?' for r in el})})
    F=sum(1 for q in prods if q['found']>=2); A=sum(1 for q in prods if q['eligible']>=2); B=sum(1 for q in prods if q['groups'])
    comp=[q for q in prods if q['eligible']>=2]
    medR=statistics.median([q['R'] for q in comp]) if comp else None
    pooled=(sum(q['D'] for q in comp)/sum(q['T'] for q in comp)) if comp and sum(q['T'] for q in comp) else None
    medG=statistics.median([q['distinct_signatures'] for q in comp]) if comp else None
    comp_sites={s:sum(1 for q in prods for v in q['groups'] if s in v) for s in SITES}
    corr={k:sum(1 for q in prods for c in q['corroboration'] if c==k) for k in ('corroborated','discordant','unassessable')}
    return prods,F,A,B,medR,pooled,medG,comp_sites,corr
main=stats(0); sens=stats(10)
if cohort=='appendix':
    # appendix products enter no denominator: row observations only, with the per-row R_p that DECISION.md permits
    prods=main[0]; out=[f'# Row observations, appendix, {date} (computed by scripts/extract2.py from data/signatures-{cohort}-{date}.json; no denominators, no headline, no sensitivity run)','',
        '| Product | Found | Eligible | T_p | D_p | R_p | Matching group | Corroboration | Listing names as displayed |','|---|---|---|---|---|---|---|---|---|']
    for q in prods: out.append(f"| {q['product']} | {q['found']} | {q['eligible']} | {q['T']} | {q['D']} | {('%.3f'%q['R']) if q['R'] is not None else 'n/a'} | {'; '.join(', '.join(v) for v in q['groups']) or 'none'} | {'; '.join(f'{c[0]} ({c[2]})' for c in q['corroboration_detail']) or 'n/a'} | {'; '.join(q['listing_names'])} |")
    out+=['','## Displayed count / one-decimal rating per profile (canonical URL in the dataset)','','| Product | Capterra | GetApp | Software Advice | G2 | Trustpilot |','|---|---|---|---|---|---|']
    for p in products:
        pr={r['site']:r for r in rows if r['product']==p}
        cell=lambda s: (f"{pr[s]['count']} / {pr[s]['rating_display']}" if pr[s].get('count') is not None and pr[s].get('rating_display') is not None else (f"{pr[s]['count']} / no rating" if pr[s].get('count') is not None else pr[s].get('status','')))
        out.append(f"| {p} | {cell('Capterra')} | {cell('GetApp')} | {cell('Software Advice')} | {cell('G2')} | {cell('Trustpilot')} |")
    open(f's3-results-{cohort}-{date}.md','w').write('\n'.join(out)+'\n'); print('\n'.join(out[:8])); sys.exit(0)
out=[f'# Results, {cohort}, {date} (computed by scripts/extract2.py from data/signatures-{cohort}-{date}.json)','']
for label,(prods,F,A,B,medR,pooled,medG,cs,corr) in (('Primary (all found profiles)',main),('Sensitivity (profiles with fewer than 10 displayed reviews excluded from eligibility; F unchanged by construction)',sens)):
    out+=[f"## {label}",f"- F, products with profiles found on at least two sites: {F} of {N}",f"- A, comparable products (at least two signature-eligible profiles): {A} of {N}",
          f"- B, comparable products with at least one exact matching signature group across two or more sites: {B} of {A}",
          f"- Median R_p across the A comparable products (zero for no match): {medR if medR is None else round(medR,3)}",f"- Pooled sum(D_p) / sum(T_p): {pooled if pooled is None else round(pooled,3)}",
          f"- Median number of distinct displayed signatures per comparable product: {medG}",
          f"- Matching groups by site: "+', '.join(f"{s} {cs[s]}" for s in SITES),
          f"- Corroboration of matching groups by displayed sub-ratings (labels shared after case and whitespace normalisation, one-decimal match): corroborated {corr['corroborated']}, discordant {corr['discordant']}, unassessable {corr['unassessable']}",'',
          '| Product | Found | Eligible | T_p | D_p | R_p | Matching group | Corroboration | Listing names as displayed |','|---|---|---|---|---|---|---|---|---|']
    for q in prods: out.append(f"| {q['product']} | {q['found']} | {q['eligible']} | {q['T']} | {q['D']} | {('%.3f'%q['R']) if q['R'] is not None else 'n/a'} | {'; '.join(', '.join(v) for v in q['groups']) or 'none'} | {'; '.join(f'{c[0]} ({c[2]})' for c in q['corroboration_detail']) or 'n/a'} | {'; '.join(q['listing_names'])} |")
    out.append('')
prods,F,A,B,medR,pooled,medG,cs,corr=main
out+=(['## Headline sentences the numbers support (site as actor; wording fixed by gate 2 r2)','',
      f'"For {B} of {A} comparable products among the {N} Capterra-listed pure-play assessment products, two or more review sites displayed at least one exact matching review signature on 2 September 2026; the median R_p across the {A} comparable products was {round(medR,3) if medR is not None else "n/a"}."','',
      (f'"Capterra and GetApp each displayed a profile in all {B} matching groups, Software Advice displayed a profile in {cs["Software Advice"]}, and G2 and Trustpilot displayed profiles in none."' if (cs["Capterra"]==B and cs["GetApp"]==B and cs["G2"]==0 and cs["Trustpilot"]==0) else f'"Capterra displayed a profile in {cs["Capterra"]} of the {B} matching groups, GetApp in {cs["GetApp"]}, Software Advice in {cs["Software Advice"]}, G2 in {cs["G2"]} and Trustpilot in {cs["Trustpilot"]}."'),'',
      f'"Across those {B} matching groups, the review sites displayed sub-ratings that yielded {corr["corroborated"]} corroborated classifications, {corr["discordant"]} discordant classifications, and {corr["unassessable"]} unassessable classifications under the prespecified check; none establishes review identity."',''] if cohort=='cohort1' else ['## Sensitivity sample: no headline (SCOPE.md: cohorts 2 and 3 and the appendix never enter a headline)',''])+[
      '## Displayed count / one-decimal rating per profile (canonical URL in the dataset)','','| Product | Capterra | GetApp | Software Advice | G2 | Trustpilot |','|---|---|---|---|---|---|']
for p in products:
    pr={r['site']:r for r in rows if r['product']==p}
    cell=lambda s: (f"{pr[s]['count']} / {pr[s]['rating_display']}" if pr[s].get('count') is not None and pr[s].get('rating_display') is not None else (f"{pr[s]['count']} / no rating" if pr[s].get('count') is not None else pr[s].get('status','')))
    out.append(f"| {p} | {cell('Capterra')} | {cell('GetApp')} | {cell('Software Advice')} | {cell('G2')} | {cell('Trustpilot')} |")
out+=['','Sub-ratings as displayed (numeric only; GetApp shows star glyphs without numbers):']
for r in rows:
    if r.get('subratings') and any(v!='stars only' for v in r['subratings'].values()): out.append(f"- {r['product']} / {r['site']}: "+', '.join(f'{k} {v}' for k,v in r['subratings'].items()))
out+=['','Trustpilot embedded star counts (1 to 5 stars) where a profile was found:']
for r in rows:
    if r['site']=='Trustpilot' and r.get('bins_counts_1to5'): out.append(f"- {r['product']}: {r['bins_counts_1to5']} (total {r['count']}, TrustScore {r['rating_raw']})")
open(f's3-results-{cohort}-{date}.md','w').write('\n'.join(out)+'\n'); print('\n'.join(out[:40]))