49 lines
2.0 KiB
Python
49 lines
2.0 KiB
Python
#!/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()
|