Backup automatico script del 2026-07-26 07:00

services/telegram-bot/webapp_alert.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""Helper: testo plain + summary state + mirror WebApp per allerte meteo."""
+
+from __future__ import annotations
+
+import logging
+import re
+import sys
+from typing import Any, Dict, Optional
+
+LOGGER = logging.getLogger("webapp_alert")
+
+
+def message_to_plain(message: str, is_html: bool = False) -> str:
+    text = message or ""
+    if is_html:
+        text = re.sub(r"<br\s*/?>", "\n", text, flags=re.I)
+        text = re.sub(r"</p\s*>", "\n", text, flags=re.I)
+        text = re.sub(r"<[^>]+>", "", text)
+    text = (
+        text.replace("*", "")
+        .replace("_", "")
+        .replace("`", "")
+        .replace("&nbsp;", " ")
+        .replace("&lt;", "<")
+        .replace("&gt;", ">")
+        .replace("&amp;", "&")
+    )
+    lines = [ln.rstrip() for ln in text.splitlines()]
+    # collassa righe vuote multiple
+    out: list[str] = []
+    blank = False
+    for ln in lines:
+        if not ln.strip():
+            if not blank:
+                out.append("")
+            blank = True
+        else:
+            out.append(ln.strip())
+            blank = False
+    return "\n".join(out).strip()
+
+
+def remember_summary(state: Optional[Dict[str, Any]], plain: str, limit: int = 3000) -> str:
+    summary = (plain or "").strip()[:limit]
+    if state is not None and summary:
+        state["summary"] = summary
+    return summary
+
+
+def publish_web_alert(
+    message: str,
+    category: str,
+    severity: str = "warning",
+    *,
+    is_html: bool = False,
+    title: Optional[str] = None,
+    state: Optional[Dict[str, Any]] = None,
+) -> bool:
+    """Salva summary nello state (se passato) e notifica la WebApp (solo canale web)."""
+    plain = message_to_plain(message, is_html=is_html)
+    if not plain:
+        return False
+    remember_summary(state, plain)
+    head = (title or plain.split("\n", 1)[0]).strip()[:80] or category
+    body = plain[:2500]
+    try:
+        shared = "/home/daniely/docker/shared"
+        if shared not in sys.path:
+            sys.path.insert(0, shared)
+        from loogle_core.alert_dispatcher import send_web
+
+        ok = send_web(head, body, category=category, severity=severity)
+        if not ok:
+            LOGGER.debug("send_web returned False for %s", category)
+        return bool(ok)
+    except Exception as exc:
+        LOGGER.warning("publish_web_alert failed (%s): %s", category, exc)
+        return False