| """Check published source identities plus split and exact-byte separation.""" |
| import hashlib |
| import json |
| import sys |
| from pathlib import Path |
|
|
|
|
| def identity_sets(rows): |
| result = {'asset': set(), 'path': set(), 'group': set(), 'sha256': set()} |
| for r in rows: |
| repo = r['repo'] |
| if r.get('asset_id'): result['asset'].add((repo, r['asset_id'])) |
| if r.get('path'): result['path'].add((repo, r['path'])) |
| for key in ('object_id', 'source_object_id'): |
| if r.get(key): result['group'].add((repo, r[key])) |
| if r.get('sha256'): result['sha256'].add(r['sha256']) |
| return result |
|
|
|
|
| def conflicts(row, blocked): |
| current = identity_sets([row]) |
| return any(current[k] & blocked[k] for k in blocked) |
|
|
|
|
| def audit(data, original): |
| data, original = Path(data), Path(original) |
| old = json.loads(original.read_text()) |
| new = json.loads((data / 'sources.json').read_text()) |
| blocked = identity_sets(old) |
| if any(conflicts(r, blocked) for r in new): |
| raise ValueError('v1 identity or mesh-byte overlap detected') |
| dev, test = [[json.loads(s) for s in (data / f).read_text().splitlines() if s.strip()] |
| for f in ('manifest.jsonl', 'test-manifest.jsonl')] |
| if not dev or not test: |
| raise ValueError('Both development and test data must be nonempty') |
| for rows in (dev, test): |
| for row in rows: |
| if conflicts(row['metadata'], blocked): |
| raise ValueError('Manifest includes a known v1 source') |
| for kind in ('image', 'target'): |
| p = (data / row[kind]).resolve() |
| if not p.is_relative_to(data.resolve()): |
| raise ValueError('Manifest path escapes dataset directory') |
| if hashlib.sha256(p.read_bytes()).hexdigest() != row[kind + '_sha256']: |
| raise ValueError('Cached file checksum mismatch') |
| for getter in (lambda r:r['metadata']['object_id'], lambda r:r['image_sha256'], lambda r:r['target_sha256']): |
| if {getter(r) for r in dev} & {getter(r) for r in test}: |
| raise ValueError('Development/test overlap detected') |
| report = {'known_v1_sources_excluded': len(old), 'new_mesh_assets':len(new), |
| 'development_samples':len(dev), 'test_samples':len(test), |
| 'test_object_groups':len({r['metadata']['object_id'] for r in test}), |
| 'checks':'source IDs, groups, source hashes, cache hashes, split groups and exact bytes', |
| 'limitation':'Semantic near duplicates and unknown upstream overlap can remain.'} |
| (data / 'v2-audit.json').write_text(json.dumps(report, indent=2)) |
| print(json.dumps(report, indent=2)) |
| return report |
|
|
| if __name__ == '__main__': |
| audit(*sys.argv[1:]) |
|
|
