download

Functions:

Name	Description
download_album	下载一个本子（album），包含其所有的章节（photo）
download_photo	下载一个章节（photo），参数同 download_album。
create_option_by_env
create_option_by_file
create_option_by_str
download_album_async	异步下载一个本子及其全部章节。
download_photo_async	异步下载一个章节。
download_batch_async	异步批量下载 album / photo。

create_option = create_option_by_file module-attribute

download_album(jm_album_id, option=None, downloader=None, *, check_exception=True, extra=None, control=None)

下载一个本子（album），包含其所有的章节（photo）

当jm_album_id不是str或int时，视为批量下载，相当于调用 download_batch(download_album, jm_album_id, option, downloader, control=control)

Parameters:

Name	Type	Description	Default
jm_album_id		本子的禁漫车号	required
option		下载选项	None
downloader		下载器类	None
check_exception		仅当 jm_album_id 是单个 ID 时生效。为 True 时检查 downloader 中的部分下载失败， 并上抛 PartialDownloadFailedException。多 ID 调用会转交 download_batch，此参数不生效； 请检查 BatchResult.failed，或自行封装 download_batch 实现所需的批量异常策略。	True
extra		下载特性（Feature），下载时动态挂载的附加行为上下文。会自动根据上下文（如 album/photo 来源）自适应参数行为。支持单个 Feature、FeatureChain、或列表	None
control		可选的取消控制器 (DownloadControl)，用于外部协作式中断下载任务	None

Returns:

Type	Description
Union[__DOWNLOAD_API_RET, Set[__DOWNLOAD_API_RET]]	对于的本子实体类，下载器（如果是上述的批量情况，返回值为download_batch的返回值）

Source code in src/jmcomic/api.py

def download_album(jm_album_id,
                   option=None,
                   downloader=None,
                   *,
                   check_exception=True,
                   extra=None,
                   control=None,
                   ) -> Union[__DOWNLOAD_API_RET, Set[__DOWNLOAD_API_RET]]:
    """
    下载一个本子（album），包含其所有的章节（photo）

    当jm_album_id不是str或int时，视为批量下载，相当于调用 download_batch(download_album, jm_album_id, option, downloader, control=control)

    :param jm_album_id: 本子的禁漫车号
    :param option: 下载选项
    :param downloader: 下载器类
    :param check_exception: 仅当 jm_album_id 是单个 ID 时生效。为 True 时检查 downloader 中的部分下载失败，
                            并上抛 PartialDownloadFailedException。多 ID 调用会转交 download_batch，此参数不生效；
                            请检查 BatchResult.failed，或自行封装 download_batch 实现所需的批量异常策略。
    :param extra: 下载特性（Feature），下载时动态挂载的附加行为上下文。会自动根据上下文（如 album/photo 来源）自适应参数行为。支持单个 Feature、FeatureChain、或列表
    :param control: 可选的取消控制器 (DownloadControl)，用于外部协作式中断下载任务
    :return: 对于的本子实体类，下载器（如果是上述的批量情况，返回值为download_batch的返回值）
    """

    if not isinstance(jm_album_id, (str, int)):
        return download_batch(
            download_album,
            jm_album_id,
            option,
            downloader,
            download_type='album',
            extra=extra,
            control=control,
        )

    return _execute_sync_download(
        lambda dler: dler.download_album(jm_album_id),
        'album',
        jm_album_id,
        option,
        downloader,
        extra,
        check_exception,
        control=control,
    )

download_photo(jm_photo_id, option=None, downloader=None, *, check_exception=True, extra=None, control=None)

下载一个章节（photo），参数同 download_album。

check_exception 仅当 jm_photo_id 是单个 ID 时生效。多 ID 场景请检查 BatchResult.failed，或自行封装 download_batch 处理批量异常。

Source code in src/jmcomic/api.py

def download_photo(jm_photo_id,
                   option=None,
                   downloader=None,
                   *,
                   check_exception=True,
                   extra=None,
                   control=None,
                   ):
    """
    下载一个章节（photo），参数同 download_album。

    check_exception 仅当 jm_photo_id 是单个 ID 时生效。多 ID 场景请检查
    BatchResult.failed，或自行封装 download_batch 处理批量异常。
    """
    if not isinstance(jm_photo_id, (str, int)):
        return download_batch(
            download_photo,
            jm_photo_id,
            option,
            downloader,
            download_type='photo',
            extra=extra,
            control=control,
        )

    return _execute_sync_download(
        lambda dler: dler.download_photo(jm_photo_id),
        'photo',
        jm_photo_id,
        option,
        downloader,
        extra,
        check_exception,
        control=control,
    )

create_option_by_env(env_name='JM_OPTION_PATH')

Source code in src/jmcomic/api.py

def create_option_by_env(env_name='JM_OPTION_PATH'):
    from .cli import get_env

    filepath = get_env(env_name, None)
    ExceptionTool.require_true(filepath is not None,
                               f'未配置环境变量: {env_name}，请配置为option的文件路径')
    return create_option_by_file(filepath)

create_option_by_file(filepath)

Source code in src/jmcomic/api.py

def create_option_by_file(filepath):
    return JmModuleConfig.option_class().from_file(filepath)

create_option_by_str(text, mode=None)

Source code in src/jmcomic/api.py

def create_option_by_str(text: str, mode=None):
    if mode is None:
        mode = PackerUtil.mode_yml
    data = PackerUtil.unpack_by_str(text, mode)[0]
    return JmModuleConfig.option_class().construct(data)

download_album_async(jm_album_id, option=None, downloader=None, *, check_exception=True, extra=None, control=None) async

异步下载一个本子及其全部章节。

当 jm_album_id 是可迭代对象时使用批量 API。单个 ID 返回 DownloadResult；其中 Downloader 的网络 client 已关闭，阻塞工作也已 收敛。通过 jm_task_context(runtime=JmAsyncRuntime(...)) 提供的 decode 执行器由调用方负责关闭。

check_exception 只对单个 ID 生效。批量场景应检查 BatchResult.failed，或按需封装 download_batch_async。

Source code in src/jmcomic/api.py

async def download_album_async(jm_album_id,
                               option=None,
                               downloader=None,
                               *,
                               check_exception=True,
                               extra=None,
                               control=None,
                               ):
    """异步下载一个本子及其全部章节。

    当 ``jm_album_id`` 是可迭代对象时使用批量 API。单个 ID 返回
    ``DownloadResult``；其中 Downloader 的网络 client 已关闭，阻塞工作也已
    收敛。通过 ``jm_task_context(runtime=JmAsyncRuntime(...))`` 提供的 decode
    执行器由调用方负责关闭。

    ``check_exception`` 只对单个 ID 生效。批量场景应检查
    ``BatchResult.failed``，或按需封装 ``download_batch_async``。
    """
    if not isinstance(jm_album_id, (str, int)):
        return await download_batch_async(download_album_async,
                                          jm_album_id,
                                          option,
                                          downloader,
                                          download_type='album',
                                          extra=extra,
                                          control=control,
                                          )

    return await _execute_async_download(
        lambda dler: dler.download_album(jm_album_id),
        'album',
        jm_album_id,
        option,
        downloader,
        extra,
        check_exception,
        control=control,
    )

download_photo_async(jm_photo_id, option=None, downloader=None, *, check_exception=True, extra=None, control=None) async

异步下载一个章节。

返回的 Downloader 已关闭网络 client，阻塞工作也已收敛。通过 jm_task_context(runtime=JmAsyncRuntime(...)) 提供的 decode 执行器由调用方负责关闭。

check_exception 只对单个 ID 生效。批量场景应检查 BatchResult.failed，或按需封装 download_batch_async。

Source code in src/jmcomic/api.py

async def download_photo_async(jm_photo_id,
                               option=None,
                               downloader=None,
                               *,
                               check_exception=True,
                               extra=None,
                               control=None,
                               ):
    """异步下载一个章节。

    返回的 Downloader 已关闭网络 client，阻塞工作也已收敛。通过
    ``jm_task_context(runtime=JmAsyncRuntime(...))`` 提供的 decode 执行器由调用方负责关闭。

    ``check_exception`` 只对单个 ID 生效。批量场景应检查
    ``BatchResult.failed``，或按需封装 ``download_batch_async``。
    """
    if not isinstance(jm_photo_id, (str, int)):
        return await download_batch_async(download_photo_async,
                                          jm_photo_id,
                                          option,
                                          downloader,
                                          download_type='photo',
                                          extra=extra,
                                          control=control,
                                          )

    return await _execute_async_download(
        lambda dler: dler.download_photo(jm_photo_id),
        'photo',
        jm_photo_id,
        option,
        downloader,
        extra,
        check_exception,
        control=control,
    )

