#!/usr/bin/env python3
"""Offline, hash-verified v2 analysis. No network, DB, or paid API access."""
import argparse
import base64
from collections import Counter, defaultdict
import csv
from decimal import Decimal, ROUND_HALF_UP
import gzip
import hashlib
import json
import io
from pathlib import Path
import re
import statistics

from policy import classify, parse, adult_label_candidate
from collect import read_body
from types import SimpleNamespace


def dump(path, data):
    Path(path).write_text(json.dumps(data, ensure_ascii=False, indent=2, sort_keys=True) + '\n', encoding='utf-8')


def sha(data):
    return hashlib.sha256(data).hexdigest()


def metric(n, d):
    return {'numerator': n, 'denominator': d, 'percent':
            str((Decimal(n)*100/Decimal(d)).quantize(Decimal('.1'), rounding=ROUND_HALF_UP)) if d else None}


def read_safe(root, relative):
    p = (root / relative).resolve()
    if not p.is_relative_to(root.resolve()):
        raise ValueError('Evidence path outside run')
    if not p.exists() and (root / 'body-aliases.json').exists():
        aliases = json.loads((root / 'body-aliases.json').read_text())
        target = aliases.get(relative)
        if target:
            p = (root / target).resolve()
            if not p.is_relative_to(root.resolve()):
                raise ValueError('Aliased evidence path outside run')
    return p.read_bytes()


def verified_records(root):
    run = json.loads((root / 'run.json').read_text())
    if 'excluded_from_analysis' in run or not run.get('ended_at') or run.get('errors'):
        raise ValueError('Incomplete or excluded run')
    cohort_bytes = (root / 'cohort.csv').read_bytes()
    if sha(cohort_bytes) != run['cohort_sha256']:
        raise ValueError('Cohort hash mismatch')
    bots_bytes = (root / 'bots.json').read_bytes()
    if sha(bots_bytes) != run['bots_sha256']:
        raise ValueError('Bot registry hash mismatch')
    bots = json.loads(bots_bytes)
    cohort = list(csv.DictReader(cohort_bytes.decode().splitlines()))
    expected = {r['domain']: int(r['rank']) for r in cohort}
    if len(expected) != len(cohort) or len(cohort) != run['cohort_count'] or run['completed'] != len(cohort):
        raise ValueError('Incomplete or duplicate cohort')
    if {p.stem for p in (root / 'records').glob('*.json')} != set(expected):
        raise ValueError('Missing/extra domain records')
    results, attempts, requests = [], [], 0
    for domain, rank in expected.items():
        result = json.loads((root / 'records' / (domain + '.json')).read_text())
        if result['domain'] != domain or result['rank'] != rank:
            raise ValueError('Record identity mismatch')
        history = [json.loads(x) for x in (root / 'attempts' / (domain + '.jsonl')).read_text().splitlines()]
        for a in history:
            if a['domain'] != domain or a['request_url'] != f'https://{domain}/' + {
                    'robots': 'robots.txt', 'llms': 'llms.txt', 'home': ''}[a['resource']]:
                raise ValueError('Unexpected primary request')
            for prefix in (() if a['resource']=='home' and run.get('public_home_metadata_only') else ('body', 'wire')):
                body = read_safe(root, a[prefix + '_path'])
                if sha(body) != a[prefix + '_sha256']:
                    raise ValueError('Body hash mismatch: ' + a[prefix + '_path'])
                if len(body) != a['bytes' if prefix == 'body' else 'wire_bytes']:
                    raise ValueError('Body length mismatch')
            if a['resource']!='home' and a.get('content_encoding') in ('gzip','deflate'):
                wire=read_safe(root,a['wire_path'])
                decoded,_,_,decode_error=read_body(SimpleNamespace(read=io.BytesIO(wire).read,
                    headers={'Content-Encoding':a['content_encoding']}),run['max_decoded_bytes'])
                if decoded != read_safe(root,a['body_path']) or (not a['truncated'] and decode_error!=a['decode_error']):
                    raise ValueError('Compressed wire/body representation mismatch')
            if a['ended_at'] < a['started_at']:
                raise ValueError('Invalid attempt timestamps')
            requests += a['request_count']
        for resource in run['resources']:
            selected = [a for a in history if a['resource'] == resource]
            if not selected or [a['number'] for a in selected] != list(range(1, len(selected)+1)):
                raise ValueError('Missing/nonsequential attempts')
            a = selected[-1]
            if result[resource]['selected_attempt'] != a['number']:
                raise ValueError('Selection is not last attempt')
            for key in ('status', 'final_url', 'body_path'):
                if result[resource][key] != a[key]:
                    raise ValueError('Selected response metadata mismatch: '+domain+'/'+resource+'/'+key)
            if resource != 'home':
                for key in ('body_sha256', 'content_type'):
                    if result[resource][key] != a[key]:
                        raise ValueError('Selected response metadata mismatch: '+domain+'/'+resource+'/'+key)
                recalculated = classify(a, read_safe(root, a['body_path']), bots)
                differences = {k: {'collected': result[resource].get(k), 'analyzed': v}
                               for k, v in recalculated.items() if result[resource].get(k) != v}
                # The original run stays immutable. A post-collection parser revision
                # applies to ALL raw bodies, and its differences are published.
                if differences:
                    result.setdefault('analysis_reclassification', {})[resource] = differences
                result[resource].update(recalculated)
            elif result['home']['body_path'] != a['body_path']:
                raise ValueError('Homepage provenance mismatch')
        attempts.extend(history)
        if run.get('public_home_metadata_only'):
            result['_home_metadata_only'] = True
        results.append(result)
    if requests != run['request_count']:
        raise ValueError('Request count mismatch')
    return run, bots, results, attempts


