Improved code quality with more idiomatic Python patterns.

owncastsentry/commands.py

@@ -83,9 +83,7 @@ class CommandHandler:
         except sqlite3.IntegrityError:
             # Room is already subscribed.
             await evt.reply(
-                "This room is already subscribed to notifications for "
+                f"This room is already subscribed to notifications for {stream_domain}."
-                + stream_domain
-                + "."
             )
             return
 
@@ -101,9 +99,8 @@ class CommandHandler:
         # All went well! Tell the user.
         self.log.info(f"[{stream_domain}] Subscription added for room {evt.room_id}.")
         await evt.reply(
-            "Subscription added! This room will receive notifications when "
+            f"Subscription added! This room will receive "
-            + stream_domain
+            f"notifications when {stream_domain} goes live."
-            + " goes live."
         )
 
     async def unsubscribe(self, evt: MessageEvent, url: str) -> None:
@@ -125,15 +122,14 @@ class CommandHandler:
                 f"[{stream_domain}] Subscription removed for room {evt.room_id}."
             )
             await evt.reply(
-                "Subscription removed! This room will no "
+                f"Subscription removed! This room will no "
-                "longer receive notifications for " + stream_domain + "."
+                f"longer receive notifications for {stream_domain}."
             )
         else:
             # No, nothing changed. Tell the user.
             await evt.reply(
-                "This room is already not subscribed to notifications for "
+                "This room is already not subscribed to "
-                + stream_domain
+                f"notifications for {stream_domain}."
-                + "."
             )
 
     def _format_duration(self, timestamp_str: str) -> str:
@@ -143,7 +139,7 @@ class CommandHandler:
         :return: Formatted duration string (e.g., "1 hour", "2 days").
         """
         try:
-            timestamp = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00"))
+            timestamp = datetime.fromisoformat(timestamp_str)
             now = datetime.now(UTC)
             delta = now - timestamp
 
@@ -158,7 +154,7 @@ class CommandHandler:
                 return f"{hours} hour{'s' if hours != 1 else ''}"
             days = seconds // 86400
             return f"{days} day{'s' if days != 1 else ''}"
-        except Exception:
+        except ValueError:
             return "unknown duration"
 
     async def subscriptions(self, evt: MessageEvent) -> None:
@@ -183,7 +179,7 @@ class CommandHandler:
 
         # Build the response message body as Markdown
         count = len(subscribed_domains)
-        body_text = f"**Subscriptions for this room ({count}):**\n\n"
+        parts = [f"**Subscriptions for this room ({count}):**\n\n"]
 
         for domain in subscribed_domains:
             # Get the stream state from the database
@@ -192,45 +188,41 @@ class CommandHandler:
                 continue
 
             # Determine stream name (use domain as fallback)
-            stream_name = stream_state.name if stream_state.name else domain
+            stream_name = stream_state.name or domain
             safe_stream_name = sanitize_for_markdown(stream_name)
 
             # Start building this stream's entry with stream name as main bullet
-            body_text += f"- **{safe_stream_name}**  \n"
+            parts.append(f"- **{safe_stream_name}**  \n")
 
             # Add title if stream is online (as a sub-bullet)
             if stream_state.status == StreamStatus.ONLINE and stream_state.title:
                 safe_title = sanitize_for_markdown(stream_state.title)
-                body_text += f"  - Title: {safe_title}  \n"
+                parts.append(f"  - Title: {safe_title}  \n")
 
             # Determine status and duration (as a sub-bullet)
-            if stream_state.status == StreamStatus.ONLINE:
+            match stream_state.status:
-                # Stream is online - use last_connect_time
+                case StreamStatus.ONLINE if stream_state.last_connect_time:
-                if stream_state.last_connect_time:
                     duration = self._format_duration(stream_state.last_connect_time)
-                    body_text += f"  - Status: Online for {duration}  \n"
+                    parts.append(f"  - Status: Online for {duration}  \n")
-            elif stream_state.status == StreamStatus.UNKNOWN:
+                case StreamStatus.UNKNOWN:
-                # Stream status is unknown - instance unreachable
+                    parts.append("  - Status: Unknown (instance unreachable)  \n")
-                body_text += "  - Status: Unknown (instance unreachable)  \n"
+                case StreamStatus.OFFLINE if stream_state.last_disconnect_time:
-            else:
-                # Stream is offline - use last_disconnect_time
-                if stream_state.last_disconnect_time:
                     duration = self._format_duration(stream_state.last_disconnect_time)
-                    body_text += f"  - Status: Offline for {duration}  \n"
+                    parts.append(f"  - Status: Offline for {duration}  \n")
-                else:
+                case StreamStatus.OFFLINE:
-                    body_text += "  - Status: Offline  \n"
+                    parts.append("  - Status: Offline  \n")
 
             # Add stream link (as a sub-bullet)
-            body_text += f"  - Link: https://{domain}\n\n"
+            parts.append(f"  - Link: https://{domain}\n\n")
 
         # Add help text for unsubscribing
-        body_text += (
+        parts.append(
             "\nTo unsubscribe from any of these Owncast "
             "instances, use `!unsubscribe <domain>`"
         )
 
         # Send the response as Markdown
-        await evt.reply(body_text, markdown=True)
+        await evt.reply("".join(parts), markdown=True)
 
     async def live(self, evt: MessageEvent) -> None:
         """List currently live streams in the current room.
