diff --git a/README.md b/README.md index 5fbb57113..f7b60c242 100644 --- a/README.md +++ b/README.md @@ -238,7 +238,7 @@ jmv 350234 -y 等 - **可扩展性强** - - 支持自定义本子/章节/图片下载前后的回调函数 + - 支持自定义本子/章节/图片下载前后事件的回调函数 - 支持自定义类：`Downloader（负责调度）` `Option（负责配置）` `Client（负责请求）` `实体类`等 - 支持自定义日志、异常监听器 - **支持Plugin插件，可以方便地扩展功能，以及使用别人的插件，目前核心内置插件有**： diff --git a/assets/docs/sources/api/command-line.md b/assets/docs/sources/api/command-line.md index 6f44be8ec..5e2d93044 100644 --- a/assets/docs/sources/api/command-line.md +++ b/assets/docs/sources/api/command-line.md @@ -1,6 +1,6 @@ # command-line -::: jmcomic.cl +::: jmcomic.cli options: members: - JmcomicUI diff --git a/assets/docs/sources/index.md b/assets/docs/sources/index.md index 89d4ce8ea..8d06270fd 100644 --- a/assets/docs/sources/index.md +++ b/assets/docs/sources/index.md @@ -65,11 +65,11 @@ - + - - - Repo Card + + + Repo Card @@ -83,4 +83,4 @@ - \ No newline at end of file + diff --git a/assets/docs/sources/option_file_syntax.md b/assets/docs/sources/option_file_syntax.md index e676e94f4..49e0975fd 100644 --- a/assets/docs/sources/option_file_syntax.md +++ b/assets/docs/sources/option_file_syntax.md @@ -186,8 +186,8 @@ plugins: # v2.5.0 引入的插件 # 可以启动一个服务器，可以在浏览器上查看本子 - # 基于flask框架，需要安装额外库: [pip install plugin_jm_server] - # 源码：https://github.com/hect0x7/plugin-jm-server + # 基于flask框架，需要安装额外库: [pip install jm-view-server] + # 源码：https://github.com/hect0x7/jm-view-server - plugin: jm_server kwargs: password: '3333' # 服务器访问密码 @@ -227,10 +227,10 @@ plugins: rule: '{Atitle}/{Aid}_cover.jpg' - after_album: # 钩子（插件被调用时机） + after_album: # 事件（插件被调用的时机） - plugin: zip # 压缩文件插件 kwargs: - # 压缩文件插件，配在不同钩子下面，效果不一样。可以选择配在 after_album 或者 after_photo 下 + # 压缩文件插件，配在不同事件下面，效果不一样。可以选择配在 after_album 或者 after_photo 下 # 配置在 after_album 下 → 整个本子合并为一个压缩文件 # 配置在 after_photo 下 → 每个章节各一个压缩文件 # （旧的 level 配置已废弃，如果你配置过level，比如level=photo, 请直接改用after_photo） @@ -239,7 +239,7 @@ plugins: suffix: zip #压缩包后缀名，默认值为zip，可以指定为zip或者7z filename_rule: Atitle # 压缩文件的命名规则 # 请注意⚠ [https://github.com/hect0x7/JMComic-Crawler-Python/issues/223#issuecomment-2045227527] - # filename_rule和所在钩子有对应关系 + # filename_rule和所在事件有对应关系 # 如果配置在 after_photo 下, filename_rule 可以写 Pxxx 和Axxx # 如果配置在 after_album 下, filename_rule 只能写 Axxx，不能写 Pxxx diff --git a/assets/docs/sources/tutorial/0_common_usage.md b/assets/docs/sources/tutorial/0_common_usage.md index d6268a413..1101cbce3 100644 --- a/assets/docs/sources/tutorial/0_common_usage.md +++ b/assets/docs/sources/tutorial/0_common_usage.md @@ -109,7 +109,7 @@ except JmcomicException as e: # 而JmDownloader有对应字段记录了这些线程发生的异常 # 使用check_exception=True参数可以使downloader主动检查是否存在下载异常 # 如果有，则当前线程会主动上抛一个PartialDownloadFailedException异常 -# 该参数主要用于主动检查部分下载失败的情况， +# 该参数主要用于主动检查部分下载失败的情况，（仅对单个本子/章节 ID 生效，传入多个 ID 时不生效。多个 ID 的场景见下） # 因为非当前线程抛出的异常（比如下载章节的线程和下载图片的线程），这些线程如果抛出异常， # 当前线程是感知不到的，try-catch下载方法download_album不能捕获到其他线程发生的异常。 try: @@ -117,6 +117,12 @@ try: except PartialDownloadFailedException as e: downloader: JmDownloader = e.downloader print(f'下载出现部分失败, 下载失败的章节: {downloader.download_failed_photo}, 下载失败的图片: {downloader.download_failed_image}') + +# 多 ID 下载不会因为某一项失败而中断，请检查 BatchResult.failed。 +# 如果需要在批量失败时抛异常或重试，建议自行封装 download_batch。 +result = download_album([123, 456, 789]) +for album_id, error in result.failed.items(): + print(f'本子 {album_id} 下载失败: {error}') ``` @@ -168,6 +174,28 @@ download_album(aid_list, option) 可参考discussions: https://github.com/hect0x7/JMComic-Crawler-Python/discussions/235 +### 一键导出全部收藏夹 + +下面的代码不会下载图片，只会把帐号中的全部收藏夹导出为 CSV，并生成 `favorites.zip`： + +```python +from jmcomic import JmOption, FavoriteFolderExportPlugin + +USERNAME = '你的禁漫帐号' +PASSWORD = '你的禁漫密码' + +option = JmOption.default() +option.build_jm_client().login(USERNAME, PASSWORD) + +FavoriteFolderExportPlugin(option).invoke( + save_dir='./', + zip_enable=True, + zip_filepath='./favorites.zip', +) +``` + +### 获取并遍历收藏夹 + ```python from jmcomic import * diff --git a/assets/docs/sources/tutorial/11_log_custom.md b/assets/docs/sources/tutorial/11_log_custom.md index 66aefb5d6..9b6e893a0 100644 --- a/assets/docs/sources/tutorial/11_log_custom.md +++ b/assets/docs/sources/tutorial/11_log_custom.md @@ -21,23 +21,45 @@ from jmcomic import disable_jm_log disable_jm_log() ``` -## 2. 日志过滤，只保留特定topic +## 2. 日志过滤，只保留特定 topic -使用插件配置 +最简单的方式是使用内置插件配置： ```yaml log: true plugins: after_init: - - plugin: log_topic_filter # 日志topic过滤插件 + - plugin: log_topic_filter # 日志 topic 过滤插件 kwargs: - whitelist: [ # 只保留api和html，这两个是Client发请求时会打的日志topic + whitelist: [ # 只保留 api 和 html，这两个是 Client 发请求时会打印的日志 topic 'api', 'html', ] ``` +这个插件底层使用标准 `logging.Filter`。jmcomic 将每条日志的 topic 放在 `LogRecord.topic` 中，需要通过代码自定义时可以直接使用 logging API： + +```python +import logging + +from jmcomic import jm_logger + + +class TopicFilter(logging.Filter): + def __init__(self, whitelist): + super().__init__() + self.whitelist = set(whitelist) + + def filter(self, record): + return getattr(record, 'topic', None) in self.whitelist + + +jm_logger.addFilter(TopicFilter({'api', 'html'})) +``` + +Filter 加在 `jm_logger` 上会作用于它的所有 Handler；如果只想过滤某个输出目标，也可以将 Filter 加到对应的 Handler 上。 + ## 3. 屏蔽插件的日志 给插件配置加上一个`log`配置项即可 @@ -52,7 +74,76 @@ plugins: whitelist: [ api, ] ``` -## 4. 深度自定义：两类不同的拦截手段 +## 4. 并发下载的日志上下文 + +场景：当你在同时运行多个下载任务时，你希望区分哪些日志属于哪些任务，或者你希望结构化展示下载任务 对应的 下载日志。 + +首先，你需要设定一个任务id，这一步是在定义你的**任务维度**。下载一个本子还是多个本子都可以算作一个任务。 + +```python +from jmcomic import download_album, jm_task_context + +with jm_task_context(task_id='task-1'): # 设置这个下载本子123的任务id为 task-1 + download_album(123) + +with jm_task_context(task_id='task-1'): # 设置这三个本子的任务id为 task-1 + download_album([123, 456, 789]) +``` + +设置了task_id以后，整个 download_album 内的日志打印都会具有这个task_id标识。 + +> jmcomic 使用 [ContextVar](https://docs.python.org/3/library/contextvars.html) 把上下文传播到库内部创建的下载线程或async task里。你设置的task_id其实就是上下文里的一个字段 + +最直接的效果是，默认终端日志会自动显示任务 ID： + +```text +[2026-01-01 12:00:00] [MainThread]:[task_id=task-1; album=123] 【album.after】本子下载完成: [123] +``` + +接下来你可以实现更高级的功能，比如收集任务对应的日志： + +给 `jm_logger` 添加一个 Handler。每当 jmcomic 产生一条日志，Handler 都可以从 `record.jm_task_context` 中取出它所属的 `task_id`，从而按任务收集日志。完整代码如下： + +```python +import logging +from collections import defaultdict + +from jmcomic import jm_logger, download_album, jm_task_context + +task_logs_dict = defaultdict(list) # 收集任务日志，任务id -> 日志列表 + + +class TaskLogHandler(logging.Handler): + def emit(self, record): + context: dict = getattr(record, 'jm_task_context', None) or {} # 通过 jm_task_context 字段取出任务上下文 + task_id = context.get('task_id') # 任务上下文里的task_id，就是你上面自定义的 task-1 + if task_id is not None: + task_logs_dict[task_id].append(self.format(record)) # 收集日志 + + +handler = TaskLogHandler() +handler.setFormatter(jm_logger.handlers[0].formatter) # 复用jmcomic默认 Handler 的日志格式，你也可以自定义日志格式 +jm_logger.addHandler(handler) + +# 自定义handler后，再正常使用下载方法 +task_id = 'task-1' +with jm_task_context(task_id=task_id): # 任务id + download_album(123) +``` + +任务上下文里，常用字段如下： + +| 字段 | 类型 | 谁来设置 | 含义 | 示例 | +| --- |--------| --- |-----------------------------------------------------------------------| --- | +| `task_id` | 你传入的类型；本例需可哈希，建议使用 `str` | 由你通过 `jm_task_context` 设置 | 你的一次下载任务标识 | `task-1` | +| `download_type` | `str` | jmcomic 自动设置 | 你使用的download入口函数类型，例如 download_album -> album | `album` / `photo` | +| `jm_id` | `str` | jmcomic 自动设置 | 你使用的download入口函数的入参，例如 download_album(123) -> 123。传入多个 ID 时，每个 ID 都在各自隔离的上下文中记录对应的 `jm_id`，并继承同一个 `task_id`。 | `123` | + +> 默认终端日志仅在 `task_id` 有值时显示任务上下文，并一同显示 `download_type` 和 `jm_id`。未设置 `task_id` 时，上下文仍会正常传递，但不会显示在默认日志中。 + +你也可以放入其他对象到任务上下文里，比如放入一个局部queue用来收集日志。在 jm_task_context 方法里传入即可 `jm_task_context(**fields)` + +## 5. 深度自定义：两类不同的拦截手段 根据你的需求复杂度，你可以选择以下方式： @@ -64,4 +155,4 @@ plugins: 适用于：需要完全重塑日志的分发逻辑，或者将日志直接桥接到不符合标准 logging 协议的第三方系统。 -详细参考文档：[模块自定义](./4_module_custom.md#自定义log) +代码示例：[模块自定义-自定义log](./4_module_custom.md#自定义log) diff --git a/assets/docs/sources/tutorial/13_export_and_feature.md b/assets/docs/sources/tutorial/13_export_and_feature.md index 99a63855b..3829c5da9 100644 --- a/assets/docs/sources/tutorial/13_export_and_feature.md +++ b/assets/docs/sources/tutorial/13_export_and_feature.md @@ -155,12 +155,12 @@ Feature (基类) └── 你的自定义 Feature ← 继承 Feature，实现任意逻辑 ``` -- **Feature 基类**：通用的附加行为抽象，不绑定任何具体实现。默认在所有生命周期钩子中执行。 +- **Feature 基类**：通用的附加行为抽象，不绑定任何具体实现。默认在所有生命周期事件中执行。 - **PluginFeature**：Feature 的子类，专门封装 jmcomic 插件。除了调用插件之外，还会根据调用来源动态适配 `filename_rule` 参数；ZIP 的打包粒度则由插件在运行时根据上下文自动推导。 ### 执行流程 -Feature **自然嵌入到 downloader 的生命周期钩子**中自动触发： +Feature **自然嵌入到 downloader 的生命周期事件**中自动触发： ```text api.download_album(extra=Feature.export_pdf) @@ -187,7 +187,7 @@ api.download_album(extra=Feature.export_pdf) > 💡 **关键点**： > -> - **执行时机**：`PluginFeature` 根据注册来源自动推导（`download_album` → `after_album`，`download_photo` → `after_photo`）。自定义 Feature 默认在所有钩子都会执行，你可以覆写 `should_invoke` 来控制。 +> - **执行时机**：`PluginFeature` 根据注册来源自动推导（`download_album` → `after_album`，`download_photo` → `after_photo`）。自定义 Feature 默认在所有事件都会执行，你可以覆写 `should_invoke` 来控制。 > - **参数自适应**：`PluginFeature` 的 `filename_rule` 前缀（A/P）会根据来源动态适配。ZIP 的打包粒度由插件根据上下文自动推导。用户显式传入的参数不会被覆盖。 ### 自定义 Feature @@ -207,4 +207,3 @@ class NotifyFeature(Feature): # 使用 download_album('123', option, extra=NotifyFeature()) ``` - diff --git a/assets/docs/sources/tutorial/4_module_custom.md b/assets/docs/sources/tutorial/4_module_custom.md index 06274c6fd..d804702cc 100644 --- a/assets/docs/sources/tutorial/4_module_custom.md +++ b/assets/docs/sources/tutorial/4_module_custom.md @@ -2,19 +2,21 @@ -## 自定义下载前后的回调函数 +## 自定义下载事件的回调函数 ```python -def custom_download_callback(): +def custom_download_event_callback(): """ - 该函数演示自定义下载时的回调函数 + 该函数演示如何自定义下载事件的回调函数。 + 公开下载 API 不接收 callback 参数；需要响应下载事件时，请自定义 Downloader。 """ # jmcomic的下载功能由 JmModuleConfig.CLASS_DOWNLOADER 这个类来负责执行 - # 这个类默认是 JmDownloader，继承了DownloadCallback - # 你可以写一个自定义类，继承JmDownloader，覆盖属于DownloadCallback的方法，来实现自定义回调 + # 这个类默认是 JmDownloader，继承了 DownloadCallback + # DownloadCallback 定义了 Downloader 内部各类下载事件的回调方法 + # 你可以继承 JmDownloader 并覆盖相应方法来响应事件 class MyDownloader(JmDownloader): - # 覆盖 album 下载完成后的回调 + # 覆盖 album 下载完成事件的回调函数 def after_album(self, album: JmAlbumDetail): print(f'album下载完毕: {album}') pass @@ -192,4 +194,4 @@ def custom_exception_listener(): # 注册监听器/回调 # 这个异常类（或者这个异常的子类）的实例将要被raise前，你的listener方法会被调用 JmModuleConfig.register_exception_listener(etype, listener) -``` \ No newline at end of file +``` diff --git a/requirements-dev.txt b/requirements-dev.txt index 07ed2aa11..f079b387c 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -5,6 +5,6 @@ Pillow psutil pycryptodome requests -plugin_jm_server +jm-view-server zhconv -img2pdf \ No newline at end of file +img2pdf diff --git a/src/jmcomic/__init__.py b/src/jmcomic/__init__.py index 121313b69..516de4e4a 100644 --- a/src/jmcomic/__init__.py +++ b/src/jmcomic/__init__.py @@ -2,8 +2,9 @@ # 被依赖方 <--- 使用方 # config <--- entity <--- toolkit <--- client <--- option <--- downloader -__version__ = '2.7.2' +__version__ = '2.7.3' +from .jm_task_context import * from .api import * from .jm_plugin import * from .jm_feature import * diff --git a/src/jmcomic/api.py b/src/jmcomic/api.py index 722d47834..b1d8ab838 100644 --- a/src/jmcomic/api.py +++ b/src/jmcomic/api.py @@ -1,26 +1,18 @@ import asyncio -import inspect from .jm_downloader import * +from .jm_task_context import bind_jm_task_context, jm_task_context __DOWNLOAD_API_RET = DownloadResult -async def _invoke_async_callback(callback, entity, downloader): - if callback is None: - return None - - is_async_callback = ( - inspect.iscoroutinefunction(callback) - or inspect.iscoroutinefunction(getattr(callback, '__call__', None)) - ) - if is_async_callback: - return await callback(entity, downloader) - - result = await asyncio.to_thread(callback, entity, downloader) - if inspect.isawaitable(result): - return await result - return result +def _download_type(download_api) -> str: + name = getattr(download_api, '__name__', download_api.__class__.__name__) + if name.endswith('_async'): + name = name[:-6] + if name.startswith('download_'): + name = name[9:] + return name def download_batch(download_api, @@ -48,21 +40,24 @@ def download_batch(download_api, result = BatchResult() + download_type = _download_type(download_api) + def _safe_download(aid): """batch 内部的单任务包装：确保异常被收集而非静默丢失""" - 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 + 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=_safe_download, + apply_each_obj_func=bind_jm_task_context(_safe_download), wait_finish=True ) @@ -72,7 +67,7 @@ def _safe_download(aid): def download_album(jm_album_id, option=None, downloader=None, - callback=None, + *, check_exception=True, extra=None, ) -> Union[__DOWNLOAD_API_RET, Set[__DOWNLOAD_API_RET]]: @@ -84,8 +79,9 @@ def download_album(jm_album_id, :param jm_album_id: 本子的禁漫车号 :param option: 下载选项 :param downloader: 下载器类 - :param callback: 返回值回调函数，可以拿到 album 和 downloader - :param check_exception: 是否检查异常, 如果为True，会检查downloader是否有下载异常，并上抛PartialDownloadFailedException + :param check_exception: 仅当 jm_album_id 是单个 ID 时生效。为 True 时检查 downloader 中的部分下载失败， + 并上抛 PartialDownloadFailedException。多 ID 调用会转交 download_batch，此参数不生效； + 请检查 BatchResult.failed，或自行封装 download_batch 实现所需的批量异常策略。 :param extra: 下载特性（Feature），下载时动态挂载的附加行为上下文。会自动根据上下文（如 album/photo 来源）自适应参数行为。支持单个 Feature、FeatureChain、或列表 :return: 对于的本子实体类，下载器（如果是上述的批量情况，返回值为download_batch的返回值） """ @@ -93,41 +89,42 @@ 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) - with new_downloader(option, downloader) as dler: - # 注册 Feature 及来源，由 downloader 在 after_album 钩子中自动执行 - dler.add_features(extra, 'download_album') - album = dler.download_album(jm_album_id) + with jm_task_context(download_type='album', jm_id=str(jm_album_id)): + with new_downloader(option, downloader) as dler: + # 注册 Feature 及来源，由 downloader 在 after_album 钩子中自动执行 + dler.add_features(extra, 'download_album') + album = dler.download_album(jm_album_id) - if callback is not None: - callback(album, dler) - if check_exception: - dler.raise_if_has_exception() - return DownloadResult(album, dler) + if check_exception: + dler.raise_if_has_exception() + return DownloadResult(album, dler) def download_photo(jm_photo_id, option=None, downloader=None, - callback=None, + *, check_exception=True, extra=None, ): """ - 下载一个章节（photo），参数同 download_album + 下载一个章节（photo），参数同 download_album。 + + check_exception 仅当 jm_photo_id 是单个 ID 时生效。多 ID 场景请检查 + BatchResult.failed，或自行封装 download_batch 处理批量异常。 """ if not isinstance(jm_photo_id, (str, int)): return download_batch(download_photo, jm_photo_id, option, downloader, extra=extra) - with new_downloader(option, downloader) as dler: - # 注册 Feature 及来源，由 downloader 在 after_photo 钩子中自动执行 - dler.add_features(extra, 'download_photo') - photo = dler.download_photo(jm_photo_id) + with jm_task_context(download_type='photo', jm_id=str(jm_photo_id)): + with new_downloader(option, downloader) as dler: + # 注册 Feature 及来源，由 downloader 在 after_photo 钩子中自动执行 + dler.add_features(extra, 'download_photo') + photo = dler.download_photo(jm_photo_id) - if callback is not None: - callback(photo, dler) - if check_exception: - dler.raise_if_has_exception() - return DownloadResult(photo, dler) + if check_exception: + dler.raise_if_has_exception() + return DownloadResult(photo, dler) def new_downloader(option=None, downloader=None) -> JmDownloader: @@ -177,7 +174,7 @@ def new_async_downloader(option=None, downloader=None): async def download_album_async(jm_album_id, option=None, downloader=None, - callback=None, + *, check_exception=True, extra=None, ): @@ -185,8 +182,9 @@ async def download_album_async(jm_album_id, 异步下载一个本子（album），包含其所有的章节（photo）。 - 支持批量下载（当 jm_album_id 为可迭代对象时） - - callback 支持同步函数和异步函数 - 返回 (album, downloader) 元组，其中 downloader 的网络和线程池资源已关闭，仅用于读取下载结果 + - check_exception 仅当 jm_album_id 是单个 ID 时生效。多 ID 场景请检查 BatchResult.failed， + 或自行封装 download_batch_async 处理批量异常 """ if not isinstance(jm_album_id, (str, int)): return await download_batch_async(download_album_async, @@ -196,28 +194,29 @@ async def download_album_async(jm_album_id, extra=extra ) - async with new_async_downloader(option, downloader) as dler: - dler.add_features(extra, 'download_album') - album = await dler.download_album(jm_album_id) + with jm_task_context(download_type='album', jm_id=str(jm_album_id)): + async with new_async_downloader(option, downloader) as dler: + dler.add_features(extra, 'download_album') + album = await dler.download_album(jm_album_id) - await _invoke_async_callback(callback, album, dler) - if check_exception: - dler.raise_if_has_exception() + if check_exception: + dler.raise_if_has_exception() - return DownloadResult(album, dler) + return DownloadResult(album, dler) async def download_photo_async(jm_photo_id, option=None, downloader=None, - callback=None, + *, check_exception=True, extra=None, ): """ 异步下载一个章节（photo）。 - callback 支持同步函数和异步函数。 返回的 downloader 已关闭网络和线程池资源，仅用于读取下载结果。 + check_exception 仅当 jm_photo_id 是单个 ID 时生效。多 ID 场景请检查 + BatchResult.failed，或自行封装 download_batch_async 处理批量异常。 """ if not isinstance(jm_photo_id, (str, int)): return await download_batch_async(download_photo_async, @@ -227,15 +226,15 @@ async def download_photo_async(jm_photo_id, extra=extra ) - async with new_async_downloader(option, downloader) as dler: - dler.add_features(extra, 'download_photo') - photo = await dler.download_photo(jm_photo_id) + with jm_task_context(download_type='photo', jm_id=str(jm_photo_id)): + async with new_async_downloader(option, downloader) as dler: + dler.add_features(extra, 'download_photo') + photo = await dler.download_photo(jm_photo_id) - await _invoke_async_callback(callback, photo, dler) - if check_exception: - dler.raise_if_has_exception() + if check_exception: + dler.raise_if_has_exception() - return DownloadResult(photo, dler) + return DownloadResult(photo, dler) async def download_batch_async(download_api, @@ -253,9 +252,14 @@ async def download_batch_async(download_api, option = JmModuleConfig.option_class().default() jm_ids = list(dict.fromkeys(JmcomicText.parse_to_jm_id(jmid) for jmid in jm_id_iter)) + 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_api(jmid, option, downloader, **kwargs) for jmid in jm_ids), + *(_download_one(jmid) for jmid in jm_ids), return_exceptions=True, ) @@ -263,7 +267,8 @@ async def download_batch_async(download_api, result = BatchResult() for jmid, r in zip(jm_ids, results): if isinstance(r, BaseException): - jm_log('async.batch.failed', f'批量下载失败: [{jmid}], 异常: [{r}]', r) + 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) diff --git a/src/jmcomic/cl.py b/src/jmcomic/cl.py index 82b8868ea..c5dbf3113 100644 --- a/src/jmcomic/cl.py +++ b/src/jmcomic/cl.py @@ -1,9 +1,11 @@ import warnings + from .cli import * + warnings.warn( "The 'jmcomic.cl' module is deprecated and renamed to 'jmcomic.cli'. " - "Please update your imports. It will be removed in a future version.", + "Please update your imports. It will be removed in version 2.7.4.", DeprecationWarning, - stacklevel=2 + stacklevel=2, ) diff --git a/src/jmcomic/jm_async_downloader.py b/src/jmcomic/jm_async_downloader.py index b2ff905bb..ee0bef025 100644 --- a/src/jmcomic/jm_async_downloader.py +++ b/src/jmcomic/jm_async_downloader.py @@ -16,6 +16,7 @@ from .jm_entity import JmAlbumDetail, JmPhotoDetail, JmImageDetail from .jm_toolkit import JmImageTool from .jm_config import jm_log +from .jm_task_context import bind_jm_task_context from .jm_option import JmOption @@ -56,6 +57,14 @@ def __init__(self, # 核心下载流程 — 对齐 sync JmDownloader # ====================================================================== + 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, + ) + async def download_album(self, album_id) -> JmAlbumDetail: """对齐 sync JmDownloader.download_album""" album = await self.client.get_album_detail(album_id) @@ -162,10 +171,8 @@ async def _download_single_image(self, image: JmImageDetail): img_bytes = img_resp.content # 提交到线程池解密并保存 - loop = asyncio.get_running_loop() if decode_image and image.scramble_id: - await loop.run_in_executor( - self._decode_pool, + await self._run_in_decode_pool( self._decode_and_save, img_bytes, int(image.scramble_id), @@ -183,8 +190,7 @@ async def _download_single_image(self, image: JmImageDetail): if qi != -1: img_url = img_url[:qi] need_convert = suffix_not_equal(img_url, img_save_path) - await loop.run_in_executor( - self._decode_pool, + await self._run_in_decode_pool( self._save_raw, img_bytes, img_save_path, @@ -227,28 +233,22 @@ def _save_raw(image_bytes, save_path, need_convert=False): # ====================================================================== async def before_album(self, album: JmAlbumDetail): - loop = asyncio.get_running_loop() - await loop.run_in_executor(self._decode_pool, super().before_album, album) + await self._run_in_decode_pool(super().before_album, album) async def after_album(self, album: JmAlbumDetail): - loop = asyncio.get_running_loop() - await loop.run_in_executor(self._decode_pool, super().after_album, album) + await self._run_in_decode_pool(super().after_album, album) async def before_photo(self, photo: JmPhotoDetail): - loop = asyncio.get_running_loop() - await loop.run_in_executor(self._decode_pool, super().before_photo, photo) + await self._run_in_decode_pool(super().before_photo, photo) async def after_photo(self, photo: JmPhotoDetail): - loop = asyncio.get_running_loop() - await loop.run_in_executor(self._decode_pool, super().after_photo, photo) + await self._run_in_decode_pool(super().after_photo, photo) async def before_image(self, image: JmImageDetail, img_save_path: str): - loop = asyncio.get_running_loop() - await loop.run_in_executor(self._decode_pool, super().before_image, image, img_save_path) + await self._run_in_decode_pool(super().before_image, image, img_save_path) async def after_image(self, image: JmImageDetail, img_save_path: str): - loop = asyncio.get_running_loop() - await loop.run_in_executor(self._decode_pool, super().after_image, image, img_save_path) + await self._run_in_decode_pool(super().after_image, image, img_save_path) def shutdown(self): """关闭解密线程池""" diff --git a/src/jmcomic/jm_client_impl.py b/src/jmcomic/jm_client_impl.py index 8db3932d4..f0410f42b 100644 --- a/src/jmcomic/jm_client_impl.py +++ b/src/jmcomic/jm_client_impl.py @@ -1,5 +1,7 @@ from threading import Lock +from .jm_task_context import bind_jm_task_context + from .jm_client_interface import * @@ -1159,7 +1161,7 @@ def get_future(self, cache_key, task): # after future done, remove it from future_dict. # cache depends on self.client instead of self.future_dict - future = self.FutureWrapper(self.executors.submit(task), + future = self.FutureWrapper(self.executors.submit(bind_jm_task_context(task)), after_done_callback=lambda: self.future_dict.pop(cache_key, None) ) diff --git a/src/jmcomic/jm_config.py b/src/jmcomic/jm_config.py index 6079d8c87..a1ad3ed50 100644 --- a/src/jmcomic/jm_config.py +++ b/src/jmcomic/jm_config.py @@ -1,11 +1,47 @@ from __future__ import annotations import logging + from common import time_stamp, field_cache, ProxyBuilder +from .jm_task_context import JM_TASK_CONTEXT, get_jm_task_context + jm_logger = logging.getLogger('jmcomic') +class JmLogFormatter(logging.Formatter): + """Format the stable correlation fields from a JM task context.""" + + @staticmethod + def _task_context_prefix(context) -> str: + if not isinstance(context, dict): + return '' + + fields = [] + task_id = context.get('task_id') + if task_id is None: + # 没有 task_id 时，默认不打印 task_context，保持日志简洁 + return '' + fields.append(f'task_id={task_id}') + + # 内置字段 + download_type = context.get('download_type') + jm_id = context.get('jm_id') + if download_type is not None and jm_id is not None: + fields.append(f'{download_type}={jm_id}') + elif download_type is not None: + fields.append(f'download_type={download_type}') + elif jm_id is not None: + fields.append(f'jm_id={jm_id}') + + return f'[{"; ".join(fields)}] ' if fields else '' + + def format(self, record): + # noinspection PyTypeChecker + record.jm_task_context_prefix = self._task_context_prefix(getattr(record, JM_TASK_CONTEXT.name, None)) + return super().format(record) + + def shuffled(lines): from random import shuffle from common import str_to_list @@ -19,7 +55,7 @@ def setup_default_jm_logger(): if not jm_logger.handlers: import sys handler = logging.StreamHandler(sys.stdout) - formatter = logging.Formatter('[%(asctime)s] [%(threadName)s]:【%(topic)s】%(message)s', datefmt='%Y-%m-%d %H:%M:%S') + formatter = JmLogFormatter(JmModuleConfig.VAR_LOG_FMT, datefmt='%Y-%m-%d %H:%M:%S') handler.setFormatter(formatter) jm_logger.addHandler(handler) jm_logger.setLevel(logging.INFO) @@ -30,7 +66,10 @@ def default_jm_logging(topic: str, msg, e: BaseException | None = None): if isinstance(msg, BaseException): e = msg msg = str(msg) - extra = {'topic': topic} + extra = { + 'topic': topic, + JM_TASK_CONTEXT.name: get_jm_task_context(), + } if e is not None: jm_logger.error(msg, extra=extra, exc_info=e) else: @@ -103,7 +142,7 @@ class JmMagicConstants: APP_TOKEN_SECRET_2 = '18comicAPPContent' APP_DATA_SECRET = '185Hcomic3PAPP7R' API_DOMAIN_SERVER_SECRET = 'diosfjckwpqpdfjkvnqQjsik' - APP_VERSION = '2.0.28' + APP_VERSION = '2.0.30' # 模块级别共用配置 @@ -259,6 +298,8 @@ class JmModuleConfig: # 当发生 oserror: [Errno 36] File name too long 时， # 把文件名限制在指定个字符以内 VAR_FILE_NAME_LENGTH_LIMIT = 100 + # 日志格式 + VAR_LOG_FMT = '[%(asctime)s] [%(threadName)s]:%(jm_task_context_prefix)s【%(topic)s】%(message)s' # 默认不打印 @classmethod def downloader_class(cls): @@ -565,8 +606,8 @@ def register_exception_listener(cls, etype, listener): disable_jm_log = JmModuleConfig.disable_jm_log -class PrettyFormatter(logging.Formatter): - """带 ANSI 颜色的日志格式化器，按 topic 前缀分配颜色""" +class PrettyFormatter(JmLogFormatter): + """带 ANSI 颜色的日志格式化器，按 topic 前缀分配颜色。""" TOPIC_COLORS = { 'album': '\033[1;36m', # 青色加粗 — 本子级别 @@ -581,7 +622,10 @@ class PrettyFormatter(logging.Formatter): RESET = '\033[0m' def __init__(self): - super().__init__(fmt='[%(asctime)s] %(message)s', datefmt='%H:%M:%S') + super().__init__( + fmt='[%(asctime)s] %(jm_task_context_prefix)s%(message)s', + datefmt='%H:%M:%S', + ) def format(self, record): topic = getattr(record, 'topic', '') @@ -590,7 +634,6 @@ def format(self, record): elif record.levelno >= logging.WARNING: color = self.WARN_COLOR else: - # 按 topic 前缀匹配颜色 color = next( (c for prefix, c in self.TOPIC_COLORS.items() if topic.startswith(prefix)), diff --git a/src/jmcomic/jm_downloader.py b/src/jmcomic/jm_downloader.py index 986196a7d..f760d5724 100644 --- a/src/jmcomic/jm_downloader.py +++ b/src/jmcomic/jm_downloader.py @@ -1,6 +1,7 @@ from typing import NamedTuple from .jm_option import * +from .jm_task_context import bind_jm_task_context def catch_exception(func): @@ -366,6 +367,8 @@ def execute_on_condition(self, if count_real == 0: return + apply = bind_jm_task_context(apply) + if count_batch >= count_real: # 一个图/章节 对应 一个线程 multi_thread_launcher( diff --git a/src/jmcomic/jm_feature.py b/src/jmcomic/jm_feature.py index 9021d113e..16d464efa 100644 --- a/src/jmcomic/jm_feature.py +++ b/src/jmcomic/jm_feature.py @@ -17,6 +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 class Feature: @@ -33,6 +34,11 @@ class Feature: export_zip: 'PluginFeature' export_long_img: 'PluginFeature' + @property + def jm_task_context(self) -> dict: + """Return the current invocation's isolated task-context snapshot.""" + return get_jm_task_context() + def should_invoke(self, feature_from: str, when: str) -> bool: """ 判断在当前钩子(when)下，根据来源(feature_from)，是否应该执行。 diff --git a/src/jmcomic/jm_plugin.py b/src/jmcomic/jm_plugin.py index 0373e762a..93721e8ca 100644 --- a/src/jmcomic/jm_plugin.py +++ b/src/jmcomic/jm_plugin.py @@ -3,6 +3,7 @@ """ from .jm_option import * +from .jm_task_context import bind_jm_task_context, get_jm_task_context class PluginValidationException(Exception): @@ -27,6 +28,11 @@ def invoke(self, **kwargs) -> None: """ raise NotImplementedError + @property + def jm_task_context(self) -> dict: + """Return the current invocation's isolated task-context snapshot.""" + return get_jm_task_context() + @classmethod def build(cls, option: JmOption) -> 'JmOptionPlugin': """ @@ -672,7 +678,7 @@ def main(self): # 一个收藏夹一个线程，导出收藏夹数据到文件 multi_thread_launcher( iter_objs=folders.items(), - apply_each_obj_func=self.handle_folder, + apply_each_obj_func=bind_jm_task_context(self.handle_folder), ) if not self.zip_enable: @@ -992,21 +998,34 @@ def invoke(self, if self.running: return - # 服务器的代码位于一个独立库：plugin_jm_server，需要独立安装 - # 源代码仓库：https://github.com/hect0x7/plugin-jm-server + # 服务器的代码位于独立库 jm-view-server，需要独立安装 + # 源代码仓库：https://github.com/hect0x7/jm-view-server try: # noinspection PyUnresolvedReferences - import plugin_jm_server - self.log(f'当前使用plugin_jm_server版本: {plugin_jm_server.__version__}') - except ImportError: - self.warning_lib_not_install('plugin_jm_server') - return + import jm_view_server as jm_server_lib + except ModuleNotFoundError as e: + if e.name != 'jm_view_server': + raise + + try: + # 兼容尚未迁移的 plugin_jm_server <= 0.2.3 + # noinspection PyUnresolvedReferences + import plugin_jm_server as jm_server_lib + except ModuleNotFoundError as legacy_error: + if legacy_error.name != 'plugin_jm_server': + raise + self.warning_lib_not_install('jm-view-server') + return + + self.log('检测到旧包 plugin_jm_server，建议升级到 jm-view-server', 'warning') + + self.log(f'当前使用 jm-view-server 版本: {jm_server_lib.__version__}') # 核心函数，启动服务器，会阻塞当前线程 def blocking_run_server(): self.server_thread = current_thread() self.enter_wait_list() - server = plugin_jm_server.JmServer(base_dir, password, **kwargs) + server = jm_server_lib.JmServer(base_dir, password, **kwargs) # run方法会阻塞当前线程直到flask退出 server.run(**run) diff --git a/src/jmcomic/jm_task_context.py b/src/jmcomic/jm_task_context.py new file mode 100644 index 000000000..57a5e9e1d --- /dev/null +++ b/src/jmcomic/jm_task_context.py @@ -0,0 +1,59 @@ +import inspect +from contextlib import contextmanager +from contextvars import ContextVar +from functools import wraps +from types import MappingProxyType +from typing import Callable, Mapping, Optional + + +__all__ = ( + 'JM_TASK_CONTEXT', + 'jm_task_context', + 'get_jm_task_context', + 'bind_jm_task_context', +) + + +_EMPTY_TASK_CONTEXT = MappingProxyType({}) +JM_TASK_CONTEXT: ContextVar[Mapping] = ContextVar( + 'jm_task_context', + default=_EMPTY_TASK_CONTEXT, +) + + +def get_jm_task_context() -> dict: + """Return a mutable snapshot of the current JM task context.""" + return dict(JM_TASK_CONTEXT.get()) + + +@contextmanager +def jm_task_context(**fields): + """Temporarily add fields to the current JM task context.""" + context = get_jm_task_context() + context.update(fields) + token = JM_TASK_CONTEXT.set(MappingProxyType(context)) + try: + yield + finally: + JM_TASK_CONTEXT.reset(token) + + +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))): + raise TypeError('bind_jm_task_context only supports synchronous callables') + + snapshot = MappingProxyType(dict( + get_jm_task_context() if context is None else context + )) + + @wraps(func) + def wrapped(*args, **kwargs): + token = JM_TASK_CONTEXT.set(snapshot) + try: + return func(*args, **kwargs) + finally: + JM_TASK_CONTEXT.reset(token) + + return wrapped diff --git a/tests/test_jmcomic/test_jm_api.py b/tests/test_jmcomic/test_jm_api.py index d9af3553a..bf46f99cc 100644 --- a/tests/test_jmcomic/test_jm_api.py +++ b/tests/test_jmcomic/test_jm_api.py @@ -1,8 +1,44 @@ +import asyncio +import inspect +from unittest.mock import AsyncMock, patch + from test_jmcomic import * class Test_Api(JmTestConfigurable): + def test_callback_is_not_public_download_api(self): + for download_api in ( + download_album, + download_photo, + download_album_async, + download_photo_async, + ): + parameters = inspect.signature(download_api).parameters + self.assertNotIn('callback', parameters) + self.assertEqual( + parameters['check_exception'].kind, + inspect.Parameter.KEYWORD_ONLY, + ) + + def test_multi_id_shortcuts_do_not_forward_check_exception(self): + for download_api in (download_album, download_photo): + expected = BatchResult() + with patch('jmcomic.api.download_batch', return_value=expected) as batch: + actual = download_api(['123', '456'], check_exception=False) + + self.assertIs(actual, expected) + self.assertNotIn('check_exception', batch.call_args.kwargs) + + for download_api in (download_album_async, download_photo_async): + expected = BatchResult() + batch = AsyncMock(return_value=expected) + with patch('jmcomic.api.download_batch_async', new=batch): + actual = asyncio.run(download_api(['123', '456'], check_exception=False)) + + self.assertIs(actual, expected) + self.assertNotIn('check_exception', batch.call_args.kwargs) + def test_download_photo_by_id(self): """ 测试jmcomic模块的api的使用 diff --git a/tests/test_jmcomic/test_jm_async_api.py b/tests/test_jmcomic/test_jm_async_api.py index 47aa774a1..995b3663d 100644 --- a/tests/test_jmcomic/test_jm_async_api.py +++ b/tests/test_jmcomic/test_jm_async_api.py @@ -6,7 +6,6 @@ """ import asyncio -import threading from test_jmcomic import * from jmcomic import ( download_album_async, download_photo_async, download_batch_async, @@ -22,13 +21,6 @@ class Test_Async_Api(JmAsyncTestConfigurable): def test_async_download_photo_by_id(self): """测试 download_photo_async：验证返回值与同步版本保持一致""" photo_id = '438516' - callback_result = {} - caller_thread = threading.get_ident() - - def callback(photo, downloader): - callback_result['photo'] = photo - callback_result['downloader'] = downloader - callback_result['thread'] = threading.get_ident() # sync sync_photo, sync_dler = download_photo(photo_id, self.option) @@ -36,26 +28,16 @@ def callback(photo, downloader): async_photo, async_dler = asyncio.run(download_photo_async( photo_id, self.option, - callback=callback, )) self.assertIsInstance(async_dler, JmAsyncDownloader, 'downloader 必须是异步版本') self.assertIsInstance(async_photo, JmPhotoDetail, '返回值必须包含 photo') self.assert_sync_async_equal(sync_photo.photo_id, async_photo.photo_id, 'photo.photo_id') - self.assertIs(callback_result['photo'], async_photo) - self.assertIs(callback_result['downloader'], async_dler) - self.assertNotEqual(callback_result['thread'], caller_thread, '同步 callback 不应阻塞事件循环线程') self.assertIsNone(async_dler.client, '顶层异步下载结束后 downloader client 应已关闭') def test_async_download_album_by_id(self): """测试 download_album_async：验证返回值与同步版本保持一致""" album_id = '438516' - callback_result = {} - - async def callback(album, downloader): - await asyncio.sleep(0) - callback_result['album'] = album - callback_result['downloader'] = downloader # sync sync_album, sync_dler = download_album(album_id, self.option) @@ -63,14 +45,11 @@ async def callback(album, downloader): async_album, async_dler = asyncio.run(download_album_async( album_id, self.option, - callback=callback, )) self.assertIsInstance(async_dler, JmAsyncDownloader, 'downloader 必须是异步版本') self.assertIsInstance(async_album, JmAlbumDetail, '返回值必须包含 album') self.assert_album_equal(sync_album, async_album) - self.assertIs(callback_result['album'], async_album) - self.assertIs(callback_result['downloader'], async_dler) self.assertIsNone(async_dler.client, '顶层异步下载结束后 downloader client 应已关闭') def test_async_batch(self): diff --git a/tests/test_jmcomic/test_jm_cli.py b/tests/test_jmcomic/test_jm_cli.py index 3db1b2370..bdbd913de 100644 --- a/tests/test_jmcomic/test_jm_cli.py +++ b/tests/test_jmcomic/test_jm_cli.py @@ -2,7 +2,7 @@ from io import StringIO from unittest.mock import patch -from jmcomic.cl import JmcomicUI, JmViewUI +from jmcomic.cli import JmcomicUI, JmViewUI class Test_Cli(JmTestConfigurable): @@ -10,6 +10,12 @@ class Test_Cli(JmTestConfigurable): album_id = '350234' + def test_cl_deprecated(self): + with self.assertWarnsRegex(DeprecationWarning, r'removed in version 2\.7\.4'): + from jmcomic.cl import JmcomicUI as DeprecatedJmcomicUI + + self.assertIs(DeprecatedJmcomicUI, JmcomicUI) + # ========== jmcomic 命令测试 ========== def test_jmcomic_parse_album_id(self): diff --git a/tests/test_jmcomic/test_jm_task_context.py b/tests/test_jmcomic/test_jm_task_context.py new file mode 100644 index 000000000..e052640f5 --- /dev/null +++ b/tests/test_jmcomic/test_jm_task_context.py @@ -0,0 +1,482 @@ +import asyncio +import logging +import threading +import unittest +from concurrent.futures import ThreadPoolExecutor +from queue import Queue + +from jmcomic import ( + BaseDownloader, + Feature, + JmAsyncDownloader, + JmDownloader, + JmModuleConfig, + JmOption, + JmOptionPlugin, + JM_TASK_CONTEXT, + PhotoConcurrentFetcherProxy, + PrettyFormatter, + 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, +) + + +class ListHandler(logging.Handler): + + def __init__(self): + super().__init__() + self.records = [] + + def emit(self, record): + self.records.append(record) + + +class Test_Jm_Task_Context(unittest.TestCase): + + 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()) + + with jm_task_context(session_id='outer'): + self.assertEqual({'session_id': 'outer'}, get_jm_task_context()) + with jm_task_context(session_id='inner', task_id='task'): + self.assertEqual( + {'session_id': 'inner', 'task_id': 'task'}, + get_jm_task_context(), + ) + self.assertEqual({'session_id': 'outer'}, get_jm_task_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({}, get_jm_task_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)) + with jm_task_context(session_id='B'): + future_b = executor.submit(bind_jm_task_context(get_jm_task_context)) + + empty = executor.submit(get_jm_task_context) + + self.assertEqual({'session_id': 'A'}, future_a.result()) + self.assertEqual({'session_id': 'B'}, future_b.result()) + self.assertEqual({}, empty.result()) + + def test_bind_rejects_async_callable(self): + async def async_work(): + return None + + with self.assertRaisesRegex(TypeError, 'synchronous callables'): + bind_jm_task_context(async_work) + + def test_default_logger_and_custom_executors_can_read_context(self): + handler = ListHandler() + original_handlers = jm_logger.handlers[:] + original_executor = JmModuleConfig.EXECUTOR_LOG + jm_logger.handlers[:] = [handler] + + try: + with jm_task_context(session_id='logger'): + default_jm_logging('test.context', 'message') + + self.assertEqual( + {'session_id': 'logger'}, + getattr(handler.records[0], JM_TASK_CONTEXT.name), + ) + + captured = [] + + def executor_two_args(topic, msg): + captured.append((topic, msg, get_jm_task_context())) + + JmModuleConfig.EXECUTOR_LOG = executor_two_args + with jm_task_context(session_id='custom-2'): + jm_log('test.custom', 'message') + + error = ValueError('failed') + + def executor_three_args(topic, msg, e): + captured.append((topic, msg, e, get_jm_task_context())) + + JmModuleConfig.EXECUTOR_LOG = executor_three_args + with jm_task_context(session_id='custom-3'): + jm_log('test.custom.error', 'message', error) + + self.assertEqual( + ('test.custom', 'message', {'session_id': 'custom-2'}), + captured[0], + ) + self.assertEqual( + ('test.custom.error', 'message', error, {'session_id': 'custom-3'}), + captured[1], + ) + finally: + JmModuleConfig.EXECUTOR_LOG = original_executor + jm_logger.handlers[:] = original_handlers + + def test_pretty_formatter_uses_topic_colors(self): + formatter = PrettyFormatter() + + def make_record(topic, context, level=logging.INFO): + record = logging.LogRecord( + name='jmcomic', + level=level, + pathname=__file__, + lineno=1, + msg=topic, + args=(), + exc_info=None, + ) + record.topic = topic + record.jm_task_context = context + return record + + task_context = { + 'task_id': 'task-A', + 'download_type': 'album', + 'jm_id': '1', + } + self.assertTrue(formatter.format( + make_record('album.before', task_context) + ).startswith(formatter.TOPIC_COLORS['album'])) + self.assertTrue(formatter.format( + make_record('image.before', task_context) + ).startswith(formatter.TOPIC_COLORS['image'])) + + self.assertTrue(formatter.format( + make_record('image.failed', task_context, logging.ERROR) + ).startswith(formatter.ERROR_COLOR)) + self.assertTrue(formatter.format( + make_record('image.warning', task_context, logging.WARNING) + ).startswith(formatter.WARN_COLOR)) + + def test_public_downloads_add_task_context_to_downloader_logs(self): + class FakeSyncDownloader: + + def __init__(self, _option): + pass + + def __enter__(self): + return self + + def __exit__(self, *_args): + pass + + def add_features(self, *_args): + pass + + def download_album(self, album_id): + jm_log('album.before', 'message') + return album_id + + def raise_if_has_exception(self): + pass + + class FakeAsyncDownloader: + + def __init__(self, _option): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + pass + + def add_features(self, *_args): + pass + + async def download_photo(self, photo_id): + jm_log('photo.before', 'message') + return photo_id + + def raise_if_has_exception(self): + pass + + handler = ListHandler() + original_handlers = jm_logger.handlers[:] + jm_logger.handlers[:] = [handler] + try: + jm_log('plugin.usage_log.log', 'unrelated background log') + with jm_task_context(session_id='session-1', task_id='task-1'): + download_album( + '123', + option=object(), + downloader=FakeSyncDownloader, + ) + asyncio.run(download_photo_async( + '456', + option=object(), + downloader=FakeAsyncDownloader, + )) + finally: + jm_logger.handlers[:] = original_handlers + + task_records = [ + record + for record in handler.records + if (getattr(record, JM_TASK_CONTEXT.name, None) or {}).get('session_id') == 'session-1' + ] + self.assertEqual( + ['album.before', 'photo.before'], + [record.topic for record in task_records], + ) + self.assertEqual( + { + 'session_id': 'session-1', + 'task_id': 'task-1', + 'download_type': 'album', + 'jm_id': '123', + }, + task_records[0].jm_task_context, + ) + self.assertEqual( + { + 'session_id': 'session-1', + 'task_id': 'task-1', + 'download_type': 'photo', + 'jm_id': '456', + }, + task_records[1].jm_task_context, + ) + + def test_plugin_invocation_can_read_current_task_context(self): + observed = [] + + class TaskAwarePlugin(JmOptionPlugin): + plugin_key = 'test_task_aware' + + def invoke(self): + observed.append(self.jm_task_context) + self.log('plugin-message') + + option = object.__new__(JmOption) + handler = ListHandler() + original_handlers = jm_logger.handlers[:] + jm_logger.handlers[:] = [handler] + try: + with jm_task_context(session_id='plugin-session', task_id='plugin-task'): + option.invoke_plugin(TaskAwarePlugin, None, {}, {}) + finally: + jm_logger.handlers[:] = original_handlers + + self.assertEqual([{ + 'session_id': 'plugin-session', + 'task_id': 'plugin-task', + }], observed) + self.assertEqual( + [ + 'plugin.invoke', + 'plugin.test_task_aware', + ], + [record.topic for record in handler.records], + ) + self.assertEqual( + [{ + 'session_id': 'plugin-session', + 'task_id': 'plugin-task', + }] * 2, + [record.jm_task_context for record in handler.records], + ) + + def test_feature_invocation_can_read_current_task_context(self): + observed = [] + + class TaskAwareFeature(Feature): + + def invoke(self, _option, feature_from, when, **_kwargs): + observed.append((feature_from, when, self.jm_task_context)) + + downloader = BaseDownloader(object()) + downloader.add_features(TaskAwareFeature(), 'download_album') + + with jm_task_context(session_id='feature-session', task_id='feature-task'): + downloader._invoke_features_for('after_album') + + self.assertEqual([( + 'download_album', + 'after_album', + { + 'session_id': 'feature-session', + 'task_id': 'feature-task', + }, + )], observed) + + def test_sync_batch_binds_parent_and_item_context(self): + def fake_download(jmid, _option, _downloader, **_kwargs): + context = get_jm_task_context() + return ( + str(jmid), + context['session_id'], + context['download_type'], + context['jm_id'], + ) + + with jm_task_context(session_id='sync-batch'): + result = download_batch(fake_download, ['1', '2'], option=object()) + + self.assertEqual( + { + ('1', 'sync-batch', 'fake_download', '1'), + ('2', 'sync-batch', 'fake_download', '2'), + }, + set(result), + ) + self.assertEqual({}, get_jm_task_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() + return context['session_id'], context['jm_id'], str(jmid) + + def run_session(session_id, jmid): + with jm_task_context(session_id=session_id): + return download_batch( + fake_download, + [jmid], + option=object(), + ) + + with ThreadPoolExecutor(max_workers=2) as executor: + future_a = executor.submit(run_session, 'session-A', '1') + future_b = executor.submit(run_session, 'session-B', '2') + + self.assertEqual( + {('session-A', '1', '1')}, + set(future_a.result()), + ) + self.assertEqual( + {('session-B', '2', '2')}, + set(future_b.result()), + ) + + def test_batch_failure_log_keeps_item_context(self): + handler = ListHandler() + original_handlers = jm_logger.handlers[:] + jm_logger.handlers[:] = [handler] + + def fail(jmid, _option, _downloader, **_kwargs): + raise ValueError(f'failed-{jmid}') + + try: + with jm_task_context(session_id='failed-session'): + result = download_batch(fail, ['404'], option=object()) + finally: + jm_logger.handlers[:] = original_handlers + + self.assertIn('404', result.failed) + self.assertEqual(1, len(handler.records)) + self.assertEqual( + { + 'session_id': 'failed-session', + 'download_type': 'fail', + 'jm_id': '404', + }, + handler.records[0].jm_task_context, + ) + + def test_sync_downloader_propagates_both_threading_branches(self): + downloader = object.__new__(JmDownloader) + BaseDownloader.__init__(downloader, object()) + + 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, + ) + + contexts = [observed.get_nowait() for _ in range(3)] + self.assertEqual( + [{ + 'session_id': f'workers-{count_batch}', + }] * 3, + contexts, + ) + + 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() + return ( + str(jmid), + context['session_id'], + context['download_type'], + context['jm_id'], + ) + + with jm_task_context(session_id='async-batch'): + batch_result = await download_batch_async( + fake_download, + ['1', '2'], + option=object(), + ) + + downloader = object.__new__(JmAsyncDownloader) + downloader._decode_pool = ThreadPoolExecutor(max_workers=1) + try: + with jm_task_context(session_id='decode-pool'): + executor_context = await downloader._run_in_decode_pool( + get_jm_task_context + ) + + loop = asyncio.get_running_loop() + leaked_context = await loop.run_in_executor( + downloader._decode_pool, + get_jm_task_context, + ) + finally: + downloader.shutdown() + + return batch_result, executor_context, leaked_context + + batch_result, executor_context, leaked_context = asyncio.run(run_test()) + + self.assertEqual( + { + ('1', 'async-batch', 'fake_download', '1'), + ('2', 'async-batch', 'fake_download', '2'), + }, + set(batch_result), + ) + self.assertEqual({'session_id': 'decode-pool'}, executor_context) + self.assertEqual({}, leaked_context) + + def test_photo_concurrent_proxy_propagates_context(self): + class FakeClient: + pass + + proxy = PhotoConcurrentFetcherProxy(FakeClient(), max_workers=1) + try: + with jm_task_context(session_id='client-proxy'): + future = proxy.get_future('context', get_jm_task_context) + + self.assertEqual({'session_id': 'client-proxy'}, future.result()) + finally: + proxy.executors.shutdown(wait=True) + + +if __name__ == '__main__': + unittest.main() diff --git a/usage/workflow_download.py b/usage/workflow_download.py index 4f8121225..22a117a31 100644 --- a/usage/workflow_download.py +++ b/usage/workflow_download.py @@ -1,5 +1,5 @@ from jmcomic import * -from jmcomic.cl import JmcomicUI +from jmcomic.cli import JmcomicUI # 下方填入你要下载的本子的id，一行一个，每行的首尾可以有空白字符 jm_albums = '''