def llms_observation(row):
    r = row['llms']
    if r['minimum_h1_candidate']:
        return 'minimum_h1_candidate'
    if r['fetch_state'] == 'not_found':
        return 'root_not_found'
    if r['fetch_state'] == 'success' and r['body_kind'] in (
            'html_or_challenge', 'error_text', 'nonconforming_text', 'empty_or_comments'):
        return 'observed_non_candidate'
    return 'unknown'


def home_label(row, root):
    """Observation labels only; these are NOT verified business categories."""
    h = row.get('home', {})
    if row.get('_home_metadata_only'):
        # Public replication can reproduce the counts of supplied annotations;
        # it cannot independently inspect the unredistributed homepage HTML.
        return h['observation']
    if not h:
        return 'not_collected'
    if h['error'] or h['status'] == 0:
        return 'unreachable_review'
    if h['status'] in (401, 403, 429):
        return 'walled_review'
    if not 200 <= h['status'] < 300:
        return 'http_status_review'
    if not h['same_authority']:
        return 'off_domain_redirect_review'
    text = read_safe(root, h['body_path']).decode('utf-8', 'replace')
    if not re.search(r'<(?:html|!doctype\s+html|head|body)\b', text[:4000], re.I):
        return 'non_html_review'
    if re.search(r'captcha|just a moment|checking your browser|access denied', h['title'], re.I):
        return 'challenge_review'
    if re.search(r'domain (?:is )?for sale|buy this domain|parked domain', h['title'], re.I):
        return 'parked_candidate_review'
    if re.search(r'^\s*(?:sign in|log in|login|登录|登入)(?:\s|$|[|—-])', h['title'], re.I):
        return 'login_candidate_review'
    if not h['title'] or len(text) < 2000:
        return 'thin_review'
    return 'content_candidate_unverified'


def restricted(row, keys):
    return any(row['robots']['policies'][key]['policy'] == 'disallow' for key in keys)


def signal_counts(rows, root):
    result = {}
    for token in ['__usage_ai-training', 'content-signal', 'claude-web', 'anthropic-ai']:
        present, valid, examples = 0, 0, []
        for r in rows:
            if r['robots']['fetch_state'] != 'success':
                continue
            body = read_safe(root, r['robots']['body_path']).decode('utf-8-sig', 'replace')
            if token in body.lower():
                present += 1
            if not r['robots']['analyzable']:
                continue
            groups, _ = parse(body)
            exists = any(token in g['agents'] for g in groups) if token != 'content-signal' else bool(
                re.search(r'^\s*content-signal\s*:\s*(?:[a-z-]+\s*=\s*(?:yes|no))(?:\s*,\s*[a-z-]+\s*=\s*(?:yes|no))*\s*(?:#.*)?$', body, re.I | re.M))
            if exists:
                valid += 1
                examples.append(r['domain'])
        result[token] = {'string_present_success_responses': present, 'syntactic_record_analyzable': valid,
                         'domains_with_syntactic_record': examples,
                         'provider_effect': 'not_evaluated'}
    return result


