snotes updates, actions, attachments, easy user space

backend/actions.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+import re
+import tomllib
+from pathlib import Path
+from urllib.parse import quote
+
+import httpx
+import structlog
+
+from backend.config import settings
+
+logger = structlog.get_logger()
+
+ACTION_LINE_RE = re.compile(r"^/(?P<kw>[A-Za-z0-9_-]+)\s*:?\s*(?P<rest>.*)$")
+
+_actions_cache: dict[str, dict] | None = None
+
+
+def load_actions(reload: bool = False) -> dict[str, dict]:
+    global _actions_cache
+    if _actions_cache is not None and not reload:
+        return _actions_cache
+
+    actions: dict[str, dict] = {}
+    path = Path(settings.actions_file)
+    if path.is_file():
+        data = tomllib.loads(path.read_text(encoding="utf-8"))
+        for item in data.get("actions", []):
+            name = item.get("name")
+            uri = item.get("uri")
+            if name and uri:
+                actions[name] = {"uri": uri, "link": item.get("link", uri)}
+    _actions_cache = actions
+    return actions
+
+
+async def _perform(uri: str) -> bool:
+    try:
+        async with httpx.AsyncClient() as client:
+            resp = await client.get(uri)
+            resp.raise_for_status()
+        return True
+    except Exception:
+        logger.exception("action_failed", uri=uri)
+        return False
+
+
+async def process_actions(body: str) -> str:
+    actions = load_actions()
+    if not actions:
+        return body
+
+    out_lines: list[str] = []
+    for line in body.split("\n"):
+        m = ACTION_LINE_RE.match(line)
+        action = actions.get(m.group("kw")) if m else None
+        if action is None:
+            out_lines.append(line)
+            continue
+
+        rest = m.group("rest").strip()
+        if not rest:
+            out_lines.append(line)
+            continue
+
+        encoded = quote(rest)
+        uri = action["uri"].replace("#string1#", encoded)
+        link = action["link"].replace("#string1#", encoded)
+        if await _perform(uri):
+            out_lines.append(f"[{rest}]({link})")
+        else:
+            out_lines.append(line)
+
+    return "\n".join(out_lines)