#!/usr/bin/env python3
"""Join frozen site-purpose labels with v2.0; produce a local review and audit pack."""
import argparse
from collections import Counter, defaultdict
import csv
import html
import json
from pathlib import Path
import shutil
from freeze import LABELS, BASES, csv_rows, dump, load, sha
BASE_SHA = {
'sites.csv': 'da47b5ae060c1a67f5131b9f40e413416f9d7a3f010456c5809316efca36dcd9',
'metrics.json': 'edbdec1742dc52d3e305b9e6fdb11c56163976561255cab6c9563a6ea147f1cc',
}
def write_csv(path, rows):
if not rows:
raise ValueError('Cannot publish a headerless CSV')
with Path(path).open('w', encoding='utf-8', newline='') as f:
w = csv.DictWriter(f, fieldnames=list(rows[0]), lineterminator='\n')
w.writeheader()
w.writerows({k: json.dumps(v, ensure_ascii=False) if isinstance(v, (list, dict)) else v
for k, v in r.items()} for r in rows)
def category_summary(rows, keys):
result = []
for category, label in LABELS.items():
subset = [r for r in rows if r['category'] == category]
valid = [r for r in subset if r['robots_analyzable'] == 1]
count = sum(r['ai_root_disallow'] for r in valid)
comparable = len(valid) >= 20 and category not in ('unknown', 'other_known')
reason = '' if comparable else ('heterogeneous_bucket' if category == 'other_known'
else 'unknown_type' if category == 'unknown' else 'analyzable_n_below_20')
result.append({
'category': category, 'label_zh': label, 'total': len(subset),
'robots_analyzable': len(valid), 'robots_unanalyzable': len(subset) - len(valid),
'ai_root_disallow': count,
'restriction_percent': round(100 * count / len(valid), 1) if comparable else None,
'rate_suppressed_reason': reason,
'llms_candidates': sum(r.get('llms_observation') == 'minimum_h1_candidate' for r in subset),
'llms_unknown': sum(r.get('llms_observation') == 'unknown' for r in subset),
'llms_candidate_percent_all_category': round(100 * sum(
r.get('llms_observation') == 'minimum_h1_candidate' for r in subset) / len(subset), 1)
if subset and category not in ('unknown', 'other_known') else None,
'borderline_n': sum(r['borderline_review_flag'] for r in subset),
})
return result
def alias_sensitivity(rows, keys):
buckets = defaultdict(list)
for r in rows:
if r['robots_analyzable']:
buckets[r['robots_final_url']].append(r)
units, groups = [], []
for url, members in sorted(buckets.items()):
categories = {r['category'] for r in members}
vectors = {tuple(r[k + '_policy'] for k in keys) for r in members}
state = 'category_conflict' if len(categories) != 1 else 'policy_conflict' if len(vectors) != 1 else 'included'
if state == 'included':
units.append(members[0])
if len(members) > 1 or state != 'included':
groups.append({'robots_final_url': url, 'member_n': len(members),
'domains': [r['domain'] for r in members], 'categories': sorted(categories),
'policy_vector_n': len(vectors), 'state': state})
categories_summary = [{k: v for k, v in g.items() if not k.startswith('llms_')}
for g in category_summary(units, keys)]
return {
'unit': 'exact final robots URL among analyzable responses; not company or unique website',
'all_url_groups': len(buckets), 'included_url_groups': len(units),
'excluded_category_conflict_groups': sum(g['state'] == 'category_conflict' for g in groups),
'excluded_policy_conflict_groups': sum(g['state'] == 'policy_conflict' for g in groups),
'included_ai_root_disallow': sum(r['ai_root_disallow'] for r in units),
'llms_collapsed_rates_computed': False,
'categories': categories_summary,
}, groups
def review_html(rows, summary):
e = lambda v: html.escape(str(v), quote=True)
options = ''.join('' for k, v in LABELS.items())
bases = ''.join('' for k, v in BASES.items())
body = []
for r in rows:
source = e(r['source_url']) if r['source_url'] else '背景知识；未逐站验证当前首页'
source_note = r['source_checked_at'] or '无本轮页面验证时间'
body.append('
'''
def analysis_html(summary):
e = lambda v: html.escape(str(v))
tr = []
for g in summary['categories']:
pct = '仅列计数' if g['restriction_percent'] is None else str(g['restriction_percent']) + '%'
tr.append('
'''
def build(frozen, base, work, output):
freeze = load(frozen / 'classification-freeze.json')
if sha(frozen / 'classifications.json') != freeze['labels_sha256']:
raise ValueError('Frozen classifications changed')
for name, digest in freeze['inputs'].items():
if (work / name).is_file() and sha(work / name) != digest:
raise ValueError('Frozen decision input changed: ' + name)
for name, value in BASE_SHA.items():
if sha(base / name) != value:
raise ValueError('Base release changed: ' + name)
labels = load(frozen / 'classifications.json')
source = csv_rows(base / 'sites.csv')
sites = {r['domain']: r for r in source}
if (len(labels) != 1000 or len({r['domain'] for r in labels}) != 1000 or len(sites) != 1000
or {r['domain'] for r in labels} != set(sites)):
raise ValueError('Join requires an exact one-to-one cohort')
metrics = load(base / 'metrics.json')
keys = [b['key'] for b in metrics['bots'] if b['group'] != 'baseline']
rows = []
for r in labels:
s = sites[r['domain']]
if r['rank'] != int(s['rank']):
raise ValueError('Rank mismatch')
valid = s['robots_analyzable'] == '1'
rows.append(r | {'robots_analyzable': int(valid),
'ai_root_disallow': int(any(s[k + '_policy'] == 'disallow' for k in keys)) if valid else None,
'robots_final_url': s['robots_final_url'], 'llms_observation': s['llms_observation'],
**{k + '_policy': s[k + '_policy'] for k in keys}})
total_valid = sum(r['robots_analyzable'] for r in rows)
total_restricted = sum(r['ai_root_disallow'] for r in rows if r['robots_analyzable'])
candidates = sum(r['llms_observation'] == 'minimum_h1_candidate' for r in rows)
if (total_valid, total_restricted, candidates) != (854, 249, 156):
raise ValueError('Frozen overall metrics not reproduced')
alias, alias_groups = alias_sensitivity(rows, keys)
bases = {basis: category_summary([r for r in rows if r['basis'] == basis], keys) for basis in BASES}
followup_run = work / 'fetch-01/run.json'
if not followup_run.exists():
followup_run = work / 'followup-run.json'
summary = {
'version': 'site-types-full-v3', 'n': len(rows),
'assigned_n': sum(r['category'] != 'unknown' for r in rows),
'unknown_n': sum(r['category'] == 'unknown' for r in rows),
'unknown_domains': [r['domain'] for r in rows if r['category'] == 'unknown'],
'basis_counts': dict(sorted(Counter(r['basis'] for r in rows).items())),
'totals': {'robots_analyzable': total_valid, 'ai_root_disallow': total_restricted,
'llms_candidates': candidates, 'llms_unknown': sum(r['llms_observation'] == 'unknown' for r in rows)},
'categories': category_summary(rows, keys), 'basis_sensitivity': bases,
'exclude_recorded_borderline_sensitivity': category_summary([r for r in rows if not r['borderline_review_flag']], keys),
'alias_sensitivity': alias, 'classification_freeze': freeze,
'base_sha256': BASE_SHA, 'followup_homepage_requests': load(followup_run),
'paid_ai_calls': 0, 'new_robots_or_llms_requests': 0,
'limits': ['Knowledge-assisted purpose labels, not formal industry/ownership verification',
'Single reviewer with prior benchmark exposure; no independent validation or accuracy estimate',
'Frozen popularity cohort, not a representative random-web sample',
'Static archived content and later follow-up may show different regional destinations',
'Exact final robots URL sensitivity does not deduplicate companies',
'Heterogeneous other/unknown buckets and analyzable cells below 20 have counts only'],
}
bot_rows = []
for g in summary['categories']:
valid = [r for r in rows if r['category'] == g['category'] and r['robots_analyzable']]
for key in keys:
n = sum(r[key + '_policy'] == 'disallow' for r in valid)
bot_rows.append({'category': g['category'], 'bot': key, 'analyzable_n': len(valid),
'root_disallow_n': n, 'percent': round(100 * n / len(valid), 1)
if not g['rate_suppressed_reason'] else None})
rank_rows = []
for low, high in ((1, 100), (101, 500), (501, 1726)):
for category in LABELS:
members = [r for r in rows if low <= r['rank'] <= high and r['category'] == category]
valid = [r for r in members if r['robots_analyzable']]
rank_rows.append({'rank_min': low, 'rank_max': high, 'category': category, 'total': len(members),
'robots_analyzable': len(valid), 'ai_root_disallow': sum(r['ai_root_disallow'] for r in valid)})
output.mkdir(parents=True, exist_ok=False)
write_csv(output / 'classifications.csv', labels)
write_csv(output / 'classified-sites.csv', rows)
write_csv(output / 'category-summary.csv', summary['categories'])
write_csv(output / 'bot-category-summary.csv', bot_rows)
write_csv(output / 'rank-category-counts.csv', rank_rows)
if alias_groups:
write_csv(output / 'alias-groups.csv', alias_groups)
dump(output / 'summary.json', summary)
(output / 'review.html').write_text(review_html(labels, summary), encoding='utf-8')
(output / 'analysis.html').write_text(analysis_html(summary), encoding='utf-8')
for name in ('classifications.json', 'classification-freeze.json'):
shutil.copyfile(frozen / name, output / name)
for name in ('knowledge-pass.json', 'archive-followup-decisions.json', 'quality-pass.json', 'source-register.json'):
shutil.copyfile(work / name, output / name)
shutil.copyfile(followup_run, output / 'followup-run.json')
return summary
if __name__ == '__main__':
p = argparse.ArgumentParser(description=__doc__)
for name in ('frozen', 'base', 'work', 'output'):
p.add_argument('--' + name, required=True, type=Path)
result = build(**vars(p.parse_args()))
print(json.dumps({'assigned': result['assigned_n'], 'unknown': result['unknown_n'],
'alias_units': result['alias_sensitivity']['included_url_groups']}))