diff --git a/app/application/transfer/jobs.py b/app/application/transfer/jobs.py new file mode 100644 index 000000000..719f30b9b --- /dev/null +++ b/app/application/transfer/jobs.py @@ -0,0 +1,852 @@ +"""进程内整理作业视图、状态心跳与作业级查询。""" + +from __future__ import annotations + +import threading +from copy import deepcopy +from pathlib import Path +from time import monotonic as _system_monotonic +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Protocol, Tuple, Union, cast + +from app.domain.context import MediaInfo, MusicInfo +from app.domain.media import normalize_music_type +from app.domain.meta.metabase import MetaBase +from app.domain.meta.metamusic import MetaMusic +from app.foundation import text as text_tools +from app.runtime.log import logger +from app.schemas.context import MediaInfo as _SchemaMediaInfo +from app.schemas.context import MetaInfo as _SchemaMetaInfo +from app.schemas.file import FileItem +from app.schemas.media import resolve_media_identity +from app.schemas.music import MusicInfo as _SchemaMusicInfo +from app.schemas.music import MusicMeta as _SchemaMusicMeta +from app.schemas.transfer import TransferJob, TransferJobTask +from app.schemas.types import MUSIC_ENTITY_ALBUM, MUSIC_ENTITY_RECORDING, MediaType + +if TYPE_CHECKING: + from app.application.transfer.workflow import TransferTask + + +def monotonic() -> float: + """读取旧 workflow 暴露的可替换时钟入口，保持失活检测注入兼容。""" + from app.application.transfer import workflow + + return workflow.monotonic() if workflow.monotonic is not monotonic else _system_monotonic() + +JobId = tuple[object, ...] +FileKey = tuple[str, str] + + +class _DictionarySerializable(Protocol): + """描述领域对象沿用的字典投影能力。""" + + def to_dict(self) -> dict[str, Any]: + """返回领域对象的字典投影。""" + + +class DirectorySize(Protocol): + """描述本地文件或目录大小读取能力。""" + + def get_directory_size(self, path: Path) -> int: + """返回真实本地路径占用的字节数。""" + ... + + +_directory_size: Optional[DirectorySize] = None + + +def configure_directory_size(reader: Optional[DirectorySize]) -> None: + """由启动组合根注入或清除本地目录大小适配器。""" + global _directory_size + _directory_size = reader + + +def _domain_to_dict(value: object) -> dict[str, Any]: + """按领域对象既有 ``to_dict`` 合同生成字典投影。""" + return cast(_DictionarySerializable, value).to_dict() + + +def _job_tasks(job: TransferJob) -> list[TransferJobTask]: + """声明进程内作业始终使用已初始化的任务列表。""" + return cast(list[TransferJobTask], job.tasks) + + +def _job_task_fileitem(task: TransferJobTask) -> FileItem: + """声明进程内作业任务始终绑定源文件。""" + return cast(FileItem, task.fileitem) + + +def _job_task_size(task: TransferJobTask) -> int: + """按既有本地目录回退规则返回已完成任务的文件大小。""" + fileitem = _job_task_fileitem(task) + if fileitem.size is not None: + return fileitem.size + if fileitem.storage == "local": + if _directory_size is None: + raise RuntimeError("本地目录大小能力尚未由启动组合根配置") + return _directory_size.get_directory_size(Path(cast(str, fileitem.path))) + return 0 + + +def _transfer_task_meta(task: "TransferTask") -> MetaBase: + """声明进入作业管理器的整理任务已经完成元数据解析。""" + return cast(MetaBase, task.meta) + + +job_lock = threading.Lock() + + +class JobManager: + """ + 作业管理器 + task任务负责一个文件的整理，job作业负责一个媒体的整理 + """ + + # 整理中的作业 + _job_view: Dict[JobId, TransferJob] = {} + # 汇总季集清单 + _season_episodes: Dict[JobId, List[int]] = {} + # 记录从 meta 作业迁移到 media 作业的关系，用于清理提前失败后残留的 media 作业 + _meta_to_media_ids: Dict[JobId, set[JobId]] = {} + # 记录任务最近一次状态心跳，供外部异步接管任务的失活检测使用 + _task_state_changed_at: Dict[FileKey, float] = {} + # 记录仍由主程序整理线程直接执行的任务，避免把阻塞中的本地任务误判为失活 + _active_executions: set[FileKey] = set() + + def __init__(self) -> None: + """初始化当前进程内的整理作业状态。""" + self._job_view = {} + self._season_episodes = {} + self._meta_to_media_ids = {} + self._task_state_changed_at = {} + self._active_executions = set() + + @staticmethod + def __get_meta_id( + meta: Optional[MetaBase] = None, + season: Optional[int] = None, + ) -> JobId: + """ + 获取元数据ID + """ + return cast(MetaBase, meta).name, season + + @staticmethod + def __get_media_id(media: Optional[Union[MediaInfo, MusicInfo]] = None, + season: Optional[int] = None) -> JobId: + """ + 获取媒体ID；音乐额外区分实体类型，并为无远端ID的曲目构造稳定身份。 + """ + if not media: + return None, season + source, media_id = resolve_media_identity(media=media) + if getattr(media, "type", None) == MediaType.MUSIC: + music_type = normalize_music_type( + getattr(media, "music_type", None), + ) or MUSIC_ENTITY_RECORDING + if source and media_id: + return "music", source, media_id, music_type + + artists = tuple( + text_tools.normalize_upper(artist) + for artist in (getattr(media, "artists", None) or []) + if text_tools.normalize_upper(artist) + ) + if music_type == MUSIC_ENTITY_ALBUM: + album_artist = text_tools.normalize_upper( + getattr(media, "album_artist", None) + or (artists[0] if artists else "") + ) + album = text_tools.normalize_upper( + getattr(media, "album", None) or getattr(media, "title", None) or "" + ) + return "music", "local", music_type, album_artist, album, getattr(media, "year", None) + + return ( + "music", + "local", + music_type, + artists, + text_tools.normalize_upper(getattr(media, "title", None) or ""), + text_tools.normalize_upper(getattr(media, "album", None) or ""), + getattr(media, "disc_number", None), + getattr(media, "track_number", None), + ) + return (source, media_id), season + + @staticmethod + def __get_file_key(fileitem: FileItem) -> Optional[Tuple[str, str]]: + """ + 获取源文件唯一键，用于跨媒体作业识别同一个整理任务。 + """ + if not fileitem or not fileitem.path: + return None + normalized_path = ( + Path(str(fileitem.path).replace("\\", "/")).as_posix().rstrip("/") or "/" + ) + return fileitem.storage or "local", normalized_path + + def __get_id(self, task: Optional[TransferTask] = None) -> JobId: + """ + 获取作业ID + """ + resolved_task = task + meta = _transfer_task_meta(resolved_task) + if resolved_task.mediainfo: + return self.__get_media_id( + media=resolved_task.mediainfo, season=meta.begin_season + ) + return self.__get_meta_id(meta=meta, season=meta.begin_season) + + def get_job_id(self, task: TransferTask) -> JobId: + """返回任务当前所属的稳定作业身份，供作业级附加状态隔离使用。""" + return self.__get_id(task) + + @staticmethod + def __get_media(task: TransferTask) -> Union[_SchemaMediaInfo, _SchemaMusicInfo]: + """ + 获取媒体信息 + """ + if task.mediainfo: + # 有媒体信息 + mediainfo = deepcopy(task.mediainfo) + mediainfo.clear() + if isinstance(mediainfo, MusicInfo): + return _SchemaMusicInfo(**_domain_to_dict(mediainfo)) + return _SchemaMediaInfo(**_domain_to_dict(mediainfo)) + else: + # 没有媒体信息 + meta = _transfer_task_meta(task) + if isinstance(meta, MetaMusic): + # 未识别的音乐按已解析元数据兜底展示；音乐年份为 int， + # 不能复用 MediaInfo（year 为 str），否则触发 pydantic 校验异常 + return _SchemaMusicInfo( + title=meta.name, + artists=list(meta.artists or []), + artist=meta.artist, + album=meta.album, + album_artist=meta.album_artist, + year=meta.year, + title_year=f"{meta.name} ({meta.year})" if meta.year else meta.name, + media_source=meta.media_source, + media_id=meta.media_id, + ) + return _SchemaMediaInfo( + title=meta.name, + year=meta.year, + title_year=f"{meta.name} ({meta.year})", + type=meta.type.value if meta.type else None, + ) + + @staticmethod + def __get_meta(task: TransferTask) -> _SchemaMetaInfo: + """ + 获取元数据 + """ + if isinstance(task.meta, MetaMusic): + return _SchemaMusicMeta(**task.meta.to_dict()) + return _SchemaMetaInfo(**_domain_to_dict(_transfer_task_meta(task))) + + def add_task(self, task: TransferTask, state: Optional[str] = "waiting") -> bool: + """ + 添加整理任务，自动分组到对应的作业中 + :return: True表示任务已添加，False表示任务无效或已存在（重复） + """ + if not all([task, task.meta, task.fileitem]): + return False + file_key = self.__get_file_key(task.fileitem) + if not file_key: + return False + with job_lock: + __mediaid__ = self.__get_id(task) + # 同一个源文件可能在识别前后落入不同作业，必须跨作业去重。 + if any( + self.__get_file_key(_job_task_fileitem(t)) == file_key + for job in self._job_view.values() + for t in _job_tasks(job) + ): + logger.debug(f"任务 {task.fileitem.name} 已存在，跳过重复添加") + return False + if __mediaid__ not in self._job_view: + self._job_view[__mediaid__] = TransferJob( + media=self.__get_media(task), + season=_transfer_task_meta(task).begin_season, + tasks=[ + TransferJobTask( + fileitem=task.fileitem, + meta=self.__get_meta(task), + downloader=task.downloader, + download_hash=task.download_hash, + state=state, + ) + ], + ) + else: + # 同锁内的跨作业检查已覆盖当前作业，直接追加通过去重的任务。 + _job_tasks(self._job_view[__mediaid__]).append( + TransferJobTask( + fileitem=task.fileitem, + meta=self.__get_meta(task), + downloader=task.downloader, + download_hash=task.download_hash, + state=state, + ) + ) + self._task_state_changed_at[file_key] = monotonic() + # 添加季集信息 + if self._season_episodes.get(__mediaid__): + self._season_episodes[__mediaid__].extend( + _transfer_task_meta(task).episode_list + ) + self._season_episodes[__mediaid__] = list( + set(self._season_episodes[__mediaid__]) + ) + else: + self._season_episodes[__mediaid__] = _transfer_task_meta(task).episode_list + return True + + def migrate_task(self, task: TransferTask) -> bool: + """ + 将任务从 meta 作业迁移到 media 作业 + """ + curr_task, source_job_id = self.__remove_task_with_job_id( + task.fileitem, preserve_execution=True + ) + if not self.add_task(task, state=curr_task.state if curr_task else "waiting"): + return False + if curr_task and task.mediainfo: + meta = _transfer_task_meta(task) + metaid = self.__get_meta_id(meta=meta, season=meta.begin_season) + mediaid = self.__get_id(task) + if source_job_id == metaid and mediaid != metaid: + with job_lock: + self._meta_to_media_ids.setdefault(metaid, set()).add(mediaid) + return True + + def __is_job_done(self, job_id: JobId) -> bool: + """ + 检查指定作业是否已完成 + """ + if job_id not in self._job_view: + return True + return all( + task.state in ["completed", "failed"] + for task in _job_tasks(self._job_view[job_id]) + ) + + def __pop_job(self, job_id: JobId) -> None: + """ + 移除指定作业和对应季集缓存 + """ + job = self._job_view.pop(job_id, None) + self._season_episodes.pop(job_id, None) + if not job: + return + for task in _job_tasks(job): + file_key = self.__get_file_key(_job_task_fileitem(task)) + if file_key: + self._task_state_changed_at.pop(file_key, None) + self._active_executions.discard(file_key) + + def __remove_done_job_groups(self, job_ids: set[JobId]) -> None: + """ + 清理已进入终态的独立作业或关联作业组。 + """ + candidates = set(job_ids) + for metaid, mediaids in list(self._meta_to_media_ids.items()): + related_ids = {metaid, *mediaids} + if not related_ids.intersection(candidates): + continue + if all(self.__is_job_done(job_id) for job_id in related_ids): + for job_id in related_ids: + self.__pop_job(job_id) + self._meta_to_media_ids.pop(metaid, None) + candidates.difference_update(related_ids) + + referenced_ids = { + job_id + for metaid, mediaids in self._meta_to_media_ids.items() + for job_id in {metaid, *mediaids} + } + for job_id in candidates - referenced_ids: + if self.__is_job_done(job_id): + self.__pop_job(job_id) + + def start_execution(self, task: TransferTask) -> None: + """ + 标记任务仍由主程序整理线程直接执行。 + + :param task: 整理任务 + """ + if not task or not task.fileitem: + return + file_key = self.__get_file_key(task.fileitem) + if not file_key: + return + with job_lock: + self._active_executions.add(file_key) + + def finish_execution(self, task: TransferTask) -> None: + """ + 结束主程序整理线程对任务的直接执行标记。 + + :param task: 整理任务 + """ + if not task or not task.fileitem: + return + file_key = self.__get_file_key(task.fileitem) + if not file_key: + return + with job_lock: + self._active_executions.discard(file_key) + + def expire_stale_running_tasks( + self, timeout_seconds: int + ) -> List[tuple[FileItem, int]]: + """ + 将外部接管后长期无心跳的运行中任务标记失败并清理作业视图。 + + 主程序整理线程仍在直接执行的任务不会被清理，以免把阻塞中的真实任务 + 误报为已终止。外部接管方可重复调用 ``running_task`` 刷新状态心跳。 + + :param timeout_seconds: 失活超时秒数，小于等于 0 时禁用 + :return: 已失活任务及其无心跳秒数 + """ + if timeout_seconds <= 0: + return [] + + current_time = monotonic() + expired: List[tuple[FileItem, int]] = [] + affected_job_ids: set[JobId] = set() + with job_lock: + for mediaid, job in self._job_view.items(): + for task in _job_tasks(job): + fileitem = _job_task_fileitem(task) + file_key = self.__get_file_key(fileitem) + if ( + not file_key + or task.state != "running" + or file_key in self._active_executions + ): + continue + updated_at = self._task_state_changed_at.get(file_key, current_time) + inactive_seconds = current_time - updated_at + if inactive_seconds < timeout_seconds: + continue + task.state = "failed" + self._task_state_changed_at[file_key] = current_time + episodes = getattr(task.meta, "episode_list", None) or [] + if mediaid in self._season_episodes: + self._season_episodes[mediaid] = list( + set(self._season_episodes[mediaid]) - set(episodes) + ) + expired.append((fileitem, int(inactive_seconds))) + affected_job_ids.add(mediaid) + + self.__remove_done_job_groups(affected_job_ids) + return expired + + def running_task(self, task: TransferTask) -> None: + """ + 设置任务为运行中，并刷新外部异步任务的状态心跳。 + """ + with job_lock: + __mediaid__ = self.__get_id(task) + if __mediaid__ not in self._job_view: + return + # 更新状态 + for t in _job_tasks(self._job_view[__mediaid__]): + if t.fileitem == task.fileitem: + t.state = "running" + file_key = self.__get_file_key(_job_task_fileitem(t)) + if file_key: + self._task_state_changed_at[file_key] = monotonic() + break + + def finish_task(self, task: TransferTask) -> None: + """ + 设置任务为完成/成功 + """ + with job_lock: + __mediaid__ = self.__get_id(task) + if __mediaid__ not in self._job_view: + return + # 更新状态 + for t in _job_tasks(self._job_view[__mediaid__]): + if t.fileitem == task.fileitem: + t.state = "completed" + file_key = self.__get_file_key(_job_task_fileitem(t)) + if file_key: + self._task_state_changed_at[file_key] = monotonic() + break + + def fail_task(self, task: TransferTask) -> None: + """ + 设置任务为失败 + """ + with job_lock: + __mediaid__ = self.__get_id(task) + if __mediaid__ not in self._job_view: + return + # 更新状态 + for t in _job_tasks(self._job_view[__mediaid__]): + if t.fileitem == task.fileitem: + t.state = "failed" + file_key = self.__get_file_key(_job_task_fileitem(t)) + if file_key: + self._task_state_changed_at[file_key] = monotonic() + break + # 移除剧集信息 + if __mediaid__ in self._season_episodes: + self._season_episodes[__mediaid__] = list( + set(self._season_episodes[__mediaid__]) + - set(_transfer_task_meta(task).episode_list) + ) + + def fail_unfinished_task(self, task: TransferTask) -> None: + """ + 将指定任务视图中的非终态任务标记为失败 + """ + if not task or not task.fileitem: + return + file_key = self.__get_file_key(task.fileitem) + if not file_key: + return + with job_lock: + for mediaid, job in self._job_view.items(): + for job_task in _job_tasks(job): + if self.__get_file_key(_job_task_fileitem(job_task)) != file_key: + continue + if job_task.state not in ["completed", "failed"]: + job_task.state = "failed" + self._task_state_changed_at[file_key] = monotonic() + if mediaid in self._season_episodes: + self._season_episodes[mediaid] = list( + set(self._season_episodes[mediaid]) + - set(_transfer_task_meta(task).episode_list) + ) + return + + def remove_task( + self, fileitem: FileItem, *, finished_only: bool = False, + ) -> Optional[TransferJobTask]: + """ + 按源文件移除任务；手动重做只清除已退出执行的终态视图。 + """ + task, _ = self.__remove_task_with_job_id(fileitem, finished_only=finished_only) + return task + + def __remove_task_with_job_id( + self, + fileitem: FileItem, + preserve_execution: bool = False, + finished_only: bool = False, + ) -> tuple[Optional[TransferJobTask], Optional[JobId]]: + """ + 根据文件项移除任务，并返回任务所在的作业ID + """ + file_key = self.__get_file_key(fileitem) + if not file_key: + return None, None + with job_lock: + for mediaid in list(self._job_view): + job = self._job_view[mediaid] + for task in _job_tasks(job): + if self.__get_file_key(_job_task_fileitem(task)) == file_key: + if finished_only and ( + task.state not in {"completed", "failed"} + or file_key in self._active_executions + ): + return None, None + _job_tasks(job).remove(task) + self._task_state_changed_at.pop(file_key, None) + if not preserve_execution: + self._active_executions.discard(file_key) + # 如果没有作业了，则移除作业 + if not _job_tasks(job): + self._job_view.pop(mediaid) + # 移除季集信息 + if mediaid in self._season_episodes: + episodes = getattr(task.meta, "episode_list", None) or [] + self._season_episodes[mediaid] = list( + set(self._season_episodes[mediaid]) + - set(episodes) + ) + return task, mediaid + return None, None + + def remove_job(self, task: TransferTask) -> Optional[TransferJob]: + """ + 移除任务对应的作业（强制，线程不安全） + """ + with job_lock: + __mediaid__ = self.__get_id(task) + if __mediaid__ in self._job_view: + job = self._job_view[__mediaid__] + self.__pop_job(__mediaid__) + return job + return None + + def try_remove_job(self, task: TransferTask) -> None: + """ + 尝试移除任务对应的作业（严格检查未完成作业，线程安全） + """ + with job_lock: + meta = _transfer_task_meta(task) + __metaid__ = self.__get_meta_id(meta=meta, season=meta.begin_season) + __mediaid__ = self.__get_media_id( + media=task.mediainfo, season=meta.begin_season + ) + + related_media_ids = set(self._meta_to_media_ids.get(__metaid__, set())) + if task.mediainfo: + related_media_ids.add(__mediaid__) + + meta_done = self.__is_job_done(__metaid__) + media_done = all( + self.__is_job_done(mediaid) for mediaid in related_media_ids + ) + + if meta_done and media_done: + remove_ids = {__metaid__, self.__get_id(task), *related_media_ids} + for job_id in remove_ids: + self.__pop_job(job_id) + self._meta_to_media_ids.pop(__metaid__, None) + + def is_done(self, task: TransferTask) -> bool: + """ + 检查任务对应的作业是否整理完成（不管成功还是失败） + """ + with job_lock: + meta = _transfer_task_meta(task) + __metaid__ = self.__get_meta_id(meta=meta, season=meta.begin_season) + __mediaid__ = self.__get_media_id( + media=task.mediainfo, season=meta.begin_season + ) + if __metaid__ in self._job_view: + meta_done = all( + task.state in ["completed", "failed"] + for task in _job_tasks(self._job_view[__metaid__]) + ) + else: + meta_done = True + if __mediaid__ in self._job_view: + media_done = all( + task.state in ["completed", "failed"] + for task in _job_tasks(self._job_view[__mediaid__]) + ) + else: + media_done = True + return meta_done and media_done + + def is_finished(self, task: TransferTask) -> bool: + """ + 检查任务对应的作业是否已完成且有成功的记录 + """ + with job_lock: + meta = _transfer_task_meta(task) + __metaid__ = self.__get_meta_id(meta=meta, season=meta.begin_season) + __mediaid__ = self.__get_media_id( + media=task.mediainfo, season=meta.begin_season + ) + if __metaid__ in self._job_view: + meta_finished = all( + task.state in ["completed", "failed"] + for task in _job_tasks(self._job_view[__metaid__]) + ) + else: + meta_finished = True + if __mediaid__ in self._job_view: + tasks = _job_tasks(self._job_view[__mediaid__]) + media_finished = all( + task.state in ["completed", "failed"] for task in tasks + ) and any(task.state == "completed" for task in tasks) + else: + media_finished = True + return meta_finished and media_finished + + def is_success(self, task: TransferTask) -> bool: + """ + 检查任务对应的作业是否全部成功 + """ + with job_lock: + meta = _transfer_task_meta(task) + __metaid__ = self.__get_meta_id(meta=meta, season=meta.begin_season) + __mediaid__ = self.__get_media_id( + media=task.mediainfo, season=meta.begin_season + ) + if __metaid__ in self._job_view: + meta_success = all( + task.state in ["completed"] + for task in _job_tasks(self._job_view[__metaid__]) + ) + else: + meta_success = True + if __mediaid__ in self._job_view: + media_success = all( + task.state in ["completed"] + for task in _job_tasks(self._job_view[__mediaid__]) + ) + else: + media_success = True + return meta_success and media_success + + def get_all_torrent_hashes(self) -> set[str]: + """ + 获取所有种子的哈希值集合 + """ + with job_lock: + return { + cast(str, task.download_hash) + for job in self._job_view.values() + for task in _job_tasks(job) + } + + def is_torrent_done(self, download_hash: str) -> bool: + """ + 检查指定种子的所有任务是否都已完成 + """ + with job_lock: + if any( + task.state not in {"completed", "failed"} + for job in self._job_view.values() + for task in _job_tasks(job) + if task.download_hash == download_hash + ): + return False + return True + + def is_torrent_success(self, download_hash: str) -> bool: + """ + 检查指定种子的所有任务是否都已成功 + """ + with job_lock: + if any( + task.state != "completed" + for job in self._job_view.values() + for task in _job_tasks(job) + if task.download_hash == download_hash + ): + return False + return True + + def has_tasks( + self, + meta: MetaBase, + mediainfo: Optional[MediaInfo] = None, + season: Optional[int] = None, + ) -> bool: + """ + 判断作业是否还有任务正在处理 + """ + with job_lock: + if mediainfo: + __mediaid__ = self.__get_media_id(media=mediainfo, season=season) + if __mediaid__ in self._job_view: + return True + + __metaid__ = self.__get_meta_id(meta=meta, season=season) + return ( + __metaid__ in self._job_view + and len(_job_tasks(self._job_view[__metaid__])) > 0 + ) + + def success_tasks( + self, media: Union[MediaInfo, MusicInfo], season: Optional[int] = None + ) -> List[TransferJobTask]: + """ + 获取作业中所有成功的任务 + """ + with job_lock: + __mediaid__ = self.__get_media_id(media=media, season=season) + if __mediaid__ not in self._job_view: + return [] + return [ + task + for task in _job_tasks(self._job_view[__mediaid__]) + if task.state == "completed" + ] + + def all_tasks( + self, media: MediaInfo, season: Optional[int] = None + ) -> List[TransferJobTask]: + """ + 获取作业中全部任务 + """ + with job_lock: + __mediaid__ = self.__get_media_id(media=media, season=season) + if __mediaid__ not in self._job_view: + return [] + return _job_tasks(self._job_view[__mediaid__]) + + def count(self, media: Union[MediaInfo, MusicInfo], season: Optional[int] = None) -> int: + """ + 获取作业中成功总数 + """ + with job_lock: + __mediaid__ = self.__get_media_id(media=media, season=season) + if __mediaid__ not in self._job_view: + return 0 + return len( + [ + task + for task in _job_tasks(self._job_view[__mediaid__]) + if task.state == "completed" + ] + ) + + def size(self, media: Union[MediaInfo, MusicInfo], season: Optional[int] = None) -> int: + """ + 获取作业中所有成功文件总大小 + """ + with job_lock: + __mediaid__ = self.__get_media_id(media=media, season=season) + if __mediaid__ not in self._job_view: + return 0 + return sum( + [ + _job_task_size(task) + for task in _job_tasks(self._job_view[__mediaid__]) + if task.state == "completed" + ] + ) + + def total(self) -> int: + """ + 获取所有任务总数 + """ + with job_lock: + return sum([len(_job_tasks(job)) for job in self._job_view.values()]) + + def pending_total(self) -> int: + """ + 获取未到终态的任务总数。 + + 作业要等关联任务全部终态才整体移除,追更/分批场景下已完成任务会 + 跨批次残留在视图中;批次统计若用全量 total() 会把历史任务计入 + 「当前共 N 个文件」并压低进度百分比,因此只数未终态任务。 + """ + with job_lock: + return sum( + 1 + for job in self._job_view.values() + for task in _job_tasks(job) + if task.state not in ("completed", "failed") + ) + + def list_jobs(self) -> List[TransferJob]: + """ + 获取所有作业的任务列表 + """ + with job_lock: + return list(self._job_view.values()) + + def season_episodes( + self, media: Union[MediaInfo, MusicInfo], season: Optional[int] = None + ) -> List[int]: + """ + 获取作业的季集清单 + """ + with job_lock: + __mediaid__ = self.__get_media_id(media=media, season=season) + return self._season_episodes.get(__mediaid__) or [] diff --git a/app/application/transfer/workflow.py b/app/application/transfer/workflow.py index a39ea5258..a7e76fd34 100644 --- a/app/application/transfer/workflow.py +++ b/app/application/transfer/workflow.py @@ -14,7 +14,6 @@ TransferJob / TransferJobTask，那两个用 app.schemas 的同名 DTO——一 视图，分开表达之后两边都不必再迁就对方。 """ import hashlib -import threading from copy import deepcopy from dataclasses import dataclass, field from pathlib import Path @@ -30,7 +29,6 @@ from typing import ( Tuple, TypeAlias, Union, - cast, ) from pydantic import BaseModel, ConfigDict, PrivateAttr @@ -38,29 +36,29 @@ from pydantic import BaseModel, ConfigDict, PrivateAttr from app.application.history import DownloadHistorySnapshot from app.application.transfer import checkpoint as checkpoint_codec from app.application.transfer.execution import TransferExecutionCheckpoint +from app.application.transfer.jobs import ( + DirectorySize, + JobManager, + _domain_to_dict, + _transfer_task_meta, + configure_directory_size, + job_lock, +) from app.application.transfer.notifications import ( TransferFailureNotification, TransferFailureNotificationAggregator, build_transfer_failure_group_key, ) from app.domain.context import MediaInfo, MusicInfo -from app.domain.media import normalize_music_type from app.domain.meta.metabase import MetaBase -from app.domain.meta.metamusic import MetaMusic -from app.foundation import text as text_tools from app.runtime.log import logger -from app.schemas.context import MediaInfo as _SchemaMediaInfo from app.schemas.context import MetaInfo as _SchemaMetaInfo from app.schemas.file import FileItem -from app.schemas.media import OptionalMediaIdentityMixin, resolve_media_identity -from app.schemas.music import MusicInfo as _SchemaMusicInfo -from app.schemas.music import MusicMeta as _SchemaMusicMeta +from app.schemas.media import OptionalMediaIdentityMixin from app.schemas.system import TransferDirectoryConf from app.schemas.tmdb import TmdbEpisode -from app.schemas.transfer import TransferInfo, TransferJob, TransferJobTask +from app.schemas.transfer import TransferInfo, TransferJob from app.schemas.types import ( - MUSIC_ENTITY_ALBUM, - MUSIC_ENTITY_RECORDING, MediaSource, MediaType, ) @@ -84,62 +82,6 @@ JobId: TypeAlias = tuple[object, ...] FileKey: TypeAlias = tuple[str, str] -class _DictionarySerializable(Protocol): - """描述领域对象沿用的字典投影能力。""" - - def to_dict(self) -> dict[str, Any]: - """返回领域对象的字典投影。""" - - -class DirectorySize(Protocol): - """描述本地文件或目录大小读取能力。""" - - def get_directory_size(self, path: Path) -> int: - """返回真实本地路径占用的字节数。""" - ... - - -_directory_size: Optional[DirectorySize] = None - - -def configure_directory_size(reader: Optional[DirectorySize]) -> None: - """由启动组合根注入或清除本地目录大小适配器。""" - global _directory_size - _directory_size = reader - - -def _domain_to_dict(value: object) -> dict[str, Any]: - """按领域对象既有 ``to_dict`` 合同生成字典投影。""" - return cast(_DictionarySerializable, value).to_dict() - - -def _job_tasks(job: TransferJob) -> list[TransferJobTask]: - """声明进程内作业始终使用已初始化的任务列表。""" - return cast(list[TransferJobTask], job.tasks) - - -def _job_task_fileitem(task: TransferJobTask) -> FileItem: - """声明进程内作业任务始终绑定源文件。""" - return cast(FileItem, task.fileitem) - - -def _job_task_size(task: TransferJobTask) -> int: - """按既有本地目录回退规则返回已完成任务的文件大小。""" - fileitem = _job_task_fileitem(task) - if fileitem.size is not None: - return fileitem.size - if fileitem.storage == "local": - if _directory_size is None: - raise RuntimeError("本地目录大小能力尚未由启动组合根配置") - return _directory_size.get_directory_size(Path(cast(str, fileitem.path))) - return 0 - - -def _transfer_task_meta(task: "TransferTask") -> MetaBase: - """声明进入作业管理器的整理任务已经完成元数据解析。""" - return cast(MetaBase, task.meta) - - TRANSFER_ADMISSION_ACCEPTED = "accepted" TRANSFER_ADMISSION_PROVIDER_PENDING = "provider_pending" TRANSFER_ADMISSION_PLANNED = "planned" @@ -1023,763 +965,3 @@ class TransferQueueService: """先处理失活任务，再返回当前整理作业视图。""" self._expire_tasks() return self._list_tasks() - - -# 作业锁：JobManager 与 TransferChain 共享，保护整理作业视图。 -job_lock = threading.Lock() - - -class JobManager: - """ - 作业管理器 - task任务负责一个文件的整理，job作业负责一个媒体的整理 - """ - - # 整理中的作业 - _job_view: Dict[JobId, TransferJob] = {} - # 汇总季集清单 - _season_episodes: Dict[JobId, List[int]] = {} - # 记录从 meta 作业迁移到 media 作业的关系，用于清理提前失败后残留的 media 作业 - _meta_to_media_ids: Dict[JobId, set[JobId]] = {} - # 记录任务最近一次状态心跳，供外部异步接管任务的失活检测使用 - _task_state_changed_at: Dict[FileKey, float] = {} - # 记录仍由主程序整理线程直接执行的任务，避免把阻塞中的本地任务误判为失活 - _active_executions: set[FileKey] = set() - - def __init__(self) -> None: - """初始化当前进程内的整理作业状态。""" - self._job_view = {} - self._season_episodes = {} - self._meta_to_media_ids = {} - self._task_state_changed_at = {} - self._active_executions = set() - - @staticmethod - def __get_meta_id( - meta: Optional[MetaBase] = None, - season: Optional[int] = None, - ) -> JobId: - """ - 获取元数据ID - """ - return cast(MetaBase, meta).name, season - - @staticmethod - def __get_media_id(media: Optional[Union[MediaInfo, MusicInfo]] = None, - season: Optional[int] = None) -> JobId: - """ - 获取媒体ID；音乐额外区分实体类型，并为无远端ID的曲目构造稳定身份。 - """ - if not media: - return None, season - source, media_id = resolve_media_identity(media=media) - if getattr(media, "type", None) == MediaType.MUSIC: - music_type = normalize_music_type( - getattr(media, "music_type", None), - ) or MUSIC_ENTITY_RECORDING - if source and media_id: - return "music", source, media_id, music_type - - artists = tuple( - text_tools.normalize_upper(artist) - for artist in (getattr(media, "artists", None) or []) - if text_tools.normalize_upper(artist) - ) - if music_type == MUSIC_ENTITY_ALBUM: - album_artist = text_tools.normalize_upper( - getattr(media, "album_artist", None) - or (artists[0] if artists else "") - ) - album = text_tools.normalize_upper( - getattr(media, "album", None) or getattr(media, "title", None) or "" - ) - return "music", "local", music_type, album_artist, album, getattr(media, "year", None) - - return ( - "music", - "local", - music_type, - artists, - text_tools.normalize_upper(getattr(media, "title", None) or ""), - text_tools.normalize_upper(getattr(media, "album", None) or ""), - getattr(media, "disc_number", None), - getattr(media, "track_number", None), - ) - return (source, media_id), season - - @staticmethod - def __get_file_key(fileitem: FileItem) -> Optional[Tuple[str, str]]: - """ - 获取源文件唯一键，用于跨媒体作业识别同一个整理任务。 - """ - if not fileitem or not fileitem.path: - return None - normalized_path = ( - Path(str(fileitem.path).replace("\\", "/")).as_posix().rstrip("/") or "/" - ) - return fileitem.storage or "local", normalized_path - - def __get_id(self, task: Optional[TransferTask] = None) -> JobId: - """ - 获取作业ID - """ - resolved_task = cast(TransferTask, task) - meta = _transfer_task_meta(resolved_task) - if resolved_task.mediainfo: - return self.__get_media_id( - media=resolved_task.mediainfo, season=meta.begin_season - ) - return self.__get_meta_id(meta=meta, season=meta.begin_season) - - def get_job_id(self, task: TransferTask) -> JobId: - """返回任务当前所属的稳定作业身份，供作业级附加状态隔离使用。""" - return self.__get_id(task) - - @staticmethod - def __get_media(task: TransferTask) -> Union[_SchemaMediaInfo, _SchemaMusicInfo]: - """ - 获取媒体信息 - """ - if task.mediainfo: - # 有媒体信息 - mediainfo = deepcopy(task.mediainfo) - mediainfo.clear() - if isinstance(mediainfo, MusicInfo): - return _SchemaMusicInfo(**_domain_to_dict(mediainfo)) - return _SchemaMediaInfo(**_domain_to_dict(mediainfo)) - else: - # 没有媒体信息 - meta = _transfer_task_meta(task) - if isinstance(meta, MetaMusic): - # 未识别的音乐按已解析元数据兜底展示；音乐年份为 int， - # 不能复用 MediaInfo（year 为 str），否则触发 pydantic 校验异常 - return _SchemaMusicInfo( - title=meta.name, - artists=list(meta.artists or []), - artist=meta.artist, - album=meta.album, - album_artist=meta.album_artist, - year=meta.year, - title_year=f"{meta.name} ({meta.year})" if meta.year else meta.name, - media_source=meta.media_source, - media_id=meta.media_id, - ) - return _SchemaMediaInfo( - title=meta.name, - year=meta.year, - title_year=f"{meta.name} ({meta.year})", - type=meta.type.value if meta.type else None, - ) - - @staticmethod - def __get_meta(task: TransferTask) -> _SchemaMetaInfo: - """ - 获取元数据 - """ - if isinstance(task.meta, MetaMusic): - return _SchemaMusicMeta(**task.meta.to_dict()) - return _SchemaMetaInfo(**_domain_to_dict(_transfer_task_meta(task))) - - def add_task(self, task: TransferTask, state: Optional[str] = "waiting") -> bool: - """ - 添加整理任务，自动分组到对应的作业中 - :return: True表示任务已添加，False表示任务无效或已存在（重复） - """ - if not all([task, task.meta, task.fileitem]): - return False - file_key = self.__get_file_key(task.fileitem) - if not file_key: - return False - with job_lock: - __mediaid__ = self.__get_id(task) - # 同一个源文件可能在识别前后落入不同作业，必须跨作业去重。 - if any( - self.__get_file_key(_job_task_fileitem(t)) == file_key - for job in self._job_view.values() - for t in _job_tasks(job) - ): - logger.debug(f"任务 {task.fileitem.name} 已存在，跳过重复添加") - return False - if __mediaid__ not in self._job_view: - self._job_view[__mediaid__] = TransferJob( - media=self.__get_media(task), - season=_transfer_task_meta(task).begin_season, - tasks=[ - TransferJobTask( - fileitem=task.fileitem, - meta=self.__get_meta(task), - downloader=task.downloader, - download_hash=task.download_hash, - state=state, - ) - ], - ) - else: - # 同锁内的跨作业检查已覆盖当前作业，直接追加通过去重的任务。 - _job_tasks(self._job_view[__mediaid__]).append( - TransferJobTask( - fileitem=task.fileitem, - meta=self.__get_meta(task), - downloader=task.downloader, - download_hash=task.download_hash, - state=state, - ) - ) - self._task_state_changed_at[file_key] = monotonic() - # 添加季集信息 - if self._season_episodes.get(__mediaid__): - self._season_episodes[__mediaid__].extend( - _transfer_task_meta(task).episode_list - ) - self._season_episodes[__mediaid__] = list( - set(self._season_episodes[__mediaid__]) - ) - else: - self._season_episodes[__mediaid__] = _transfer_task_meta(task).episode_list - return True - - def migrate_task(self, task: TransferTask) -> bool: - """ - 将任务从 meta 作业迁移到 media 作业 - """ - curr_task, source_job_id = self.__remove_task_with_job_id( - task.fileitem, preserve_execution=True - ) - if not self.add_task(task, state=curr_task.state if curr_task else "waiting"): - return False - if curr_task and task.mediainfo: - meta = _transfer_task_meta(task) - metaid = self.__get_meta_id(meta=meta, season=meta.begin_season) - mediaid = self.__get_id(task) - if source_job_id == metaid and mediaid != metaid: - with job_lock: - self._meta_to_media_ids.setdefault(metaid, set()).add(mediaid) - return True - - def __is_job_done(self, job_id: JobId) -> bool: - """ - 检查指定作业是否已完成 - """ - if job_id not in self._job_view: - return True - return all( - task.state in ["completed", "failed"] - for task in _job_tasks(self._job_view[job_id]) - ) - - def __pop_job(self, job_id: JobId) -> None: - """ - 移除指定作业和对应季集缓存 - """ - job = self._job_view.pop(job_id, None) - self._season_episodes.pop(job_id, None) - if not job: - return - for task in _job_tasks(job): - file_key = self.__get_file_key(_job_task_fileitem(task)) - if file_key: - self._task_state_changed_at.pop(file_key, None) - self._active_executions.discard(file_key) - - def __remove_done_job_groups(self, job_ids: set[JobId]) -> None: - """ - 清理已进入终态的独立作业或关联作业组。 - """ - candidates = set(job_ids) - for metaid, mediaids in list(self._meta_to_media_ids.items()): - related_ids = {metaid, *mediaids} - if not related_ids.intersection(candidates): - continue - if all(self.__is_job_done(job_id) for job_id in related_ids): - for job_id in related_ids: - self.__pop_job(job_id) - self._meta_to_media_ids.pop(metaid, None) - candidates.difference_update(related_ids) - - referenced_ids = { - job_id - for metaid, mediaids in self._meta_to_media_ids.items() - for job_id in {metaid, *mediaids} - } - for job_id in candidates - referenced_ids: - if self.__is_job_done(job_id): - self.__pop_job(job_id) - - def start_execution(self, task: TransferTask) -> None: - """ - 标记任务仍由主程序整理线程直接执行。 - - :param task: 整理任务 - """ - if not task or not task.fileitem: - return - file_key = self.__get_file_key(task.fileitem) - if not file_key: - return - with job_lock: - self._active_executions.add(file_key) - - def finish_execution(self, task: TransferTask) -> None: - """ - 结束主程序整理线程对任务的直接执行标记。 - - :param task: 整理任务 - """ - if not task or not task.fileitem: - return - file_key = self.__get_file_key(task.fileitem) - if not file_key: - return - with job_lock: - self._active_executions.discard(file_key) - - def expire_stale_running_tasks( - self, timeout_seconds: int - ) -> List[tuple[FileItem, int]]: - """ - 将外部接管后长期无心跳的运行中任务标记失败并清理作业视图。 - - 主程序整理线程仍在直接执行的任务不会被清理，以免把阻塞中的真实任务 - 误报为已终止。外部接管方可重复调用 ``running_task`` 刷新状态心跳。 - - :param timeout_seconds: 失活超时秒数，小于等于 0 时禁用 - :return: 已失活任务及其无心跳秒数 - """ - if timeout_seconds <= 0: - return [] - - current_time = monotonic() - expired: List[tuple[FileItem, int]] = [] - affected_job_ids: set[JobId] = set() - with job_lock: - for mediaid, job in self._job_view.items(): - for task in _job_tasks(job): - fileitem = _job_task_fileitem(task) - file_key = self.__get_file_key(fileitem) - if ( - not file_key - or task.state != "running" - or file_key in self._active_executions - ): - continue - updated_at = self._task_state_changed_at.get(file_key, current_time) - inactive_seconds = current_time - updated_at - if inactive_seconds < timeout_seconds: - continue - task.state = "failed" - self._task_state_changed_at[file_key] = current_time - episodes = getattr(task.meta, "episode_list", None) or [] - if mediaid in self._season_episodes: - self._season_episodes[mediaid] = list( - set(self._season_episodes[mediaid]) - set(episodes) - ) - expired.append((fileitem, int(inactive_seconds))) - affected_job_ids.add(mediaid) - - self.__remove_done_job_groups(affected_job_ids) - return expired - - def running_task(self, task: TransferTask) -> None: - """ - 设置任务为运行中，并刷新外部异步任务的状态心跳。 - """ - with job_lock: - __mediaid__ = self.__get_id(task) - if __mediaid__ not in self._job_view: - return - # 更新状态 - for t in _job_tasks(self._job_view[__mediaid__]): - if t.fileitem == task.fileitem: - t.state = "running" - file_key = self.__get_file_key(_job_task_fileitem(t)) - if file_key: - self._task_state_changed_at[file_key] = monotonic() - break - - def finish_task(self, task: TransferTask) -> None: - """ - 设置任务为完成/成功 - """ - with job_lock: - __mediaid__ = self.__get_id(task) - if __mediaid__ not in self._job_view: - return - # 更新状态 - for t in _job_tasks(self._job_view[__mediaid__]): - if t.fileitem == task.fileitem: - t.state = "completed" - file_key = self.__get_file_key(_job_task_fileitem(t)) - if file_key: - self._task_state_changed_at[file_key] = monotonic() - break - - def fail_task(self, task: TransferTask) -> None: - """ - 设置任务为失败 - """ - with job_lock: - __mediaid__ = self.__get_id(task) - if __mediaid__ not in self._job_view: - return - # 更新状态 - for t in _job_tasks(self._job_view[__mediaid__]): - if t.fileitem == task.fileitem: - t.state = "failed" - file_key = self.__get_file_key(_job_task_fileitem(t)) - if file_key: - self._task_state_changed_at[file_key] = monotonic() - break - # 移除剧集信息 - if __mediaid__ in self._season_episodes: - self._season_episodes[__mediaid__] = list( - set(self._season_episodes[__mediaid__]) - - set(_transfer_task_meta(task).episode_list) - ) - - def fail_unfinished_task(self, task: TransferTask) -> None: - """ - 将指定任务视图中的非终态任务标记为失败 - """ - if not task or not task.fileitem: - return - file_key = self.__get_file_key(task.fileitem) - if not file_key: - return - with job_lock: - for mediaid, job in self._job_view.items(): - for job_task in _job_tasks(job): - if self.__get_file_key(_job_task_fileitem(job_task)) != file_key: - continue - if job_task.state not in ["completed", "failed"]: - job_task.state = "failed" - self._task_state_changed_at[file_key] = monotonic() - if mediaid in self._season_episodes: - self._season_episodes[mediaid] = list( - set(self._season_episodes[mediaid]) - - set(_transfer_task_meta(task).episode_list) - ) - return - - def remove_task( - self, fileitem: FileItem, *, finished_only: bool = False, - ) -> Optional[TransferJobTask]: - """ - 按源文件移除任务；手动重做只清除已退出执行的终态视图。 - """ - task, _ = self.__remove_task_with_job_id(fileitem, finished_only=finished_only) - return task - - def __remove_task_with_job_id( - self, - fileitem: FileItem, - preserve_execution: bool = False, - finished_only: bool = False, - ) -> tuple[Optional[TransferJobTask], Optional[JobId]]: - """ - 根据文件项移除任务，并返回任务所在的作业ID - """ - file_key = self.__get_file_key(fileitem) - if not file_key: - return None, None - with job_lock: - for mediaid in list(self._job_view): - job = self._job_view[mediaid] - for task in _job_tasks(job): - if self.__get_file_key(_job_task_fileitem(task)) == file_key: - if finished_only and ( - task.state not in {"completed", "failed"} - or file_key in self._active_executions - ): - return None, None - _job_tasks(job).remove(task) - self._task_state_changed_at.pop(file_key, None) - if not preserve_execution: - self._active_executions.discard(file_key) - # 如果没有作业了，则移除作业 - if not _job_tasks(job): - self._job_view.pop(mediaid) - # 移除季集信息 - if mediaid in self._season_episodes: - episodes = getattr(task.meta, "episode_list", None) or [] - self._season_episodes[mediaid] = list( - set(self._season_episodes[mediaid]) - - set(episodes) - ) - return task, mediaid - return None, None - - def remove_job(self, task: TransferTask) -> Optional[TransferJob]: - """ - 移除任务对应的作业（强制，线程不安全） - """ - with job_lock: - __mediaid__ = self.__get_id(task) - if __mediaid__ in self._job_view: - job = self._job_view[__mediaid__] - self.__pop_job(__mediaid__) - return job - return None - - def try_remove_job(self, task: TransferTask) -> None: - """ - 尝试移除任务对应的作业（严格检查未完成作业，线程安全） - """ - with job_lock: - meta = _transfer_task_meta(task) - __metaid__ = self.__get_meta_id(meta=meta, season=meta.begin_season) - __mediaid__ = self.__get_media_id( - media=task.mediainfo, season=meta.begin_season - ) - - related_media_ids = set(self._meta_to_media_ids.get(__metaid__, set())) - if task.mediainfo: - related_media_ids.add(__mediaid__) - - meta_done = self.__is_job_done(__metaid__) - media_done = all( - self.__is_job_done(mediaid) for mediaid in related_media_ids - ) - - if meta_done and media_done: - remove_ids = {__metaid__, self.__get_id(task), *related_media_ids} - for job_id in remove_ids: - self.__pop_job(job_id) - self._meta_to_media_ids.pop(__metaid__, None) - - def is_done(self, task: TransferTask) -> bool: - """ - 检查任务对应的作业是否整理完成（不管成功还是失败） - """ - with job_lock: - meta = _transfer_task_meta(task) - __metaid__ = self.__get_meta_id(meta=meta, season=meta.begin_season) - __mediaid__ = self.__get_media_id( - media=task.mediainfo, season=meta.begin_season - ) - if __metaid__ in self._job_view: - meta_done = all( - task.state in ["completed", "failed"] - for task in _job_tasks(self._job_view[__metaid__]) - ) - else: - meta_done = True - if __mediaid__ in self._job_view: - media_done = all( - task.state in ["completed", "failed"] - for task in _job_tasks(self._job_view[__mediaid__]) - ) - else: - media_done = True - return meta_done and media_done - - def is_finished(self, task: TransferTask) -> bool: - """ - 检查任务对应的作业是否已完成且有成功的记录 - """ - with job_lock: - meta = _transfer_task_meta(task) - __metaid__ = self.__get_meta_id(meta=meta, season=meta.begin_season) - __mediaid__ = self.__get_media_id( - media=task.mediainfo, season=meta.begin_season - ) - if __metaid__ in self._job_view: - meta_finished = all( - task.state in ["completed", "failed"] - for task in _job_tasks(self._job_view[__metaid__]) - ) - else: - meta_finished = True - if __mediaid__ in self._job_view: - tasks = _job_tasks(self._job_view[__mediaid__]) - media_finished = all( - task.state in ["completed", "failed"] for task in tasks - ) and any(task.state == "completed" for task in tasks) - else: - media_finished = True - return meta_finished and media_finished - - def is_success(self, task: TransferTask) -> bool: - """ - 检查任务对应的作业是否全部成功 - """ - with job_lock: - meta = _transfer_task_meta(task) - __metaid__ = self.__get_meta_id(meta=meta, season=meta.begin_season) - __mediaid__ = self.__get_media_id( - media=task.mediainfo, season=meta.begin_season - ) - if __metaid__ in self._job_view: - meta_success = all( - task.state in ["completed"] - for task in _job_tasks(self._job_view[__metaid__]) - ) - else: - meta_success = True - if __mediaid__ in self._job_view: - media_success = all( - task.state in ["completed"] - for task in _job_tasks(self._job_view[__mediaid__]) - ) - else: - media_success = True - return meta_success and media_success - - def get_all_torrent_hashes(self) -> set[str]: - """ - 获取所有种子的哈希值集合 - """ - with job_lock: - return { - cast(str, task.download_hash) - for job in self._job_view.values() - for task in _job_tasks(job) - } - - def is_torrent_done(self, download_hash: str) -> bool: - """ - 检查指定种子的所有任务是否都已完成 - """ - with job_lock: - if any( - task.state not in {"completed", "failed"} - for job in self._job_view.values() - for task in _job_tasks(job) - if task.download_hash == download_hash - ): - return False - return True - - def is_torrent_success(self, download_hash: str) -> bool: - """ - 检查指定种子的所有任务是否都已成功 - """ - with job_lock: - if any( - task.state != "completed" - for job in self._job_view.values() - for task in _job_tasks(job) - if task.download_hash == download_hash - ): - return False - return True - - def has_tasks( - self, - meta: MetaBase, - mediainfo: Optional[MediaInfo] = None, - season: Optional[int] = None, - ) -> bool: - """ - 判断作业是否还有任务正在处理 - """ - with job_lock: - if mediainfo: - __mediaid__ = self.__get_media_id(media=mediainfo, season=season) - if __mediaid__ in self._job_view: - return True - - __metaid__ = self.__get_meta_id(meta=meta, season=season) - return ( - __metaid__ in self._job_view - and len(_job_tasks(self._job_view[__metaid__])) > 0 - ) - - def success_tasks( - self, media: Union[MediaInfo, MusicInfo], season: Optional[int] = None - ) -> List[TransferJobTask]: - """ - 获取作业中所有成功的任务 - """ - with job_lock: - __mediaid__ = self.__get_media_id(media=media, season=season) - if __mediaid__ not in self._job_view: - return [] - return [ - task - for task in _job_tasks(self._job_view[__mediaid__]) - if task.state == "completed" - ] - - def all_tasks( - self, media: MediaInfo, season: Optional[int] = None - ) -> List[TransferJobTask]: - """ - 获取作业中全部任务 - """ - with job_lock: - __mediaid__ = self.__get_media_id(media=media, season=season) - if __mediaid__ not in self._job_view: - return [] - return _job_tasks(self._job_view[__mediaid__]) - - def count(self, media: Union[MediaInfo, MusicInfo], season: Optional[int] = None) -> int: - """ - 获取作业中成功总数 - """ - with job_lock: - __mediaid__ = self.__get_media_id(media=media, season=season) - if __mediaid__ not in self._job_view: - return 0 - return len( - [ - task - for task in _job_tasks(self._job_view[__mediaid__]) - if task.state == "completed" - ] - ) - - def size(self, media: Union[MediaInfo, MusicInfo], season: Optional[int] = None) -> int: - """ - 获取作业中所有成功文件总大小 - """ - with job_lock: - __mediaid__ = self.__get_media_id(media=media, season=season) - if __mediaid__ not in self._job_view: - return 0 - return sum( - [ - _job_task_size(task) - for task in _job_tasks(self._job_view[__mediaid__]) - if task.state == "completed" - ] - ) - - def total(self) -> int: - """ - 获取所有任务总数 - """ - with job_lock: - return sum([len(_job_tasks(job)) for job in self._job_view.values()]) - - def pending_total(self) -> int: - """ - 获取未到终态的任务总数。 - - 作业要等关联任务全部终态才整体移除,追更/分批场景下已完成任务会 - 跨批次残留在视图中;批次统计若用全量 total() 会把历史任务计入 - 「当前共 N 个文件」并压低进度百分比,因此只数未终态任务。 - """ - with job_lock: - return sum( - 1 - for job in self._job_view.values() - for task in _job_tasks(job) - if task.state not in ("completed", "failed") - ) - - def list_jobs(self) -> List[TransferJob]: - """ - 获取所有作业的任务列表 - """ - with job_lock: - return list(self._job_view.values()) - - def season_episodes( - self, media: Union[MediaInfo, MusicInfo], season: Optional[int] = None - ) -> List[int]: - """ - 获取作业的季集清单 - """ - with job_lock: - __mediaid__ = self.__get_media_id(media=media, season=season) - return self._season_episodes.get(__mediaid__) or [] diff --git a/tests/test_application_transfer_architecture.py b/tests/test_application_transfer_architecture.py new file mode 100644 index 000000000..442571970 --- /dev/null +++ b/tests/test_application_transfer_architecture.py @@ -0,0 +1,59 @@ +"""应用层整理模块的职责边界门禁。""" + +import ast +import importlib +from pathlib import Path + + +PROJECT_ROOT = Path(__file__).parents[1] +TRANSFER_ROOT = PROJECT_ROOT / "app" / "application" / "transfer" +WORKFLOW_PATH = TRANSFER_ROOT / "workflow.py" +JOBS_PATH = TRANSFER_ROOT / "jobs.py" + + +def test_job_manager_has_one_canonical_owner_and_legacy_exports() -> None: + """JobManager 与作业锁只能在 jobs.py 定义，workflow.py 只保留兼容导出。""" + workflow_tree = ast.parse( + WORKFLOW_PATH.read_text(encoding="utf-8-sig"), + filename=str(WORKFLOW_PATH), + ) + jobs_tree = ast.parse( + JOBS_PATH.read_text(encoding="utf-8-sig"), + filename=str(JOBS_PATH), + ) + + workflow_classes = { + node.name + for node in workflow_tree.body + if isinstance(node, ast.ClassDef) + } + jobs_classes = { + node.name + for node in jobs_tree.body + if isinstance(node, ast.ClassDef) + } + workflow_assignments = { + target.id + for node in workflow_tree.body + if isinstance(node, ast.Assign) + for target in node.targets + if isinstance(target, ast.Name) + } + jobs_assignments = { + target.id + for node in jobs_tree.body + if isinstance(node, ast.Assign) + for target in node.targets + if isinstance(target, ast.Name) + } + + assert "JobManager" in jobs_classes + assert "JobManager" not in workflow_classes + assert "job_lock" in jobs_assignments + assert "job_lock" not in workflow_assignments + + workflow = importlib.import_module("app.application.transfer.workflow") + jobs = importlib.import_module("app.application.transfer.jobs") + assert workflow.JobManager is jobs.JobManager + assert workflow.job_lock is jobs.job_lock + assert workflow.configure_directory_size is jobs.configure_directory_size