285 lines
10 KiB
Python
285 lines
10 KiB
Python
# Copyright 2026 Logan Fick
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
"""Notification service for sending Matrix messages."""
|
|
|
|
import asyncio
|
|
import time
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from mautrix.types import MessageType, TextMessageEventContent
|
|
|
|
from .utils import (
|
|
CLEANUP_DELETE_THRESHOLD,
|
|
CLEANUP_WARNING_THRESHOLD,
|
|
SECONDS_BETWEEN_NOTIFICATIONS,
|
|
sanitize_for_plain_text,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
import logging
|
|
|
|
from .database import SubscriptionRepository
|
|
|
|
|
|
class NotificationService:
|
|
"""Service for sending Matrix notifications about stream events."""
|
|
|
|
def __init__(
|
|
self,
|
|
client: Any,
|
|
subscription_repo: SubscriptionRepository,
|
|
logger: logging.Logger,
|
|
) -> None:
|
|
"""Initialize the notification service.
|
|
|
|
:param client: The Matrix client for sending messages.
|
|
:param subscription_repo: Repository for managing subscriptions.
|
|
:param logger: Logger instance for debugging.
|
|
"""
|
|
self.client = client
|
|
self.subscription_repo = subscription_repo
|
|
self.log = logger
|
|
|
|
# Cache for tracking when notifications were last sent
|
|
self.notification_timers_cache: dict[str, float] = {}
|
|
|
|
async def notify_stream_live(
|
|
self,
|
|
domain: str,
|
|
name: str,
|
|
title: str,
|
|
tags: list[str],
|
|
*,
|
|
title_change: bool = False,
|
|
) -> None:
|
|
"""Send notifications to rooms subscribed to a stream.
|
|
|
|
:param domain: The stream domain to send notifications for.
|
|
:param name: The stream name to include in the message.
|
|
:param title: The stream title to include in the message.
|
|
:param tags: List of stream tags to include in the message.
|
|
:param title_change: Whether this is a title change notification.
|
|
"""
|
|
# Has enough time passed since the last notification was sent?
|
|
if not self._can_notify(domain):
|
|
seconds_since_last = round(
|
|
time.time() - self.notification_timers_cache[domain]
|
|
)
|
|
self.log.info(
|
|
f"[{domain}] Not sending notifications. Only "
|
|
f"{seconds_since_last} of required "
|
|
f"{SECONDS_BETWEEN_NOTIFICATIONS} seconds have "
|
|
f"passed since last notification."
|
|
)
|
|
return
|
|
|
|
# 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 = [
|
|
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"{failed_notifications} failed."
|
|
)
|
|
|
|
async def _send_notification(
|
|
self, room_id: str, body_text: str, domain: str
|
|
) -> None:
|
|
"""Send a notification to a single room.
|
|
|
|
:param room_id: The Matrix room ID to send to.
|
|
:param body_text: The message body text.
|
|
:param domain: The stream domain (for logging).
|
|
:raises Exception: If sending fails.
|
|
"""
|
|
try:
|
|
content = TextMessageEventContent(msgtype=MessageType.TEXT, body=body_text)
|
|
await self.client.send_message(room_id, content)
|
|
except Exception as exception:
|
|
self.log.warning(
|
|
f"[{domain}] Failed to send notification "
|
|
f"message to room [{room_id}]: {exception}"
|
|
)
|
|
raise
|
|
|
|
def _format_message(
|
|
self, name: str, title: str, domain: str, tags: list[str], title_change: bool
|
|
) -> str:
|
|
"""Format the notification message body.
|
|
|
|
:param name: The stream name.
|
|
:param title: The stream title.
|
|
:param domain: The stream domain.
|
|
:param tags: List of stream tags.
|
|
:param title_change: Whether this is a title change notification.
|
|
:return: Formatted message body.
|
|
"""
|
|
# Use name if available, fallback to domain
|
|
stream_name = name if name else 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!"
|
|
else:
|
|
body_text = "🎥 " + safe_stream_name + " is now live!"
|
|
|
|
# Add title if present
|
|
if title != "":
|
|
safe_title = sanitize_for_plain_text(title)
|
|
body_text += "\nStream Title: " + safe_title
|
|
|
|
# Add stream URL
|
|
body_text += "\n\nTo tune in, visit: https://" + domain + "/"
|
|
|
|
# Add tags if present
|
|
if tags:
|
|
safe_tags = []
|
|
for tag in tags:
|
|
safe_tag = sanitize_for_plain_text(tag)
|
|
if safe_tag and not safe_tag.startswith("."):
|
|
safe_tags.append(safe_tag)
|
|
|
|
if safe_tags:
|
|
body_text += "\n\n"
|
|
body_text += " ".join("#" + tag for tag in safe_tags)
|
|
|
|
return body_text
|
|
|
|
def get_last_notification_time(self, domain: str) -> float:
|
|
"""Get the timestamp of the last notification sent for a domain.
|
|
|
|
:param domain: The stream domain.
|
|
:return: Unix timestamp of last notification, or 0 if never notified.
|
|
"""
|
|
return self.notification_timers_cache.get(domain, 0)
|
|
|
|
def _can_notify(self, domain: str) -> bool:
|
|
"""Check if enough time has passed to send another notification.
|
|
|
|
:param domain: The stream domain.
|
|
:return: True if notification can be sent, False otherwise.
|
|
"""
|
|
if domain not in self.notification_timers_cache:
|
|
return True
|
|
|
|
seconds_since_last = round(time.time() - self.notification_timers_cache[domain])
|
|
return bool(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()
|
|
|
|
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
|
|
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 {remaining_days} more days "
|
|
f"({delete_days} days total), this subscription "
|
|
f"will be automatically removed."
|
|
)
|
|
|
|
# Send to all rooms in parallel
|
|
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)."
|
|
)
|
|
|
|
async def send_cleanup_deletion(self, domain: str) -> None:
|
|
"""Send cleanup deletion notification to all subscribed rooms.
|
|
|
|
: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"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
|
|
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)."
|
|
)
|