download_batch_async(download_api, jm_id_iter, option=None, downloader=None, download_type='batch', control=None, **kwargs) async

异步批量下载 album / photo。 - 容错机制：单个 album/photo 失败不会中止整批，也不会丢失其它已完成结果。 - 返回 BatchResult，保持 FIFO 顺序，支持按车号检索，失败项收集在 result.failed 中。

Parameters:

Name	Type	Description	Default
control		可选的取消控制器 (DownloadControl)，用于外部协作式中断下载任务	None

Source code in src/jmcomic/api.py

async def download_batch_async(
        download_api,
        jm_id_iter,
        option=None,
        downloader=None,
        download_type='batch',
        control=None,
        **kwargs,
) -> BatchResult:
    """
    异步批量下载 album / photo。
    - 容错机制：单个 album/photo 失败不会中止整批，也不会丢失其它已完成结果。
    - 返回 BatchResult，保持 FIFO 顺序，支持按车号检索，失败项收集在 result.failed 中。
    :param control: 可选的取消控制器 (DownloadControl)，用于外部协作式中断下载任务
    """
    option = _ensure_option(option)
    jm_ids = list(dict.fromkeys(JmcomicText.parse_to_jm_id(jmid) for jmid in jm_id_iter))
    result = BatchResult()
    if len(jm_ids) == 0:
        return result

    async def _download_one(jmid):
        with jm_task_context(download_type=download_type, jm_id=str(jmid)):
            return await download_api(jmid, option, downloader, **kwargs)

    runtime = JTC.get_runtime()
    if runtime is not None and not isinstance(runtime, JmAsyncRuntime):
        raise TypeError('async batch API requires JmAsyncRuntime')

    with jm_task_context(option=option, control=control):
        tasks = [asyncio.create_task(_download_one(jmid)) for jmid in jm_ids]
        # 等待所有异步任务执行完成
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # 检查是否有任务被取消
        cancellation_error = None
        for item in results:
            if isinstance(item, DownloadCancelledException):
                cancellation_error = item
                break

        # 若发生取消，收集已完成成果与未完成 ID 后抛出，不当成下载失败记录进 failed
        if cancellation_error is not None:
            succeeded_ids = {
                str(jmid)
                for jmid, item in zip(jm_ids, results)
                if not isinstance(item, BaseException) and item is not None
            }
            cancellation_error.context[ExceptionTool.CONTEXT_KEY_COMPLETED_RESULTS] = [
                item for item in results
                if not isinstance(item, BaseException) and item is not None
            ]
            cancellation_error.context[ExceptionTool.CONTEXT_KEY_UNFINISHED_IDS] = [
                str(jmid) for jmid in jm_ids
                if str(jmid) not in succeeded_ids
            ]
            raise cancellation_error

        # 外部协程取消信号
        for item in results:
            if isinstance(item, asyncio.CancelledError):
                raise item

        # 记录失败项并收集成功结果
        for jmid, item in zip(jm_ids, results):
            if isinstance(item, BaseException):
                with jm_task_context(download_type=download_type, jm_id=str(jmid)):
                    jm_log(
                        'async.batch.failed',
                        f'批量下载失败: [{jmid}], 异常: [{item}]',
                        item,
                    )
                result.failed[str(jmid)] = item
            else:
                result.add(item)

        return result

异步下载器 —— 对齐 sync JmDownloader

设计原则： - 继承 JmDownloader，复用回调体系（before_album/after_album 等）和 plugin 调用 - 下载 IO（asyncio）与 CPU 解密（ThreadPoolExecutor）流水线化 - 通过 asyncio.Semaphore 控制并发

JmAsyncDownloader

Bases: BaseDownloader

全异步流水线下载器。

核心设计： - 下载 IO 与 CPU 解密（ThreadPoolExecutor）完全流水线化 - 通过 asyncio.Semaphore 实现并发控制 - 继承 JmDownloader 的回调体系和 Plugin 调用

Methods:

Name	Description
download_album	对齐 sync JmDownloader.download_album
download_by_album_detail	异步下载整个本子。
download_by_photo_detail	异步下载一个章节的所有图片。
download_photo	对齐 sync JmDownloader.download_photo
shutdown	关闭解密线程池

Source code in src/jmcomic/jm_async_downloader.py

