Add notifications module and update response scaffold for pattern notifications

Author: njbbaer

src/notifications.py

@@ -0,0 +1,20 @@
+import asyncio
+import contextvars
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .telegram.telegram_context import TelegramContext
+
+_ctx: contextvars.ContextVar["TelegramContext | None"] = contextvars.ContextVar(
+    "telegram_context", default=None
+)
+
+
+def set_context(ctx: "TelegramContext") -> None:
+    _ctx.set(ctx)
+
+
+def send(message: str) -> None:
+    ctx = _ctx.get()
+    if ctx:
+        asyncio.create_task(ctx.send_message(message))  # noqa: RUF006

src/response_scaffold.py

@@ -1,5 +1,6 @@
 import re
 
+from . import notifications
 from .scaffold_config import ScaffoldConfig
 
 
@@ -36,8 +37,10 @@ def _transform(self) -> str:
         return re.sub(r"\n{3,}", "\n\n", content).strip()
 
     def _apply_replace_patterns(self, content: str) -> str:
-        for pattern, replacement in self.config.replace_patterns.items():
-            content = re.sub(pattern, replacement, content, flags=re.DOTALL)
+        for rp in self.config.replace_patterns:
+            if rp.notify and re.search(rp.pattern, content, flags=re.DOTALL):
+                notifications.send(rp.notify)
+            content = re.sub(rp.pattern, rp.replacement, content, flags=re.DOTALL)
         return content
 
     def _validate_required_tags(self, content: str) -> None:

src/scaffold_config.py

@@ -2,18 +2,34 @@
 from typing import Any
 
 
+@dataclass
+class ReplacePattern:
+    pattern: str
+    replacement: str
+    notify: str | None = None
+
+
 @dataclass
 class ScaffoldConfig:
     require_tags: set[str] = field(default_factory=set)
     delete_tags: set[str] = field(default_factory=set)
     rename_tags: dict[str, str] = field(default_factory=dict)
-    replace_patterns: dict[str, str] = field(default_factory=dict)
+    replace_patterns: list[ReplacePattern] = field(default_factory=list)
 
     @classmethod
     def from_dict(cls, data: dict[str, Any]) -> "ScaffoldConfig":
+        raw_patterns = data.get("replace_patterns", [])
+        patterns = [
+            ReplacePattern(
+                pattern=p["pattern"],
+                replacement=p["replacement"],
+                notify=p.get("notify"),
+            )
+            for p in raw_patterns
+        ]
         return cls(
             require_tags=set(data.get("require_tags", [])),
             delete_tags=set(data.get("delete_tags", [])),
             rename_tags=data.get("rename_tags", {}),
-            replace_patterns=data.get("replace_patterns", {}),
+            replace_patterns=patterns,
         )

src/telegram/message_handler.py

@@ -5,6 +5,7 @@
 import backoff
 from telegram.error import TimedOut
 
+from .. import notifications
 from .telegram_context import TelegramContext
 
 
@@ -13,6 +14,7 @@ def message_handler(
 ) -> Callable[..., Awaitable[Any]]:
     async def wrapper(self, update, context, *args, **kwargs) -> Any:
         ctx = TelegramContext(self.app, update, context)
+        notifications.set_context(ctx)
         typing_task = asyncio.create_task(_loop_send_typing_action(ctx))
         try:
             await func(self, ctx, *args, **kwargs)

tests/unit/test_response_scaffold.py

@@ -1,7 +1,7 @@
 import pytest
 
 from src.response_scaffold import ResponseScaffold
-from src.scaffold_config import ScaffoldConfig
+from src.scaffold_config import ReplacePattern, ScaffoldConfig
 
 
 def test_deletes_tags():
@@ -38,7 +38,8 @@ def test_display_returns_content_outside_tags():
 
 
 def test_apply_replace_patterns():
-    config = ScaffoldConfig(replace_patterns={r"foo\s+bar": "baz"})
+    pattern = ReplacePattern(pattern=r"foo\s+bar", replacement="baz")
+    config = ScaffoldConfig(replace_patterns=[pattern])
     content = "before foo   bar after"
     scaffold = ResponseScaffold(content, config)
     assert scaffold.transformed_content == "before baz after"