#!/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"", "\n", text, flags=re.I) text = re.sub(r"", "\n", text, flags=re.I) text = re.sub(r"<[^>]+>", "", text) text = ( text.replace("*", "") .replace("_", "") .replace("`", "") .replace(" ", " ") .replace("<", "<") .replace(">", ">") .replace("&", "&") ) 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