Fix Telegram delivery with an authenticated free-tier relay

.dockerignore

@@ -0,0 +1,13 @@
+.cache
+.git
+__pycache__
+*.pyc
+.env
+.env.*
+*.sqlite3
+*.sqlite3-*
+.local-*
+.wrangler
+node_modules
+.dev.vars
+.dev.vars.*

.gitignore

@@ -0,0 +1,12 @@
+.cache/
+__pycache__/
+*.pyc
+.env
+.env.*
+*.sqlite3
+*.sqlite3-*
+.local-*
+.wrangler/
+node_modules/
+.dev.vars
+.dev.vars.*

README.md

@@ -14,8 +14,9 @@ independent DeepSeek classifier for potential topic alerts. Text only; no image,
 OCR, subtitle, or original-page fetching. The default model is deepseek-flash
 with thinking disabled. A positive result is a signal for a person to assess.
 
-Set these Space Secrets: WEBHOOK_SECRET, TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID,
-DEEPSEEK_API_KEY and ALERT_MENTIONS. CLASSIFIER_PROMPT overrides the generic
+Set these Space Secrets: WEBHOOK_SECRET, TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID
+and DEEPSEEK_API_KEY. ALERT_MENTIONS optionally adds user mentions to alerts.
+CLASSIFIER_PROMPT overrides the generic
 default instructions and belongs in a Secret when it contains internal context.
 Never commit credentials or confidential classifier instructions to this repo.
 
@@ -37,3 +38,40 @@ or an alert. AI failures remain pending and are retried, never recorded as NO.
 Normal posts are sent silently; alerts request a notification, which still
 depends on recipients' Telegram settings. Mentions are explicit @usernames.
 Post text is split across messages when necessary; its complete text is checked.
+
+## Delivery health and independent workers
+
+`/health` now distinguishes successful delivery, an untested Telegram connection,
+delivery errors, missing configuration, and stopped workers. The public response
+contains fixed error categories, never tokens, chat identifiers or post content.
+It is not an end-to-end delivery guarantee. `/live` is a separate process liveness
+check, so a Telegram outage does not restart the process and discard its queue.
+
+Telegram forwarding runs whenever its token and chat ID are configured, even if
+DeepSeek is not configured yet. AI processing starts independently when its key
+is available. `ALERT_MENTIONS` is optional. Valid authenticated webhooks are stored
+even while downstream configuration is incomplete, and workers resume delivery
+after configuration is corrected and the service restarted with the same database.
+
+## Optional free Telegram relay
+
+The `telegram-relay/` directory contains a Cloudflare Worker for installations
+where the Space cannot reach Telegram directly. It supports the Workers Free
+plan within its quotas. Configure `TELEGRAM_RELAY_URL` and
+`TELEGRAM_RELAY_SECRET` together to use it. Otherwise requests go directly to
+Telegram as before. The Worker holds its own bot token and allows only this
+monitor's five Bot API methods for one numeric chat ID. Both normal delivery and
+authenticated diagnostics use the configured transport.
+
+When the relay is configured, read-only `/diagnostics` and `/status` also accept
+`Authorization: Bearer TELEGRAM_RELAY_SECRET`. The existing webhook-secret
+routes continue to work. This does not authorize incoming publication webhooks.
+
+See `telegram-relay/README.md` for deployment and secret configuration. Verify
+the relay's read-only getMe, getChat and getChatMember checks before rebuilding
+the Space, then verify an actual publication arrives after deployment. Neither
+the bot token nor the shared secret belongs in source control.
+
+The Worker does not fix HF sleeping or the temporary queue. Do not rebuild the
+running Space until pending publications have been recovered or their loss and
+replay from Medialogia have been explicitly handled. Keep only one app instance.

diagnose_telegram.py