class JmAsyncDownloader(BaseDownloader):
    """
    全异步流水线下载器。

    核心设计：
    - 下载 IO 与 CPU 解密（ThreadPoolExecutor）完全流水线化
    - 通过 asyncio.Semaphore 实现并发控制
    - 继承 JmDownloader 的回调体系和 Plugin 调用
    """

    def __init__(self,
                 option: JmOption,
                 image_concurrency: int | None = None,
                 photo_concurrency: int | None = None,
                 decode_worker: int | None = None,
                 ) -> None:
        super().__init__(option)
        # 提取图片并发配置（使用 is None 判断，避免 0 被 or 静默替换为默认值）
        image_concurrency = int(image_concurrency if image_concurrency is not None else option.download.threading.image)
        if image_concurrency <= 0:
            raise ValueError(f"image_concurrency must be > 0, got {image_concurrency}")

        photo_concurrency = int(photo_concurrency if photo_concurrency is not None else option.download.threading.photo)
        if photo_concurrency <= 0:
            raise ValueError(f"photo_concurrency must be > 0, got {photo_concurrency}")

        self._image_concurrency = image_concurrency
        self._image_semaphore = asyncio.Semaphore(image_concurrency)
        self._photo_semaphore = asyncio.Semaphore(photo_concurrency)

        if decode_worker is None:
            # 对齐 ThreadPoolExecutor 的默认容量，并由调用点明确交给 Runtime。
            decode_worker = min(32, (os.cpu_count() or 1) + 4)
        elif (
                isinstance(decode_worker, bool)
                or not isinstance(decode_worker, int)
                or decode_worker <= 0
        ):
            raise ValueError(
                f'decode_worker must be a positive integer, got {decode_worker!r}'
            )
        self._decode_worker = decode_worker
        self._decode_pool: ThreadPoolExecutor | None = None

    @classmethod
    def use(cls, *args, **kwargs):
        before_class = JmModuleConfig.async_downloader_class()
        JmModuleConfig.CLASS_ASYNC_DOWNLOADER = cls
        after_class = JmModuleConfig.async_downloader_class()
        jm_log(
            'async_downloader.use',
            f'替换 Async Downloader class: '
            f'[{before_class.__module__}.{before_class.__qualname__}] -> '
            f'[{after_class.__module__}.{after_class.__qualname__}]'
        )

    # ======================================================================
    # 核心下载流程 — 对齐 sync JmDownloader
    # ======================================================================

    async def _run_in_decode_pool(self, func, *args):
        runtime = JTC.get_runtime()
        if isinstance(runtime, JmAsyncRuntime):
            executor = runtime.executor(JmRuntime.LEVEL_DECODE)
        else:
            if self._decode_pool is None:
                self._decode_pool = ThreadPoolExecutor(
                    max_workers=self._decode_worker,
                    thread_name_prefix='jm-async-decode',
                )
            executor = self._decode_pool

        worker = bind_jm_task_context(func)
        future = executor.submit(worker, *args)
        waiter = asyncio.wrap_future(future)
        try:
            return await asyncio.shield(waiter)
        except asyncio.CancelledError:
            # 线程任务无法可靠中断，协程取消后仍需等待写盘等操作收尾。
            try:
                await waiter
            except (asyncio.CancelledError, DownloadCancelledException):
                pass
            except BaseException as drain_error:
                jm_log('dler.cancel.drain.exception', drain_error)
            raise

    @record_download_duration('album_started_at')
    async def download_album(self, album_id) -> JmAlbumDetail:
        """对齐 sync JmDownloader.download_album"""
        self.raise_if_cancelled()
        album = await self.client.get_album_detail(album_id)
        self.raise_if_cancelled()
        self.begin_manifest(album)
        try:
            await self.download_by_album_detail(album)
        finally:
            self.finish_manifest(album)
        return album

    @record_download_duration('album_started_at')
    async def download_by_album_detail(self, album: JmAlbumDetail):
        """
        异步下载整个本子。
        对齐 sync JmDownloader.download_by_album_detail 的回调链路。
        """
        album.save_path = self.option.dir_rule.decide_album_root_dir(album)
        await self.before_album(album)
        if album.skip:
            return

        photos = list(self.do_filter(album))

        # 即使过滤后 photos 为空，也要执行 after_album（对齐 sync：execute_on_condition
        # 在 count_real==0 时提前返回，但调用方仍会走到 after_album，触发其插件与 Feature）。
        if photos:
            # photo 级并发由 _photo_semaphore 控制（默认 3），包裹整段 photo 下载（见 download_by_photo_detail）。
            photo_tasks = [self._safe_download_photo(photo) for photo in photos]
            # 对齐 sync 最大容错：普通异常由子任务记录，取消异常等待全部结束后再检查
            await asyncio.gather(*photo_tasks, return_exceptions=True)
            self.raise_if_cancelled()

        await self.after_album(album)

    async def _safe_download_photo(self, photo: JmPhotoDetail):
        """包装 download_by_photo_detail，对齐 sync @catch_exception 的异常记录"""
        try:
            await self.download_by_photo_detail(photo)
        except DownloadCancelledException:
            raise
        except Exception as e:
            jm_log('photo.failed', f'章节下载失败: [{photo.id}], 异常: [{e}]', e)
            self.download_failed_photo.append((photo, e))
            self.raise_if_cancelled()

    @record_download_duration('photo_started_at')
    async def download_photo(self, photo_id) -> JmPhotoDetail:
        """对齐 sync JmDownloader.download_photo"""
        self.raise_if_cancelled()
        photo = await self.client.get_photo_detail(photo_id)
        self.raise_if_cancelled()
        self.begin_manifest(photo)
        try:
            await self.download_by_photo_detail(photo)
        finally:
            self.finish_manifest(photo)
        return photo

    @record_download_duration('photo_started_at')
    async def download_by_photo_detail(self, photo: JmPhotoDetail):
        """
        异步下载一个章节的所有图片。
        对齐 sync JmDownloader.download_by_photo_detail 的回调链路。
        """
        self.raise_if_cancelled()
        photo.save_path = self.option.decide_image_save_dir(photo)
        # _photo_semaphore 包裹整段 photo 下载（check_photo + 全部图片），
        # 真正限制「同时下载的章节数」（对齐 sync：每个 photo 占用 photo 线程池一个槽位）。
        # 章节内图片再由共享的 _image_semaphore 二级限流。
        async with self._photo_semaphore:
            # 排队期间可能被取消；执行 I/O 前必须再次检查。
            self.raise_if_cancelled()
            await self.client.check_photo(photo)

            await self.before_photo(photo)
            if photo.skip:
                return

            images = self.do_filter(photo)
            image_list = list(images) if images is not None else []

            # 即使过滤后图片为空，也要执行 after_photo（对齐 sync，触发 after_photo 插件与 Feature）。
            if image_list:
                # 直接创建所有下载协程，由 _image_semaphore 实现滑动窗口流控
                download_tasks = [
                    self._safe_download_image(image)
                    for image in image_list
                ]
                # 对齐 sync 最大容错：普通异常由子任务记录，取消异常等待全部结束后再检查
                await asyncio.gather(*download_tasks, return_exceptions=True)
                self.raise_if_cancelled()

            await self.after_photo(photo)

    async def _safe_download_image(self, image: JmImageDetail):
        """
        包装 _download_single_image，对齐 sync @catch_exception 的异常记录。
        异常由此统一捕获和记录，不再在内部重复 try/except。
        """
        try:
            await self._download_single_image(image)
        except DownloadCancelledException:
            raise
        except Exception as e:
            jm_log('image.failed', f'图片下载失败: [{image.download_url}], 异常: [{e}]', e)
            self.download_failed_image.append((image, e))
            self.raise_if_cancelled()

    @record_download_duration('image_started_at')
    async def _download_single_image(self, image: JmImageDetail):
        """
        下载并解密单张图片的完整流程。
        对齐 sync JmDownloader.download_by_image_detail 的逻辑。
        """
        img_save_path = self.option.decide_image_filepath(image)
        image.save_path = img_save_path
        image.exists = os.path.exists(img_save_path)
        image.cache = self.option.decide_download_cache(image)

        await self.before_image(image, img_save_path)
        if image.skip:
            return

        if image.cache and image.exists:
            await self.after_image(image, img_save_path)
            return

        decode_image = self.option.decide_download_image_decode(image)

        # 异步下载图片（受 image semaphore 限流，并将解密写盘过程也锁入信号量范围内，防大字节积压）
        async with self._image_semaphore:
            # 排队期间可能被取消；执行 I/O 前必须再次检查。
            self.raise_if_cancelled()
            img_resp = await self.client.get_jm_image(image.download_url)
            img_bytes = img_resp.content

            # 提交到线程池解密并保存
            if decode_image and image.scramble_id:
                await self._run_in_decode_pool(
                    self._decode_and_save,
                    img_bytes,
                    int(image.scramble_id),
                    int(image.aid),
                    image.img_file_name,
                    img_save_path,
                )
            else:
                # 不解密保存。对齐 sync transfer_to(decode_image=False)：
                # 当目标后缀与原图后缀不一致时，需经 PIL 做格式转换。
                # 与 sync 一致：比较后缀前先剥离 url 的 ?query 部分，避免 query 干扰后缀判定。
                from common import suffix_not_equal
                img_url = image.download_url
                qi = img_url.find('?')
                if qi != -1:
                    img_url = img_url[:qi]
                need_convert = suffix_not_equal(img_url, img_save_path)
                await self._run_in_decode_pool(
                    self._save_raw,
                    img_bytes,
                    img_save_path,
                    need_convert,
                )

        await self.after_image(image, img_save_path)

    # ======================================================================
    # 磁盘写入（在线程池中执行）
    # ======================================================================

    @staticmethod
    def _decode_and_save(image_bytes, scramble_id, aid, img_file_name, save_path):
        """
        解密图片并保存到磁盘（在线程池中执行）。
        与 Sync 版对齐的直接写文件方式。
        保存目录已由 decide_image_filepath(ensure_exists=True) 创建，此处不重复 makedirs。
        """
        num = JmImageTool.get_num(scramble_id, aid, img_file_name)
        img_src = JmImageTool.open_image(image_bytes)
        JmImageTool.decode_and_save(num, img_src, save_path)

    @staticmethod
    def _save_raw(image_bytes, save_path, need_convert=False):
        """
        不解密保存。
        - need_convert=False：直接写原始字节（如 .gif，或后缀与原图一致时）。
        - need_convert=True：经 PIL 按 save_path 后缀做格式转换（对齐 sync save_resp_img）。
        保存目录已由 decide_image_filepath(ensure_exists=True) 创建，此处不重复 makedirs。
        """
        if need_convert:
            JmImageTool.save_image(JmImageTool.open_image(image_bytes), save_path)
        else:
            with open(save_path, 'wb') as f:
                f.write(image_bytes)

    # ======================================================================
    # 生命周期
    # ======================================================================

    async def _dispatch_plugin(self, group: str, **kwargs):
        plugins = getattr(self.option, 'plugins', None)
        if plugins is not None and not plugins.get(group):
            return
        await asyncio.to_thread(self.option.call_all_plugin, group, **kwargs)

    async def _dispatch_feature(self, when: str, **kwargs):
        if len(self._feature_list) == 0:
            return
        await asyncio.to_thread(self._invoke_features_for, when, **kwargs)

    async def before_album(self, album: JmAlbumDetail):
        self.raise_if_cancelled()
        DownloadCallback.before_album(self, album)
        self.download_success_dict.setdefault(album, {})
        await self._dispatch_plugin('before_album', album=album, downloader=self)
        self.raise_if_cancelled()

    async def after_album(self, album: JmAlbumDetail):
        self.raise_if_cancelled()
        DownloadCallback.after_album(self, album)
        await self._dispatch_plugin('after_album', album=album, downloader=self)
        self.raise_if_cancelled()
        await self._dispatch_feature('after_album', album=album, downloader=self)

    async def before_photo(self, photo: JmPhotoDetail):
        self.raise_if_cancelled()
        DownloadCallback.before_photo(self, photo)
        self.download_success_dict.setdefault(photo.from_album, {})
        self.download_success_dict[photo.from_album].setdefault(photo, [])
        await self._dispatch_plugin('before_photo', photo=photo, downloader=self)
        self.raise_if_cancelled()

    async def after_photo(self, photo: JmPhotoDetail):
        self.raise_if_cancelled()
        DownloadCallback.after_photo(self, photo)
        await self._dispatch_plugin('after_photo', photo=photo, downloader=self)
        self.raise_if_cancelled()
        await self._dispatch_feature('after_photo', photo=photo, downloader=self)

    async def before_image(self, image: JmImageDetail, img_save_path: str):
        self.raise_if_cancelled()
        DownloadCallback.before_image(self, image, img_save_path)
        await self._dispatch_plugin('before_image', image=image, downloader=self)
        self.raise_if_cancelled()

    async def after_image(self, image: JmImageDetail, img_save_path: str):
        # 1. 记录下载成功图片
        photo = image.from_photo
        album = photo.from_album
        self.download_success_dict[album][photo].append((image.save_path, image))

        # 2. 日志回调
        DownloadCallback.after_image(self, image, img_save_path)

        # 3. 检查取消
        self.raise_if_cancelled()

        # 4. 插件回调
        await self._dispatch_plugin('after_image', image=image, downloader=self)

    def shutdown(self):
        """关闭解密线程池"""
        if self._decode_pool is not None:
            self._decode_pool.shutdown(wait=False)
            self._decode_pool = None

    async def __aenter__(self):
        self.client = self.option.new_jm_async_client(
            max_clients=self._image_concurrency
        )
        try:
            await self.client.setup()
        except BaseException:
            try:
                if self.client is not None:
                    await self.client.close()
            except BaseException as cleanup_error:
                jm_log('dler.cleanup.exception',
                       f'初始化失败后的资源清理也发生异常: {cleanup_error}', cleanup_error)
            finally:
                self.client = None
                self.shutdown()
            raise
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        try:
            if self.client is not None:
                await self.client.close()
        finally:
            self.shutdown()

        if exc_type is not None:
            jm_log('dler.exception',
                   f'{self.__class__.__name__} Exit with exception: {exc_type, str(exc_val)}')

