"""Review existing collector request intents and settled history. No network or dispatch command.""" import argparse import os import json import os from pathlib import Path import re import sys sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'server/voc-library')) from industry_monitor import load from monitoring_dispatch import batch_preflight, claim, import_closeout, preflight, projection, release from monitoring_store import need def json_arg(value): need(isinstance(value, str) and value, 'JSON_ARGUMENT_REQUIRED') return json.loads(Path(value[1:]).read_text('utf-8-sig') if value.startswith('@') else value) def main(): p = argparse.ArgumentParser(description=__doc__) p.add_argument('command', choices=['status', 'preflight', 'batch', 'claim', 'release', 'import-closeout', 'import-ledger']) sources = p.add_mutually_exclusive_group(required=not bool(os.environ.get('KSF_UNIFIED_DB_CONFIG'))) sources.add_argument('--snapshot'); sources.add_argument('--registry'); sources.add_argument('--db') p.add_argument('--generation', required=True) p.add_argument('--store', required=True) p.add_argument('--request', help='JSON or @file; batch expects an array') p.add_argument('--archive', type=Path) p.add_argument('--sha256') p.add_argument('--historical-archive', type=Path) p.add_argument('--historical-sha256') p.add_argument('--owner') p.add_argument('--lease', type=int, default=1800) p.add_argument('--lease-file', type=Path) p.add_argument('--out', type=Path) args = p.parse_args() config = {k: getattr(args, k) for k in ('snapshot', 'registry', 'db') if getattr(args, k)} data = load(config) need(data['generation'] == args.generation, 'LIBRARY_VERSION_CHANGED') if args.out: need(not args.out.exists(), 'OUTPUT_ALREADY_EXISTS') if args.command == 'import-ledger': from monitoring_ledger import register need(args.archive and args.sha256 and args.historical_archive and args.historical_sha256, 'LEDGER_ARCHIVES_AND_SHA_REQUIRED') result = register(args.store, args.archive, args.sha256, args.historical_archive, args.historical_sha256) elif args.command == 'import-closeout': need(args.archive and args.sha256, 'CLOSEOUT_ARCHIVE_AND_SHA_REQUIRED') result = import_closeout(args.store, args.archive, args.sha256) elif args.command == 'status': result = {'ok': True, **projection(args.store, args.generation)} elif args.command == 'preflight': result = preflight(args.store, json_arg(args.request), args.generation) elif args.command == 'batch': result = batch_preflight(args.store, json_arg(args.request), args.generation) elif args.command == 'claim': need(args.out is not None, 'PRIVATE_LEASE_OUTPUT_REQUIRED') result = claim(args.store, json_arg(args.request), args.generation, args.owner, args.lease) else: need(args.lease_file is not None, 'PRIVATE_LEASE_FILE_REQUIRED') result = release(args.store, json.loads(args.lease_file.read_text('utf-8-sig')), args.generation) if args.out: fd = os.open(args.out, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, 'w', encoding='utf-8') as stream: json.dump(result, stream, ensure_ascii=False, indent=2) stream.write('\n') # Token is restricted to the explicitly requested private lease file. public = {k: v for k, v in result.items() if k not in ('leaseToken', 'owner')} print(json.dumps(public, ensure_ascii=False)) if __name__ == '__main__': try: main() except Exception as exc: message = str(exc) code = message if isinstance(exc, ValueError) and re.fullmatch('[A-Z][A-Z0-9_]+', message) else 'REQUEST_GUARD_FAILED' print(json.dumps({'ok': False, 'error': code, 'dispatchAllowed': False, 'paidCalls': 0})) sys.exit(1)