File size: 4,973 Bytes
375880c 5a60197 91b8be2 375880c 5a60197 91b8be2 5a60197 375880c 5a60197 91b8be2 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 | """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 # stable, the UI and the tests key on it
level: str # info | warning | blocking
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 # what answering it sets: type, action, rule, data, group ...
text: str
options: list[dict] = field(default_factory=list) # [{value, label}]
blocking: bool = True # True: the draft must not be saved until answered
default: Any = None # what the draft currently uses
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)
# Set when the draft goes into an existing policy (merge.py): {id, name, active} and the changes.
mergeInto: dict | None = None
mergeOps: list[dict] = field(default_factory=list)
# Which part of the sentence this draft came from: answers are keyed "<slot>@<requestIndex>".
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 # still waiting for an answer: nothing to save yet
# Applied by the dashboard to the policy as it is at save time, then saved with PATCH.
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
|