download_album(album_id) async

对齐 sync JmDownloader.download_album

Source code in src/jmcomic/jm_async_downloader.py

@record_download_duration('album_started_at')
async def download_album(self, album_id) -> JmAlbumDetail:
    """对齐 sync JmDownloader.download_album"""
    self.raise_if_cancelled()
    album = await self.client.get_album_detail(album_id)
    self.raise_if_cancelled()
    self.begin_manifest(album)
    try:
        await self.download_by_album_detail(album)
    finally:
        self.finish_manifest(album)
    return album

download_by_album_detail(album) async

异步下载整个本子。 对齐 sync JmDownloader.download_by_album_detail 的回调链路。

Source code in src/jmcomic/jm_async_downloader.py

@record_download_duration('album_started_at')
async def download_by_album_detail(self, album: JmAlbumDetail):
    """
    异步下载整个本子。
    对齐 sync JmDownloader.download_by_album_detail 的回调链路。
    """
    album.save_path = self.option.dir_rule.decide_album_root_dir(album)
    await self.before_album(album)
    if album.skip:
        return

    photos = list(self.do_filter(album))

    # 即使过滤后 photos 为空，也要执行 after_album（对齐 sync：execute_on_condition
    # 在 count_real==0 时提前返回，但调用方仍会走到 after_album，触发其插件与 Feature）。
    if photos:
        # photo 级并发由 _photo_semaphore 控制（默认 3），包裹整段 photo 下载（见 download_by_photo_detail）。
        photo_tasks = [self._safe_download_photo(photo) for photo in photos]
        # 对齐 sync 最大容错：普通异常由子任务记录，取消异常等待全部结束后再检查
        await asyncio.gather(*photo_tasks, return_exceptions=True)
        self.raise_if_cancelled()

    await self.after_album(album)

download_by_photo_detail(photo) async

异步下载一个章节的所有图片。 对齐 sync JmDownloader.download_by_photo_detail 的回调链路。

Source code in src/jmcomic/jm_async_downloader.py

@record_download_duration('photo_started_at')
async def download_by_photo_detail(self, photo: JmPhotoDetail):
    """
    异步下载一个章节的所有图片。
    对齐 sync JmDownloader.download_by_photo_detail 的回调链路。
    """
    self.raise_if_cancelled()
    photo.save_path = self.option.decide_image_save_dir(photo)
    # _photo_semaphore 包裹整段 photo 下载（check_photo + 全部图片），
    # 真正限制「同时下载的章节数」（对齐 sync：每个 photo 占用 photo 线程池一个槽位）。
    # 章节内图片再由共享的 _image_semaphore 二级限流。
    async with self._photo_semaphore:
        # 排队期间可能被取消；执行 I/O 前必须再次检查。
        self.raise_if_cancelled()
        await self.client.check_photo(photo)

        await self.before_photo(photo)
        if photo.skip:
            return

        images = self.do_filter(photo)
        image_list = list(images) if images is not None else []

        # 即使过滤后图片为空，也要执行 after_photo（对齐 sync，触发 after_photo 插件与 Feature）。
        if image_list:
            # 直接创建所有下载协程，由 _image_semaphore 实现滑动窗口流控
            download_tasks = [
                self._safe_download_image(image)
                for image in image_list
            ]
            # 对齐 sync 最大容错：普通异常由子任务记录，取消异常等待全部结束后再检查
            await asyncio.gather(*download_tasks, return_exceptions=True)
            self.raise_if_cancelled()

        await self.after_photo(photo)

download_photo(photo_id) async

对齐 sync JmDownloader.download_photo

Source code in src/jmcomic/jm_async_downloader.py

@record_download_duration('photo_started_at')
async def download_photo(self, photo_id) -> JmPhotoDetail:
    """对齐 sync JmDownloader.download_photo"""
    self.raise_if_cancelled()
    photo = await self.client.get_photo_detail(photo_id)
    self.raise_if_cancelled()
    self.begin_manifest(photo)
    try:
        await self.download_by_photo_detail(photo)
    finally:
        self.finish_manifest(photo)
    return photo

shutdown()

关闭解密线程池

Source code in src/jmcomic/jm_async_downloader.py

def shutdown(self):
    """关闭解密线程池"""
    if self._decode_pool is not None:
        self._decode_pool.shutdown(wait=False)
        self._decode_pool = None

BaseDownloader

Bases: DownloadCallback

不含 I/O 调度的公共基类，负责回调、钩子、Features 注册等无 I/O 通用逻辑。

Methods:

Name	Description
add_features	为当前顶层下载注册 Feature。
do_filter	该方法可用于过滤本子/章节，默认不会做过滤。
raise_if_cancelled	在当前下载作用域已取消时抛出异常，子类可重写该检查点。

Source code in src/jmcomic/jm_downloader.py

