From 1c878a029fa7523cedad40b2a59af65ad2225316 Mon Sep 17 00:00:00 2001 From: hect0x7 <93357912+hect0x7@users.noreply.github.com> Date: Thu, 10 Sep 2026 00:47:25 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=94=AF=E6=8C=81=E5=8D=8F=E4=BD=9C?= =?UTF-8?q?=E5=BC=8F=E5=8F=96=E6=B6=88=E4=B8=8B=E8=BD=BD=E4=B8=8E=E5=85=B1?= =?UTF-8?q?=E4=BA=AB=20Runtime=20=E8=B0=83=E5=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .github/workflows/benchmark.yml | 6 +- .github/workflows/close_specific_pr.yml | 4 +- .github/workflows/download.yml | 6 +- .github/workflows/download_dispatch.yml | 8 +- .github/workflows/export_favorites.yml | 8 +- .github/workflows/release.yml | 8 +- .github/workflows/release_auto.yml | 8 +- .github/workflows/test_api.yml | 4 +- .github/workflows/test_html.yml | 4 +- CHANGELOG.md | 16 + assets/docs/mkdocs.yml | 1 + assets/docs/sources/api/download.md | 34 ++ .../docs/sources/tutorial/0_common_usage.md | 67 ++- .../docs/sources/tutorial/14_async_usage.md | 76 +++- .../sources/tutorial/16_shared_executors.md | 152 +++++++ pyproject.toml | 2 +- src/jmcomic/__init__.py | 4 +- src/jmcomic/api.py | 332 ++++++++++---- src/jmcomic/cli.py | 5 +- src/jmcomic/jm_async_downloader.py | 114 ++++- src/jmcomic/jm_client_impl.py | 6 + src/jmcomic/jm_config.py | 4 +- src/jmcomic/jm_downloader.py | 147 ++++-- src/jmcomic/jm_exception.py | 33 +- src/jmcomic/jm_feature.py | 4 +- src/jmcomic/jm_plugin.py | 6 +- src/jmcomic/jm_runtime.py | 246 ++++++++++ src/jmcomic/jm_task_context.py | 119 ++++- tests/test_jmcomic/test_jm_api.py | 3 + tests/test_jmcomic/test_jm_async_custom.py | 30 +- tests/test_jmcomic/test_jm_cancellation.py | 422 ++++++++++++++++++ tests/test_jmcomic/test_jm_cli.py | 46 +- tests/test_jmcomic/test_jm_client.py | 10 +- .../test_jmcomic/test_jm_download_manifest.py | 194 ++++++-- .../test_jmcomic/test_jm_download_progress.py | 39 +- tests/test_jmcomic/test_jm_exception.py | 44 ++ tests/test_jmcomic/test_jm_release.py | 47 +- tests/test_jmcomic/test_jm_runtime.py | 267 +++++++++++ tests/test_jmcomic/test_jm_task_context.py | 401 +++++++++++++++-- 39 files changed, 2530 insertions(+), 397 deletions(-) create mode 100644 assets/docs/sources/tutorial/16_shared_executors.md create mode 100644 src/jmcomic/jm_runtime.py create mode 100644 tests/test_jmcomic/test_jm_cancellation.py create mode 100644 tests/test_jmcomic/test_jm_runtime.py 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..897920702 100644 --- a/.github/workflows/download_dispatch.yml +++ b/.github/workflows/download_dispatch.yml @@ -108,9 +108,11 @@ jobs: JM_DOWNLOAD_DIR: /home/runner/work/jmcomic/download/ steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v7 + with: + persist-credentials: false - name: Set up Python 3.11 - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: "3.11" @@ -153,7 +155,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..4375e3bc7 100644 --- a/.github/workflows/export_favorites.yml +++ b/.github/workflows/export_favorites.yml @@ -48,9 +48,11 @@ jobs: ZIP_FP: /home/runner/work/jmcomic/download/export.7z steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v7 + with: + persist-credentials: false - name: Set up Python 3.11 - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: "3.11" @@ -71,7 +73,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..9458b5856 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -13,10 +13,12 @@ jobs: id-token: write contents: write steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v7 + with: + persist-credentials: false - name: Set up Python 3.11 - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: "3.11" @@ -31,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/release_auto.yml b/.github/workflows/release_auto.yml index 0ef2d6cb1..c3829d119 100644 --- a/.github/workflows/release_auto.yml +++ b/.github/workflows/release_auto.yml @@ -14,10 +14,12 @@ jobs: contents: write if: startsWith(github.event.head_commit.message, 'v') steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v7 + with: + persist-credentials: false - name: Set up Python 3.11 - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: "3.11" @@ -33,7 +35,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 a9f87541e..164a51489 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,22 @@ 条目分类参考 [Keep a Changelog](https://keepachangelog.com/zh-CN/1.0.0/)， 版本号遵循 [语义化版本](https://semver.org/lang/zh-CN/)。 +## [2.7.7] - 2026-09-10 + +### Summary + +本次更新支持协作式取消下载与共享 Runtime 调度。 + +### Added +- 新增 `JmSimpleRuntime`、`JmSyncRuntime` 与 `JmAsyncRuntime`；裸同步 Downloader 的局部调度使用单池 Runtime，顶层同步 API 或自定义同步调度可复用 `id/photo/image` 三层线程池，异步下载可复用 `decode` 线程池。 +- 新增 `DownloadControl` 和 `DownloadCancelledException`，支持通过任务上下文协作式取消同步与异步下载。 +- 新增 `JTC` (Jm Task Context) 门面类，统一提供上下文、Runtime、Option 及取消控制器的快捷读取方法。 + +### Changed +- Python 3.9 保留安装兼容，但不再纳入 CI。 +- 下载调度统一使用标准库 Executor；顶层 API 和裸同步 Downloader 的临时调度显式关闭自己创建的 Runtime，`jm_task_context` 只传播字段；外部 Runtime 和 Executor 仍由调用方关闭。 +- 顶层下载会把 Runtime 和 Option 作为公开字段直接放入任务上下文；Runtime 不依赖 Context 或 Option，未配置的层级由实际调用点传入默认 worker 数。 + ## [2.7.6] - 2026-09-09 ### 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..ee37905ba 100644 --- a/assets/docs/sources/api/download.md +++ b/assets/docs/sources/api/download.md @@ -17,3 +17,37 @@ options: members: - JmAsyncDownloader + +::: jmcomic.jm_downloader + options: + members: + - BaseDownloader + - JmDownloader + +::: 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 + +::: jmcomic.jm_task_context.JTC + options: + members: + - get_runtime + - get_option + - get_control + - get_context diff --git a/assets/docs/sources/tutorial/0_common_usage.md b/assets/docs/sources/tutorial/0_common_usage.md index 0da5a4428..2df2eedec 100644 --- a/assets/docs/sources/tutorial/0_common_usage.md +++ b/assets/docs/sources/tutorial/0_common_usage.md @@ -455,11 +455,61 @@ cl = JmApiClient( ``` +## 取消下载 + +> 如果你想中途取消下载，最简单粗暴的方式是直接杀死进程，比如 `ctrl+c`，关闭终端窗口 等 +> +> 但是在 GUI、Web服务这类场景里，就不适合这么关闭。 +> +> 于是，jmcomic 提供了一种优雅停止的写法，可以使用 `DownloadControl` 这个类。 +> +> 这种写法更加可控： +> +> - **不影响主程序**：只停当前下载，GUI 界面或 Web 服务依然正常运作。 +> - **多任务统一取消**：多个本子一起下载时，可以被统一取消。 +> - **可携带取消理由**：调用 `cancel()` 传入的原因可以直接捕获，方便做提示和日志。 + +写法就两步： + +1. 创建 DownloadControl 对象，传入任务上下文 +2. 调用 DownloadControl 类的 cancel() 方法 + +```python +from threading import Thread +from time import sleep +from jmcomic import DownloadCancelledException, DownloadControl, download_album, jm_task_context + +# 创建取消控制器 +my_control = DownloadControl() + +def run_download(): + try: + # 使用 with jm_task_context 创建任务上下文，并传入 control 参数 + with jm_task_context(control=my_control): + # 这里还可以写多个 download_album(xxx)，都会统一被取消，因为属于同一个 任务上下文 + download_album('123456') + except DownloadCancelledException as e: + print(f'下载已取消: {e.reason}') + +# 使用单独的下载线程执行下载 +# 主线程负责取消 +t = Thread(target=run_download) +t.start() + +# 模拟一段时间后，需要取消下载 +sleep(2) +# 可以传入取消原因，下载线程可通过上面的 e.reason 获取 +my_control.cancel("不想要了，取消掉吧") +# 等待下载线程结束 +t.join() +``` + + ## 下载返回值 `download_album` 和 `download_photo` 下载完成后，单个 ID 返回 `DownloadResult`，多个 ID 返回 `BatchResult`。 -从 `result.detail` 可以取得下载的本子/章节的实体类： +从 `DownloadResult` 的 `detail` 字段可以取得下载的本子/章节的实体类： ```python from jmcomic import download_album, download_photo @@ -559,24 +609,26 @@ print('长图导出文件:', png_filepath_list) ### 批量下载的返回值 -传入多个 ID 时，返回值是 `BatchResult`。每一项成功下载对应一个 `DownloadResult`，失败任务则记录在 `failed` 中： +传入多个 ID 时，返回值是 `BatchResult`，通过这个对象可以取得成功和失败的下载结果：
完整示例：处理批量下载结果 ```python -from jmcomic import download_album +from jmcomic import download_album, DownloadResult # 同时下载多个本子 batch_result = download_album(['123', '456', '789']) -# BatchResult 继承 set，成功结果没有输入顺序保证 +# BatchResult 继承 set，直接遍历是只遍历成功结果 for result in batch_result: + result: DownloadResult + # result 的用法同上 album = result.detail - # 通过实体 ID 识别当前结果，不要用遍历位置对应输入列表 print(f'JM{album.id} 下载到: {album.save_path}') -# failed 的键是下载失败的 ID，值是记录失败原因的异常对象 +# failed 负责存放失败的下载结果，类型是dict +# key是下载失败的 ID，value是记录失败原因的异常对象 for album_id, error in batch_result.failed.items(): print(f'JM{album_id} 下载失败: {error}') @@ -587,7 +639,8 @@ print('是否全部成功:', batch_result.all_succeeded)
-下载单个 ID 时，请求本子失败会直接抛出异常；如果只有部分章节或图片失败，会在任务结束后汇总抛出 `PartialDownloadFailedException`，此时不会返回 `DownloadResult`。批量下载则继续执行其他任务，并把失败项集中放进 `batch_result.failed`。 +--- + ### 速查表 diff --git a/assets/docs/sources/tutorial/14_async_usage.md b/assets/docs/sources/tutorial/14_async_usage.md index 5d1342526..08c776341 100644 --- a/assets/docs/sources/tutorial/14_async_usage.md +++ b/assets/docs/sources/tutorial/14_async_usage.md @@ -234,9 +234,9 @@ client: async_impl: async_api ``` -## 9. 查看下载耗时 +## 9. 下载返回值 -异步下载完成后，可以直接查看自己总共等了多久，也可以继续查看具体是哪个本子、章节或图片比较慢。所有 `duration` 的单位都是秒。 +用法与同步版本完全一致，返回的 `DownloadResult` 包含了本子实体、保存路径、耗时以及下载清单（`manifest`）等专属字段： ```python import asyncio @@ -247,25 +247,67 @@ async def main(): result = await jmcomic.download_album_async('438696') album = result.detail - # 从调用 download_album_async 到返回，总共等了多久 - print(f'总共等待: {result.duration:.3f} 秒') + # 1. 实体与保存路径 + print(f'本子: JM{album.id} - {album.title}') + print(f'保存目录: {album.save_path}') - # 如果下载比较慢，可以继续查看具体慢在哪里 - print(f'下载本子用了: {album.duration:.3f} 秒') - for photo in album: - print(f'下载章节 {photo.id} 用了: {photo.duration:.3f} 秒') - for image in photo: - print(f'处理图片 {image.img_file_name} 用了: {image.duration:.3f} 秒') + # 2. 耗时统计（单位：秒） + print(f'总耗时: {result.duration:.3f} 秒, 本子处理耗时: {album.duration:.3f} 秒') + + # 3. 下载清单 manifest + manifest = result.manifest + print(f'全部本子图片的磁盘路径: {manifest.image_filepath_list}') asyncio.run(main()) ``` -| 字段 | 它告诉你什么 | -| --- | --- | -| `result.duration` | 从调用异步下载方法到返回，你总共等了多久 | -| `album.duration` | 下载这个本子花了多久，包含获取本子信息和整理下载结果 | -| `photo.duration` | 下载这个章节花了多久，包含获取或补全章节信息 | -| `image.duration` | 处理这张图片花了多久，包含检查缓存、下载、解密和保存 | +## 10. 取消下载 + +下面示例代码模拟了一个 GUI 场景中，有两个按钮，一个开始下载，一个取消下载，基于 `DownloadControl` 实现。 + +```python +import asyncio +from jmcomic import DownloadCancelledException, DownloadControl, download_album_async, jm_task_context + +# 界面持有的当前下载控制器与任务 +current_control = None +current_task = None + + +# 按钮 1：点击【开始下载】 +def on_btn_download_click(album_id: str) -> asyncio.Task: + global current_control, current_task + current_control = DownloadControl() + + async def _run(): + try: + with jm_task_context(control=current_control): + await download_album_async(album_id) + except DownloadCancelledException as e: + print(f'[UI 提示] 下载已取消: {e.reason}') + + current_task = asyncio.create_task(_run()) + return current_task -下载器可能同时处理多个章节或多张图片，所以把它们的耗时全部相加，不会得到本子的耗时，这是正常现象。同步下载中的这些字段含义相同。 + +# 按钮 2：点击【取消下载】 +def on_btn_cancel_click(): + if current_control: + current_control.cancel("用户点击了界面的取消按钮") + + +async def main(): + # 模拟用户点击【开始下载】 + task = on_btn_download_click('123456') + + # 模拟用户在界面等待一段时间后，点击了【取消下载】 + await asyncio.sleep(2) + on_btn_cancel_click() + + # 等待下载任务响应取消并安全收尾 + await task + + +asyncio.run(main()) +``` 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..c4153f698 --- /dev/null +++ b/assets/docs/sources/tutorial/16_shared_executors.md @@ -0,0 +1,152 @@ +# 复用下载 Runtime 与共享线程池 + +> 日常使用 JMComic 下载时，通常不需要手动配置线程池，顶层下载 API 会自动管理并在任务结束后安全释放。 +> +> 但在以下两种场景下，你可以使用 `Runtime` 来精细控制： +> +> 1. **批量下载时防卡死**：一次性下载几十上百个本子，希望严格限制并发（例如“最多同时下 2 个本子，所有图片最多 8 个线程”），避免把带宽或电脑卡爆。 +> 2. **复用已有的线程池**：你的程序（如 Web 服务、后台调度器）本身已经维护了全局线程池，希望 JMComic 直接复用，避免重复创建和销毁线程。 + +--- + +## 什么是 Runtime？ + +平时我们直接调用 `download_album('123456')` 时，完全不需要传任何 Runtime 参数。 + +这是因为 JMComic 已经在后台自动创建了一个默认的 Runtime（下载运行时），负责拉起线程池，并在下载结束后自动释放。 + +**Runtime 的定位很简单：它就是专门负责管理下载过程中“线程池与并发调度”的管家。** + +当你需要**自定义线程数**，或者希望**复用已有线程池**时，就需要显式创建一个 Runtime，并通过 `jm_task_context` 传递给下载任务： + +```python +from jmcomic import JmSyncRuntime, download_album, jm_task_context + +# 1. 创建你定制的 Runtime +runtime = JmSyncRuntime(...) + +# 2. 通过 jm_task_context 注入给下载任务 +with jm_task_context(runtime=runtime): + download_album(...) +``` + +接下来分别介绍同步与异步的具体用法。 + +--- + +## 1. 同步下载：使用 JmSyncRuntime + +同步下载时，任务分为三个层级：本子、章节、图片。`JmSyncRuntime` 允许你分别控制这三层的并发线程数。 + +### 场景 A：自定义各层并发数 + +通过 `JmSyncRuntime` 指定每一层的最大线程数，并通过 `jm_task_context` 传入下载任务： + +```python +from jmcomic import JmSyncRuntime, download_album, jm_task_context + +# 1. 创建 Runtime，指定各层并发上限 +runtime = JmSyncRuntime( + id_workers=2, # 最多同时下载 2 个本子 + photo_workers=3, # 所有本子合计最多同时处理 3 个章节 + image_workers=8, # 所有章节合计最多同时下载 8 张图片 +) + +try: + # 2. 绑定到任务上下文并执行下载 + with jm_task_context(runtime=runtime): + download_album(['123456', '789012', '345678']) +finally: + # 3. 任务完成后释放线程池 + runtime.close() +``` + +> [!TIP] +> 如果某一层不传参数（例如不传 `photo_workers`），该层会使用系统默认值。 + +### 场景 B：复用外部已有的线程池 + +如果你的程序已经有现成的 `ThreadPoolExecutor`，可以直接传给 Runtime 复用： + +```python +from concurrent.futures import ThreadPoolExecutor +from jmcomic import JmSyncRuntime, download_album, jm_task_context + +# 假设项目中已有维护好的线程池 +with ThreadPoolExecutor(max_workers=2) as id_pool, \ + ThreadPoolExecutor(max_workers=3) as photo_pool, \ + ThreadPoolExecutor(max_workers=8) as image_pool: + + runtime = JmSyncRuntime( + id_executor=id_pool, + photo_executor=photo_pool, + image_executor=image_pool, + ) + try: + with jm_task_context(runtime=runtime): + download_album(['123456', '789012']) + finally: + # 释放 Runtime。遵循“谁创建谁关闭”原则，外部传入的线程池不会被关闭 + runtime.close() +``` + +--- + +## 2. 异步下载：使用 JmAsyncRuntime + +异步下载与同步不同：网络请求完全由 `asyncio` 协程高效处理，不占线程；**只有图片的解密、反混淆拼接与写盘（CPU 计算与文件 I/O）需要用到后台线程池**。 + +因此，`JmAsyncRuntime` 只需要管理一个 `decode`（图片解码）线程池，配置更加简单。 + +### 场景 A：指定解码并发线程数 + +```python +import asyncio +from jmcomic import JmAsyncRuntime, download_album_async, jm_task_context + + +async def main(): + # 限制最多同时使用 4 个线程进行图片解密与保存 + runtime = JmAsyncRuntime(decode_workers=4) + try: + with jm_task_context(runtime=runtime): + await download_album_async(['123456', '789012']) + finally: + runtime.close() + + +asyncio.run(main()) +``` + +### 场景 B：复用外部已有的解码线程池 + +```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 my_decode_pool: + runtime = JmAsyncRuntime(decode_executor=my_decode_pool) + try: + with jm_task_context(runtime=runtime): + await download_album_async('123456') + finally: + runtime.close() + + +asyncio.run(main()) +``` + +--- + +## 3. 常见问题与速查 + +| 需求 | 推荐做法 | 说明 | +| :--- | :--- | :--- | +| **同步批量控制并发** | `JmSyncRuntime(id_workers=..., image_workers=...)` | 避免一次性下几十个本子把资源占满 | +| **异步控制解密并发** | `JmAsyncRuntime(decode_workers=...)` | 仅限制图片解密写盘线程数，网络请求仍走协程 | +| **复用已有线程池** | 传入 `*_executor=你的线程池` | 遵循“谁创建谁关闭”，`runtime.close()` 不会关闭外部线程池 | +| **自省当前 Runtime** | `JTC.get_runtime()` | 在插件或上下文内部随时自省当前生效的 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 25d6c7b18..12e66a95c 100644 --- a/src/jmcomic/__init__.py +++ b/src/jmcomic/__init__.py @@ -2,8 +2,10 @@ # 被依赖方 <--- 使用方 # config <--- entity <--- toolkit <--- client <--- option <--- downloader -__version__ = '2.7.6' +__version__ = '2.7.7' +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..e5f3c3a4d 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, + JTC, + 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 = JTC.get_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(...))`` 提供的 decode + 执行器由调用方负责关闭。 - - 支持批量下载（当 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(...))`` 提供的 decode 执行器由调用方负责关闭。 + + ``check_exception`` 只对单个 ID 生效。批量场景应检查 + ``BatchResult.failed``，或按需封装 ``download_batch_async``。 """ if not isinstance(jm_photo_id, (str, int)): return await download_batch_async(download_photo_async, @@ -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_downloader.py b/src/jmcomic/jm_async_downloader.py index 3f89e24e0..f277eeb18 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, + JTC, + 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,37 @@ 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 = JTC.get_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('decode', 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 (asyncio.CancelledError, DownloadCancelledException): + pass + except BaseException as drain_error: + jm_log('dler.cancel.drain.exception', drain_error) + raise @record_download_duration('album_started_at') async def download_album(self, album_id) -> JmAlbumDetail: """对齐 sync JmDownloader.download_album""" - 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 +143,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 +154,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 +183,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 +207,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 +221,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 +245,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 +285,7 @@ async def _download_single_image(self, image: JmImageDetail): ) await self.after_image(image, img_save_path) + self.raise_if_cancelled() # ====================================================================== # 磁盘写入（在线程池中执行） @@ -278,17 +339,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 = JTC.get_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 +373,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 31d8fbeec..659a25cb0 100644 --- a/src/jmcomic/jm_client_impl.py +++ b/src/jmcomic/jm_client_impl.py @@ -274,6 +274,12 @@ def add_favorite_album(self, }, ) + 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: diff --git a/src/jmcomic/jm_config.py b/src/jmcomic/jm_config.py index 0ac46a38a..dd070a9ca 100644 --- a/src/jmcomic/jm_config.py +++ b/src/jmcomic/jm_config.py @@ -4,7 +4,7 @@ from common import time_stamp, field_cache, ProxyBuilder -from .jm_task_context import JM_TASK_CONTEXT, get_jm_task_context +from .jm_task_context import JM_TASK_CONTEXT, JTC jm_logger = logging.getLogger('jmcomic') @@ -68,7 +68,7 @@ def default_jm_logging(topic: str, msg, e: BaseException | None = None): msg = str(msg) extra = { 'topic': topic, - JM_TASK_CONTEXT.name: get_jm_task_context(), + JM_TASK_CONTEXT.name: JTC.get_context(), } if e is not None: jm_logger.error(msg, extra=extra, exc_info=e) diff --git a/src/jmcomic/jm_downloader.py b/src/jmcomic/jm_downloader.py index 2846e547b..86a26b07b 100644 --- a/src/jmcomic/jm_downloader.py +++ b/src/jmcomic/jm_downloader.py @@ -5,7 +5,12 @@ 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, + JTC, + jm_task_context, +) def record_download_duration(context_key: str, clock=None): @@ -28,7 +33,7 @@ async def async_wrapper(*args, **kwargs): entity = get_entity(args, kwargs) detail_call = isinstance(entity, Downloadable) # 顶层 ID 下载负责完整耗时，内部 detail 调用复用同一个计时上下文。 - if detail_call and get_jm_task_context().get(context_key) is not None: + if detail_call and JTC.get_context().get(context_key) is not None: return await func(*args, **kwargs) started_at = get_time() @@ -46,7 +51,7 @@ def wrapper(*args, **kwargs): entity = get_entity(args, kwargs) detail_call = isinstance(entity, Downloadable) # 顶层 ID 下载负责完整耗时，内部 detail 调用复用同一个计时上下文。 - if detail_call and get_jm_task_context().get(context_key) is not None: + if detail_call and JTC.get_context().get(context_key) is not None: return func(*args, **kwargs) started_at = get_time() @@ -79,9 +84,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 +205,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 = JTC.get_control() + return control is not None and control.is_cancelled + + @classmethod + def raise_if_cancelled(cls) -> None: + """在当前下载作用域已取消时抛出异常，子类可重写该检查点。""" + control = JTC.get_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 +231,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,35 +255,50 @@ 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): - super().after_image(image, img_save_path) - self.option.call_all_plugin( - 'after_image', - image=image, - downloader=self, - ) + cancellation_error = None + try: + self.raise_if_cancelled() + super().after_image(image, img_save_path) + self.option.call_all_plugin( + 'after_image', + image=image, + downloader=self, + ) + except DownloadCancelledException as error: + cancellation_error = error + + # 正常完成或取消时登记；普通插件异常直接向外传播，不进入成功清单。 photo = image.from_photo album = photo.from_album self.download_success_dict.get(album).get(photo).append((image.save_path, image)) + if cancellation_error is not None: + raise cancellation_error + self.raise_if_cancelled() def begin_manifest(self, detail: DetailEntity) -> DownloadManifest: manifest = DownloadManifest() @@ -310,7 +354,7 @@ def finish_manifest(self, detail: DetailEntity) -> DownloadManifest: def _require_feature_context() -> str: from .jm_toolkit import ExceptionTool - download_type = get_jm_task_context().get('download_type') + download_type = JTC.get_context().get('download_type') ExceptionTool.require_true( download_type in ('album', 'photo'), 'Feature 注册与执行必须位于下载任务上下文中，请使用 ' @@ -355,12 +399,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 +487,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 +509,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 +539,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 +557,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 +568,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 = JTC.get_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..7f1ba4f16 100644 --- a/src/jmcomic/jm_exception.py +++ b/src/jmcomic/jm_exception.py @@ -94,6 +94,27 @@ 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,20 +219,24 @@ 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) + + def raise_original(msg, context=None, etype=etype): + return old(msg, context, etype) + + raises(raise_original, msg, context) cls.raises = new diff --git a/src/jmcomic/jm_feature.py b/src/jmcomic/jm_feature.py index d93bf7dac..b4614dc65 100644 --- a/src/jmcomic/jm_feature.py +++ b/src/jmcomic/jm_feature.py @@ -17,7 +17,7 @@ download_album(id, option, extra=Feature.export_pdf + Feature.export_zip) """ from .jm_plugin import * -from .jm_task_context import get_jm_task_context +from .jm_task_context import JTC class Feature: @@ -37,7 +37,7 @@ class Feature: @property def jm_task_context(self) -> dict: """Return the current invocation's isolated task-context snapshot.""" - return get_jm_task_context() + return JTC.get_context() def should_invoke(self, when: str) -> bool: """ diff --git a/src/jmcomic/jm_plugin.py b/src/jmcomic/jm_plugin.py index 203036776..9726b475d 100644 --- a/src/jmcomic/jm_plugin.py +++ b/src/jmcomic/jm_plugin.py @@ -8,7 +8,7 @@ from .jm_option import * from .jm_async_downloader import JmAsyncDownloader from .jm_downloader import JmDownloader -from .jm_task_context import bind_jm_task_context, get_jm_task_context +from .jm_task_context import bind_jm_task_context, JTC class PluginValidationException(Exception): @@ -36,7 +36,7 @@ def invoke(self, **kwargs) -> None: @property def jm_task_context(self) -> dict: """Return the current invocation's isolated task-context snapshot.""" - return get_jm_task_context() + return JTC.get_context() @classmethod def build(cls, option: JmOption) -> 'JmOptionPlugin': @@ -931,7 +931,7 @@ class DownloadProgressPlugin(JmOptionPlugin): @staticmethod def cli_no_progress_notice(): - if not get_jm_task_context().get('cli_no_progress'): + if not JTC.get_context().get('cli_no_progress'): return '' return ( diff --git a/src/jmcomic/jm_runtime.py b/src/jmcomic/jm_runtime.py new file mode 100644 index 000000000..c6e89654d --- /dev/null +++ b/src/jmcomic/jm_runtime.py @@ -0,0 +1,246 @@ +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)) + finally: + if wait_finish: + concurrent.futures.wait(futures) + + return futures + + def close(self) -> None: + """ + 关闭 Runtime 自建的 Executor；调用方传入的 Executor 保持可用。 + """ + with self._lock: + if self._closed: + return + self._closed = True + owned_executors = tuple(self._owned_executors) + self._owned_executors.clear() + + for executor in owned_executors: + executor.shutdown(wait=True) + + +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 的 decode Executor。 + """ + + def __init__( + self, + *, + decode_workers: Optional[int] = None, + decode_executor: Optional[Executor] = None, + ): + super().__init__( + configs={ + 'decode': normalize_jm_executor_config( + 'decode', decode_workers, decode_executor + ), + }, + ) diff --git a/src/jmcomic/jm_task_context.py b/src/jmcomic/jm_task_context.py index 57a5e9e1d..1b3022dbe 100644 --- a/src/jmcomic/jm_task_context.py +++ b/src/jmcomic/jm_task_context.py @@ -2,14 +2,19 @@ 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', - 'jm_task_context', + 'DownloadControl', + 'JTC', 'get_jm_task_context', + 'jm_task_context', 'bind_jm_task_context', ) @@ -21,16 +26,104 @@ ) -def get_jm_task_context() -> dict: - """Return a mutable snapshot of the current JM task context.""" - return dict(JM_TASK_CONTEXT.get()) +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 + + +class JTC: + """ + JMComic 任务上下文统一门面 (Jm Task Context)。 + """ + + @classmethod + def get_context(cls) -> dict: + """ + 返回当前 JM 任务上下文的可变快照。 + """ + return dict(JM_TASK_CONTEXT.get()) + + @classmethod + def get_runtime(cls) -> Optional[JmRuntime]: + """ + 返回当前任务绑定的 JmRuntime；没有活动 Runtime 时返回 None。 + """ + return cls.get_context().get('runtime') + + @classmethod + def get_option(cls): + """ + 返回当前任务绑定的 JmOption；没有活动 Option 时返回 None。 + """ + return cls.get_context().get('option') + + @classmethod + def get_control(cls) -> Optional[DownloadControl]: + """ + 返回当前任务绑定的取消控制器 DownloadControl；未设置时返回 None。 + """ + control = cls.get_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 + + +# 兼容已发布的上下文查询入口，内部统一使用 JTC。 +get_jm_task_context = JTC.get_context @contextmanager -def jm_task_context(**fields): - """Temporarily add fields to the current JM task context.""" - context = get_jm_task_context() +def jm_task_context(*, option=None, runtime=None, **fields): + """ + 临时绑定任务字段；只传播 Option/Runtime，不管理资源生命周期。 + """ + context = JTC.get_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,13 +132,17 @@ 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( - get_jm_task_context() if context is None else context + JTC.get_context() if context is None else context )) @wraps(func) 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..78631295d 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__(decode_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(JTC.get_runtime()) + with self.assertRaisesRegex(RuntimeError, 'JmRuntime is closed'): + runtime.executor('decode', 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..04e0313cd --- /dev/null +++ b/tests/test_jmcomic/test_jm_cancellation.py @@ -0,0 +1,422 @@ +import asyncio +from concurrent.futures import Future +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, + JmAsyncRuntime, + JmDownloader, + bind_jm_task_context, + download_album, + download_batch, + download_batch_async, + JTC, + 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(JTC.get_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(JTC.get_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) + + async def run_inline(func, *args): + return func(*args) + + downloader._run_in_decode_pool = run_inline + downloader._save_raw = Mock(side_effect=lambda *_args: control.cancel('save finished')) + 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' + image.index = 1 + image.from_photo = Mock() + photo = image.from_photo + album = photo.from_album + photo.is_album.return_value = False + downloader.download_success_dict = {album: {photo: []}} + downloader.manifest_dict = {} + downloader.begin_manifest(photo) + downloader.before_image = AsyncMock() + + 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(downloader.finish_manifest(photo).image_filepath_list, ['image.jpg']) + downloader._save_raw.assert_called_once() + downloader.option.call_all_plugin.assert_not_called() + + def test_sync_image_saved_during_cancellation_is_in_manifest(self): + control = DownloadControl() + downloader = object.__new__(JmDownloader) + downloader.option = Mock() + 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 + downloader.client = Mock() + downloader.client.download_by_image_detail.side_effect = ( + lambda *_args, **_kwargs: control.cancel('save finished') + ) + downloader.before_image = Mock() + image = Mock(spec=JmImageDetail) + image.skip = False + image.index = 1 + image.from_photo = Mock() + photo = image.from_photo + album = photo.from_album + photo.is_album.return_value = False + downloader.download_success_dict = {album: {photo: []}} + downloader.manifest_dict = {} + downloader.begin_manifest(photo) + + with jm_task_context(control=control): + with self.assertRaises(DownloadCancelledException): + downloader.download_by_image_detail(image) + + self.assertEqual(downloader.finish_manifest(photo).image_filepath_list, ['image.jpg']) + downloader.client.download_by_image_detail.assert_called_once() + downloader.option.call_all_plugin.assert_not_called() + + async def test_cancelled_decode_logs_worker_failure_and_preserves_cancellation(self): + future = Future() + submitted = asyncio.Event() + executor = Mock() + + def submit(*_args): + submitted.set() + return future + + executor.submit.side_effect = submit + runtime = JmAsyncRuntime(decode_workers=1) + downloader = object.__new__(JmAsyncDownloader) + downloader._decode_worker = 1 + error = OSError('save failed') + try: + with patch.object(runtime, 'executor', return_value=executor): + with patch('jmcomic.jm_async_downloader.jm_log') as log: + with jm_task_context(runtime=runtime): + task = asyncio.create_task(downloader._run_in_decode_pool(lambda: None)) + await asyncio.wait_for(submitted.wait(), timeout=1) + task.cancel() + await asyncio.sleep(0) + future.set_exception(error) + with self.assertRaises(asyncio.CancelledError): + await task + log.assert_called_once_with('dler.cancel.drain.exception', error) + finally: + runtime.close() + + 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..40c4cd7ba 100644 --- a/tests/test_jmcomic/test_jm_cli.py +++ b/tests/test_jmcomic/test_jm_cli.py @@ -1,10 +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.jm_task_context import get_jm_task_context +from jmcomic import JmSyncRuntime, JTC, jm_task_context class Test_Cli(JmTestConfigurable): @@ -60,7 +62,7 @@ def parse_arg(): ui.option_path = None def create_default_option(): - self.assertTrue(get_jm_task_context().get('cli_no_progress')) + self.assertTrue(JTC.get_context().get('cli_no_progress')) return option with patch.object(ui, 'parse_arg', side_effect=parse_arg), \ @@ -70,7 +72,7 @@ def create_default_option(): patch('jmcomic.api.jm_log'): ui.main() - self.assertNotIn('cli_no_progress', get_jm_task_context()) + self.assertNotIn('cli_no_progress', JTC.get_context()) def test_jmcomic_falls_back_without_rich(self): ui = JmcomicUI() @@ -107,6 +109,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, + JTC.get_runtime(), + JTC.get_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 dfc36e388..bcc368ad0 100644 --- a/tests/test_jmcomic/test_jm_client.py +++ b/tests/test_jmcomic/test_jm_client.py @@ -257,17 +257,9 @@ def search_and_test(expected_result, params): raise AssertionError(elist) def test_comment_count(self): - aid = 'JM438516' + aid = '350234' album = self.client.get_album_detail(aid) self.assertGreater(album.comment_count, 0) - page = self.client.search_site('无修正') - for i in range(3): - aid, _atitle = page[i] - self.assertGreaterEqual( - self.client.get_album_detail(aid).comment_count, - 0, - aid, - ) def test_album_pagination(self): album_id = '302820' diff --git a/tests/test_jmcomic/test_jm_download_manifest.py b/tests/test_jmcomic/test_jm_download_manifest.py index 48037c20b..0bb0f7a79 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 @@ -75,7 +77,7 @@ def decide_image_batch_count(self, _photo): def call_all_plugin(self, group, **kwargs): self.plugin_event_list.append((group, kwargs)) - self.context_event_list.append((group, get_jm_task_context())) + self.context_event_list.append((group, JTC.get_context())) if group == 'after_image' and self.after_image_callback is not None: self.after_image_callback(kwargs['image']) @@ -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,120 @@ def do_filter(self, detail): class Test_Download_Manifest(unittest.TestCase): + def test_after_image_error_does_not_register_sync_success(self): + with TemporaryDirectory() as temp_dir: + album, photo, images = new_album_photo_images() + option = ContractOption(temp_dir) + downloader = ContractSyncDownloader(option, album, photo, images) + error = ValueError('image plugin failed') + + def fail(_image): + raise error + + option.after_image_callback = fail + with self.assertRaises(ValueError) as caught: + downloader.download_album(album.id) + + self.assertIs(caught.exception, error) + self.assertTrue(os.path.isfile(images[0].save_path)) + self.assertEqual(downloader.download_success_dict[album][photo], []) + self.assertEqual(downloader.download_failed_image, [(images[0], error)]) + self.assertEqual(downloader.manifest_dict[album].image_filepath_list, []) + + def test_after_image_error_does_not_register_async_success(self): + async def run_test(temp_dir): + album, photo, images = new_album_photo_images() + option = ContractOption(temp_dir) + downloader = ContractAsyncDownloader(option, album, photo, images) + error = ValueError('image plugin failed') + + def fail(_image): + raise error + + option.after_image_callback = fail + runtime = JmAsyncRuntime() + try: + with jm_task_context(runtime=runtime): + await downloader.download_album(album.id) + finally: + runtime.close() + + self.assertTrue(os.path.isfile(images[0].save_path)) + self.assertEqual(downloader.download_success_dict[album][photo], []) + self.assertEqual(downloader.download_failed_image, [(images[0], error)]) + self.assertEqual(downloader.manifest_dict[album].image_filepath_list, []) + + with TemporaryDirectory() as temp_dir: + asyncio.run(run_test(temp_dir)) + + def test_after_image_direct_cancellation_keeps_saved_image(self): + with TemporaryDirectory() as temp_dir: + album, photo, images = new_album_photo_images() + option = ContractOption(temp_dir) + downloader = ContractSyncDownloader(option, album, photo, images) + error = DownloadCancelledException('image plugin cancelled') + + def cancel(_image): + raise error + + option.after_image_callback = cancel + with self.assertRaises(DownloadCancelledException) as caught: + downloader.download_album(album.id) + + self.assertIs(caught.exception, error) + self.assertEqual(downloader.download_failed_image, []) + self.assertEqual( + downloader.manifest_dict[album].image_filepath_list, + [images[0].save_path], + ) + + 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() @@ -794,7 +914,7 @@ def test_entity_timing_contexts_are_nested_and_do_not_leak(self): [2, 2], [context['photo_started_at'] for context in context_by_group['before_image']], ) - self.assertEqual({}, get_jm_task_context()) + self.assertEqual({}, JTC.get_context()) def test_sync_download_album_duration_includes_detail_and_manifest(self): clock = {'now': 10.0} @@ -803,20 +923,20 @@ def test_sync_download_album_duration_includes_detail_and_manifest(self): downloader = object.__new__(JmDownloader) def get_album_detail(_album_id): - contexts.append(get_jm_task_context()) + contexts.append(JTC.get_context()) clock['now'] = 20.0 return album def begin_manifest(_album): - contexts.append(get_jm_task_context()) + contexts.append(JTC.get_context()) clock['now'] = 30.0 def download_by_album_detail(_album): - contexts.append(get_jm_task_context()) + contexts.append(JTC.get_context()) clock['now'] = 40.0 def finish_manifest(_album): - contexts.append(get_jm_task_context()) + contexts.append(JTC.get_context()) clock['now'] = 50.0 downloader.client = SimpleNamespace(get_album_detail=get_album_detail) @@ -830,7 +950,7 @@ def finish_manifest(_album): self.assertIs(result, album) self.assertEqual(40.0, album.duration) self.assertEqual([10.0] * 4, [context.get('album_started_at') for context in contexts]) - self.assertEqual({}, get_jm_task_context()) + self.assertEqual({}, JTC.get_context()) def test_sync_download_photo_duration_includes_detail_and_manifest(self): clock = {'now': 10.0} @@ -839,20 +959,20 @@ def test_sync_download_photo_duration_includes_detail_and_manifest(self): downloader = object.__new__(JmDownloader) def get_photo_detail(_photo_id): - contexts.append(get_jm_task_context()) + contexts.append(JTC.get_context()) clock['now'] = 20.0 return photo def begin_manifest(_photo): - contexts.append(get_jm_task_context()) + contexts.append(JTC.get_context()) clock['now'] = 30.0 def download_by_photo_detail(_photo): - contexts.append(get_jm_task_context()) + contexts.append(JTC.get_context()) clock['now'] = 40.0 def finish_manifest(_photo): - contexts.append(get_jm_task_context()) + contexts.append(JTC.get_context()) clock['now'] = 50.0 downloader.client = SimpleNamespace(get_photo_detail=get_photo_detail) @@ -866,7 +986,7 @@ def finish_manifest(_photo): self.assertIs(result, photo) self.assertEqual(40.0, photo.duration) self.assertEqual([10.0] * 4, [context.get('photo_started_at') for context in contexts]) - self.assertEqual({}, get_jm_task_context()) + self.assertEqual({}, JTC.get_context()) def test_async_download_album_duration_includes_detail_and_manifest(self): async def run_test(): @@ -876,20 +996,20 @@ async def run_test(): downloader = object.__new__(JmAsyncDownloader) async def get_album_detail(_album_id): - contexts.append(get_jm_task_context()) + contexts.append(JTC.get_context()) clock['now'] = 20.0 return album def begin_manifest(_album): - contexts.append(get_jm_task_context()) + contexts.append(JTC.get_context()) clock['now'] = 30.0 async def download_by_album_detail(_album): - contexts.append(get_jm_task_context()) + contexts.append(JTC.get_context()) clock['now'] = 40.0 def finish_manifest(_album): - contexts.append(get_jm_task_context()) + contexts.append(JTC.get_context()) clock['now'] = 50.0 downloader.client = SimpleNamespace(get_album_detail=get_album_detail) @@ -903,7 +1023,7 @@ def finish_manifest(_album): self.assertIs(result, album) self.assertEqual(40.0, album.duration) self.assertEqual([10.0] * 4, [context.get('album_started_at') for context in contexts]) - self.assertEqual({}, get_jm_task_context()) + self.assertEqual({}, JTC.get_context()) asyncio.run(run_test()) @@ -915,20 +1035,20 @@ async def run_test(): downloader = object.__new__(JmAsyncDownloader) async def get_photo_detail(_photo_id): - contexts.append(get_jm_task_context()) + contexts.append(JTC.get_context()) clock['now'] = 20.0 return photo def begin_manifest(_photo): - contexts.append(get_jm_task_context()) + contexts.append(JTC.get_context()) clock['now'] = 30.0 async def download_by_photo_detail(_photo): - contexts.append(get_jm_task_context()) + contexts.append(JTC.get_context()) clock['now'] = 40.0 def finish_manifest(_photo): - contexts.append(get_jm_task_context()) + contexts.append(JTC.get_context()) clock['now'] = 50.0 downloader.client = SimpleNamespace(get_photo_detail=get_photo_detail) @@ -942,7 +1062,7 @@ def finish_manifest(_photo): self.assertIs(result, photo) self.assertEqual(40.0, photo.duration) self.assertEqual([10.0] * 4, [context.get('photo_started_at') for context in contexts]) - self.assertEqual({}, get_jm_task_context()) + self.assertEqual({}, JTC.get_context()) asyncio.run(run_test()) @@ -1032,21 +1152,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..5273bdba2 100644 --- a/tests/test_jmcomic/test_jm_exception.py +++ b/tests/test_jmcomic/test_jm_exception.py @@ -4,6 +4,50 @@ 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_replaced_exception_executor_preserves_requested_type(self): + class IsolatedExceptionTool(ExceptionTool): + pass + + IsolatedExceptionTool.replace_old_exception_executor( + lambda old, msg, context: old(msg=msg, context=context) + ) + + with self.assertRaises(DownloadCancelledException) as caught: + IsolatedExceptionTool.require_true( + False, 'cancel probe', etype=DownloadCancelledException, + ) + self.assertEqual(caught.exception.reason, 'cancel probe') + + def test_replaced_exception_executor_can_override_requested_type(self): + class IsolatedExceptionTool(ExceptionTool): + pass + + context = {'detail': 'override probe'} + IsolatedExceptionTool.replace_old_exception_executor( + lambda old, msg, context: old(msg, context, JmcomicException) + ) + + with self.assertRaises(JmcomicException) as caught: + IsolatedExceptionTool.raises( + 'override probe', context, DownloadCancelledException, + ) + self.assertIs(type(caught.exception), JmcomicException) + self.assertEqual(caught.exception.context, context) + 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_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..bb2ede3d3 --- /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, + JTC, + 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(JTC.get_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(JTC.get_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('decode', 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..298d803f5 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,8 +9,11 @@ from jmcomic import ( BaseDownloader, + JmAsyncRuntime, DownloadManifest, + DownloadControl, Feature, + JTC, JmAsyncDownloader, JmDownloader, JmModuleConfig, @@ -18,13 +22,14 @@ 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_jm_task_context, jm_log, jm_task_context, jm_logger, @@ -41,39 +46,234 @@ 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( + JTC.get_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(JTC.get_runtime(), runtime) + with jm_task_context(jm_id='123'): + self.assertEqual( + JTC.get_context(), + { + 'task_id': 'job-42', + 'runtime': runtime, + 'jm_id': '123', + }, + ) + + public_bound = executor.submit(bind_jm_task_context( + lambda: ( + JTC.get_context(), + JTC.get_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( + JTC.get_context(), + {'task_id': 'job-42', 'option': option}, + ) + self.assertIs(JTC.get_option(), option) + future = executor.submit(bind_jm_task_context(JTC.get_option)) + + self.assertIs(future.result(timeout=1), option) + + self.assertIsNone(JTC.get_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(JTC.get_context(), {'task_id': 'job-42'}) + self.assertIsNone(JTC.get_option()) + self.assertIsNone(JTC.get_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(JTC.get_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(decode_workers=1, decode_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(decode_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(JTC.get_control)) + + self.assertIs(future.result(timeout=1), control) + self.assertIsNone(executor.submit(JTC.get_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) def test_nested_context_restores_on_normal_and_exception_exit(self): - self.assertEqual({}, get_jm_task_context()) + self.assertEqual({}, JTC.get_context()) with jm_task_context(session_id='outer'): - self.assertEqual({'session_id': 'outer'}, get_jm_task_context()) + self.assertEqual({'session_id': 'outer'}, JTC.get_context()) with jm_task_context(session_id='inner', task_id='task'): self.assertEqual( {'session_id': 'inner', 'task_id': 'task'}, - get_jm_task_context(), + JTC.get_context(), ) - self.assertEqual({'session_id': 'outer'}, get_jm_task_context()) + self.assertEqual({'session_id': 'outer'}, JTC.get_context()) with self.assertRaisesRegex(RuntimeError, 'stop'): with jm_task_context(task_id='failed'): raise RuntimeError('stop') - self.assertEqual({'session_id': 'outer'}, get_jm_task_context()) + self.assertEqual({'session_id': 'outer'}, JTC.get_context()) - self.assertEqual({}, get_jm_task_context()) + self.assertEqual({}, JTC.get_context()) def test_bound_context_survives_thread_pool_and_does_not_leak(self): with ThreadPoolExecutor(max_workers=1) as executor: with jm_task_context(session_id='A'): - future_a = executor.submit(bind_jm_task_context(get_jm_task_context)) + future_a = executor.submit(bind_jm_task_context(JTC.get_context)) with jm_task_context(session_id='B'): - future_b = executor.submit(bind_jm_task_context(get_jm_task_context)) + future_b = executor.submit(bind_jm_task_context(JTC.get_context)) - empty = executor.submit(get_jm_task_context) + empty = executor.submit(JTC.get_context) self.assertEqual({'session_id': 'A'}, future_a.result()) self.assertEqual({'session_id': 'B'}, future_b.result()) @@ -104,7 +304,7 @@ def test_default_logger_and_custom_executors_can_read_context(self): captured = [] def executor_two_args(topic, msg): - captured.append((topic, msg, get_jm_task_context())) + captured.append((topic, msg, JTC.get_context())) JmModuleConfig.EXECUTOR_LOG = executor_two_args with jm_task_context(session_id='custom-2'): @@ -113,7 +313,7 @@ def executor_two_args(topic, msg): error = ValueError('failed') def executor_three_args(topic, msg, e): - captured.append((topic, msg, e, get_jm_task_context())) + captured.append((topic, msg, e, JTC.get_context())) JmModuleConfig.EXECUTOR_LOG = executor_three_args with jm_task_context(session_id='custom-3'): @@ -267,10 +467,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('decode', 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 @@ -291,19 +498,22 @@ def download_album(self, _album_id): detail = Detail() self.begin_manifest(detail) try: - observed_contexts.append(get_jm_task_context()) + observed_contexts.append(JTC.get_context()) + observed_options.append(JTC.get_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({}, get_jm_task_context()) + self.assertEqual([option], observed_options) + self.assertEqual({}, JTC.get_context()) + self.assertIsNone(JTC.get_option()) def test_async_result_duration_uses_task_context_and_finishes_after_downloader_exit(self): async def run_test(): @@ -321,7 +531,7 @@ def is_album(): class FakeDownloader(BaseDownloader): async def __aenter__(self): - observed_contexts.append(get_jm_task_context()) + observed_contexts.append(JTC.get_context()) return self async def __aexit__(self, *_args): @@ -344,7 +554,7 @@ async def download_photo(self, _photo_id): self.assertEqual(30.0, result.manifest.duration) self.assertEqual(30.0, result.duration) self.assertEqual(99.0, result.detail.duration) - self.assertEqual({}, get_jm_task_context()) + self.assertEqual({}, JTC.get_context()) def test_plugin_invocation_can_read_current_task_context(self): observed = [] @@ -440,7 +650,7 @@ def test_plugin_feature_uses_download_type_context(self): def test_sync_batch_binds_parent_and_item_context(self): def fake_download(jmid, _option, _downloader, **_kwargs): - context = get_jm_task_context() + context = JTC.get_context() return ( str(jmid), context['session_id'], @@ -458,14 +668,14 @@ def fake_download(jmid, _option, _downloader, **_kwargs): }, set(result), ) - self.assertEqual({}, get_jm_task_context()) + self.assertEqual({}, JTC.get_context()) def test_concurrent_sync_sessions_do_not_cross(self): barrier = threading.Barrier(2) def fake_download(jmid, _option, _downloader, **_kwargs): barrier.wait(timeout=2) - context = get_jm_task_context() + context = JTC.get_context() return context['session_id'], context['jm_id'], str(jmid) def run_session(session_id, jmid): @@ -497,9 +707,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,42 +721,58 @@ 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(JTC.get_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(): async def fake_download(jmid, _option, _downloader, **_kwargs): await asyncio.sleep(0) - context = get_jm_task_context() + context = JTC.get_context() return ( str(jmid), context['session_id'], @@ -560,21 +787,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 decode_executor: + runtime = JmAsyncRuntime(decode_executor=decode_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( + 'decode', + 1, + ).submit( + bind_jm_task_context(JTC.get_context), + ) + executor_context = await asyncio.wrap_future(future) loop = asyncio.get_running_loop() leaked_context = await loop.run_in_executor( - downloader._decode_pool, - get_jm_task_context, + decode_executor, + JTC.get_context, ) - finally: - downloader.shutdown() return batch_result, executor_context, leaked_context @@ -587,7 +816,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): @@ -597,12 +827,91 @@ class FakeClient: proxy = PhotoConcurrentFetcherProxy(FakeClient(), max_workers=1) try: with jm_task_context(session_id='client-proxy'): - future = proxy.get_future('context', get_jm_task_context) + future = proxy.get_future('context', JTC.get_context) self.assertEqual({'session_id': 'client-proxy'}, future.result()) 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(decode_executor=executor) + with jm_task_context(session_id='async-runtime', runtime=runtime): + future = runtime.executor( + 'decode', + 1, + ).submit( + bind_jm_task_context( + lambda: (8, JTC.get_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()) + + def test_jtc_facade_getters(self): + # 根作用域无状态 + self.assertIsNone(JTC.get_runtime()) + self.assertIsNone(JTC.get_option()) + self.assertIsNone(JTC.get_control()) + self.assertEqual(JTC.get_context(), {}) + + runtime = JmSyncRuntime() + option = object() + control = DownloadControl() + + try: + with jm_task_context(runtime=runtime, option=option, control=control, custom_field='val'): + self.assertIs(JTC.get_runtime(), runtime) + self.assertIs(JTC.get_option(), option) + self.assertIs(JTC.get_control(), control) + ctx = JTC.get_context() + self.assertIs(ctx['runtime'], runtime) + self.assertIs(ctx['option'], option) + self.assertIs(ctx['control'], control) + self.assertEqual(ctx['custom_field'], 'val') + finally: + runtime.close() + + # 退出作用域后恢复 + self.assertIsNone(JTC.get_runtime()) + self.assertIsNone(JTC.get_option()) + self.assertIsNone(JTC.get_control()) + if __name__ == '__main__': unittest.main()