@@ -0,0 +1,65 @@
+import getpass
+import json
+import os
+from pathlib import Path
+import re
+import urllib.error
+import urllib.request
+
+from receiver import Clients, Config, network_error_category
+
+
+def summarize(report):
+    fields = {"method", "ok", "error", "seconds", "chat_type", "permissions",
+              "membership", "is_member", "can_send_messages", "can_post_messages"}
+    return {"status": report.get("status"),
+            "checks": [{key: value for key, value in check.items() if key in fields}
+                       for check in report.get("checks", [])]}
+
+
+def compare(token, chat_id, webhook_secret):
+    config = Config(secret=webhook_secret or "a" * 32,
+                    telegram_token=token, chat_id=chat_id)
+    report = {"mac": summarize(Clients(config).diagnose_telegram())}
+    if webhook_secret:
+        try:
+            request = urllib.request.Request(
+                "https://pranalytics-prtooltest.hf.space/diagnostics/" + webhook_secret)
+            with urllib.request.urlopen(request, timeout=40) as response:
+                report["space"] = summarize(json.loads(response.read(65536)))
+        except urllib.error.HTTPError as exc:
+            report["space"] = {"error": "diagnostics_http_" + str(exc.code)}
+            exc.close()
+        except OSError as exc:
+            report["space"] = {"error": "diagnostics_" + network_error_category(exc)}
+        except (ValueError, TypeError, AttributeError):
+            report["space"] = {"error": "diagnostics_invalid_response"}
+    return report
+
+
+def main():
+    print("Проверка только читает настройки. Сообщения не отправляются.")
+    print("Значения вводятся скрыто и не сохраняются. Используйте значения, заданные при настройке HF.")
+    token = getpass.getpass("TELEGRAM_BOT_TOKEN: ").strip()
+    chat_id = getpass.getpass("TELEGRAM_CHAT_ID: ").strip()
+    webhook_secret = getpass.getpass("WEBHOOK_SECRET (Enter — пропустить проверку Space): ").strip()
+    if not re.fullmatch(r"\d+:[A-Za-z0-9_-]+", token):
+        raise SystemExit("Неверный формат TELEGRAM_BOT_TOKEN.")
+    if not re.fullmatch(r"-?\d+|@[A-Za-z0-9_]+", chat_id):
+        raise SystemExit("Неверный формат TELEGRAM_CHAT_ID.")
+    if webhook_secret and not re.fullmatch(r"[A-Za-z0-9_-]{24,}", webhook_secret):
+        raise SystemExit("Неверный формат WEBHOOK_SECRET.")
+    result = json.dumps(compare(token, chat_id, webhook_secret), ensure_ascii=False, indent=2)
+    destination = Path(__file__).with_name(".local-diagnostics.json")
+    descriptor = os.open(destination, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW, 0o600)
+    with os.fdopen(descriptor, "w") as report_file:
+        report_file.write(result + "\n")
+    print(result)
+    print("Обезличенный отчёт сохранён. Вернитесь в чат и напишите: готово.")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except (KeyboardInterrupt, EOFError):
+        raise SystemExit("Проверка отменена.") from None

receiver.py

@@ -28,7 +28,7 @@ from urllib.parse import urlsplit
 from zoneinfo import ZoneInfo
 
 LOG = logging.getLogger("monitor")
