mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-31 04:14:38 +08:00
326 lines
11 KiB
Python
326 lines
11 KiB
Python
"""
|
|
订阅领域的请求级 command/query 依赖。
|
|
"""
|
|
|
|
from collections.abc import AsyncIterator, Mapping
|
|
from contextlib import asynccontextmanager
|
|
from typing import cast
|
|
|
|
from fastapi import Depends
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.adapters.external.server import MoviePilotServerHelper
|
|
from app.api.context import (
|
|
get_async_session,
|
|
get_background_task_registry,
|
|
get_host_runtime,
|
|
get_subscription_history_repository,
|
|
get_subscription_outbox,
|
|
get_subscription_outbox_store,
|
|
get_subscription_repository,
|
|
get_subscription_session,
|
|
get_subscription_transaction,
|
|
get_sync_session,
|
|
resolve_background_task_registry,
|
|
)
|
|
from app.application.outbox import AsyncOutboxDispatchStore, AsyncOutboxStager
|
|
from app.application.scheduling import start_scheduler_job
|
|
from app.application.servarr import ServarrSubscriptionService
|
|
from app.application.subscription.contract import (
|
|
SessionSubscriptionPort,
|
|
SubscriptionHistoryStagingPort,
|
|
SubscriptionQueryPort,
|
|
SubscriptionStagingPort,
|
|
)
|
|
from app.application.subscription.delete import (
|
|
AsyncUnitOfWork as DeleteUnitOfWork,
|
|
)
|
|
from app.application.subscription.delete import (
|
|
DeleteSubscribeCommand,
|
|
)
|
|
from app.application.subscription.execution import SubscriptionSearchRepository
|
|
from app.application.subscription.identity import (
|
|
DeleteSubscriptionsByIdentityCommand,
|
|
)
|
|
from app.application.subscription.mutation import (
|
|
SubscriptionMutationService,
|
|
)
|
|
from app.application.subscription.query import SubscriptionQueryService
|
|
from app.application.subscription.search import (
|
|
SearchSubscriptionsCommand,
|
|
SubscriptionSearchSubmission,
|
|
)
|
|
from app.application.subscription.status import (
|
|
SubscriptionExecutionReadRepository,
|
|
SubscriptionExecutionStatusService,
|
|
)
|
|
from app.application.subscription.write import (
|
|
SubscriptionBatchWritePort,
|
|
)
|
|
from app.runtime.events import eventmanager
|
|
from app.runtime.log import logger
|
|
from app.runtime.tasks import TaskRegistry
|
|
from app.schemas.common import JsonData
|
|
from app.schemas.types import EventType
|
|
from app.startup.composition.context import HostRuntime
|
|
|
|
|
|
async def _publish_subscribe_deleted(
|
|
payload: Mapping[str, JsonData],
|
|
) -> None:
|
|
"""
|
|
通过宿主事件总线发布已提交的订阅删除事件。
|
|
"""
|
|
event_payload = payload if isinstance(payload, dict) else dict(payload)
|
|
await eventmanager.async_send_event(EventType.SubscribeDeleted, event_payload)
|
|
|
|
|
|
async def _publish_subscribe_modified(payload: Mapping[str, JsonData]) -> None:
|
|
"""
|
|
通过宿主事件总线发布已提交的订阅修改事件。
|
|
"""
|
|
await eventmanager.async_send_event(EventType.SubscribeModified, dict(payload))
|
|
|
|
|
|
async def _report_subscribe_deleted(payload: Mapping[str, JsonData]) -> bool:
|
|
"""
|
|
把只读删除快照复制给远端上报边界。
|
|
"""
|
|
return bool(await MoviePilotServerHelper.async_sub_done_durable(dict(payload)))
|
|
|
|
|
|
def get_delete_subscribe_command(
|
|
repository_port: SubscriptionStagingPort = Depends(get_subscription_repository),
|
|
unit_of_work: object = Depends(get_subscription_transaction),
|
|
outbox: AsyncOutboxStager = Depends(get_subscription_outbox),
|
|
dispatch_store: AsyncOutboxDispatchStore = Depends(get_subscription_outbox_store),
|
|
) -> DeleteSubscribeCommand:
|
|
"""
|
|
组装请求级订阅删除用例及其具体适配器。
|
|
"""
|
|
return DeleteSubscribeCommand(
|
|
repository=repository_port,
|
|
unit_of_work=cast(DeleteUnitOfWork, unit_of_work),
|
|
publish_deleted=_publish_subscribe_deleted,
|
|
report_deleted=_report_subscribe_deleted,
|
|
outbox=outbox,
|
|
dispatch_store=dispatch_store,
|
|
)
|
|
|
|
|
|
def _log_subscribe_deleted_event_error(
|
|
subscribe_id: int,
|
|
error: Exception,
|
|
) -> None:
|
|
"""
|
|
记录按媒体身份删除时的单条事件失败并允许后续事件继续。
|
|
"""
|
|
logger.error(
|
|
f"发送订阅删除事件失败：{subscribe_id} - {error}",
|
|
exc_info=True,
|
|
)
|
|
|
|
|
|
def get_delete_subscriptions_by_identity_command(
|
|
repository_port: SubscriptionStagingPort = Depends(get_subscription_repository),
|
|
unit_of_work: object = Depends(get_subscription_transaction),
|
|
outbox: AsyncOutboxStager = Depends(get_subscription_outbox),
|
|
dispatch_store: AsyncOutboxDispatchStore = Depends(get_subscription_outbox_store),
|
|
) -> DeleteSubscriptionsByIdentityCommand:
|
|
"""
|
|
组装请求级按媒体身份删除订阅用例。
|
|
"""
|
|
return DeleteSubscriptionsByIdentityCommand(
|
|
repository=repository_port,
|
|
unit_of_work=cast(DeleteUnitOfWork, unit_of_work),
|
|
publish_deleted=_publish_subscribe_deleted,
|
|
handle_event_error=_log_subscribe_deleted_event_error,
|
|
outbox=outbox,
|
|
dispatch_store=dispatch_store,
|
|
)
|
|
|
|
|
|
def _resume_submitted_subscription_search(
|
|
subscribe_ids: tuple[int, ...],
|
|
) -> None:
|
|
"""
|
|
唤醒搜索队列，并优先处理本次手工选择的订阅。
|
|
"""
|
|
start_scheduler_job(
|
|
"subscribe_search_queue",
|
|
limit=max(1, len(subscribe_ids)),
|
|
manual_sids=subscribe_ids,
|
|
)
|
|
|
|
|
|
def get_search_subscriptions_command(
|
|
task_registry: TaskRegistry = Depends(get_background_task_registry),
|
|
db: AsyncSession = Depends(get_async_session),
|
|
runtime: HostRuntime = Depends(get_host_runtime),
|
|
) -> SearchSubscriptionsCommand:
|
|
"""
|
|
组装手工搜索用例，请求内只入队，实际搜索交给后台继续。
|
|
"""
|
|
|
|
registry = resolve_background_task_registry(task_registry)
|
|
search_repository = get_subscription_search_repository(runtime)
|
|
|
|
async def submit_search(
|
|
subscribe_ids: tuple[int, ...],
|
|
single: bool,
|
|
) -> SubscriptionSearchSubmission:
|
|
"""
|
|
在线程 owner 中完成轻量入队，并返回前端可立即跟踪的批次。
|
|
"""
|
|
enqueue_task = registry.create_sync(
|
|
search_repository.enqueue,
|
|
subscription_ids=subscribe_ids,
|
|
source="manual",
|
|
priority=120 if single else 100,
|
|
owner="api.subscribe.search.enqueue",
|
|
)
|
|
enqueued = await enqueue_task
|
|
registry.create_sync(
|
|
_resume_submitted_subscription_search,
|
|
subscribe_ids,
|
|
owner="api.subscribe.search.run",
|
|
)
|
|
return SubscriptionSearchSubmission(
|
|
batch_ids=enqueued.active_batch_ids,
|
|
target_count=len(subscribe_ids),
|
|
queued_count=enqueued.created_count,
|
|
ongoing_count=enqueued.coalesced_count,
|
|
single=single,
|
|
)
|
|
|
|
return SearchSubscriptionsCommand(
|
|
repository=runtime.subscription.repository(db),
|
|
submit_search=submit_search,
|
|
)
|
|
|
|
|
|
def get_subscription_query_service(
|
|
db: AsyncSession = Depends(get_async_session),
|
|
runtime: HostRuntime = Depends(get_host_runtime),
|
|
) -> SubscriptionQueryService:
|
|
"""
|
|
组装订阅和订阅历史异步查询服务。
|
|
"""
|
|
return SubscriptionQueryService(
|
|
repository=runtime.subscription.repository(db),
|
|
async_repository=runtime.subscription.repository(db),
|
|
history_repository=runtime.subscription.history_repository(db),
|
|
)
|
|
|
|
|
|
def get_subscription_execution_status_service(
|
|
db: AsyncSession = Depends(get_async_session),
|
|
runtime: HostRuntime = Depends(get_host_runtime),
|
|
task_registry: TaskRegistry = Depends(get_background_task_registry),
|
|
) -> SubscriptionExecutionStatusService:
|
|
"""
|
|
组装请求级订阅执行状态投影与取消服务。
|
|
"""
|
|
factory = runtime.subscription.execution_status_repository
|
|
if factory is None:
|
|
raise RuntimeError("订阅执行状态仓储未注册")
|
|
repository = factory(db)
|
|
search_repository = get_subscription_search_repository(runtime)
|
|
registry = resolve_background_task_registry(task_registry)
|
|
|
|
async def request_cancel(batch_id: str) -> bool:
|
|
"""
|
|
在线程 owner 内提交同步队列写入，并等待其真实终态。
|
|
"""
|
|
task = registry.create_sync(
|
|
search_repository.request_cancel,
|
|
batch_id,
|
|
owner="api.subscribe.execution.cancel",
|
|
)
|
|
return bool(await task)
|
|
|
|
return SubscriptionExecutionStatusService(
|
|
repository=cast(SubscriptionExecutionReadRepository, repository),
|
|
request_cancel=request_cancel,
|
|
)
|
|
|
|
|
|
def get_subscription_search_repository(
|
|
runtime: HostRuntime = Depends(get_host_runtime),
|
|
) -> SubscriptionSearchRepository:
|
|
"""
|
|
返回宿主组合根持有的订阅搜索队列端口。
|
|
"""
|
|
if runtime.subscription.search_repository is None:
|
|
raise RuntimeError("订阅搜索队列未注册")
|
|
return cast(SubscriptionSearchRepository, runtime.subscription.search_repository)
|
|
|
|
|
|
def get_subscription_mutation_service(
|
|
repository_port: SessionSubscriptionPort = Depends(get_subscription_repository),
|
|
history_repository: SubscriptionHistoryStagingPort = Depends(get_subscription_history_repository),
|
|
unit_of_work: object = Depends(get_subscription_transaction),
|
|
outbox: AsyncOutboxStager = Depends(get_subscription_outbox),
|
|
dispatch_store: AsyncOutboxDispatchStore = Depends(get_subscription_outbox_store),
|
|
) -> SubscriptionMutationService:
|
|
"""
|
|
组装异步订阅写服务。
|
|
"""
|
|
return SubscriptionMutationService(
|
|
repository=repository_port,
|
|
history_repository=history_repository,
|
|
unit_of_work=cast(DeleteUnitOfWork, unit_of_work),
|
|
outbox=outbox,
|
|
dispatch_store=dispatch_store,
|
|
publish_modified=_publish_subscribe_modified,
|
|
)
|
|
|
|
|
|
def get_servarr_subscription_batch_writer(
|
|
repository_port: SubscriptionStagingPort = Depends(get_subscription_repository),
|
|
unit_of_work: object = Depends(get_subscription_transaction),
|
|
outbox: AsyncOutboxStager = Depends(get_subscription_outbox),
|
|
dispatch_store: AsyncOutboxDispatchStore = Depends(get_subscription_outbox_store),
|
|
runtime: HostRuntime = Depends(get_host_runtime),
|
|
) -> SubscriptionBatchWritePort:
|
|
"""
|
|
组装 Servarr 多季订阅使用的请求级原子批量写端口。
|
|
"""
|
|
return runtime.subscription.batch_writer(
|
|
repository=repository_port,
|
|
unit_of_work=cast(DeleteUnitOfWork, unit_of_work),
|
|
outbox=outbox,
|
|
dispatch_store=dispatch_store,
|
|
)
|
|
|
|
|
|
def get_servarr_subscription_service(
|
|
async_db: AsyncSession = Depends(get_subscription_session),
|
|
db: Session = Depends(get_sync_session),
|
|
unit_of_work: object = Depends(get_subscription_transaction),
|
|
runtime: HostRuntime = Depends(get_host_runtime),
|
|
) -> ServarrSubscriptionService:
|
|
"""
|
|
组装 Servarr 兼容路由的请求级订阅数据用例。
|
|
"""
|
|
return ServarrSubscriptionService(
|
|
async_repository=runtime.subscription.repository(async_db),
|
|
sync_repository=runtime.subscription.repository(db),
|
|
unit_of_work=cast(DeleteUnitOfWork, unit_of_work),
|
|
)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def subscription_log_query_scope(runtime: HostRuntime) -> AsyncIterator[SubscriptionQueryService]:
|
|
"""
|
|
日志权限轮询使用短会话，每次检查后释放连接，避免 SSE 占用数据库连接池。
|
|
"""
|
|
sessions = runtime.subscription.async_session()
|
|
try:
|
|
session = await anext(sessions)
|
|
repository = cast(SubscriptionQueryPort, runtime.subscription.repository(session))
|
|
yield SubscriptionQueryService(repository, async_repository=repository)
|
|
finally:
|
|
await sessions.aclose()
|