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

"""Prepare and coordinate local evidence audits. Does not call VOC or issue collection requests."""
import argparse
import json
from pathlib import Path
import sys

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'server/voc-library'))
from industry_monitor import load, Monitor
from monitoring_store import need
from monitoring_tasks import seed, claim, release, finish, reconcile, read_tasks


def json_arg(value):
    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=['seed', 'list', 'claim', 'release', 'receipt', 'reconcile'])
    p.add_argument('--store', required=True)
    p.add_argument('--snapshot')
    p.add_argument('--registry')
    p.add_argument('--db')
    p.add_argument('--generation', required=True)
    p.add_argument('--filters', default='{}', help='JSON or @file')
    p.add_argument('--kinds', default='listings,reviews,rankings')
    p.add_argument('--task-id')
    p.add_argument('--owner')
    p.add_argument('--lease', type=int, default=1800)
    p.add_argument('--lease-file', help='Private local claim JSON; token is never accepted through HTTP')
    p.add_argument('--receipt', help='JSON or @file')
    p.add_argument('--page', type=int, default=1)
    p.add_argument('--state')
    p.add_argument('--kind')
    args = p.parse_args()
    try:
        data = load({k: getattr(args, k) for k in ('snapshot', 'registry', 'db') if getattr(args, k)})
        need(data['generation'] == args.generation, 'LIBRARY_VERSION_CHANGED')
        monitor, filters = Monitor(data), json_arg(args.filters)
        products = monitor.selected(filters)
        if args.command == 'seed':
            result = seed(args.store, monitor, filters, args.kinds.split(','))
        elif args.command == 'list':
            result = {'ok': True, **read_tasks(args.store, args.generation, filters, {p['record_id'] for p in products}, args.state, args.kind, args.page)}
        elif args.command == 'reconcile':
            result = reconcile(args.store, args.generation)
        elif args.command == 'claim':
            need(args.task_id and args.owner, 'TASK_AND_OWNER_REQUIRED')
            result = claim(args.store, args.task_id, args.owner, args.generation, args.lease)
        else:
            need(args.lease_file, 'PRIVATE_LEASE_FILE_REQUIRED')
            lease = json.loads(Path(args.lease_file).read_text('utf-8-sig'))
            need(not args.task_id or args.task_id == lease['taskId'], 'TASK_LEASE_MISMATCH')
            arguments = [args.store, lease['taskId'], lease['owner'], lease['leaseToken'], args.generation]
            if args.command == 'release':
                result = release(*arguments)
            else:
                need(args.receipt, 'TASK_RECEIPT_REQUIRED')
                result = finish(*arguments, json_arg(args.receipt))
        print(json.dumps(result, ensure_ascii=False))
    except Exception as error:
        print(json.dumps({'ok': False, 'error': str(error) if isinstance(error, ValueError) else 'TASK_OPERATION_FAILED'}, ensure_ascii=False))
        sys.exit(1)


if __name__ == '__main__':
    main()