| """Response shapes. Everything the copilot returns is built from these and is JSON-serialisable.""" |
| from __future__ import annotations |
|
|
| from dataclasses import dataclass, field |
| from typing import Any |
|
|
| from .rules import RuleDecision |
|
|
|
|
| @dataclass |
| class Issue: |
| code: str |
| level: str |
| message: str |
| detail: dict = field(default_factory=dict) |
|
|
| def to_json(self) -> dict: |
| d = {'code': self.code, 'level': self.level, 'message': self.message} |
| if self.detail: |
| d['detail'] = self.detail |
| return d |
|
|
|
|
| @dataclass |
| class Question: |
| slot: str |
| text: str |
| options: list[dict] = field(default_factory=list) |
| blocking: bool = True |
| default: Any = None |
|
|
| def to_json(self) -> dict: |
| d = {'slot': self.slot, 'text': self.text, 'options': self.options, 'blocking': self.blocking} |
| if self.default is not None: |
| d['default'] = self.default |
| return d |
|
|
|
|
| @dataclass |
| class Draft: |
| policyType: str |
| policyBody: dict | None |
| ruleDecision: RuleDecision = field(default_factory=RuleDecision) |
| newBlockLists: list[dict] = field(default_factory=list) |
| issues: list[Issue] = field(default_factory=list) |
| questions: list[Question] = field(default_factory=list) |
| summary: str = '' |
| spans: list[dict] = field(default_factory=list) |
| |
| mergeInto: dict | None = None |
| mergeOps: list[dict] = field(default_factory=list) |
| |
| requestIndex: int = 0 |
|
|
| @property |
| def status(self) -> str: |
| if self.policyBody is None: |
| return 'invalid' |
| if any(q.blocking for q in self.questions) or any(i.level == 'blocking' for i in self.issues): |
| return 'needs_input' |
| return 'ready' |
|
|
| def issue(self, code: str, level: str, message: str, **detail) -> None: |
| if not any(i.code == code and i.detail == detail for i in self.issues): |
| self.issues.append(Issue(code, level, message, detail)) |
|
|
| def ask(self, q: Question) -> None: |
| if not any(x.slot == q.slot for x in self.questions): |
| self.questions.append(q) |
|
|
| def save_steps(self) -> list[dict]: |
| """The calls the dashboard makes when the admin saves, in order.""" |
| steps: list[dict] = [] |
| if self.ruleDecision.new_rule and '$NEW_RULE' in self.ruleDecision.rule_ids: |
| steps.append({'method': 'POST', 'path': '/api/admin/detection-rules', 'body': self.ruleDecision.new_rule, |
| 'resultId': 'rule.id', 'replaces': '$NEW_RULE'}) |
| for bl in self.newBlockLists: |
| steps.append({'method': 'POST', 'path': '/api/admin/block-lists', 'body': bl['body'], |
| 'resultId': 'list.id', 'replaces': bl['placeholder']}) |
| if self.policyBody is not None and self.mergeInto: |
| if not self.mergeOps: |
| return steps |
| |
| steps.append({'method': 'PATCH', 'path': f'/api/admin/policies/{self.mergeInto["id"]}', |
| 'merge': self.mergeOps}) |
| elif self.policyBody is not None: |
| steps.append({'method': 'POST', 'path': '/api/admin/policies', 'body': self.policyBody}) |
| return steps |
|
|
| def to_json(self) -> dict: |
| return { |
| 'status': self.status, |
| 'policyType': self.policyType, |
| 'summary': self.summary, |
| 'ruleDecision': self.ruleDecision.to_json(), |
| 'policyBody': self.policyBody, |
| 'saveSteps': self.save_steps(), |
| 'mergeInto': self.mergeInto, |
| 'requestIndex': self.requestIndex, |
| 'questions': [q.to_json() for q in self.questions], |
| 'issues': [i.to_json() for i in self.issues], |
| 'spans': self.spans, |
| } |
|
|
|
|
| def get_path(d: dict, path: str) -> Any: |
| cur: Any = d |
| for p in path.split('.'): |
| if not isinstance(cur, dict) or p not in cur: |
| return None |
| cur = cur[p] |
| return cur |
|
|
|
|
| def set_path(d: dict, path: str, value: Any) -> None: |
| """Set a dotted path that already exists in the default body. Never invents keys.""" |
| parts = path.split('.') |
| cur = d |
| for p in parts[:-1]: |
| nxt = cur.get(p) |
| if not isinstance(nxt, dict): |
| raise KeyError(path) |
| cur = nxt |
| if parts[-1] not in cur: |
| raise KeyError(path) |
| cur[parts[-1]] = value |
|
|
