mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-31 12:24:39 +08:00
951cdc5b27
* fix: arbitrate subscription identity and restore workflow state * test: preserve legacy empty-owner subscription transactions * fix: align shared atomic configuration method with lifecycle prerequisite --------- Co-authored-by: jxxghp <jxxghp@gmail.com>
477 lines
15 KiB
Python
477 lines
15 KiB
Python
"""
|
|
订阅写操作用例及其数据端口。
|
|
"""
|
|
|
|
from collections.abc import Awaitable, Callable
|
|
from contextlib import AbstractAsyncContextManager, AbstractContextManager
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Literal, cast
|
|
from uuid import uuid4
|
|
|
|
from app.application.classification.reference import (
|
|
normalize_classification_reference_payload,
|
|
)
|
|
from app.application.outbox import (
|
|
SUBSCRIBE_MODIFIED_TOPIC,
|
|
AsyncOutboxDispatchStore,
|
|
AsyncOutboxStager,
|
|
OutboxDispatchStore,
|
|
OutboxIntent,
|
|
OutboxStager,
|
|
SyncUnitOfWork,
|
|
deliver_async_outbox_effect,
|
|
deliver_outbox_effect,
|
|
)
|
|
from app.application.subscription.contract import (
|
|
SessionSubscriptionPort,
|
|
SubscriptionHistorySnapshot,
|
|
SubscriptionHistoryStagingPort,
|
|
SubscriptionPatch,
|
|
SubscriptionSnapshot,
|
|
)
|
|
from app.application.subscription.delete import AsyncUnitOfWork
|
|
from app.application.subscription.observability import subscription_log_scope
|
|
from app.runtime.log import logger
|
|
from app.schemas.common import JsonData
|
|
from app.schemas.event import SubscribeModifiedEventData
|
|
from app.schemas.types import MUSIC_ENTITY_ALBUM
|
|
|
|
SubscribeModifiedPublisher = Callable[[dict[str, JsonData]], Awaitable[None]]
|
|
SyncSubscribeModifiedPublisher = Callable[[dict[str, JsonData]], None]
|
|
|
|
|
|
def _normalized_subscription_patch(
|
|
subscribe: SubscriptionSnapshot,
|
|
payload: dict[str, JsonData],
|
|
) -> SubscriptionPatch:
|
|
"""
|
|
按订阅媒体类型规范化人工分类稳定引用并构造写入补丁。
|
|
"""
|
|
normalized = normalize_classification_reference_payload(
|
|
payload,
|
|
media_type=subscribe.type,
|
|
)
|
|
identity_fields = ("type", "media_source", "media_id", "season", "episode_group", "music_type")
|
|
if any(field in payload and payload[field] != getattr(subscribe, field) for field in identity_fields):
|
|
# 旧目标的下载和洗版事实不属于新作品、季或剧集组，须与身份修改在同一事务清除。
|
|
normalized.update({
|
|
"note": [], "episode_priority": {}, "current_priority": None,
|
|
"lack_episode": payload.get("total_episode", subscribe.total_episode),
|
|
"last_update": None, "last_search": None,
|
|
"current_audio_format": None, "current_bitrate": None,
|
|
"current_bit_depth": None, "current_sample_rate": None,
|
|
})
|
|
if subscribe.music_type == MUSIC_ENTITY_ALBUM:
|
|
normalized["downloaded_tracks"] = []
|
|
return SubscriptionPatch(cast(dict[str, JsonData], normalized))
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SubscriptionActor:
|
|
"""
|
|
订阅写操作的权限主体。
|
|
|
|
:param name: 发起订阅变更的本地用户名
|
|
:param is_superuser: 操作人是否具有超级管理员权限
|
|
"""
|
|
|
|
name: str
|
|
is_superuser: bool
|
|
|
|
|
|
SubscribeHistoryDeletionStatus = Literal["deleted", "not_found", "forbidden"]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SubscriptionMutation:
|
|
"""
|
|
一次订阅变更前后的稳定快照。
|
|
|
|
:param snapshot: 更新后的稳定订阅快照
|
|
:param old: 更新前的字段字典
|
|
:param new: 更新后的字段字典
|
|
:param event_published: 本次变更事件是否已完成发布
|
|
:param business_committed: 订阅业务事务是否已成功提交
|
|
:param pending_effects: 提交后仍等待投递的副作用标识
|
|
"""
|
|
|
|
snapshot: SubscriptionSnapshot
|
|
old: dict[str, JsonData]
|
|
new: dict[str, JsonData]
|
|
event_published: bool = False
|
|
business_committed: bool = False
|
|
pending_effects: tuple[str, ...] = ()
|
|
|
|
|
|
def _modified_event(
|
|
subscribe_id: int,
|
|
scene: str,
|
|
old: dict[str, JsonData],
|
|
updated: SubscriptionSnapshot,
|
|
) -> tuple[str, dict[str, JsonData]]:
|
|
"""
|
|
构造订阅修改 intent 共用的稳定键和事件快照。
|
|
"""
|
|
event_payload = SubscribeModifiedEventData(
|
|
subscribe_id=subscribe_id,
|
|
old_subscribe_info=old,
|
|
subscribe_info=updated.to_dict(),
|
|
scene=scene,
|
|
).to_dict()
|
|
event_key = _modified_event_key(subscribe_id, scene)
|
|
event_payload["idempotency_key"] = event_key
|
|
return event_key, event_payload
|
|
|
|
|
|
def _reset_payload(subscribe: SubscriptionSnapshot) -> dict[str, JsonData]:
|
|
"""
|
|
构造同步与异步重置共享的订阅字段补丁。
|
|
"""
|
|
return {
|
|
"note": [],
|
|
"downloaded_tracks": [] if subscribe.music_type == MUSIC_ENTITY_ALBUM else None,
|
|
"lack_episode": subscribe.total_episode,
|
|
"current_priority": None,
|
|
"current_audio_format": None,
|
|
"current_bitrate": None,
|
|
"current_bit_depth": None,
|
|
"current_sample_rate": None,
|
|
"episode_priority": {},
|
|
"manual_total_episode": 0,
|
|
"state": "R",
|
|
}
|
|
|
|
|
|
class SyncSubscriptionMutationService:
|
|
"""
|
|
用一个同步 UoW 原子提交订阅修改和 durable 事件 intent。
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
repository: SessionSubscriptionPort,
|
|
unit_of_work: SyncUnitOfWork,
|
|
outbox: OutboxStager,
|
|
dispatch_store: OutboxDispatchStore,
|
|
publish_modified: SyncSubscribeModifiedPublisher,
|
|
) -> None:
|
|
"""
|
|
注入同一 Session 的仓储、事务、outbox 与提交后发布端口。
|
|
"""
|
|
self._repository = repository
|
|
self._unit_of_work = unit_of_work
|
|
self._outbox = outbox
|
|
self._dispatch_store = dispatch_store
|
|
self._publish_modified = publish_modified
|
|
|
|
def get_accessible(
|
|
self,
|
|
subscribe_id: int,
|
|
actor: SubscriptionActor,
|
|
) -> SubscriptionSnapshot | None:
|
|
"""
|
|
同步读取当前主体可访问的订阅。
|
|
"""
|
|
subscribe = self._repository.get(subscribe_id)
|
|
return subscribe if SubscriptionMutationService.can_access(subscribe, actor) else None
|
|
|
|
def update(
|
|
self,
|
|
subscribe_id: int,
|
|
payload: dict[str, JsonData],
|
|
actor: SubscriptionActor,
|
|
existing: SubscriptionSnapshot | None = None,
|
|
scene: str = "update",
|
|
) -> SubscriptionMutation | None:
|
|
"""
|
|
同步更新订阅，并在同一事务暂存可恢复的修改事件。
|
|
"""
|
|
subscribe = existing or self.get_accessible(subscribe_id, actor)
|
|
if subscribe and not SubscriptionMutationService.can_access(subscribe, actor):
|
|
return None
|
|
if not subscribe:
|
|
return None
|
|
old = subscribe.to_dict()
|
|
try:
|
|
updated = self._repository.stage_update(
|
|
subscribe_id,
|
|
_normalized_subscription_patch(subscribe, payload),
|
|
)
|
|
if not updated:
|
|
self._unit_of_work.rollback()
|
|
return None
|
|
event_key, event_payload = _modified_event(
|
|
subscribe_id,
|
|
scene,
|
|
old,
|
|
updated,
|
|
)
|
|
self._outbox.stage(
|
|
OutboxIntent(
|
|
event_key=event_key,
|
|
topic=SUBSCRIBE_MODIFIED_TOPIC,
|
|
payload=event_payload,
|
|
),
|
|
datetime.now(timezone.utc),
|
|
)
|
|
self._unit_of_work.commit()
|
|
except Exception:
|
|
self._unit_of_work.rollback()
|
|
raise
|
|
|
|
delivered = deliver_outbox_effect(
|
|
self._dispatch_store,
|
|
event_key,
|
|
lambda: self._publish_modified(event_payload),
|
|
)
|
|
return SubscriptionMutation(
|
|
snapshot=updated,
|
|
old=old,
|
|
new=updated.to_dict(),
|
|
event_published=delivered,
|
|
business_committed=True,
|
|
pending_effects=() if delivered else (event_key,),
|
|
)
|
|
|
|
def update_status(
|
|
self,
|
|
subscribe_id: int,
|
|
state: str,
|
|
actor: SubscriptionActor,
|
|
) -> SubscriptionMutation | None:
|
|
"""
|
|
同步更新订阅状态并返回前后快照。
|
|
"""
|
|
return self.update(
|
|
subscribe_id,
|
|
{"state": state},
|
|
actor,
|
|
scene="status",
|
|
)
|
|
|
|
def reset(
|
|
self,
|
|
subscribe_id: int,
|
|
actor: SubscriptionActor,
|
|
) -> SubscriptionMutation | None:
|
|
"""
|
|
同步重置订阅进度和手工集数标记。
|
|
"""
|
|
subscribe = self.get_accessible(subscribe_id, actor)
|
|
if not subscribe:
|
|
return None
|
|
with subscription_log_scope(subscribe_id, "reset"):
|
|
result = self.update(
|
|
subscribe_id,
|
|
_reset_payload(subscribe),
|
|
actor,
|
|
existing=subscribe,
|
|
scene="reset",
|
|
)
|
|
logger.scoped_info("订阅进度已重置，下次搜索将重新检查缺失内容" if result is not None else "本次重置未完成，订阅记录可能已变更")
|
|
return result
|
|
|
|
|
|
class SubscriptionMutationService:
|
|
"""
|
|
编排订阅访问控制、更新和历史删除。
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
repository: SessionSubscriptionPort,
|
|
unit_of_work: AsyncUnitOfWork,
|
|
outbox: AsyncOutboxStager,
|
|
dispatch_store: AsyncOutboxDispatchStore,
|
|
publish_modified: SubscribeModifiedPublisher,
|
|
history_repository: SubscriptionHistoryStagingPort | None = None,
|
|
) -> None:
|
|
"""
|
|
注入订阅数据、事务与 durable 事件端口。
|
|
"""
|
|
self._repository = repository
|
|
self._history_repository = history_repository
|
|
self._unit_of_work = unit_of_work
|
|
self._outbox = outbox
|
|
self._dispatch_store = dispatch_store
|
|
self._publish_modified = publish_modified
|
|
|
|
async def get_accessible(
|
|
self,
|
|
subscribe_id: int,
|
|
actor: SubscriptionActor,
|
|
) -> SubscriptionSnapshot | None:
|
|
"""
|
|
读取当前主体可访问的订阅。
|
|
"""
|
|
subscribe = await self._repository.async_get(subscribe_id)
|
|
return subscribe if self.can_access(subscribe, actor) else None
|
|
|
|
async def update(
|
|
self,
|
|
subscribe_id: int,
|
|
payload: dict[str, JsonData],
|
|
actor: SubscriptionActor,
|
|
existing: SubscriptionSnapshot | None = None,
|
|
scene: str = "update",
|
|
) -> SubscriptionMutation | None:
|
|
"""
|
|
更新订阅，并在同一事务暂存可恢复的 SubscribeModified 事件。
|
|
"""
|
|
subscribe = existing or await self.get_accessible(subscribe_id, actor)
|
|
if subscribe and not self.can_access(subscribe, actor):
|
|
return None
|
|
if not subscribe:
|
|
return None
|
|
old = subscribe.to_dict()
|
|
try:
|
|
updated = await self._repository.async_stage_update(
|
|
subscribe_id,
|
|
_normalized_subscription_patch(subscribe, payload),
|
|
)
|
|
if not updated:
|
|
await self._unit_of_work.rollback()
|
|
return None
|
|
event_key, event_payload = _modified_event(
|
|
subscribe_id,
|
|
scene,
|
|
old,
|
|
updated,
|
|
)
|
|
await self._outbox.stage(
|
|
OutboxIntent(
|
|
event_key=event_key,
|
|
topic=SUBSCRIBE_MODIFIED_TOPIC,
|
|
payload=event_payload,
|
|
),
|
|
datetime.now(timezone.utc),
|
|
)
|
|
await self._unit_of_work.commit()
|
|
except Exception:
|
|
await self._unit_of_work.rollback()
|
|
raise
|
|
|
|
async def publish_event() -> None:
|
|
"""
|
|
发布本次事务已持久化的订阅修改事件。
|
|
"""
|
|
await self._publish_modified(event_payload)
|
|
|
|
delivered = await deliver_async_outbox_effect(
|
|
self._dispatch_store,
|
|
event_key,
|
|
publish_event,
|
|
)
|
|
return SubscriptionMutation(
|
|
snapshot=updated,
|
|
old=old,
|
|
new=updated.to_dict(),
|
|
event_published=delivered,
|
|
business_committed=True,
|
|
pending_effects=() if delivered else (event_key,),
|
|
)
|
|
|
|
async def update_status(
|
|
self,
|
|
subscribe_id: int,
|
|
state: str,
|
|
actor: SubscriptionActor,
|
|
) -> SubscriptionMutation | None:
|
|
"""
|
|
更新订阅状态并返回前后快照。
|
|
"""
|
|
return await self.update(
|
|
subscribe_id,
|
|
{"state": state},
|
|
actor,
|
|
scene="status",
|
|
)
|
|
|
|
async def reset(
|
|
self,
|
|
subscribe_id: int,
|
|
actor: SubscriptionActor,
|
|
) -> SubscriptionMutation | None:
|
|
"""
|
|
重置订阅进度和手工集数标记。
|
|
"""
|
|
subscribe = await self.get_accessible(subscribe_id, actor)
|
|
if not subscribe:
|
|
return None
|
|
with subscription_log_scope(subscribe_id, "reset"):
|
|
result = await self.update(
|
|
subscribe_id,
|
|
_reset_payload(subscribe),
|
|
actor,
|
|
existing=subscribe,
|
|
scene="reset",
|
|
)
|
|
logger.scoped_info("订阅进度已重置，下次搜索将重新检查缺失内容" if result is not None else "本次重置未完成，订阅记录可能已变更")
|
|
return result
|
|
|
|
async def delete_history(
|
|
self,
|
|
history_id: int,
|
|
actor: SubscriptionActor,
|
|
) -> bool:
|
|
"""
|
|
删除当前主体可访问的订阅历史。
|
|
"""
|
|
return (await self.delete_history_with_status(history_id, actor)) == "deleted"
|
|
|
|
async def delete_history_with_status(
|
|
self,
|
|
history_id: int,
|
|
actor: SubscriptionActor,
|
|
) -> SubscribeHistoryDeletionStatus:
|
|
"""
|
|
删除订阅历史并返回可供 HTTP 层判断的结果状态。
|
|
"""
|
|
if self._history_repository is None:
|
|
raise RuntimeError("订阅历史数据端口未配置")
|
|
history = await self._history_repository.async_get(history_id)
|
|
if history is None:
|
|
return "not_found"
|
|
if not self.can_access(history, actor):
|
|
return "forbidden"
|
|
try:
|
|
await self._history_repository.stage_delete(history_id)
|
|
await self._unit_of_work.commit()
|
|
except Exception:
|
|
await self._unit_of_work.rollback()
|
|
raise
|
|
return "deleted"
|
|
|
|
@staticmethod
|
|
def can_access(
|
|
subscribe: SubscriptionSnapshot | SubscriptionHistorySnapshot | None,
|
|
actor: SubscriptionActor,
|
|
) -> bool:
|
|
"""
|
|
判断主体是否可访问订阅或订阅历史。
|
|
"""
|
|
if not subscribe:
|
|
return False
|
|
if actor.is_superuser:
|
|
return True
|
|
username = subscribe.username
|
|
return bool(username) and username == actor.name
|
|
|
|
|
|
def _modified_event_key(subscribe_id: int, scene: str) -> str:
|
|
"""
|
|
为一次订阅修改生成重试期间稳定且跨多次相同变更不碰撞的幂等键。
|
|
"""
|
|
return f"subscribe.modified:{subscribe_id}:{scene}:{uuid4().hex}:v1"
|
|
|
|
|
|
SubscriptionMutationScope = Callable[
|
|
[],
|
|
AbstractAsyncContextManager[SubscriptionMutationService],
|
|
]
|
|
SyncSubscriptionMutationScope = Callable[
|
|
[],
|
|
AbstractContextManager[SyncSubscriptionMutationService],
|
|
]
|