XGH
scoped-relay
CodeCommitsAxiomsAsk

e7a14c9 · Ship broadcast cards and an agent skill · 27 days ago

python126 lines · 3.1 KB
1"""Room messages. Four kinds, one shape. No labeled dumps."""
2
3from __future__ import annotations
4
5from dataclasses import dataclass
6
7from mapping import Mapping, Work
8
9KINDS = ("kickoff", "nudge", "blocked", "done")
10STATES = ("active", "blocked", "closed")
11EXPECT = {
12 "kickoff": "active",
13 "nudge": "active",
14 "blocked": "blocked",
15 "done": "closed",
16}
17CTA = {
18 "kickoff": "Reply in this thread. One status is enough.",
19 "nudge": "Still open. Reply in this thread.",
20 "blocked": "Blocked. Reply here only if that changed.",
21 "done": "Closed.",
22}
23
24
25class BroadcastError(ValueError):
26 pass
27
28
29@dataclass(frozen=True)
30class Broadcast:
31 kind: str
32 work_id: str
33 title: str
34 state: str
35 owner: str = ""
36 assignee: str = ""
37 quote: str = ""
38 quote_from: str = ""
39 platform: str = "slack"
40
41
42def render(b: Broadcast) -> str:
43 if b.kind not in KINDS:
44 raise BroadcastError(f"unknown kind: {b.kind}")
45 if b.state not in STATES:
46 raise BroadcastError(f"unknown state: {b.state}")
47 if b.state != EXPECT[b.kind]:
48 raise BroadcastError(f"{b.kind} requires state {EXPECT[b.kind]}")
49 title = _line(b.title)
50 work_id = _line(b.work_id)
51 if not title or not work_id:
52 raise BroadcastError("title and work id are required")
53 lines = [_bold(title, b.platform), f"`{work_id}` · {b.state}"]
54 who = _who(_line(b.owner), _line(b.assignee))
55 if who:
56 lines.append("")
57 lines.extend(who)
58 quote = _line(b.quote)
59 if quote:
60 lines.append("")
61 lines.append("> " + quote)
62 frm = _line(b.quote_from)
63 if frm:
64 lines.append("> — " + frm)
65 lines.append("")
66 lines.append(CTA[b.kind])
67 text = "\n".join(lines) + "\n"
68 for banned in ("broadcast:", "state:", "owner:", "in flight:", "context:"):
69 if banned in text:
70 raise BroadcastError(f"labeled dump leaked: {banned}")
71 return text
72
73
74def kickoff_for(
75 work: Work,
76 *,
77 assignee: str = "",
78 quote: str = "",
79 quote_from: str = "",
80 platform: str = "slack"
81) -> str:
82 return render(
83 Broadcast(
84 kind="kickoff",
85 work_id=work.id,
86 title=work.name or work.id,
87 state=work.state,
88 owner=work.owner,
89 assignee=assignee,
90 quote=quote,
91 quote_from=quote_from,
92 platform=platform,
93 )
94 )
95
96
97def preview_posts(mapping: Mapping, platform: str = "slack") -> str:
98 """The kickoff each `post` row would drop in its room. Empty if nothing posts."""
99 chunks = [
100 kickoff_for(w, platform=platform)
101 for w, action in mapping.plan()
102 if action == "post"
103 ]
104 if not chunks:
105 return ""
106 return "# would post\n\n" + "\n".join(chunks)
107
108
109def _who(owner: str, assignee: str) -> list[str]:
110 lines: list[str] = []
111 if owner:
112 lines.append(f"{owner} owns it")
113 if assignee:
114 lines.append(f"{assignee} is on it")
115 return lines
116
117
118def _bold(text: str, platform: str) -> str:
119 if platform == "discord":
120 return f"**{text}**"
121 return f"*{text}*"
122
123
124def _line(text: str) -> str:
125 return " ".join(text.split())
126