diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml index e534db734..dd928be5e 100644 --- a/.github/workflows/benchmark.yml +++ b/.github/workflows/benchmark.yml @@ -19,10 +19,10 @@ jobs: timeout-minutes: 15 steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v7 - name: Set up Python - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: '3.14' @@ -49,7 +49,7 @@ jobs: - name: Upload Benchmark Report if: always() - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v7 with: name: benchmark-report path: PERFORMANCE_REPORT.md diff --git a/.github/workflows/close_specific_pr.yml b/.github/workflows/close_specific_pr.yml index 8ed49d3eb..ce890e337 100644 --- a/.github/workflows/close_specific_pr.yml +++ b/.github/workflows/close_specific_pr.yml @@ -20,7 +20,7 @@ jobs: runs-on: ubuntu-latest steps: - name: Checkout code - uses: actions/checkout@v4 + uses: actions/checkout@v6 - name: Close PR run: | @@ -36,4 +36,4 @@ jobs: gh pr comment ${{ github.event.pull_request.number }} --repo ${{ github.repository }} --body '${{ env.comment }}' gh pr close ${{ github.event.pull_request.number }} --repo ${{ github.repository }} env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} \ No newline at end of file + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} diff --git a/.github/workflows/download.yml b/.github/workflows/download.yml index 380d4bc03..9ea493025 100644 --- a/.github/workflows/download.yml +++ b/.github/workflows/download.yml @@ -29,9 +29,9 @@ jobs: UPLOAD_NAME: 'Click me to download' steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v7 - name: Set up Python 3.11 - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: "3.11" @@ -69,7 +69,7 @@ jobs: mv "../$ZIP_NAME" . - name: 上传结果 - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v7 with: name: ${{ env.UPLOAD_NAME }} path: ${{ env.JM_DOWNLOAD_DIR }}/${{ env.ZIP_NAME }} diff --git a/.github/workflows/download_dispatch.yml b/.github/workflows/download_dispatch.yml index ae02fb8a0..a1c1596c4 100644 --- a/.github/workflows/download_dispatch.yml +++ b/.github/workflows/download_dispatch.yml @@ -108,9 +108,9 @@ jobs: JM_DOWNLOAD_DIR: /home/runner/work/jmcomic/download/ steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v7 - name: Set up Python 3.11 - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: "3.11" @@ -153,7 +153,7 @@ jobs: mv "../$ZIP_NAME" . - name: 上传结果 - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v7 with: name: ${{ env.UPLOAD_NAME }} path: ${{ env.JM_DOWNLOAD_DIR }}/${{ env.ZIP_NAME }} diff --git a/.github/workflows/export_favorites.yml b/.github/workflows/export_favorites.yml index a00cdf64c..b653deb79 100644 --- a/.github/workflows/export_favorites.yml +++ b/.github/workflows/export_favorites.yml @@ -48,9 +48,9 @@ jobs: ZIP_FP: /home/runner/work/jmcomic/download/export.7z steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v7 - name: Set up Python 3.11 - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: "3.11" @@ -71,7 +71,7 @@ jobs: python workflow_export_favorites.py - name: 上传结果 - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v7 with: name: '导出的收藏夹' path: ${{ env.ZIP_FP }} diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 8f17f7722..61f62a939 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -13,10 +13,10 @@ jobs: id-token: write contents: write steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v7 - name: Set up Python 3.11 - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: "3.11" @@ -31,7 +31,7 @@ jobs: python -m build - name: Create Release - uses: softprops/action-gh-release@v2 + uses: softprops/action-gh-release@v3 env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} with: diff --git a/.github/workflows/release_auto.yml b/.github/workflows/release_auto.yml index 0ef2d6cb1..d0fcf4aec 100644 --- a/.github/workflows/release_auto.yml +++ b/.github/workflows/release_auto.yml @@ -14,10 +14,10 @@ jobs: contents: write if: startsWith(github.event.head_commit.message, 'v') steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v7 - name: Set up Python 3.11 - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: "3.11" @@ -33,7 +33,7 @@ jobs: python -m build - name: Create Release - uses: softprops/action-gh-release@v2 + uses: softprops/action-gh-release@v3 env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} with: diff --git a/.github/workflows/test_api.yml b/.github/workflows/test_api.yml index e0e5d7558..35245d21f 100644 --- a/.github/workflows/test_api.yml +++ b/.github/workflows/test_api.yml @@ -27,10 +27,10 @@ jobs: steps: - name: Checkout repository - uses: actions/checkout@v4 + uses: actions/checkout@v7 - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: ${{ matrix.python-version }} diff --git a/.github/workflows/test_html.yml b/.github/workflows/test_html.yml index 8f44d7c9b..f8a45b50c 100644 --- a/.github/workflows/test_html.yml +++ b/.github/workflows/test_html.yml @@ -27,10 +27,10 @@ jobs: steps: - name: Checkout repository - uses: actions/checkout@v4 + uses: actions/checkout@v7 - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: ${{ matrix.python-version }} diff --git a/CHANGELOG.md b/CHANGELOG.md index 67b9680fc..11b84e2ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,17 @@ 条目分类参考 [Keep a Changelog](https://keepachangelog.com/zh-CN/1.0.0/)， 版本号遵循 [语义化版本](https://semver.org/lang/zh-CN/)。 +## [Unreleased] + +### Added +- 新增 `JmSimpleRuntime`、`JmSyncRuntime` 与 `JmAsyncRuntime`；裸同步 Downloader 的局部调度使用单池 Runtime，顶层同步 API 或自定义同步调度可复用 `id/photo/image` 三层线程池，异步下载可复用 `blocking` 线程池。 +- 新增 `DownloadControl` 和 `DownloadCancelledException`，支持通过任务上下文协作式取消同步与异步下载。 + +### Changed +- Python 3.9 保留安装兼容，但不再纳入 CI。 +- 下载调度统一使用标准库 Executor；顶层 API 和裸同步 Downloader 的临时调度显式关闭自己创建的 Runtime，`jm_task_context` 只传播字段；外部 Runtime 和 Executor 仍由调用方关闭。 +- 顶层下载会把 Runtime 和 Option 作为公开字段直接放入任务上下文；Runtime 不依赖 Context 或 Option，未配置的层级由实际调用点传入默认 worker 数。 + ## [2.7.5] - 2026-08-25 ### Summary diff --git a/assets/docs/mkdocs.yml b/assets/docs/mkdocs.yml index 003684fcb..182055fa5 100644 --- a/assets/docs/mkdocs.yml +++ b/assets/docs/mkdocs.yml @@ -58,6 +58,7 @@ nav: - tutorial/12_domain_strategy.md - tutorial/13_export_and_feature.md - tutorial/14_async_usage.md + - tutorial/16_shared_executors.md - tutorial/15_download_progress.md plugins: diff --git a/assets/docs/sources/api/download.md b/assets/docs/sources/api/download.md index ce7d19654..92817233a 100644 --- a/assets/docs/sources/api/download.md +++ b/assets/docs/sources/api/download.md @@ -17,3 +17,47 @@ options: members: - JmAsyncDownloader + +::: jmcomic.jm_downloader + options: + members: + - BaseDownloader + - JmDownloader + +## 下载 Runtime 与取消控制 + +::: jmcomic.jm_runtime + options: + members: + - JmRuntime + - JmSimpleRuntime + - JmSyncRuntime + - JmAsyncRuntime + +::: jmcomic.jm_exception + options: + members: + - DownloadCancelledException + +::: jmcomic.jm_task_context + options: + members: + - DownloadControl + - jm_task_context + - bind_jm_task_context + - get_jm_task_context + - get_current_control + - get_jm_runtime + - get_current_option + +同步顶层 API 默认创建 `JmSyncRuntime`，并在调用结束时显式关闭。需要让多个顶层调用复用线程池时，先创建 `JmSyncRuntime`，通过 `jm_task_context(runtime=runtime)` 传播，并在任务完成后显式调用 `runtime.close()`；异步调用改用 `JmAsyncRuntime`。裸同步 Downloader 没有 Runtime 时，会为每次局部调度创建并关闭只有一个线程池的 `JmSimpleRuntime`。完整示例见[复用下载 Runtime](../tutorial/16_shared_executors.md)。 + +Runtime 只负责 Executor 的配置、调度和生命周期。`JmSimpleRuntime.multi_thread_launcher()` 不要求下载层级；`JmSyncRuntime` 的 launcher 使用 `id/photo/image` 层级。未显式配置容量时，由下载调用点把 Option 或 Downloader 已解析出的默认 worker 数传给 Runtime。 + +使用 `get_jm_runtime()` 可以读取当前任务激活的 Runtime；没有激活 Runtime 时返回 `None`。 + +Runtime 和 Option 以公开字段 `runtime`、`option` 直接保存在 `JM_TASK_CONTEXT` 中。`get_jm_task_context()` 返回包含这两个字段的完整副本；`get_jm_runtime()` 和 `get_current_option()` 是读取它们的便捷方法。自定义日志处理器也可以从 LogRecord 的任务上下文中访问这两个公开字段，默认文本日志不会展开对象内容。 + +`jm_task_context` 只负责字段传播与作用域恢复，不会关闭 Runtime。谁创建 Runtime，谁显式调用 `runtime.close()`；Runtime 只会关闭自身创建的 Executor，不会关闭调用方注入的 Executor。 + +取消检查统一由 `BaseDownloader.raise_if_cancelled()` 这个 `classmethod` 执行。自定义 Downloader 可以重写它来调整检查策略；Client 和顶层 API 不直接读取取消状态。 diff --git a/assets/docs/sources/tutorial/0_common_usage.md b/assets/docs/sources/tutorial/0_common_usage.md index 2093e91e5..dcb07a627 100644 --- a/assets/docs/sources/tutorial/0_common_usage.md +++ b/assets/docs/sources/tutorial/0_common_usage.md @@ -574,6 +574,62 @@ print('是否全部成功:', batch_result.all_succeeded) 下载单个 ID 时，请求本子失败会直接抛出异常；如果只有部分章节或图片失败，会在任务结束后汇总抛出 `PartialDownloadFailedException`，此时不会返回 `DownloadResult`。批量下载则继续执行其他任务，并把失败项集中放进 `batch_result.failed`。 +### 取消下载 + +下载开始后，可以从按钮、定时器或其他线程调用 `DownloadControl.cancel()` 请求停止。JMComic 收到请求后不会再开始下载新的章节和图片，但会先把当前正在处理的图片完整保存，避免留下损坏文件；随后下载会抛出 `DownloadCancelledException`，调用方可以据此提示用户任务已经取消。 + +下面的同步示例启动一个下载线程，并在两秒后从主线程请求取消： + +```python +from threading import Thread +from time import sleep + +from jmcomic import DownloadCancelledException, DownloadControl, download_album, jm_task_context + +control = DownloadControl() + +def run_download(): + try: + # ContextVar 不会自动进入用户创建的新线程， + # 因此 context 必须在实际调用下载的线程内建立。 + with jm_task_context(control=control): + download_album('123') + except DownloadCancelledException as e: + print('取消原因:', e.reason) + +thread = Thread(target=run_download) +thread.start() + +# GUI 按钮、请求处理器或其他线程调用 +sleep(2) +control.cancel('用户取消') +thread.join() +``` + +异步 API 使用完全相同的 `DownloadControl` 和 context。`asyncio.create_task()` 会自动复制当前 Context，因此 Task 会看到同一个 `control`。`DownloadControl` 的协作式取消路径不会调用 `asyncio.Task.cancel()`；如果要保留“当前图片完整写入后再停止”的保证，只调用 `control.cancel()`。如果调用方直接取消外层异步批量 Task，批量 API 会取消并等待其内部子 Task 完成清理，再透传 `CancelledError`；这条外部取消路径不保证当前图片完整写入。 + +```python +import asyncio + +from jmcomic import DownloadCancelledException, DownloadControl, download_album_async, jm_task_context + +async def main(): + control = DownloadControl() + with jm_task_context(control=control): + task = asyncio.create_task(download_album_async('123')) + + await asyncio.sleep(2) + control.cancel('用户取消') + try: + await task + except DownloadCancelledException as e: + print('取消原因:', e.reason) + +asyncio.run(main()) +``` + +取消不会删除已经完整写入的图片。同步 HTTP 请求以及已经运行的解密或写盘不能被强制中断，因此取消会在当前不可中断操作完成或超时后生效。取消后不会执行整章、整本完成回调及 PDF/ZIP 等导出 Feature。 + ### 速查表 | 你的需求 | 推荐写法 | diff --git a/assets/docs/sources/tutorial/14_async_usage.md b/assets/docs/sources/tutorial/14_async_usage.md index dfac607f9..007cd7f80 100644 --- a/assets/docs/sources/tutorial/14_async_usage.md +++ b/assets/docs/sources/tutorial/14_async_usage.md @@ -19,9 +19,12 @@ import jmcomic async def main(): # 异步下载单个本子 - album, downloader = await jmcomic.download_album_async('438696') + result = await jmcomic.download_album_async('438696') + album = result.detail + downloader = result.downloader - # 返回的 downloader 已释放网络连接和线程池，只用于读取下载结果 + # 返回时该 Downloader 的网络 client 已关闭；如果传入共享 blocking + # 执行器，它仍由创建它的调用方管理 print(downloader.download_failed_image) # 异步下载单章节 @@ -52,6 +55,8 @@ async def main(): asyncio.run(main()) ``` +异步下载的网络 I/O 仍由 event loop 和 Semaphore 控制；解密、PIL、写盘及同步 hook 使用标准线程池。每个 `JmAsyncDownloader` 仍独立创建并关闭自己的 client、session 和并发控制；需要显式复用线程池时，请参阅[复用下载 Runtime](16_shared_executors.md)。 + ## 3. 异步获取实体类，并发请求 ### 💡 关于 async with 和自动初始化 diff --git a/assets/docs/sources/tutorial/16_shared_executors.md b/assets/docs/sources/tutorial/16_shared_executors.md new file mode 100644 index 000000000..763b8f71f --- /dev/null +++ b/assets/docs/sources/tutorial/16_shared_executors.md @@ -0,0 +1,109 @@ +# 复用下载 Runtime + +普通下载不需要手动创建 Runtime。同步顶层 API 使用临时 `JmSyncRuntime` 并在返回前显式关闭；裸同步 Downloader 没有 Runtime 时，每次局部调度使用一个临时 `JmSimpleRuntime`。只有当你希望多个下载调用复用同一组线程池，或想明确控制各层并发数时，才需要 `jm_task_context`： + +```python +from jmcomic import JmSyncRuntime, download_album, jm_task_context + + +runtime = JmSyncRuntime( + id_workers=2, + photo_workers=3, + image_workers=8, +) +try: + with jm_task_context(runtime=runtime): + download_album(['123', '456']) +finally: + runtime.close() +``` + +这就是分层同步 Runtime 的完整公开用法。`JmSyncRuntime` 负责 `id`、`photo` 和 `image` 三层调度；任务上下文只传播 Runtime，不管理资源，创建 Runtime 的代码负责显式关闭它。 + +如果只需要直接调度一层独立任务，也可以使用 `JmSimpleRuntime(workers=...)` 或 `JmSimpleRuntime(executor=...)`。它的 `multi_thread_launcher()` 不接收 `level`。不要把同一个单池 Runtime 用于 album 的 photo/image 两层同步嵌套调度，否则外层 worker 等待内层 worker 时可能耗尽线程。 + +## Runtime 默认容量从哪里来 + +Runtime 本身不读取 Option，也不保存下载任务状态。它只管理 Executor 的配置、创建、调度和关闭。 + +如果你没有给某一层配置 `*_workers`，真正发起调度的调用点会把已经解析好的默认值传给 Runtime：`photo` 和 `image` 使用 Option 中的 `download.threading.photo/image`，批量 ID 使用本次 ID 数量，异步阻塞池使用 Downloader 的 `decode_worker`。这样 Runtime 不需要反向依赖 Context 或 Option。 + +你也可以直接使用公开的 `multi_thread_launcher()`，但普通下载通常不需要这样做。`wait_finish=True` 会等待所有 Future 完成；worker 异常保存在对应 Future 中，由需要结果的调用方通过 `future.result()` 读取： + +```python +from jmcomic import JmSyncRuntime, jm_task_context + + +runtime = JmSyncRuntime(id_workers=2) +try: + with jm_task_context(runtime=runtime): + futures = runtime.multi_thread_launcher( + [123, 456], + str, + level='id', + ) + print([future.result() for future in futures]) +finally: + runtime.close() +``` + +## 借用已有线程池 + +如果应用已经管理自己的 `ThreadPoolExecutor`，可以把它交给 Runtime。Runtime 只借用外部 Executor，不会替你关闭： + +```python +from concurrent.futures import ThreadPoolExecutor + +from jmcomic import JmSyncRuntime, download_album, jm_task_context + + +with ThreadPoolExecutor(max_workers=2) as id_executor, \ + ThreadPoolExecutor(max_workers=3) as photo_executor, \ + ThreadPoolExecutor(max_workers=8) as image_executor: + runtime = JmSyncRuntime( + id_executor=id_executor, + photo_executor=photo_executor, + image_executor=image_executor, + ) + try: + with jm_task_context(runtime=runtime): + download_album(['123', '456']) + finally: + runtime.close() +``` + +三个同步层级会互相等待，因此必须使用不同的 Executor。任务上下文需要在线程间传播，所以不支持 `ProcessPoolExecutor`。 + +## 异步下载 + +异步网络请求由 event loop 和 Semaphore 并发；图片解密、PIL、写盘和同步 hook 交给 `blocking` Executor： + +```python +import asyncio +from concurrent.futures import ThreadPoolExecutor + +from jmcomic import JmAsyncRuntime, download_album_async, jm_task_context + + +async def main(): + with ThreadPoolExecutor(max_workers=4) as blocking_executor: + runtime = JmAsyncRuntime(blocking_executor=blocking_executor) + try: + with jm_task_context(runtime=runtime): + await download_album_async('123') + finally: + runtime.close() + + +asyncio.run(main()) +``` + +共享 `blocking` Executor 不会共享 Downloader、网络 Client、Session 或 Semaphore。每个顶层调用仍有自己的 Downloader 和 Manifest。 + +## Context 中保存什么 + +项目只有一个 `JM_TASK_CONTEXT`。Runtime 和 Option 直接使用公开字段 `runtime`、`option` 保存，和 `download_type`、`jm_id` 以及调用方附加字段处于同一个 context mapping 中。`get_jm_task_context()` 返回包含所有字段的完整副本。 + +可以直接从 context 副本读取 `runtime`、`option`，也可以使用便捷方法 `get_jm_runtime()`、`get_current_option()`。`bind_jm_task_context()` 会把完整上下文快照传播到工作线程。默认文本日志不会展开 Runtime 或 Option，但自定义日志处理器可以读取这两个公开字段。 + +嵌套上下文中的 `runtime=None` 和 `option=None` 表示继承父上下文。同步下载 API 只接受 `JmSyncRuntime`，异步 API 只接受 `JmAsyncRuntime`；同一任务作用域不能替换成另一个 Runtime。`JmSimpleRuntime` 用于独立的单层调度，不作为分层同步下载的共享 context Runtime。 diff --git a/pyproject.toml b/pyproject.toml index 007c68baa..70922cf53 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,4 +41,4 @@ jmcomic = "jmcomic.cli:main" jmv = "jmcomic.cli:view_main" [tool.setuptools.dynamic] -version = {attr = "jmcomic.__version__"} \ No newline at end of file +version = {attr = "jmcomic.__version__"} diff --git a/src/jmcomic/__init__.py b/src/jmcomic/__init__.py index fdf04d29a..080f708da 100644 --- a/src/jmcomic/__init__.py +++ b/src/jmcomic/__init__.py @@ -4,6 +4,8 @@ __version__ = '2.7.5' +from .jm_exception import DownloadCancelledException +from .jm_runtime import * from .jm_task_context import * from .api import * from .jm_plugin import * diff --git a/src/jmcomic/api.py b/src/jmcomic/api.py index 6ef842835..a5c252241 100644 --- a/src/jmcomic/api.py +++ b/src/jmcomic/api.py @@ -2,7 +2,13 @@ from time import perf_counter from .jm_downloader import * -from .jm_task_context import bind_jm_task_context, jm_task_context +from .jm_exception import DownloadCancelledException +from .jm_runtime import JmAsyncRuntime, JmSyncRuntime +from .jm_task_context import ( + bind_jm_task_context, + get_jm_runtime, + jm_task_context, +) __DOWNLOAD_API_RET = DownloadResult @@ -22,12 +28,27 @@ def _finish_download_result(detail, dler, task_started_at): return DownloadResult(detail, dler) -def download_batch(download_api, - jm_id_iter: Union[Iterable, Generator], - option=None, - downloader=None, - **kwargs, - ) -> BatchResult: +def _ensure_option(option): + return JmModuleConfig.option_class().default() if option is None else option + + +def _resolve_runtime(runtime_type, error_message): + runtime = get_jm_runtime() + if runtime is None: + runtime = runtime_type() + return runtime, True + if not isinstance(runtime, runtime_type): + raise TypeError(error_message) + return runtime, False + + +def download_batch( + download_api, + jm_id_iter: Union[Iterable, Generator], + option=None, + downloader=None, + **kwargs, +) -> BatchResult: """ 批量下载 album / photo @@ -40,35 +61,66 @@ def download_batch(download_api, :param option: 下载选项，所有的jmid共用一个option :param downloader: 下载器类 """ - from common import multi_thread_launcher - - if option is None: - option = JmModuleConfig.option_class().default() - + option = _ensure_option(option) + jm_ids = list({ + JmcomicText.parse_to_jm_id(jmid) + for jmid in jm_id_iter + }) result = BatchResult() - + if len(jm_ids) == 0: + return result download_type = _download_type(download_api) - def _safe_download(aid): - """batch 内部的单任务包装：确保异常被收集而非静默丢失""" + def _download_one(aid): with jm_task_context(download_type=download_type, jm_id=str(aid)): try: - ret = download_api(aid, option, downloader, **kwargs) - result.add(ret) - except Exception as e: - jm_log('batch.failed', f'批量下载失败: [{aid}], 异常: [{e}]', e) - result.failed[str(aid)] = e - - multi_thread_launcher( - iter_objs=set( - JmcomicText.parse_to_jm_id(jmid) - for jmid in jm_id_iter - ), - apply_each_obj_func=bind_jm_task_context(_safe_download), - wait_finish=True + item = download_api(aid, option, downloader, **kwargs) + return aid, item, None + except Exception as error: + return aid, None, error + + runtime, runtime_created_here = _resolve_runtime( + JmSyncRuntime, + 'sync batch API requires JmSyncRuntime', ) - return result + cancellation_error = None + try: + with jm_task_context(option=option, runtime=runtime): + worker = bind_jm_task_context(_download_one) + futures = runtime.multi_thread_launcher( + iter_objs=jm_ids, + apply_each_obj_func=worker, + wait_finish=True, + level='id', + default_workers=len(jm_ids), + ) + + for future in futures: + aid, item, error = future.result() + if error is None: + result.add(item) + continue + + if isinstance(error, DownloadCancelledException): + if cancellation_error is None: + cancellation_error = error + continue + + with jm_task_context(download_type=download_type, jm_id=str(aid)): + jm_log( + 'batch.failed', + f'批量下载失败: [{aid}], 异常: [{error}]', + error, + ) + result.failed[str(aid)] = error + + if cancellation_error is not None: + raise cancellation_error + return result + finally: + if runtime_created_here: + runtime.close() def download_album(jm_album_id, @@ -96,17 +148,31 @@ def download_album(jm_album_id, if not isinstance(jm_album_id, (str, int)): return download_batch(download_album, jm_album_id, option, downloader, extra=extra) + option = _ensure_option(option) task_started_at = perf_counter() - with jm_task_context(download_type='album', jm_id=str(jm_album_id), task_started_at=task_started_at): - with new_downloader(option, downloader) as dler: - # 下载类型已记录在 TaskContext 中，Feature 会据此选择执行钩子 - dler.add_features(extra) - album = dler.download_album(jm_album_id) - - if check_exception: - dler.raise_if_has_exception() + runtime, runtime_created_here = _resolve_runtime( + JmSyncRuntime, + 'sync API requires JmSyncRuntime', + ) - return _finish_download_result(album, dler, task_started_at) + try: + with jm_task_context( + option=option, + runtime=runtime, + download_type='album', + jm_id=str(jm_album_id), + task_started_at=task_started_at, + ): + with new_downloader(option, downloader) as dler: + dler.add_features(extra) + album = dler.download_album(jm_album_id) + if check_exception: + dler.raise_if_has_exception() + + return _finish_download_result(album, dler, task_started_at) + finally: + if runtime_created_here: + runtime.close() def download_photo(jm_photo_id, @@ -125,17 +191,31 @@ def download_photo(jm_photo_id, if not isinstance(jm_photo_id, (str, int)): return download_batch(download_photo, jm_photo_id, option, downloader, extra=extra) + option = _ensure_option(option) task_started_at = perf_counter() - with jm_task_context(download_type='photo', jm_id=str(jm_photo_id), task_started_at=task_started_at): - with new_downloader(option, downloader) as dler: - # 下载类型已记录在 TaskContext 中，Feature 会据此选择执行钩子 - dler.add_features(extra) - photo = dler.download_photo(jm_photo_id) - - if check_exception: - dler.raise_if_has_exception() + runtime, runtime_created_here = _resolve_runtime( + JmSyncRuntime, + 'sync API requires JmSyncRuntime', + ) - return _finish_download_result(photo, dler, task_started_at) + try: + with jm_task_context( + option=option, + runtime=runtime, + download_type='photo', + jm_id=str(jm_photo_id), + task_started_at=task_started_at, + ): + with new_downloader(option, downloader) as dler: + dler.add_features(extra) + photo = dler.download_photo(jm_photo_id) + if check_exception: + dler.raise_if_has_exception() + + return _finish_download_result(photo, dler, task_started_at) + finally: + if runtime_created_here: + runtime.close() def new_downloader(option=None, downloader=None) -> JmDownloader: @@ -188,13 +268,15 @@ async def download_album_async(jm_album_id, check_exception=True, extra=None, ): - """ - 异步下载一个本子（album），包含其所有的章节（photo）。 + """异步下载一个本子及其全部章节。 + + 当 ``jm_album_id`` 是可迭代对象时使用批量 API。单个 ID 返回 + ``DownloadResult``；其中 Downloader 的网络 client 已关闭，阻塞工作也已 + 收敛。通过 ``jm_task_context(runtime=JmAsyncRuntime(...))`` 提供的 blocking + 执行器由调用方负责关闭。 - - 支持批量下载（当 jm_album_id 为可迭代对象时） - - 返回 (album, downloader) 元组，其中 downloader 的网络和线程池资源已关闭，仅用于读取下载结果 - - check_exception 仅当 jm_album_id 是单个 ID 时生效。多 ID 场景请检查 BatchResult.failed， - 或自行封装 download_batch_async 处理批量异常 + ``check_exception`` 只对单个 ID 生效。批量场景应检查 + ``BatchResult.failed``，或按需封装 ``download_batch_async``。 """ if not isinstance(jm_album_id, (str, int)): return await download_batch_async(download_album_async, @@ -204,16 +286,31 @@ async def download_album_async(jm_album_id, extra=extra ) + option = _ensure_option(option) task_started_at = perf_counter() - with jm_task_context(download_type='album', jm_id=str(jm_album_id), task_started_at=task_started_at): - async with new_async_downloader(option, downloader) as dler: - dler.add_features(extra) - album = await dler.download_album(jm_album_id) - - if check_exception: - dler.raise_if_has_exception() + runtime, runtime_created_here = _resolve_runtime( + JmAsyncRuntime, + 'async API requires JmAsyncRuntime', + ) - return _finish_download_result(album, dler, task_started_at) + try: + with jm_task_context( + option=option, + runtime=runtime, + download_type='album', + jm_id=str(jm_album_id), + task_started_at=task_started_at, + ): + async with new_async_downloader(option, downloader) as dler: + dler.add_features(extra) + album = await dler.download_album(jm_album_id) + if check_exception: + dler.raise_if_has_exception() + + return _finish_download_result(album, dler, task_started_at) + finally: + if runtime_created_here: + runtime.close() async def download_photo_async(jm_photo_id, @@ -223,11 +320,13 @@ async def download_photo_async(jm_photo_id, check_exception=True, extra=None, ): - """ - 异步下载一个章节（photo）。 - 返回的 downloader 已关闭网络和线程池资源，仅用于读取下载结果。 - check_exception 仅当 jm_photo_id 是单个 ID 时生效。多 ID 场景请检查 - BatchResult.failed，或自行封装 download_batch_async 处理批量异常。 + """异步下载一个章节。 + + 返回的 Downloader 已关闭网络 client，阻塞工作也已收敛。通过 + ``jm_task_context(runtime=JmAsyncRuntime(...))`` 提供的 blocking 执行器由调用方负责关闭。 + + ``check_exception`` 只对单个 ID 生效。批量场景应检查 + ``BatchResult.failed``，或按需封装 ``download_batch_async``。 """ if not isinstance(jm_photo_id, (str, int)): return await download_batch_async(download_photo_async, @@ -237,52 +336,87 @@ async def download_photo_async(jm_photo_id, extra=extra ) + option = _ensure_option(option) task_started_at = perf_counter() - with jm_task_context(download_type='photo', jm_id=str(jm_photo_id), task_started_at=task_started_at): - async with new_async_downloader(option, downloader) as dler: - dler.add_features(extra) - photo = await dler.download_photo(jm_photo_id) - - if check_exception: - dler.raise_if_has_exception() - - return _finish_download_result(photo, dler, task_started_at) - + runtime, runtime_created_here = _resolve_runtime( + JmAsyncRuntime, + 'async API requires JmAsyncRuntime', + ) -async def download_batch_async(download_api, - jm_id_iter, - option=None, - downloader=None, - **kwargs, - ) -> BatchResult: + try: + with jm_task_context( + option=option, + runtime=runtime, + download_type='photo', + jm_id=str(jm_photo_id), + task_started_at=task_started_at, + ): + async with new_async_downloader(option, downloader) as dler: + dler.add_features(extra) + photo = await dler.download_photo(jm_photo_id) + if check_exception: + dler.raise_if_has_exception() + + return _finish_download_result(photo, dler, task_started_at) + finally: + if runtime_created_here: + runtime.close() + + +async def download_batch_async( + download_api, + jm_id_iter, + option=None, + downloader=None, + **kwargs, +) -> BatchResult: """ 异步批量下载 album / photo。 - 容错机制：单个 album/photo 失败不会中止整批，也不会丢失其它已完成结果。 - 返回 BatchResult(set)，失败项收集在 result.failed 中。 """ - if option is None: - option = JmModuleConfig.option_class().default() - + 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 download_type = _download_type(download_api) 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) - results = await asyncio.gather( - *(_download_one(jmid) for jmid in jm_ids), - return_exceptions=True, + runtime, runtime_created_here = _resolve_runtime( + JmAsyncRuntime, + 'async API requires JmAsyncRuntime', ) - # 失败不抛出，但要记录到 result.failed，便于调用者排查 - result = BatchResult() - for jmid, r in zip(jm_ids, results): - if isinstance(r, BaseException): - with jm_task_context(download_type=download_type, jm_id=str(jmid)): - jm_log('async.batch.failed', f'批量下载失败: [{jmid}], 异常: [{r}]', r) - result.failed[str(jmid)] = r - else: - result.add(r) - - return result + try: + with jm_task_context(option=option, runtime=runtime): + tasks = [asyncio.create_task(_download_one(jmid)) for jmid in jm_ids] + results = await asyncio.gather(*tasks, return_exceptions=True) + + for item in results: + if isinstance(item, DownloadCancelledException): + raise item + + 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 + finally: + if runtime_created_here: + runtime.close() diff --git a/src/jmcomic/cli.py b/src/jmcomic/cli.py index ad1871db8..ffea99858 100644 --- a/src/jmcomic/cli.py +++ b/src/jmcomic/cli.py @@ -135,6 +135,7 @@ def option_has_download_progress(option): def run(self, option): from .api import download_album, download_photo + from .jm_task_context import bind_jm_task_context from common import MultiTaskLauncher if len(self.album_id_list) == 0: @@ -146,11 +147,11 @@ def run(self, option): launcher = MultiTaskLauncher() launcher.create_task( - target=download_album, + target=bind_jm_task_context(download_album), args=(self.album_id_list, option) ) launcher.create_task( - target=download_photo, + target=bind_jm_task_context(download_photo), args=(self.photo_id_list, option) ) diff --git a/src/jmcomic/jm_async_client.py b/src/jmcomic/jm_async_client.py index d7ac2b27e..a558448d3 100644 --- a/src/jmcomic/jm_async_client.py +++ b/src/jmcomic/jm_async_client.py @@ -681,8 +681,11 @@ async def add_favorite_album(self, album_id, folder_id='0'): 将指定图集加入用户的收藏夹。 注意：移动端没有提供 folder_id 参数。 """ - # 服务端实现上使用带 body 的 GET 请求方式 - resp = await self.req_api('/favorite', data={'aid': album_id}) + resp = await self.req_api( + self.API_FAVORITE, + get=False, + data={'aid': album_id}, + ) data = resp.model_data if data.status != 'ok': ExceptionTool.raises_resp(data.msg, resp) diff --git a/src/jmcomic/jm_async_downloader.py b/src/jmcomic/jm_async_downloader.py index 3f89e24e0..e6eb4449d 100644 --- a/src/jmcomic/jm_async_downloader.py +++ b/src/jmcomic/jm_async_downloader.py @@ -10,13 +10,19 @@ import asyncio import os -from concurrent.futures import ThreadPoolExecutor +from contextlib import ExitStack from .jm_downloader import BaseDownloader, record_download_duration from .jm_entity import JmAlbumDetail, JmPhotoDetail, JmImageDetail from .jm_toolkit import JmImageTool from .jm_config import JmModuleConfig, jm_log -from .jm_task_context import bind_jm_task_context +from .jm_exception import DownloadCancelledException +from .jm_runtime import JmAsyncRuntime +from .jm_task_context import ( + bind_jm_task_context, + get_jm_runtime, + jm_task_context, +) from .jm_option import JmOption @@ -50,8 +56,19 @@ def __init__(self, self._image_semaphore = asyncio.Semaphore(image_concurrency) self._photo_semaphore = asyncio.Semaphore(photo_concurrency) - # 解密线程池（CPU 密集操作卸载） - self._decode_pool = ThreadPoolExecutor(max_workers=decode_worker, thread_name_prefix='jm-async-decode') + 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._runtime_context = ExitStack() @classmethod def use(cls, *args, **kwargs): @@ -70,17 +87,35 @@ def use(cls, *args, **kwargs): # ====================================================================== async def _run_in_decode_pool(self, func, *args): - loop = asyncio.get_running_loop() - return await loop.run_in_executor( - self._decode_pool, - bind_jm_task_context(func), - *args, - ) + runtime = get_jm_runtime() + if not isinstance(runtime, JmAsyncRuntime): + raise RuntimeError( + 'async downloader requires jm_task_context(' + 'runtime=JmAsyncRuntime(...))' + ) + worker = bind_jm_task_context(func) + executor = runtime.executor('blocking', self._decode_worker) + future = executor.submit(worker, *args) + waiter = asyncio.wrap_future(future) + try: + return await asyncio.shield(waiter) + except asyncio.CancelledError: + # 线程任务无法可靠中断，协程取消后仍需等待写盘等操作收尾。 + try: + await waiter + except BaseException: + pass + raise @record_download_duration('album_started_at') async def download_album(self, album_id) -> JmAlbumDetail: """对齐 sync JmDownloader.download_album""" - album = await self.client.get_album_detail(album_id) + self.raise_if_cancelled() + try: + album = await self.client.get_album_detail(album_id) + except Exception: + self.raise_if_cancelled() + raise self.begin_manifest(album) try: await self.download_by_album_detail(album) @@ -106,7 +141,10 @@ async def download_by_album_detail(self, album: JmAlbumDetail): if photos: # photo 级并发由 _photo_semaphore 控制（默认 3），包裹整段 photo 下载（见 download_by_photo_detail）。 photo_tasks = [self._safe_download_photo(photo) for photo in photos] - await asyncio.gather(*photo_tasks) + results = await asyncio.gather(*photo_tasks, return_exceptions=True) + for item in results: + if isinstance(item, BaseException): + raise item await self.after_album(album) @@ -114,14 +152,22 @@ 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: + self.raise_if_cancelled() jm_log('photo.failed', f'章节下载失败: [{photo.id}], 异常: [{e}]', e) self.download_failed_photo.append((photo, e)) @record_download_duration('photo_started_at') async def download_photo(self, photo_id) -> JmPhotoDetail: """对齐 sync JmDownloader.download_photo""" - photo = await self.client.get_photo_detail(photo_id) + self.raise_if_cancelled() + try: + photo = await self.client.get_photo_detail(photo_id) + except Exception: + self.raise_if_cancelled() + raise self.begin_manifest(photo) try: await self.download_by_photo_detail(photo) @@ -135,11 +181,14 @@ 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) @@ -156,7 +205,10 @@ async def download_by_photo_detail(self, photo: JmPhotoDetail): self._safe_download_image(image) for image in image_list ] - await asyncio.gather(*download_tasks) + results = await asyncio.gather(*download_tasks, return_exceptions=True) + for item in results: + if isinstance(item, BaseException): + raise item await self.after_photo(photo) @@ -167,7 +219,10 @@ async def _safe_download_image(self, image: JmImageDetail): """ try: await self._download_single_image(image) + except DownloadCancelledException: + raise except Exception as e: + self.raise_if_cancelled() jm_log('image.failed', f'图片下载失败: [{image.download_url}], 异常: [{e}]', e) self.download_failed_image.append((image, e)) @@ -188,12 +243,15 @@ async def _download_single_image(self, image: JmImageDetail): if image.cache and image.exists: await self.after_image(image, img_save_path) + self.raise_if_cancelled() 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 @@ -225,6 +283,7 @@ async def _download_single_image(self, image: JmImageDetail): ) await self.after_image(image, img_save_path) + self.raise_if_cancelled() # ====================================================================== # 磁盘写入（在线程池中执行） @@ -278,17 +337,30 @@ async def after_image(self, image: JmImageDetail, img_save_path: str): await self._run_in_decode_pool(super().after_image, image, img_save_path) def shutdown(self): - """关闭解密线程池""" - self._decode_pool.shutdown(wait=False) + """关闭 __aenter__ 建立的任务上下文和自建 Runtime。""" + self._runtime_context.close() async def __aenter__(self): - # 创建并独占一个 async client（含 AsyncSession）。 - self.client = self.option.new_jm_async_client(max_clients=self._image_concurrency) + runtime = get_jm_runtime() + if runtime is None: + runtime = JmAsyncRuntime() + self._runtime_context.callback(runtime.close) + self._runtime_context.enter_context( + jm_task_context(runtime=runtime) + ) + if not isinstance(runtime, JmAsyncRuntime): + raise TypeError('async downloader requires JmAsyncRuntime') + try: + # 创建并独占一个 async client（含 AsyncSession）。 + self.client = self.option.new_jm_async_client( + max_clients=self._image_concurrency + ) await self.client.setup() except BaseException: try: - await self.client.close() + 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) @@ -299,8 +371,6 @@ async def __aenter__(self): return self async def __aexit__(self, exc_type, exc_val, exc_tb): - # 关闭顺序：先关网络 client（释放 AsyncSession / libcurl multi handle / 后台任务）， - # 再关解密线程池。两者都要在异常路径下保证释放。 try: if self.client is not None: await self.client.close() diff --git a/src/jmcomic/jm_client_impl.py b/src/jmcomic/jm_client_impl.py index c183a2624..449fd46c1 100644 --- a/src/jmcomic/jm_client_impl.py +++ b/src/jmcomic/jm_client_impl.py @@ -268,15 +268,29 @@ def add_favorite_album(self, folder_id='0', ): data = { - 'album_id': album_id, - 'fid': folder_id, + 'album_id': str(album_id), + 'fid': str(folder_id), } - resp = self.get_jm_html( + resp = self.post( '/ajax/favorite_album', data=data, + headers={ + 'accept': 'application/json, text/javascript, */*; q=0.01', + 'content-type': 'application/x-www-form-urlencoded; charset=UTF-8', + 'sec-fetch-dest': 'empty', + 'sec-fetch-mode': 'cors', + 'sec-fetch-site': 'same-origin', + 'x-requested-with': 'XMLHttpRequest', + }, ) + if resp.status_code != 200: + self.check_special_http_code(resp) + self.raise_request_error(resp) + + self.require_resp_success_else_raise(resp, '/ajax/favorite_album') + res = resp.json() if res['status'] != 1: @@ -967,7 +981,8 @@ def add_favorite_album(self, 移动端没有提供folder_id参数 """ resp = self.req_api( - '/favorite', + self.API_FAVORITE, + get=False, data={ 'aid': album_id, }, diff --git a/src/jmcomic/jm_downloader.py b/src/jmcomic/jm_downloader.py index 2846e547b..41925276f 100644 --- a/src/jmcomic/jm_downloader.py +++ b/src/jmcomic/jm_downloader.py @@ -5,7 +5,14 @@ from time import perf_counter from .jm_option import * -from .jm_task_context import bind_jm_task_context, get_jm_task_context, jm_task_context +from .jm_runtime import JmSimpleRuntime, JmSyncRuntime +from .jm_task_context import ( + bind_jm_task_context, + get_current_control, + get_jm_runtime, + get_jm_task_context, + jm_task_context, +) def record_download_duration(context_key: str, clock=None): @@ -79,9 +86,13 @@ def catch_exception(func): @wraps(func) def wrapper(self, *args, **kwargs): self: JmDownloader + self.raise_if_cancelled() try: return func(self, *args, **kwargs) except Exception as e: + if isinstance(e, DownloadCancelledException): + raise + self.raise_if_cancelled() detail: JmBaseEntity = args[0] if detail.is_image(): detail: JmImageDetail @@ -196,9 +207,25 @@ def all_success(self) -> bool: def has_download_failures(self): return len(self.download_failed_image) != 0 or len(self.download_failed_photo) != 0 + def is_cancelled(self) -> bool: + control = get_current_control() + return control is not None and control.is_cancelled + + @classmethod + def raise_if_cancelled(cls) -> None: + """在当前下载作用域已取消时抛出异常，子类可重写该检查点。""" + control = get_current_control() + if control is None or not control.is_cancelled: + return + raise DownloadCancelledException( + control.reason, + {'control': control, 'reason': control.reason}, + ) + # 下面是回调方法 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( @@ -206,18 +233,22 @@ def before_album(self, album: JmAlbumDetail): 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, []) @@ -226,26 +257,32 @@ def before_photo(self, photo: JmPhotoDetail): 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): + self.raise_if_cancelled() super().after_image(image, img_save_path) self.option.call_all_plugin( 'after_image', @@ -255,6 +292,7 @@ def after_image(self, image: JmImageDetail, img_save_path): photo = image.from_photo album = photo.from_album self.download_success_dict.get(album).get(photo).append((image.save_path, image)) + self.raise_if_cancelled() def begin_manifest(self, detail: DetailEntity) -> DownloadManifest: manifest = DownloadManifest() @@ -355,12 +393,17 @@ def _invoke_features_for(self, when: str, **kwargs): 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}]', + 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: @@ -438,7 +481,12 @@ def create_client(self): @record_download_duration('album_started_at') def download_album(self, album_id): - album = self.client.get_album_detail(album_id) + self.raise_if_cancelled() + try: + album = self.client.get_album_detail(album_id) + except Exception: + self.raise_if_cancelled() + raise self.begin_manifest(album) try: self.download_by_album_detail(album) @@ -455,13 +503,18 @@ def download_by_album_detail(self, album: JmAlbumDetail): self.execute_on_condition( iter_objs=album, apply=self.download_by_photo_detail, - count_batch=self.option.decide_photo_batch_count(album) + count_batch=self.option.decide_photo_batch_count(album), ) self.after_album(album) @record_download_duration('photo_started_at') def download_photo(self, photo_id): - photo = self.client.get_photo_detail(photo_id) + self.raise_if_cancelled() + try: + photo = self.client.get_photo_detail(photo_id) + except Exception: + self.raise_if_cancelled() + raise self.begin_manifest(photo) try: self.download_by_photo_detail(photo) @@ -480,7 +533,7 @@ def download_by_photo_detail(self, photo: JmPhotoDetail): self.execute_on_condition( iter_objs=photo, apply=self.download_by_image_detail, - count_batch=self.option.decide_image_batch_count(photo) + count_batch=self.option.decide_image_batch_count(photo), ) self.after_photo(photo) @@ -498,6 +551,7 @@ def download_by_image_detail(self, image: JmImageDetail): if image.cache and image.exists: self.after_image(image, img_save_path) + self.raise_if_cancelled() return decode_image = self.option.decide_download_image_decode(image) @@ -508,36 +562,43 @@ def download_by_image_detail(self, image: JmImageDetail): ) self.after_image(image, img_save_path) + self.raise_if_cancelled() + + def execute_on_condition(self, iter_objs, apply, count_batch): + """使用当前 Runtime 调度本子或章节的下载。""" + runtime = get_jm_runtime() + if runtime is not None and not isinstance(runtime, JmSyncRuntime): + raise TypeError('sync downloader requires JmSyncRuntime') + level = None if runtime is None else ( + 'photo' if iter_objs.is_album() else 'image' + ) - def execute_on_condition(self, - iter_objs: DetailEntity, - apply: Callable, - count_batch: int, - ): - """ - 调度本子/章节的下载 - """ iter_objs = self.do_filter(iter_objs) - count_real = len(iter_objs) - - if count_real == 0: + if len(iter_objs) == 0: return - apply = bind_jm_task_context(apply) + if isinstance(count_batch, bool) or not isinstance(count_batch, int) or count_batch <= 0: + raise ValueError(f'local download limit must be > 0, got {count_batch!r}') - if count_batch >= count_real: - # 一个图/章节 对应 一个线程 - multi_thread_launcher( - iter_objs=iter_objs, - apply_each_obj_func=apply, - ) - else: - # 创建batch个线程的线程池 - thread_pool_executor( - iter_objs=iter_objs, - apply_each_obj_func=apply, - max_workers=count_batch, - ) + worker = bind_jm_task_context(apply) + + if runtime is None: + runtime = JmSimpleRuntime(workers=count_batch) + try: + runtime.multi_thread_launcher( + iter_objs=iter_objs, + apply_each_obj_func=worker, + ) + finally: + runtime.close() + return + + runtime.multi_thread_launcher( + iter_objs=iter_objs, + apply_each_obj_func=worker, + level=level, + default_workers=count_batch, + ) # 下面是对with语法的支持 diff --git a/src/jmcomic/jm_exception.py b/src/jmcomic/jm_exception.py index f88392e44..56c164d39 100644 --- a/src/jmcomic/jm_exception.py +++ b/src/jmcomic/jm_exception.py @@ -94,6 +94,25 @@ def downloader(self): return self.from_context(ExceptionTool.CONTEXT_KEY_DOWNLOADER) +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('reason', msg) + super().__init__(msg, context) + + @property + def control(self): + return self.from_context('control') + + @property + def reason(self) -> str: + return str(self.context.get('reason', 'download cancelled')) + + class ExceptionTool: """ 抛异常的工具 @@ -198,17 +217,17 @@ def raise_missing(cls, ) @classmethod - def require_true(cls, case: bool, msg: str): + def require_true(cls, case: bool, msg: str, etype=None): if case: return - cls.raises(msg) + cls.raises(msg, etype=etype) @classmethod def replace_old_exception_executor(cls, raises: Callable[[Callable, str, dict], None]): old = cls.raises - def new(msg, context=None, _etype=None): + def new(msg, context=None, etype=None): if context is None: context = {} raises(old, msg, context) diff --git a/src/jmcomic/jm_runtime.py b/src/jmcomic/jm_runtime.py new file mode 100644 index 000000000..fc9b664e7 --- /dev/null +++ b/src/jmcomic/jm_runtime.py @@ -0,0 +1,236 @@ +import concurrent.futures +from concurrent.futures import Executor, ProcessPoolExecutor, ThreadPoolExecutor +from threading import RLock +from typing import Callable, Dict, Optional, Set, Tuple + +from common import process_single_arg_to_args_and_kwargs + + +__all__ = ( + 'JmRuntime', + 'JmSimpleRuntime', + 'JmSyncRuntime', + 'JmAsyncRuntime', +) + + +def validate_jm_workers(name: str, workers: int) -> int: + if isinstance(workers, bool) or not isinstance(workers, int) or workers <= 0: + raise ValueError(f'{name} must be a positive integer, got {workers!r}') + return workers + + +def validate_jm_executor(name: str, executor: Executor) -> Executor: + if not isinstance(executor, Executor): + raise TypeError(f'{name} must be a concurrent.futures.Executor') + + rejected_types = [ProcessPoolExecutor] + interpreter_pool = getattr(concurrent.futures, 'InterpreterPoolExecutor', None) + if interpreter_pool is not None: + rejected_types.append(interpreter_pool) + if isinstance(executor, tuple(rejected_types)): + raise TypeError( + f'{name} must be a same-process thread-based Executor; ' + f'{executor.__class__.__name__} is not supported' + ) + return executor + + +def normalize_jm_executor_config( + name: str, + workers: Optional[int], + executor: Optional[Executor], +) -> Tuple[Optional[int], Optional[Executor]]: + if workers is not None and executor is not None: + raise ValueError(f'{name}_workers and {name}_executor are mutually exclusive') + if executor is not None: + executor = validate_jm_executor(f'{name}_executor', executor) + if workers is not None: + workers = validate_jm_workers(f'{name}_workers', workers) + return workers, executor + + +class JmRuntime: + """轻量 Runtime：管理 Executor 的配置、创建、调度和关闭。""" + + _default_level = None + + 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 + + def executor( + self, + level: Optional[str] = None, + default_workers: Optional[int] = None, + ) -> Executor: + """返回指定层级的 Executor；未配置时按调用点给出的默认并发数创建。""" + if level is None: + level = self._default_level + if level is None: + raise TypeError('level is required for this Runtime') + + if default_workers is not None: + default_workers = validate_jm_workers( + f'{level}_default_workers', + default_workers, + ) + + 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: + workers = default_workers + self._workers[level] = workers + + 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, + default_workers: Optional[int] = None, + ): + """使用指定层级的 Executor 批量提交任务，并按需等待完成。""" + executor = self.executor(level, default_workers) + 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)) + except BaseException: + if wait_finish: + concurrent.futures.wait(futures) + raise + + 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) + + +class JmSimpleRuntime(JmRuntime): + """单 Executor Runtime：用于一次同步下载中的局部并发。""" + + _default_level = 'default' + + 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={ + 'default': (workers, executor), + }, + ) + + +class JmSyncRuntime(JmRuntime): + """同步下载 Runtime：分别管理 id、photo、image 三层 Executor。""" + + def __init__( + self, + *, + id_workers: Optional[int] = None, + id_executor: Optional[Executor] = None, + photo_workers: Optional[int] = None, + photo_executor: Optional[Executor] = None, + image_workers: Optional[int] = None, + image_executor: Optional[Executor] = None, + ): + external_executors = [ + executor + for executor in (id_executor, photo_executor, image_executor) + if executor is not None + ] + if len({id(executor) for executor in external_executors}) != len(external_executors): + raise ValueError( + 'id/photo/image levels must use different executor objects' + ) + + super().__init__( + configs={ + 'id': normalize_jm_executor_config('id', id_workers, id_executor), + 'photo': normalize_jm_executor_config( + 'photo', photo_workers, photo_executor + ), + 'image': normalize_jm_executor_config( + 'image', image_workers, image_executor + ), + }, + ) + + +class JmAsyncRuntime(JmRuntime): + """异步下载 Runtime：管理解密、图片处理和同步 hook 的 blocking Executor。""" + + def __init__( + self, + *, + blocking_workers: Optional[int] = None, + blocking_executor: Optional[Executor] = None, + ): + super().__init__( + configs={ + 'blocking': normalize_jm_executor_config( + 'blocking', blocking_workers, blocking_executor + ), + }, + ) diff --git a/src/jmcomic/jm_task_context.py b/src/jmcomic/jm_task_context.py index 57a5e9e1d..8a1d94ad2 100644 --- a/src/jmcomic/jm_task_context.py +++ b/src/jmcomic/jm_task_context.py @@ -2,14 +2,21 @@ from contextlib import contextmanager from contextvars import ContextVar from functools import wraps +from threading import Event, Lock from types import MappingProxyType from typing import Callable, Mapping, Optional +from .jm_runtime import JmRuntime + __all__ = ( 'JM_TASK_CONTEXT', + 'DownloadControl', 'jm_task_context', 'get_jm_task_context', + 'get_current_control', + 'get_jm_runtime', + 'get_current_option', 'bind_jm_task_context', ) @@ -21,16 +28,82 @@ ) +class DownloadControl: + """在线程间共享、线程安全且幂等的下载取消信号。""" + + def __init__(self): + self._event = Event() + self._lock = Lock() + self._reason = 'download cancelled' + + @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) -> str: + with self._lock: + return self._reason + + def get_jm_task_context() -> dict: - """Return a mutable snapshot of the current JM task context.""" + """返回当前 JM 任务上下文的可变快照。""" return dict(JM_TASK_CONTEXT.get()) +def get_current_control() -> Optional[DownloadControl]: + """返回当前 JM 任务上下文中的取消控制器。""" + control = get_jm_task_context().get('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 + + +def get_jm_runtime() -> Optional[JmRuntime]: + """返回当前任务的 Runtime；没有活动 Runtime 时返回 None。""" + return get_jm_task_context().get('runtime') + + +def get_current_option(): + """返回当前任务的 Option；没有活动 Option 时返回 None。""" + return get_jm_task_context().get('option') + + @contextmanager -def jm_task_context(**fields): - """Temporarily add fields to the current JM task context.""" +def jm_task_context(*, option=None, runtime=None, **fields): + """临时绑定任务字段；只传播 Option/Runtime，不管理资源生命周期。""" context = get_jm_task_context() + parent_runtime = context.get('runtime') + + if runtime is not None and not isinstance(runtime, JmRuntime): + raise TypeError('runtime must be JmSyncRuntime, JmAsyncRuntime, or None') + if ( + runtime is not None + and parent_runtime is not None + and runtime is not parent_runtime + ): + raise RuntimeError('another JmRuntime is already active in this task context') + context.update(fields) + if runtime is not None: + context['runtime'] = runtime + if option is not None: + context['option'] = option + token = JM_TASK_CONTEXT.set(MappingProxyType(context)) try: yield @@ -39,9 +112,11 @@ def jm_task_context(**fields): def bind_jm_task_context(func: Callable, context: Optional[Mapping] = None) -> Callable: - """Bind a synchronous callable to a snapshot of the current task context.""" - if (inspect.iscoroutinefunction(func) - or inspect.iscoroutinefunction(getattr(func, '__call__', None))): + """把完整任务上下文快照绑定到同步可调用对象。""" + if ( + inspect.iscoroutinefunction(func) + or inspect.iscoroutinefunction(getattr(func, '__call__', None)) + ): raise TypeError('bind_jm_task_context only supports synchronous callables') snapshot = MappingProxyType(dict( diff --git a/tests/test_jmcomic/test_jm_api.py b/tests/test_jmcomic/test_jm_api.py index bf46f99cc..2cd4b437c 100644 --- a/tests/test_jmcomic/test_jm_api.py +++ b/tests/test_jmcomic/test_jm_api.py @@ -16,6 +16,9 @@ def test_callback_is_not_public_download_api(self): ): parameters = inspect.signature(download_api).parameters self.assertNotIn('callback', parameters) + self.assertNotIn('runtime', parameters) + self.assertNotIn('workers', parameters) + self.assertNotIn('executor', parameters) self.assertEqual( parameters['check_exception'].kind, inspect.Parameter.KEYWORD_ONLY, diff --git a/tests/test_jmcomic/test_jm_async_custom.py b/tests/test_jmcomic/test_jm_async_custom.py index e6cb93677..3c4dc2843 100644 --- a/tests/test_jmcomic/test_jm_async_custom.py +++ b/tests/test_jmcomic/test_jm_async_custom.py @@ -8,6 +8,7 @@ from jmcomic.jm_async_downloader import JmAsyncDownloader from jmcomic.jm_client_interface import AsyncJmcomicClient import asyncio +from unittest.mock import patch class Test_Async_Custom(JmAsyncTestConfigurable): @@ -213,15 +214,23 @@ def test_async_setup_checks_cookies_for_each_session(self): AsyncJmApiClient._has_setup_domain = old_setup_domain def test_async_downloader_cleanup_when_setup_fails(self): - """真实 AsyncSession 初始化失败时，downloader 应回收 client 和线程池""" + """真实会话初始化失败时，也要关闭 Downloader 自建线程池。""" old_auto_update = JmModuleConfig.FLAG_API_CLIENT_AUTO_UPDATE_DOMAIN old_require_cookies = JmModuleConfig.FLAG_API_CLIENT_REQUIRE_COOKIES old_app_cookies = JmModuleConfig.APP_COOKIES old_updated_domains = JmModuleConfig.DOMAIN_API_UPDATED_LIST old_setup_domain = AsyncJmApiClient._has_setup_domain + class ProbeAsyncRuntime(JmAsyncRuntime): + instances = [] + + def __init__(self): + super().__init__(blocking_workers=1) + self.instances.append(self) + loop = asyncio.new_event_loop() downloader = None + runtime = None try: JmModuleConfig.FLAG_API_CLIENT_AUTO_UPDATE_DOMAIN = False JmModuleConfig.FLAG_API_CLIENT_REQUIRE_COOKIES = True @@ -241,14 +250,21 @@ def test_async_downloader_cleanup_when_setup_fails(self): photo_concurrency=1, decode_worker=1, ) - with self.assertRaises(RequestRetryAllFailException): - loop.run_until_complete(downloader.__aenter__()) - + with patch( + 'jmcomic.jm_async_downloader.JmAsyncRuntime', + ProbeAsyncRuntime, + ): + with self.assertRaises(RequestRetryAllFailException): + loop.run_until_complete(downloader.__aenter__()) + + runtime = ProbeAsyncRuntime.instances[0] self.assertIsNone(downloader.client) - self.assertTrue(downloader._decode_pool._shutdown) + self.assertIsNone(get_jm_runtime()) + with self.assertRaisesRegex(RuntimeError, 'JmRuntime is closed'): + runtime.executor('blocking', 1) finally: - if downloader is not None and not downloader._decode_pool._shutdown: - downloader.shutdown() + if runtime is not None: + runtime.close() loop.close() JmModuleConfig.FLAG_API_CLIENT_AUTO_UPDATE_DOMAIN = old_auto_update JmModuleConfig.FLAG_API_CLIENT_REQUIRE_COOKIES = old_require_cookies diff --git a/tests/test_jmcomic/test_jm_cancellation.py b/tests/test_jmcomic/test_jm_cancellation.py new file mode 100644 index 000000000..8f395128e --- /dev/null +++ b/tests/test_jmcomic/test_jm_cancellation.py @@ -0,0 +1,353 @@ +import asyncio +from types import SimpleNamespace +from threading import Thread +import unittest +from unittest.mock import AsyncMock, Mock, patch + +from jmcomic import ( + BaseDownloader, + DownloadCancelledException, + DownloadControl, + JmImageDetail, + JmAsyncDownloader, + JmDownloader, + bind_jm_task_context, + download_album, + download_batch, + download_batch_async, + get_current_control, + jm_task_context, +) + + +class Test_Cancellation(unittest.IsolatedAsyncioTestCase): + + def test_control_is_idempotent_and_visible_across_threads(self): + control = DownloadControl() + seen = [] + + with jm_task_context(control=control): + worker = bind_jm_task_context( + lambda: seen.append(get_current_control()) + ) + thread = Thread(target=worker) + thread.start() + thread.join() + + self.assertEqual(seen, [control]) + self.assertFalse(control.is_cancelled) + self.assertTrue(control.cancel(123)) + self.assertFalse(control.cancel('ignored second reason')) + self.assertTrue(control.is_cancelled) + self.assertEqual(control.reason, '123') + + with jm_task_context(control=control): + with self.assertRaises(DownloadCancelledException) as caught: + BaseDownloader.raise_if_cancelled() + + self.assertIs(caught.exception.control, control) + self.assertEqual(caught.exception.reason, '123') + self.assertEqual( + caught.exception.context, + {'control': control, 'reason': '123'}, + ) + + def test_downloader_classmethod_is_overrideable(self): + BaseDownloader.raise_if_cancelled() + + control = DownloadControl() + control.cancel() + with jm_task_context(control=control): + with self.assertRaises(DownloadCancelledException): + BaseDownloader.raise_if_cancelled() + + seen = [] + + class CustomDownloader(BaseDownloader): + + @classmethod + def raise_if_cancelled(cls) -> None: + seen.append(cls) + + object.__new__(CustomDownloader).raise_if_cancelled() + self.assertEqual(seen, [CustomDownloader]) + + def test_exception_keeps_normal_jmcomic_constructor_shape(self): + control = DownloadControl() + exception = DownloadCancelledException( + 'explicit reason', + {'control': control, 'reason': 'explicit reason'}, + ) + + self.assertIs(exception.control, control) + self.assertEqual(exception.reason, 'explicit reason') + self.assertEqual(str(exception), 'explicit reason') + + def test_sync_api_uses_control_from_caller_context(self): + control = DownloadControl() + seen = [] + + class ProbeDownloader: + + def __init__(self, _option): + self.manifest_dict = {} + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def add_features(self, _extra): + pass + + def download_album(self, _album_id): + seen.append(get_current_control()) + control.cancel('api probe') + BaseDownloader.raise_if_cancelled() + + with jm_task_context(control=control): + with self.assertRaises(DownloadCancelledException): + download_album('123', option=object(), downloader=ProbeDownloader) + + self.assertEqual(seen, [control]) + + def test_sync_batch_raises_cancellation_instead_of_failed_result(self): + control = DownloadControl() + + def cancel_first(jmid, *_args, **_kwargs): + control.cancel(f'stopped at {jmid}') + BaseDownloader.raise_if_cancelled() + + with jm_task_context(control=control): + with self.assertRaises(DownloadCancelledException): + download_batch(cancel_first, ['123', '456'], option=object()) + + def test_sync_batch_propagates_manual_cancellation_without_control(self): + def stop(_jmid, *_args, **_kwargs): + raise DownloadCancelledException('manual stop') + + with self.assertRaisesRegex(DownloadCancelledException, 'manual stop'): + download_batch(stop, ['123'], option=object()) + + async def test_empty_batches_do_not_invoke_downloader_cancellation(self): + control = DownloadControl() + control.cancel('already stopped') + + with jm_task_context(control=control): + self.assertEqual( + download_batch(lambda *_args: None, [], option=object()), + set(), + ) + self.assertEqual( + await download_batch_async(lambda *_args: None, [], option=object()), + set(), + ) + + async def test_async_batch_prefers_control_cancellation_over_task_cancellation(self): + async def stop_differently(jmid, *_args, **_kwargs): + if str(jmid) == '1': + raise asyncio.CancelledError('external cancellation') + raise DownloadCancelledException('control cancellation') + + with self.assertRaisesRegex(DownloadCancelledException, 'control cancellation'): + await download_batch_async(stop_differently, ['1', '2'], option=object()) + + async def test_async_batch_waits_for_cancelled_children_to_finish(self): + ready = asyncio.Event() + never = asyncio.Event() + started = [] + finished = [] + + async def wait_until_cancelled(jmid, *_args, **_kwargs): + started.append(str(jmid)) + if len(started) == 2: + ready.set() + try: + await never.wait() + finally: + await asyncio.sleep(0) + finished.append(str(jmid)) + + batch = asyncio.create_task(download_batch_async( + wait_until_cancelled, + ['1', '2'], + option=object(), + )) + await ready.wait() + batch.cancel() + + with self.assertRaises(asyncio.CancelledError): + await batch + + self.assertCountEqual(finished, ['1', '2']) + + def test_feature_chain_stops_at_cancellation_boundary(self): + control = DownloadControl() + option = Mock() + downloader = BaseDownloader(option) + first = Mock() + second = Mock() + first.should_invoke.return_value = True + second.should_invoke.return_value = True + first.invoke.side_effect = lambda *_args, **_kwargs: control.cancel('feature stop') + downloader._feature_list = [first, second] + + with jm_task_context(download_type='album', control=control): + with self.assertRaises(DownloadCancelledException): + downloader._invoke_features_for('after_album', album=object()) + + first.invoke.assert_called_once() + second.invoke.assert_not_called() + + def test_sync_downloader_converts_client_failure_after_cancellation(self): + control = DownloadControl() + downloader = object.__new__(JmDownloader) + downloader.client = Mock() + + def fail_after_cancel(_album_id): + control.cancel('请求期间取消') + raise RuntimeError('请求失败') + + downloader.client.get_album_detail.side_effect = fail_after_cancel + + with jm_task_context(control=control): + with self.assertRaisesRegex(DownloadCancelledException, '请求期间取消'): + downloader.download_album('123') + + async def test_async_downloader_converts_client_failure_after_cancellation(self): + control = DownloadControl() + downloader = object.__new__(JmAsyncDownloader) + downloader.client = Mock() + + async def fail_after_cancel(_album_id): + control.cancel('异步请求期间取消') + raise RuntimeError('请求失败') + + downloader.client.get_album_detail = fail_after_cancel + + with jm_task_context(control=control): + with self.assertRaisesRegex(DownloadCancelledException, '异步请求期间取消'): + await downloader.download_album('123') + + async def test_control_cancellation_is_not_recorded_as_image_failure(self): + control = DownloadControl() + control.cancel('stop before image') + downloader = object.__new__(JmAsyncDownloader) + downloader.download_failed_image = [] + + async def cancelled(_image): + downloader.raise_if_cancelled() + + downloader._download_single_image = cancelled + + with jm_task_context(control=control): + with self.assertRaises(DownloadCancelledException): + await downloader._safe_download_image(object()) + + self.assertEqual(downloader.download_failed_image, []) + + async def test_external_task_cancellation_is_not_swallowed(self): + downloader = object.__new__(JmAsyncDownloader) + + async def wait_forever(_image): + await asyncio.Event().wait() + + downloader._download_single_image = wait_forever + task = asyncio.create_task(downloader._safe_download_image(object())) + await asyncio.sleep(0) + task.cancel() + + with self.assertRaises(asyncio.CancelledError): + await task + + async def test_current_async_image_is_recorded_before_control_cancellation(self): + control = DownloadControl() + downloader = object.__new__(JmAsyncDownloader) + downloader.option = Mock() + downloader.client = Mock() + downloader.client.get_jm_image = AsyncMock() + downloader.client.get_jm_image.return_value.content = b'image' + downloader._image_semaphore = asyncio.Semaphore(1) + downloader._run_in_decode_pool = AsyncMock() + downloader.option.decide_image_filepath.return_value = 'image.jpg' + downloader.option.decide_download_cache.return_value = False + downloader.option.decide_download_image_decode.return_value = False + + image = Mock(spec=JmImageDetail) + image.skip = False + image.scramble_id = None + image.download_url = 'https://example.invalid/image.jpg' + recorded = [] + + async def after_image(current_image, _path): + recorded.append(current_image) + control.cancel('stop after current image') + + downloader.before_image = AsyncMock() + downloader.after_image = after_image + + with patch('jmcomic.jm_async_downloader.os.path.exists', return_value=False): + with jm_task_context(control=control): + with self.assertRaises(DownloadCancelledException): + await downloader._download_single_image(image) + + self.assertEqual(recorded, [image]) + downloader._run_in_decode_pool.assert_awaited_once() + + async def test_image_cancelled_while_waiting_for_semaphore_never_requests(self): + control = DownloadControl() + downloader = object.__new__(JmAsyncDownloader) + downloader.option = Mock() + downloader.client = Mock() + downloader.client.get_jm_image = AsyncMock() + downloader._image_semaphore = asyncio.Semaphore(1) + await downloader._image_semaphore.acquire() + downloader.option.decide_image_filepath.return_value = 'image.jpg' + downloader.option.decide_download_cache.return_value = False + downloader.option.decide_download_image_decode.return_value = False + before_reached = asyncio.Event() + + async def before_image(_image, _path): + before_reached.set() + + downloader.before_image = before_image + downloader.after_image = AsyncMock() + image = SimpleNamespace( + skip=False, + scramble_id=None, + download_url='https://example.invalid/image.jpg', + ) + + with patch('jmcomic.jm_async_downloader.os.path.exists', return_value=False): + with jm_task_context(control=control): + task = asyncio.create_task(downloader._download_single_image(image)) + await asyncio.wait_for(before_reached.wait(), timeout=1) + await asyncio.sleep(0) + control.cancel('queued image') + downloader._image_semaphore.release() + with self.assertRaises(DownloadCancelledException): + await task + + downloader.client.get_jm_image.assert_not_awaited() + + async def test_photo_cancelled_while_waiting_for_semaphore_never_requests(self): + control = DownloadControl() + downloader = object.__new__(JmAsyncDownloader) + downloader.option = Mock() + downloader.option.decide_image_save_dir.return_value = 'photo' + downloader.client = Mock() + downloader.client.check_photo = AsyncMock() + downloader._photo_semaphore = asyncio.Semaphore(1) + await downloader._photo_semaphore.acquire() + photo = SimpleNamespace() + + with jm_task_context(control=control): + task = asyncio.create_task(downloader.download_by_photo_detail(photo)) + await asyncio.sleep(0) + control.cancel('queued photo') + downloader._photo_semaphore.release() + with self.assertRaises(DownloadCancelledException): + await task + + downloader.client.check_photo.assert_not_awaited() diff --git a/tests/test_jmcomic/test_jm_cli.py b/tests/test_jmcomic/test_jm_cli.py index d20f16c1d..9dd1beb58 100644 --- a/tests/test_jmcomic/test_jm_cli.py +++ b/tests/test_jmcomic/test_jm_cli.py @@ -1,9 +1,12 @@ +from concurrent.futures import ThreadPoolExecutor + from test_jmcomic import * from io import StringIO from types import SimpleNamespace from unittest.mock import MagicMock, patch from jmcomic.cli import JmcomicUI, JmViewUI +from jmcomic import JmSyncRuntime, get_jm_runtime, jm_task_context from jmcomic.jm_task_context import get_jm_task_context @@ -107,6 +110,44 @@ def test_jmcomic_parse_mixed(self): self.assertEqual(ui.album_id_list, [self.album_id]) self.assertEqual(ui.photo_id_list, [self.album_id]) + def test_jmcomic_mixed_threads_inherit_runtime(self): + ui = JmcomicUI() + ui.album_id_list = ['1'] + ui.photo_id_list = ['2'] + option = SimpleNamespace() + observed = [] + + def record(ids, actual_option): + observed.append(( + ids, + actual_option, + get_jm_runtime(), + get_jm_task_context(), + )) + + with ThreadPoolExecutor(max_workers=1) as executor: + runtime = JmSyncRuntime(id_executor=executor) + with patch('jmcomic.api.download_album', side_effect=record), \ + patch('jmcomic.api.download_photo', side_effect=record): + with jm_task_context(task_id='cli-job', runtime=runtime): + ui.run(option) + + self.assertCountEqual( + [(ids, actual_option) for ids, actual_option, _executor, _context in observed], + [(['1'], option), (['2'], option)], + ) + self.assertEqual( + [seen for _ids, _option, seen, _context in observed], + [runtime, runtime], + ) + self.assertEqual( + [context for _ids, _option, _executor, context in observed], + [ + {'task_id': 'cli-job', 'runtime': runtime}, + {'task_id': 'cli-job', 'runtime': runtime}, + ], + ) + def test_jmcomic_download_album(self): """jmcomic 真实下载 album 350234""" JustDownloadSpecificCountImage.count = 5 diff --git a/tests/test_jmcomic/test_jm_client.py b/tests/test_jmcomic/test_jm_client.py index aefe69467..85abb6d89 100644 --- a/tests/test_jmcomic/test_jm_client.py +++ b/tests/test_jmcomic/test_jm_client.py @@ -1,11 +1,42 @@ import asyncio from types import SimpleNamespace +from unittest.mock import Mock from test_jmcomic import * class Test_Client(JmTestConfigurable): + def test_add_favorite_album_uses_ajax_form_post(self): + client = object.__new__(JmHtmlClient) + response = SimpleNamespace( + status_code=200, + url='https://example.com/ajax/favorite_album', + redirect_count=0, + text='{"status": 1}', + json=lambda: {'status': 1}, + ) + client.post = Mock(return_value=response) + + result = JmHtmlClient.add_favorite_album(client, 21, 123) + + self.assertIs(result, response) + client.post.assert_called_once_with( + '/ajax/favorite_album', + data={ + 'album_id': '21', + 'fid': '123', + }, + headers={ + 'accept': 'application/json, text/javascript, */*; q=0.01', + 'content-type': 'application/x-www-form-urlencoded; charset=UTF-8', + 'sec-fetch-dest': 'empty', + 'sec-fetch-mode': 'cors', + 'sec-fetch-site': 'same-origin', + 'x-requested-with': 'XMLHttpRequest', + }, + ) + def test_download_image(self): jm_photo_id = 'JM438516' photo = self.client.get_photo_detail(jm_photo_id) diff --git a/tests/test_jmcomic/test_jm_download_manifest.py b/tests/test_jmcomic/test_jm_download_manifest.py index 48037c20b..cdb156d04 100644 --- a/tests/test_jmcomic/test_jm_download_manifest.py +++ b/tests/test_jmcomic/test_jm_download_manifest.py @@ -59,7 +59,9 @@ def decide_image_save_dir(self, _photo): return os.path.join(self.base_dir, 'album', 'photo') def decide_image_filepath(self, image): - return os.path.join(self.decide_image_save_dir(image.from_photo), image.filename) + save_dir = self.decide_image_save_dir(image.from_photo) + os.makedirs(save_dir, exist_ok=True) + return os.path.join(save_dir, image.filename) def decide_download_cache(self, _image): return True @@ -143,6 +145,10 @@ async def get_photo_detail(self, _photo_id): async def check_photo(self, _photo): return None + async def get_jm_image(self, _url): + self.image_download_count += 1 + return SimpleNamespace(content=b'image') + class ContractAsyncDownloader(JmAsyncDownloader): @@ -163,6 +169,53 @@ def do_filter(self, detail): class Test_Download_Manifest(unittest.TestCase): + def test_current_sync_image_is_recorded_before_cancellation(self): + with TemporaryDirectory() as temp_dir: + album, photo, image_list = new_album_photo_images() + option = ContractOption(temp_dir) + downloader = ContractSyncDownloader(option, album, photo, image_list) + control = DownloadControl() + option.after_image_callback = lambda image: control.cancel('stop after image') + + with jm_task_context(control=control): + with self.assertRaises(DownloadCancelledException): + downloader.download_album(album.id) + + image = image_list[0] + self.assertTrue(os.path.isfile(image.save_path)) + self.assertEqual( + downloader.download_success_dict[album][photo], + [(image.save_path, image)], + ) + self.assertEqual(downloader.download_failed_image, []) + + def test_current_async_image_is_recorded_before_cancellation(self): + async def run_test(temp_dir): + album, photo, image_list = new_album_photo_images() + option = ContractOption(temp_dir) + downloader = ContractAsyncDownloader(option, album, photo, image_list) + control = DownloadControl() + option.after_image_callback = lambda image: control.cancel('stop after image') + os.makedirs(option.decide_image_save_dir(photo), exist_ok=True) + + runtime = JmAsyncRuntime() + try: + with jm_task_context(control=control, runtime=runtime): + with self.assertRaises(DownloadCancelledException): + await downloader.download_album(album.id) + finally: + runtime.close() + + image = image_list[0] + self.assertEqual( + downloader.download_success_dict[album][photo], + [(image.save_path, image)], + ) + self.assertEqual(downloader.download_failed_image, []) + + with TemporaryDirectory() as temp_dir: + asyncio.run(run_test(temp_dir)) + def test_downloadable_defaults(self): album, photo, image_list = new_album_photo_images() @@ -1032,21 +1085,23 @@ async def run_test(temp_dir): f.write(b'cached') downloader = ContractAsyncDownloader(option, album, photo, image_list) + runtime = JmAsyncRuntime() try: - await downloader.download_album(album.id) - - self.assertEqual(album.save_path, option.dir_rule.decide_album_root_dir(album)) - self.assertEqual(photo.save_path, option.decide_image_save_dir(photo)) - self.assertEqual(image.save_path, filepath) - self.assertIsInstance(album.duration, float) - self.assertIsInstance(photo.duration, float) - self.assertIsInstance(image.duration, float) - after_image_events = [event for event, _ in option.plugin_event_list if event == 'after_image'] - self.assertEqual(after_image_events, ['after_image']) - self.assertEqual(downloader.download_success_dict[album][photo], [(filepath, image)]) - self.assertEqual(downloader.manifest_dict[album].image_filepath_list, [filepath]) + with jm_task_context(runtime=runtime): + await downloader.download_album(album.id) finally: - downloader.shutdown() + runtime.close() + + self.assertEqual(album.save_path, option.dir_rule.decide_album_root_dir(album)) + self.assertEqual(photo.save_path, option.decide_image_save_dir(photo)) + self.assertEqual(image.save_path, filepath) + self.assertIsInstance(album.duration, float) + self.assertIsInstance(photo.duration, float) + self.assertIsInstance(image.duration, float) + after_image_events = [event for event, _ in option.plugin_event_list if event == 'after_image'] + self.assertEqual(after_image_events, ['after_image']) + self.assertEqual(downloader.download_success_dict[album][photo], [(filepath, image)]) + self.assertEqual(downloader.manifest_dict[album].image_filepath_list, [filepath]) with TemporaryDirectory() as temp_dir: asyncio.run(run_test(temp_dir)) diff --git a/tests/test_jmcomic/test_jm_download_progress.py b/tests/test_jmcomic/test_jm_download_progress.py index 57b151e4d..6e81b30de 100644 --- a/tests/test_jmcomic/test_jm_download_progress.py +++ b/tests/test_jmcomic/test_jm_download_progress.py @@ -3,7 +3,6 @@ import importlib.util import logging import os -import time import unittest from io import StringIO from pathlib import Path @@ -28,10 +27,6 @@ from jmcomic.jm_task_context import jm_task_context -PROJECT_DIR = Path(__file__).resolve().parents[2] -DOCUMENT_FILE = PROJECT_DIR / 'assets' / 'docs' / 'sources' / 'tutorial' / '15_download_progress.md' -PLUGIN_FILE = PROJECT_DIR / 'src' / 'jmcomic' / 'jm_plugin.py' -DOWNLOADER_FILE = PROJECT_DIR / 'src' / 'jmcomic' / 'jm_downloader.py' RICH_INSTALLED = importlib.util.find_spec('rich') is not None @@ -159,7 +154,7 @@ class AfterAsyncDownloader(JmAsyncDownloader): JmModuleConfig.CLASS_ASYNC_DOWNLOADER = original_downloader @unittest.skipUnless(RICH_INSTALLED, '需要安装 rich 才能测试彩色进度插件') - def test_sync_progress_is_rendered_before_download_finishes(self): + def test_sync_progress_updates_before_download_finishes(self): from rich.console import Console album = create_album() @@ -186,19 +181,24 @@ def test_sync_progress_is_rendered_before_download_finishes(self): ): downloader = ProgressDownloader(option) downloader.before_album(album) - rendered_before_wait = ui_output.getvalue() - time.sleep(0.2) - self.assertEqual(rendered_before_wait, ui_output.getvalue()) downloader.before_photo(photo) image.save_path = 'mock.jpg' downloader.after_image(image, image.save_path) - rendered_during_download = ui_output.getvalue() - self.assertIn('本子-JM123456', rendered_during_download) - self.assertIn('章节-JM101', rendered_during_download) - self.assertIn('1/2', rendered_during_download) - self.assertNotIn('✓ 本子-JM123456', rendered_during_download) + album_task = next( + task for task in downloader.progress.tasks + if '本子-JM123456' in task.description + ) + chapter_task = next( + task for task in downloader.progress.tasks + if '章节-JM101' in task.description + ) + self.assertEqual(0, album_task.completed) + self.assertEqual(1, chapter_task.completed) + self.assertEqual(2, chapter_task.total) + self.assertFalse(album_task.finished) + self.assertFalse(chapter_task.finished) finally: if downloader is not None: downloader.stop_progress() @@ -470,7 +470,7 @@ def fake_download_image(downloader, image): self.assertIn('检测到命令行参数 --no-progress', rendered) self.assertIn('当前 Option 已配置 download_progress', rendered) self.assertIn('JMComic Logs', rendered) - self.assertIn('album.before', rendered) + self.assertIn('album.after', rendered) self.assertIn('章节-JM101', rendered) self.assertIn('章节-JM102', rendered) self.assertIn('✓ 本子-JM123456', rendered) @@ -548,18 +548,11 @@ async def fake_download_image(downloader, image): self.assertIn('详细日志', rendered) self.assertIn('✓ 下载完成：本子-JM123456，章节 2/2，图片 5/5', rendered) - def test_plugin_is_registered_and_documentation_only_shows_usage(self): - document = DOCUMENT_FILE.read_text(encoding='utf-8') + def test_plugin_is_registered(self): self.assertIs( DownloadProgressPlugin, JmModuleConfig.REGISTRY_PLUGIN['download_progress'], ) - self.assertIn('plugin: download_progress', document) - self.assertIn('download_album_async', document) - self.assertNotIn('class ProgressDownloader', document) - self.assertNotIn('class DownloadProgressPlugin', document) - self.assertIn('class ProgressDownloader', PLUGIN_FILE.read_text(encoding='utf-8')) - self.assertNotIn('class ProgressDownloader', DOWNLOADER_FILE.read_text(encoding='utf-8')) if __name__ == '__main__': diff --git a/tests/test_jmcomic/test_jm_exception.py b/tests/test_jmcomic/test_jm_exception.py index cc4450a1f..4d2e2834f 100644 --- a/tests/test_jmcomic/test_jm_exception.py +++ b/tests/test_jmcomic/test_jm_exception.py @@ -4,6 +4,20 @@ class Test_RequestRetryAllFailException(unittest.TestCase): + def test_require_true_keeps_replaced_exception_executor_compatible(self): + class IsolatedExceptionTool(ExceptionTool): + pass + + seen = [] + IsolatedExceptionTool.replace_old_exception_executor( + lambda old, msg, context: seen.append((old, msg, context)) + ) + + IsolatedExceptionTool.require_true(False, 'compatibility probe') + + self.assertEqual(len(seen), 1) + self.assertEqual(seen[0][1:], ('compatibility probe', {})) + def test_sync_client_collects_each_failed_request(self): client = object.__new__(AbstractJmClient) client.domain_list = ['api-one.example', 'api-two.example'] diff --git a/tests/test_jmcomic/test_jm_favorite.py b/tests/test_jmcomic/test_jm_favorite.py new file mode 100644 index 000000000..134656e09 --- /dev/null +++ b/tests/test_jmcomic/test_jm_favorite.py @@ -0,0 +1,39 @@ +import asyncio +import unittest +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +from jmcomic import AsyncJmApiClient, JmApiClient + + +class Test_Favorite(unittest.TestCase): + + def test_api_add_favorite_album_uses_post(self): + client = object.__new__(JmApiClient) + response = SimpleNamespace(model_data=SimpleNamespace(status='ok')) + client.req_api = Mock(return_value=response) + client.require_resp_status_ok = Mock() + + result = JmApiClient.add_favorite_album(client, 21) + + self.assertIs(result, response) + client.req_api.assert_called_once_with( + client.API_FAVORITE, + get=False, + data={'aid': 21}, + ) + client.require_resp_status_ok.assert_called_once_with(response) + + def test_async_api_add_favorite_album_uses_post(self): + client = object.__new__(AsyncJmApiClient) + response = SimpleNamespace(model_data=SimpleNamespace(status='ok')) + client.req_api = AsyncMock(return_value=response) + + result = asyncio.run(AsyncJmApiClient.add_favorite_album(client, 21)) + + self.assertIs(result, response) + client.req_api.assert_awaited_once_with( + client.API_FAVORITE, + get=False, + data={'aid': 21}, + ) diff --git a/tests/test_jmcomic/test_jm_release.py b/tests/test_jmcomic/test_jm_release.py index 500f8d522..b46b2b678 100644 --- a/tests/test_jmcomic/test_jm_release.py +++ b/tests/test_jmcomic/test_jm_release.py @@ -114,45 +114,14 @@ def test_release_entry_count_uses_changelog_bullets(self): self.assertEqual(release.count_release_entries(body), 3) - def test_workflow_keeps_master_v_prefix_trigger_without_generated_notes(self): - workflow = (PROJECT_ROOT / ".github" / "workflows" / "release_auto.yml").read_text(encoding="utf-8") - - self.assertIn("branches:\n - master", workflow) - self.assertIn("startsWith(github.event.head_commit.message, 'v')", workflow) - self.assertIn('python .github/release.py "$commit_message"', workflow) - self.assertNotIn("generate_release_notes:", workflow) - - def test_manual_workflow_reads_source_version_from_master(self): - workflow = (PROJECT_ROOT / ".github" / "workflows" / "release.yml").read_text(encoding="utf-8") - - self.assertIn("workflow_dispatch:", workflow) - self.assertNotIn("types: [ published ]", workflow) - self.assertIn("github.ref_name == 'master'", workflow) - self.assertIn("python .github/release.py\n", workflow) - self.assertIn("softprops/action-gh-release@v2", workflow) - self.assertIn("pypa/gh-action-pypi-publish@release/v1", workflow) - - def test_release_workflows_build_before_creating_release(self): - for filename in ("release.yml", "release_auto.yml"): - with self.subTest(filename=filename): - workflow = (PROJECT_ROOT / ".github" / "workflows" / filename).read_text(encoding="utf-8") - - self.assertLess(workflow.index("- name: Build\n"), workflow.index("- name: Create Release\n")) - - def test_test_workflows_watch_development_requirements(self): - for filename in ("test_api.yml", "test_html.yml"): - with self.subTest(filename=filename): - workflow = (PROJECT_ROOT / ".github" / "workflows" / filename).read_text(encoding="utf-8") - - self.assertIn(" - '.github/requirements-dev.txt'", workflow) - - def test_contributing_allows_only_formal_release_prs_to_master(self): - contributing = (PROJECT_ROOT / ".github" / "CONTRIBUTING.md").read_text(encoding="utf-8") - - self.assertIn("普通 PR 禁止直飞 master", contributing) - self.assertIn("发版专线 (仅限版本发布)", contributing) - self.assertIn("任意一项缺失，都不得指向或合并到 `master`", contributing) - self.assertNotIn("本项目不接受任何直接指向 `master` 分支的 PR", contributing) + def test_python_39_remains_install_compatible(self): + pyproject = (PROJECT_ROOT / "pyproject.toml").read_text(encoding="utf-8") + setup_py = (PROJECT_ROOT / "setup.py").read_text(encoding="utf-8") + + self.assertIn('requires-python = ">=3.9"', pyproject) + self.assertIn('python_requires=">=3.9"', setup_py) + self.assertIn("Programming Language :: Python :: 3.9", pyproject) + self.assertIn("Programming Language :: Python :: 3.9", setup_py) if __name__ == "__main__": diff --git a/tests/test_jmcomic/test_jm_runtime.py b/tests/test_jmcomic/test_jm_runtime.py new file mode 100644 index 000000000..6851be639 --- /dev/null +++ b/tests/test_jmcomic/test_jm_runtime.py @@ -0,0 +1,267 @@ +import asyncio +import threading +import time +import unittest +from concurrent.futures import ThreadPoolExecutor + +from jmcomic import ( + DownloadCancelledException, + JmAsyncRuntime, + JmSimpleRuntime, + JmSyncRuntime, + download_batch, + download_batch_async, + get_jm_runtime, + jm_task_context, +) + + +class Test_Jm_Runtime(unittest.TestCase): + + def test_launcher_expands_arguments_and_returns_done_futures(self): + runtime = JmSyncRuntime(id_workers=2) + try: + positional = runtime.multi_thread_launcher( + [(1, 2), (3, 4)], + lambda left, right: left + right, + level='id', + ) + keyword = runtime.multi_thread_launcher( + [{'left': 5, 'right': 6}], + lambda left, right: left + right, + level='id', + ) + + self.assertEqual([3, 7], [future.result() for future in positional]) + self.assertEqual([11], [future.result() for future in keyword]) + self.assertTrue(all(future.done() for future in positional + keyword)) + finally: + runtime.close() + + def test_launcher_can_return_without_waiting(self): + started = threading.Event() + release = threading.Event() + runtime = JmSyncRuntime(id_workers=1) + + def blocked(_item): + started.set() + release.wait(timeout=2) + + try: + futures = runtime.multi_thread_launcher( + [1], + blocked, + False, + level='id', + ) + self.assertTrue(started.wait(timeout=1)) + self.assertFalse(futures[0].done()) + finally: + release.set() + runtime.close() + + def test_launcher_waits_for_all_tasks_and_keeps_worker_error_in_future(self): + sibling_started = threading.Event() + release_sibling = threading.Event() + launcher_done = threading.Event() + futures = [] + runtime = JmSyncRuntime(id_workers=2) + + def work(item): + if item == 'error': + raise ValueError('worker failed') + sibling_started.set() + release_sibling.wait(timeout=2) + + def launch(): + try: + futures.extend(runtime.multi_thread_launcher( + ['error', 'sibling'], + work, + level='id', + )) + finally: + launcher_done.set() + + thread = threading.Thread(target=launch) + thread.start() + try: + self.assertTrue(sibling_started.wait(timeout=1)) + self.assertFalse(launcher_done.is_set()) + release_sibling.set() + self.assertTrue(launcher_done.wait(timeout=1)) + self.assertEqual(2, len(futures)) + with self.assertRaisesRegex(ValueError, 'worker failed'): + futures[0].result() + self.assertIsNone(futures[1].result()) + finally: + release_sibling.set() + thread.join(timeout=1) + runtime.close() + + def test_launcher_waits_for_submitted_tasks_when_submit_fails(self): + first_started = threading.Event() + release_first = threading.Event() + launcher_done = threading.Event() + raised = [] + + class FailSecondSubmitExecutor(ThreadPoolExecutor): + + def __init__(self): + super().__init__(max_workers=1) + self.submit_count = 0 + + def submit(self, fn, /, *args, **kwargs): + self.submit_count += 1 + if self.submit_count == 2: + raise RuntimeError('submit failed') + return super().submit(fn, *args, **kwargs) + + executor = FailSecondSubmitExecutor() + runtime = JmSyncRuntime(id_executor=executor) + + def blocked(_item): + first_started.set() + release_first.wait(timeout=2) + + def launch(): + try: + runtime.multi_thread_launcher( + [1, 2], + blocked, + level='id', + ) + except BaseException as error: + raised.append(error) + finally: + launcher_done.set() + + thread = threading.Thread(target=launch) + thread.start() + try: + self.assertTrue(first_started.wait(timeout=1)) + self.assertFalse(launcher_done.is_set()) + release_first.set() + self.assertTrue(launcher_done.wait(timeout=1)) + self.assertEqual(1, len(raised)) + self.assertIsInstance(raised[0], RuntimeError) + self.assertEqual('submit failed', str(raised[0])) + finally: + release_first.set() + thread.join(timeout=1) + runtime.close() + executor.shutdown(wait=True) + + def test_runtime_does_not_close_external_executor(self): + executor = ThreadPoolExecutor(max_workers=1) + try: + runtime = JmSyncRuntime(id_executor=executor) + futures = runtime.multi_thread_launcher( + [1, 2], + lambda item: item * 2, + level='id', + ) + self.assertEqual([2, 4], [future.result() for future in futures]) + + runtime.close() + self.assertEqual(9, executor.submit(lambda: 9).result(timeout=1)) + finally: + executor.shutdown(wait=True) + + def test_simple_runtime_uses_one_executor_without_level(self): + executor = ThreadPoolExecutor(max_workers=1) + try: + runtime = JmSimpleRuntime(executor=executor) + futures = runtime.multi_thread_launcher( + [1, 2], + lambda item: item * 2, + ) + self.assertEqual([2, 4], [future.result() for future in futures]) + + runtime.close() + self.assertEqual(9, executor.submit(lambda: 9).result(timeout=1)) + finally: + executor.shutdown(wait=True) + + def test_sync_batch_explicitly_closes_runtime_created_by_api(self): + observed = [] + + def download_one(jmid, *_args, **_kwargs): + observed.append(get_jm_runtime()) + return jmid + + self.assertEqual( + {'1'}, + set(download_batch(download_one, ['1'], option=object())), + ) + self.assertEqual(1, len(observed)) + with self.assertRaisesRegex(RuntimeError, 'JmRuntime is closed'): + observed[0].executor('id', 1) + + def test_async_batch_explicitly_closes_runtime_created_by_api(self): + observed = [] + + async def download_one(jmid, *_args, **_kwargs): + observed.append(get_jm_runtime()) + return jmid + + result = asyncio.run(download_batch_async( + download_one, + ['1'], + option=object(), + )) + + self.assertEqual({'1'}, set(result)) + self.assertEqual(1, len(observed)) + self.assertIsInstance(observed[0], JmAsyncRuntime) + with self.assertRaisesRegex(RuntimeError, 'JmRuntime is closed'): + observed[0].executor('blocking', 1) + + def test_sync_batch_collects_failure_after_siblings_finish(self): + completed = [] + + def download_one(jmid, *_args, **_kwargs): + if jmid == '404': + raise ValueError('missing') + time.sleep(0.02) + completed.append(jmid) + return jmid + + result = download_batch( + download_one, + ['200', '404'], + option=object(), + ) + + self.assertEqual({'200'}, set(result)) + self.assertEqual(['200'], completed) + self.assertIsInstance(result.failed['404'], ValueError) + + def test_sync_batch_waits_for_siblings_before_raising_cancellation(self): + barrier = threading.Barrier(2) + completed = [] + + def download_one(jmid, *_args, **_kwargs): + barrier.wait(timeout=1) + if jmid == '1': + raise DownloadCancelledException('stop') + time.sleep(0.02) + completed.append(jmid) + return jmid + + with ThreadPoolExecutor(max_workers=2) as executor: + runtime = JmSyncRuntime(id_executor=executor) + with jm_task_context(runtime=runtime): + with self.assertRaisesRegex(DownloadCancelledException, 'stop'): + download_batch( + download_one, + ['1', '2'], + option=object(), + ) + + self.assertEqual(['2'], completed) + self.assertEqual(9, executor.submit(lambda: 9).result(timeout=1)) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_jmcomic/test_jm_task_context.py b/tests/test_jmcomic/test_jm_task_context.py index 89d585958..a5cea3111 100644 --- a/tests/test_jmcomic/test_jm_task_context.py +++ b/tests/test_jmcomic/test_jm_task_context.py @@ -1,6 +1,7 @@ import asyncio import logging import threading +import time import unittest from concurrent.futures import ThreadPoolExecutor from queue import Queue @@ -8,7 +9,9 @@ from jmcomic import ( BaseDownloader, + JmAsyncRuntime, DownloadManifest, + DownloadControl, Feature, JmAsyncDownloader, JmDownloader, @@ -18,12 +21,17 @@ JM_TASK_CONTEXT, PhotoConcurrentFetcherProxy, PrettyFormatter, + JmSimpleRuntime, + JmSyncRuntime, bind_jm_task_context, default_jm_logging, download_album, download_batch, download_batch_async, download_photo_async, + get_current_control, + get_current_option, + get_jm_runtime, get_jm_task_context, jm_log, jm_task_context, @@ -41,8 +49,203 @@ def emit(self, record): self.records.append(record) +class ExecutorProbeDownloader(JmDownloader): + + def __init__(self): + BaseDownloader.__init__(self, object()) + self.work = lambda _item: None + + def download_by_image_detail(self, image): + self.work(image) + + +class PhotoDetails(list): + + @staticmethod + def is_album(): + return False + + class Test_Jm_Task_Context(unittest.TestCase): + def test_context_exposes_runtime_and_borrows_executor(self): + with ThreadPoolExecutor(max_workers=1) as executor: + runtime = JmSyncRuntime(id_executor=executor) + with jm_task_context(task_id='job-42', runtime=runtime): + self.assertEqual( + get_jm_task_context(), + {'task_id': 'job-42', 'runtime': runtime}, + ) + raw_context = JM_TASK_CONTEXT.get() + self.assertEqual(raw_context['task_id'], 'job-42') + self.assertIs(raw_context['runtime'], runtime) + with self.assertRaises(TypeError): + raw_context['mutable'] = True + self.assertIs(get_jm_runtime(), runtime) + with jm_task_context(jm_id='123'): + self.assertEqual( + get_jm_task_context(), + { + 'task_id': 'job-42', + 'runtime': runtime, + 'jm_id': '123', + }, + ) + + public_bound = executor.submit(bind_jm_task_context( + lambda: ( + get_jm_task_context(), + get_jm_runtime(), + ) + )) + self.assertEqual( + public_bound.result(timeout=1), + ({'task_id': 'job-42', 'runtime': runtime}, runtime), + ) + + self.assertEqual(executor.submit(lambda: 7).result(timeout=1), 7) + + def test_option_is_public_and_propagated(self): + option = object() + with ThreadPoolExecutor(max_workers=1) as executor: + with jm_task_context(option=option, task_id='job-42'): + self.assertEqual( + get_jm_task_context(), + {'task_id': 'job-42', 'option': option}, + ) + self.assertIs(get_current_option(), option) + future = executor.submit(bind_jm_task_context(get_current_option)) + + self.assertIs(future.result(timeout=1), option) + + self.assertIsNone(get_current_option()) + + def test_none_option_runtime_and_nested_runtime_rules(self): + with jm_task_context(option=None, runtime=None, task_id='job-42'): + self.assertEqual(get_jm_task_context(), {'task_id': 'job-42'}) + self.assertIsNone(get_current_option()) + self.assertIsNone(get_jm_runtime()) + + outer = JmSyncRuntime(id_workers=1) + inner = JmSyncRuntime(id_workers=1) + try: + with jm_task_context(runtime=outer): + with jm_task_context(runtime=outer): + self.assertIs(get_jm_runtime(), outer) + with self.assertRaisesRegex(RuntimeError, 'already active'): + with jm_task_context(runtime=inner): + self.fail('nested runtime must be rejected') + finally: + outer.close() + inner.close() + + runtime = JmSyncRuntime(id_workers=1) + with jm_task_context(runtime=runtime): + pass + futures = runtime.multi_thread_launcher( + [1], + lambda item: item, + level='id', + ) + self.assertEqual(1, futures[0].result()) + runtime.close() + with self.assertRaisesRegex(RuntimeError, 'JmRuntime is closed'): + runtime.multi_thread_launcher( + [1], + lambda item: item, + level='id', + ) + + def test_runtime_configuration_is_explicit_and_validated(self): + with self.assertRaisesRegex(TypeError, 'runtime must be'): + with jm_task_context(runtime='metadata'): + self.fail('runtime must be a Runtime instance') + + with ThreadPoolExecutor(max_workers=1) as executor: + with self.assertRaisesRegex(ValueError, 'mutually exclusive'): + JmSyncRuntime(id_workers=1, id_executor=executor) + with self.assertRaisesRegex(ValueError, 'different executor'): + JmSyncRuntime(id_executor=executor, photo_executor=executor) + with self.assertRaisesRegex(ValueError, 'mutually exclusive'): + JmAsyncRuntime(blocking_workers=1, blocking_executor=executor) + with self.assertRaisesRegex(ValueError, 'mutually exclusive'): + JmSimpleRuntime(workers=1, executor=executor) + + with self.assertRaisesRegex(ValueError, 'positive integer'): + JmSyncRuntime(photo_workers=0) + with self.assertRaisesRegex(ValueError, 'positive integer'): + JmAsyncRuntime(blocking_workers=True) + with self.assertRaisesRegex(ValueError, 'positive integer'): + JmSimpleRuntime(workers=0) + + def test_sync_runtime_reuses_one_id_pool_across_batches(self): + worker_threads = [] + + def fake_download(jmid, _option, _downloader, **_kwargs): + worker_threads.append(threading.get_ident()) + return str(jmid) + + runtime = JmSyncRuntime(id_workers=1, photo_workers=1, image_workers=1) + try: + with jm_task_context(runtime=runtime): + self.assertEqual( + {'1', '2'}, + set(download_batch(fake_download, ['1', '2'], option=object())), + ) + self.assertEqual( + {'3'}, + set(download_batch(fake_download, ['3'], option=object())), + ) + + self.assertEqual(1, len(set(worker_threads))) + finally: + runtime.close() + + def test_sync_runtime_accepts_call_site_worker_defaults(self): + with ThreadPoolExecutor(max_workers=1) as id_executor: + runtime = JmSyncRuntime(id_executor=id_executor) + try: + with jm_task_context(runtime=runtime): + photo_futures = runtime.multi_thread_launcher( + [1, 2], + lambda item: item * 2, + level='photo', + default_workers=2, + ) + self.assertEqual( + [future.result() for future in photo_futures], + [2, 4], + ) + image_futures = runtime.multi_thread_launcher( + [1], + lambda item: item + 1, + level='image', + default_workers=4, + ) + self.assertEqual( + [future.result() for future in image_futures], + [2], + ) + finally: + runtime.close() + + self.assertEqual(id_executor.submit(lambda: 5).result(timeout=1), 5) + + def test_download_control_is_propagated_by_public_binder(self): + control = DownloadControl() + + with ThreadPoolExecutor(max_workers=1) as executor: + with jm_task_context(control=control): + future = executor.submit(bind_jm_task_context(get_current_control)) + + self.assertIs(future.result(timeout=1), control) + self.assertIsNone(executor.submit(get_current_control).result(timeout=1)) + + self.assertTrue(control.cancel('user requested')) + self.assertFalse(control.cancel('ignored')) + self.assertTrue(control.is_cancelled) + self.assertEqual(control.reason, 'user requested') + def test_public_context_var_names_log_record_field(self): self.assertEqual('jm_task_context', JM_TASK_CONTEXT.name) @@ -267,10 +470,17 @@ async def download_photo(self, _photo_id): ) self.assertIsInstance(album_context.get('task_started_at'), float) self.assertIsInstance(photo_context.get('task_started_at'), float) + self.assertIsInstance(album_context['runtime'], JmSyncRuntime) + with self.assertRaisesRegex(RuntimeError, 'JmRuntime is closed'): + album_context['runtime'].executor('photo', 1) + with self.assertRaisesRegex(RuntimeError, 'JmRuntime is closed'): + photo_context['runtime'].executor('blocking', 1) def test_sync_result_duration_uses_task_context_and_finishes_after_downloader_exit(self): clock = {'now': 10.0} observed_contexts = [] + observed_options = [] + option = object() class Detail: duration = 99.0 @@ -292,18 +502,21 @@ def download_album(self, _album_id): self.begin_manifest(detail) try: observed_contexts.append(get_jm_task_context()) + observed_options.append(get_current_option()) finally: self.finish_manifest(detail) return detail with patch('jmcomic.api.perf_counter', side_effect=lambda: clock['now'], create=True): - result = download_album('123', option=object(), downloader=FakeDownloader) + result = download_album('123', option=option, downloader=FakeDownloader) self.assertEqual(10.0, observed_contexts[0].get('task_started_at')) self.assertEqual(10.0, result.manifest.duration) self.assertEqual(10.0, result.duration) self.assertEqual(99.0, result.detail.duration) + self.assertEqual([option], observed_options) self.assertEqual({}, get_jm_task_context()) + self.assertIsNone(get_current_option()) def test_async_result_duration_uses_task_context_and_finishes_after_downloader_exit(self): async def run_test(): @@ -497,9 +710,10 @@ def test_batch_failure_log_keeps_item_context(self): def fail(jmid, _option, _downloader, **_kwargs): raise ValueError(f'failed-{jmid}') + option = object() try: with jm_task_context(session_id='failed-session'): - result = download_batch(fail, ['404'], option=object()) + result = download_batch(fail, ['404'], option=option) finally: jm_logger.handlers[:] = original_handlers @@ -510,36 +724,52 @@ def fail(jmid, _option, _downloader, **_kwargs): if getattr(record, 'topic', None) == 'batch.failed' ] self.assertEqual(1, len(batch_failure_records)) + log_context = batch_failure_records[0].jm_task_context self.assertEqual( { 'session_id': 'failed-session', 'download_type': 'fail', 'jm_id': '404', }, - batch_failure_records[0].jm_task_context, + { + key: log_context[key] + for key in ('session_id', 'download_type', 'jm_id') + }, ) + self.assertIs(log_context['option'], option) + self.assertIsInstance(log_context['runtime'], JmSyncRuntime) - def test_sync_downloader_propagates_both_threading_branches(self): + def test_sync_downloader_uses_and_closes_temporary_simple_runtime(self): downloader = object.__new__(JmDownloader) BaseDownloader.__init__(downloader, object()) + created = [] + + class ProbeRuntime(JmSimpleRuntime): + + def __init__(self, **kwargs): + super().__init__(**kwargs) + created.append(self) for count_batch in (3, 1): observed = Queue() - with jm_task_context(session_id=f'workers-{count_batch}'): - downloader.execute_on_condition( - iter_objs=[1, 2, 3], - apply=lambda _item: observed.put(get_jm_task_context()), - count_batch=count_batch, - ) + with patch('jmcomic.jm_downloader.JmSimpleRuntime', ProbeRuntime): + with jm_task_context(session_id=f'workers-{count_batch}'): + downloader.execute_on_condition( + iter_objs=[1, 2, 3], + apply=lambda _item: observed.put(get_jm_task_context()), + count_batch=count_batch, + ) contexts = [observed.get_nowait() for _ in range(3)] self.assertEqual( - [{ - 'session_id': f'workers-{count_batch}', - }] * 3, - contexts, + [f'workers-{count_batch}'] * 3, + [context['session_id'] for context in contexts], ) + self.assertTrue(all('runtime' not in context for context in contexts)) + self.assertTrue(all('option' not in context for context in contexts)) + with self.assertRaisesRegex(RuntimeError, 'JmRuntime is closed'): + created[-1].executor() def test_async_batch_and_decode_pool_propagate_context(self): async def run_test(): @@ -560,21 +790,23 @@ async def fake_download(jmid, _option, _downloader, **_kwargs): option=object(), ) - downloader = object.__new__(JmAsyncDownloader) - downloader._decode_pool = ThreadPoolExecutor(max_workers=1) - try: + with ThreadPoolExecutor(max_workers=1) as blocking_executor: + runtime = JmAsyncRuntime(blocking_executor=blocking_executor) with jm_task_context(session_id='decode-pool'): - executor_context = await downloader._run_in_decode_pool( - get_jm_task_context - ) + with jm_task_context(runtime=runtime): + future = runtime.executor( + 'blocking', + 1, + ).submit( + bind_jm_task_context(get_jm_task_context), + ) + executor_context = await asyncio.wrap_future(future) loop = asyncio.get_running_loop() leaked_context = await loop.run_in_executor( - downloader._decode_pool, + blocking_executor, get_jm_task_context, ) - finally: - downloader.shutdown() return batch_result, executor_context, leaked_context @@ -587,7 +819,8 @@ async def fake_download(jmid, _option, _downloader, **_kwargs): }, set(batch_result), ) - self.assertEqual({'session_id': 'decode-pool'}, executor_context) + self.assertEqual('decode-pool', executor_context['session_id']) + self.assertIsInstance(executor_context['runtime'], JmAsyncRuntime) self.assertEqual({}, leaked_context) def test_photo_concurrent_proxy_propagates_context(self): @@ -603,6 +836,56 @@ class FakeClient: finally: proxy.executors.shutdown(wait=True) + def test_sync_runtime_executor_capacity_controls_image_concurrency(self): + downloader = ExecutorProbeDownloader() + lock = threading.Lock() + active = 0 + maximum = 0 + + def limited_work(_item): + nonlocal active, maximum + with lock: + active += 1 + maximum = max(maximum, active) + time.sleep(0.02) + with lock: + active -= 1 + + downloader.work = limited_work + downloader.do_filter = lambda details: list(details) + with ThreadPoolExecutor(max_workers=2) as executor: + runtime = JmSyncRuntime(image_executor=executor) + with jm_task_context(runtime=runtime): + downloader.execute_on_condition( + PhotoDetails(range(8)), + downloader.download_by_image_detail, + count_batch=8, + ) + runtime.close() + self.assertEqual(executor.submit(lambda: 1).result(timeout=1), 1) + + self.assertEqual(maximum, 2) + + def test_async_runtime_borrows_shared_executor(self): + async def scenario(): + with ThreadPoolExecutor(max_workers=1) as executor: + runtime = JmAsyncRuntime(blocking_executor=executor) + with jm_task_context(session_id='async-runtime', runtime=runtime): + future = runtime.executor( + 'blocking', + 1, + ).submit( + bind_jm_task_context( + lambda: (8, get_jm_task_context()['session_id']) + ), + ) + value = await asyncio.wrap_future(future) + + self.assertEqual(value, (8, 'async-runtime')) + self.assertEqual(executor.submit(lambda: 9).result(timeout=1), 9) + + asyncio.run(scenario()) + if __name__ == '__main__': unittest.main()