| """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 {} |
| 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() |
|
|