def summarize(rows, bots):
    keys = [b['key'] for b in bots if b['group'] != 'baseline']
    subset = [r for r in rows if r['robots']['analyzable']]
    n = len(subset)
    m = {'version': 'v2.0', 'scope': 'declared root path policies, fixed filtered domain cohort',
         'sample': {'total': len(rows), 'analyzable': n, 'outside_subset': len(rows)-n,
                    'rank_min': min(r['rank'] for r in rows), 'rank_max': max(r['rank'] for r in rows)},
         'ai_token_count': len(keys), 'all_token_count': len(bots),
         'ai_any': metric(sum(restricted(r, keys) for r in subset), n),
         'fetch_states': dict(sorted(Counter(r['robots']['fetch_state'] for r in rows).items())),
         'body_kinds_success': dict(sorted(Counter(r['robots']['body_kind'] for r in rows if r['robots']['fetch_state']=='success').items())),
         'llms_observations': dict(sorted(Counter(llms_observation(r) for r in rows).items())),
         'llms_candidates_all_cohort': metric(sum(r['llms']['minimum_h1_candidate'] for r in rows), len(rows)),
         'bots': [b | {'restricted': metric(sum(restricted(r, [b['key']]) for r in subset), n)} for b in bots],
         'distribution': dict(sorted(Counter(sum(restricted(r, [k]) for k in keys) for r in subset).items())),
         'home_observations': dict(sorted(Counter(r.get('home_observation', 'not_collected') for r in rows).items()))}
    m['ai_named'] = sum(any(r['robots']['policies'][k]['policy']=='disallow' and r['robots']['policies'][k]['source']=='named' for k in keys) for r in subset)
    m['ai_wildcard_only'] = m['ai_any']['numerator'] - m['ai_named']
    pairs = []
    for first, second in [('gptbot','oai_searchbot'), ('claudebot','claude_searchbot')]:
        matrix = [{'first_restricted': a, 'second_restricted': b,
                   'count': sum(restricted(r,[first])==a and restricted(r,[second])==b for r in subset)}
                  for a,b in [(True,True),(True,False),(False,True),(False,False)]]
        pairs.append({'first':first,'second':second,'cells':matrix,
                      'first_restricted_second_not':metric(matrix[1]['count'],sum(restricted(r,[first]) for r in subset))})
    m['pairs'] = pairs
    m['llms_cross_tab'] = []
    for label in ['minimum_h1_candidate', 'observed_non_candidate', 'root_not_found', 'unknown']:
        group = [r for r in subset if llms_observation(r)==label]
        m['llms_cross_tab'].append({'observation':label, 'n':len(group),
                                    'ai_restricted':metric(sum(restricted(r,keys) for r in group),len(group))})
    m['rank_buckets'] = []
    for low,high in [(1,100),(101,500),(501,1000),(1001,m['sample']['rank_max'])]:
        group = [r for r in subset if low<=r['rank']<=high]
        m['rank_buckets'].append({'label':f'{low}–{high}','analyzable':len(group),
                                 'ai_any':metric(sum(restricted(r,keys) for r in group),len(group))})
    aliases = defaultdict(list)
    for r in subset:
        aliases[r['robots']['final_url']].append(r)
    consistent = [v for v in aliases.values() if len({tuple(r['robots']['policies'][b['key']]['policy']=='disallow' for b in bots) for r in v})==1]
    conflicts = [v for v in aliases.values() if v not in consistent]
    m['alias_sensitivity'] = {'grouping':'exact final robots URL; not companies',
        'domain_rate':m['ai_any'], 'unique_final_urls':len(aliases),
        'conflicting_final_urls':len(conflicts), 'conflicting_domains':[r['domain'] for v in conflicts for r in v],
        'consistent_url_rate':metric(sum(restricted(v[0],keys) for v in consistent),len(consistent)),
        'groups':[{'url':u,'domains':[r['domain'] for r in v]} for u,v in aliases.items() if len(v)>1]}
    return m


def write_csv(path, rows, bots):
    flat = []
    for r in rows:
        x = {'domain':r['domain'],'rank':r['rank']}
        for resource in ('robots','llms'):
            for k in ('status','fetch_state','body_kind','selected_attempt','final_url','body_sha256','content_type'):
                x[resource+'_'+k] = r[resource][k]
        x.update(robots_analyzable=int(r['robots']['analyzable']),
                 protocol_default=r['robots']['protocol_default'],
                 llms_observation=llms_observation(r),home_observation=r['home_observation'],
                 home_final_url=r.get('home',{}).get('final_url',''),
                 home_title=r.get('home',{}).get('title',''),
                 adult_label_review_flag=int(adult_label_candidate(r['domain'])))
        for b in bots:
            for k in ('policy','source'):
                x[b['key']+'_'+k]=r['robots']['policies'][b['key']][k]
        flat.append(x)
    with Path(path).open('w',encoding='utf-8',newline='') as f:
        writer=csv.DictWriter(f,fieldnames=list(flat[0]),lineterminator='\n')
        writer.writeheader();writer.writerows(flat)