class BaseDownloader(DownloadCallback):
    """
    不含 I/O 调度的公共基类，负责回调、钩子、Features 注册等无 I/O 通用逻辑。
    """

    def __init__(self, option: JmOption):
        self.option = option
        self.client = None
        # 下载成功的记录dict
        self.download_success_dict: Dict[JmAlbumDetail, Dict[JmPhotoDetail, List[Tuple[str, JmImageDetail]]]] = {}
        # 下载失败的记录list
        self.download_failed_image: List[Tuple[JmImageDetail, BaseException]] = []
        self.download_failed_photo: List[Tuple[JmPhotoDetail, BaseException]] = []
        # 每次顶层下载对应的聚合清单
        self.manifest_dict: Dict[DetailEntity, DownloadManifest] = {}
        # 当前顶层下载注册的 Feature 列表
        self._feature_list: List = []

    def do_filter(self, detail: DetailEntity):
        """
        该方法可用于过滤本子/章节，默认不会做过滤。
        例如:
        只想下载 本子的最新一章，返回 [album[-1]]
        只想下载 章节的前10张图片，返回 [photo[:10]]

        :param detail: 可能是本子或者章节，需要自行使用 isinstance / detail.is_xxx 判断
        :returns: 只想要下载的 本子的章节 或 章节的图片
        """
        return detail

    @property
    def all_success(self) -> bool:
        """
        是否成功下载了全部图片

        该属性需要等到downloader的全部download_xxx方法完成后才有意义。

        注意！如果使用了filter机制，例如通过filter只下载3张图片，那么all_success也会为False
        """
        if self.has_download_failures:
            return False

        for album, photo_dict in self.download_success_dict.items():
            if len(album) != len(photo_dict):
                return False

            for photo, image_list in photo_dict.items():
                if len(photo) != len(image_list):
                    return False

        return True

    @property
    def has_download_failures(self):
        return len(self.download_failed_image) != 0 or len(self.download_failed_photo) != 0

    @property
    def is_cancelled(self) -> bool:
        return JTC.is_cancelled()

    @classmethod
    def raise_if_cancelled(cls) -> None:
        """在当前下载作用域已取消时抛出异常，子类可重写该检查点。"""
        JTC.raise_if_cancelled()

    # 下面是回调方法

    def before_album(self, album: JmAlbumDetail):
        self.raise_if_cancelled()
        super().before_album(album)
        self.download_success_dict.setdefault(album, {})
        self.option.call_all_plugin(
            'before_album',
            album=album,
            downloader=self,
        )
        self.raise_if_cancelled()

    def after_album(self, album: JmAlbumDetail):
        self.raise_if_cancelled()
        super().after_album(album)
        self.option.call_all_plugin(
            'after_album',
            album=album,
            downloader=self,
        )
        self.raise_if_cancelled()
        # 触发匹配 after_album 的 Feature
        self._invoke_features_for('after_album', album=album, downloader=self)

    def before_photo(self, photo: JmPhotoDetail):
        self.raise_if_cancelled()
        super().before_photo(photo)
        self.download_success_dict.setdefault(photo.from_album, {})
        self.download_success_dict[photo.from_album].setdefault(photo, [])
        self.option.call_all_plugin(
            'before_photo',
            photo=photo,
            downloader=self,
        )
        self.raise_if_cancelled()

    def after_photo(self, photo: JmPhotoDetail):
        self.raise_if_cancelled()
        super().after_photo(photo)
        self.option.call_all_plugin(
            'after_photo',
            photo=photo,
            downloader=self,
        )
        self.raise_if_cancelled()
        # 触发匹配 after_photo 的 Feature
        self._invoke_features_for('after_photo', photo=photo, downloader=self)

    def before_image(self, image: JmImageDetail, img_save_path):
        self.raise_if_cancelled()
        super().before_image(image, img_save_path)
        self.option.call_all_plugin(
            'before_image',
            image=image,
            downloader=self,
        )
        self.raise_if_cancelled()

    def after_image(self, image: JmImageDetail, img_save_path):
        # 1. 记录下载成功图片
        photo = image.from_photo
        album = photo.from_album
        self.download_success_dict[album][photo].append((image.save_path, image))

        # 2. 日志回调
        super().after_image(image, img_save_path)

        # 3. 检查取消
        self.raise_if_cancelled()

        # 4. 插件回调
        self.option.call_all_plugin(
            'after_image',
            image=image,
            downloader=self,
        )

    def begin_manifest(self, detail: DetailEntity) -> DownloadManifest:
        manifest = DownloadManifest()
        self.manifest_dict[detail] = manifest
        return manifest

    def resolve_manifest_detail(self, detail: DetailEntity) -> Optional[DetailEntity]:
        if detail in self.manifest_dict:
            return detail

        if detail.is_photo() and detail.from_album in self.manifest_dict:
            return detail.from_album

        return None

    def record_export_filepath(self, detail: DetailEntity, filepath: str) -> None:
        manifest_detail = self.resolve_manifest_detail(detail)
        if manifest_detail is None:
            from .jm_toolkit import ExceptionTool
            ExceptionTool.raises(f'当前实体没有活动的下载清单: {detail}')

        suffix = os.path.splitext(filepath)[1].lower().lstrip('.')
        if suffix == '':
            return

        manifest = self.manifest_dict[manifest_detail]
        manifest.export_filepath_dict.setdefault(suffix, []).append(filepath)

    def finish_manifest(self, detail: DetailEntity) -> DownloadManifest:
        manifest = self.manifest_dict[detail]
        if detail.is_album():
            success_dict = self.download_success_dict.get(detail, {})
            success_groups = [
                success_list
                for _, success_list in sorted(
                    success_dict.items(),
                    key=lambda item: item[0].index,
                )
            ]
        else:
            success_groups = [
                self.download_success_dict.get(detail.from_album, {}).get(detail, [])
            ]

        manifest.image_filepath_list = [
            image.save_path
            for success_list in success_groups
            for _, image in sorted(success_list, key=lambda item: item[1].index)
        ]
        return manifest

    @staticmethod
    def _require_feature_context() -> str:
        from .jm_toolkit import ExceptionTool

        download_type = JTC.get_context_value('download_type')
        ExceptionTool.require_true(
            download_type in ('album', 'photo'),
            'Feature 注册与执行必须位于下载任务上下文中，请使用 '
            "jm_task_context(download_type='album') 或 "
            "jm_task_context(download_type='photo') 进行包裹",
        )
        return download_type

    def add_features(self, features):
        """
        为当前顶层下载注册 Feature。

        :param features: Feature / FeatureChain / list / None
        """
        if features is None:
            return

        from .jm_feature import FeatureChain, Feature
        from .jm_toolkit import ExceptionTool
        self._require_feature_context()

        if isinstance(features, list):
            for f in features:
                self.add_features(f)
        elif isinstance(features, FeatureChain):
            for f in features.to_list():
                self._feature_list.append(f)
        elif isinstance(features, Feature):
            self._feature_list.append(features)
        else:
            ExceptionTool.raises(f'不支持的 extra 类型: {type(features)}，请传入 Feature / FeatureChain / list / None')

    def _invoke_features_for(self, when: str, **kwargs):
        """
        在指定钩子(when)中触发匹配的 Feature。

        :param when: 当前钩子名，如 'after_album', 'after_photo'
        :param kwargs: album, photo, downloader 等上下文
        """
        if len(self._feature_list) == 0:
            return

        download_type = self._require_feature_context()
        for feature in self._feature_list:
            self.raise_if_cancelled()
            if feature.should_invoke(when):
                try:
                    feature.invoke(self.option, when=when, **kwargs)
                except DownloadCancelledException:
                    raise
                except Exception as e:
                    jm_log('downloader.feature.exception',
                           f'Feature执行失败: [{feature}], 下载类型: [{download_type}], 异常: [{e}]',
                           e)
        self.raise_if_cancelled()

    def raise_if_has_exception(self):
        if not self.has_download_failures:
            return
        msg_ls = ['部分下载失败', '', '']

        if len(self.download_failed_photo) != 0:
            msg_ls[1] = f'共{len(self.download_failed_photo)}个章节下载失败: {self.download_failed_photo}'

        if len(self.download_failed_image) != 0:
            msg_ls[2] = f'共{len(self.download_failed_image)}个图片下载失败: {self.download_failed_image}'

        ExceptionTool.raises(
            '\n'.join(msg_ls),
            {ExceptionTool.CONTEXT_KEY_DOWNLOADER: self},
            PartialDownloadFailedException,
        )

all_success property

是否成功下载了全部图片

该属性需要等到downloader的全部download_xxx方法完成后才有意义。

注意！如果使用了filter机制，例如通过filter只下载3张图片，那么all_success也会为False

add_features(features)

为当前顶层下载注册 Feature。

Parameters:

Name	Type	Description	Default
features		Feature / FeatureChain / list / None	required

Source code in src/jmcomic/jm_downloader.py

def add_features(self, features):
    """
    为当前顶层下载注册 Feature。

    :param features: Feature / FeatureChain / list / None
    """
    if features is None:
        return

    from .jm_feature import FeatureChain, Feature
    from .jm_toolkit import ExceptionTool
    self._require_feature_context()

    if isinstance(features, list):
        for f in features:
            self.add_features(f)
    elif isinstance(features, FeatureChain):
        for f in features.to_list():
            self._feature_list.append(f)
    elif isinstance(features, Feature):
        self._feature_list.append(features)
    else:
        ExceptionTool.raises(f'不支持的 extra 类型: {type(features)}，请传入 Feature / FeatureChain / list / None')

do_filter(detail)

该方法可用于过滤本子/章节，默认不会做过滤。 例如: 只想下载 本子的最新一章，返回 [album[-1]] 只想下载 章节的前10张图片，返回 [photo[:10]]

Parameters:

Name	Type	Description	Default
detail	DetailEntity	可能是本子或者章节，需要自行使用 isinstance / detail.is_xxx 判断	required

Returns:

Type	Description
	只想要下载的 本子的章节 或 章节的图片

Source code in src/jmcomic/jm_downloader.py

def do_filter(self, detail: DetailEntity):
    """
    该方法可用于过滤本子/章节，默认不会做过滤。
    例如:
    只想下载 本子的最新一章，返回 [album[-1]]
    只想下载 章节的前10张图片，返回 [photo[:10]]

    :param detail: 可能是本子或者章节，需要自行使用 isinstance / detail.is_xxx 判断
    :returns: 只想要下载的 本子的章节 或 章节的图片
    """
    return detail

raise_if_cancelled() classmethod

在当前下载作用域已取消时抛出异常，子类可重写该检查点。

Source code in src/jmcomic/jm_downloader.py

@classmethod
def raise_if_cancelled(cls) -> None:
    """在当前下载作用域已取消时抛出异常，子类可重写该检查点。"""
    JTC.raise_if_cancelled()

JmDownloader

Bases: BaseDownloader

JmDownloader = BaseDownloader + 同步 I/O 调度逻辑

Methods:

Name	Description
create_client	创建该downloader使用的client。
execute_on_condition	使用当前 Runtime 调度本子或章节的下载。
use	让本类替换JmModuleConfig.CLASS_DOWNLOADER

Source code in src/jmcomic/jm_downloader.py

