mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-31 12:24:39 +08:00
6d4ce83837
严格检查全部宿主源码，清零类型错误预算并移除按模块类型豁免。 按原因、所属方法、错误码及语法指纹锁定必要兼容例外和站点类型桩， CI 阻断新增忽略、检查范围退化、本地桩替代真实接口及新的类型错误。 修复空值、异步和供应方类型边界，保持插件与 API 兼容合同， 并补充存储、订阅、通知、Agent 和工作流边界的回归覆盖。
875 lines
35 KiB
Python
875 lines
35 KiB
Python
"""
|
||
订阅创建与批量写入编排
|
||
"""
|
||
|
||
from collections.abc import Awaitable, Callable, Sequence
|
||
from typing import Any, Dict, Optional, Tuple, cast
|
||
|
||
from app.application.subscription.write import (
|
||
SubscriptionBatchWritePort,
|
||
SubscriptionCreateRequest,
|
||
add_subscribe,
|
||
async_add_subscribe,
|
||
build_subscription_create_request,
|
||
)
|
||
from app.chain.media import MediaChain
|
||
from app.chain.subscribe.context import _SubscribeCreateContext, _SubscribePostCommitContext
|
||
from app.chain.subscribe.contract import _SubscribeOwnerBase
|
||
from app.chain.subscribe.identity import media_recognize_kwargs
|
||
from app.domain.context import MediaInfo
|
||
from app.domain.meta.metamusic import MetaMusic
|
||
from app.domain.metainfo import MetaInfo
|
||
from app.runtime.log import logger
|
||
from app.runtime.settings import get_runtime_setting
|
||
from app.schemas.common import JsonData
|
||
from app.schemas.media import normalize_media_source, resolve_media_identity
|
||
from app.schemas.types import (
|
||
MediaSource,
|
||
MediaType,
|
||
NotificationChannel,
|
||
)
|
||
|
||
|
||
def _build_post_commit_callback(
|
||
owner: Any,
|
||
context: _SubscribePostCommitContext,
|
||
) -> Callable[[int], bool]:
|
||
"""
|
||
冻结同步订阅提交后的副作用上下文。
|
||
"""
|
||
|
||
def after_commit(subscribe_id: int) -> bool:
|
||
"""
|
||
把同步提交后的副作用委托给通知组件。
|
||
"""
|
||
return cast(bool, owner._SubscribeChain__post_subscribe_added(subscribe_id, context))
|
||
|
||
return after_commit
|
||
|
||
|
||
def _build_async_post_commit_callback(
|
||
owner: Any,
|
||
context: _SubscribePostCommitContext,
|
||
) -> Callable[[int], Awaitable[bool]]:
|
||
"""
|
||
冻结异步订阅提交后的副作用上下文。
|
||
"""
|
||
|
||
async def after_commit(subscribe_id: int) -> bool:
|
||
"""
|
||
把异步提交后的副作用委托给通知组件。
|
||
"""
|
||
return cast(bool, await owner._SubscribeChain__async_post_subscribe_added(subscribe_id, context))
|
||
|
||
return after_commit
|
||
|
||
|
||
class SubscribeCreateOwner(_SubscribeOwnerBase):
|
||
"""
|
||
订阅创建与批量写入编排，作为 SubscribeChain 的单一职责组件。
|
||
"""
|
||
|
||
def _SubscribeChain__get_default_kwargs(self, mtype: MediaType, **kwargs: Any) -> dict[str, Any]:
|
||
"""
|
||
合并调用方选项与对应媒体类型的订阅默认配置。
|
||
|
||
:param mtype: 订阅的媒体类型，音乐会关闭电视剧整包洗版和 IMDb 搜索选项
|
||
:param kwargs: 调用方提供的质量、关键词、站点、目录和洗版等订阅选项
|
||
:return: 合并后的订阅选项字典，空值是否回退按各配置项的既有规则处理
|
||
"""
|
||
defaults = {
|
||
"quality": self._SubscribeChain__get_default_subscribe_config(mtype, "quality")
|
||
if not kwargs.get("quality")
|
||
else kwargs.get("quality"),
|
||
"resolution": self._SubscribeChain__get_default_subscribe_config(mtype, "resolution")
|
||
if not kwargs.get("resolution")
|
||
else kwargs.get("resolution"),
|
||
"effect": self._SubscribeChain__get_default_subscribe_config(mtype, "effect")
|
||
if not kwargs.get("effect")
|
||
else kwargs.get("effect"),
|
||
"audio_quality": self._SubscribeChain__get_default_subscribe_config(mtype, "audio_quality")
|
||
if not kwargs.get("audio_quality")
|
||
else kwargs.get("audio_quality"),
|
||
"audio_format": self._SubscribeChain__get_default_subscribe_config(mtype, "audio_format")
|
||
if not kwargs.get("audio_format")
|
||
else kwargs.get("audio_format"),
|
||
"min_bitrate": self._SubscribeChain__get_default_subscribe_config(mtype, "min_bitrate")
|
||
if not kwargs.get("min_bitrate")
|
||
else kwargs.get("min_bitrate"),
|
||
"min_bit_depth": self._SubscribeChain__get_default_subscribe_config(mtype, "min_bit_depth")
|
||
if not kwargs.get("min_bit_depth")
|
||
else kwargs.get("min_bit_depth"),
|
||
"min_sample_rate": self._SubscribeChain__get_default_subscribe_config(mtype, "min_sample_rate")
|
||
if not kwargs.get("min_sample_rate")
|
||
else kwargs.get("min_sample_rate"),
|
||
"include": self._SubscribeChain__get_default_subscribe_config(mtype, "include")
|
||
if not kwargs.get("include")
|
||
else kwargs.get("include"),
|
||
"exclude": self._SubscribeChain__get_default_subscribe_config(mtype, "exclude")
|
||
if not kwargs.get("exclude")
|
||
else kwargs.get("exclude"),
|
||
"best_version": self._SubscribeChain__get_default_subscribe_config(mtype, "best_version")
|
||
if kwargs.get("best_version") is None
|
||
else kwargs.get("best_version"),
|
||
"best_version_full": self._SubscribeChain__get_default_subscribe_config(mtype, "best_version_full")
|
||
if kwargs.get("best_version_full") is None
|
||
else kwargs.get("best_version_full"),
|
||
"search_imdbid": self._SubscribeChain__get_default_subscribe_config(mtype, "search_imdbid")
|
||
if not kwargs.get("search_imdbid")
|
||
else kwargs.get("search_imdbid"),
|
||
"sites": self._SubscribeChain__get_default_subscribe_config(mtype, "sites") or None
|
||
if not kwargs.get("sites")
|
||
else kwargs.get("sites"),
|
||
"downloader": self._SubscribeChain__get_default_subscribe_config(mtype, "downloader")
|
||
if not kwargs.get("downloader")
|
||
else kwargs.get("downloader"),
|
||
"save_path": self._SubscribeChain__get_default_subscribe_config(mtype, "save_path")
|
||
if not kwargs.get("save_path")
|
||
else kwargs.get("save_path"),
|
||
"filter_groups": self._SubscribeChain__get_default_subscribe_config(mtype, "filter_groups")
|
||
if not kwargs.get("filter_groups")
|
||
else kwargs.get("filter_groups"),
|
||
}
|
||
if mtype == MediaType.MUSIC:
|
||
# 音乐允许按音质洗版，但没有电视剧整包洗版和 IMDB 搜索语义。
|
||
defaults.update(
|
||
{
|
||
"best_version_full": 0,
|
||
"search_imdbid": 0,
|
||
}
|
||
)
|
||
return defaults
|
||
|
||
@staticmethod
|
||
def _SubscribeChain__build_subscribe_create_context(
|
||
title: str,
|
||
year: str,
|
||
mtype: Optional[MediaType],
|
||
episode_group: Optional[str],
|
||
season: Optional[int],
|
||
channel: Optional[NotificationChannel],
|
||
source: Optional[str],
|
||
userid: Optional[str],
|
||
username: Optional[str],
|
||
message: bool,
|
||
exist_ok: bool,
|
||
media_source: Optional[MediaSource],
|
||
media_id: Optional[str],
|
||
options: Dict[str, Any],
|
||
) -> Tuple[Optional[_SubscribeCreateContext], Optional[str]]:
|
||
"""
|
||
规范订阅新增输入，并在任何识别 I/O 前拒绝不完整的显式媒体身份。
|
||
"""
|
||
explicit_identity = media_source is not None or media_id is not None
|
||
media_source, media_id = resolve_media_identity(
|
||
media_source=media_source,
|
||
media_id=media_id,
|
||
)
|
||
if explicit_identity and (not media_source or not media_id):
|
||
return None, "媒体来源和媒体 ID 必须同时提供"
|
||
|
||
metainfo = MetaMusic.parse_query(title) if mtype == MediaType.MUSIC else MetaInfo(title)
|
||
if year:
|
||
metainfo.year = year
|
||
if mtype:
|
||
metainfo.type = mtype
|
||
if season is not None:
|
||
metainfo.type = MediaType.TV
|
||
metainfo.begin_season = season
|
||
if mtype == MediaType.MUSIC and media_id:
|
||
metainfo.media_id = str(media_id)
|
||
|
||
return _SubscribeCreateContext(
|
||
title=title,
|
||
year=year,
|
||
mtype=mtype,
|
||
episode_group=episode_group,
|
||
season=season,
|
||
channel=channel,
|
||
source=source,
|
||
userid=userid,
|
||
username=username,
|
||
message=bool(message),
|
||
exist_ok=bool(exist_ok),
|
||
options=options,
|
||
explicit_identity=explicit_identity,
|
||
media_source=media_source,
|
||
media_id=media_id,
|
||
requested_music_type=options.get("music_type"),
|
||
metainfo=metainfo,
|
||
), None
|
||
|
||
@staticmethod
|
||
def _SubscribeChain__normalize_recognized_subscribe_media(context: _SubscribeCreateContext) -> None:
|
||
"""
|
||
保留非 TMDB 影视标题清洗与从标题补季号的历史行为。
|
||
"""
|
||
mediainfo = context.mediainfo
|
||
if context.mtype == MediaType.MUSIC or not mediainfo or mediainfo.media_source == MediaSource.TMDB:
|
||
return
|
||
meta = MetaInfo(mediainfo.title or "")
|
||
mediainfo.title = meta.name
|
||
if context.season is None:
|
||
context.season = meta.begin_season
|
||
|
||
@staticmethod
|
||
def _SubscribeChain__tmdb_conversion_identity(
|
||
context: _SubscribeCreateContext,
|
||
) -> Optional[Tuple[MediaSource, str]]:
|
||
"""
|
||
识别源为 TMDB 时，返回需要换成 TMDB 整剧身份的影视来源身份；其余情况返回空。
|
||
"""
|
||
mediainfo = context.mediainfo
|
||
if not mediainfo or mediainfo.type not in (MediaType.MOVIE, MediaType.TV):
|
||
return None
|
||
source, media_id = resolve_media_identity(media=mediainfo)
|
||
if not source or not media_id or source == MediaSource.TMDB:
|
||
return None
|
||
if normalize_media_source(get_runtime_setting("RECOGNIZE_SOURCE", None)) != MediaSource.TMDB:
|
||
return None
|
||
return source, media_id
|
||
|
||
@staticmethod
|
||
def _SubscribeChain__apply_converted_subscribe_media(
|
||
context: _SubscribeCreateContext,
|
||
source: MediaSource,
|
||
media_id: str,
|
||
tmdbinfo: Optional[Dict[str, Any]],
|
||
converted: Optional[MediaInfo],
|
||
) -> None:
|
||
"""
|
||
转换成功才替换订阅目标；豆瓣季条目的季号来自标题或投影结果，整剧身份本身不带季。
|
||
"""
|
||
tmdb_id = tmdbinfo.get("id") if tmdbinfo else None
|
||
if not tmdb_id or not converted:
|
||
logger.warning(
|
||
f"未能将 {source}:{media_id} 转换为 TMDB 身份，订阅沿用原身份："
|
||
f"{context.mediainfo.title_year if context.mediainfo else context.title}"
|
||
)
|
||
return
|
||
if context.season is None and converted.type == MediaType.TV:
|
||
projected_season = tmdbinfo.get("season") if tmdbinfo else None
|
||
if isinstance(projected_season, int) and projected_season > 0:
|
||
context.season = projected_season
|
||
logger.info(f"订阅身份已由 {source}:{media_id} 转换为 TMDB {tmdb_id}：{converted.title_year}")
|
||
context.mediainfo = converted
|
||
|
||
def _SubscribeChain__convert_subscribe_media_to_tmdb(self, context: _SubscribeCreateContext) -> None:
|
||
"""
|
||
同步把豆瓣、Bangumi 等影视身份换成 TMDB，和豆瓣想看、豆瓣榜单的订阅口径一致。
|
||
|
||
非 TMDB 来源把每一季当成独立条目，标题和年份都是季级的；若直接作为订阅主身份，
|
||
下载记录与整理会按季标题生成独立剧集目录，媒体库里就会出现多部同名剧。
|
||
"""
|
||
identity = self._SubscribeChain__tmdb_conversion_identity(context)
|
||
mediainfo = context.mediainfo
|
||
if not identity or mediainfo is None:
|
||
return
|
||
source, media_id = identity
|
||
tmdbinfo = MediaChain().convert_media_identity(
|
||
target_source=MediaSource.TMDB,
|
||
media_source=source,
|
||
media_id=media_id,
|
||
mtype=mediainfo.type,
|
||
season=context.season,
|
||
)
|
||
converted = None
|
||
if tmdbinfo and tmdbinfo.get("id"):
|
||
converted = MediaChain().recognize_media(
|
||
meta=context.metainfo,
|
||
mtype=mediainfo.type,
|
||
media_source=MediaSource.TMDB,
|
||
media_id=str(tmdbinfo.get("id")),
|
||
episode_group=context.episode_group,
|
||
cache=False,
|
||
)
|
||
self._SubscribeChain__apply_converted_subscribe_media(context, source, media_id, tmdbinfo, converted)
|
||
|
||
async def _SubscribeChain__async_convert_subscribe_media_to_tmdb(
|
||
self,
|
||
context: _SubscribeCreateContext,
|
||
) -> None:
|
||
"""
|
||
异步版本的 TMDB 身份转换，与同步入口共享判定和替换规则。
|
||
"""
|
||
identity = self._SubscribeChain__tmdb_conversion_identity(context)
|
||
mediainfo = context.mediainfo
|
||
if not identity or mediainfo is None:
|
||
return
|
||
source, media_id = identity
|
||
tmdbinfo = await MediaChain().async_convert_media_identity(
|
||
target_source=MediaSource.TMDB,
|
||
media_source=source,
|
||
media_id=media_id,
|
||
mtype=mediainfo.type,
|
||
season=context.season,
|
||
)
|
||
converted = None
|
||
if tmdbinfo and tmdbinfo.get("id"):
|
||
converted = await MediaChain().async_recognize_media(
|
||
meta=context.metainfo,
|
||
mtype=mediainfo.type,
|
||
media_source=MediaSource.TMDB,
|
||
media_id=str(tmdbinfo.get("id")),
|
||
episode_group=context.episode_group,
|
||
cache=False,
|
||
)
|
||
self._SubscribeChain__apply_converted_subscribe_media(context, source, media_id, tmdbinfo, converted)
|
||
|
||
def _SubscribeChain__recognize_subscribe_media(self, context: _SubscribeCreateContext) -> Optional[str]:
|
||
"""
|
||
同步识别订阅目标；显式身份失败时禁止按标题换成另一个媒体。
|
||
"""
|
||
if context.media_source and context.media_id:
|
||
context.mediainfo = MediaChain().recognize_media(
|
||
meta=context.metainfo,
|
||
mtype=context.mtype,
|
||
media_source=context.media_source,
|
||
media_id=context.media_id,
|
||
music_type=context.requested_music_type,
|
||
episode_group=context.episode_group,
|
||
cache=False,
|
||
)
|
||
self._SubscribeChain__normalize_recognized_subscribe_media(context)
|
||
if not context.mediainfo and not context.explicit_identity:
|
||
context.mediainfo = MediaChain().recognize_by_meta(
|
||
context.metainfo,
|
||
media_source=context.media_source,
|
||
episode_group=context.episode_group,
|
||
obtain_images=False,
|
||
music_type=context.requested_music_type,
|
||
)
|
||
if context.mtype == MediaType.MUSIC and context.mediainfo and not context.mediainfo.media_source:
|
||
context.mediainfo = None
|
||
self._SubscribeChain__convert_subscribe_media_to_tmdb(context)
|
||
return self._SubscribeChain__validate_recognized_subscribe_media(context)
|
||
|
||
async def _SubscribeChain__async_recognize_subscribe_media(
|
||
self,
|
||
context: _SubscribeCreateContext,
|
||
) -> Optional[str]:
|
||
"""
|
||
异步识别订阅目标，并与同步入口共享相同的规范化和校验规则。
|
||
"""
|
||
if context.media_source and context.media_id:
|
||
context.mediainfo = await MediaChain().async_recognize_media(
|
||
meta=context.metainfo,
|
||
mtype=context.mtype,
|
||
media_source=context.media_source,
|
||
media_id=context.media_id,
|
||
music_type=context.requested_music_type,
|
||
episode_group=context.episode_group,
|
||
cache=False,
|
||
)
|
||
self._SubscribeChain__normalize_recognized_subscribe_media(context)
|
||
if not context.mediainfo and not context.explicit_identity:
|
||
context.mediainfo = await MediaChain().async_recognize_by_meta(
|
||
context.metainfo,
|
||
media_source=context.media_source,
|
||
episode_group=context.episode_group,
|
||
obtain_images=False,
|
||
music_type=context.requested_music_type,
|
||
)
|
||
if context.mtype == MediaType.MUSIC and context.mediainfo and not context.mediainfo.media_source:
|
||
context.mediainfo = None
|
||
await self._SubscribeChain__async_convert_subscribe_media_to_tmdb(context)
|
||
return self._SubscribeChain__validate_recognized_subscribe_media(context)
|
||
|
||
def _SubscribeChain__validate_recognized_subscribe_media(
|
||
self,
|
||
context: _SubscribeCreateContext,
|
||
) -> Optional[str]:
|
||
"""
|
||
校验识别结果以及音乐订阅实体，返回兼容旧入口的错误文案。
|
||
"""
|
||
if not context.mediainfo:
|
||
logger.warning(
|
||
f"未识别到媒体信息，标题：{context.title}，媒体来源：{context.media_source}，"
|
||
f"媒体 ID：{context.media_id}"
|
||
)
|
||
return "未识别到媒体信息"
|
||
if context.mtype != MediaType.MUSIC:
|
||
return None
|
||
music_error = self._validate_music_subscribe_target(
|
||
context.mediainfo,
|
||
requested_music_type=context.requested_music_type,
|
||
)
|
||
if music_error:
|
||
logger.warning(f"音乐订阅目标校验失败：{context.title} - {music_error}")
|
||
return cast(Optional[str], music_error)
|
||
|
||
def _SubscribeChain__prepare_subscribe_episodes(self, context: _SubscribeCreateContext) -> Optional[str]:
|
||
"""
|
||
同步补齐电视剧季集信息，并保持外部集数刷新只能扩展创建目标。
|
||
"""
|
||
mediainfo = context.mediainfo
|
||
if mediainfo is None:
|
||
return "未识别到媒体信息"
|
||
if mediainfo.type != MediaType.TV:
|
||
context.season = None
|
||
return None
|
||
if context.season is None:
|
||
context.season = 1
|
||
if not context.options.get("total_episode"):
|
||
if not mediainfo.seasons or context.episode_group:
|
||
mediainfo = MediaChain().recognize_media(
|
||
mtype=mediainfo.type,
|
||
**media_recognize_kwargs(mediainfo),
|
||
episode_group=context.episode_group,
|
||
cache=False,
|
||
)
|
||
context.mediainfo = mediainfo
|
||
if mediainfo is None:
|
||
return "未识别到媒体信息"
|
||
error = self._SubscribeChain__validate_subscribe_seasons(context)
|
||
if error:
|
||
return error
|
||
current_total_episode = len(mediainfo.seasons.get(context.season) or [])
|
||
total_episode = self._SubscribeChain__apply_episodes_refresh(
|
||
current_total_episode,
|
||
season=context.season,
|
||
mediainfo=mediainfo,
|
||
media_source=resolve_media_identity(media=mediainfo)[0],
|
||
media_id=resolve_media_identity(media=mediainfo)[1],
|
||
scene="create",
|
||
)
|
||
error = self._SubscribeChain__store_subscribe_episode_total(
|
||
context,
|
||
current_total_episode,
|
||
total_episode,
|
||
)
|
||
if error:
|
||
return error
|
||
self._SubscribeChain__fill_subscribe_lack_episode(context)
|
||
return None
|
||
|
||
async def _SubscribeChain__async_prepare_subscribe_episodes(
|
||
self,
|
||
context: _SubscribeCreateContext,
|
||
) -> Optional[str]:
|
||
"""
|
||
异步补齐电视剧季集信息，并复用同步入口的结果校验和字段写入规则。
|
||
"""
|
||
mediainfo = context.mediainfo
|
||
if mediainfo is None:
|
||
return "未识别到媒体信息"
|
||
if mediainfo.type != MediaType.TV:
|
||
context.season = None
|
||
return None
|
||
if context.season is None:
|
||
context.season = 1
|
||
if not context.options.get("total_episode"):
|
||
if not mediainfo.seasons or context.episode_group:
|
||
mediainfo = await MediaChain().async_recognize_media(
|
||
mtype=mediainfo.type,
|
||
**media_recognize_kwargs(mediainfo),
|
||
episode_group=context.episode_group,
|
||
cache=False,
|
||
)
|
||
context.mediainfo = mediainfo
|
||
if mediainfo is None:
|
||
return "未识别到媒体信息"
|
||
error = self._SubscribeChain__validate_subscribe_seasons(context)
|
||
if error:
|
||
return error
|
||
current_total_episode = len(mediainfo.seasons.get(context.season) or [])
|
||
total_episode = await self._SubscribeChain__async_apply_episodes_refresh(
|
||
current_total_episode,
|
||
season=context.season,
|
||
mediainfo=mediainfo,
|
||
media_source=resolve_media_identity(media=mediainfo)[0],
|
||
media_id=resolve_media_identity(media=mediainfo)[1],
|
||
scene="create",
|
||
)
|
||
error = self._SubscribeChain__store_subscribe_episode_total(
|
||
context,
|
||
current_total_episode,
|
||
total_episode,
|
||
)
|
||
if error:
|
||
return error
|
||
self._SubscribeChain__fill_subscribe_lack_episode(context)
|
||
return None
|
||
|
||
@staticmethod
|
||
def _SubscribeChain__validate_subscribe_seasons(context: _SubscribeCreateContext) -> Optional[str]:
|
||
"""
|
||
校验补充识别结果是否仍包含创建电视剧订阅所需的季集信息。
|
||
"""
|
||
if not context.mediainfo:
|
||
logger.error("媒体信息识别失败！")
|
||
return "媒体信息识别失败"
|
||
if not context.mediainfo.seasons:
|
||
logger.error(f"媒体信息中没有季集信息，标题：{context.title}")
|
||
return "媒体信息中没有季集信息"
|
||
return None
|
||
|
||
@staticmethod
|
||
def _SubscribeChain__store_subscribe_episode_total(
|
||
context: _SubscribeCreateContext,
|
||
current_total_episode: int,
|
||
total_episode: int,
|
||
) -> Optional[str]:
|
||
"""
|
||
写入创建场景最终集数，阻止外部刷新把可靠的当前集数向下覆盖。
|
||
"""
|
||
if current_total_episode and total_episode < current_total_episode:
|
||
total_episode = current_total_episode
|
||
if not total_episode:
|
||
logger.error(f"未获取到总集数，标题：{context.title}")
|
||
return f"未获取到第 {context.season} 季的总集数"
|
||
context.options["total_episode"] = total_episode
|
||
return None
|
||
|
||
@staticmethod
|
||
def _SubscribeChain__fill_subscribe_lack_episode(context: _SubscribeCreateContext) -> None:
|
||
"""
|
||
未显式指定缺失集数时沿用总集数，保持旧创建默认值。
|
||
"""
|
||
if not context.options.get("lack_episode"):
|
||
context.options["lack_episode"] = context.options.get("total_episode")
|
||
|
||
def _SubscribeChain__finalize_subscribe_create_context(self, context: _SubscribeCreateContext) -> None:
|
||
"""
|
||
同步补图并写入规范媒体身份和订阅默认配置。
|
||
"""
|
||
mediainfo = context.mediainfo
|
||
if mediainfo is None:
|
||
return
|
||
if mediainfo.type != MediaType.MUSIC:
|
||
self.obtain_images(mediainfo=mediainfo)
|
||
self._SubscribeChain__apply_subscribe_create_defaults(context)
|
||
|
||
async def _SubscribeChain__async_finalize_subscribe_create_context(
|
||
self,
|
||
context: _SubscribeCreateContext,
|
||
) -> None:
|
||
"""
|
||
异步补图并写入规范媒体身份和订阅默认配置。
|
||
"""
|
||
mediainfo = context.mediainfo
|
||
if mediainfo is None:
|
||
return
|
||
if mediainfo.type != MediaType.MUSIC:
|
||
await self.async_obtain_images(mediainfo=mediainfo)
|
||
self._SubscribeChain__apply_subscribe_create_defaults(context)
|
||
|
||
def _SubscribeChain__apply_subscribe_create_defaults(self, context: _SubscribeCreateContext) -> None:
|
||
"""
|
||
以最终识别结果覆盖身份字段，并补齐当前媒体类型的默认订阅参数。
|
||
"""
|
||
mediainfo = context.mediainfo
|
||
if mediainfo is None:
|
||
return
|
||
context.media_source, context.media_id = resolve_media_identity(media=mediainfo)
|
||
context.options.update(
|
||
{
|
||
"media_source": context.media_source,
|
||
"media_id": context.media_id,
|
||
}
|
||
)
|
||
if mediainfo.type is not None:
|
||
context.options.update(self._SubscribeChain__get_default_kwargs(mediainfo.type, **context.options))
|
||
|
||
@staticmethod
|
||
def _SubscribeChain__subscribe_post_commit_context(
|
||
context: _SubscribeCreateContext,
|
||
notification: Optional[dict[str, Any]] = None,
|
||
) -> _SubscribePostCommitContext:
|
||
"""
|
||
从创建阶段状态冻结提交后副作用需要的最小快照。
|
||
"""
|
||
if context.mediainfo is None:
|
||
raise ValueError("订阅提交上下文缺少媒体信息")
|
||
return _SubscribePostCommitContext(
|
||
title=context.title,
|
||
year=context.year,
|
||
metainfo=context.metainfo,
|
||
mediainfo=context.mediainfo,
|
||
media_source=context.media_source,
|
||
media_id=context.media_id,
|
||
season=context.season,
|
||
channel=context.channel,
|
||
source=context.source,
|
||
userid=context.userid,
|
||
username=context.username,
|
||
message=context.message,
|
||
notification=notification,
|
||
)
|
||
|
||
def _SubscribeChain__persist_subscribe_create(self, context: _SubscribeCreateContext) -> Tuple[Optional[int], str]:
|
||
"""
|
||
同步提交订阅，并在提交成功后按原顺序执行消息、事件和统计。
|
||
"""
|
||
if context.mediainfo is None:
|
||
return None, "未识别到媒体信息"
|
||
mediainfo = context.mediainfo
|
||
post_commit_context = self._SubscribeChain__subscribe_post_commit_context(
|
||
context,
|
||
self._SubscribeChain__build_subscribe_notification(context),
|
||
)
|
||
try:
|
||
sid, err_msg = add_subscribe(
|
||
mediainfo=mediainfo,
|
||
subscribe_oper=self.subscription_repository,
|
||
season=context.season,
|
||
username=context.username,
|
||
after_commit=_build_post_commit_callback(self, post_commit_context),
|
||
notification=post_commit_context.notification,
|
||
occurrence_id=post_commit_context.occurrence_id,
|
||
**context.options,
|
||
)
|
||
except ValueError as error:
|
||
logger.error(f"订阅分类设置无效：{error}", exc_info=True)
|
||
err_msg = "订阅分类设置无效，请重新选择分类后重试"
|
||
self._SubscribeChain__notify_subscribe_create_failure(context, err_msg)
|
||
return None, err_msg
|
||
if not sid:
|
||
self._SubscribeChain__notify_subscribe_create_failure(context, err_msg)
|
||
return None, err_msg
|
||
return sid, err_msg
|
||
|
||
async def _SubscribeChain__async_persist_subscribe_create(
|
||
self,
|
||
context: _SubscribeCreateContext,
|
||
) -> Tuple[Optional[int], str]:
|
||
"""
|
||
异步提交订阅，并在提交成功后按原顺序执行消息、事件和统计。
|
||
"""
|
||
if context.mediainfo is None:
|
||
return None, "未识别到媒体信息"
|
||
mediainfo = context.mediainfo
|
||
post_commit_context = self._SubscribeChain__subscribe_post_commit_context(
|
||
context,
|
||
self._SubscribeChain__build_subscribe_notification(context),
|
||
)
|
||
try:
|
||
sid, err_msg = await async_add_subscribe(
|
||
mediainfo=mediainfo,
|
||
subscribe_oper=self.subscription_repository,
|
||
season=context.season,
|
||
username=context.username,
|
||
after_commit=_build_async_post_commit_callback(self, post_commit_context),
|
||
notification=post_commit_context.notification,
|
||
occurrence_id=post_commit_context.occurrence_id,
|
||
**context.options,
|
||
)
|
||
except ValueError as error:
|
||
logger.error(f"订阅分类设置无效：{error}", exc_info=True)
|
||
err_msg = "订阅分类设置无效，请重新选择分类后重试"
|
||
await self._SubscribeChain__async_notify_subscribe_create_failure(
|
||
context,
|
||
err_msg,
|
||
)
|
||
return None, err_msg
|
||
if not sid:
|
||
await self._SubscribeChain__async_notify_subscribe_create_failure(context, err_msg)
|
||
return None, err_msg
|
||
return sid, err_msg
|
||
|
||
def add(
|
||
self,
|
||
title: str,
|
||
year: str,
|
||
mtype: Optional[MediaType] = None,
|
||
episode_group: Optional[str] = None,
|
||
season: Optional[int] = None,
|
||
channel: Optional[NotificationChannel] = None,
|
||
source: Optional[str] = None,
|
||
userid: Optional[str] = None,
|
||
username: Optional[str] = None,
|
||
message: Optional[bool] = True,
|
||
exist_ok: Optional[bool] = False,
|
||
media_source: Optional[MediaSource] = None,
|
||
media_id: Optional[str] = None,
|
||
**kwargs: Any,
|
||
) -> Tuple[Optional[int], str]:
|
||
"""
|
||
识别媒体信息并添加订阅
|
||
"""
|
||
logger.info(f"开始添加订阅，标题：{title} ...")
|
||
context, error = self._SubscribeChain__build_subscribe_create_context(
|
||
title,
|
||
year,
|
||
mtype,
|
||
episode_group,
|
||
season,
|
||
channel,
|
||
source,
|
||
userid,
|
||
username,
|
||
bool(message),
|
||
bool(exist_ok),
|
||
media_source,
|
||
media_id,
|
||
kwargs,
|
||
)
|
||
if error or context is None:
|
||
return None, error or "订阅准备失败"
|
||
error = self._SubscribeChain__recognize_subscribe_media(context)
|
||
if error:
|
||
return None, error
|
||
if context.mediainfo is None:
|
||
return None, "未识别到媒体信息"
|
||
error = self._SubscribeChain__prepare_subscribe_episodes(context)
|
||
if error:
|
||
return None, error
|
||
self._SubscribeChain__finalize_subscribe_create_context(context)
|
||
return self._SubscribeChain__persist_subscribe_create(context)
|
||
|
||
async def async_add(
|
||
self,
|
||
title: str,
|
||
year: str,
|
||
mtype: Optional[MediaType] = None,
|
||
episode_group: Optional[str] = None,
|
||
season: Optional[int] = None,
|
||
channel: Optional[NotificationChannel] = None,
|
||
source: Optional[str] = None,
|
||
userid: Optional[str] = None,
|
||
username: Optional[str] = None,
|
||
message: Optional[bool] = True,
|
||
exist_ok: Optional[bool] = False,
|
||
media_source: Optional[MediaSource] = None,
|
||
media_id: Optional[str] = None,
|
||
**kwargs: Any,
|
||
) -> Tuple[Optional[int], str]:
|
||
"""
|
||
异步识别媒体信息并添加订阅
|
||
"""
|
||
logger.info(f"开始添加订阅，标题：{title} ...")
|
||
context, error = await self._SubscribeChain__async_prepare_subscribe_create(
|
||
title,
|
||
year,
|
||
mtype,
|
||
episode_group,
|
||
season,
|
||
channel,
|
||
source,
|
||
userid,
|
||
username,
|
||
bool(message),
|
||
bool(exist_ok),
|
||
media_source,
|
||
media_id,
|
||
kwargs,
|
||
)
|
||
if error or context is None or context.mediainfo is None:
|
||
return None, error or "订阅准备失败"
|
||
return await self._SubscribeChain__async_persist_subscribe_create(context)
|
||
|
||
async def _SubscribeChain__async_prepare_subscribe_create(
|
||
self,
|
||
title: str,
|
||
year: str,
|
||
mtype: Optional[MediaType],
|
||
episode_group: Optional[str],
|
||
season: Optional[int],
|
||
channel: Optional[NotificationChannel],
|
||
source: Optional[str],
|
||
userid: Optional[str],
|
||
username: Optional[str],
|
||
message: bool,
|
||
exist_ok: bool,
|
||
media_source: Optional[MediaSource],
|
||
media_id: Optional[str],
|
||
options: Dict[str, Any],
|
||
) -> Tuple[Optional[_SubscribeCreateContext], Optional[str]]:
|
||
"""
|
||
执行异步新增的识别、季集补齐和默认配置准备，但不触发持久化。
|
||
"""
|
||
context, error = self._SubscribeChain__build_subscribe_create_context(
|
||
title,
|
||
year,
|
||
mtype,
|
||
episode_group,
|
||
season,
|
||
channel,
|
||
source,
|
||
userid,
|
||
username,
|
||
message,
|
||
exist_ok,
|
||
media_source,
|
||
media_id,
|
||
options,
|
||
)
|
||
if error or context is None:
|
||
return None, error
|
||
error = await self._SubscribeChain__async_recognize_subscribe_media(context)
|
||
if error:
|
||
return None, error
|
||
error = await self._SubscribeChain__async_prepare_subscribe_episodes(context)
|
||
if error:
|
||
return None, error
|
||
await self._SubscribeChain__async_finalize_subscribe_create_context(context)
|
||
return context, None
|
||
|
||
async def async_add_batch(
|
||
self,
|
||
*,
|
||
title: str,
|
||
year: str,
|
||
seasons: Sequence[int],
|
||
batch_writer: SubscriptionBatchWritePort,
|
||
mtype: Optional[MediaType] = None,
|
||
episode_group: Optional[str] = None,
|
||
channel: Optional[NotificationChannel] = None,
|
||
source: Optional[str] = None,
|
||
userid: Optional[str] = None,
|
||
username: Optional[str] = None,
|
||
message: Optional[bool] = True,
|
||
exist_ok: Optional[bool] = False,
|
||
media_source: Optional[MediaSource] = None,
|
||
media_id: Optional[str] = None,
|
||
**kwargs: JsonData,
|
||
) -> Tuple[Optional[int], str]:
|
||
"""
|
||
先完整准备各季订阅，再交给共享事务批量写端口原子提交。
|
||
"""
|
||
requests: list[SubscriptionCreateRequest] = []
|
||
for season in seasons:
|
||
context, error = await self._SubscribeChain__async_prepare_subscribe_create(
|
||
title,
|
||
year,
|
||
mtype,
|
||
episode_group,
|
||
season,
|
||
channel,
|
||
source,
|
||
userid,
|
||
username,
|
||
bool(message),
|
||
bool(exist_ok),
|
||
media_source,
|
||
media_id,
|
||
dict(kwargs),
|
||
)
|
||
if error or context is None:
|
||
return None, error or "订阅准备失败"
|
||
assert context.mediainfo is not None
|
||
notification = self._SubscribeChain__build_subscribe_notification(context)
|
||
post_commit_context = self._SubscribeChain__subscribe_post_commit_context(
|
||
context,
|
||
notification,
|
||
)
|
||
|
||
request = build_subscription_create_request(
|
||
context.mediainfo,
|
||
notification=notification,
|
||
after_commit=_build_async_post_commit_callback(self, post_commit_context),
|
||
occurrence_id=post_commit_context.occurrence_id,
|
||
season=context.season,
|
||
username=context.username,
|
||
**context.options,
|
||
)
|
||
if request is None:
|
||
return None, "媒体身份不完整"
|
||
requests.append(request)
|
||
|
||
results = await batch_writer.async_add(requests)
|
||
if not results:
|
||
return None, "未提供订阅季"
|
||
result = results[-1]
|
||
return result.subscribe_id or None, result.message
|