Spark
395ab6e14c
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'.
118 lines
4.2 KiB
Python
118 lines
4.2 KiB
Python
"""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)
|