File size: 4,805 Bytes
375880c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | """Try the copilot in a terminal.
python -m copilot.cli # interactive, typical test tenant
python -m copilot.cli --context my_tenant.json # your own context file
python -m copilot.cli "block aadhaar in chatgpt" # one request, then exit
python -m copilot.cli --json "block usb for interns" # raw response JSON
Questions are asked one by one; your answers are sent back exactly as the dashboard would.
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from .pipeline import parse
DEFAULT_CONTEXT = Path(__file__).parent.parent / 'corpus' / 'contexts' / 'typical.json'
def load_context(path: str | None) -> dict:
if not path and not DEFAULT_CONTEXT.exists():
return {} # installed without the test tenants: an empty tenant (built-in rules only)
p = Path(path) if path else DEFAULT_CONTEXT
try:
return json.loads(p.read_text(encoding='utf-8'))
except (OSError, json.JSONDecodeError) as e:
print(f'Could not read context {p}: {e}. Using an empty tenant.', file=sys.stderr)
return {}
def show(r: dict, full: bool) -> None:
print(f'\n[{r["status"]}] {r["message"]}')
for i in r.get('issues', []):
print(f' ({i["code"]}) {i["message"]}')
for n, d in enumerate(r['drafts']):
print(f'\n Draft {n}: {d["policyType"] or "?"} - {d["status"]}')
print(f' {d["summary"]}')
if d['policyBody']:
print(f' name: {d["policyBody"]["name"]}')
rd = d['ruleDecision']
if rd['mode'] != 'none':
print(f' detection rule: {rd["mode"]} {rd["ruleIds"]}')
for i in d['issues']:
if i['code'] != 'CREATED_INACTIVE':
print(f' {i["level"]:>8}: {i["message"]}')
if full and d['policyBody']:
for step in d['saveSteps']:
print(f' {step["method"]} {step["path"]}')
print(' ' + json.dumps(step['body'], indent=2).replace('\n', '\n '))
if r.get('examples') and not r['drafts']:
print('\n Try for example:')
for e in r['examples'][:4]:
print(f' - {e}')
def ask(r: dict, answers: dict) -> bool:
"""Ask every open question. Returns False when there was nothing to ask."""
asked = False
for n, d in enumerate(r['drafts']):
for q in d['questions']:
key = f'{q["slot"]}@{n}'
if key in answers:
continue
asked = True
print(f'\n Draft {n} asks: {q["text"]}' + ('' if q['blocking'] else ' (optional, Enter to skip)'))
for k, o in enumerate(q['options'][:12]):
print(f' {k + 1}. {o["label"]}')
raw = input(' > ').strip()
if not raw:
answers[key] = q.get('default')
continue
if raw.isdigit() and 1 <= int(raw) <= len(q['options']):
answers[key] = q['options'][int(raw) - 1]['value']
elif q['slot'] in ('data', 'connector', 'mcp', 'sites'):
answers[key] = [x.strip() for x in raw.split(',') if x.strip()]
else:
answers[key] = raw
return asked
def run_one(text: str, ctx: dict, as_json: bool, interactive: bool) -> None:
answers: dict = {}
r = parse(text, ctx, answers)
if as_json:
print(json.dumps(r, indent=2, ensure_ascii=False))
return
if not interactive:
show(r, full=True)
return
show(r, full=False)
while r['status'] == 'needs_input' and ask(r, answers):
r = parse(text, ctx, answers)
show(r, full=False)
if r['drafts'] and any(d['policyBody'] for d in r['drafts']):
if input('\nShow the exact request bodies? [y/N] ').strip().lower() == 'y':
show(r, full=True)
def main() -> None:
ap = argparse.ArgumentParser(description='Policy Copilot in the terminal')
ap.add_argument('text', nargs='*')
ap.add_argument('--context')
ap.add_argument('--json', action='store_true')
args = ap.parse_args()
for stream in (sys.stdout, sys.stderr):
try:
stream.reconfigure(encoding='utf-8', errors='replace')
except (AttributeError, ValueError):
pass
ctx = load_context(args.context)
if args.text:
run_one(' '.join(args.text), ctx, args.json, interactive=False)
return
print('Policy Copilot. Describe a policy (empty line to quit).')
while True:
try:
text = input('\n> ').strip()
except (EOFError, KeyboardInterrupt):
break
if not text:
break
run_one(text, ctx, False, interactive=True)
if __name__ == '__main__':
main()
|