kangshifu-board/scripts/manage-industry-dispatch.py

"""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)