tegwick
745d74abd9
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a070b5-4994-7271-bd8b-7c3dbcedec4b
117 lines
5.2 KiB
Python
117 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate migration assignments, optionally against the pinned donor Git blob."""
|
|
|
|
import argparse
|
|
from collections import Counter, defaultdict
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
def validate(ledger, inventory):
|
|
errors = []
|
|
expected = [entry['heading'] for entry in inventory['entries']]
|
|
entries = ledger['entries']
|
|
counts = Counter(entry['source_heading'] for entry in entries)
|
|
if len(set(expected)) != len(expected):
|
|
errors.append('Duplicate source inventory headings')
|
|
for heading in expected:
|
|
if counts[heading] != 1:
|
|
errors.append(f'{heading}: expected one mapping, found {counts[heading]}')
|
|
for heading in counts.keys() - set(expected):
|
|
errors.append(f'Unknown source heading: {heading}')
|
|
|
|
owners = defaultdict(set)
|
|
unowned = sum(counts[heading] == 0 for heading in expected)
|
|
target_count = 0
|
|
for entry in entries:
|
|
heading = entry['source_heading']
|
|
disposition = entry['disposition']
|
|
targets = entry['targets']
|
|
if disposition not in {'own', 'import', 'split', 'retire'}:
|
|
errors.append(f'{heading}: invalid disposition')
|
|
if disposition in {'own', 'import'} and len(targets) != 1:
|
|
errors.append(f'{heading}: own/import requires exactly one target')
|
|
if disposition == 'split' and len(targets) < 2:
|
|
errors.append(f'{heading}: split requires at least two targets')
|
|
if disposition == 'retire' and (targets or not entry.get('migration_note')):
|
|
errors.append(f'{heading}: retirement requires rationale and no targets')
|
|
if not targets and disposition != 'retire':
|
|
unowned += 1
|
|
if not entry.get('authority') or not entry.get('implementation_tasks'):
|
|
errors.append(f'{heading}: missing authority or implementation task')
|
|
if heading.startswith('Non-Canonical Convenience Term: ') and any(
|
|
target.get('kind') != 'convenience_term' for target in targets
|
|
):
|
|
errors.append(f'{heading}: convenience term promoted to canonical concept')
|
|
names = [target['concept'] for target in targets]
|
|
if len(set(names)) != len(names):
|
|
errors.append(f'{heading}: duplicate split targets')
|
|
target_count += len(targets)
|
|
|
|
all_targets = [target for entry in entries for target in entry['targets']]
|
|
all_targets += ledger.get('additional_required_concepts', [])
|
|
for target in all_targets:
|
|
name = target['concept']
|
|
owner = target.get('owner', {})
|
|
canon, model = owner.get('canon'), owner.get('model')
|
|
if not canon or not model:
|
|
unowned += 1
|
|
errors.append(f'{name}: missing owner')
|
|
continue
|
|
registration = ledger['model_registry'].get(model)
|
|
if not registration or registration['canon'] != canon:
|
|
errors.append(f'{name}: owner not in model registry')
|
|
owners[name].add((canon, model))
|
|
multiple = sorted(name for name, assigned in owners.items() if len(assigned) > 1)
|
|
for name in multiple:
|
|
errors.append(f'{name}: multiple owners {sorted(owners[name])}')
|
|
return {
|
|
'source_entries': len(expected),
|
|
'mapped_entries': len(entries),
|
|
'source_targets': target_count,
|
|
'additional_required_concepts': len(ledger.get('additional_required_concepts', [])),
|
|
'dispositions': dict(sorted(Counter(e['disposition'] for e in entries).items())),
|
|
'unowned_concepts': unowned,
|
|
'multiply_owned_concepts': len(multiple),
|
|
'errors': errors,
|
|
}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--ledger', type=Path, default=Path(__file__).resolve().parents[1] / 'ledger/concept-ownership.json')
|
|
parser.add_argument('--source-repo', type=Path, help='Donor checkout, under either old or new name')
|
|
args = parser.parse_args()
|
|
try:
|
|
ledger = json.loads(args.ledger.read_text())
|
|
inventory = json.loads((args.ledger.parent / ledger['source_inventory']).read_text())
|
|
result = validate(ledger, inventory)
|
|
result['source_blob_verified'] = False
|
|
if args.source_repo:
|
|
raw = subprocess.check_output([
|
|
'git', '-C', str(args.source_repo), 'show',
|
|
f"{inventory['commit']}:{inventory['path']}",
|
|
])
|
|
headings = [
|
|
{'heading': line[3:], 'line': number}
|
|
for number, line in enumerate(raw.decode().splitlines(), 1)
|
|
if line.startswith('## ')
|
|
]
|
|
if hashlib.sha256(raw).hexdigest() != inventory['sha256']:
|
|
result['errors'].append('Source blob SHA-256 mismatch')
|
|
elif headings != inventory['entries']:
|
|
result['errors'].append('Inventory differs from pinned source headings')
|
|
else:
|
|
result['source_blob_verified'] = True
|
|
result['valid'] = not result['errors']
|
|
except (OSError, ValueError, KeyError, TypeError, subprocess.CalledProcessError) as exc:
|
|
result = {'valid': False, 'errors': [str(exc)]}
|
|
print(json.dumps(result, indent=2))
|
|
return 0 if result['valid'] else 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|