@@ -271,28 +263,28 @@ class CommandHandler:
 
         # Build the response message body as Markdown
         count = len(live_streams)
-        body_text = f"**Live Owncast instances ({count}):**\n\n"
+        parts = [f"**Live Owncast instances ({count}):**\n\n"]
 
         for domain, stream_state in live_streams:
             # Determine stream name (use domain as fallback)
-            stream_name = stream_state.name if stream_state.name else domain
+            stream_name = stream_state.name or domain
             safe_stream_name = sanitize_for_markdown(stream_name)
 
             # Start building this stream's entry with stream name as main bullet
-            body_text += f"- **{safe_stream_name}**  \n"
+            parts.append(f"- **{safe_stream_name}**  \n")
 
             # Add title (should be present for live streams)
             if stream_state.title:
                 safe_title = sanitize_for_markdown(stream_state.title)
-                body_text += f"  - Title: {safe_title}  \n"
+                parts.append(f"  - Title: {safe_title}  \n")
 
             # Add status with duration
             if stream_state.last_connect_time:
                 duration = self._format_duration(stream_state.last_connect_time)
-                body_text += f"  - Online for {duration}  \n"
+                parts.append(f"  - Online for {duration}  \n")
 
             # Add stream link
-            body_text += f"  - Link: https://{domain}\n\n"
+            parts.append(f"  - Link: https://{domain}\n\n")
 
         # Send the response as Markdown
-        await evt.reply(body_text.rstrip(), markdown=True)
+        await evt.reply("".join(parts).rstrip(), markdown=True)

owncastsentry/health_checker.py

@@ -22,32 +22,10 @@ if TYPE_CHECKING:
 
     from mautrix.util.async_db import Database
 
+    from .models import UpdateResult
     from .owncast_client import OwncastClient
 
 
-@dataclass
-class UpdateResult:
-    """Result of a stream update cycle."""
-
-    total_streams: int
-    successful_checks: int
-    failed_checks: int
-
-    @property
-    def http_healthy(self) -> bool:
-        """Determine HTTP health based on update results.
-
-        HTTP is considered healthy if:
-        - No streams are subscribed (nothing to check), OR
-        - At least one stream check succeeded
-
-        :return: True if HTTP is considered healthy.
-        """
-        if self.total_streams == 0:
-            return True
-        return self.successful_checks > 0
-
-
 @dataclass
 class HealthStatus:
     """Represents the health status of the plugin."""
@@ -92,7 +70,7 @@ class HealthChecker:
             async with self.db.acquire() as conn:  # type: ignore[var-annotated]
                 await conn.fetchval("SELECT 1")
             return True
-        except Exception as e:
+        except Exception as e:  # broad catch - DB backends raise varied errors
             self.log.warning(f"Database health check failed: {e}")
             return False
 
@@ -126,7 +104,7 @@ class HealthChecker:
         )
 
         # Skip endpoint notification if not configured
-        if not endpoint or not endpoint.strip():
+        if not endpoint.strip():
             self.log.debug("Health check endpoint not configured, skipping report.")
             return
 

owncastsentry/models.py

@@ -91,6 +91,29 @@ class StreamState:
         )
 
 
