feat(capture): capture éclair — API à jeton dédié + écoute ntfy (issue #12)
POST /api/capture (X-Capture-Token ou Bearer; 503 si non configuré; exempté Basic Auth/CSRF car machine-à-machine), JSON ou texte libre (1re ligne = titre), dépôt par défaut KANBANTE_CAPTURE_REPO. Module app/capture.py: parsing partagé + abonné ntfy en tâche de fond (reconnexion, curseur kv_store anti-rejeu, tag 'kanban'... 'kanbante' ignoré). _do_capture commun: issue créée, carte placée, événement 'created' source 'capture'.
This commit is contained in:
+117
@@ -0,0 +1,117 @@
|
||||
"""Quick capture (roadmap #12): turn an external message into an issue.
|
||||
|
||||
Two entry points share the same parsing:
|
||||
- POST /api/capture (token-authenticated, in app.main) — curl, a
|
||||
shortcut, any automation;
|
||||
- an ntfy *listener*: messages published to a dedicated topic become
|
||||
issues in the default capture repo. Capture from a phone is then
|
||||
just "share to the ntfy app".
|
||||
|
||||
The listener is a background task with reconnect + dedupe: the last
|
||||
processed ntfy message id is kept (by the caller, in kv_store) and
|
||||
passed back as `since`, so a restart never re-ingests old messages.
|
||||
Messages tagged "kanbante" are skipped — Kanbante never ingests its
|
||||
own posts.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
|
||||
import httpx
|
||||
|
||||
from .config import get_settings
|
||||
|
||||
logger = logging.getLogger("kanbante.capture")
|
||||
|
||||
MAX_TITLE_CHARS = 255
|
||||
|
||||
|
||||
def parse_capture_text(text: str) -> tuple[str, str]:
|
||||
"""First non-empty line = title, the rest = body."""
|
||||
lines = (text or "").strip().splitlines()
|
||||
lines = [line.rstrip() for line in lines]
|
||||
while lines and not lines[0].strip():
|
||||
lines.pop(0)
|
||||
if not lines:
|
||||
return "", ""
|
||||
title = lines[0].strip()[:MAX_TITLE_CHARS]
|
||||
body = "\n".join(lines[1:]).strip()
|
||||
return title, body
|
||||
|
||||
|
||||
def parse_ntfy_message(msg: dict) -> tuple[str, str] | None:
|
||||
"""Extract (title, body) from an ntfy message, or None to skip it."""
|
||||
if msg.get("event") not in (None, "message"):
|
||||
return None
|
||||
if "kanbante" in (msg.get("tags") or []):
|
||||
return None
|
||||
title = str(msg.get("title") or "").strip()
|
||||
message = str(msg.get("message") or "").strip()
|
||||
if title and message:
|
||||
return title[:MAX_TITLE_CHARS], message
|
||||
if message:
|
||||
return parse_capture_text(message)
|
||||
if title:
|
||||
return title[:MAX_TITLE_CHARS], ""
|
||||
return None
|
||||
|
||||
|
||||
async def ntfy_capture_loop(create_issue, get_last_id, set_last_id) -> None:
|
||||
"""Subscribe to the capture topic forever (reconnect with backoff).
|
||||
|
||||
create_issue(title, body) is an async callback; get_last_id /
|
||||
set_last_id persist the ntfy cursor. Returns quietly when the
|
||||
listener is no longer configured.
|
||||
"""
|
||||
seen: set[str] = set()
|
||||
while True:
|
||||
settings = get_settings()
|
||||
base = settings.ntfy_url.rstrip("/")
|
||||
topic = settings.capture_ntfy_topic
|
||||
if not base or not topic:
|
||||
return
|
||||
since = get_last_id() or "latest"
|
||||
headers = {}
|
||||
if settings.ntfy_token:
|
||||
headers["Authorization"] = f"Bearer {settings.ntfy_token}"
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=None) as client:
|
||||
async with client.stream(
|
||||
"GET", f"{base}/{topic}/json?since={since}", headers=headers
|
||||
) as response:
|
||||
response.raise_for_status()
|
||||
logger.info(
|
||||
"ntfy capture listening on %s/%s (since=%s)",
|
||||
base, topic, since,
|
||||
)
|
||||
async for line in response.aiter_lines():
|
||||
if not line.strip():
|
||||
continue
|
||||
try:
|
||||
msg = json.loads(line)
|
||||
except ValueError:
|
||||
continue
|
||||
msg_id = str(msg.get("id") or "")
|
||||
if msg_id:
|
||||
if msg_id in seen:
|
||||
continue
|
||||
seen.add(msg_id)
|
||||
if len(seen) > 500:
|
||||
seen = set(list(seen)[-250:])
|
||||
set_last_id(msg_id)
|
||||
parsed = parse_ntfy_message(msg)
|
||||
if parsed is None:
|
||||
continue
|
||||
title, body = parsed
|
||||
try:
|
||||
await create_issue(title, body)
|
||||
except Exception:
|
||||
logger.exception("capture failed for %r", title)
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.warning(
|
||||
"ntfy capture stream interrupted (%s); retrying in 10s", exc
|
||||
)
|
||||
await asyncio.sleep(10)
|
||||
Reference in New Issue
Block a user