class JmDownloader(BaseDownloader):
    """
    JmDownloader = BaseDownloader + 同步 I/O 调度逻辑
    """

    def __init__(self, option: JmOption):
        super().__init__(option)
        self.client = self.create_client()

    def create_client(self):
        """
        创建该downloader使用的client。
        """
        return self.option.build_jm_client()

    @record_download_duration('album_started_at')
    def download_album(self, album_id):
        self.raise_if_cancelled()
        album = self.client.get_album_detail(album_id)
        self.raise_if_cancelled()
        self.begin_manifest(album)
        try:
            self.download_by_album_detail(album)
        finally:
            self.finish_manifest(album)
        return album

    @record_download_duration('album_started_at')
    def download_by_album_detail(self, album: JmAlbumDetail):
        album.save_path = self.option.dir_rule.decide_album_root_dir(album)
        self.before_album(album)
        if album.skip:
            return
        self.execute_on_condition(
            iter_objs=album,
            apply=self.download_by_photo_detail,
            count_batch=self.option.decide_photo_batch_count(album),
            level=JmRuntime.LEVEL_PHOTO,
        )
        self.after_album(album)

    @record_download_duration('photo_started_at')
    def download_photo(self, photo_id):
        self.raise_if_cancelled()
        photo = self.client.get_photo_detail(photo_id)
        self.raise_if_cancelled()
        self.begin_manifest(photo)
        try:
            self.download_by_photo_detail(photo)
        finally:
            self.finish_manifest(photo)
        return photo

    @catch_exception
    @record_download_duration('photo_started_at')
    def download_by_photo_detail(self, photo: JmPhotoDetail):
        photo.save_path = self.option.decide_image_save_dir(photo)
        self.client.check_photo(photo)
        self.before_photo(photo)
        if photo.skip:
            return
        self.execute_on_condition(
            iter_objs=photo,
            apply=self.download_by_image_detail,
            count_batch=self.option.decide_image_batch_count(photo),
            level=JmRuntime.LEVEL_IMAGE,
        )
        self.after_photo(photo)

    @catch_exception
    @record_download_duration('image_started_at')
    def download_by_image_detail(self, image: JmImageDetail):
        img_save_path = self.option.decide_image_filepath(image)
        image.save_path = img_save_path
        image.exists = file_exists(img_save_path)
        image.cache = self.option.decide_download_cache(image)

        self.before_image(image, img_save_path)
        if image.skip:
            return

        if image.cache and image.exists:
            self.after_image(image, img_save_path)
            return

        decode_image = self.option.decide_download_image_decode(image)
        self.client.download_by_image_detail(
            image,
            img_save_path,
            decode_image=decode_image,
        )

        self.after_image(image, img_save_path)

    def execute_on_condition(self, iter_objs, apply, count_batch, level=None):
        """使用当前 Runtime 调度本子或章节的下载。"""
        runtime = JTC.get_runtime()
        ExceptionTool.require_true(
            runtime is None or isinstance(runtime, JmSyncRuntime),
            'sync downloader requires JmSyncRuntime',
        )

        iter_objs = self.do_filter(iter_objs)
        if len(iter_objs) == 0:
            return

        ExceptionTool.require_true(
            not isinstance(count_batch, bool) and isinstance(count_batch, int) and count_batch > 0,
            f'local download limit must be > 0, got {count_batch!r}',
        )

        self.raise_if_cancelled()

        if runtime is not None and level is not None and runtime.has_executor(level):
            runtime.multi_thread_launcher(
                iter_objs=iter_objs,
                apply_each_obj_func=bind_jm_task_context(apply),
                level=level,
            )
        else:
            with JmTempRuntime(workers=count_batch) as temp_launcher:
                temp_launcher.multi_thread_launcher(
                    iter_objs=iter_objs,
                    apply_each_obj_func=bind_jm_task_context(apply),
                )

        # 批处理线程等待结束后，统一在此处检查取消状态
        self.raise_if_cancelled()

    # 下面是对with语法的支持

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            jm_log('dler.exception',
                   f'{self.__class__.__name__} Exit with exception: {exc_type, str(exc_val)}'
                   )

    @classmethod
    def use(cls, *args, **kwargs):
        """
        让本类替换JmModuleConfig.CLASS_DOWNLOADER
        """
        before_class = JmModuleConfig.downloader_class()
        JmModuleConfig.CLASS_DOWNLOADER = cls
        after_class = JmModuleConfig.downloader_class()
        jm_log(
            'downloader.use',
            f'替换 Downloader class: '
            f'[{before_class.__module__}.{before_class.__qualname__}] -> '
            f'[{after_class.__module__}.{after_class.__qualname__}]'
        )

create_client()

创建该downloader使用的client。

Source code in src/jmcomic/jm_downloader.py

def create_client(self):
    """
    创建该downloader使用的client。
    """
    return self.option.build_jm_client()

execute_on_condition(iter_objs, apply, count_batch, level=None)

使用当前 Runtime 调度本子或章节的下载。

Source code in src/jmcomic/jm_downloader.py

def execute_on_condition(self, iter_objs, apply, count_batch, level=None):
    """使用当前 Runtime 调度本子或章节的下载。"""
    runtime = JTC.get_runtime()
    ExceptionTool.require_true(
        runtime is None or isinstance(runtime, JmSyncRuntime),
        'sync downloader requires JmSyncRuntime',
    )

    iter_objs = self.do_filter(iter_objs)
    if len(iter_objs) == 0:
        return

    ExceptionTool.require_true(
        not isinstance(count_batch, bool) and isinstance(count_batch, int) and count_batch > 0,
        f'local download limit must be > 0, got {count_batch!r}',
    )

    self.raise_if_cancelled()

    if runtime is not None and level is not None and runtime.has_executor(level):
        runtime.multi_thread_launcher(
            iter_objs=iter_objs,
            apply_each_obj_func=bind_jm_task_context(apply),
            level=level,
        )
    else:
        with JmTempRuntime(workers=count_batch) as temp_launcher:
            temp_launcher.multi_thread_launcher(
                iter_objs=iter_objs,
                apply_each_obj_func=bind_jm_task_context(apply),
            )

    # 批处理线程等待结束后，统一在此处检查取消状态
    self.raise_if_cancelled()

use(*args, **kwargs) classmethod

让本类替换JmModuleConfig.CLASS_DOWNLOADER

Source code in src/jmcomic/jm_downloader.py

@classmethod
def use(cls, *args, **kwargs):
    """
    让本类替换JmModuleConfig.CLASS_DOWNLOADER
    """
    before_class = JmModuleConfig.downloader_class()
    JmModuleConfig.CLASS_DOWNLOADER = cls
    after_class = JmModuleConfig.downloader_class()
    jm_log(
        'downloader.use',
        f'替换 Downloader class: '
        f'[{before_class.__module__}.{before_class.__qualname__}] -> '
        f'[{after_class.__module__}.{after_class.__qualname__}]'
    )

JmRuntime

轻量 Runtime：管理 Executor 的配置、创建、调度和关闭。

【生命周期与关闭规则规范】： 1. with runtime 上下文管理： 离开 with 作用域（正常完成或异常退出）时，会自动调用 runtime.close()， 关闭由此 Runtime 自身创建并拥有的所有 Executor 线程池。 2. 用户外传 Executor： 若在构造时传入了用户自定义的 executor，Runtime 在 close() 时绝不会关闭它， 生命周期全权由调用方自行管理（严格遵循“谁创建、谁负责关闭”原则）。 3. 不使用 with（裸调用）： 绝不会自动关闭。必须由调用方显式调用 runtime.close()，或作为长效单例常驻复用。

Methods:

Name	Description
close	关闭 Runtime 自建的 Executor；调用方传入的 Executor 保持可用。
executor	返回指定层级的 Executor；未显式配置 workers 时按系统默认并发数创建。
has_executor	检查指定层级是否已由外部显式配置（workers 或 executor），或已初始化 Executor。
multi_thread_launcher	使用指定层级的 Executor 批量提交任务，并按需等待完成。

Source code in src/jmcomic/jm_runtime.py

