#!/usr/bin/env python3 """Consult one decision with its private scoped credential, never execute its plan.""" import argparse import hashlib import json from pathlib import Path import re from control import Client def endpoint(cfg, suffix=''): decision_id=cfg.get('decision_id') if not isinstance(decision_id,str) or not re.fullmatch(r'[A-Za-z0-9_.:-]{1,128}',decision_id): raise ValueError('Invalid bound decision') return '/api/decisions/'+decision_id+suffix def execute(cfg,action,payload=None): client=Client(cfg['control']) if action=='show':return client.call(endpoint(cfg)) if action=='history':return client.call(endpoint(cfg,'/history')) if action=='edit': if not isinstance(payload,dict) or 'expected_revision' not in payload or not payload.get('reason'): raise ValueError('Edit requires current revision and reason') return client.call(endpoint(cfg),payload,'PATCH') if action=='ask': if not isinstance(payload,dict) or not payload.get('question'): raise ValueError('A factual question is required') key=payload.get('idempotency_key') or 'fact-'+hashlib.sha256( json.dumps(payload,sort_keys=True,ensure_ascii=True).encode()).hexdigest()[:32] return client.call(endpoint(cfg,'/facts'),{**payload,'idempotency_key':key}) raise ValueError('Unsupported consultation action') def main(): parser=argparse.ArgumentParser(description=__doc__) parser.add_argument('--config',required=True) parser.add_argument('action',choices=['show','history','edit','ask']) parser.add_argument('--payload-file') parser.add_argument('--question') args=parser.parse_args() cfg=json.loads(Path(args.config).read_text(encoding='utf-8-sig')) payload=json.loads(Path(args.payload_file).read_text(encoding='utf-8-sig')) if args.payload_file else None if args.question:payload={'question':args.question} print(json.dumps(execute(cfg,args.action,payload),ensure_ascii=True)) if __name__=='__main__':main()