diff --git a/tests/test_publishers.py b/tests/test_publishers.py
index 0d199d1e..723d1354 100644
--- a/tests/test_publishers.py
+++ b/tests/test_publishers.py
@@ -4,16 +4,18 @@
import pytest
from weather_briefing.api_client import LoggedAsyncClient
-from weather_briefing.models import RenderedMessage
-from weather_briefing.publishers import (
+from weather_briefing.data import resources as data_resources
+from weather_briefing.data.resources import ReferenceDataError
+from weather_briefing.delivery import (
DeliveryError,
DeliveryProvider,
+ PlainTextRenderer,
StdoutPublisher,
TelegramPublisher,
- _split_message,
)
-from weather_briefing.reference_data import ReferenceDataError
-from weather_briefing.render import PlainTextRenderer
+from weather_briefing.delivery.telegram import split_message
+from weather_briefing.delivery.telegram_reference import telegram_error_classification
+from weather_briefing.models import RenderedMessage
class NoopPublisher:
@@ -70,13 +72,11 @@ def test_delivery_provider_applies_platform_limit_without_leaking_it_into_config
async def test_telegram_publisher_validates_error_metadata_on_construction(monkeypatch) -> None:
- def fail_validation() -> None:
- raise ReferenceDataError("invalid Telegram metadata")
-
- monkeypatch.setattr("weather_briefing.publishers.telegram_error_classification", fail_validation)
+ monkeypatch.setattr(data_resources, "load_reference_data", lambda filename: {})
+ telegram_error_classification.cache_clear()
async with httpx.AsyncClient() as client:
- with pytest.raises(ReferenceDataError, match="invalid Telegram metadata"):
+ with pytest.raises(ReferenceDataError, match="supported fields"):
TelegramPublisher(client, "runtime-token", "runtime-chat")
@@ -93,7 +93,17 @@ def test_delivery_error_rejects_non_boolean_channel_availability(value) -> None:
def test_split_message_prefers_line_boundary() -> None:
- assert _split_message("first line\nsecond line", 12) == ("first line", "\nsecond line")
+ assert split_message("first line\nsecond line", 12) == ("first line", "\nsecond line")
+
+
+def test_split_message_preserves_markup_without_visible_text() -> None:
+ assert split_message("", 3) == ("",)
+
+
+@pytest.mark.parametrize(("body", "limit"), (("", 0), ("body", 0), ("body", -1)))
+def test_split_message_rejects_non_positive_limit(body: str, limit: int) -> None:
+ with pytest.raises(ValueError, match="must be positive"):
+ split_message(body, limit)
@pytest.mark.parametrize(
@@ -122,7 +132,7 @@ def test_split_message_balances_html_tags(
limit: int,
expected: tuple[str, ...],
) -> None:
- assert _split_message(body, limit) == expected
+ assert split_message(body, limit) == expected
async def test_telegram_publisher_uses_runtime_values(caplog) -> None:
diff --git a/tests/test_reference_data.py b/tests/test_reference_data.py
index eaf11b84..d34e9846 100644
--- a/tests/test_reference_data.py
+++ b/tests/test_reference_data.py
@@ -6,6 +6,7 @@
from weather_briefing import __version__
from weather_briefing.air_quality import health_guidance
+from weather_briefing.data import resources as data_resources
from weather_briefing.data.resources import (
ReferenceDataError,
load_reference_data,
@@ -14,10 +15,10 @@
reference_value,
)
from weather_briefing.data.service_endpoints import NOMINATIM_USER_AGENT
+from weather_briefing.delivery.telegram_reference import telegram_error_classification
from weather_briefing.localization import localization_table
from weather_briefing.reference_data import (
open_meteo_weather_code_descriptions,
- telegram_error_classification,
)
@@ -214,7 +215,7 @@ def test_reference_string_rejects_invalid_value(monkeypatch, value) -> None:
),
)
def test_telegram_error_classification_rejects_invalid_data(monkeypatch, value, message) -> None:
- monkeypatch.setattr("weather_briefing.reference_data.load_reference_data", lambda filename: value)
+ monkeypatch.setattr(data_resources, "load_reference_data", lambda filename: value)
telegram_error_classification.cache_clear()
with pytest.raises(ReferenceDataError, match=message):
diff --git a/tests/test_render.py b/tests/test_render.py
index afc06a47..49604e28 100644
--- a/tests/test_render.py
+++ b/tests/test_render.py
@@ -1,6 +1,7 @@
import pendulum
import pytest
+from weather_briefing.delivery import PlainTextRenderer, TelegramHTMLRenderer
from weather_briefing.models import (
Advice,
AdviceTopic,
@@ -10,7 +11,6 @@
SourceDocument,
Warning,
)
-from weather_briefing.render import PlainTextRenderer, TelegramHTMLRenderer
def test_render_briefing_uses_safe_telegram_html() -> None:
diff --git a/weather_briefing/delivery/__init__.py b/weather_briefing/delivery/__init__.py
new file mode 100644
index 00000000..b80b510a
--- /dev/null
+++ b/weather_briefing/delivery/__init__.py
@@ -0,0 +1,16 @@
+"""Delivery contracts, renderers, and platform adapters."""
+
+from .base import DeliveryError, DeliveryProvider, RenderedTextDiagnostics
+from .renderers import PlainTextRenderer, TelegramHTMLRenderer
+from .stdout import StdoutPublisher
+from .telegram import TelegramPublisher
+
+__all__ = [
+ "DeliveryError",
+ "DeliveryProvider",
+ "PlainTextRenderer",
+ "RenderedTextDiagnostics",
+ "StdoutPublisher",
+ "TelegramHTMLRenderer",
+ "TelegramPublisher",
+]
diff --git a/weather_briefing/delivery/base.py b/weather_briefing/delivery/base.py
new file mode 100644
index 00000000..bdd44ad4
--- /dev/null
+++ b/weather_briefing/delivery/base.py
@@ -0,0 +1,121 @@
+"""Platform-neutral message delivery composition."""
+
+from __future__ import annotations
+
+import logging
+import re
+from dataclasses import dataclass
+from typing import Protocol
+
+from ..models import Article, BriefingResult, RenderedMessage, SourceDocument
+from .renderers import MessageRenderer
+
+_LOGGER = logging.getLogger("weather_briefing.publishers")
+_SAFE_DELIVERY_REASON = re.compile(r"[a-z0-9]+(?:-[a-z0-9]+)*")
+
+
+class Publisher(Protocol):
+ """Transport a rendered message to its destination."""
+
+ async def publish(
+ self,
+ message: RenderedMessage,
+ *,
+ single_message: bool = False,
+ silent: bool = False,
+ ) -> None:
+ """Publish one rendered message with delivery hints."""
+ ...
+
+
+class RenderedTextDiagnostics(Protocol):
+ """Expose the runtime switch for sensitive rendered-text logging."""
+
+ def rendered_text_logging_enabled(self) -> bool:
+ """Return whether sensitive rendered-text logging is enabled."""
+ ...
+
+
+@dataclass(frozen=True, slots=True)
+class DeliveryProvider:
+ """Bind a platform renderer to its message transport."""
+
+ renderer: MessageRenderer
+ publisher: Publisher
+ single_message_limit: int | None = None
+ diagnostics: RenderedTextDiagnostics | None = None
+
+ def briefing_limit(self, configured_limit: int) -> int:
+ """Clamp a configured briefing limit to the platform limit."""
+ if self.single_message_limit is None:
+ return configured_limit
+ return min(configured_limit, self.single_message_limit)
+
+ def render_briefing(
+ self,
+ result: BriefingResult,
+ reference_articles: tuple[Article, ...],
+ context: tuple[SourceDocument, ...],
+ ) -> RenderedMessage:
+ """Render a briefing with the configured platform renderer."""
+ return self.renderer.render_briefing(result, reference_articles, context)
+
+ async def publish_rendered(
+ self,
+ message: RenderedMessage,
+ *,
+ single_message: bool = False,
+ silent: bool = False,
+ ) -> None:
+ """Publish an already rendered message with delivery hints."""
+ log_rendered_text(self.diagnostics, "briefing", message.body)
+ await self.publisher.publish(message, single_message=single_message, silent=silent)
+
+ async def publish_verbatim(self, article: Article, *, silent: bool = False) -> None:
+ """Render and publish one cleaned article without summarization."""
+ message = self.renderer.render_verbatim(article)
+ _LOGGER.debug(
+ "Rendered verbatim message: visible_characters=%d payload_characters=%d",
+ message.visible_length,
+ len(message.body),
+ )
+ log_rendered_text(self.diagnostics, "verbatim", message.body)
+ await self.publisher.publish(message, silent=silent)
+
+ async def publish_alert(self, title: str, body: str) -> None:
+ """Render and publish an operational alert."""
+ message = self.renderer.render_alert(title, body)
+ log_rendered_text(self.diagnostics, "alert", message.body)
+ await self.publisher.publish(message)
+
+
+class DeliveryError(RuntimeError):
+ """Raised without exposing private delivery endpoint details."""
+
+ def __init__(self, message: str, *, reason: str, channel_unavailable: bool = False) -> None:
+ """Describe a delivery failure using a safe structured reason."""
+ if not isinstance(reason, str) or _SAFE_DELIVERY_REASON.fullmatch(reason) is None:
+ raise ValueError("Delivery error reason must be a lowercase kebab-case label")
+ if not isinstance(channel_unavailable, bool):
+ raise TypeError("channel_unavailable must be a bool")
+ super().__init__(message)
+ self.reason = reason
+ self.channel_unavailable = channel_unavailable
+
+
+def log_rendered_text(diagnostics: RenderedTextDiagnostics | None, stage: str, body: str) -> None:
+ """Log application-owned rendered text when its temporary switch is enabled."""
+ if rendered_text_logging_enabled(diagnostics):
+ _LOGGER.debug("Sensitive rendered text diagnostic: stage=%s body=%r", stage, body)
+
+
+def rendered_text_logging_enabled(diagnostics: RenderedTextDiagnostics | None) -> bool:
+ """Read the diagnostic switch without affecting delivery on failure."""
+ if diagnostics is None:
+ return False
+ try:
+ enabled = diagnostics.rendered_text_logging_enabled()
+ except Exception:
+ _LOGGER.warning("Rendered text diagnostic state check failed", exc_info=True)
+ return False
+ return enabled and _LOGGER.isEnabledFor(logging.DEBUG)
diff --git a/weather_briefing/delivery/renderers.py b/weather_briefing/delivery/renderers.py
new file mode 100644
index 00000000..b9066b66
--- /dev/null
+++ b/weather_briefing/delivery/renderers.py
@@ -0,0 +1,219 @@
+"""Platform-specific rendering of validated briefing results."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from html import escape, unescape
+from typing import Protocol
+
+from bs4 import BeautifulSoup
+
+from ..languages import LanguageSupport
+from ..localization import localization_table
+from ..models import (
+ Advice,
+ Article,
+ BriefingResult,
+ Conclusion,
+ RenderedMessage,
+ SourceDocument,
+)
+
+_BRIEFING_LABELS = localization_table("briefing")
+_BRIEFING_LANGUAGE_SUPPORT = LanguageSupport(
+ default="en",
+ supported=tuple(_BRIEFING_LABELS),
+)
+
+
+class MessageRenderer(Protocol):
+ """Render platform-neutral briefing data for one delivery platform."""
+
+ def render_briefing(
+ self,
+ result: BriefingResult,
+ reference_articles: tuple[Article, ...],
+ context: tuple[SourceDocument, ...],
+ ) -> RenderedMessage:
+ """Render a validated briefing and its citable references."""
+ ...
+
+ def render_verbatim(self, article: Article) -> RenderedMessage:
+ """Render an article without summarizing its cleaned content."""
+ ...
+
+ def render_alert(self, title: str, body: str) -> RenderedMessage:
+ """Render an operational alert."""
+ ...
+
+
+class TelegramHTMLRenderer:
+ """Render briefings as Telegram-compatible HTML."""
+
+ def render_briefing(
+ self,
+ result: BriefingResult,
+ reference_articles: tuple[Article, ...],
+ context: tuple[SourceDocument, ...],
+ ) -> RenderedMessage:
+ """Render a sourced briefing as Telegram HTML."""
+ labels = _briefing_labels(result.output_language)
+ source_links = {
+ article.id: _html_link(article.url, _article_source_name(article)) for article in reference_articles
+ }
+ source_links.update({document.id: _html_link(document.url, document.name) for document in context})
+ lines = [
+ f"{_html_text(result.headline)} "
+ f"{_html_attribution(result.headline_source_ids, source_links, labels)}",
+ "",
+ ]
+ lines.extend(_html_items(labels["weather"], result.conclusions, source_links, labels))
+ if result.active_warnings:
+ lines.extend([f"{labels['warnings']}", ""])
+ lines.extend(
+ (
+ f"• {_html_text(warning.title)}{labels['status_open']}"
+ f"{_html_text(warning.status)}{labels['status_close']}"
+ f"{labels['detail_separator']}{_html_text(warning.detail)} "
+ f"{_html_attribution(warning.source_ids, source_links, labels)}"
+ )
+ for warning in result.active_warnings
+ )
+ lines.append("")
+ lines.extend(_html_items(labels["disasters"], result.disaster_tracking, source_links, labels))
+ lines.extend(_html_items(labels["advice"], result.advice, source_links, labels))
+ return _html_message("\n".join(lines).strip())
+
+ def render_verbatim(self, article: Article) -> RenderedMessage:
+ """Render cleaned article content as Telegram HTML."""
+ return _html_message(
+ "\n".join(
+ (
+ f"{_html_text(article.title)}",
+ "",
+ _html_text(article.content),
+ )
+ )
+ )
+
+ def render_alert(self, title: str, body: str) -> RenderedMessage:
+ """Render an escaped Telegram HTML alert."""
+ return _html_message(f"{_html_text(title)}\n\n{_html_text(body)}")
+
+
+class PlainTextRenderer:
+ """Render briefings for stdout and other plain-text transports."""
+
+ def render_briefing(
+ self,
+ result: BriefingResult,
+ reference_articles: tuple[Article, ...],
+ context: tuple[SourceDocument, ...],
+ ) -> RenderedMessage:
+ """Render a sourced briefing as plain text."""
+ labels = _briefing_labels(result.output_language)
+ source_references = {
+ article.id: f"{_article_source_name(article)}: {article.url}" for article in reference_articles
+ }
+ source_references.update({document.id: f"{document.name}: {document.url}" for document in context})
+ lines = [
+ f"{result.headline} {_plain_attribution(result.headline_source_ids, source_references, labels)}",
+ "",
+ ]
+ lines.extend(_plain_items(labels["weather"], result.conclusions, source_references, labels))
+ if result.active_warnings:
+ lines.extend([labels["warnings"], ""])
+ for warning in result.active_warnings:
+ sources = _plain_attribution(warning.source_ids, source_references, labels)
+ lines.append(
+ f"- {warning.title}{labels['status_open']}{warning.status}{labels['status_close']}"
+ f"{labels['detail_separator']}{warning.detail} {sources}"
+ )
+ lines.append("")
+ lines.extend(_plain_items(labels["disasters"], result.disaster_tracking, source_references, labels))
+ lines.extend(_plain_items(labels["advice"], result.advice, source_references, labels))
+ return _plain_message("\n".join(lines).strip())
+
+ def render_verbatim(self, article: Article) -> RenderedMessage:
+ """Render cleaned article content as plain text."""
+ return _plain_message(f"{article.title}\n\n{article.content}")
+
+ def render_alert(self, title: str, body: str) -> RenderedMessage:
+ """Render an operational alert as plain text."""
+ return _plain_message(f"{title}\n\n{body}")
+
+
+def _html_text(value: str) -> str:
+ return escape(unescape(value), quote=False)
+
+
+def _article_source_name(article: Article) -> str:
+ return article.source_name.strip() or article.source_id
+
+
+def _html_link(url: str, label: str) -> str:
+ return f'{_html_text(label)}'
+
+
+def _html_items(
+ title: str,
+ items: tuple[Conclusion | Advice, ...],
+ source_links: dict[str, str],
+ labels: Mapping[str, str],
+) -> list[str]:
+ if not items:
+ return []
+ lines = [f"{_html_text(title)}", ""]
+ lines.extend(
+ f"• {_html_text(item.text)} {_html_attribution(item.source_ids, source_links, labels)}" for item in items
+ )
+ lines.append("")
+ return lines
+
+
+def _plain_items(
+ title: str,
+ items: tuple[Conclusion | Advice, ...],
+ source_references: dict[str, str],
+ labels: Mapping[str, str],
+) -> list[str]:
+ if not items:
+ return []
+ lines = [title, ""]
+ lines.extend(f"- {item.text} {_plain_attribution(item.source_ids, source_references, labels)}" for item in items)
+ lines.append("")
+ return lines
+
+
+def _html_attribution(
+ source_ids: tuple[str, ...],
+ source_links: dict[str, str],
+ labels: Mapping[str, str],
+) -> str:
+ sources = labels["html_source_separator"].join(dict.fromkeys(source_links[source_id] for source_id in source_ids))
+ return labels["attribution"].format(sources=sources)
+
+
+def _plain_attribution(
+ source_ids: tuple[str, ...],
+ source_references: dict[str, str],
+ labels: Mapping[str, str],
+) -> str:
+ sources = labels["plain_source_separator"].join(
+ dict.fromkeys(source_references[source_id] for source_id in source_ids)
+ )
+ return labels["attribution"].format(sources=sources)
+
+
+def _briefing_labels(language: str) -> Mapping[str, str]:
+ selected = _BRIEFING_LANGUAGE_SUPPORT.match(language)
+ return _BRIEFING_LABELS[selected]
+
+
+def _html_message(body: str) -> RenderedMessage:
+ visible = BeautifulSoup(body, "html.parser").get_text()
+ return RenderedMessage(body=body, visible_length=len(visible))
+
+
+def _plain_message(body: str) -> RenderedMessage:
+ return RenderedMessage(body=body, visible_length=len(body))
diff --git a/weather_briefing/delivery/stdout.py b/weather_briefing/delivery/stdout.py
new file mode 100644
index 00000000..dc9f6c72
--- /dev/null
+++ b/weather_briefing/delivery/stdout.py
@@ -0,0 +1,19 @@
+"""Standard-output delivery adapter."""
+
+from __future__ import annotations
+
+from ..models import RenderedMessage
+
+
+class StdoutPublisher:
+ """Write rendered messages to standard output."""
+
+ async def publish(
+ self,
+ message: RenderedMessage,
+ *,
+ single_message: bool = False,
+ silent: bool = False,
+ ) -> None:
+ """Print the rendered body and ignore platform delivery hints."""
+ print(message.body)
diff --git a/weather_briefing/delivery/telegram.py b/weather_briefing/delivery/telegram.py
new file mode 100644
index 00000000..583689eb
--- /dev/null
+++ b/weather_briefing/delivery/telegram.py
@@ -0,0 +1,229 @@
+"""Telegram Bot API delivery adapter."""
+
+from __future__ import annotations
+
+import logging
+from html.parser import HTMLParser
+
+import httpx
+
+from ..api_client import api_call_extensions
+from ..models import RenderedMessage
+from . import telegram_reference
+from .base import DeliveryError, RenderedTextDiagnostics, rendered_text_logging_enabled
+
+_LOGGER = logging.getLogger("weather_briefing.publishers")
+
+
+class TelegramPublisher:
+ """Publish rendered HTML messages through the Telegram Bot API."""
+
+ MAX_MESSAGE_LENGTH = 4096
+
+ def __init__(
+ self,
+ client: httpx.AsyncClient,
+ token: str,
+ chat_id: str,
+ diagnostics: RenderedTextDiagnostics | None = None,
+ ) -> None:
+ """Configure Telegram delivery and optional sensitive-text diagnostics."""
+ telegram_reference.telegram_error_classification()
+ self._client = client
+ self._url = f"https://api.telegram.org/bot{token}/sendMessage"
+ self._chat_id = chat_id
+ self._diagnostics = diagnostics
+
+ async def publish(
+ self,
+ message: RenderedMessage,
+ *,
+ single_message: bool = False,
+ silent: bool = False,
+ ) -> None:
+ """Publish one message, splitting it only when allowed."""
+ if single_message and message.visible_length > self.MAX_MESSAGE_LENGTH:
+ raise DeliveryError(
+ "Telegram single message exceeds the platform limit",
+ reason="message-too-long",
+ )
+ chunks = (message.body,) if single_message else split_message(message.body, self.MAX_MESSAGE_LENGTH)
+ _LOGGER.info(
+ "Telegram delivery prepared: visible_characters=%d payload_characters=%d chunks=%d "
+ "single_message=%s silent=%s",
+ message.visible_length,
+ len(message.body),
+ len(chunks),
+ single_message,
+ silent,
+ )
+ log_rendered_text = rendered_text_logging_enabled(self._diagnostics)
+ for index, chunk in enumerate(chunks, start=1):
+ if log_rendered_text:
+ _LOGGER.debug(
+ "Sensitive rendered text diagnostic: stage=telegram-chunk-%d-of-%d body=%r",
+ index,
+ len(chunks),
+ chunk,
+ )
+ try:
+ response = await self._client.post(
+ self._url,
+ json={
+ "chat_id": self._chat_id,
+ "text": chunk,
+ "parse_mode": "HTML",
+ "link_preview_options": {"is_disabled": True},
+ "disable_notification": silent,
+ },
+ extensions=api_call_extensions(
+ "telegram",
+ "send-message",
+ response_error_handled=True,
+ ),
+ )
+ response.raise_for_status()
+ except httpx.HTTPStatusError as exc:
+ reason, channel_unavailable = telegram_error_reason(exc.response)
+ _LOGGER.warning(
+ "Telegram delivery rejected index=%d/%d message_visible_characters=%d payload_characters=%d "
+ "status_code=%d reason=%s",
+ index,
+ len(chunks),
+ message.visible_length,
+ len(chunk),
+ exc.response.status_code,
+ reason,
+ )
+ raise DeliveryError(
+ f"Telegram delivery failed ({reason})",
+ reason=reason,
+ channel_unavailable=channel_unavailable,
+ ) from None
+ except httpx.RequestError as exc:
+ _LOGGER.info(
+ "Telegram delivery request failed index=%d/%d message_visible_characters=%d payload_characters=%d "
+ "reason=%s",
+ index,
+ len(chunks),
+ message.visible_length,
+ len(chunk),
+ type(exc).__name__,
+ )
+ raise DeliveryError(
+ "Telegram delivery failed (request-error)",
+ reason="request-error",
+ ) from None
+ _LOGGER.debug(
+ "Telegram chunk accepted: index=%d/%d payload_characters=%d",
+ index,
+ len(chunks),
+ len(chunk),
+ )
+
+
+def telegram_error_reason(response: httpx.Response) -> tuple[str, bool]:
+ """Classify a Telegram API error without logging its response body."""
+ classification = telegram_reference.telegram_error_classification()
+ try:
+ payload = response.json()
+ except ValueError:
+ payload = None
+
+ if isinstance(payload, dict):
+ parameters = payload.get("parameters")
+ if isinstance(parameters, dict) and type(parameters.get("migrate_to_chat_id")) is int:
+ reason = classification.parameter_reasons["migrate_to_chat_id"]
+ return reason, reason in classification.channel_unavailable_reasons
+
+ description = payload.get("description")
+ if isinstance(description, str):
+ normalized = description.casefold()
+ for marker, reason in classification.description_markers:
+ if marker in normalized:
+ return reason, reason in classification.channel_unavailable_reasons
+
+ reason = classification.status_reasons.get(response.status_code, "api-error")
+ return reason, reason in classification.channel_unavailable_reasons
+
+
+def split_message(body: str, limit: int) -> tuple[str, ...]:
+ """Split Telegram HTML into independently valid chunks."""
+ if limit <= 0:
+ raise ValueError("Message split limit must be positive")
+ if len(body) <= limit:
+ return (body,)
+ chunker = _TelegramHTMLChunker(limit)
+ chunker.feed(body)
+ chunker.close()
+ return chunker.finish() or (body,)
+
+
+class _TelegramHTMLChunker(HTMLParser):
+ """Split Telegram HTML while making every chunk independently valid."""
+
+ def __init__(self, limit: int) -> None:
+ super().__init__(convert_charrefs=False)
+ self._limit = limit
+ self._chunks: list[str] = []
+ self._parts: list[str] = []
+ self._open_tags: list[tuple[str, str]] = []
+ self._visible_length = 0
+
+ def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+ if self._visible_length == self._limit:
+ self._finish_chunk()
+ start_tag = self.get_starttag_text()
+ assert start_tag is not None
+ self._parts.append(start_tag)
+ self._open_tags.append((tag, start_tag))
+
+ def handle_endtag(self, tag: str) -> None:
+ self._parts.append(f"")
+ self._open_tags.pop()
+
+ def handle_data(self, data: str) -> None:
+ while True:
+ available = self._limit - self._visible_length
+ if available == 0:
+ self._finish_chunk()
+ available = self._limit
+ if len(data) <= available:
+ self._parts.append(data)
+ self._visible_length += len(data)
+ return
+ split_at = data.rfind("\n", 0, available + 1)
+ if split_at > 0:
+ self._parts.append(data[:split_at])
+ self._visible_length += split_at
+ data = data[split_at:]
+ else:
+ self._parts.append(data[:available])
+ self._visible_length += available
+ data = data[available:]
+ self._finish_chunk()
+
+ def handle_entityref(self, name: str) -> None:
+ self._append_entity(f"&{name};")
+
+ def handle_charref(self, name: str) -> None:
+ self._append_entity(f"&#{name};")
+
+ def finish(self) -> tuple[str, ...]:
+ """Flush and return all accumulated chunks."""
+ self._finish_chunk()
+ return tuple(self._chunks)
+
+ def _append_entity(self, value: str) -> None:
+ if self._visible_length == self._limit:
+ self._finish_chunk()
+ self._parts.append(value)
+ self._visible_length += 1
+
+ def _finish_chunk(self) -> None:
+ if self._visible_length == 0:
+ return
+ closing_tags = (f"" for tag, _ in reversed(self._open_tags))
+ self._chunks.append("".join((*self._parts, *closing_tags)))
+ self._parts = [start_tag for _, start_tag in self._open_tags]
+ self._visible_length = 0
diff --git a/weather_briefing/delivery/telegram_reference.py b/weather_briefing/delivery/telegram_reference.py
new file mode 100644
index 00000000..f0b15b74
--- /dev/null
+++ b/weather_briefing/delivery/telegram_reference.py
@@ -0,0 +1,99 @@
+"""Validated Telegram delivery classification metadata."""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping
+from dataclasses import dataclass
+from functools import cache
+from types import MappingProxyType
+
+from ..data import resources
+from ..data.resources import ReferenceDataError
+
+_CLASSIFICATION_REASON = re.compile(r"[a-z0-9]+(?:-[a-z0-9]+)*")
+
+
+@dataclass(frozen=True, slots=True)
+class TelegramErrorClassification:
+ """Validated Telegram API error mappings and delivery metadata."""
+
+ description_markers: tuple[tuple[str, str], ...]
+ parameter_reasons: Mapping[str, str]
+ status_reasons: Mapping[int, str]
+ channel_unavailable_reasons: frozenset[str]
+
+
+@cache
+def telegram_error_classification() -> TelegramErrorClassification:
+ """Return validated Telegram API error classification data."""
+ value = resources.load_reference_data("telegram_error_classification.json")
+ markers = value.get("description_markers")
+ parameters = value.get("parameter_reasons")
+ statuses = value.get("status_reasons")
+ unavailable_reasons = value.get("channel_unavailable_reasons")
+ if set(value) != {
+ "channel_unavailable_reasons",
+ "description_markers",
+ "parameter_reasons",
+ "status_reasons",
+ }:
+ raise ReferenceDataError("Telegram error classification must contain the supported fields")
+ if not isinstance(markers, dict) or not markers:
+ raise ReferenceDataError("Telegram description markers must map normalized strings to reasons")
+ validated_markers: list[tuple[str, str]] = []
+ for marker, reason in markers.items():
+ if (
+ not isinstance(marker, str)
+ or not marker.strip()
+ or marker != marker.casefold()
+ or not isinstance(reason, str)
+ or _CLASSIFICATION_REASON.fullmatch(reason) is None
+ ):
+ raise ReferenceDataError("Telegram description markers must map normalized strings to reasons")
+ validated_markers.append((marker, reason))
+
+ if not isinstance(parameters, dict) or set(parameters) != {"migrate_to_chat_id"}:
+ raise ReferenceDataError("Telegram parameters must map supported API fields to reasons")
+ migration_reason = parameters.get("migrate_to_chat_id")
+ if not isinstance(migration_reason, str) or _CLASSIFICATION_REASON.fullmatch(migration_reason) is None:
+ raise ReferenceDataError("Telegram parameters must map supported API fields to reasons")
+
+ if not isinstance(statuses, dict) or not statuses:
+ raise ReferenceDataError("Telegram statuses must map HTTP error codes to reasons")
+ validated_statuses: dict[int, str] = {}
+ for status, reason in statuses.items():
+ if (
+ not isinstance(status, str)
+ or not status.isascii()
+ or not status.isdigit()
+ or not 400 <= int(status) <= 599
+ or not isinstance(reason, str)
+ or _CLASSIFICATION_REASON.fullmatch(reason) is None
+ ):
+ raise ReferenceDataError("Telegram statuses must map HTTP error codes to reasons")
+ validated_statuses[int(status)] = reason
+
+ known_reasons = {
+ *(reason for _, reason in validated_markers),
+ migration_reason,
+ *validated_statuses.values(),
+ }
+ if not isinstance(unavailable_reasons, list) or not unavailable_reasons:
+ raise ReferenceDataError("Telegram channel availability must reference known reasons")
+ validated_unavailable_reasons: list[str] = []
+ for reason in unavailable_reasons:
+ if not isinstance(reason, str) or _CLASSIFICATION_REASON.fullmatch(reason) is None:
+ raise ReferenceDataError("Telegram channel availability must reference known reasons")
+ validated_unavailable_reasons.append(reason)
+ if len(set(validated_unavailable_reasons)) != len(validated_unavailable_reasons) or not set(
+ validated_unavailable_reasons
+ ).issubset(known_reasons):
+ raise ReferenceDataError("Telegram channel availability must reference known reasons")
+
+ return TelegramErrorClassification(
+ description_markers=tuple(validated_markers),
+ parameter_reasons=MappingProxyType({"migrate_to_chat_id": migration_reason}),
+ status_reasons=MappingProxyType(validated_statuses),
+ channel_unavailable_reasons=frozenset(validated_unavailable_reasons),
+ )
diff --git a/weather_briefing/publishers.py b/weather_briefing/publishers.py
index ce0d5c1a..e4712a39 100644
--- a/weather_briefing/publishers.py
+++ b/weather_briefing/publishers.py
@@ -1,352 +1,19 @@
-"""Delivery composition and message publisher adapters."""
-
-from __future__ import annotations
-
-import logging
-import re
-from dataclasses import dataclass
-from html.parser import HTMLParser
-from typing import Protocol
-
-import httpx
-
-from .api_client import api_call_extensions
-from .models import Article, BriefingResult, RenderedMessage, SourceDocument
-from .reference_data import telegram_error_classification
-from .render import MessageRenderer
-
-_LOGGER = logging.getLogger("weather_briefing.publishers")
-_SAFE_DELIVERY_REASON = re.compile(r"[a-z0-9]+(?:-[a-z0-9]+)*")
-
-
-class Publisher(Protocol):
- """Transport a rendered message to its destination."""
-
- async def publish(
- self,
- message: RenderedMessage,
- *,
- single_message: bool = False,
- silent: bool = False,
- ) -> None:
- """Publish one rendered message with delivery hints."""
- ...
-
-
-class RenderedTextDiagnostics(Protocol):
- """Expose the runtime switch for sensitive rendered-text logging."""
-
- def rendered_text_logging_enabled(self) -> bool:
- """Return whether sensitive rendered-text logging is enabled."""
- ...
-
-
-@dataclass(frozen=True, slots=True)
-class DeliveryProvider:
- """Bind a platform renderer to its message transport."""
-
- renderer: MessageRenderer
- publisher: Publisher
- single_message_limit: int | None = None
- diagnostics: RenderedTextDiagnostics | None = None
-
- def briefing_limit(self, configured_limit: int) -> int:
- """Clamp a configured briefing limit to the platform limit."""
- if self.single_message_limit is None:
- return configured_limit
- return min(configured_limit, self.single_message_limit)
-
- def render_briefing(
- self,
- result: BriefingResult,
- reference_articles: tuple[Article, ...],
- context: tuple[SourceDocument, ...],
- ) -> RenderedMessage:
- """Render a briefing with the configured platform renderer."""
- return self.renderer.render_briefing(result, reference_articles, context)
-
- async def publish_rendered(
- self,
- message: RenderedMessage,
- *,
- single_message: bool = False,
- silent: bool = False,
- ) -> None:
- """Publish an already rendered message with delivery hints."""
- _log_rendered_text(self.diagnostics, "briefing", message.body)
- await self.publisher.publish(message, single_message=single_message, silent=silent)
-
- async def publish_verbatim(self, article: Article, *, silent: bool = False) -> None:
- """Render and publish one cleaned article without summarization."""
- message = self.renderer.render_verbatim(article)
- _LOGGER.debug(
- "Rendered verbatim message: visible_characters=%d payload_characters=%d",
- message.visible_length,
- len(message.body),
- )
- _log_rendered_text(self.diagnostics, "verbatim", message.body)
- await self.publisher.publish(message, silent=silent)
-
- async def publish_alert(self, title: str, body: str) -> None:
- """Render and publish an operational alert."""
- message = self.renderer.render_alert(title, body)
- _log_rendered_text(self.diagnostics, "alert", message.body)
- await self.publisher.publish(message)
-
-
-class StdoutPublisher:
- """Write rendered messages to standard output."""
-
- async def publish(
- self,
- message: RenderedMessage,
- *,
- single_message: bool = False,
- silent: bool = False,
- ) -> None:
- """Print the rendered body and ignore platform delivery hints."""
- print(message.body)
-
-
-class DeliveryError(RuntimeError):
- """Raised without exposing private delivery endpoint details."""
-
- def __init__(self, message: str, *, reason: str, channel_unavailable: bool = False) -> None:
- """Describe a delivery failure using a safe structured reason."""
- if not isinstance(reason, str) or _SAFE_DELIVERY_REASON.fullmatch(reason) is None:
- raise ValueError("Delivery error reason must be a lowercase kebab-case label")
- if not isinstance(channel_unavailable, bool):
- raise TypeError("channel_unavailable must be a bool")
- super().__init__(message)
- self.reason = reason
- self.channel_unavailable = channel_unavailable
-
-
-class TelegramPublisher:
- """Publish rendered HTML messages through the Telegram Bot API."""
-
- MAX_MESSAGE_LENGTH = 4096
-
- def __init__(
- self,
- client: httpx.AsyncClient,
- token: str,
- chat_id: str,
- diagnostics: RenderedTextDiagnostics | None = None,
- ) -> None:
- """Configure Telegram delivery and optional sensitive-text diagnostics."""
- telegram_error_classification()
- self._client = client
- self._url = f"https://api.telegram.org/bot{token}/sendMessage"
- self._chat_id = chat_id
- self._diagnostics = diagnostics
-
- async def publish(
- self,
- message: RenderedMessage,
- *,
- single_message: bool = False,
- silent: bool = False,
- ) -> None:
- """Publish one message, splitting it only when allowed."""
- if single_message and message.visible_length > self.MAX_MESSAGE_LENGTH:
- raise DeliveryError(
- "Telegram single message exceeds the platform limit",
- reason="message-too-long",
- )
- chunks = (message.body,) if single_message else _split_message(message.body, self.MAX_MESSAGE_LENGTH)
- _LOGGER.info(
- "Telegram delivery prepared: visible_characters=%d payload_characters=%d chunks=%d "
- "single_message=%s silent=%s",
- message.visible_length,
- len(message.body),
- len(chunks),
- single_message,
- silent,
- )
- log_rendered_text = _rendered_text_logging_enabled(self._diagnostics)
- for index, chunk in enumerate(chunks, start=1):
- if log_rendered_text:
- _LOGGER.debug(
- "Sensitive rendered text diagnostic: stage=telegram-chunk-%d-of-%d body=%r",
- index,
- len(chunks),
- chunk,
- )
- try:
- response = await self._client.post(
- self._url,
- json={
- "chat_id": self._chat_id,
- "text": chunk,
- "parse_mode": "HTML",
- "link_preview_options": {"is_disabled": True},
- "disable_notification": silent,
- },
- extensions=api_call_extensions(
- "telegram",
- "send-message",
- response_error_handled=True,
- ),
- )
- response.raise_for_status()
- except httpx.HTTPStatusError as exc:
- reason, channel_unavailable = _telegram_error_reason(exc.response)
- _LOGGER.warning(
- "Telegram delivery rejected index=%d/%d message_visible_characters=%d payload_characters=%d "
- "status_code=%d reason=%s",
- index,
- len(chunks),
- message.visible_length,
- len(chunk),
- exc.response.status_code,
- reason,
- )
- raise DeliveryError(
- f"Telegram delivery failed ({reason})",
- reason=reason,
- channel_unavailable=channel_unavailable,
- ) from None
- except httpx.RequestError as exc:
- _LOGGER.info(
- "Telegram delivery request failed index=%d/%d message_visible_characters=%d payload_characters=%d "
- "reason=%s",
- index,
- len(chunks),
- message.visible_length,
- len(chunk),
- type(exc).__name__,
- )
- raise DeliveryError(
- "Telegram delivery failed (request-error)",
- reason="request-error",
- ) from None
- _LOGGER.debug(
- "Telegram chunk accepted: index=%d/%d payload_characters=%d",
- index,
- len(chunks),
- len(chunk),
- )
-
-
-def _log_rendered_text(
- diagnostics: RenderedTextDiagnostics | None,
- stage: str,
- body: str,
-) -> None:
- if _rendered_text_logging_enabled(diagnostics):
- _LOGGER.debug("Sensitive rendered text diagnostic: stage=%s body=%r", stage, body)
-
-
-def _rendered_text_logging_enabled(diagnostics: RenderedTextDiagnostics | None) -> bool:
- if diagnostics is None:
- return False
- try:
- enabled = diagnostics.rendered_text_logging_enabled()
- except Exception:
- _LOGGER.warning("Rendered text diagnostic state check failed", exc_info=True)
- return False
- return enabled and _LOGGER.isEnabledFor(logging.DEBUG)
-
-
-def _telegram_error_reason(response: httpx.Response) -> tuple[str, bool]:
- """Classify a Telegram API error without logging its response body."""
- classification = telegram_error_classification()
- try:
- payload = response.json()
- except ValueError:
- payload = None
-
- if isinstance(payload, dict):
- parameters = payload.get("parameters")
- if isinstance(parameters, dict) and type(parameters.get("migrate_to_chat_id")) is int:
- reason = classification.parameter_reasons["migrate_to_chat_id"]
- return reason, reason in classification.channel_unavailable_reasons
-
- description = payload.get("description")
- if isinstance(description, str):
- normalized = description.casefold()
- for marker, reason in classification.description_markers:
- if marker in normalized:
- return reason, reason in classification.channel_unavailable_reasons
-
- reason = classification.status_reasons.get(response.status_code, "api-error")
- return reason, reason in classification.channel_unavailable_reasons
-
-
-def _split_message(body: str, limit: int) -> tuple[str, ...]:
- if len(body) <= limit:
- return (body,)
- chunker = _TelegramHTMLChunker(limit)
- chunker.feed(body)
- chunker.close()
- return chunker.finish()
-
-
-class _TelegramHTMLChunker(HTMLParser):
- """Split Telegram HTML while making every chunk independently valid."""
-
- def __init__(self, limit: int) -> None:
- super().__init__(convert_charrefs=False)
- self._limit = limit
- self._chunks: list[str] = []
- self._parts: list[str] = []
- self._open_tags: list[tuple[str, str]] = []
- self._visible_length = 0
-
- def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
- if self._visible_length == self._limit:
- self._finish_chunk()
- start_tag = self.get_starttag_text()
- assert start_tag is not None
- self._parts.append(start_tag)
- self._open_tags.append((tag, start_tag))
-
- def handle_endtag(self, tag: str) -> None:
- self._parts.append(f"")
- self._open_tags.pop()
-
- def handle_data(self, data: str) -> None:
- while True:
- available = self._limit - self._visible_length
- if available == 0:
- self._finish_chunk()
- available = self._limit
- if len(data) <= available:
- self._parts.append(data)
- self._visible_length += len(data)
- return
- split_at = data.rfind("\n", 0, available + 1)
- if split_at > 0:
- self._parts.append(data[:split_at])
- self._visible_length += split_at
- data = data[split_at:]
- else:
- self._parts.append(data[:available])
- self._visible_length += available
- data = data[available:]
- self._finish_chunk()
-
- def handle_entityref(self, name: str) -> None:
- self._append_entity(f"&{name};")
-
- def handle_charref(self, name: str) -> None:
- self._append_entity(f"&#{name};")
-
- def finish(self) -> tuple[str, ...]:
- self._finish_chunk()
- return tuple(self._chunks)
-
- def _append_entity(self, value: str) -> None:
- if self._visible_length == self._limit:
- self._finish_chunk()
- self._parts.append(value)
- self._visible_length += 1
-
- def _finish_chunk(self) -> None:
- if self._visible_length == 0:
- return
- closing_tags = (f"" for tag, _ in reversed(self._open_tags))
- self._chunks.append("".join((*self._parts, *closing_tags)))
- self._parts = [start_tag for _, start_tag in self._open_tags]
- self._visible_length = 0
+"""Compatibility exports for delivery providers."""
+
+from .delivery import (
+ DeliveryError,
+ DeliveryProvider,
+ RenderedTextDiagnostics,
+ StdoutPublisher,
+ TelegramPublisher,
+)
+from .delivery.base import Publisher
+
+__all__ = [
+ "DeliveryError",
+ "DeliveryProvider",
+ "Publisher",
+ "RenderedTextDiagnostics",
+ "StdoutPublisher",
+ "TelegramPublisher",
+]
diff --git a/weather_briefing/reference_data.py b/weather_briefing/reference_data.py
index 8c4ff459..bec76daf 100644
--- a/weather_briefing/reference_data.py
+++ b/weather_briefing/reference_data.py
@@ -1,10 +1,8 @@
-"""Compatibility exports for reference data awaiting feature migration."""
+"""Compatibility exports for reference data awaiting weather migration."""
from __future__ import annotations
-import re
from collections.abc import Mapping
-from dataclasses import dataclass
from functools import cache
from types import MappingProxyType
@@ -17,18 +15,6 @@
)
from .localization import localization_table
-_CLASSIFICATION_REASON = re.compile(r"[a-z0-9]+(?:-[a-z0-9]+)*")
-
-
-@dataclass(frozen=True, slots=True)
-class TelegramErrorClassification:
- """Validated Telegram API error mappings and delivery metadata."""
-
- description_markers: tuple[tuple[str, str], ...]
- parameter_reasons: Mapping[str, str]
- status_reasons: Mapping[int, str]
- channel_unavailable_reasons: frozenset[str]
-
@cache
def open_meteo_weather_code_descriptions() -> Mapping[int, str]:
@@ -56,89 +42,12 @@ def open_meteo_weather_code_descriptions() -> Mapping[int, str]:
return MappingProxyType(validated)
-@cache
-def telegram_error_classification() -> TelegramErrorClassification:
- """Return validated Telegram API error classification data."""
- value = load_reference_data("telegram_error_classification.json")
- markers = value.get("description_markers")
- parameters = value.get("parameter_reasons")
- statuses = value.get("status_reasons")
- unavailable_reasons = value.get("channel_unavailable_reasons")
- if set(value) != {
- "channel_unavailable_reasons",
- "description_markers",
- "parameter_reasons",
- "status_reasons",
- }:
- raise ReferenceDataError("Telegram error classification must contain the supported fields")
- if not isinstance(markers, dict) or not markers:
- raise ReferenceDataError("Telegram description markers must map normalized strings to reasons")
- validated_markers: list[tuple[str, str]] = []
- for marker, reason in markers.items():
- if (
- not isinstance(marker, str)
- or not marker.strip()
- or marker != marker.casefold()
- or not isinstance(reason, str)
- or _CLASSIFICATION_REASON.fullmatch(reason) is None
- ):
- raise ReferenceDataError("Telegram description markers must map normalized strings to reasons")
- validated_markers.append((marker, reason))
-
- if not isinstance(parameters, dict) or set(parameters) != {"migrate_to_chat_id"}:
- raise ReferenceDataError("Telegram parameters must map supported API fields to reasons")
- migration_reason = parameters.get("migrate_to_chat_id")
- if not isinstance(migration_reason, str) or _CLASSIFICATION_REASON.fullmatch(migration_reason) is None:
- raise ReferenceDataError("Telegram parameters must map supported API fields to reasons")
-
- if not isinstance(statuses, dict) or not statuses:
- raise ReferenceDataError("Telegram statuses must map HTTP error codes to reasons")
- validated_statuses: dict[int, str] = {}
- for status, reason in statuses.items():
- if (
- not isinstance(status, str)
- or not status.isascii()
- or not status.isdigit()
- or not 400 <= int(status) <= 599
- or not isinstance(reason, str)
- or _CLASSIFICATION_REASON.fullmatch(reason) is None
- ):
- raise ReferenceDataError("Telegram statuses must map HTTP error codes to reasons")
- validated_statuses[int(status)] = reason
-
- known_reasons = {
- *(reason for _, reason in validated_markers),
- migration_reason,
- *validated_statuses.values(),
- }
- if not isinstance(unavailable_reasons, list) or not unavailable_reasons:
- raise ReferenceDataError("Telegram channel availability must reference known reasons")
- validated_unavailable_reasons: list[str] = []
- for reason in unavailable_reasons:
- if not isinstance(reason, str) or _CLASSIFICATION_REASON.fullmatch(reason) is None:
- raise ReferenceDataError("Telegram channel availability must reference known reasons")
- validated_unavailable_reasons.append(reason)
- if len(set(validated_unavailable_reasons)) != len(validated_unavailable_reasons) or not set(
- validated_unavailable_reasons
- ).issubset(known_reasons):
- raise ReferenceDataError("Telegram channel availability must reference known reasons")
-
- return TelegramErrorClassification(
- description_markers=tuple(validated_markers),
- parameter_reasons=MappingProxyType({"migrate_to_chat_id": migration_reason}),
- status_reasons=MappingProxyType(validated_statuses),
- channel_unavailable_reasons=frozenset(validated_unavailable_reasons),
- )
-
-
__all__ = [
"ReferenceDataError",
- "TelegramErrorClassification",
"load_reference_data",
"localization_table",
"open_meteo_weather_code_descriptions",
"reference_string",
"reference_string_tuple",
"reference_value",
- "telegram_error_classification",
]
diff --git a/weather_briefing/render.py b/weather_briefing/render.py
index 2315798e..f1d67bbc 100644
--- a/weather_briefing/render.py
+++ b/weather_briefing/render.py
@@ -1,219 +1,6 @@
-"""Platform-specific rendering of validated briefing results."""
+"""Compatibility exports for delivery renderers."""
-from __future__ import annotations
+from .delivery import PlainTextRenderer, TelegramHTMLRenderer
+from .delivery.renderers import MessageRenderer
-from collections.abc import Mapping
-from html import escape, unescape
-from typing import Protocol
-
-from bs4 import BeautifulSoup
-
-from .languages import LanguageSupport
-from .models import (
- Advice,
- Article,
- BriefingResult,
- Conclusion,
- RenderedMessage,
- SourceDocument,
-)
-from .reference_data import localization_table
-
-_BRIEFING_LABELS = localization_table("briefing")
-_BRIEFING_LANGUAGE_SUPPORT = LanguageSupport(
- default="en",
- supported=tuple(_BRIEFING_LABELS),
-)
-
-
-class MessageRenderer(Protocol):
- """Render platform-neutral briefing data for one delivery platform."""
-
- def render_briefing(
- self,
- result: BriefingResult,
- reference_articles: tuple[Article, ...],
- context: tuple[SourceDocument, ...],
- ) -> RenderedMessage:
- """Render a validated briefing and its citable references."""
- ...
-
- def render_verbatim(self, article: Article) -> RenderedMessage:
- """Render an article without summarizing its cleaned content."""
- ...
-
- def render_alert(self, title: str, body: str) -> RenderedMessage:
- """Render an operational alert."""
- ...
-
-
-class TelegramHTMLRenderer:
- """Render briefings as Telegram-compatible HTML."""
-
- def render_briefing(
- self,
- result: BriefingResult,
- reference_articles: tuple[Article, ...],
- context: tuple[SourceDocument, ...],
- ) -> RenderedMessage:
- """Render a sourced briefing as Telegram HTML."""
- labels = _briefing_labels(result.output_language)
- source_links = {
- article.id: _html_link(article.url, _article_source_name(article)) for article in reference_articles
- }
- source_links.update({document.id: _html_link(document.url, document.name) for document in context})
- lines = [
- f"{_html_text(result.headline)} "
- f"{_html_attribution(result.headline_source_ids, source_links, labels)}",
- "",
- ]
- lines.extend(_html_items(labels["weather"], result.conclusions, source_links, labels))
- if result.active_warnings:
- lines.extend([f"{labels['warnings']}", ""])
- lines.extend(
- (
- f"• {_html_text(warning.title)}{labels['status_open']}"
- f"{_html_text(warning.status)}{labels['status_close']}"
- f"{labels['detail_separator']}{_html_text(warning.detail)} "
- f"{_html_attribution(warning.source_ids, source_links, labels)}"
- )
- for warning in result.active_warnings
- )
- lines.append("")
- lines.extend(_html_items(labels["disasters"], result.disaster_tracking, source_links, labels))
- lines.extend(_html_items(labels["advice"], result.advice, source_links, labels))
- return _html_message("\n".join(lines).strip())
-
- def render_verbatim(self, article: Article) -> RenderedMessage:
- """Render cleaned article content as Telegram HTML."""
- return _html_message(
- "\n".join(
- (
- f"{_html_text(article.title)}",
- "",
- _html_text(article.content),
- )
- )
- )
-
- def render_alert(self, title: str, body: str) -> RenderedMessage:
- """Render an escaped Telegram HTML alert."""
- return _html_message(f"{_html_text(title)}\n\n{_html_text(body)}")
-
-
-class PlainTextRenderer:
- """Render briefings for stdout and other plain-text transports."""
-
- def render_briefing(
- self,
- result: BriefingResult,
- reference_articles: tuple[Article, ...],
- context: tuple[SourceDocument, ...],
- ) -> RenderedMessage:
- """Render a sourced briefing as plain text."""
- labels = _briefing_labels(result.output_language)
- source_references = {
- article.id: f"{_article_source_name(article)}: {article.url}" for article in reference_articles
- }
- source_references.update({document.id: f"{document.name}: {document.url}" for document in context})
- lines = [
- f"{result.headline} {_plain_attribution(result.headline_source_ids, source_references, labels)}",
- "",
- ]
- lines.extend(_plain_items(labels["weather"], result.conclusions, source_references, labels))
- if result.active_warnings:
- lines.extend([labels["warnings"], ""])
- for warning in result.active_warnings:
- sources = _plain_attribution(warning.source_ids, source_references, labels)
- lines.append(
- f"- {warning.title}{labels['status_open']}{warning.status}{labels['status_close']}"
- f"{labels['detail_separator']}{warning.detail} {sources}"
- )
- lines.append("")
- lines.extend(_plain_items(labels["disasters"], result.disaster_tracking, source_references, labels))
- lines.extend(_plain_items(labels["advice"], result.advice, source_references, labels))
- return _plain_message("\n".join(lines).strip())
-
- def render_verbatim(self, article: Article) -> RenderedMessage:
- """Render cleaned article content as plain text."""
- return _plain_message(f"{article.title}\n\n{article.content}")
-
- def render_alert(self, title: str, body: str) -> RenderedMessage:
- """Render an operational alert as plain text."""
- return _plain_message(f"{title}\n\n{body}")
-
-
-def _html_text(value: str) -> str:
- return escape(unescape(value), quote=False)
-
-
-def _article_source_name(article: Article) -> str:
- return article.source_name.strip() or article.source_id
-
-
-def _html_link(url: str, label: str) -> str:
- return f'{_html_text(label)}'
-
-
-def _html_items(
- title: str,
- items: tuple[Conclusion | Advice, ...],
- source_links: dict[str, str],
- labels: Mapping[str, str],
-) -> list[str]:
- if not items:
- return []
- lines = [f"{_html_text(title)}", ""]
- lines.extend(
- f"• {_html_text(item.text)} {_html_attribution(item.source_ids, source_links, labels)}" for item in items
- )
- lines.append("")
- return lines
-
-
-def _plain_items(
- title: str,
- items: tuple[Conclusion | Advice, ...],
- source_references: dict[str, str],
- labels: Mapping[str, str],
-) -> list[str]:
- if not items:
- return []
- lines = [title, ""]
- lines.extend(f"- {item.text} {_plain_attribution(item.source_ids, source_references, labels)}" for item in items)
- lines.append("")
- return lines
-
-
-def _html_attribution(
- source_ids: tuple[str, ...],
- source_links: dict[str, str],
- labels: Mapping[str, str],
-) -> str:
- sources = labels["html_source_separator"].join(dict.fromkeys(source_links[source_id] for source_id in source_ids))
- return labels["attribution"].format(sources=sources)
-
-
-def _plain_attribution(
- source_ids: tuple[str, ...],
- source_references: dict[str, str],
- labels: Mapping[str, str],
-) -> str:
- sources = labels["plain_source_separator"].join(
- dict.fromkeys(source_references[source_id] for source_id in source_ids)
- )
- return labels["attribution"].format(sources=sources)
-
-
-def _briefing_labels(language: str) -> Mapping[str, str]:
- selected = _BRIEFING_LANGUAGE_SUPPORT.match(language)
- return _BRIEFING_LABELS[selected]
-
-
-def _html_message(body: str) -> RenderedMessage:
- visible = BeautifulSoup(body, "html.parser").get_text()
- return RenderedMessage(body=body, visible_length=len(visible))
-
-
-def _plain_message(body: str) -> RenderedMessage:
- return RenderedMessage(body=body, visible_length=len(body))
+__all__ = ["MessageRenderer", "PlainTextRenderer", "TelegramHTMLRenderer"]