Refine notification events and Telegram delivery

backend/alembic/versions/0003_notification_center_events.py

@@ -0,0 +1,138 @@
+"""Create notification center event storage.
+
+Revision ID: 0003_notification_center_events
+Revises: 0002_scope_douban_identity
+Create Date: 2026-06-03
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0003_notification_center_events"
+down_revision = "0002_scope_douban_identity"
+branch_labels = None
+depends_on = None
+
+
+PRUNED_EVENT_TYPES = (
+    "download.started",
+    "download.task.downloader_changed",
+    "download.task.storage_change_started",
+    "download.task.storage_changed",
+    "media.import.started",
+    "media_server_sync.started",
+    "danmu.generate.started",
+    "subscription.enabled",
+    "subscription.disabled",
+    "subscription.ended.manual",
+    "subscription.run.completed",
+    "subscription.run.failed",
+    "follow.enabled",
+    "follow.disabled",
+    "pilot.episode.queued",
+)
+
+
+def _has_table(table_name: str) -> bool:
+    bind = op.get_bind()
+    return sa.inspect(bind).has_table(table_name)
+
+
+def _prune_events(event_types: tuple[str, ...]) -> None:
+    if not event_types or not _has_table("events"):
+        return
+    bind = op.get_bind()
+    params = {f"type_{index}": value for index, value in enumerate(event_types)}
+    placeholders = ", ".join(f":type_{index}" for index in range(len(event_types)))
+    if _has_table("event_dispatches"):
+        bind.execute(
+            sa.text(
+                f"""
+                DELETE FROM event_dispatches
+                WHERE event_id IN (
+                    SELECT id FROM events WHERE type IN ({placeholders})
+                )
+                """
+            ),
+            params,
+        )
+    if _has_table("event_acknowledgements"):
+        bind.execute(
+            sa.text(
+                f"""
+                DELETE FROM event_acknowledgements
+                WHERE event_id IN (
+                    SELECT id FROM events WHERE type IN ({placeholders})
+                )
+                """
+            ),
+            params,
+        )
+    bind.execute(sa.text(f"DELETE FROM events WHERE type IN ({placeholders})"), params)
+
+
+def upgrade() -> None:
+    if _has_table("alerts"):
+        op.drop_table("alerts")
+    if not _has_table("event_acknowledgements"):
+        op.create_table(
+            "event_acknowledgements",
+            sa.Column("event_id", sa.Text(), primary_key=True),
+            sa.Column("acknowledged_at", sa.Text(), nullable=False),
+        )
+        op.create_index("ix_event_acknowledgements_acknowledged_at", "event_acknowledgements", ["acknowledged_at"])
+    _prune_events(PRUNED_EVENT_TYPES)
+
+
+def downgrade() -> None:
+    if _has_table("event_acknowledgements"):
+        op.drop_index("ix_event_acknowledgements_acknowledged_at", table_name="event_acknowledgements")
+        op.drop_table("event_acknowledgements")
+    if _has_table("alerts"):
+        return
+    op.create_table(
+        "alerts",
+        sa.Column("id", sa.Text(), primary_key=True),
+        sa.Column("fingerprint", sa.Text(), nullable=False),
+        sa.Column("status", sa.Text(), nullable=False),
+        sa.Column("severity", sa.Text(), nullable=False),
+        sa.Column("category", sa.Text(), nullable=False),
+        sa.Column("message_key", sa.Text(), nullable=False),
+        sa.Column("message_params_json", sa.JSON(), nullable=False),
+        sa.Column("target_type", sa.Text(), nullable=True),
+        sa.Column("target_id", sa.Text(), nullable=True),
+        sa.Column("media_id", sa.Text(), nullable=True),
+        sa.Column("media_season_number", sa.Integer(), nullable=True),
+        sa.Column("media_title", sa.Text(), nullable=True),
+        sa.Column("media_year", sa.Integer(), nullable=True),
+        sa.Column("task_id", sa.Text(), nullable=True),
+        sa.Column("action_id", sa.Text(), nullable=True),
+        sa.Column("occurrence_count", sa.Integer(), nullable=False, server_default="1"),
+        sa.Column("first_seen_at", sa.Text(), nullable=False),
+        sa.Column("last_seen_at", sa.Text(), nullable=False),
+        sa.Column("acknowledged_at", sa.Text(), nullable=True),
+        sa.Column("resolved_at", sa.Text(), nullable=True),
+        sa.Column("created_at", sa.Text(), nullable=False),
+        sa.Column("updated_at", sa.Text(), nullable=False),
+        sa.UniqueConstraint("fingerprint", name="uq_alerts_fingerprint"),
+    )
+    op.create_index("ix_alerts_fingerprint", "alerts", ["fingerprint"])
+    op.create_index("ix_alerts_status", "alerts", ["status"])
+    op.create_index("ix_alerts_severity", "alerts", ["severity"])
+    op.create_index("ix_alerts_category", "alerts", ["category"])
+    op.create_index("ix_alerts_target_type", "alerts", ["target_type"])
+    op.create_index("ix_alerts_target_id", "alerts", ["target_id"])
+    op.create_index("ix_alerts_media_id", "alerts", ["media_id"])
+    op.create_index("ix_alerts_media_season_number", "alerts", ["media_season_number"])
+    op.create_index("ix_alerts_task_id", "alerts", ["task_id"])
+    op.create_index("ix_alerts_action_id", "alerts", ["action_id"])
+    op.create_index("ix_alerts_first_seen_at", "alerts", ["first_seen_at"])
+    op.create_index("ix_alerts_last_seen_at", "alerts", ["last_seen_at"])
+    op.create_index("ix_alerts_acknowledged_at", "alerts", ["acknowledged_at"])
+    op.create_index("ix_alerts_resolved_at", "alerts", ["resolved_at"])
+    op.create_index("ix_alerts_updated_at", "alerts", ["updated_at"])
+    op.create_index("ix_alerts_status_ack_severity", "alerts", ["status", "acknowledged_at", "severity"])
+    op.create_index("ix_alerts_target", "alerts", ["target_type", "target_id"])

backend/app/addons/descriptors.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from app.addons.registry import AddonDescriptor, AddonJobSpec, AddonRegistry
-from app.schemas.config import AddonsConfig
+from app.schemas.config import AddonsConfig, NotificationChannelConfig
 from app.services.application.workflows.danmu import danmu_application_service
 from app.services.application.workflows.notifications import notification_application_service
 from app.services.platform.auth_provider_service import auth_provider_service
@@ -13,9 +13,15 @@ def _auth_enabled(config: AddonsConfig) -> bool:
 
 
 def _notifications_enabled(config: AddonsConfig) -> bool:
-    return config.notifications.enabled and any(
-        notification_channel_service.supports(channel.type) for channel in config.notifications.channels
-    )
+    if not config.notifications.enabled:
+        return False
+    return any(_notification_channel_ready(channel) for channel in config.notifications.channels)
+
+
+def _notification_channel_ready(channel: NotificationChannelConfig) -> bool:
+    if not channel.enabled or not notification_channel_service.supports(channel.type):
+        return False
+    return notification_channel_service.get_channel(channel.type).is_configured(channel)
 
 
 def _notification_event_patterns() -> list[str]:

backend/app/api/v1/api.py

@@ -1,6 +1,6 @@
 from fastapi import APIRouter
 
-from app.api.v1 import actions, alerts, auth, calendar, commands, config, discover, events, addons, library, logs, media, media_management, resource, scheduler, subscription, task
+from app.api.v1 import actions, auth, calendar, commands, config, discover, events, addons, library, logs, media, media_management, resource, scheduler, subscription, task
 from app.api.v1.config.test_directory import router as test_directory_router
 from app.api.v1.resource import download_history, parser
 
@@ -22,7 +22,6 @@
 api_router.include_router(commands.router, tags=["commands"])
 api_router.include_router(events.router, tags=["events"])
 api_router.include_router(actions.router, tags=["actions"])
-api_router.include_router(alerts.router, tags=["alerts"])
 api_router.include_router(logs.router, tags=["logs"])
 api_router.include_router(addons.router, tags=["addons"])
 api_router.include_router(media_management.router, tags=["media-management"])

backend/app/api/v1/config/read_tabs.py

@@ -65,6 +65,8 @@ class SystemTabAuthResponse(BaseModel):
 
 
 class SystemTabResponse(BaseModel):
+    locale: str = "zh-CN"
+    public_base_url: str = ""
     auth: SystemTabAuthResponse
     download: DownloadConfig
     logging: LoggingConfig

backend/app/api/v1/config/test_connection.py

@@ -8,11 +8,13 @@
     PathMapping,
     ProwlarrConfig,
     RTorrentConfig,
+    TelegramNotificationChannelConfig,
     TMDBConfig,
     QBittorrentConfig,
 )
 from app.schemas.exception import ServiceTypeException, TestConnectionException
 from app.services.integration import douban as douban_integration
+from app.services.integration import notifications as notifications_integration
 from app.services.integration import tmdb as tmdb_integration
 from app.services.integration.download.gateway import download_gateway
 from app.services.integration.indexer import indexer_gateway
@@ -27,6 +29,7 @@
 SERVICE_TYPE_JELLYFIN = "jellyfin"
 SERVICE_TYPE_TMDB = "themoviedb"
 SERVICE_TYPE_DOUBAN = "douban"
+SERVICE_TYPE_TELEGRAM = "telegram"
 SUPPORTED_SERVICE_TYPES = [
     SERVICE_TYPE_QBITTORRENT,
     SERVICE_TYPE_RTORRENT,
@@ -35,6 +38,7 @@
     SERVICE_TYPE_JELLYFIN,
     SERVICE_TYPE_TMDB,
     SERVICE_TYPE_DOUBAN,
+    SERVICE_TYPE_TELEGRAM,
 ]
 
 
@@ -47,6 +51,8 @@ class TestConnectionConfig(BaseModel):
     username: str | None = None
     password: str | None = None
     api_key: str | None = None
+    bot_token: str | None = None
+    chat_id: str | None = None
     path_mappings: list[PathMapping] = []
 
 
@@ -91,6 +97,12 @@ async def test_service_connection(payload: TestServiceConnectionRequest):
     elif service_type == SERVICE_TYPE_DOUBAN:
         config = DoubanConfig.model_validate(config_data)
         success = await douban_integration.test_connection_for_config(config)
+    elif service_type == SERVICE_TYPE_TELEGRAM:
+        config = TelegramNotificationChannelConfig.model_validate(config_data)
+        try:
+            success = await notifications_integration.test_telegram_connection_for_config(config)
+        except RuntimeError as exc:
+            raise TestConnectionException(service_name=service_type, reason=str(exc)) from exc
     else:
         raise ServiceTypeException(service_type=service_type, supported_types=SUPPORTED_SERVICE_TYPES)
 

backend/app/api/v1/events/list.py

@@ -2,9 +2,8 @@
 from pydantic import BaseModel
 
 from app.api.deps import OptionalMediaIDParam
-from app.schemas.constants.event_types import EventTypes
 from app.schemas.media_id import MediaID
-from app.schemas.domain.event import Event, EventLevel, EventSource, EventType
+from app.schemas.domain.event import Event, EventCenterResponse, EventLevel, EventSource, EventType
 from app.services.audit.event_service import event_service
 
 router = APIRouter()
@@ -21,6 +20,11 @@ class EventFilterOptionsResponse(BaseModel):
     sources: list[str]
 
 
+class EventAcknowledgeResponse(BaseModel):
+    ok: bool
+    acknowledged_count: int = 0
+
+
 @router.get("/", response_model=EventListResponse)
 async def list_events(
     limit: int = Query(default=50, ge=1, le=200),
@@ -53,6 +57,22 @@ async def list_events(
     return EventListResponse(total=total, items=items)
 
 
+@router.get("/center", response_model=EventCenterResponse)
+async def get_event_center() -> EventCenterResponse:
+    return event_service.get_center()
+
+
+@router.post("/center/acknowledge-all", response_model=EventAcknowledgeResponse)
+async def acknowledge_event_center() -> EventAcknowledgeResponse:
+    return EventAcknowledgeResponse(ok=True, acknowledged_count=event_service.acknowledge_attention_events())
+
+
+@router.post("/{event_id}/acknowledge", response_model=EventAcknowledgeResponse)
+async def acknowledge_event(event_id: str) -> EventAcknowledgeResponse:
+    acknowledged = event_service.acknowledge_event(event_id)
+    return EventAcknowledgeResponse(ok=acknowledged, acknowledged_count=1 if acknowledged else 0)
+
+
 @router.get("/filter-options", response_model=EventFilterOptionsResponse)
 async def get_event_filter_options(
     media_id: MediaID | None = Depends(OptionalMediaIDParam),
@@ -65,31 +85,7 @@ async def get_event_filter_options(
     _ = (media_id, season_number, task_id, subscription_id, action_id, keyword)
     return EventFilterOptionsResponse(
         levels=[level.value for level in EventLevel],
-        types=[
-            EventTypes.DOWNLOAD_STARTED.value,
-            EventTypes.DOWNLOAD_COMPLETED.value,
-            EventTypes.DOWNLOAD_FAILED.value,
-            EventTypes.MEDIA_IMPORT_STARTED.value,
-            EventTypes.MEDIA_IMPORT_COMPLETED.value,
-            EventTypes.MEDIA_IMPORT_FAILED.value,
-            EventTypes.MEDIA_SERVER_SYNC_STARTED.value,
-            EventTypes.MEDIA_SERVER_SYNC_COMPLETED.value,
-            EventTypes.MEDIA_SERVER_SYNC_FAILED.value,
-            EventTypes.DANMU_GENERATE_STARTED.value,
-            EventTypes.DANMU_GENERATE_COMPLETED.value,
-            EventTypes.DANMU_GENERATE_FAILED.value,
-            EventTypes.MEDIA_DELETED.value,
-            EventTypes.SUBSCRIPTION_ENABLED.value,
-            EventTypes.SUBSCRIPTION_DISABLED.value,
-            EventTypes.FOLLOW_ENABLED.value,
-            EventTypes.FOLLOW_DISABLED.value,
-            EventTypes.FOLLOW_RELEASED.value,
-            EventTypes.FOLLOW_DIGITAL_RELEASED.value,
-            EventTypes.FOLLOW_PHYSICAL_RELEASED.value,
-            EventTypes.SUBSCRIPTION_RUN_COMPLETED.value,
-            EventTypes.SUBSCRIPTION_RUN_FAILED.value,
-            EventTypes.PILOT_EPISODE_QUEUED.value,
-        ],
+        types=[event_type.value for event_type in EventType],
         sources=[source.value for source in EventSource],
     )
 

backend/app/clients/__init__.py

@@ -4,5 +4,3 @@
 """
 
 from app.clients.base import BaseClient
-from app.clients.factory import ClientFactory, ClientType
-

backend/app/core/feature_flags.py

@@ -14,7 +14,3 @@ def env_flag_enabled(name: str, *, default: bool = False) -> bool:
 
 def oidc_auth_enabled() -> bool:
     return env_flag_enabled("AETHERA_EXPERIMENTAL_OIDC_AUTH")
-
-
-def telegram_notifications_enabled() -> bool:
-    return env_flag_enabled("AETHERA_EXPERIMENTAL_TELEGRAM_NOTIFICATIONS")