From 395ab6e14c7066446a8ed2fe2ff2a3c0aa71f555 Mon Sep 17 00:00:00 2001 From: Spark Date: Sun, 11 Oct 2026 12:40:31 -0400 Subject: [PATCH] =?UTF-8?q?feat(capture):=20capture=20=C3=A9clair=20?= =?UTF-8?q?=E2=80=94=20API=20=C3=A0=20jeton=20d=C3=A9di=C3=A9=20+=20=C3=A9?= =?UTF-8?q?coute=20ntfy=20(issue=20#12)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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'. --- app/capture.py | 117 +++++++++++++++++++++++++++++++++ app/config.py | 23 ++++++- app/main.py | 175 ++++++++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 313 insertions(+), 2 deletions(-) create mode 100644 app/capture.py diff --git a/app/capture.py b/app/capture.py new file mode 100644 index 0000000..9f55ed6 --- /dev/null +++ b/app/capture.py @@ -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) diff --git a/app/config.py b/app/config.py index 3ebd2e2..68858ad 100644 --- a/app/config.py +++ b/app/config.py @@ -36,6 +36,21 @@ Environment variables: the daily briefing is sent to ntfy. Empty (default) = no automatic send, the /briefing page stays available. + KANBANTE_CAPTURE_REPO Default repo for quick capture + (POST /api/capture, the /capture page + and the ntfy capture listener), e.g. + "Projets/kanbante". Empty = captures + must name their repo explicitly. + KANBANTE_CAPTURE_TOKEN Dedicated token for POST /api/capture + (header X-Capture-Token or Bearer). + Empty (default) = the endpoint is + disabled (503). Distinct from the + Gitea token; revocable on its own. + KANBANTE_CAPTURE_NTFY_TOPIC ntfy topic listened to for captures: + each message published there becomes + an issue in KANBANTE_CAPTURE_REPO. + Uses KANBANTE_NTFY_URL/_TOKEN as the + server. Empty (default) = no listener. """ import os @@ -44,7 +59,7 @@ from pathlib import Path BASE_DIR = Path(__file__).resolve().parent.parent -APP_VERSION = "0.6.0" +APP_VERSION = "0.7.0" def _env_bool(name: str, default: bool) -> bool: @@ -100,6 +115,9 @@ class Settings: ntfy_topic: str = "" ntfy_token: str = "" briefing_time: str = "" + capture_repo: str = "" + capture_token: str = "" + capture_ntfy_topic: str = "" @property def auth_enabled(self) -> bool: @@ -129,4 +147,7 @@ def get_settings() -> Settings: ntfy_topic=os.getenv("KANBANTE_NTFY_TOPIC", ""), ntfy_token=os.getenv("KANBANTE_NTFY_TOKEN", ""), briefing_time=os.getenv("KANBANTE_BRIEFING_TIME", ""), + capture_repo=os.getenv("KANBANTE_CAPTURE_REPO", ""), + capture_token=os.getenv("KANBANTE_CAPTURE_TOKEN", ""), + capture_ntfy_topic=os.getenv("KANBANTE_CAPTURE_NTFY_TOPIC", ""), ) diff --git a/app/main.py b/app/main.py index 18fe33a..fdb8c74 100644 --- a/app/main.py +++ b/app/main.py @@ -35,6 +35,7 @@ from starlette.middleware.base import BaseHTTPMiddleware from app.config import APP_VERSION, get_settings from app.gitea_client import GiteaClient from app.briefing import format_briefing_text, ntfy_configured, send_ntfy +from app.capture import ntfy_capture_loop, parse_capture_text from app.models import ( KANBAN_COLUMNS, column_entry_times, @@ -139,8 +140,30 @@ async def lifespan(app: FastAPI): "GITEA_TOKEN is not set: only public Gitea content will be visible." ) scheduler = asyncio.create_task(_briefing_scheduler()) + capture_task = None + if settings.capture_ntfy_topic and settings.ntfy_url: + async def _capture_create(title: str, body: str) -> None: + repo = get_settings().capture_repo + if not repo: + logger.warning("ntfy capture: no KANBANTE_CAPTURE_REPO, skipped") + return + await _do_capture(title, body, repo, "backlog") + + capture_task = asyncio.create_task( + ntfy_capture_loop( + _capture_create, + lambda: get_kv("capture_ntfy_last_id"), + lambda value: set_kv("capture_ntfy_last_id", value), + ) + ) + logger.info( + "ntfy capture listener started (topic %s)", + settings.capture_ntfy_topic, + ) yield scheduler.cancel() + if capture_task is not None: + capture_task.cancel() await gitea.aclose() @@ -162,7 +185,7 @@ class AuthMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): # /api/health: liveness for Docker. /webhooks/gitea: authenticated # by its HMAC signature, not by Basic auth (Gitea cannot log in). - if request.url.path in ("/api/health", "/webhooks/gitea"): + if request.url.path in ("/api/health", "/webhooks/gitea", "/api/capture"): return await call_next(request) settings = get_settings() if not settings.auth_enabled: @@ -1003,6 +1026,156 @@ async def sse_event_stream(request: Request): hub.unsubscribe(queue) +async def _do_capture( + title: str, body: str, repo_full_name: str, column: str = "backlog" +) -> dict: + """Create a Gitea issue from a capture and place its card. + + Shared by POST /api/capture, the /capture page and the ntfy + listener. Raises ValueError for bad input (callers map to 400). + """ + title = (title or "").strip() + if not title or len(title) > 255: + raise ValueError("Title required (max 255 chars)") + if column not in KANBAN_COLUMNS: + raise ValueError("Invalid column") + parts = _split_full_name(repo_full_name) + if parts is None: + raise ValueError("Invalid repo (expected owner/name)") + owner, repo = parts + issue = await gitea.create_issue(owner, repo, title, body or "") + move_issue(repo_full_name, issue["number"], column) + log_event( + repo_full_name, issue["number"], "created", + to_column=column, source="capture", + ) + return { + "number": issue["number"], + "title": issue.get("title", title), + "html_url": issue.get("html_url", ""), + "repo_full_name": repo_full_name, + "column": column, + } + + +@app.post("/api/capture") +async def api_capture(request: Request): + """Quick capture endpoint (roadmap #12). + + Auth: dedicated capture token (X-Capture-Token header or + Authorization: Bearer), distinct from the Gitea token. Exempt + from Basic Auth and CSRF — it is a machine-to-machine endpoint. + Disabled (503) until KANBANTE_CAPTURE_TOKEN is set. + """ + settings = get_settings() + if not settings.capture_token: + return JSONResponse( + {"ok": False, "error": "Capture is not configured " + "(set KANBANTE_CAPTURE_TOKEN)"}, + status_code=503, + ) + provided = request.headers.get("x-capture-token", "") + auth = request.headers.get("authorization", "") + if not provided and auth.lower().startswith("bearer "): + provided = auth[7:].strip() + if not provided or not secrets.compare_digest(provided, settings.capture_token): + return JSONResponse( + {"ok": False, "error": "Invalid capture token"}, status_code=401 + ) + + content_type = request.headers.get("content-type", "") + title = body = text = repo = "" + column = "backlog" + if "application/json" in content_type: + try: + payload = await request.json() + except Exception: + payload = {} + if isinstance(payload, dict): + title = str(payload.get("title") or "") + body = str(payload.get("body") or "") + text = str(payload.get("text") or "") + repo = str(payload.get("repo") or "") + column = str(payload.get("column") or "backlog") + elif "application/x-www-form-urlencoded" in content_type or ( + "multipart/form-data" in content_type + ): + form = await request.form() + title = str(form.get("title") or "") + body = str(form.get("body") or "") + text = str(form.get("text") or "") + repo = str(form.get("repo") or "") + column = str(form.get("column") or "backlog") + else: # text/plain and friends: the whole body is the capture + text = (await request.body()).decode("utf-8", errors="replace") + + if not title and text: + title, rest = parse_capture_text(text) + body = body or rest + repo = repo or settings.capture_repo + if not repo: + return JSONResponse( + {"ok": False, "error": "No repo given and KANBANTE_CAPTURE_REPO " + "is not set"}, + status_code=400, + ) + try: + created = await _do_capture(title, body, repo, column) + except ValueError as exc: + return JSONResponse({"ok": False, "error": str(exc)}, status_code=400) + return JSONResponse({"ok": True, "issue": created}, status_code=201) + + +@app.get("/capture", response_class=HTMLResponse) +async def capture_page(request: Request): + """Quick-capture page: form, status and usage instructions.""" + settings = get_settings() + return render( + request, + "capture.html", + capture_repo=settings.capture_repo, + api_enabled=bool(settings.capture_token), + ntfy_topic=settings.capture_ntfy_topic, + ntfy_url=settings.ntfy_url, + created=None, + error=None, + ) + + +@app.post("/capture", response_class=HTMLResponse) +async def capture_submit( + request: Request, + repo_full_name: str = Form(""), + title: str = Form(""), + body: str = Form(""), + column: str = Form("backlog"), + csrf_token: str = Form(""), +): + """Web form capture (session + CSRF, like every other form).""" + if not _csrf_valid(request, csrf_token): + return _csrf_error() + settings = get_settings() + repo_full_name = repo_full_name.strip() or settings.capture_repo + created, error = None, None + if not repo_full_name: + error = "Aucun dépôt : renseigne le dépôt ou définis KANBANTE_CAPTURE_REPO." + else: + try: + created = await _do_capture(title, body, repo_full_name, column) + except ValueError as exc: + error = str(exc) + return render( + request, + "capture.html", + capture_repo=settings.capture_repo, + api_enabled=bool(settings.capture_token), + ntfy_topic=settings.capture_ntfy_topic, + ntfy_url=settings.ntfy_url, + created=created, + error=error, + ) + + @app.get("/api/stream") async def live_stream(request: Request): """Server-Sent Events: browsers reload softly on Gitea activity."""