""" 整理剧集命名与格式规则推荐。 """ from pathlib import Path from typing import List, Optional, Tuple, Union, cast from app.application.configuration import ( get_configured_system_config, ) from app.application.formatting import EpisodeFormatRecommendation, EpisodeFormatRuleHelper from app.application.history.retry import max_failed_retries from app.application.transfer.feedback import TransferFailureNotification, classify_transfer_failure from app.application.transfer.workflow import TransferTask from app.chain._contracts import TransferMixinHost from app.chain.storage import StorageChain from app.chain.transfer.contract import _TransferOwnerBase from app.domain.context import MediaInfo, MusicInfo from app.domain.meta.metabase import MetaBase from app.runtime.log import logger from app.schemas.file import FileItem from app.schemas.tmdb import TmdbEpisode from app.schemas.transfer import EpisodeFormatRule as _SchemaEpisodeFormatRule from app.schemas.transfer import TransferInfo from app.schemas.types import ( MediaType, SystemConfigKey, ) class EpisodeFormatMixin(_TransferOwnerBase): """ 提供剧集命名建议、格式规则选择和样例发现。 """ __mixin_host_protocol__ = TransferMixinHost def recommend_name( self, meta: Optional[MetaBase], mediainfo: Union[MediaInfo, MusicInfo], ) -> Optional[str]: """ 获取重命名后的名称 :param meta: 元数据 :param mediainfo: 媒体信息 :return: 重命名后的名称（含目录） """ if meta is None: return None # 获取集信息，供重命名模块使用 episodes_info: Optional[List[TmdbEpisode]] = None if mediainfo.type == MediaType.TV: # 第零季表示特别篇，不能按空值回退为第一季。 season_num = mediainfo.season if season_num is None and meta.season_seq: if meta.season_seq.isdigit(): season_num = int(meta.season_seq) if season_num is None: season_num = 1 episodes_info = self.run_module( "tmdb_episodes", tmdbid=mediainfo.tmdb_id, season=season_num, episode_group=mediainfo.episode_group, ) if episodes_info: return cast( Optional[str], self.run_module( "recommend_name", meta=meta, mediainfo=mediainfo, episodes_info=episodes_info, ), ) # 电影或无集信息时保持原有参数集，避免影响旧签名的模块实现 return cast( Optional[str], self.run_module("recommend_name", meta=meta, mediainfo=mediainfo), ) def recommend_episode_format( self, fileitem: Optional[FileItem], fileitems: Optional[List[FileItem]] = None, ) -> Tuple[bool, str, Optional[EpisodeFormatRecommendation]]: """ 根据目录样本推荐集数定位模板 """ if not fileitem and not fileitems: logger.warn("推荐集数定位模板失败：缺少目录参数") return False, "缺少目录参数", None rules = self._get_episode_format_rules() if fileitems: state, errmsg, sample_files = self._get_selected_episode_format_sample_files( fileitems ) if not state: logger.warn(f"推荐集数定位模板失败：{errmsg}") return False, errmsg, None target_path = sample_files[0].path if sample_files else None else: if not fileitem or not fileitem.path: logger.warn("推荐集数定位模板失败：缺少目录参数") return False, "缺少目录参数", None directory = self._resolve_episode_format_directory(fileitem) if not directory or directory.type != "dir": logger.warn(f"推荐集数定位模板失败：目录不存在 - {fileitem.path}") return False, "目录不存在", None sample_files = self._get_episode_format_sample_files(directory) target_path = directory.path logger.info( f"开始匹配集数定位规则：{target_path}，规则数 {len(rules)}，样本数 {len(sample_files)}" ) state, errmsg, data = EpisodeFormatRuleHelper().recommend( rules=rules, sample_files=sample_files, ) if not state: logger.warn(f"集数定位模板推荐失败：{target_path} - {errmsg}") return state, errmsg, data logger.info( f"集数定位模板推荐成功：{target_path} - 规则 {data.get('rule_name') if data else None}" ) return state, errmsg, data @staticmethod def _get_episode_format_rules() -> List[_SchemaEpisodeFormatRule]: """ 获取启用的集数定位规则 """ rule_items = get_configured_system_config().get(SystemConfigKey.EpisodeFormatRuleTable) or [] rules: List[_SchemaEpisodeFormatRule] = [] for item in rule_items: if not isinstance(item, dict): continue try: rule = _SchemaEpisodeFormatRule(**item) except Exception as err: logger.warn(f"忽略无效的集数定位规则：{err}") continue if rule.enabled: rules.append(rule) return sorted(rules, key=lambda item: item.order) def _resolve_episode_format_directory( self, fileitem: FileItem ) -> Optional[FileItem]: """ 将文件或目录入参归一化为目录对象 """ if not fileitem.path: return None storage_chain = StorageChain() if fileitem.type == "dir": return storage_chain.get_item(fileitem) source_path = Path(fileitem.path) parent_item = FileItem( storage=fileitem.storage, path=source_path.parent.as_posix(), type="dir", name=source_path.parent.name, ) return storage_chain.get_item(parent_item) def _get_selected_episode_format_sample_files( self, fileitems: List[FileItem] ) -> Tuple[bool, str, List[FileItem]]: """ 获取当前选择文件中可参与模板推荐的样本文件。 """ if not fileitems: return False, "没有可用于识别的样本文件", [] expected_dir_key: Optional[Tuple[str, str]] = None selected_files: List[FileItem] = [] seen_files = set() for item in fileitems: if not item or not item.path or item.type != "file": return False, "当前选择不满足智能识别条件", [] dir_key = ( item.storage or "local", Path(item.path).parent.as_posix(), ) if expected_dir_key is None: expected_dir_key = dir_key elif dir_key != expected_dir_key: return False, "当前选择不满足智能识别条件", [] file_key = (item.storage or "local", item.path) if file_key in seen_files: continue seen_files.add(file_key) if not ( self._is_media_file(item) or self._is_subtitle_file(item) or self._is_audio_file(item) ): continue if self._is_hidden_or_recycle_path(item.path): continue selected_files.append(item) if not selected_files: return False, "没有可用于识别的样本文件", [] return True, "", selected_files def _get_episode_format_sample_files( self, directory: FileItem ) -> List[FileItem]: """ 获取目录下可参与模板推荐的样本文件。 推荐结果最终会在手动整理链路中作为 `episode_format` 交由 `FormatParser` 过滤主视频、字幕和外挂音频，因此这里需要把 同目录下的主视频、字幕和外挂音频一起纳入推荐流程。 """ file_items = StorageChain().list_files(directory, recursion=False) or [] sample_files: List[FileItem] = [] for item in file_items: if not item or item.type != "file": continue if not ( self._is_media_file(item) or self._is_subtitle_file(item) or self._is_audio_file(item) ): continue if self._is_hidden_or_recycle_path(item.path): continue sample_files.append(item) return sample_files def build_failure_notification( task: TransferTask, transferinfo: TransferInfo, history_id: Optional[int], *, manual_identity: bool = False, retry_count: Optional[int] = None, auto_paused: bool = False, ) -> TransferFailureNotification: """ 从整理工作项构造失败通知快照，不触发聚合或消息发送。 """ from app.runtime.errors import public_error_message feedback = classify_transfer_failure(transferinfo.message) return TransferFailureNotification( media_title=( task.mediainfo.title_year if task.mediainfo else task.fileitem.name if task.fileitem else None ) or "未知媒体", season_episode=getattr(task.meta, "season_episode", "") or "", reason=public_error_message(transferinfo.message, context="transfer") or "整理失败", history_id=history_id, image=( task.mediainfo.get_message_image() if task.mediainfo and hasattr(task.mediainfo, "get_message_image") else None ), username=task.username, manual_identity=manual_identity, task_id=task.admission_task_id, source_path=task.fileitem.path if task.fileitem else None, target_path=transferinfo.target_item.path if transferinfo.target_item else None, failure_stage=transferinfo.failure_stage or feedback.stage.value, recovery_action=transferinfo.recovery_action or feedback.action, retry_count=retry_count, max_retries=max_failed_retries(), auto_paused=auto_paused, cleanup_status=transferinfo.cleanup_status, cleanup_error=transferinfo.cleanup_error, )