Rhinox13/chatapi

1from __future__ import annotations
2
3from concurrent.futures import ThreadPoolExecutor
4from email.header import Header
5from logging import Logger
6from urllib import error, request
7
8from ..repositories import SystemConfigStore, UserStore
9from .url_safety import validate_public_http_url
10
11_executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="ntfy")
12
13
14class _NoRedirectHandler(request.HTTPRedirectHandler):
15    def redirect_request(self, req, fp, code, msg, headers, newurl):  # type: ignore[override]
16        return None
17
18
19_opener = request.build_opener(_NoRedirectHandler)
20
21
22def _encode_header(value: str) -> str:
23    try:
24        value.encode("latin-1")
25        return value
26    except UnicodeEncodeError:
27        return Header(value, "utf-8").encode()
28
29
30def notify_new_message(
31    system_config_store: SystemConfigStore,
32    user_store: UserStore,
33    owner_id: str,
34    *,
35    conversation_title: str,
36    message_text: str,
37    logger: Logger,
38) -> None:
39    url = user_store.get_effective_ntfy_url(owner_id)
40    text = message_text.strip()
41    if not url or not text:
42        return
43    user = user_store.get_user(owner_id)
44    allow_private = system_config_store.is_ntfy_private_url_allowed_for_role(user.role if user else "")
45    safety = validate_public_http_url(url, allow_private=allow_private)
46    if not safety.ok:
47        logger.warning("Skipped unsafe ntfy notification URL: %s", safety.reason or "unsafe URL")
48        return
49    title_fallback = system_config_store.get_effective_title("ChatAPI")
50
51    def send() -> None:
52        body = text.encode("utf-8")
53        req = request.Request(
54            url,
55            data=body,
56            method="POST",
57            headers={
58                "Content-Type": "text/plain; charset=utf-8",
59                "Title": _encode_header(conversation_title[:80] or title_fallback),
60            },
61        )
62        try:
63            with _opener.open(req, timeout=5) as response:
64                response.read(1)
65        except error.HTTPError as exc:
66            if 300 <= exc.code < 400:
67                logger.warning("Skipped ntfy notification redirect to %s", exc.headers.get("Location", ""))
68                return
69            logger.exception("Failed to send ntfy notification")
70        except Exception:
71            logger.exception("Failed to send ntfy notification")
72
73    _executor.submit(send)
74