6f12c9a · Accept mapping URLs · 5 days ago
1
"""Parse an Axiom mapping markdown file (front matter + table)."""2
3
from __future__ import annotations4
5
from dataclasses import dataclass6
from pathlib import Path7
from urllib.request import Request, urlopen8
9
ROLES = {"management", "room", "reports"}10
DIRECTIONS = {"in", "out", "both"}11
PLATFORMS = {"slack", "discord"}12
13
14
@dataclass(frozen=True)15
class Channel:16
id: str17
name: str18
role: str19
direction: str20
21
22
@dataclass(frozen=True)23
class Mapping:24
workspace: str25
platform: str26
scope: str27
agent: str28
collect_seconds: int29
channels: tuple[Channel, ...]30
31
def ids(self) -> set[str]:32
return {c.id for c in self.channels}33
34
def by_role(self, role: str) -> tuple[Channel, ...]:35
return tuple(c for c in self.channels if c.role == role)36
37
38
class MappingError(ValueError):39
pass40
41
42
def read_source(source: str) -> str:43
if source.startswith(("http://", "https://")):44
req = Request(source, headers={"User-Agent": "axiom-relay/0.2"})45
with urlopen(req, timeout=15) as resp:46
return resp.read().decode("utf-8")47
return Path(source).read_text(encoding="utf-8")48
49
50
def load_mapping_from(source: str) -> Mapping:51
return load_mapping(read_source(source))52
53
54
def load_mapping(text: str) -> Mapping:55
meta, table = _split_front_matter(text)56
channels = _parse_table(table)57
_validate(meta, channels)58
return Mapping(59
workspace=meta["workspace"],60
platform=meta["platform"],61
scope=meta["scope"],62
agent=meta["agent"],63
collect_seconds=int(meta.get("collect_seconds", 45)),64
channels=channels,65
)66
67
68
def _split_front_matter(text: str) -> tuple[dict[str, str], str]:69
if not text.startswith("---"):70
raise MappingError("mapping must start with YAML front matter")71
rest = text[3:]72
end = rest.find("\n---")73
if end < 0:74
raise MappingError("unterminated front matter")75
raw, body = rest[:end], rest[end + 4 :]76
meta: dict[str, str] = {}77
for line in raw.splitlines():78
if not line.strip() or ":" not in line:79
continue80
key, value = line.split(":", 1)81
meta[key.strip()] = value.strip()82
return meta, body83
84
85
def _parse_table(body: str) -> tuple[Channel, ...]:86
rows: list[Channel] = []87
header: list[str] | None = None88
for raw in body.splitlines():89
line = raw.strip()90
if not line.startswith("|"):91
continue92
cells = [c.strip() for c in line.strip("|").split("|")]93
if header is None:94
header = [c.lower() for c in cells]95
continue96
if set("".join(cells)) <= set("-: "):97
continue98
data = dict(zip(header, cells))99
rows.append(100
Channel(101
id=data["id"],102
name=data.get("name", ""),103
role=data["role"],104
direction=data["direction"],105
)106
)107
return tuple(rows)108
109
110
def _validate(meta: dict[str, str], channels: tuple[Channel, ...]) -> None:111
for key in ("workspace", "platform", "scope", "agent"):112
if not meta.get(key):113
raise MappingError(f"missing {key}")114
if meta["platform"] not in PLATFORMS:115
raise MappingError(f"unknown platform {meta['platform']!r}")116
ids = [c.id for c in channels]117
if len(ids) != len(set(ids)):118
raise MappingError("duplicate channel id")119
for ch in channels:120
if ch.role not in ROLES:121
raise MappingError(f"unknown role {ch.role!r}")122
if ch.direction not in DIRECTIONS:123
raise MappingError(f"unknown direction {ch.direction!r}")124
if not any(c.role == "management" and c.direction in {"in", "both"} for c in channels):125
raise MappingError("need a management channel with direction in/both")126
if not any(c.role == "reports" and c.direction in {"out", "both"} for c in channels):127
raise MappingError("need a reports channel with direction out/both")128