axiom

b4e18a0 · Protocol states are only active, blocked, closed · 27 days ago

python207 lines · 6.8 KB

1"""Notion is the live work ledger. Channels stay the markdown wall.

3The agent reads the Projects database and writes back:

4 - Broadcast TS + Broadcast Permalink after a post

5 - Last Status Update after a gather (the breadcrumb trail)

7Stdlib HTTP only. Token from NOTION_TOKEN. Dry-run queries and prints

8writes; it does not PATCH.

9"""

11from __future__ import annotations

13import json

14import os

15from typing import Any, Callable

16from urllib.error import HTTPError, URLError

17from urllib.request import Request, urlopen

19from mapping import Mapping, Work

21API = "https://api.notion.com/v1"

22NOTION_VERSION = "2022-06-28"

23MAX_BLOCK = 2000

25PROPS = {

26 "name": "Name",

27 "id": "Scope key",

28 "room": "Admin Channel",

29 "state": "Status",

30 "anchor": "Broadcast TS",

31 "owner": "Owner",

32 "permalink": "Broadcast Permalink",

33 "status_update": "Last Status Update",

34}

36Transport = Callable[[str, str, dict[str, str], bytes | None], dict[str, Any]]

39class NotionError(RuntimeError):

40 pass

43def default_transport(method: str, url: str, headers: dict[str, str], body: bytes | None) -> dict[str, Any]:

44 req = Request(url, data=body, method=method, headers=headers)

45 try:

46 with urlopen(req, timeout=20) as resp:

47 return json.loads(resp.read().decode("utf-8"))

48 except HTTPError as err:

49 raise NotionError(f"notion {method} {err.code}") from err

50 except URLError as err:

51 raise NotionError("notion unreachable") from err

54class NotionClient:

55 def __init__(self, token: str, transport: Transport | None = None) -> None:

56 if not token:

57 raise NotionError("NOTION_TOKEN is required")

58 self.token = token

59 self.transport = transport or default_transport

61 def _headers(self) -> dict[str, str]:

62 return {

63 "Authorization": "Bearer " + self.token,

64 "Notion-Version": NOTION_VERSION,

65 "Content-Type": "application/json",

66 "User-Agent": "axiom-relay/0.3",

67 }

69 def _call(self, method: str, path: str, payload: dict[str, Any] | None = None) -> dict[str, Any]:

70 body = json.dumps(payload).encode("utf-8") if payload is not None else None

71 return self.transport(method, API + path, self._headers(), body)

73 def query_work(self, database_id: str) -> tuple[Work, ...]:

74 rows: list[Work] = []

75 cursor: str | None = None

76 while True:

77 payload: dict[str, Any] = {"page_size": 100}

78 if cursor:

79 payload["start_cursor"] = cursor

80 data = self._call("POST", "/databases/" + database_id + "/query", payload)

81 for page in data.get("results", []):

82 w = work_from_page(page)

83 if w:

84 rows.append(w)

85 if not data.get("has_more"):

86 break

87 cursor = data.get("next_cursor")

88 return tuple(rows)

90 def write_anchor(self, page_id: str, ts: str, permalink: str) -> None:

91 if not ts:

92 raise NotionError("refusing empty Broadcast TS")

93 self._call(

94 "PATCH",

95 "/pages/" + page_id,

96 {

97 "properties": {

98 PROPS["anchor"]: rich_text(ts),

99 PROPS["permalink"]: rich_text(permalink),

100 }

101 },

102 )

103

104 def write_trail(self, page_id: str, trail: str) -> None:

105 self._call(

106 "PATCH",

107 "/pages/" + page_id,

108 {"properties": {PROPS["status_update"]: rich_text(clip(trail))}},

109 )

110

111

112def clip(text: str) -> str:

113 if len(text) <= MAX_BLOCK:

114 return text

115 suffix = "\n(truncated)"

116 return text[: MAX_BLOCK - len(suffix)] + suffix

117

118

119def rich_text(content: str) -> dict[str, Any]:

120 return {"rich_text": [{"type": "text", "text": {"content": content}}]}

121

122

123def _plain(prop: dict[str, Any] | None) -> str:

124 if not prop:

125 return ""

126 kind = prop.get("type")

127 if kind == "title":

128 return "".join(t.get("plain_text", "") for t in prop.get("title", []))

129 if kind == "rich_text":

130 return "".join(t.get("plain_text", "") for t in prop.get("rich_text", []))

131 if kind in {"select", "status"}:

132 node = prop.get(kind) or {}

133 return str(node.get("name") or "")

134 if kind == "url":

135 return str(prop.get("url") or "")

136 if kind == "number" and prop.get("number") is not None:

137 return str(prop["number"])

138 return ""

139

140

141def normalize_state(raw: str) -> str:

142 """Protocol states only. A company word is returned unchanged so the plan can gap it."""

143 return raw.strip().lower()

144

145

146def work_from_page(page: dict[str, Any]) -> Work | None:

147 props = page.get("properties") or {}

148 name = _plain(props.get(PROPS["name"]))

149 key = _plain(props.get(PROPS["id"])) or name.lower().replace(" ", "-")

150 if not key:

151 return None

152 return Work(

153 id=key,

154 name=name or key,

155 room=_plain(props.get(PROPS["room"])),

156 state=normalize_state(_plain(props.get(PROPS["state"]))),

157 anchor=_plain(props.get(PROPS["anchor"])),

158 owner=_plain(props.get(PROPS["owner"])),

159 page_id=str(page.get("id") or ""),

160 )

161

162

163def load_work_from_notion(mapping: Mapping, database_id: str, client: NotionClient) -> Mapping:

164 return mapping.replace_work(client.query_work(database_id))

165

166

167def apply_writes(

168 client: NotionClient,

169 plan: tuple[tuple[Work, str], ...],

170 posts: dict[str, tuple[str, str]],

171 trails: dict[str, str],

172 *,

173 dry_run: bool,

174) -> list[str]:

175 """Write anchors and trails back to Notion. Dry-run returns the plan, no PATCH.

176

177 posts maps work id -> (thread_ts, permalink) from the adapter.

178 trails maps work id -> breadcrumb text for gather.

179 A post with no ts does not PATCH. Empty is not recoverable; missing is.

180 """

181 lines: list[str] = []

182 for w, action in plan:

183 if not w.page_id:

184 lines.append(f"notion {w.id:16} skip no page id")

185 continue

186 if action == "post":

187 ts, permalink = posts.get(w.id, ("", ""))

188 if not ts:

189 lines.append(f"notion {w.id:16} POST no ts — leave empty, retry next run")

190 continue

191 lines.append(f"notion {w.id:16} POST write Broadcast TS + Permalink")

192 if not dry_run:

193 client.write_anchor(w.page_id, ts, permalink)

194 elif action == "gather":

195 lines.append(f"notion {w.id:16} GATHER write Last Status Update")

196 if not dry_run:

197 client.write_trail(w.page_id, trails.get(w.id, "silent"))

198 elif action == "gap":

199 lines.append(f"notion {w.id:16} GAP no write")

200 else:

201 lines.append(f"notion {w.id:16} SKIP closed")

202 return lines

203

204

205def client_from_env(transport: Transport | None = None) -> NotionClient:

206 return NotionClient(os.environ.get("NOTION_TOKEN", ""), transport)

207