+@dataclass
+class UpdateResult:
+    """Result of a stream update cycle."""
+
+    total_streams: int
+    successful_checks: int
+    failed_checks: int
+
+    @property
+    def http_healthy(self) -> bool:
+        """Determine HTTP health based on update results.
+
+        HTTP is considered healthy if:
+        - No streams are subscribed (nothing to check), OR
+        - At least one stream check succeeded
+
+        :return: True if HTTP is considered healthy.
+        """
+        if self.total_streams == 0:
+            return True
+        return self.successful_checks > 0
+
+
 @dataclass
 class StreamConfig:
     """Represents the configuration of an Owncast stream."""

owncastsentry/notification_service.py

@@ -21,8 +21,8 @@ from typing import TYPE_CHECKING, Any
 from mautrix.types import MessageType, TextMessageEventContent
 
 from .utils import (
-    CLEANUP_DELETE_THRESHOLD,
+    CLEANUP_DELETE_DAYS,
-    CLEANUP_WARNING_THRESHOLD,
+    CLEANUP_WARNING_DAYS,
     SECONDS_BETWEEN_NOTIFICATIONS,
     sanitize_for_plain_text,
 )
@@ -88,36 +88,18 @@ class NotificationService:
         # Record that we're sending a notification now
         self._record_notification(domain)
 
-        # Get a list of room IDs with active subscriptions to the stream domain
-        room_ids = await self.subscription_repo.get_subscribed_rooms(domain)
-
         # Build the notification message
         body_text = self._format_message(name, title, domain, tags, title_change)
 
-        # Set up counters for statistics
-        successful_notifications = 0
-        failed_notifications = 0
-
         # Send notifications to all subscribed rooms in parallel
