kanbante/app/capture.py

"""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)