def enrich(metrics, rows, bots, context):
    if not context.exists():
        return
    metrics.update(published='2026-11-09', modified=metrics['window']['ended_at'][:10],
                   dataset_published=metrics['window']['ended_at'][:10])
    for key, name in [('provenance','provenance.json'),('reference','reference-comparison.json'),
                      ('sample_review','sample-review.json')]:
        metrics[key]=json.loads((context/name).read_text())
    if metrics['reference']['compared'] != metrics['sample']['analyzable'] * len(bots):
        raise ValueError('Incomplete reference comparison')
    if metrics['provenance']['cohort_sha256'] != metrics['integrity']['cohort_sha256']:
        raise ValueError('Sample provenance mismatch')
    baseline=list(csv.DictReader((context/'legacy-sites.csv').open()))
    if sha((context/'legacy-sites.csv').read_bytes()) != metrics['provenance']['published_v11_csv_sha256']:
        raise ValueError('Historical baseline hash mismatch')
    if {r['domain'] for r in baseline} != {r['domain'] for r in rows}:
        raise ValueError('Historical comparison uses a different cohort')
    old={r['domain']:r for r in baseline if r['robots_parse_ok']=='1'}
    keys=[b['key'] for b in bots if b['group']!='baseline']
    old_blocked=lambda r:any(r.get(k)=='blocked' for k in keys)
    common=[r for r in rows if r['robots']['analyzable'] and r['domain'] in old]
    metrics['legacy_comparison']={'historical_analyzable':len(old),
        'historical_rate':metric(sum(old_blocked(r) for r in old.values()),len(old)),
        'common_analyzable':len(common),
        'common_historical_rate':metric(sum(old_blocked(old[r['domain']]) for r in common),len(common)),
        'common_new_rate':metric(sum(restricted(r,keys) for r in common),len(common)),
        'interpretation':'Both time and validation changed; not a causal policy-change estimate.'}


def main():
    parser=argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--run',required=True,type=Path)
    parser.add_argument('--output',required=True,type=Path)
    parser.add_argument('--context',type=Path)
    args=parser.parse_args()
    run,bots,rows,attempts=verified_records(args.run)
    for r in rows:
        r['home_observation']=home_label(r,args.run)
    metrics=summarize(rows,bots)
    metrics.update(window={'started_at':run['started_at'],'ended_at':run['ended_at']},
                   request_count=run['request_count'],attempt_count=len(attempts),
                   signals=signal_counts(rows,args.run),
                   collector_to_analysis_reclassifications={r['domain']:r['analysis_reclassification']
                        for r in rows if r.get('analysis_reclassification')},
                   integrity={'core_robots_llms_attempt_body_and_wire_hashes_verified':True,
                              'public_home_evidence':'response metadata and supplied observation labels; original HTML retained locally',
                              'cohort_sha256':run['cohort_sha256'],'bots_sha256':run['bots_sha256']})
    by_domain={r['domain']:r for r in rows}
    selected=[a for a in attempts if a['resource']=='robots' and
              by_domain[a['domain']]['robots']['selected_attempt']==a['number']]
    metrics['operational']={'redirected_selected_robots':sum(a['redirected'] for a in selected),
        'formal_https_to_http_robots':sum(a['final_url'].startswith('http://') for a in selected),
        'median_complete_analyzable_bytes':statistics.median(
            a['bytes'] for a in selected if by_domain[a['domain']]['robots']['analyzable'])}
    enrich(metrics,rows,bots,args.context or args.run/'context')
    args.output.mkdir(parents=True,exist_ok=False)
    dump(args.output/'metrics.json',metrics)
    write_csv(args.output/'sites.csv',rows,bots)
    dump(args.output/'records.json',rows)
    dump(args.output/'run.json',run)
    dump(args.output/'bots.json',bots)
    (args.output/'cohort.csv').write_bytes((args.run/'cohort.csv').read_bytes())
    for f in Path(__file__).parent.glob('*.py'):
        (args.output/f.name).write_bytes(f.read_bytes())
    (args.output/'method.md').write_bytes((Path(__file__).parent/'README.md').read_bytes())
    dump(args.output/'manifest.json',{'version':'v2.0','files':{
        f.name:sha(f.read_bytes()) for f in sorted(args.output.iterdir()) if f.is_file()}})
    print(json.dumps({'total':len(rows),'analyzable':metrics['sample']['analyzable'],
                      'ai_any':metrics['ai_any'],'output':str(args.output)}))


if __name__=='__main__':
    main()