-        tasks = [
+        successful, failed = await self._broadcast_to_rooms(domain, body_text)
-            self._send_notification(room_id, body_text, domain) for room_id in room_ids
-        ]
-
-        results = await asyncio.gather(*tasks, return_exceptions=True)
-
-        # Count successes and failures
-        for result in results:
-            if isinstance(result, Exception):
-                failed_notifications += 1
-            else:
-                successful_notifications += 1
 
         # Log completion
         notification_type = "title change" if title_change else "going live"
         self.log.info(
             f"[{domain}] Completed sending {notification_type} "
-            f"notifications! {successful_notifications} succeeded, "
+            f"notifications! {successful} succeeded, "
-            f"{failed_notifications} failed."
+            f"{failed} failed."
         )
 
     async def _send_notification(
@@ -153,36 +135,36 @@ class NotificationService:
         :return: Formatted message body.
         """
         # Use name if available, fallback to domain
-        stream_name = name if name else domain
+        stream_name = name or domain
         safe_stream_name = sanitize_for_plain_text(stream_name)
 
         # Choose message based on notification type
         if title_change:
-            body_text = "📝 " + safe_stream_name + " has changed its stream title!"
+            parts = [f"📝 {safe_stream_name} has changed its stream title!"]
         else:
-            body_text = "🎥 " + safe_stream_name + " is now live!"
+            parts = [f"🎥 {safe_stream_name} is now live!"]
 
         # Add title if present
-        if title != "":
+        if title:
             safe_title = sanitize_for_plain_text(title)
-            body_text += "\nStream Title: " + safe_title
+            parts.append(f"\nStream Title: {safe_title}")
 
         # Add stream URL
-        body_text += "\n\nTo tune in, visit: https://" + domain + "/"
+        parts.append(f"\n\nTo tune in, visit: https://{domain}/")
 
         # Add tags if present
         if tags:
-            safe_tags = []
+            safe_tags = [
-            for tag in tags:
+                safe_tag
-                safe_tag = sanitize_for_plain_text(tag)
+                for tag in tags
-                if safe_tag and not safe_tag.startswith("."):
+                if (safe_tag := sanitize_for_plain_text(tag))
-                    safe_tags.append(safe_tag)
+                and not safe_tag.startswith(".")
+            ]
 
             if safe_tags:
-                body_text += "\n\n"
+                parts.append(f"\n\n{' '.join(f'#{tag}' for tag in safe_tags)}")
-                body_text += " ".join("#" + tag for tag in safe_tags)
 
-        return body_text
+        return "".join(parts)
 
     def get_last_notification_time(self, domain: str) -> float:
         """Get the timestamp of the last notification sent for a domain.
@@ -201,48 +183,50 @@ class NotificationService:
         if domain not in self.notification_timers_cache:
             return True
 
-        seconds_since_last = round(time.time() - self.notification_timers_cache[domain])
+        seconds_since_last = round(
-        return bool(seconds_since_last >= SECONDS_BETWEEN_NOTIFICATIONS)
+            time.monotonic() - self.notification_timers_cache[domain]
+        )
+        return seconds_since_last >= SECONDS_BETWEEN_NOTIFICATIONS
 
     def _record_notification(self, domain: str) -> None:
         """Record that a notification was sent at the current time.
 
         :param domain: The stream domain.
         """
-        self.notification_timers_cache[domain] = time.time()
+        self.notification_timers_cache[domain] = time.monotonic()
+
+    async def _broadcast_to_rooms(self, domain: str, body_text: str) -> tuple[int, int]:
+        """Send a message to all rooms subscribed to a domain.
+
+        :param domain: The stream domain.
+        :param body_text: The message body text.
+        :return: Tuple of (successful, failed) counts.
+        """
+        room_ids = await self.subscription_repo.get_subscribed_rooms(domain)
+        tasks = [
+            self._send_notification(room_id, body_text, domain) for room_id in room_ids
+        ]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        failed = sum(1 for r in results if isinstance(r, Exception))
+        successful = len(results) - failed
+        return successful, failed
 
     async def send_cleanup_warning(self, domain: str) -> None:
         """Send cleanup warning notification to all subscribed rooms.
 
         :param domain: The stream domain.
         """
-        # Get all subscribed rooms
+        remaining_days = CLEANUP_DELETE_DAYS - CLEANUP_WARNING_DAYS
-        room_ids = await self.subscription_repo.get_subscribed_rooms(domain)
-
-        # Build the warning message
-        warning_days = CLEANUP_WARNING_THRESHOLD // (24 * 60)
-        delete_days = CLEANUP_DELETE_THRESHOLD // (24 * 60)
-        remaining_days = delete_days - warning_days
         body_text = (
             "⚠️  Warning: Subscription Cleanup Scheduled\n\n"
             f"The Owncast instance at {domain} has been "
-            f"unreachable for {warning_days} days. If it remains "
+            f"unreachable for {CLEANUP_WARNING_DAYS} days. If it remains "
             f"unreachable for {remaining_days} more days "
-            f"({delete_days} days total), this subscription "
+            f"({CLEANUP_DELETE_DAYS} days total), this subscription "
             f"will be automatically removed."
         )
 
-        # Send to all rooms in parallel
+        successful, failed = await self._broadcast_to_rooms(domain, body_text)
-        tasks = [
-            self._send_notification(room_id, body_text, domain) for room_id in room_ids
-        ]
-
-        results = await asyncio.gather(*tasks, return_exceptions=True)
-
-        # Count successes and failures
-        successful = sum(1 for r in results if not isinstance(r, Exception))
-        failed = sum(1 for r in results if isinstance(r, Exception))
-
         self.log.info(
             f"[{domain}] Sent cleanup warning to {successful} rooms ({failed} failed)."
         )
@@ -252,32 +236,17 @@ class NotificationService:
 
         :param domain: The stream domain.
         """
-        # Get all subscribed rooms
-        room_ids = await self.subscription_repo.get_subscribed_rooms(domain)
-
-        # Build the deletion message
-        delete_days = CLEANUP_DELETE_THRESHOLD // (24 * 60)
         body_text = (
             "🗑️  Subscription Automatically Removed\n\n"
             f"The Owncast instance at {domain} has been "
-            f"unreachable for {delete_days} days and has been "
+            f"unreachable for {CLEANUP_DELETE_DAYS} days and has been "
             f"automatically removed from subscriptions in this "
             f"room.\n\n"
             f"If the instance comes online again and you want to "
             f"resubscribe, run `!subscribe {domain}`."
         )
 
-        # Send to all rooms in parallel
+        successful, failed = await self._broadcast_to_rooms(domain, body_text)
-        tasks = [
-            self._send_notification(room_id, body_text, domain) for room_id in room_ids
-        ]
-
-        results = await asyncio.gather(*tasks, return_exceptions=True)
-
-        # Count successes and failures
-        successful = sum(1 for r in results if not isinstance(r, Exception))
-        failed = sum(1 for r in results if isinstance(r, Exception))
-
         self.log.info(
             f"[{domain}] Sent cleanup deletion notice to "
             f"{successful} rooms ({failed} failed)."

owncastsentry/owncast_client.py

@@ -14,17 +14,21 @@
 
 """HTTP client for querying Owncast instance APIs."""
 
-import json
+from typing import TYPE_CHECKING, Any
-from typing import TYPE_CHECKING
 
 import aiohttp
 
+from .models import StreamConfig, StreamState
+from .utils import (
+    OWNCAST_CONFIG_PATH,
+    OWNCAST_STATUS_PATH,
+    REQUIRED_STATUS_FIELDS,
+    user_agent,
+)
+
 if TYPE_CHECKING:
     import logging
 
-from .models import StreamConfig, StreamState
-from .utils import OWNCAST_CONFIG_PATH, OWNCAST_STATUS_PATH, user_agent
-
 
 class OwncastClient:
     """HTTP client for communicating with Owncast instances."""
@@ -55,6 +59,37 @@ class OwncastClient:
             connector=connector,
         )
 
+    async def _fetch_json(self, domain: str, path: str) -> dict[str, Any] | None:
+        """Fetch and parse JSON from an Owncast API endpoint.
+
+        :param domain: The domain to query.
+        :param path: The API path to request.
+        :return: Parsed JSON response, or None on error.
+        """
+        url = f"https://{domain}{path}"
+        try:
+            async with self.session.get(url, allow_redirects=False) as response:
+                if response.status != 200:
+                    self.log.warning(
+                        f"[{domain}] Response to request on "
+                        f"{path} was not 200, "
+                        f"got {response.status} instead."
+                    )
+                    return None
+                try:
+                    result: dict[str, Any] = await response.json()
+                    return result
+                except (ValueError, aiohttp.ContentTypeError) as e:
+                    self.log.warning(
+                        f"[{domain}] Rejecting response to request on "
+                        f"{path} as could not be "
+                        f"interpreted as JSON: {e}"
+                    )
+                    return None
+        except (aiohttp.ClientError, TimeoutError, OSError) as e:
+            self.log.warning(f"[{domain}] Error making GET request to {path}: {e}")
+            return None
+
     async def get_stream_state(self, domain: str) -> StreamState | None:
         """Get the current stream state for a given domain.
 
@@ -65,51 +100,17 @@ class OwncastClient:
         :return: A StreamState if available, None on error.
         """
         self.log.debug(f"[{domain}] Fetching current stream state...")
-        status_url = "https://" + domain + OWNCAST_STATUS_PATH
+        new_state = await self._fetch_json(domain, OWNCAST_STATUS_PATH)
-
+        if new_state is None:
-        # Make a request to the endpoint
-        try:
-            async with self.session.request(
-                "GET", status_url, allow_redirects=False
-            ) as response:
-                # Check the response code is success
-                if response.status != 200:
-                    self.log.warning(
-                        f"[{domain}] Response to request on "
-                        f"{OWNCAST_STATUS_PATH} was not 200, "
-                        f"got {response.status} instead."
-                    )
-                    return None
-
-                # Try to interpret the response as JSON
-                try:
-                    new_state = json.loads(await response.read())
-                except Exception as e:
-                    self.log.warning(
-                        f"[{domain}] Rejecting response to request on "
-                        f"{OWNCAST_STATUS_PATH} as could not be "
-                        f"interpreted as JSON: {e}"
-                    )
-                    return None
-        except Exception as e:
-            self.log.warning(
-                f"[{domain}] Error making GET request to {OWNCAST_STATUS_PATH}: {e}"
-            )
             return None
 
         # Validate the response contains all basic info needed
-        required_fields = [
+        missing = REQUIRED_STATUS_FIELDS - new_state.keys()
-            "lastConnectTime",
+        if missing:
-            "lastDisconnectTime",
-            "streamTitle",
-            "online",
-        ]
-        for field in required_fields:
-            if field not in new_state:
             self.log.warning(
-                    f"[{domain}] Rejecting response to request "
+                f"[{domain}] Rejecting response to request on "
-                    f"on {OWNCAST_STATUS_PATH} as it does not "
+                f"{OWNCAST_STATUS_PATH} as it is missing "
-                    f"have {field} field."
+                f"fields: {', '.join(sorted(missing))}"
             )
             return None
 
@@ -125,39 +126,10 @@ class OwncastClient:
         :return: A StreamConfig, or None if fetch failed.
         """
         self.log.debug(f"[{domain}] Fetching current stream config...")
-        config_url = "https://" + domain + OWNCAST_CONFIG_PATH
+        config = await self._fetch_json(domain, OWNCAST_CONFIG_PATH)
-
+        if config is None:
-        # Make a request to the endpoint
-        try:
-            async with self.session.request(
-                "GET", config_url, allow_redirects=False
-            ) as response:
-                # Check the response code is success
-                if response.status != 200:
-                    self.log.warning(
-                        f"[{domain}] Response to request on "
-                        f"{OWNCAST_CONFIG_PATH} was not 200, "
-                        f"got {response.status} instead."
-                    )
             return None
 
-                # Try to interpret the response as JSON
-                try:
-                    config = json.loads(await response.read())
-                except Exception as e:
-                    self.log.warning(
-                        f"[{domain}] Rejecting response to request on "
-                        f"{OWNCAST_CONFIG_PATH} as could not be "
-                        f"interpreted as JSON: {e}"
-                    )
-                    return None
-        except Exception as e:
-            self.log.warning(
-                f"[{domain}] Error making GET request to {OWNCAST_CONFIG_PATH}: {e}"
-            )
-            return None
-
-        # Create StreamConfig from response (fields are truncated to max lengths)
         return StreamConfig.from_api_response(config)
 
     async def validate_instance(self, domain: str) -> bool:

owncastsentry/stream_monitor.py

@@ -18,8 +18,7 @@ import asyncio
 import time
 from typing import TYPE_CHECKING
 
-from .health_checker import UpdateResult
+from .models import StreamState, UpdateResult
-from .models import StreamState
 from .utils import (
     CLEANUP_DELETE_THRESHOLD,
     CLEANUP_WARNING_THRESHOLD,
@@ -77,17 +76,14 @@ class StreamMonitor:
         total_streams = len(subscribed_domains)
 
         # Build a list of async tasks for each stream domain
-        tasks = [
+        # Run all stream updates in parallel and collect results
-            asyncio.create_task(self.update_stream(domain))
+        results = await asyncio.gather(
-            for domain in subscribed_domains
+            *(self.update_stream(domain) for domain in subscribed_domains)
-        ]
+        )
-
-        # Run the tasks in parallel and collect results
-        results = await asyncio.gather(*tasks)
 
         # Count successes and failures
-        successful_checks = sum(1 for result in results if result is True)
+        successful_checks = results.count(True)
-        failed_checks = sum(1 for result in results if result is False)
+        failed_checks = results.count(False)
 
         self.log.debug(
             f"Update complete. {successful_checks}/{total_streams} succeeded, "
@@ -159,8 +155,7 @@ class StreamMonitor:
         await self.stream_repo.reset_failure_counter(domain)
 
         # Initialize timer cache entries to prevent KeyError on first access
-        if domain not in self.offline_timer_cache:
+        self.offline_timer_cache.setdefault(domain, 0)
-            self.offline_timer_cache[domain] = 0
 
         # Does the last known stream state lack connect/disconnect?
         if (
@@ -184,7 +179,7 @@ class StreamMonitor:
 
             # Calculate seconds since the stream last went offline
             seconds_since_last_offline = round(
-                time.time() - self.offline_timer_cache[domain]
+                time.monotonic() - self.offline_timer_cache[domain]
             )
 
             # Have we queried this stream before?
@@ -278,7 +273,7 @@ class StreamMonitor:
         ):
             # Yep. This stream is now offline. Log it.
             update_database = True
-            self.offline_timer_cache[domain] = time.time()
+            self.offline_timer_cache[domain] = time.monotonic()
             self.log.info(f"[{domain}] Stream is now offline.")
 
         # Update the database with current stream state, if needed.

owncastsentry/utils.py

@@ -23,6 +23,16 @@ OWNCAST_STATUS_PATH = "/api/status"
 # Path to GetWebConfig API call on Owncast instances
 OWNCAST_CONFIG_PATH = "/api/config"
 
+# Fields that must be present in an Owncast status API response
+REQUIRED_STATUS_FIELDS = frozenset(
+    {
+        "lastConnectTime",
+        "lastDisconnectTime",
+        "streamTitle",
+        "online",
+    }
+)
+
 
 def user_agent(version: str) -> str:
     """Build the User-Agent header string for HTTP requests.
@@ -50,9 +60,13 @@ SECONDS_BETWEEN_NOTIFICATIONS = 20 * 60  # 20 minutes in seconds
 #   online after, it's treated as regular going live.
 TEMPORARY_OFFLINE_NOTIFICATION_COOLDOWN = 7 * 60  # 7 min in seconds
 
-# Counter thresholds for auto-cleanup (60-second polling intervals)
+# Auto-cleanup timing (days of continuous unreachability)
-CLEANUP_WARNING_THRESHOLD = 83 * 24 * 60  # 119,520 cycles = 83 days
+CLEANUP_WARNING_DAYS = 83
-CLEANUP_DELETE_THRESHOLD = 90 * 24 * 60  # 129,600 cycles = 90 days
+CLEANUP_DELETE_DAYS = 90
+
+# Counter thresholds derived from days (60-second polling intervals)
+CLEANUP_WARNING_THRESHOLD = CLEANUP_WARNING_DAYS * 24 * 60
+CLEANUP_DELETE_THRESHOLD = CLEANUP_DELETE_DAYS * 24 * 60
 
 # Failure counter threshold for treating stream status as "unknown"
 UNKNOWN_STATUS_THRESHOLD = 15
@@ -109,7 +123,7 @@ def domainify(url: str) -> str:
 
     # Prepend // if no scheme so urlparse treats input as netloc
     if not url.startswith(("http://", "https://", "//")):
-        url = "//" + url
+        url = f"//{url}"
 
     parsed = urlparse(url)
     domain = (parsed.netloc or parsed.path).lower()
@@ -133,6 +147,9 @@ def truncate(text: str, max_length: int) -> str:
     return text[:max_length]
 
 
+_MARKDOWN_ESCAPE_TABLE = str.maketrans({c: f"\\{c}" for c in r"\*_[]()~`#+-=|{}.!<>&"})
+
+
 def escape_markdown(text: str) -> str:
     """Escape Markdown special characters to prevent injection attacks.
 
@@ -146,38 +163,7 @@ def escape_markdown(text: str) -> str:
     if not text:
         return text
 
-    # Escape Markdown special characters by prefixing with backslash
+    return text.translate(_MARKDOWN_ESCAPE_TABLE)
-    # Covers: formatting (*_~`), links ([]()), headings (#), lists (-+),
-    # blockquotes (>), code blocks (```), and other special characters
-    special_chars = {
-        "\\": "\\\\",  # Backslash must be first to avoid double-escaping
-        "*": "\\*",
-        "_": "\\_",
-        "[": "\\[",
-        "]": "\\]",
-        "(": "\\(",
-        ")": "\\)",
-        "~": "\\~",
-        "`": "\\`",
-        "#": "\\#",
-        "+": "\\+",
-        "-": "\\-",
-        "=": "\\=",
-        "|": "\\|",
-        "{": "\\{",
-        "}": "\\}",
-        ".": "\\.",
-        "!": "\\!",
-        "<": "\\<",
-        ">": "\\>",
-        "&": "\\&",
-    }
-
-    escaped_text = text
-    for char, replacement in special_chars.items():
-        escaped_text = escaped_text.replace(char, replacement)
-
-    return escaped_text
 
 
 def sanitize_for_plain_text(text: str) -> str:
@@ -216,11 +202,4 @@ def sanitize_for_markdown(text: str) -> str:
     if not text:
         return text
 
-    # Remove newlines and carriage returns to prevent multi-line injection
+    return escape_markdown(sanitize_for_plain_text(text))
-    sanitized = text.replace("\n", " ").replace("\r", " ")
-
-    # Collapse multiple spaces into single space
-    sanitized = " ".join(sanitized.split())
-
-    # Escape Markdown special characters
-    return escape_markdown(sanitized)

tests/test_health_checker.py

@@ -22,7 +22,8 @@ from typing import TYPE_CHECKING
 import pytest
 from aioresponses import aioresponses
 
-from owncastsentry.health_checker import HealthChecker, HealthStatus, UpdateResult
+from owncastsentry.health_checker import HealthChecker, HealthStatus
+from owncastsentry.models import UpdateResult
 from owncastsentry.owncast_client import OwncastClient
 
 if TYPE_CHECKING:

tests/test_notification_service.py

@@ -62,7 +62,7 @@ class TestCanNotify:
         service = _make_service(
             client=_StubMatrixClient(), subscription_repo=subscription_repo
         )
-        service.notification_timers_cache["example.com"] = time.time()
+        service.notification_timers_cache["example.com"] = time.monotonic()
         assert service._can_notify("example.com") is False
 
     def test_after_cooldown_allowed(
@@ -74,7 +74,7 @@ class TestCanNotify:
         )
         # Subtract an extra second to ensure the cooldown has fully elapsed
         service.notification_timers_cache["example.com"] = (
-            time.time() - SECONDS_BETWEEN_NOTIFICATIONS - 1
+            time.monotonic() - SECONDS_BETWEEN_NOTIFICATIONS - 1
         )
         assert service._can_notify("example.com") is True
 
@@ -253,7 +253,7 @@ class TestNotifyStreamLive:
         """Skip sending when the domain is within the rate-limit cooldown."""
         client = _StubMatrixClient()
         service = _make_service(client=client, subscription_repo=subscription_repo)
-        service.notification_timers_cache["example.com"] = time.time()
+        service.notification_timers_cache["example.com"] = time.monotonic()
 
         await stream_repo.create("example.com")
         await subscription_repo.add("example.com", "!room1:matrix.org")

tests/test_stream_monitor.py

@@ -337,7 +337,7 @@ class TestUpdateStreamBriefOffline:
         )
 
         # Recently offline (within cooldown)
-        monitor.offline_timer_cache["brief.com"] = time.time() - 60
+        monitor.offline_timer_cache["brief.com"] = time.monotonic() - 60
 
         result = await monitor.update_stream("brief.com")
         assert result is True
@@ -375,7 +375,7 @@ class TestUpdateStreamBriefOffline:
         )
 
         # Recently offline (within cooldown)
-        monitor.offline_timer_cache["brief.com"] = time.time() - 60
+        monitor.offline_timer_cache["brief.com"] = time.monotonic() - 60
 
         result = await monitor.update_stream("brief.com")
         assert result is True
@@ -424,7 +424,7 @@ class TestUpdateStreamTitleChange:
         monitor.offline_timer_cache["title.com"] = 0
         # Subtract an extra second to ensure the cooldown has fully elapsed
         notification_service.notification_timers_cache["title.com"] = (
-            time.time() - SECONDS_BETWEEN_NOTIFICATIONS - 1
+            time.monotonic() - SECONDS_BETWEEN_NOTIFICATIONS - 1
         )
 
         result = await monitor.update_stream("title.com")
@@ -473,7 +473,7 @@ class TestUpdateStreamTitleChange:
         monitor.offline_timer_cache["title.com"] = 0
         # Subtract an extra second to ensure the cooldown has fully elapsed
         notification_service.notification_timers_cache["title.com"] = (
-            time.time() - SECONDS_BETWEEN_NOTIFICATIONS - 1
+            time.monotonic() - SECONDS_BETWEEN_NOTIFICATIONS - 1
         )
 
         result = await monitor.update_stream("title.com")
@@ -518,7 +518,7 @@ class TestUpdateStreamTitleChange:
 
         # Offline timer is MORE recent than last notification,
         # and both are old enough to pass rate limiting
-        now = time.time()
+        now = time.monotonic()
         monitor.offline_timer_cache["title.com"] = (
             now - SECONDS_BETWEEN_NOTIFICATIONS - 100
         )
@@ -571,9 +571,9 @@ class TestUpdateStreamGoesOffline:
         )
         monitor.offline_timer_cache["offline.com"] = 0
 
-        before = time.time()
+        before = time.monotonic()
         result = await monitor.update_stream("offline.com")
-        after = time.time()
+        after = time.monotonic()
 
         assert result is True
         assert before <= monitor.offline_timer_cache["offline.com"] <= after