-VERSION = "1.1-diagnostics"
+VERSION = "1.3-telegram-relay"
 MAX_BYTES = 1024 * 1024
 AI_CHUNK = 12000
 DEFAULT_PROMPT = """Ты классификатор публикаций для мониторинга нового внешнего вида
@@ -156,6 +156,8 @@ class Config:
     database_url: str = ""
     db_path: str = "/tmp/medialogia-monitor/queue.sqlite3"
     thread_id: str = ""
+    telegram_relay_url: str = ""
+    telegram_relay_secret: str = ""
 
     @classmethod
     def from_env(cls):
@@ -168,9 +170,21 @@ class Config:
                    model=os.getenv("DEEPSEEK_MODEL", "deepseek-flash").strip(),
                    database_url=os.getenv("DATABASE_URL", "").strip(),
                    db_path=os.getenv("MONITOR_DB_PATH", "/tmp/medialogia-monitor/queue.sqlite3"),
-                   thread_id=os.getenv("TELEGRAM_THREAD_ID", "").strip())
+                   thread_id=os.getenv("TELEGRAM_THREAD_ID", "").strip(),
+                   telegram_relay_url=os.getenv("TELEGRAM_RELAY_URL", "").strip(),
+                   telegram_relay_secret=os.getenv("TELEGRAM_RELAY_SECRET", "").strip())
 
     def validate(self):
+        if bool(self.telegram_relay_url) != bool(self.telegram_relay_secret):
+            raise ValueError("TELEGRAM_RELAY_URL and TELEGRAM_RELAY_SECRET must be set together")
+        if self.telegram_relay_url:
+            relay = urlsplit(self.telegram_relay_url)
+            if (relay.scheme != "https" or not relay.hostname or relay.username or relay.password
+                    or relay.query or relay.fragment or relay.path != "/telegram"
+                    or any(ord(char) < 33 for char in self.telegram_relay_url)):
+                raise ValueError("TELEGRAM_RELAY_URL: expected an HTTPS /telegram endpoint")
+            if not re.fullmatch(r"[A-Za-z0-9_-]{32,}", self.telegram_relay_secret):
+                raise ValueError("TELEGRAM_RELAY_SECRET: use at least 32 ASCII letters/digits/-/_")
         if not re.fullmatch(r"[A-Za-z0-9_-]{24,}", self.secret):
             raise ValueError("WEBHOOK_SECRET: use at least 24 ASCII letters/digits/-/_")
         if self.chat_id and not re.fullmatch(r"-?\d+|@[A-Za-z0-9_]+", self.chat_id):
@@ -189,8 +203,7 @@ class Config:
 
     def missing(self):
         return [name for name, value in (("TELEGRAM_BOT_TOKEN", self.telegram_token),
-                ("TELEGRAM_CHAT_ID", self.chat_id), ("DEEPSEEK_API_KEY", self.deepseek_key),
-                ("ALERT_MENTIONS", self.mentions)) if not value]
+                ("TELEGRAM_CHAT_ID", self.chat_id), ("DEEPSEEK_API_KEY", self.deepseek_key)) if not value]
 
 
 class Store:
@@ -318,7 +331,9 @@ def network_error_category(exc):
 
 def request_json(url, payload, headers=None, service="api", timeout=45):
     request = urllib.request.Request(url, data=json.dumps(payload, ensure_ascii=False).encode(),
-                                     headers={"Content-Type": "application/json", **(headers or {})}, method="POST")
+                                     headers={"Content-Type": "application/json", "Accept": "application/json",
+                                              "User-Agent": "MedialogiaMonitor/" + VERSION,
+                                              **(headers or {})}, method="POST")
     try:
         with urllib.request.urlopen(request, timeout=timeout) as response:
             result = json.loads(response.read(2 * MAX_BYTES))
@@ -353,6 +368,16 @@ class Clients:
     def __init__(self, config):
         self.config, self.last_tg = config, 0.0
         self.diagnostics_lock = threading.Lock()
+        self.telegram_health = {"status": "not_checked"}
+
+    def telegram_request(self, method, payload, timeout):
+        if self.config.telegram_relay_url:
+            return request_json(self.config.telegram_relay_url,
+                                {"method": method, "payload": payload},
+                                {"Authorization": "Bearer " + self.config.telegram_relay_secret},
+                                service="telegram", timeout=timeout)
+        return request_json("https://api.telegram.org/bot" + self.config.telegram_token + "/" + method,
+                            payload, service="telegram", timeout=timeout)
 
     def diagnose_telegram(self):
         """Read-only Bot API checks; no publication, queue changes or AI calls."""
@@ -369,9 +394,7 @@ class Clients:
                     payload["user_id"] = bot_id
                 started = time.monotonic()
                 try:
-                    result = request_json(
-                        "https://api.telegram.org/bot" + self.config.telegram_token + "/" + method,
-                        payload, service="telegram", timeout=10)
+                    result = self.telegram_request(method, payload, timeout=10)
                     if not result.get("ok") or not isinstance(result.get("result"), dict):
                         raise ServiceError("telegram_invalid_response")
                     data = result["result"]
@@ -423,10 +446,14 @@ class Clients:
         data = {"chat_id": self.config.chat_id, **payload}
         if method == "sendMessage" and self.config.thread_id:
             data["message_thread_id"] = int(self.config.thread_id)
-        result = request_json("https://api.telegram.org/bot" + self.config.telegram_token + "/" + method,
-                              data, service="telegram", timeout=35)
-        if not isinstance(result, dict) or not result.get("ok"):
-            raise ServiceError("telegram_invalid_response")
+        try:
+            result = self.telegram_request(method, data, timeout=35)
+            if not isinstance(result, dict) or not result.get("ok"):
+                raise ServiceError("telegram_invalid_response")
+        except ServiceError as exc:
+            self.telegram_health = {"status": "error", "error": exc.code}
+            raise
+        self.telegram_health = {"status": "ok"}
         return result.get("result")
 
 
@@ -509,15 +536,40 @@ class Monitor:
             self.stop.wait(0.25 if row else 1)
 
     def start(self):
-        if not self.config.missing():
-            for lane in ("ai", "tg"):
+        for lane, enabled in (("ai", bool(self.config.deepseek_key)),
+                              ("tg", bool(self.config.telegram_token and self.config.chat_id))):
+            if enabled:
                 thread = threading.Thread(target=self.worker, args=(lane,), daemon=True, name=lane)
                 self.threads.append(thread)
                 thread.start()
 
+    def health(self):
+        missing = self.config.missing()
+        telegram = dict(self.clients.telegram_health)
+        try:
+            queue = self.store.stats()
+        except Exception:
+            return {"status": "storage_unavailable", "version": VERSION}
+        workers = {thread.name: thread.is_alive() for thread in self.threads}
+        expected = {lane for lane, enabled in (("ai", bool(self.config.deepseek_key)),
+                    ("tg", bool(self.config.telegram_token and self.config.chat_id))) if enabled}
+        if missing:
+            state = "configuration_required"
+        elif not all(workers.get(lane, False) for lane in expected):
+            state = "worker_unavailable"
+        elif queue["telegram_errors"] or queue["ai_errors"] or telegram["status"] == "error":
+            state = "monitor_degraded"
+        elif telegram["status"] == "not_checked":
+            state = "monitor_waiting_for_delivery"
+        else:
+            state = "monitor_ready"
+        return {"status": state, "version": VERSION, "missing": missing,
+                "storage": "postgresql" if self.config.database_url else "sqlite",
+                "telegram": telegram, "workers": workers}
+
     def status(self):
         return {"mode": "monitor", "version": VERSION, "missing": self.config.missing(), "model": self.config.model,
-                "storage": "postgresql" if self.config.database_url else "temporary_sqlite",
+                "storage": "postgresql" if self.config.database_url else "sqlite",
                 "queue": self.store.stats(), "workers": {t.name: t.is_alive() for t in self.threads}}
 
 
@@ -546,14 +598,21 @@ def make_server(monitor, host="0.0.0.0", port=7860):
         def authorized(self, prefix):
             return secrets.compare_digest(urlsplit(self.path).path.encode(), (prefix + monitor.config.secret).encode())
 
+        def relay_authorized(self):
+            secret = monitor.config.telegram_relay_secret
+            return bool(secret) and secrets.compare_digest(
+                self.headers.get("Authorization", "").encode(), ("Bearer " + secret).encode())
+
         def do_GET(self):
+            if urlsplit(self.path).path == "/live":
+                return self.reply(200, {"status": "alive", "version": VERSION})
             if urlsplit(self.path).path in ("/", "/health"):
-                missing = monitor.config.missing()
-                return self.reply(200, {"status": "configuration_required" if missing else "monitor_ready",
-                    "version": VERSION, "missing": missing, "storage": "postgresql" if monitor.config.database_url else "temporary_sqlite"})
-            if self.authorized("/diagnostics/"):
+                health = monitor.health()
+                return self.reply(503 if health["status"] in ("storage_unavailable", "worker_unavailable") else 200, health)
+            if self.authorized("/diagnostics/") or (urlsplit(self.path).path == "/diagnostics" and self.relay_authorized()):
                 return self.reply(200, monitor.clients.diagnose_telegram())
-            if self.authorized("/status/") or self.authorized("/samples/"):
+            if (self.authorized("/status/") or self.authorized("/samples/")
+                    or (urlsplit(self.path).path == "/status" and self.relay_authorized())):
                 try:
                     return self.reply(200, monitor.status())
                 except Exception:
@@ -563,8 +622,6 @@ def make_server(monitor, host="0.0.0.0", port=7860):
         def do_POST(self):
             if not self.authorized("/medialogia/"):
                 return self.reply(404, {"error": "not_found"})
-            if monitor.config.missing():
-                return self.reply(503, {"error": "configuration_required"})
             try:
                 size = int(self.headers["Content-Length"])
             except (ValueError, TypeError):
@@ -598,7 +655,7 @@ if __name__ == "__main__":
     app = Monitor(Config.from_env())
     server = make_server(app, port=int(os.getenv("PORT", "7860")))
     app.start()
-    LOG.info("Monitor started; storage=%s; missing=%s", "postgresql" if app.config.database_url else "temporary_sqlite", app.config.missing())
+    LOG.info("Monitor started; storage=%s; missing=%s", "postgresql" if app.config.database_url else "sqlite", app.config.missing())
 
     def shutdown(*_args):
         app.stop.set()
@@ -610,5 +667,8 @@ if __name__ == "__main__":
         server.serve_forever()
     finally:
         app.stop.set()
+        deadline = time.monotonic() + 75
+        for thread in app.threads:
+            thread.join(timeout=max(0, deadline - time.monotonic()))
         server.server_close()
 

telegram-relay/README.md

@@ -0,0 +1,32 @@
+# Telegram delivery relay
+
+This optional Worker forwards the monitor's five Bot API methods to Telegram.
+It requires its own shared secret and accepts chat operations only for one
+numeric chat ID. It does not accept arbitrary destination URLs or bot tokens.
+Request logging is disabled. Do not enable logging of headers or request bodies.
+
+Use the Cloudflare Workers Free plan. Its quotas apply; this is not a guarantee
+of continuous delivery or permanent availability from Hugging Face. HF sleeping
+and temporary SQLite storage remain separate limitations.
+
+After signing into a free Cloudflare account:
+
+```
+npx wrangler login
+npx wrangler deploy
+npx wrangler secret put TELEGRAM_BOT_TOKEN
+npx wrangler secret put TELEGRAM_CHAT_ID
+npx wrangler secret put RELAY_SECRET
+```
+
+Run these commands from this directory. Enter values interactively, never as
+command-line arguments. Use the same bot token and numeric chat ID as the
+monitor. Generate a random shared secret with at least 32 URL-safe characters.
+Store the same secret in HF as `TELEGRAM_RELAY_SECRET` and the deployed HTTPS
+Worker URL ending in `/telegram` as `TELEGRAM_RELAY_URL`. Existing Telegram
+settings remain present in the Space. Never use an anonymous third-party proxy.
+
+Before changing the running Space, verify getMe, getChat and getChatMember via
+the deployed Worker. Preserve or explicitly handle the pending temporary queue
+before rebuilding HF. After updating, rerun the Space's authenticated diagnostics
+and confirm an actual publication arrives in Telegram.

telegram-relay/worker.mjs

@@ -0,0 +1,111 @@
+const allowedMethods = new Set(["getMe", "getChat", "getChatMember", "sendMessage", "editMessageText"]);
+const maxBytes = 65536;
+
+function reply(status, description) {
+  return Response.json({ ok: false, description }, {
+    status,
+    headers: { "Cache-Control": "no-store" },
+  });
+}
+
+async function authorized(actual, expected) {
+  const encoder = new TextEncoder();
+  const [actualHash, expectedHash] = await Promise.all([
+    crypto.subtle.digest("SHA-256", encoder.encode(actual)),
+    crypto.subtle.digest("SHA-256", encoder.encode(expected)),
+  ]);
+  const actualBytes = new Uint8Array(actualHash);
+  const expectedBytes = new Uint8Array(expectedHash);
+  let difference = 0;
+  for (let index = 0; index < actualBytes.length; index += 1) {
+    difference |= actualBytes[index] ^ expectedBytes[index];
+  }
+  return difference === 0;
+}
+
+async function readBody(request) {
+  if (!request.body) throw new Error("empty_body");
+  const reader = request.body.getReader();
+  const chunks = [];
+  let size = 0;
+  while (true) {
+    const { value, done } = await reader.read();
+    if (done) break;
+    size += value.byteLength;
+    if (size > maxBytes) {
+      await reader.cancel();
+      throw new Error("payload_too_large");
+    }
+    chunks.push(value);
+  }
+  const body = new Uint8Array(size);
+  let offset = 0;
+  for (const chunk of chunks) {
+    body.set(chunk, offset);
+    offset += chunk.byteLength;
+  }
+  return JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(body));
+}
+
+export default {
+  async fetch(request, environment) {
+    const url = new URL(request.url);
+    if (url.pathname !== "/telegram" || url.search || request.method !== "POST") {
+      return reply(404, "not_found");
+    }
+    if (!/^\d+:[A-Za-z0-9_-]+$/.test(environment.TELEGRAM_BOT_TOKEN || "")
+        || !/^[A-Za-z0-9_-]{32,}$/.test(environment.RELAY_SECRET || "")
+        || !/^-?\d+$/.test(environment.TELEGRAM_CHAT_ID || "")) {
+      return reply(503, "configuration_required");
+    }
+    if (!await authorized(request.headers.get("Authorization") || "", "Bearer " + environment.RELAY_SECRET)) {
+      return reply(401, "unauthorized");
+    }
+    let input;
+    try {
+      input = await readBody(request);
+    } catch (error) {
+      return reply(error.message === "payload_too_large" ? 413 : 400, "invalid_payload");
+    }
+    if (!input || typeof input !== "object" || !allowedMethods.has(input.method)
+        || !input.payload || typeof input.payload !== "object" || Array.isArray(input.payload)) {
+      return reply(400, "invalid_method_or_payload");
+    }
+    if (input.method === "getMe") {
+      input.payload = {};
+    } else if (String(input.payload.chat_id) !== environment.TELEGRAM_CHAT_ID) {
+      return reply(403, "chat_not_allowed");
+    }
+    try {
+      const response = await fetch(
+        "https://api.telegram.org/bot" + environment.TELEGRAM_BOT_TOKEN + "/" + input.method,
+        {
+          method: "POST",
+          headers: { "Content-Type": "application/json" },
+          body: JSON.stringify(input.payload),
+          redirect: "manual",
+          signal: AbortSignal.timeout(8000),
+        },
+      );
+      if (response.status >= 300 && response.status < 400) return reply(502, "telegram_unexpected_redirect");
+      const body = await response.text();
+      let data;
+      try {
+        data = JSON.parse(body);
+      } catch {
+        return reply(502, "telegram_upstream_non_json_http_" + response.status);
+      }
+      if (!data || typeof data.ok !== "boolean") return reply(502, "telegram_invalid_response");
+      return new Response(body, {
+        status: response.status,
+        headers: { "Content-Type": "application/json", "Cache-Control": "no-store" },
+      });
+    } catch (error) {
+      if (typeof AbortSignal.timeout !== "function") return reply(502, "telegram_timeout_api_unavailable");
+      if (String(error.message).includes("Network connection lost")) return reply(502, "telegram_network_connection_lost");
+      if (String(error.message).includes("resolve")) return reply(502, "telegram_dns_error");
+      const category = ["TimeoutError", "AbortError", "TypeError"].includes(error.name) ? error.name : "unknown";
+      return reply(502, "telegram_upstream_" + category);
+    }
+  },
+};

telegram-relay/wrangler.toml

@@ -0,0 +1,8 @@
+name = "pranalytics-telegram-relay"
+main = "worker.mjs"
+compatibility_date = "2026-10-23"
+workers_dev = true
+preview_urls = false
+
+[observability]
+enabled = false