class JmRuntime:
    """
    轻量 Runtime：管理 Executor 的配置、创建、调度和关闭。

    【生命周期与关闭规则规范】：
    1. with runtime 上下文管理：
       离开 with 作用域（正常完成或异常退出）时，会自动调用 runtime.close()，
       关闭由此 Runtime 自身创建并拥有的所有 Executor 线程池。
    2. 用户外传 Executor：
       若在构造时传入了用户自定义的 executor，Runtime 在 close() 时绝不会关闭它，
       生命周期全权由调用方自行管理（严格遵循“谁创建、谁负责关闭”原则）。
    3. 不使用 with（裸调用）：
       绝不会自动关闭。必须由调用方显式调用 runtime.close()，或作为长效单例常驻复用。
    """

    LEVEL_ALBUM = 'album'
    LEVEL_PHOTO = 'photo'
    LEVEL_IMAGE = 'image'
    LEVEL_DECODE = 'decode'
    LEVEL_DEFAULT = 'default'

    def __init__(
            self,
            configs: Dict[str, Tuple[Optional[int], Optional[Executor]]],
    ):
        self._workers = {
            level: config[0]
            for level, config in configs.items()
        }
        self._executors = {
            level: config[1]
            for level, config in configs.items()
        }
        self._owned_executors: Set[Executor] = set()
        self._lock = RLock()
        self._closed = False
        self._ctx_cm = None

    def has_executor(self, level: Optional[str] = None) -> bool:
        """
        检查指定层级是否已由外部显式配置（workers 或 executor），或已初始化 Executor。
        """
        if level is None:
            raise TypeError('level is required for this Runtime')
        with self._lock:
            return self._workers.get(level) is not None or self._executors.get(level) is not None

    def executor(
            self,
            level: Optional[str] = None,
    ) -> Executor:
        """
        返回指定层级的 Executor；未显式配置 workers 时按系统默认并发数创建。
        """
        if level is None:
            raise TypeError('level is required for this Runtime')

        with self._lock:
            if self._closed:
                raise RuntimeError('JmRuntime is closed')

            try:
                executor = self._executors[level]
            except KeyError as error:
                raise ValueError(f'unknown runtime level: {level!r}') from error

            if executor is not None:
                return executor

            workers = self._workers[level]
            if workers is None:
                executor = ThreadPoolExecutor(thread_name_prefix=f'jm-{level}')
            else:
                executor = ThreadPoolExecutor(
                    max_workers=workers,
                    thread_name_prefix=f'jm-{level}',
                )

            self._executors[level] = executor
            self._owned_executors.add(executor)
            return executor

    def multi_thread_launcher(
            self,
            iter_objs,
            apply_each_obj_func: Callable,
            wait_finish=True,
            *,
            level: Optional[str] = None,
    ):
        """
        使用指定层级的 Executor 批量提交任务，并按需等待完成。
        """
        executor = self.executor(level)
        futures = []

        try:
            for obj in iter_objs:
                args, kwargs = process_single_arg_to_args_and_kwargs(obj)
                futures.append(executor.submit(apply_each_obj_func, *args, **kwargs))
        finally:
            if wait_finish:
                concurrent.futures.wait(futures)

        return futures

    def close(self) -> None:
        """
        关闭 Runtime 自建的 Executor；调用方传入的 Executor 保持可用。
        """
        with self._lock:
            if self._closed:
                return
            self._closed = True
            owned_executors = tuple(self._owned_executors)
            self._owned_executors.clear()

        for executor in owned_executors:
            executor.shutdown(wait=True)

    def __enter__(self):
        from .jm_task_context import jm_task_context
        self._ctx_cm = jm_task_context(runtime=self)
        self._ctx_cm.__enter__()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        try:
            if self._ctx_cm is not None:
                self._ctx_cm.__exit__(exc_type, exc_val, exc_tb)
                self._ctx_cm = None
        finally:
            self.close()

    async def __aenter__(self):
        return self.__enter__()

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        return self.__exit__(exc_type, exc_val, exc_tb)

close()

关闭 Runtime 自建的 Executor；调用方传入的 Executor 保持可用。

Source code in src/jmcomic/jm_runtime.py

def close(self) -> None:
    """
    关闭 Runtime 自建的 Executor；调用方传入的 Executor 保持可用。
    """
    with self._lock:
        if self._closed:
            return
        self._closed = True
        owned_executors = tuple(self._owned_executors)
        self._owned_executors.clear()

    for executor in owned_executors:
        executor.shutdown(wait=True)

executor(level=None)

返回指定层级的 Executor；未显式配置 workers 时按系统默认并发数创建。

Source code in src/jmcomic/jm_runtime.py

def executor(
        self,
        level: Optional[str] = None,
) -> Executor:
    """
    返回指定层级的 Executor；未显式配置 workers 时按系统默认并发数创建。
    """
    if level is None:
        raise TypeError('level is required for this Runtime')

    with self._lock:
        if self._closed:
            raise RuntimeError('JmRuntime is closed')

        try:
            executor = self._executors[level]
        except KeyError as error:
            raise ValueError(f'unknown runtime level: {level!r}') from error

        if executor is not None:
            return executor

        workers = self._workers[level]
        if workers is None:
            executor = ThreadPoolExecutor(thread_name_prefix=f'jm-{level}')
        else:
            executor = ThreadPoolExecutor(
                max_workers=workers,
                thread_name_prefix=f'jm-{level}',
            )

        self._executors[level] = executor
        self._owned_executors.add(executor)
        return executor

has_executor(level=None)

检查指定层级是否已由外部显式配置（workers 或 executor），或已初始化 Executor。

Source code in src/jmcomic/jm_runtime.py

def has_executor(self, level: Optional[str] = None) -> bool:
    """
    检查指定层级是否已由外部显式配置（workers 或 executor），或已初始化 Executor。
    """
    if level is None:
        raise TypeError('level is required for this Runtime')
    with self._lock:
        return self._workers.get(level) is not None or self._executors.get(level) is not None

multi_thread_launcher(iter_objs, apply_each_obj_func, wait_finish=True, *, level=None)

使用指定层级的 Executor 批量提交任务，并按需等待完成。

Source code in src/jmcomic/jm_runtime.py

def multi_thread_launcher(
        self,
        iter_objs,
        apply_each_obj_func: Callable,
        wait_finish=True,
        *,
        level: Optional[str] = None,
):
    """
    使用指定层级的 Executor 批量提交任务，并按需等待完成。
    """
    executor = self.executor(level)
    futures = []

    try:
        for obj in iter_objs:
            args, kwargs = process_single_arg_to_args_and_kwargs(obj)
            futures.append(executor.submit(apply_each_obj_func, *args, **kwargs))
    finally:
        if wait_finish:
            concurrent.futures.wait(futures)

    return futures

JmTempRuntime

Bases: JmRuntime

临时 Runtime：用于单次同步下载中的局部并发（章节层、图片层或批量任务的本子层）。 退出即销毁自建线程池，不自动绑定上下文，不跨任务共享。

Source code in src/jmcomic/jm_runtime.py

class JmTempRuntime(JmRuntime):
    """
    临时 Runtime：用于单次同步下载中的局部并发（章节层、图片层或批量任务的本子层）。
    退出即销毁自建线程池，不自动绑定上下文，不跨任务共享。
    """

    def __init__(
            self,
            *,
            workers: Optional[int] = None,
            executor: Optional[Executor] = None,
    ):
        if workers is not None and executor is not None:
            raise ValueError('workers and executor are mutually exclusive')
        if workers is not None:
            workers = validate_jm_workers('workers', workers)
        if executor is not None:
            executor = validate_jm_executor('executor', executor)

        super().__init__(
            configs={
                JmRuntime.LEVEL_DEFAULT: (workers, executor),
            },
        )

    def has_executor(self, level: Optional[str] = JmRuntime.LEVEL_DEFAULT) -> bool:
        return super().has_executor(level or JmRuntime.LEVEL_DEFAULT)

    def executor(
            self,
            level: Optional[str] = JmRuntime.LEVEL_DEFAULT,
    ) -> Executor:
        return super().executor(level or JmRuntime.LEVEL_DEFAULT)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        self.close()

JmSyncRuntime

Bases: JmRuntime

同步下载 Runtime：分别管理 album、photo、image 三层 Executor。

Source code in src/jmcomic/jm_runtime.py

class JmSyncRuntime(JmRuntime):
    """
    同步下载 Runtime：分别管理 album、photo、image 三层 Executor。
    """

    def __init__(
            self,
            *,
            album_workers: Optional[int] = None,
            album_executor: Optional[Executor] = None,
            photo_workers: Optional[int] = None,
            photo_executor: Optional[Executor] = None,
            image_workers: Optional[int] = None,
            image_executor: Optional[Executor] = None,
            # 兼容别名
            id_workers: Optional[int] = None,
            id_executor: Optional[Executor] = None,
    ):
        if album_workers is None and id_workers is not None:
            album_workers = id_workers
        if album_executor is None and id_executor is not None:
            album_executor = id_executor

        external_executors = [
            executor
            for executor in (album_executor, photo_executor, image_executor)
            if executor is not None
        ]
        if len({id(executor) for executor in external_executors}) != len(external_executors):
            raise ValueError(
                'album/photo/image levels must use different executor objects, '
                'as sharing one executor across nested levels causes thread pool starvation deadlock'
            )

        album_config = normalize_jm_executor_config(JmRuntime.LEVEL_ALBUM, album_workers, album_executor)
        super().__init__(
            configs={
                JmRuntime.LEVEL_ALBUM: album_config,
                JmRuntime.LEVEL_PHOTO: normalize_jm_executor_config(
                    JmRuntime.LEVEL_PHOTO, photo_workers, photo_executor
                ),
                JmRuntime.LEVEL_IMAGE: normalize_jm_executor_config(
                    JmRuntime.LEVEL_IMAGE, image_workers, image_executor
                ),
            },
        )

JmAsyncRuntime

Bases: JmRuntime

异步下载 Runtime：管理图片解密、处理和同步 hook 的 decode Executor。

Source code in src/jmcomic/jm_runtime.py

class JmAsyncRuntime(JmRuntime):
    """
    异步下载 Runtime：管理图片解密、处理和同步 hook 的 decode Executor。
    """

    def __init__(
            self,
            *,
            decode_workers: Optional[int] = None,
            decode_executor: Optional[Executor] = None,
    ):
        super().__init__(
            configs={
                JmRuntime.LEVEL_DECODE: normalize_jm_executor_config(
                    JmRuntime.LEVEL_DECODE, decode_workers, decode_executor
                ),
            },
        )

    def has_executor(self, level: Optional[str] = JmRuntime.LEVEL_DECODE) -> bool:
        return super().has_executor(level or JmRuntime.LEVEL_DECODE)

    def executor(
            self,
            level: Optional[str] = JmRuntime.LEVEL_DECODE,
    ) -> Executor:
        return super().executor(level or JmRuntime.LEVEL_DECODE)

DownloadCancelledException

Bases: JmcomicException

Methods:

Name	Description
__init__	按 JMComic 异常的统一结构创建下载取消异常。

Source code in src/jmcomic/jm_exception.py

class DownloadCancelledException(JmcomicException):
    description = '下载任务已取消'

    def __init__(self, msg='download cancelled', context=None):
        """
        按 JMComic 异常的统一结构创建下载取消异常。
        """
        msg = str(msg or 'download cancelled')
        context = dict(context or {})
        context.setdefault(ExceptionTool.CONTEXT_KEY_REASON, msg)
        super().__init__(msg, context)

    @property
    def control(self):
        return self.from_context(ExceptionTool.CONTEXT_KEY_CONTROL)

    @property
    def reason(self) -> str:
        return str(self.context.get(ExceptionTool.CONTEXT_KEY_REASON, 'download cancelled'))

    @property
    def completed_results(self):
        return list(self.context.get(ExceptionTool.CONTEXT_KEY_COMPLETED_RESULTS) or [])

    @property
    def unfinished_ids(self):
        return list(self.context.get(ExceptionTool.CONTEXT_KEY_UNFINISHED_IDS) or [])

__init__(msg='download cancelled', context=None)

按 JMComic 异常的统一结构创建下载取消异常。

Source code in src/jmcomic/jm_exception.py

def __init__(self, msg='download cancelled', context=None):
    """
    按 JMComic 异常的统一结构创建下载取消异常。
    """
    msg = str(msg or 'download cancelled')
    context = dict(context or {})
    context.setdefault(ExceptionTool.CONTEXT_KEY_REASON, msg)
    super().__init__(msg, context)

Functions:

Name	Description
bind_jm_task_context	把完整任务上下文快照绑定到同步可调用对象。

jm_task_context = JmTaskContext module-attribute

DownloadControl

在线程间共享、线程安全且幂等的下载取消信号。

Source code in src/jmcomic/jm_task_context.py

class DownloadControl:
    """
    在线程间共享、线程安全且幂等的下载取消信号。
    """

    def __init__(self):
        self._event = Event()
        self._lock = Lock()
        self._reason: Optional[str] = None

    @property
    def is_cancelled(self) -> bool:
        return self._event.is_set()

    def cancel(self, reason: str = 'download cancelled') -> bool:
        normalized_reason = str(reason or 'download cancelled')
        with self._lock:
            if self._event.is_set():
                return False
            self._reason = normalized_reason
            self._event.set()
            return True

    @property
    def reason(self) -> Optional[str]:
        with self._lock:
            return self._reason

bind_jm_task_context(func, context=None)

把完整任务上下文快照绑定到同步可调用对象。

Source code in src/jmcomic/jm_task_context.py

def bind_jm_task_context(func: Callable, context: Optional[Mapping] = None) -> Callable:
    """
    把完整任务上下文快照绑定到同步可调用对象。
    """
    if (
            inspect.iscoroutinefunction(func)
            or inspect.iscoroutinefunction(getattr(func, '__call__', None))
    ):
        raise TypeError('bind_jm_task_context only supports synchronous callables')

    snapshot = MappingProxyType(dict(
        JTC.get_context() if context is None else context
    ))

    @wraps(func)
    def wrapped(*args, **kwargs):
        token = JM_TASK_CONTEXT.set(snapshot)
        try:
            return func(*args, **kwargs)
        finally:
            JM_TASK_CONTEXT.reset(token)

    return wrapped

JMComic 任务上下文统一门面 (Jm Task Context)。

Methods:

Name	Description
get_runtime	返回当前任务绑定的 JmRuntime；没有活动 Runtime 时返回 None。
get_option	返回当前任务绑定的 JmOption；没有活动 Option 时返回 None。
get_control	返回当前任务绑定的取消控制器 DownloadControl；未设置时返回 None。
get_context	返回当前 JM 任务上下文的可变快照。

Source code in src/jmcomic/jm_task_context.py

class JTC:
    """
    JMComic 任务上下文统一门面 (Jm Task Context)。
    """

    @classmethod
    def get_context(cls) -> dict:
        """
        返回当前 JM 任务上下文的可变快照。
        """
        return dict(JM_TASK_CONTEXT.get())

    @classmethod
    def get_context_value(cls, key: str, default=None):
        """
        获取当前 JM 任务上下文指定 key 的值，避免全量 dict 拷贝。
        """
        return JM_TASK_CONTEXT.get().get(key, default)

    @classmethod
    def get_runtime(cls) -> Optional[JmRuntime]:
        """
        返回当前任务绑定的 JmRuntime；没有活动 Runtime 时返回 None。
        """
        return cls.get_context_value('runtime')

    @classmethod
    def get_option(cls):
        """
        返回当前任务绑定的 JmOption；没有活动 Option 时返回 None。
        """
        return cls.get_context_value('option')

    @classmethod
    def get_control(cls) -> Optional[DownloadControl]:
        """
        返回当前任务绑定的取消控制器 DownloadControl；未设置时返回 None。
        """
        control = cls.get_context_value('control')
        if control is None:
            return None
        if not isinstance(control, DownloadControl):
            raise TypeError(
                'jm_task_context control must be DownloadControl, '
                f'got {type(control)}'
            )
        return control

    @classmethod
    def is_cancelled(cls) -> bool:
        """
        当前任务是否已被标记取消。
        """
        control = cls.get_control()
        return control is not None and control.is_cancelled

    @classmethod
    def raise_if_cancelled(cls) -> None:
        """
        若当前任务已被标记取消，直接抛出 DownloadCancelledException。
        """
        control = cls.get_control()
        if control is None or not control.is_cancelled:
            return
        from .jm_exception import DownloadCancelledException, ExceptionTool
        raise DownloadCancelledException(
            control.reason,
            {
                ExceptionTool.CONTEXT_KEY_CONTROL: control,
                ExceptionTool.CONTEXT_KEY_REASON: control.reason,
            },
        )

get_runtime() classmethod

返回当前任务绑定的 JmRuntime；没有活动 Runtime 时返回 None。

Source code in src/jmcomic/jm_task_context.py

@classmethod
def get_runtime(cls) -> Optional[JmRuntime]:
    """
    返回当前任务绑定的 JmRuntime；没有活动 Runtime 时返回 None。
    """
    return cls.get_context_value('runtime')

get_option() classmethod

返回当前任务绑定的 JmOption；没有活动 Option 时返回 None。

Source code in src/jmcomic/jm_task_context.py

@classmethod
def get_option(cls):
    """
    返回当前任务绑定的 JmOption；没有活动 Option 时返回 None。
    """
    return cls.get_context_value('option')

get_control() classmethod

返回当前任务绑定的取消控制器 DownloadControl；未设置时返回 None。

Source code in src/jmcomic/jm_task_context.py

@classmethod
def get_control(cls) -> Optional[DownloadControl]:
    """
    返回当前任务绑定的取消控制器 DownloadControl；未设置时返回 None。
    """
    control = cls.get_context_value('control')
    if control is None:
        return None
    if not isinstance(control, DownloadControl):
        raise TypeError(
            'jm_task_context control must be DownloadControl, '
            f'got {type(control)}'
        )
    return control

get_context() classmethod

返回当前 JM 任务上下文的可变快照。

Source code in src/jmcomic/jm_task_context.py

@classmethod
def get_context(cls) -> dict:
    """
    返回当前 JM 任务上下文的可变快照。
    """
    return dict(JM_TASK_CONTEXT.get())