client

JmHtmlClient

Bases: AbstractJmClient

Methods:

Name	Description
build_search_url	构建网页搜索/分类的URL
get_jm_html	请求禁漫网页的入口
login	返回response响应对象
raise_request_error	请求如果失败，统一由该方法抛出异常
require_resp_success_else_raise	:param resp: 响应对象
search	网页搜索API

Source code in src/jmcomic/jm_client_impl.py

class JmHtmlClient(AbstractJmClient):
    client_key = 'html'

    func_to_cache = ['search', 'fetch_detail_entity']

    API_SEARCH = '/search/photos'
    API_CATEGORY = '/albums'

    def add_favorite_album(self,
                           album_id,
                           folder_id='0',
                           ):
        data = {
            'album_id': album_id,
            'fid': folder_id,
        }

        resp = self.get_jm_html(
            '/ajax/favorite_album',
            data=data,
        )

        res = resp.json()

        if res['status'] != 1:
            msg = parse_unicode_escape_text(res['msg'])
            error_msg = PatternTool.match_or_default(msg, JmcomicText.pattern_ajax_favorite_msg, msg)
            # 此圖片已經在您最喜愛的清單！

            self.raise_request_error(
                resp,
                error_msg
            )

        return resp

    def get_album_detail(self, album_id) -> JmAlbumDetail:
        return self.fetch_detail_entity(album_id, 'album')

    def get_photo_detail(self,
                         photo_id,
                         fetch_album=True,
                         fetch_scramble_id=True,
                         ) -> JmPhotoDetail:
        photo: JmPhotoDetail = self.fetch_detail_entity(photo_id, 'photo')

        # 一并获取该章节的所处本子
        if fetch_album:
            photo.from_album = self.get_album_detail(photo.album_id)

        return photo

    def fetch_detail_entity(self, jmid, prefix) -> DetailType:
        # 参数校验
        jmid = JmcomicText.parse_to_jm_id(jmid)

        # 请求
        resp = self.get_jm_html(f"/{prefix}/{jmid}")

        # 用 JmcomicText 解析 html，返回实体类
        if prefix == 'album':
            return JmcomicText.analyse_jm_album_html(resp.text)

        if prefix == 'photo':
            return JmcomicText.analyse_jm_photo_html(resp.text)

        raise ValueError(f"不支持的 prefix 类型: {prefix}")

    def search(self,
               search_query: str,
               page: int,
               main_tag: int,
               order_by: str,
               time: str,
               category: str,
               sub_category: Optional[str],
               ) -> JmSearchPage:
        """
        网页搜索API
        """
        params = {
            'main_tag': main_tag,
            'search_query': search_query,
            'page': page,
            'o': order_by,
            't': time,
        }

        url = self.build_search_url(self.API_SEARCH, category, sub_category)

        resp = self.get_jm_html(
            self.append_params_to_url(url, params),
            allow_redirects=True,
        )

        # 检查是否发生了重定向
        # 因为如果搜索的是禁漫车号，会直接跳转到本子详情页面
        if resp.redirect_count != 0 and '/album/' in resp.url:
            album = JmcomicText.analyse_jm_album_html(resp.text)
            return JmSearchPage.wrap_single_album(album)
        else:
            return JmPageTool.parse_html_to_search_page(resp.text)

    @classmethod
    def build_search_url(cls, base: str, category: str, sub_category: Optional[str]):
        """
        构建网页搜索/分类的URL

        示例：
        :param base: "/search/photos"
        :param category CATEGORY_DOUJIN
        :param sub_category SUB_DOUJIN_CG
        :return "/search/photos/doujin/sub/CG"
        """
        if category == JmMagicConstants.CATEGORY_ALL:
            return base

        if sub_category is None:
            return f'{base}/{category}'
        else:
            return f'{base}/{category}/sub/{sub_category}'

    def categories_filter(self,
                          page: int,
                          time: str,
                          category: str,
                          order_by: str,
                          sub_category: Optional[str] = None,
                          ) -> JmCategoryPage:
        params = {
            'page': page,
            'o': order_by,
            't': time,
        }

        url = self.build_search_url(self.API_CATEGORY, category, sub_category)

        resp = self.get_jm_html(
            self.append_params_to_url(url, params),
            allow_redirects=True,
        )

        return JmPageTool.parse_html_to_category_page(resp.text)

    # -- 帐号管理 --

    def login(self,
              username,
              password,
              id_remember='on',
              login_remember='on',
              ):
        """
        返回response响应对象
        """

        data = {
            'username': username,
            'password': password,
            'id_remember': id_remember,
            'login_remember': login_remember,
            'submit_login': '',
        }

        resp = self.post('/login',
                         data=data,
                         allow_redirects=False,
                         )

        if resp.status_code != 200:
            ExceptionTool.raises_resp(f'登录失败，状态码为{resp.status_code}', resp)

        orig_cookies = self.get_meta_data('cookies') or {}
        new_cookies = dict(resp.cookies)
        # 重复登录下存在bug，AVS会丢失
        if 'AVS' in orig_cookies and 'AVS' not in new_cookies:
            return resp

        self['cookies'] = new_cookies
        self._username = username

        return resp

    def favorite_folder(self,
                        page=1,
                        order_by=JmMagicConstants.ORDER_BY_LATEST,
                        folder_id='0',
                        username='',
                        ) -> JmFavoritePage:
        if username == '':
            ExceptionTool.require_true(self._username is not None, 'favorite_folder方法需要传username参数')
            username = self._username

        resp = self.get_jm_html(
            f'/user/{username}/favorite/albums',
            params={
                'page': page,
                'o': order_by,
                'folder': folder_id,
            }
        )

        return JmPageTool.parse_html_to_favorite_page(resp.text)

    # noinspection PyTypeChecker
    def get_username_from_cookies(self) -> str:
        # cookies = self.get_meta_data('cookies', None)
        # if not cookies:
        #     ExceptionTool.raises('未登录，无法获取到对应的用户名，请给favorite方法传入username参数')
        # 解析cookies，可能需要用到 phpserialize，比较麻烦，暂不实现
        pass

    def get_jm_html(self, url, require_200=True, **kwargs):
        """
        请求禁漫网页的入口
        """
        resp = self.get(url, **kwargs)

        if require_200 is True and resp.status_code != 200:
            # 检查是否是特殊的状态码（JmModuleConfig.JM_ERROR_STATUS_CODE）
            # 如果是，直接抛出异常
            self.check_special_http_code(resp)
            # 运行到这里说明上一步没有抛异常，说明是未知状态码，抛异常兜底处理
            self.raise_request_error(resp)

        # 检查请求是否成功
        self.require_resp_success_else_raise(resp, url)

        return resp

    def update_request_with_specify_domain(self, kwargs: dict, domain: Optional[str], is_image=False):
        if is_image:
            return

        latest_headers = kwargs.get('headers', None)
        base_headers = self.get_meta_data('headers', None) or JmModuleConfig.new_html_headers(domain)
        base_headers.update(latest_headers or {})
        kwargs['headers'] = base_headers

    @classmethod
    def raise_request_error(cls, resp, msg: Optional[str] = None):
        """
        请求如果失败，统一由该方法抛出异常
        """
        if msg is None:
            msg_tail = '' if JmModuleConfig.FLAG_DUMP_HTML_ON_REGEX_ERROR else ('，可通过设置 '
                                                                                'JmModuleConfig.FLAG_DUMP_HTML_ON_REGEX_ERROR = '
                                                                                'True 将响应文本保存到文件')
            msg = f"请求失败，" \
                  f"响应状态码为{resp.status_code}，" \
                  f"URL=[{resp.url}]，" \
                  + (f"响应文本=[{resp.text}]" if len(resp.text) < 200 else
                     f'响应文本过长(len={len(resp.text)})，不打印{msg_tail}'
                     )

            # 当 flag 开启时，将过长的响应文本持久化到文件，方便debug
            if len(resp.text) >= 200 and JmModuleConfig.FLAG_DUMP_HTML_ON_REGEX_ERROR:
                dump_path = ExceptionTool.dump_html_to_file(resp.text, msg)
                if dump_path is not None:
                    msg += f'\n已将响应文本持久化到文件: [{dump_path}]'

        ExceptionTool.raises_resp(msg, resp)

    def album_comment(self,
                      video_id,
                      comment,
                      originator='',
                      status='true',
                      comment_id=None,
                      **kwargs,
                      ) -> JmAlbumCommentResp:
        data = {
            'video_id': video_id,
            'comment': comment,
            'originator': originator,
            'status': status,
        }

        # 处理回复评论
        if comment_id is not None:
            data.pop('status')
            data['comment_id'] = comment_id
            data['is_reply'] = 1
            data['forum_subject'] = 1

        jm_log('album.comment',
               f'{video_id}: [{comment}]' +
               (f' to ({comment_id})' if comment_id is not None else '')
               )

        resp = self.post('/ajax/album_comment', data=data)

        ret = JmAlbumCommentResp(resp)
        jm_log('album.comment', f'{video_id}: [{comment}] ← ({ret.model().cid})')

        return ret

    @classmethod
    def require_resp_success_else_raise(cls, resp, url: str):
        """
        :param resp: 响应对象
        :param url: /photo/12412312
        """
        resp_url: str = resp.url

        # 1. 是否是特殊的内容
        cls.check_special_text(resp)

        # 2. 检查响应发送重定向，重定向url是否表示错误网页，即 /error/xxx
        if resp.redirect_count == 0 or '/error/' not in resp_url:
            return

        # 3. 检查错误类型
        def match_case(error_path):
            return resp_url.endswith(error_path) and not url.endswith(error_path)

        # 3.1 album_missing
        if match_case('/error/album_missing'):
            ExceptionTool.raise_missing(resp, JmcomicText.parse_to_jm_id(url))

        # 3.2 user_missing
        if match_case('/error/user_missing'):
            ExceptionTool.raises_resp('此用戶名稱不存在，或者你没有登录，請再次確認使用名稱', resp)

        # 3.3 invalid_module
        if match_case('/error/invalid_module'):
            ExceptionTool.raises_resp('發生了無法預期的錯誤。若問題持續發生，請聯繫客服支援', resp)

    @classmethod
    def check_special_text(cls, resp):
        html = resp.text
        url = resp.url

        if len(html) > 500:
            return

        for content, reason in JmModuleConfig.JM_ERROR_RESPONSE_TEXT.items():
            if content not in html:
                continue

            cls.raise_request_error(
                resp,
                f'{reason}({content})'
                + (f': {url}' if url is not None else '')
            )

    @classmethod
    def check_special_http_code(cls, resp):
        code = resp.status_code
        url = resp.url

        error_msg = JmModuleConfig.JM_ERROR_STATUS_CODE.get(int(code), None)
        if error_msg is None:
            return

        cls.raise_request_error(
            resp,
            f"请求失败，"
            f"响应状态码为{code}，"
            f'原因为: [{error_msg}], '
            + (f'URL=[{url}]' if url is not None else '')
        )

build_search_url(base, category, sub_category) classmethod

构建网页搜索/分类的URL

示例：

Parameters:

Name	Type	Description	Default
base	str	"/search/photos"	required

Source code in src/jmcomic/jm_client_impl.py

@classmethod
def build_search_url(cls, base: str, category: str, sub_category: Optional[str]):
    """
    构建网页搜索/分类的URL

    示例：
    :param base: "/search/photos"
    :param category CATEGORY_DOUJIN
    :param sub_category SUB_DOUJIN_CG
    :return "/search/photos/doujin/sub/CG"
    """
    if category == JmMagicConstants.CATEGORY_ALL:
        return base

    if sub_category is None:
        return f'{base}/{category}'
    else:
        return f'{base}/{category}/sub/{sub_category}'

get_jm_html(url, require_200=True, **kwargs)

请求禁漫网页的入口

Source code in src/jmcomic/jm_client_impl.py

def get_jm_html(self, url, require_200=True, **kwargs):
    """
    请求禁漫网页的入口
    """
    resp = self.get(url, **kwargs)

    if require_200 is True and resp.status_code != 200:
        # 检查是否是特殊的状态码（JmModuleConfig.JM_ERROR_STATUS_CODE）
        # 如果是，直接抛出异常
        self.check_special_http_code(resp)
        # 运行到这里说明上一步没有抛异常，说明是未知状态码，抛异常兜底处理
        self.raise_request_error(resp)

    # 检查请求是否成功
    self.require_resp_success_else_raise(resp, url)

    return resp

login(username, password, id_remember='on', login_remember='on')

返回response响应对象

Source code in src/jmcomic/jm_client_impl.py

def login(self,
          username,
          password,
          id_remember='on',
          login_remember='on',
          ):
    """
    返回response响应对象
    """

    data = {
        'username': username,
        'password': password,
        'id_remember': id_remember,
        'login_remember': login_remember,
        'submit_login': '',
    }

    resp = self.post('/login',
                     data=data,
                     allow_redirects=False,
                     )

    if resp.status_code != 200:
        ExceptionTool.raises_resp(f'登录失败，状态码为{resp.status_code}', resp)

    orig_cookies = self.get_meta_data('cookies') or {}
    new_cookies = dict(resp.cookies)
    # 重复登录下存在bug，AVS会丢失
    if 'AVS' in orig_cookies and 'AVS' not in new_cookies:
        return resp

    self['cookies'] = new_cookies
    self._username = username

    return resp

raise_request_error(resp, msg=None) classmethod

请求如果失败，统一由该方法抛出异常

Source code in src/jmcomic/jm_client_impl.py

@classmethod
def raise_request_error(cls, resp, msg: Optional[str] = None):
    """
    请求如果失败，统一由该方法抛出异常
    """
    if msg is None:
        msg_tail = '' if JmModuleConfig.FLAG_DUMP_HTML_ON_REGEX_ERROR else ('，可通过设置 '
                                                                            'JmModuleConfig.FLAG_DUMP_HTML_ON_REGEX_ERROR = '
                                                                            'True 将响应文本保存到文件')
        msg = f"请求失败，" \
              f"响应状态码为{resp.status_code}，" \
              f"URL=[{resp.url}]，" \
              + (f"响应文本=[{resp.text}]" if len(resp.text) < 200 else
                 f'响应文本过长(len={len(resp.text)})，不打印{msg_tail}'
                 )

        # 当 flag 开启时，将过长的响应文本持久化到文件，方便debug
        if len(resp.text) >= 200 and JmModuleConfig.FLAG_DUMP_HTML_ON_REGEX_ERROR:
            dump_path = ExceptionTool.dump_html_to_file(resp.text, msg)
            if dump_path is not None:
                msg += f'\n已将响应文本持久化到文件: [{dump_path}]'

    ExceptionTool.raises_resp(msg, resp)

require_resp_success_else_raise(resp, url) classmethod

Parameters:

Name	Type	Description	Default
resp		响应对象	required
url	str	/photo/12412312	required

Source code in src/jmcomic/jm_client_impl.py

@classmethod
def require_resp_success_else_raise(cls, resp, url: str):
    """
    :param resp: 响应对象
    :param url: /photo/12412312
    """
    resp_url: str = resp.url

    # 1. 是否是特殊的内容
    cls.check_special_text(resp)

    # 2. 检查响应发送重定向，重定向url是否表示错误网页，即 /error/xxx
    if resp.redirect_count == 0 or '/error/' not in resp_url:
        return

    # 3. 检查错误类型
    def match_case(error_path):
        return resp_url.endswith(error_path) and not url.endswith(error_path)

    # 3.1 album_missing
    if match_case('/error/album_missing'):
        ExceptionTool.raise_missing(resp, JmcomicText.parse_to_jm_id(url))

    # 3.2 user_missing
    if match_case('/error/user_missing'):
        ExceptionTool.raises_resp('此用戶名稱不存在，或者你没有登录，請再次確認使用名稱', resp)

    # 3.3 invalid_module
    if match_case('/error/invalid_module'):
        ExceptionTool.raises_resp('發生了無法預期的錯誤。若問題持續發生，請聯繫客服支援', resp)

search(search_query, page, main_tag, order_by, time, category, sub_category)

网页搜索API

Source code in src/jmcomic/jm_client_impl.py

def search(self,
           search_query: str,
           page: int,
           main_tag: int,
           order_by: str,
           time: str,
           category: str,
           sub_category: Optional[str],
           ) -> JmSearchPage:
    """
    网页搜索API
    """
    params = {
        'main_tag': main_tag,
        'search_query': search_query,
        'page': page,
        'o': order_by,
        't': time,
    }

    url = self.build_search_url(self.API_SEARCH, category, sub_category)

    resp = self.get_jm_html(
        self.append_params_to_url(url, params),
        allow_redirects=True,
    )

    # 检查是否发生了重定向
    # 因为如果搜索的是禁漫车号，会直接跳转到本子详情页面
    if resp.redirect_count != 0 and '/album/' in resp.url:
        album = JmcomicText.analyse_jm_album_html(resp.text)
        return JmSearchPage.wrap_single_album(album)
    else:
        return JmPageTool.parse_html_to_search_page(resp.text)

JmApiClient

Bases: AbstractJmClient

Methods:

Name	Description
add_favorite_album	移动端没有提供folder_id参数
categories_filter	移动端不支持 sub_category
fetch_detail_entity	Fetches a JM entity (album or chapter) by its JM ID and returns it as an instance of clazz.
fetch_photo_additional_field	获取章节的额外信息
fetch_scramble_id	请求scramble_id
get_scramble_id	带有缓存的fetch_scramble_id，缓存位于 JmModuleConfig.SCRAMBLE_CACHE
login	{
raise_if_resp_should_retry	该方法会判断resp返回值是否是json格式，
require_resp_status_ok	检查返回数据中的status字段是否为ok
require_resp_success	:param resp: 响应对象
search	移动端暂不支持 category和sub_category
setting	禁漫app的setting请求

Source code in src/jmcomic/jm_client_impl.py

class JmApiClient(AbstractJmClient):
    client_key = 'api'
    func_to_cache = ['search', 'fetch_detail_entity']

    API_SEARCH = '/search'
    API_CATEGORIES_FILTER = '/categories/filter'
    API_ALBUM = '/album'
    API_CHAPTER = '/chapter'
    API_SCRAMBLE = '/chapter_view_template'
    API_FAVORITE = '/favorite'

    def search(self,
               search_query: str,
               page: int,
               main_tag: int,
               order_by: str,
               time: str,
               category: str,
               sub_category: Optional[str],
               ) -> JmSearchPage:
        """
        移动端暂不支持 category和sub_category
        """
        params = {
            'main_tag': main_tag,
            'search_query': search_query,
            'page': page,
            'o': order_by,
            't': time,
        }

        resp = self.req_api(self.append_params_to_url(self.API_SEARCH, params))

        # 直接搜索禁漫车号，发生重定向的响应数据 resp.model_data
        # {
        #   "search_query": "310311",
        #   "total": 1,
        #   "redirect_aid": "310311",
        #   "content": []
        # }
        data = resp.model_data
        if data.get('redirect_aid', None) is not None:
            aid = data.redirect_aid
            return JmSearchPage.wrap_single_album(self.get_album_detail(aid))

        return JmPageTool.parse_api_to_search_page(data)

    def categories_filter(self,
                          page: int,
                          time: str,
                          category: str,
                          order_by: str,
                          sub_category: Optional[str] = None,
                          ):
        """
        移动端不支持 sub_category
        """
        # o: mv, mv_m, mv_w, mv_t
        o = f'{order_by}_{time}' if time != JmMagicConstants.TIME_ALL else order_by

        params = {
            'page': page,
            'order': '',  # 该参数为空
            'c': category,
            'o': o,
        }

        resp = self.req_api(self.append_params_to_url(self.API_CATEGORIES_FILTER, params))

        return JmPageTool.parse_api_to_search_page(resp.model_data)

    def get_album_detail(self, album_id) -> JmAlbumDetail:
        return self.fetch_detail_entity(album_id,
                                        JmModuleConfig.album_class(),
                                        )

    def get_photo_detail(self,
                         photo_id,
                         fetch_album=True,
                         fetch_scramble_id=True,
                         ) -> JmPhotoDetail:
        photo: JmPhotoDetail = self.fetch_detail_entity(photo_id,
                                                        JmModuleConfig.photo_class(),
                                                        )
        if fetch_album or fetch_scramble_id:
            self.fetch_photo_additional_field(photo, fetch_album, fetch_scramble_id)

        return photo

    def get_scramble_id(self, photo_id, album_id=None):
        """
        带有缓存的fetch_scramble_id，缓存位于 JmModuleConfig.SCRAMBLE_CACHE
        """
        cache = JmModuleConfig.SCRAMBLE_CACHE
        if photo_id in cache:
            return cache[photo_id]

        if album_id is not None and album_id in cache:
            return cache[album_id]

        scramble_id = self.fetch_scramble_id(photo_id)
        cache[photo_id] = scramble_id
        if album_id is not None:
            cache[album_id] = scramble_id

        return scramble_id

    def fetch_detail_entity(self, jmid, clazz: Type[DetailType]) -> DetailType:
        """
        Fetches a JM entity (album or chapter) by its JM ID and returns it as an instance of `clazz`.

        Parameters:
            jmid (str | int): JM ID or value parseable to a JM ID.
            clazz (type): Entity class to parse the response into (e.g., `JmAlbumDetail` or a chapter/detail class).

        Returns:
            object: An instance of `clazz` populated from the API response data.

        Raises:
            Exception: Raised via ExceptionTool.raise_missing if the API response lacks required data.
        """
        jmid = JmcomicText.parse_to_jm_id(jmid)
        url = self.API_ALBUM if issubclass(clazz, JmAlbumDetail) else self.API_CHAPTER
        resp = self.req_api(self.append_params_to_url(
            url,
            {
                'id': jmid
            })
        )

        if not resp.encoded_data or resp.res_data.get('name') is None:
            ExceptionTool.raise_missing(resp, jmid)

        return JmApiAdaptTool.parse_entity(resp.res_data, clazz)

    def fetch_scramble_id(self, photo_id):
        """
        请求scramble_id
        """
        photo_id: str = JmcomicText.parse_to_jm_id(photo_id)
        resp = self.req_api(self.append_params_to_url(
            self.API_SCRAMBLE,
            params={
                'id': photo_id,
                'mode': 'vertical',
                'page': '0',
                'app_img_shunt': '1',
                'express': 'off',
                'v': time_stamp(),
            }),
            require_success=False,
        )

        scramble_id = PatternTool.match_or_default(resp.text,
                                                   JmcomicText.pattern_html_album_scramble_id,
                                                   None,
                                                   )
        if scramble_id is None:
            jm_log('api.scramble', f'未匹配到scramble_id，响应文本：{resp.text}')
            scramble_id = str(JmMagicConstants.SCRAMBLE_220980)

        return scramble_id

    def fetch_photo_additional_field(self, photo: JmPhotoDetail, fetch_album: bool, fetch_scramble_id: bool):
        """
        获取章节的额外信息
        1. scramble_id
        2. album
        如果都需要获取，会排队，效率低

        todo: 改进实现 (had polished by FutureClientProxy)
        1. 直接开两个线程跑
        2. 开两个线程，但是开之前检查重复性
        3. 线程池，也要检查重复性
        23做法要改不止一处地方
        """
        if fetch_album:
            photo.from_album = self.get_album_detail(photo.album_id)

        if fetch_scramble_id:
            # 同album的scramble_id相同
            photo.scramble_id = self.get_scramble_id(photo.photo_id, photo.album_id)

    def setting(self) -> JmApiResp:
        """
        禁漫app的setting请求
        """
        resp = self.req_api('/setting')

        # 检查禁漫最新的版本号
        setting_ver = str(resp.model_data.jm3_version)
        # 禁漫接口的版本 > jmcomic库内置版本
        if (
                JmModuleConfig.FLAG_USE_VERSION_NEWER_IF_BEHIND
                and JmcomicText.compare_versions(setting_ver, JmMagicConstants.APP_VERSION) == 1
        ):
            jm_log('api.setting', f'change APP_VERSION from [{JmMagicConstants.APP_VERSION}] to [{setting_ver}]')
            JmMagicConstants.APP_VERSION = setting_ver

        return resp

    def login(self,
              username,
              password,
              ) -> JmApiResp:
        """
        {
          "uid": "123",
          "username": "x",
          "email": "x",
          "emailverified": "yes",
          "photo": "x",
          "fname": "",
          "gender": "x",
          "message": "Welcome x!",
          "coin": 123,
          "album_favorites": 123,
          "s": "x",
          "level_name": "x",
          "level": 1,
          "nextLevelExp": 123,
          "exp": "123",
          "expPercent": 123,
          "badges": [],
          "album_favorites_max": 123
        }

        """
        resp = self.req_api('/login', False, data={
            'username': username,
            'password': password,
        })

        cookies = dict(resp.resp.cookies)
        cookies.update({'AVS': resp.res_data['s']})
        self['cookies'] = cookies

        return resp

    def favorite_folder(self,
                        page=1,
                        order_by=JmMagicConstants.ORDER_BY_LATEST,
                        folder_id='0',
                        username='',
                        ) -> JmFavoritePage:
        resp = self.req_api(
            self.API_FAVORITE,
            params={
                'page': page,
                'folder_id': folder_id,
                'o': order_by,
            }
        )

        return JmPageTool.parse_api_to_favorite_page(resp.model_data)

    def add_favorite_album(self,
                           album_id,
                           folder_id='0',
                           ):
        """
        移动端没有提供folder_id参数
        """
        resp = self.req_api(
            '/favorite',
            data={
                'aid': album_id,
            },
        )

        self.require_resp_status_ok(resp)

        return resp

    # noinspection PyMethodMayBeStatic
    def require_resp_status_ok(self, resp: JmApiResp):
        """
        检查返回数据中的status字段是否为ok
        """
        data = resp.model_data
        if data.status != 'ok':
            ExceptionTool.raises_resp(data.msg, resp)

    def req_api(self, url, get=True, require_success=True, **kwargs) -> JmApiResp:
        ts = self.decide_headers_and_ts(kwargs, url)

        if get:
            resp = self.get(url, **kwargs)
        else:
            resp = self.post(url, **kwargs)

        resp = JmApiResp(resp, ts)

        if require_success:
            self.require_resp_success(resp, url)

        return resp

    def update_request_with_specify_domain(self, kwargs: dict, domain: Optional[str], is_image=False):
        if is_image:
            # 设置APP端的图片请求headers
            kwargs['headers'] = {**JmModuleConfig.APP_HEADERS_TEMPLATE, **JmModuleConfig.APP_HEADERS_IMAGE}

    # noinspection PyMethodMayBeStatic
    def decide_headers_and_ts(self, kwargs, url):
        # 获取时间戳
        if url == self.API_SCRAMBLE:
            # /chapter_view_template
            # 这个接口很特殊，用的密钥 18comicAPPContent 而不是 18comicAPP
            # 如果用后者，则会返回403信息
            ts = time_stamp()
            token, tokenparam = JmCryptoTool.token_and_tokenparam(ts, secret=JmMagicConstants.APP_TOKEN_SECRET_2)

        elif JmModuleConfig.FLAG_USE_FIX_TIMESTAMP:
            ts, token, tokenparam = JmModuleConfig.get_fix_ts_token_tokenparam()

        else:
            ts = time_stamp()
            token, tokenparam = JmCryptoTool.token_and_tokenparam(ts)

        # 设置headers
        headers = kwargs.get('headers', None) or JmModuleConfig.APP_HEADERS_TEMPLATE.copy()
        headers.update({
            'token': token,
            'tokenparam': tokenparam,
        })
        kwargs['headers'] = headers

        return ts

    @classmethod
    def require_resp_success(cls, resp: JmApiResp, url: Optional[str] = None):
        """

        :param resp: 响应对象
        :param url: 请求路径，例如 /setting
        """
        resp.require_success()

        # 1. 检查是否 album_missing
        # json: {'code': 200, 'data': []}
        # 最新api已不存在这种情况，无需检查
        # 2. 是否是特殊的内容
        # 暂无

    def raise_if_resp_should_retry(self, resp, is_image):
        """
        该方法会判断resp返回值是否是json格式，
        如果不是，大概率是禁漫内部异常，需要进行重试

        由于完整的json格式校验会有性能开销，所以只做简单的检查，
        只校验第一个有效字符是不是 '{'，如果不是，就认为异常数据，需要重试
        """
        resp = super().raise_if_resp_should_retry(resp, is_image)

        if isinstance(resp, JmResp):
            # 不对包装过的resp对象做校验，包装者自行校验
            # 例如图片请求
            return resp

        code = resp.status_code
        if code >= 500:
            msg = JmModuleConfig.JM_ERROR_STATUS_CODE.get(code, f'HTTP状态码: {code}')
            ExceptionTool.raises_resp(f"禁漫API异常响应, {msg}", resp)

        url = resp.request.url

        if self.API_SCRAMBLE in url:
            # /chapter_view_template 这个接口不是返回json数据，不做检查
            return resp

        text = resp.text
        for char in text:
            if char not in (' ', '\n', '\t'):
                # 找到第一个有效字符
                ExceptionTool.require_true(
                    char == '{',
                    f'请求不是json格式，强制重试！响应文本: [{JmcomicText.limit_text(text, 200)}]'
                )
                return resp

        ExceptionTool.raises_resp(f'响应无数据！request_url=[{url}]', resp)

    def after_init(self):
        # 自动更新禁漫API域名
        if JmModuleConfig.FLAG_API_CLIENT_AUTO_UPDATE_DOMAIN:
            new_server_list = self.fetch_latest_api_domain_for_module()
            self.update_old_api_domain(new_server_list)

        # 保证拥有cookies，因为移动端要求必须携带cookies，否则会直接跳转同一本子【禁漫娘】
        if JmModuleConfig.FLAG_API_CLIENT_REQUIRE_COOKIES:
            self.ensure_have_cookies()

    client_update_domain_lock = Lock()

    def req_api_domain_server(self, url):
        resp = self.postman.get(url)
        text: str = resp.text
        # 去掉开头非ascii字符
        while text and not text[0].isascii():
            text = text[1:]
        res_json = JmCryptoTool.decode_resp_data(text, '', JmMagicConstants.API_DOMAIN_SERVER_SECRET)
        res_data = json_loads(res_json)

        # 检查返回值
        if not res_data.get('Server', None):
            jm_log('api.update_domain.empty',
                   f'获取禁漫最新API域名失败, 返回值: {res_json}')
            return None
        else:
            return res_data['Server']

    def update_old_api_domain(self, new_server_list: List[str]):
        if new_server_list and sorted(self.domain_list) == sorted(JmModuleConfig.DOMAIN_API_LIST):
            self.domain_list = new_server_list

    def fetch_latest_api_domain_for_module(self):
        if JmModuleConfig.DOMAIN_API_UPDATED_LIST is not None:
            return JmModuleConfig.DOMAIN_API_UPDATED_LIST

        with self.client_update_domain_lock:
            # double check
            if JmModuleConfig.DOMAIN_API_UPDATED_LIST is not None:
                return JmModuleConfig.DOMAIN_API_UPDATED_LIST

            # 遍历多个域名服务器
            for url in JmModuleConfig.API_URL_DOMAIN_SERVER_LIST:
                try:
                    # 获取域名列表
                    new_server_list = self.req_api_domain_server(url)
                    if new_server_list is None:
                        continue
                    old_server_list = JmModuleConfig.DOMAIN_API_LIST
                    jm_log('api.update_domain.success',
                           f'获取到最新的API域名，替换jmcomic内置域名：(new){new_server_list} ---→ (old){old_server_list}'
                           )
                    JmModuleConfig.DOMAIN_API_UPDATED_LIST = new_server_list
                    return new_server_list
                except Exception as e:
                    jm_log('api.update_domain.error',
                           f'通过[{url}]自动更新API域名失败，尝试下一个地址。'
                           f'可通过代码[JmModuleConfig.FLAG_API_CLIENT_AUTO_UPDATE_DOMAIN=False]关闭自动更新API域名. 异常： {e}'
                           )
                    continue

            # 走到这里，说明没有获取到域名更新
            # 为了本方法不被重复执行，把新域名字段修改为空列表
            # 空列表相当于一个done标识
            JmModuleConfig.DOMAIN_API_UPDATED_LIST = []
            return JmModuleConfig.DOMAIN_API_UPDATED_LIST

    client_init_cookies_lock = Lock()

    def ensure_have_cookies(self):
        if self.get_meta_data('cookies'):
            return

        with self.client_init_cookies_lock:
            if self.get_meta_data('cookies'):
                return

            self['cookies'] = self.get_cookies()

    @field_cache("APP_COOKIES", obj=JmModuleConfig)
    def get_cookies(self):
        resp = self.setting()
        cookies = dict(resp.resp.cookies)
        return cookies

add_favorite_album(album_id, folder_id='0')

移动端没有提供folder_id参数

Source code in src/jmcomic/jm_client_impl.py

def add_favorite_album(self,
                       album_id,
                       folder_id='0',
                       ):
    """
    移动端没有提供folder_id参数
    """
    resp = self.req_api(
        '/favorite',
        data={
            'aid': album_id,
        },
    )

    self.require_resp_status_ok(resp)

    return resp

categories_filter(page, time, category, order_by, sub_category=None)

移动端不支持 sub_category

Source code in src/jmcomic/jm_client_impl.py

def categories_filter(self,
                      page: int,
                      time: str,
                      category: str,
                      order_by: str,
                      sub_category: Optional[str] = None,
                      ):
    """
    移动端不支持 sub_category
    """
    # o: mv, mv_m, mv_w, mv_t
    o = f'{order_by}_{time}' if time != JmMagicConstants.TIME_ALL else order_by

    params = {
        'page': page,
        'order': '',  # 该参数为空
        'c': category,
        'o': o,
    }

    resp = self.req_api(self.append_params_to_url(self.API_CATEGORIES_FILTER, params))

    return JmPageTool.parse_api_to_search_page(resp.model_data)

fetch_detail_entity(jmid, clazz)

Fetches a JM entity (album or chapter) by its JM ID and returns it as an instance of clazz.

Parameters: jmid (str | int): JM ID or value parseable to a JM ID. clazz (type): Entity class to parse the response into (e.g., JmAlbumDetail or a chapter/detail class).

Returns: object: An instance of clazz populated from the API response data.

Raises: Exception: Raised via ExceptionTool.raise_missing if the API response lacks required data.

Source code in src/jmcomic/jm_client_impl.py

def fetch_detail_entity(self, jmid, clazz: Type[DetailType]) -> DetailType:
    """
    Fetches a JM entity (album or chapter) by its JM ID and returns it as an instance of `clazz`.

    Parameters:
        jmid (str | int): JM ID or value parseable to a JM ID.
        clazz (type): Entity class to parse the response into (e.g., `JmAlbumDetail` or a chapter/detail class).

    Returns:
        object: An instance of `clazz` populated from the API response data.

    Raises:
        Exception: Raised via ExceptionTool.raise_missing if the API response lacks required data.
    """
    jmid = JmcomicText.parse_to_jm_id(jmid)
    url = self.API_ALBUM if issubclass(clazz, JmAlbumDetail) else self.API_CHAPTER
    resp = self.req_api(self.append_params_to_url(
        url,
        {
            'id': jmid
        })
    )

    if not resp.encoded_data or resp.res_data.get('name') is None:
        ExceptionTool.raise_missing(resp, jmid)

    return JmApiAdaptTool.parse_entity(resp.res_data, clazz)

fetch_photo_additional_field(photo, fetch_album, fetch_scramble_id)

获取章节的额外信息 1. scramble_id 2. album 如果都需要获取，会排队，效率低

todo: 改进实现 (had polished by FutureClientProxy) 1. 直接开两个线程跑 2. 开两个线程，但是开之前检查重复性 3. 线程池，也要检查重复性 23做法要改不止一处地方

Source code in src/jmcomic/jm_client_impl.py

def fetch_photo_additional_field(self, photo: JmPhotoDetail, fetch_album: bool, fetch_scramble_id: bool):
    """
    获取章节的额外信息
    1. scramble_id
    2. album
    如果都需要获取，会排队，效率低

    todo: 改进实现 (had polished by FutureClientProxy)
    1. 直接开两个线程跑
    2. 开两个线程，但是开之前检查重复性
    3. 线程池，也要检查重复性
    23做法要改不止一处地方
    """
    if fetch_album:
        photo.from_album = self.get_album_detail(photo.album_id)

    if fetch_scramble_id:
        # 同album的scramble_id相同
        photo.scramble_id = self.get_scramble_id(photo.photo_id, photo.album_id)

fetch_scramble_id(photo_id)

请求scramble_id

Source code in src/jmcomic/jm_client_impl.py

def fetch_scramble_id(self, photo_id):
    """
    请求scramble_id
    """
    photo_id: str = JmcomicText.parse_to_jm_id(photo_id)
    resp = self.req_api(self.append_params_to_url(
        self.API_SCRAMBLE,
        params={
            'id': photo_id,
            'mode': 'vertical',
            'page': '0',
            'app_img_shunt': '1',
            'express': 'off',
            'v': time_stamp(),
        }),
        require_success=False,
    )

    scramble_id = PatternTool.match_or_default(resp.text,
                                               JmcomicText.pattern_html_album_scramble_id,
                                               None,
                                               )
    if scramble_id is None:
        jm_log('api.scramble', f'未匹配到scramble_id，响应文本：{resp.text}')
        scramble_id = str(JmMagicConstants.SCRAMBLE_220980)

    return scramble_id

get_scramble_id(photo_id, album_id=None)

带有缓存的fetch_scramble_id，缓存位于 JmModuleConfig.SCRAMBLE_CACHE

Source code in src/jmcomic/jm_client_impl.py

def get_scramble_id(self, photo_id, album_id=None):
    """
    带有缓存的fetch_scramble_id，缓存位于 JmModuleConfig.SCRAMBLE_CACHE
    """
    cache = JmModuleConfig.SCRAMBLE_CACHE
    if photo_id in cache:
        return cache[photo_id]

    if album_id is not None and album_id in cache:
        return cache[album_id]

    scramble_id = self.fetch_scramble_id(photo_id)
    cache[photo_id] = scramble_id
    if album_id is not None:
        cache[album_id] = scramble_id

    return scramble_id

login(username, password)

{ "uid": "123", "username": "x", "email": "x", "emailverified": "yes", "photo": "x", "fname": "", "gender": "x", "message": "Welcome x!", "coin": 123, "album_favorites": 123, "s": "x", "level_name": "x", "level": 1, "nextLevelExp": 123, "exp": "123", "expPercent": 123, "badges": [], "album_favorites_max": 123 }

Source code in src/jmcomic/jm_client_impl.py

def login(self,
          username,
          password,
          ) -> JmApiResp:
    """
    {
      "uid": "123",
      "username": "x",
      "email": "x",
      "emailverified": "yes",
      "photo": "x",
      "fname": "",
      "gender": "x",
      "message": "Welcome x!",
      "coin": 123,
      "album_favorites": 123,
      "s": "x",
      "level_name": "x",
      "level": 1,
      "nextLevelExp": 123,
      "exp": "123",
      "expPercent": 123,
      "badges": [],
      "album_favorites_max": 123
    }

    """
    resp = self.req_api('/login', False, data={
        'username': username,
        'password': password,
    })

    cookies = dict(resp.resp.cookies)
    cookies.update({'AVS': resp.res_data['s']})
    self['cookies'] = cookies

    return resp

raise_if_resp_should_retry(resp, is_image)

该方法会判断resp返回值是否是json格式， 如果不是，大概率是禁漫内部异常，需要进行重试

由于完整的json格式校验会有性能开销，所以只做简单的检查， 只校验第一个有效字符是不是 '{'，如果不是，就认为异常数据，需要重试

Source code in src/jmcomic/jm_client_impl.py

def raise_if_resp_should_retry(self, resp, is_image):
    """
    该方法会判断resp返回值是否是json格式，
    如果不是，大概率是禁漫内部异常，需要进行重试

    由于完整的json格式校验会有性能开销，所以只做简单的检查，
    只校验第一个有效字符是不是 '{'，如果不是，就认为异常数据，需要重试
    """
    resp = super().raise_if_resp_should_retry(resp, is_image)

    if isinstance(resp, JmResp):
        # 不对包装过的resp对象做校验，包装者自行校验
        # 例如图片请求
        return resp

    code = resp.status_code
    if code >= 500:
        msg = JmModuleConfig.JM_ERROR_STATUS_CODE.get(code, f'HTTP状态码: {code}')
        ExceptionTool.raises_resp(f"禁漫API异常响应, {msg}", resp)

    url = resp.request.url

    if self.API_SCRAMBLE in url:
        # /chapter_view_template 这个接口不是返回json数据，不做检查
        return resp

    text = resp.text
    for char in text:
        if char not in (' ', '\n', '\t'):
            # 找到第一个有效字符
            ExceptionTool.require_true(
                char == '{',
                f'请求不是json格式，强制重试！响应文本: [{JmcomicText.limit_text(text, 200)}]'
            )
            return resp

    ExceptionTool.raises_resp(f'响应无数据！request_url=[{url}]', resp)

require_resp_status_ok(resp)

检查返回数据中的status字段是否为ok

Source code in src/jmcomic/jm_client_impl.py

def require_resp_status_ok(self, resp: JmApiResp):
    """
    检查返回数据中的status字段是否为ok
    """
    data = resp.model_data
    if data.status != 'ok':
        ExceptionTool.raises_resp(data.msg, resp)

require_resp_success(resp, url=None) classmethod

Parameters:

Name	Type	Description	Default
resp	JmApiResp	响应对象	required
url	Optional[str]	请求路径，例如 /setting	None

Source code in src/jmcomic/jm_client_impl.py

@classmethod
def require_resp_success(cls, resp: JmApiResp, url: Optional[str] = None):
    """

    :param resp: 响应对象
    :param url: 请求路径，例如 /setting
    """
    resp.require_success()

search(search_query, page, main_tag, order_by, time, category, sub_category)

移动端暂不支持 category和sub_category

Source code in src/jmcomic/jm_client_impl.py

def search(self,
           search_query: str,
           page: int,
           main_tag: int,
           order_by: str,
           time: str,
           category: str,
           sub_category: Optional[str],
           ) -> JmSearchPage:
    """
    移动端暂不支持 category和sub_category
    """
    params = {
        'main_tag': main_tag,
        'search_query': search_query,
        'page': page,
        'o': order_by,
        't': time,
    }

    resp = self.req_api(self.append_params_to_url(self.API_SEARCH, params))

    # 直接搜索禁漫车号，发生重定向的响应数据 resp.model_data
    # {
    #   "search_query": "310311",
    #   "total": 1,
    #   "redirect_aid": "310311",
    #   "content": []
    # }
    data = resp.model_data
    if data.get('redirect_aid', None) is not None:
        aid = data.redirect_aid
        return JmSearchPage.wrap_single_album(self.get_album_detail(aid))

    return JmPageTool.parse_api_to_search_page(data)

setting()

禁漫app的setting请求

Source code in src/jmcomic/jm_client_impl.py

def setting(self) -> JmApiResp:
    """
    禁漫app的setting请求
    """
    resp = self.req_api('/setting')

    # 检查禁漫最新的版本号
    setting_ver = str(resp.model_data.jm3_version)
    # 禁漫接口的版本 > jmcomic库内置版本
    if (
            JmModuleConfig.FLAG_USE_VERSION_NEWER_IF_BEHIND
            and JmcomicText.compare_versions(setting_ver, JmMagicConstants.APP_VERSION) == 1
    ):
        jm_log('api.setting', f'change APP_VERSION from [{JmMagicConstants.APP_VERSION}] to [{setting_ver}]')
        JmMagicConstants.APP_VERSION = setting_ver

    return resp

AsyncJmcomicClient

异步客户端接口基类，对标 sync 的 JmcomicClient。

• 所有方法签名和 sync 版完全对齐
• 通过 REGISTRY_ASYNC_CLIENT 注册（配置项: client.async_impl）
• 由 JmOption.new_async_client() 工厂方法创建

Methods:

Name	Description
check_photo	检查 photo 的 from_album / page_arr / data_original_domain 是否齐全，
favorite_folder_gen	见 search_gen
search_gen	异步搜索结果的生成器。

Source code in src/jmcomic/jm_client_interface.py

class AsyncJmcomicClient:
    """
    异步客户端接口基类，对标 sync 的 JmcomicClient。

    - 所有方法签名和 sync 版完全对齐
    - 通过 REGISTRY_ASYNC_CLIENT 注册（配置项: client.async_impl）
    - 由 JmOption.new_async_client() 工厂方法创建
    """

    client_key = None

    # -- JmDetailClient --

    async def get_album_detail(self, album_id) -> JmAlbumDetail:
        raise NotImplementedError

    async def get_photo_detail(self,
                               photo_id,
                               fetch_album=True,
                               fetch_scramble_id=True,
                               ) -> JmPhotoDetail:
        raise NotImplementedError

    async def check_photo(self, photo: JmPhotoDetail):
        """
        检查 photo 的 from_album / page_arr / data_original_domain 是否齐全，
        缺失则请求补全。对齐 sync JmDetailClient.check_photo。
        """
        # 检查 from_album
        if photo.from_album is None:
            photo.from_album = await self.get_album_detail(photo.album_id)

        # 检查 page_arr 和 data_original_domain
        if photo.page_arr is None or photo.data_original_domain is None:
            new = await self.get_photo_detail(photo.photo_id, False)
            new.from_album = photo.from_album
            photo.__dict__.update(new.__dict__)

    # -- JmSearchAlbumClient --

    async def search(self,
                     search_query: str,
                     page: int,
                     main_tag: int,
                     order_by: str,
                     time: str,
                     category: str,
                     sub_category: Optional[str],
                     ) -> JmSearchPage:
        raise NotImplementedError

    async def search_site(self,
                          search_query: str,
                          page: int = 1,
                          order_by: str = JmMagicConstants.ORDER_BY_LATEST,
                          time: str = JmMagicConstants.TIME_ALL,
                          category: str = JmMagicConstants.CATEGORY_ALL,
                          sub_category: Optional[str] = None,
                          ):
        return await self.search(search_query, page, 0, order_by, time, category, sub_category)

    async def search_work(self,
                          search_query: str,
                          page: int = 1,
                          order_by: str = JmMagicConstants.ORDER_BY_LATEST,
                          time: str = JmMagicConstants.TIME_ALL,
                          category: str = JmMagicConstants.CATEGORY_ALL,
                          sub_category: Optional[str] = None,
                          ):
        return await self.search(search_query, page, 1, order_by, time, category, sub_category)

    async def search_author(self,
                            search_query: str,
                            page: int = 1,
                            order_by: str = JmMagicConstants.ORDER_BY_LATEST,
                            time: str = JmMagicConstants.TIME_ALL,
                            category: str = JmMagicConstants.CATEGORY_ALL,
                            sub_category: Optional[str] = None,
                            ):
        return await self.search(search_query, page, 2, order_by, time, category, sub_category)

    async def search_tag(self,
                         search_query: str,
                         page: int = 1,
                         order_by: str = JmMagicConstants.ORDER_BY_LATEST,
                         time: str = JmMagicConstants.TIME_ALL,
                         category: str = JmMagicConstants.CATEGORY_ALL,
                         sub_category: Optional[str] = None,
                         ):
        return await self.search(search_query, page, 3, order_by, time, category, sub_category)

    async def search_actor(self,
                           search_query: str,
                           page: int = 1,
                           order_by: str = JmMagicConstants.ORDER_BY_LATEST,
                           time: str = JmMagicConstants.TIME_ALL,
                           category: str = JmMagicConstants.CATEGORY_ALL,
                           sub_category: Optional[str] = None,
                           ):
        return await self.search(search_query, page, 4, order_by, time, category, sub_category)

    async def do_page_iter(self, params: dict, page: int, get_page_method):
        from math import inf
        from typing import Optional, Dict

        def update(value: Optional[Dict], page: int, page_content):
            if value is None:
                return page + 1, page_content.page_count

            ExceptionTool.require_true(isinstance(value, dict), 'require dict params')

            # 根据外界传递的参数，更新params和page
            page = value.get('page', page)
            params.update(value)

            return page, inf

        total = inf
        while page <= total:
            params['page'] = page
            page_content = await get_page_method(**params)
            value = yield page_content
            page, total = update(value, page, page_content)

    async def search_gen(self,
                         search_query: str,
                         main_tag=0,
                         page: int = 1,
                         order_by: str = JmMagicConstants.ORDER_BY_LATEST,
                         time: str = JmMagicConstants.TIME_ALL,
                         category: str = JmMagicConstants.CATEGORY_ALL,
                         sub_category: Optional[str] = None,
                         ):
        """
        异步搜索结果的生成器。
        使用示例:
        ```
        async for page in client.search_gen('无修正'):
            pass
        ```
        同时支持外界 asend 参数改变搜索的设定:
        ```
        gen = client.search_gen('MANA')
        page_1 = await gen.asend(None)
        page_3 = await gen.asend({'page': 3})
        ```
        """
        params = {
            'search_query': search_query,
            'main_tag': main_tag,
            'order_by': order_by,
            'time': time,
            'category': category,
            'sub_category': sub_category,
        }

        aiter = self.do_page_iter(params, page, self.search)
        value = None
        while True:
            try:
                page_content = await aiter.asend(value)
                value = yield page_content
            except StopAsyncIteration:
                break

    # -- JmCategoryClient --

    async def categories_filter(self,
                                page: int,
                                time: str,
                                category: str,
                                order_by: str,
                                sub_category: Optional[str] = None,
                                ) -> JmCategoryPage:
        raise NotImplementedError

    async def month_ranking(self,
                            page: int = 1,
                            category: str = JmMagicConstants.CATEGORY_ALL,
                            ):
        return await self.categories_filter(page, JmMagicConstants.TIME_MONTH, category,
                                            JmMagicConstants.ORDER_BY_VIEW)

    async def week_ranking(self,
                           page: int = 1,
                           category: str = JmMagicConstants.CATEGORY_ALL,
                           ):
        return await self.categories_filter(page, JmMagicConstants.TIME_WEEK, category,
                                            JmMagicConstants.ORDER_BY_VIEW)

    async def day_ranking(self,
                          page: int = 1,
                          category: str = JmMagicConstants.CATEGORY_ALL,
                          ):
        return await self.categories_filter(page, JmMagicConstants.TIME_TODAY, category,
                                            JmMagicConstants.ORDER_BY_VIEW)

    # -- JmUserClient --

    async def login(self, username: str, password: str):
        raise NotImplementedError

    async def favorite_folder(self,
                              page=1,
                              order_by=JmMagicConstants.ORDER_BY_LATEST,
                              folder_id='0',
                              username='',
                              ) -> JmFavoritePage:
        raise NotImplementedError

    async def favorite_folder_gen(self,
                                  page=1,
                                  order_by=JmMagicConstants.ORDER_BY_LATEST,
                                  folder_id='0',
                                  username='',
                                  ):
        """
        见 search_gen
        """
        params = {
            'order_by': order_by,
            'folder_id': folder_id,
            'username': username,
        }

        aiter = self.do_page_iter(params, page, self.favorite_folder)
        value = None
        while True:
            try:
                page_content = await aiter.asend(value)
                value = yield page_content
            except StopAsyncIteration:
                break

    async def add_favorite_album(self, album_id, folder_id='0'):
        raise NotImplementedError

    async def album_comment(self,
                            video_id,
                            comment,
                            originator='',
                            status='true',
                            comment_id=None,
                            **kwargs,
                            ) -> JmAlbumCommentResp:
        raise NotImplementedError

    # -- 域名 / 缓存管理 --

    def get_domain_list(self) -> List[str]:
        raise NotImplementedError

    def set_domain_list(self, domain_list: List[str]):
        raise NotImplementedError

    def set_cache_dict(self, cache_dict: Optional[Dict]):
        raise NotImplementedError

    def get_cache_dict(self) -> Optional[Dict]:
        raise NotImplementedError

    # -- 生命周期 --

    async def close(self):
        raise NotImplementedError

    async def __aenter__(self):
        await self.setup()
        return self

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

    async def setup(self):
        pass

    async def get_jm_image(self, download_url):
        raise NotImplementedError

check_photo(photo) async

检查 photo 的 from_album / page_arr / data_original_domain 是否齐全， 缺失则请求补全。对齐 sync JmDetailClient.check_photo。

Source code in src/jmcomic/jm_client_interface.py

async def check_photo(self, photo: JmPhotoDetail):
    """
    检查 photo 的 from_album / page_arr / data_original_domain 是否齐全，
    缺失则请求补全。对齐 sync JmDetailClient.check_photo。
    """
    # 检查 from_album
    if photo.from_album is None:
        photo.from_album = await self.get_album_detail(photo.album_id)

    # 检查 page_arr 和 data_original_domain
    if photo.page_arr is None or photo.data_original_domain is None:
        new = await self.get_photo_detail(photo.photo_id, False)
        new.from_album = photo.from_album
        photo.__dict__.update(new.__dict__)

favorite_folder_gen(page=1, order_by=JmMagicConstants.ORDER_BY_LATEST, folder_id='0', username='') async

见 search_gen

Source code in src/jmcomic/jm_client_interface.py

async def favorite_folder_gen(self,
                              page=1,
                              order_by=JmMagicConstants.ORDER_BY_LATEST,
                              folder_id='0',
                              username='',
                              ):
    """
    见 search_gen
    """
    params = {
        'order_by': order_by,
        'folder_id': folder_id,
        'username': username,
    }

    aiter = self.do_page_iter(params, page, self.favorite_folder)
    value = None
    while True:
        try:
            page_content = await aiter.asend(value)
            value = yield page_content
        except StopAsyncIteration:
            break

search_gen(search_query, main_tag=0, page=1, order_by=JmMagicConstants.ORDER_BY_LATEST, time=JmMagicConstants.TIME_ALL, category=JmMagicConstants.CATEGORY_ALL, sub_category=None) async

异步搜索结果的生成器。 使用示例:

async for page in client.search_gen('无修正'):
    pass

同时支持外界 asend 参数改变搜索的设定:

gen = client.search_gen('MANA')
page_1 = await gen.asend(None)
page_3 = await gen.asend({'page': 3})

Source code in src/jmcomic/jm_client_interface.py

async def search_gen(self,
                     search_query: str,
                     main_tag=0,
                     page: int = 1,
                     order_by: str = JmMagicConstants.ORDER_BY_LATEST,
                     time: str = JmMagicConstants.TIME_ALL,
                     category: str = JmMagicConstants.CATEGORY_ALL,
                     sub_category: Optional[str] = None,
                     ):
    """
    异步搜索结果的生成器。
    使用示例:
    ```
    async for page in client.search_gen('无修正'):
        pass
    ```
    同时支持外界 asend 参数改变搜索的设定:
    ```
    gen = client.search_gen('MANA')
    page_1 = await gen.asend(None)
    page_3 = await gen.asend({'page': 3})
    ```
    """
    params = {
        'search_query': search_query,
        'main_tag': main_tag,
        'order_by': order_by,
        'time': time,
        'category': category,
        'sub_category': sub_category,
    }

    aiter = self.do_page_iter(params, page, self.search)
    value = None
    while True:
        try:
            page_content = await aiter.asend(value)
            value = yield page_content
        except StopAsyncIteration:
            break

异步 jmcomic API 客户端模块

提供禁漫移动端接口的异步访问能力，基于 curl_cffi 与 asyncio 构建高性能网络通信层。

AsyncJmApiClient

Bases: AsyncJmcomicClient

禁漫移动端异步 API 客户端。

继承 AsyncJmcomicClient 接口，提供全面的异步网络通信能力， 涵盖图集、章节、搜索、登录与收藏夹等功能模块。 通过异步会话管理与并发请求调度，显著提升网络 I/O 的处理性能与吞吐量。

Methods:

Name	Description
add_favorite_album	将指定图集加入用户的收藏夹。
album_comment	提交图集评论内容
auto_update_domain	通过查询中心服务器下发的配置动态刷新本地的接口可用域名列表
before_retry	每次请求失败且即将进入重试前的拦截回调，子类可重写以加入自定义的副作用逻辑（例如告警或统计）。
categories_filter	获取指定分类下的图集列表数据。
ensure_have_cookies	初始化基础 Cookies 信息，当不存在时从服务端的 setting 接口拉取
favorite_folder	获取收藏夹内特定目录的图集数据分页。
fetch_scramble_id	向服务端发起实时请求，提取指定图片的 scramble_id 解析参数
get_album_detail	获取图集详情信息
get_jm_image	异步下载指定 URL 的图片原始字节数据。
get_photo_detail	获取指定图片的详细数据及其前置依赖关联信息
get_scramble_id	获取指定图片的 scramble_id（支持内存级缓存）
login	使用账户密码执行系统登录
req_api	核心的 API 请求封装方法。
search	发起全局搜索请求，提取并包装为搜索结果分页对象。
setting	获取服务端的环境配置（包含应用版本等参数）
setup	异步初始化入口，应在使用前调用。

Source code in src/jmcomic/jm_async_client.py

class AsyncJmApiClient(AsyncJmcomicClient):
    """
    禁漫移动端异步 API 客户端。

    继承 AsyncJmcomicClient 接口，提供全面的异步网络通信能力，
    涵盖图集、章节、搜索、登录与收藏夹等功能模块。
    通过异步会话管理与并发请求调度，显著提升网络 I/O 的处理性能与吞吐量。
    """

    client_key = 'async_api'

    # 核心 API 路径定义
    API_SEARCH = '/search'
    API_CATEGORIES_FILTER = '/categories/filter'
    API_ALBUM = '/album'
    API_CHAPTER = '/chapter'
    API_SCRAMBLE = '/chapter_view_template'
    API_FAVORITE = '/favorite'

    # 缓存未命中标记
    _SENTINEL = object()

    # 类级别初始化标记与锁，防止并发更新域名
    _has_setup_domain_and_cookies = False
    _setup_lock = asyncio.Lock()

    def __init__(self, option: JmOption, max_clients=None, **kwargs):
        self.option = option
        self._domain_list = self._resolve_domain_list()
        self._retry_times = option.client.get('retry_times', 5) or 5
        self._timeout = option.client.get('timeout', 30) or 30
        # AsyncSession 句柄池大小：优先用调用方（下载器）传入的实际图片并发，
        # 否则回退到 option 配置；避免因默认限制导致真实并发被隐式压低。
        if max_clients:
            self._max_clients_hint = int(max_clients)
        else:
            try:
                self._max_clients_hint = int(option.download.threading.image) or 10
            except Exception:
                self._max_clients_hint = 10

        self._session: AsyncSession | None = None
        self._session_lock = asyncio.Lock()
        # 缓存默认关闭，由外部配置决定是否启用。
        self._cache: dict | None = None
        self._username: str | None = None

        # 接收并保存额外的会话级元数据参数
        self._meta_kwargs = kwargs
        self._has_setup = False

    # ======================================================================
    # 域名管理
    # ======================================================================

    def _resolve_domain_list(self) -> list[str]:
        """解析并返回可用的 API 域名列表"""
        updated = JmModuleConfig.DOMAIN_API_UPDATED_LIST
        if updated:
            return list(updated)
        domain = self.option.client.domain
        if hasattr(domain, 'get'):
            domain_list = domain.get('api', [])
        elif isinstance(domain, list):
            domain_list = domain
        elif isinstance(domain, str):
            domain_list = [d.strip() for d in domain.split('\n') if d.strip()]
        else:
            domain_list = []
        if domain_list:
            return domain_list
        return list(JmModuleConfig.DOMAIN_API_LIST)

    def get_domain_list(self) -> list[str]:
        return self._domain_list

    def set_domain_list(self, domain_list: list[str]):
        self._domain_list = domain_list

    # ======================================================================
    # 缓存
    # ======================================================================

    def set_cache_dict(self, cache_dict: dict | None):
        self._cache = cache_dict

    def get_cache_dict(self) -> dict | None:
        return self._cache

    def _cache_get(self, key):
        """从缓存获取，未命中返回 sentinel"""
        if self._cache is None:
            return self._SENTINEL
        return self._cache.get(key, self._SENTINEL)

    def _cache_set(self, key, value):
        """写入缓存"""
        if self._cache is not None:
            self._cache[key] = value

    # 说明：异步缓存不采用动态方法包裹（Monkey Patching）的方式，避免缓存协程对象引发复用异常。
    # 而是直接在 _fetch_detail_entity / search 内部通过 _cache_get/_cache_set 进行结果级缓存操作。
    # 启停状态由 self._cache 对象驱动。

    # ======================================================================
    # Session 管理
    # ======================================================================

    async def _ensure_session(self):
        """懒加载 AsyncSession，确保在 event loop 中初始化"""
        if self._session is not None:
            return
        async with self._session_lock:
            if self._session is not None:
                return

            # 提取应用配置中预设的网络通信元数据信息（如代理配置与全局 Headers）
            from copy import deepcopy
            postman_conf = deepcopy(self.option.client.get('postman', {}))
            meta_data = postman_conf.get('meta_data', {})
            if self._meta_kwargs:
                meta_data.update(self._meta_kwargs)

            kwargs = {
                'timeout': self._timeout,
                'impersonate': meta_data.get('impersonate', 'chrome'),
                # 让 AsyncSession 的句柄池大小与本下载器的图片并发对齐，
                # 避免因默认限制导致真实并发被隐式压低。
                'max_clients': max(self._max_clients_hint, 1),
            }

            proxies = meta_data.get('proxies', None)
            if proxies is not None:
                # 字符串形式的代理需经 ProxyBuilder 转 dict
                if isinstance(proxies, str):
                    from common import ProxyBuilder
                    proxies = ProxyBuilder.build_by_str(proxies)
                kwargs['proxies'] = proxies

            if meta_data.get('headers'):
                kwargs['headers'] = meta_data['headers']

            # 加载预配置或已持久化的历史会话 Cookies
            if meta_data.get('cookies'):
                kwargs['cookies'] = meta_data['cookies']

            # noinspection PyArgumentList
            self._session = AsyncSession(**kwargs)

    # ======================================================================
    # 核心请求基础设施
    # ======================================================================

    def _build_api_url(self, path: str, domain: str) -> str:
        prot = JmModuleConfig.PROT
        if domain.startswith(prot):
            return f'{domain}{path}'
        return f'{prot}{domain}{path}'

    def _build_api_headers(self, path: str) -> tuple:
        """构建对应接口所需的 API 请求头部信息与时间戳"""
        headers = dict(JmModuleConfig.APP_HEADERS_TEMPLATE)

        if path == self.API_SCRAMBLE:
            ts = time_stamp()
            token, tokenparam = JmCryptoTool.token_and_tokenparam(
                ts, secret=JmMagicConstants.APP_TOKEN_SECRET_2
            )
        elif JmModuleConfig.FLAG_USE_FIX_TIMESTAMP:
            ts, token, tokenparam = JmModuleConfig.get_fix_ts_token_tokenparam()
        else:
            ts = time_stamp()
            token, tokenparam = JmCryptoTool.token_and_tokenparam(ts)

        headers['token'] = token
        headers['tokenparam'] = tokenparam
        return headers, ts

    async def _request_with_retry(self,
                                  url_path: str,
                                  headers: dict,
                                  get: bool = True,
                                  is_api: bool = True,
                                  **kwargs,
                                  ):
        """
        带域名切换机制的请求重试策略。
        机制：在当前域名下重试指定的次数，如全数失败则切换至备选域名，直至遍历完所有可用域名。
        """
        domain_list = self._domain_list
        if not domain_list:
            ExceptionTool.raises("无可用 API 域名列表")

        for domain_index, domain in enumerate(domain_list):
            url = self._build_api_url(url_path, domain)

            for retry in range(self._retry_times + 1):
                # 记录重试信息
                if domain_index != 0 or retry != 0:
                    jm_log('req.retry',
                           f'次数: [{retry}/{self._retry_times}], '
                           f'域名: [{domain_index} of {domain_list}], '
                           f'路径: [{url}]')

                # 记录请求日志
                jm_log(self.client_key, self._decode_url_for_log(url))

                try:
                    if get:
                        # noinspection PyUnresolvedReferences
                        resp = await self._session.get(url, headers=headers, **kwargs)
                    else:
                        # noinspection PyUnresolvedReferences
                        resp = await self._session.post(url, headers=headers, **kwargs)

                    # 校验 API 响应的有效性并决定是否触发重试
                    if is_api:
                        self._raise_if_resp_should_retry(resp)

                    return resp
                except Exception as e:
                    self.before_retry(e, url, retry, domain_index)

        # 所有域名都失败
        msg = f"请求重试全部失败: [{url_path}], {domain_list}"
        jm_log('req.fallback', msg)
        ExceptionTool.raises(msg, {}, RequestRetryAllFailException)

    # noinspection PyMethodMayBeStatic,PyUnusedLocal
    def before_retry(self, e, url, retry, domain_index):
        """
        每次请求失败且即将进入重试前的拦截回调，子类可重写以加入自定义的副作用逻辑（例如告警或统计）。
        """
        jm_log('req.error', str(e), e)

    def _decode_url_for_log(self, url: str) -> str:
        """将 URL 转换为适合在日志中显示的解码格式"""
        if not JmModuleConfig.FLAG_DECODE_URL_WHEN_LOGGING or '/search/' not in url:
            return url

        from urllib.parse import unquote
        return unquote(url.replace('+', ' '))

    @staticmethod
    def _raise_if_resp_should_retry(resp):
        """内部校验 API 响应报文内容，若存在异常格式或无法处理的数据则抛出异常以触发重试"""
        code = resp.status_code
        if code >= 500:
            msg = JmModuleConfig.JM_ERROR_STATUS_CODE.get(code, f'HTTP状态码: {code}')
            ExceptionTool.raises_resp(f"禁漫API异常响应, {msg}", resp)

        url = getattr(resp, 'url', '')
        if AsyncJmApiClient.API_SCRAMBLE in str(url):
            # /chapter_view_template 这个接口不是返回json数据，不做检查
            return

        # 检查响应的第一个有效字符是否为 '{'（JSON 格式）
        text = resp.text
        for char in text:
            if char not in (' ', '\n', '\t'):
                ExceptionTool.require_true(
                    char == '{',
                    f'请求不是json格式，强制重试！响应文本: [{JmcomicText.limit_text(text, 200)}]'
                )
                return
        ExceptionTool.raises_resp(f'响应无数据！', resp)

    @staticmethod
    def _require_resp_success(resp: JmApiResp):
        """断言响应状态必须为成功"""
        resp.require_success()

    async def req_api(self,
                      url: str,
                      get: bool = True,
                      require_success: bool = True,
                      params: dict | None = None,
                      **kwargs,
                      ) -> JmApiResp:
        """
        核心的 API 请求封装方法。
        处理参数拼装、请求发送与重试，并返回统一的 JmApiResp 响应对象。
        """
        # /setting 是 setup() 内部初始化调用的接口，跳过 setup 防止 asyncio.Lock 不可重入死锁
        if url != '/setting':
            await self.setup()
        else:
            await self._ensure_session()

        # 构建 headers 和时间戳
        headers, ts = self._build_api_headers(url)
        # 合并外部传入的 headers
        ext_headers = kwargs.pop('headers', None)
        if ext_headers:
            headers.update(ext_headers)

        # 构建 URL 路径
        url_path = url
        if params:
            url_path = f'{url}?{urlencode(params)}'

        # 带域名重试的请求（不硬编码 timeout，使用 session 级别的配置）
        resp = await self._request_with_retry(
            url_path, headers, get=get, is_api=True, **kwargs,
        )

        # 封装为 JmApiResp，复用完整的校验链
        api_resp = JmApiResp(resp, ts)
        if require_success:
            self._require_resp_success(api_resp)

        return api_resp

    # ======================================================================
    # 详情数据接口获取方法
    # ======================================================================

    async def _fetch_detail_entity(self, jmid, clazz: type[DetailType]) -> DetailType:
        """发起详情页数据请求并解析为指定的实体类型"""
        jmid = JmcomicText.parse_to_jm_id(jmid)

        # 缓存检查
        cache_key = ('detail', jmid, clazz)
        cached = self._cache_get(cache_key)
        if cached is not self._SENTINEL:
            # noinspection PyTypeChecker
            return cached

        url = self.API_ALBUM if issubclass(clazz, JmAlbumDetail) else self.API_CHAPTER
        resp = await self.req_api(url, params={'id': jmid})

        if not resp.encoded_data or resp.res_data.get('name') is None:
            ExceptionTool.raise_missing(resp, jmid)

        result = JmApiAdaptTool.parse_entity(resp.res_data, clazz)
        self._cache_set(cache_key, result)
        return result

    async def get_album_detail(self, album_id) -> JmAlbumDetail:
        """获取图集详情信息"""
        return await self._fetch_detail_entity(album_id, JmModuleConfig.album_class())

    async def get_photo_detail(self,
                               photo_id,
                               fetch_album=True,
                               fetch_scramble_id=True,
                               ) -> JmPhotoDetail:
        """获取指定图片的详细数据及其前置依赖关联信息"""
        photo = await self._fetch_detail_entity(photo_id, JmModuleConfig.photo_class())
        if fetch_album or fetch_scramble_id:
            await self._fetch_photo_additional_field(photo, fetch_album, fetch_scramble_id)
        return photo

    async def _fetch_photo_additional_field(self, photo: JmPhotoDetail,
                                            fetch_album: bool,
                                            fetch_scramble_id: bool):
        """并发获取图片从属的图集信息与 scramble_id 加解密参数。"""
        tasks = {}
        if fetch_album:
            tasks['album'] = self.get_album_detail(photo.album_id)
        if fetch_scramble_id:
            tasks['scramble'] = self.get_scramble_id(photo.photo_id, photo.album_id)

        if not tasks:
            return

        keys = list(tasks.keys())
        results = await asyncio.gather(*tasks.values())
        result_map = dict(zip(keys, results))

        if 'album' in result_map:
            photo.from_album = result_map['album']
        if 'scramble' in result_map:
            photo.scramble_id = result_map['scramble']

    # check_photo 继承自 AsyncJmcomicClient 基类

    # ======================================================================
    # 图片解码参数 Scramble ID 获取接口
    # ======================================================================

    async def get_scramble_id(self, photo_id, album_id=None) -> str:
        """获取指定图片的 scramble_id（支持内存级缓存）"""
        cache = JmModuleConfig.SCRAMBLE_CACHE
        if photo_id in cache:
            return cache[photo_id]
        if album_id is not None and album_id in cache:
            return cache[album_id]

        scramble_id = await self.fetch_scramble_id(photo_id)
        cache[photo_id] = scramble_id
        if album_id is not None:
            cache[album_id] = scramble_id
        return scramble_id

    async def fetch_scramble_id(self, photo_id) -> str:
        """向服务端发起实时请求，提取指定图片的 scramble_id 解析参数"""
        photo_id = JmcomicText.parse_to_jm_id(photo_id)
        resp = await self.req_api(
            self.API_SCRAMBLE,
            params={
                'id': photo_id,
                'mode': 'vertical',
                'page': '0',
                'app_img_shunt': '1',
                'express': 'off',
                'v': time_stamp(),
            },
            require_success=False,
        )

        scramble_id = PatternTool.match_or_default(
            resp.text, JmcomicText.pattern_html_album_scramble_id, None
        )
        if scramble_id is None:
            jm_log('api.scramble', f'未匹配到scramble_id，响应文本：{resp.text}')
            scramble_id = str(JmMagicConstants.SCRAMBLE_220980)

        return scramble_id

    # ======================================================================
    # 环境配置与认证管理
    # ======================================================================

    async def ensure_have_cookies(self):
        """初始化基础 Cookies 信息，当不存在时从服务端的 setting 接口拉取"""
        # noinspection PyUnresolvedReferences
        if self._session and self._session.cookies:
            return
        # 复用全局缓存
        if JmModuleConfig.APP_COOKIES is not None:
            await self._ensure_session()
            # noinspection PyUnresolvedReferences
            self._session.cookies.update(JmModuleConfig.APP_COOKIES)
            return
        resp = await self.setting()
        cookies = dict(resp.resp.cookies)
        JmModuleConfig.APP_COOKIES = cookies
        # noinspection PyUnresolvedReferences,PyTypeChecker
        self._session.cookies.update(cookies)

    async def setting(self) -> JmApiResp:
        """获取服务端的环境配置（包含应用版本等参数）"""
        resp = await self.req_api('/setting')

        setting_ver = str(resp.model_data.jm3_version)
        if (
                JmModuleConfig.FLAG_USE_VERSION_NEWER_IF_BEHIND
                and JmcomicText.compare_versions(setting_ver, JmMagicConstants.APP_VERSION) == 1
        ):
            jm_log('api.setting',
                   f'change APP_VERSION from [{JmMagicConstants.APP_VERSION}] to [{setting_ver}]')
            JmMagicConstants.APP_VERSION = setting_ver

        return resp

    # ======================================================================
    # 搜索与分类接口
    # ======================================================================

    async def search(self,
                     search_query: str,
                     page: int,
                     main_tag: int,
                     order_by: str,
                     time: str,
                     category: str,
                     sub_category: str | None,
                     ) -> JmSearchPage:
        """
        发起全局搜索请求，提取并包装为搜索结果分页对象。
        注意：移动端暂不支持 category 和 sub_category。
        """
        # 缓存检查
        cache_key = ('search', search_query, page, main_tag, order_by, time)
        # noinspection PyTypeChecker
        cached: JmSearchPage = self._cache_get(cache_key)
        if cached is not self._SENTINEL:
            return cached

        params = {
            'main_tag': main_tag,
            'search_query': search_query,
            'page': page,
            'o': order_by,
            't': time,
        }
        resp = await self.req_api(self.API_SEARCH, params=params)

        data = resp.model_data
        if data.get('redirect_aid', None) is not None:
            aid = data.redirect_aid
            result = JmSearchPage.wrap_single_album(await self.get_album_detail(aid))
        else:
            result = JmPageTool.parse_api_to_search_page(data)

        self._cache_set(cache_key, result)
        return result

    # search_site / search_work / search_author / search_tag / search_actor
    # 继承自 AsyncJmcomicClient 基类，默认值由基类便捷方法提供

    # ======================================================================
    # 分类过滤接口
    # ======================================================================

    async def categories_filter(self,
                                page: int,
                                time: str,
                                category: str,
                                order_by: str,
                                sub_category: str | None = None,
                                ) -> JmCategoryPage:
        """
        获取指定分类下的图集列表数据。
        注意：移动端不支持 sub_category。
        """
        o = f'{order_by}_{time}' if time != JmMagicConstants.TIME_ALL else order_by
        params = {
            'page': page,
            'order': '',
            'c': category,
            'o': o,
        }
        resp = await self.req_api(self.API_CATEGORIES_FILTER, params=params)
        return JmPageTool.parse_api_to_search_page(resp.model_data)

    # month_ranking / week_ranking / day_ranking
    # 继承自 AsyncJmcomicClient 基类

    # ======================================================================
    # 用户资产与登录接口
    # ======================================================================

    async def login(self, username: str, password: str) -> JmApiResp:
        """使用账户密码执行系统登录"""
        resp = await self.req_api('/login', False, data={
            'username': username,
            'password': password,
        })
        cookies = dict(resp.resp.cookies)
        cookies.update({'AVS': resp.res_data['s']})
        # noinspection PyUnresolvedReferences,PyTypeChecker
        self._session.cookies.update(cookies)
        # 同步到 Option 配置，确保 cookies 持久化
        self.option.update_cookies(cookies)
        self._username = username
        return resp

    async def favorite_folder(self,
                              page=1,
                              order_by=JmMagicConstants.ORDER_BY_LATEST,
                              folder_id='0',
                              username='',
                              ) -> JmFavoritePage:
        """获取收藏夹内特定目录的图集数据分页。"""
        resp = await self.req_api(
            self.API_FAVORITE,
            params={
                'page': page,
                'folder_id': folder_id,
                'o': order_by,
            }
        )
        return JmPageTool.parse_api_to_favorite_page(resp.model_data)

    async def add_favorite_album(self, album_id, folder_id='0'):
        """
        将指定图集加入用户的收藏夹。
        注意：移动端没有提供 folder_id 参数。
        """
        # 服务端实现上使用带 body 的 GET 请求方式
        resp = await self.req_api('/favorite', data={'aid': album_id})
        data = resp.model_data
        if data.status != 'ok':
            ExceptionTool.raises_resp(data.msg, resp)
        return resp

    async def album_comment(self,
                            video_id,
                            comment,
                            originator='',
                            status='true',
                            comment_id=None,
                            **kwargs,
                            ) -> JmAlbumCommentResp:
        """提交图集评论内容"""
        # 移动端 API 没有评论接口，此方法仅为接口完整性保留
        raise NotImplementedError('移动端 API 不支持评论功能，请使用网页端 JmHtmlClient')

    # ======================================================================
    # 图片下载
    # ======================================================================

    async def get_jm_image(self, img_url: str) -> JmImageResp:
        """
        异步下载指定 URL 的图片原始字节数据。
        """
        await self.setup()
        headers = {**JmModuleConfig.APP_HEADERS_TEMPLATE, **JmModuleConfig.APP_HEADERS_IMAGE}

        last_error = None
        for retry in range(self._retry_times + 1):
            try:
                # noinspection PyUnresolvedReferences
                resp = await self._session.get(img_url, headers=headers, timeout=self._timeout)
                # 对图片资源的数据进行基础有效性校验
                img_resp = JmImageResp(resp)
                if resp.status_code != 200 or len(resp.content) == 0:
                    img_resp.require_success()  # 会抛出描述性异常
                return img_resp
            except Exception as e:
                last_error = e
                jm_log('req.error',
                       f'图片下载失败: [{img_url}], Retry=[{retry}/{self._retry_times}], Error=[{e}]')
                if retry < self._retry_times:
                    await asyncio.sleep(0.3)

        raise ExceptionTool.raises(f'图片下载重试全部失败: {last_error}', {}, RequestRetryAllFailException)

    # ======================================================================
    # 域名与状态自动刷新
    # ======================================================================

    async def auto_update_domain(self):
        """通过查询中心服务器下发的配置动态刷新本地的接口可用域名列表"""
        if not JmModuleConfig.FLAG_API_CLIENT_AUTO_UPDATE_DOMAIN:
            return

        if JmModuleConfig.DOMAIN_API_UPDATED_LIST is not None:
            if JmModuleConfig.DOMAIN_API_UPDATED_LIST:
                self._domain_list = list(JmModuleConfig.DOMAIN_API_UPDATED_LIST)
            return

        # 尝试从域名服务器获取最新域名
        await self._ensure_session()
        for url in JmModuleConfig.API_URL_DOMAIN_SERVER_LIST:
            try:
                # noinspection PyUnresolvedReferences
                resp = await self._session.get(url, timeout=10)
                text = resp.text
                while text and not text[0].isascii():
                    text = text[1:]
                res_json = JmCryptoTool.decode_resp_data(
                    text, '', JmMagicConstants.API_DOMAIN_SERVER_SECRET
                )
                res_data = json.loads(res_json)
                new_server_list = res_data.get('Server', None)
                if not new_server_list:
                    continue

                jm_log('api.update_domain.success',
                       f'获取到最新的API域名: {new_server_list}')
                JmModuleConfig.DOMAIN_API_UPDATED_LIST = new_server_list
                if sorted(self._domain_list) == sorted(JmModuleConfig.DOMAIN_API_LIST):
                    self._domain_list = new_server_list
                return
            except Exception as e:
                jm_log('api.update_domain.error', f'通过[{url}]自动更新API域名失败: {e}')
                continue

        JmModuleConfig.DOMAIN_API_UPDATED_LIST = []

    # ======================================================================
    # 资源生命周期控制
    # ======================================================================

    async def setup(self):
        """
        异步初始化入口，应在使用前调用。
        __aenter__ 会自动调用此方法。
        """
        if self._has_setup:
            return

        await self._ensure_session()

        cls = self.__class__
        async with cls._setup_lock:
            if not cls._has_setup_domain_and_cookies:
                await self.auto_update_domain()
                if JmModuleConfig.FLAG_API_CLIENT_REQUIRE_COOKIES:
                    await self.ensure_have_cookies()
                cls._has_setup_domain_and_cookies = True
            else:
                # 即使已经初始化过域名和 cookie，也需要将已保存的全局 DOMAIN 和 COOKIES 赋值到当前 client
                if JmModuleConfig.DOMAIN_API_UPDATED_LIST:
                    self._domain_list = list(JmModuleConfig.DOMAIN_API_UPDATED_LIST)
                if JmModuleConfig.FLAG_API_CLIENT_REQUIRE_COOKIES and JmModuleConfig.APP_COOKIES:
                    # noinspection PyUnresolvedReferences
                    self._session.cookies.update(JmModuleConfig.APP_COOKIES)

        self._has_setup = True

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

    async def close(self):
        if self._session:
            await self._session.close()
            self._session = None

    async def __aenter__(self):
        await self.setup()
        return self

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

add_favorite_album(album_id, folder_id='0') async

将指定图集加入用户的收藏夹。 注意：移动端没有提供 folder_id 参数。

Source code in src/jmcomic/jm_async_client.py

async def add_favorite_album(self, album_id, folder_id='0'):
    """
    将指定图集加入用户的收藏夹。
    注意：移动端没有提供 folder_id 参数。
    """
    # 服务端实现上使用带 body 的 GET 请求方式
    resp = await self.req_api('/favorite', data={'aid': album_id})
    data = resp.model_data
    if data.status != 'ok':
        ExceptionTool.raises_resp(data.msg, resp)
    return resp

album_comment(video_id, comment, originator='', status='true', comment_id=None, **kwargs) async

提交图集评论内容

Source code in src/jmcomic/jm_async_client.py

async def album_comment(self,
                        video_id,
                        comment,
                        originator='',
                        status='true',
                        comment_id=None,
                        **kwargs,
                        ) -> JmAlbumCommentResp:
    """提交图集评论内容"""
    # 移动端 API 没有评论接口，此方法仅为接口完整性保留
    raise NotImplementedError('移动端 API 不支持评论功能，请使用网页端 JmHtmlClient')

auto_update_domain() async

通过查询中心服务器下发的配置动态刷新本地的接口可用域名列表

Source code in src/jmcomic/jm_async_client.py

async def auto_update_domain(self):
    """通过查询中心服务器下发的配置动态刷新本地的接口可用域名列表"""
    if not JmModuleConfig.FLAG_API_CLIENT_AUTO_UPDATE_DOMAIN:
        return

    if JmModuleConfig.DOMAIN_API_UPDATED_LIST is not None:
        if JmModuleConfig.DOMAIN_API_UPDATED_LIST:
            self._domain_list = list(JmModuleConfig.DOMAIN_API_UPDATED_LIST)
        return

    # 尝试从域名服务器获取最新域名
    await self._ensure_session()
    for url in JmModuleConfig.API_URL_DOMAIN_SERVER_LIST:
        try:
            # noinspection PyUnresolvedReferences
            resp = await self._session.get(url, timeout=10)
            text = resp.text
            while text and not text[0].isascii():
                text = text[1:]
            res_json = JmCryptoTool.decode_resp_data(
                text, '', JmMagicConstants.API_DOMAIN_SERVER_SECRET
            )
            res_data = json.loads(res_json)
            new_server_list = res_data.get('Server', None)
            if not new_server_list:
                continue

            jm_log('api.update_domain.success',
                   f'获取到最新的API域名: {new_server_list}')
            JmModuleConfig.DOMAIN_API_UPDATED_LIST = new_server_list
            if sorted(self._domain_list) == sorted(JmModuleConfig.DOMAIN_API_LIST):
                self._domain_list = new_server_list
            return
        except Exception as e:
            jm_log('api.update_domain.error', f'通过[{url}]自动更新API域名失败: {e}')
            continue

    JmModuleConfig.DOMAIN_API_UPDATED_LIST = []

before_retry(e, url, retry, domain_index)

每次请求失败且即将进入重试前的拦截回调，子类可重写以加入自定义的副作用逻辑（例如告警或统计）。

Source code in src/jmcomic/jm_async_client.py

def before_retry(self, e, url, retry, domain_index):
    """
    每次请求失败且即将进入重试前的拦截回调，子类可重写以加入自定义的副作用逻辑（例如告警或统计）。
    """
    jm_log('req.error', str(e), e)

categories_filter(page, time, category, order_by, sub_category=None) async

获取指定分类下的图集列表数据。 注意：移动端不支持 sub_category。

Source code in src/jmcomic/jm_async_client.py

async def categories_filter(self,
                            page: int,
                            time: str,
                            category: str,
                            order_by: str,
                            sub_category: str | None = None,
                            ) -> JmCategoryPage:
    """
    获取指定分类下的图集列表数据。
    注意：移动端不支持 sub_category。
    """
    o = f'{order_by}_{time}' if time != JmMagicConstants.TIME_ALL else order_by
    params = {
        'page': page,
        'order': '',
        'c': category,
        'o': o,
    }
    resp = await self.req_api(self.API_CATEGORIES_FILTER, params=params)
    return JmPageTool.parse_api_to_search_page(resp.model_data)

ensure_have_cookies() async

初始化基础 Cookies 信息，当不存在时从服务端的 setting 接口拉取

Source code in src/jmcomic/jm_async_client.py

async def ensure_have_cookies(self):
    """初始化基础 Cookies 信息，当不存在时从服务端的 setting 接口拉取"""
    # noinspection PyUnresolvedReferences
    if self._session and self._session.cookies:
        return
    # 复用全局缓存
    if JmModuleConfig.APP_COOKIES is not None:
        await self._ensure_session()
        # noinspection PyUnresolvedReferences
        self._session.cookies.update(JmModuleConfig.APP_COOKIES)
        return
    resp = await self.setting()
    cookies = dict(resp.resp.cookies)
    JmModuleConfig.APP_COOKIES = cookies
    # noinspection PyUnresolvedReferences,PyTypeChecker
    self._session.cookies.update(cookies)

favorite_folder(page=1, order_by=JmMagicConstants.ORDER_BY_LATEST, folder_id='0', username='') async

获取收藏夹内特定目录的图集数据分页。

Source code in src/jmcomic/jm_async_client.py

async def favorite_folder(self,
                          page=1,
                          order_by=JmMagicConstants.ORDER_BY_LATEST,
                          folder_id='0',
                          username='',
                          ) -> JmFavoritePage:
    """获取收藏夹内特定目录的图集数据分页。"""
    resp = await self.req_api(
        self.API_FAVORITE,
        params={
            'page': page,
            'folder_id': folder_id,
            'o': order_by,
        }
    )
    return JmPageTool.parse_api_to_favorite_page(resp.model_data)

fetch_scramble_id(photo_id) async

向服务端发起实时请求，提取指定图片的 scramble_id 解析参数

Source code in src/jmcomic/jm_async_client.py

async def fetch_scramble_id(self, photo_id) -> str:
    """向服务端发起实时请求，提取指定图片的 scramble_id 解析参数"""
    photo_id = JmcomicText.parse_to_jm_id(photo_id)
    resp = await self.req_api(
        self.API_SCRAMBLE,
        params={
            'id': photo_id,
            'mode': 'vertical',
            'page': '0',
            'app_img_shunt': '1',
            'express': 'off',
            'v': time_stamp(),
        },
        require_success=False,
    )

    scramble_id = PatternTool.match_or_default(
        resp.text, JmcomicText.pattern_html_album_scramble_id, None
    )
    if scramble_id is None:
        jm_log('api.scramble', f'未匹配到scramble_id，响应文本：{resp.text}')
        scramble_id = str(JmMagicConstants.SCRAMBLE_220980)

    return scramble_id

get_album_detail(album_id) async

获取图集详情信息

Source code in src/jmcomic/jm_async_client.py

async def get_album_detail(self, album_id) -> JmAlbumDetail:
    """获取图集详情信息"""
    return await self._fetch_detail_entity(album_id, JmModuleConfig.album_class())

get_jm_image(img_url) async

异步下载指定 URL 的图片原始字节数据。

Source code in src/jmcomic/jm_async_client.py

async def get_jm_image(self, img_url: str) -> JmImageResp:
    """
    异步下载指定 URL 的图片原始字节数据。
    """
    await self.setup()
    headers = {**JmModuleConfig.APP_HEADERS_TEMPLATE, **JmModuleConfig.APP_HEADERS_IMAGE}

    last_error = None
    for retry in range(self._retry_times + 1):
        try:
            # noinspection PyUnresolvedReferences
            resp = await self._session.get(img_url, headers=headers, timeout=self._timeout)
            # 对图片资源的数据进行基础有效性校验
            img_resp = JmImageResp(resp)
            if resp.status_code != 200 or len(resp.content) == 0:
                img_resp.require_success()  # 会抛出描述性异常
            return img_resp
        except Exception as e:
            last_error = e
            jm_log('req.error',
                   f'图片下载失败: [{img_url}], Retry=[{retry}/{self._retry_times}], Error=[{e}]')
            if retry < self._retry_times:
                await asyncio.sleep(0.3)

    raise ExceptionTool.raises(f'图片下载重试全部失败: {last_error}', {}, RequestRetryAllFailException)

get_photo_detail(photo_id, fetch_album=True, fetch_scramble_id=True) async

获取指定图片的详细数据及其前置依赖关联信息

Source code in src/jmcomic/jm_async_client.py

async def get_photo_detail(self,
                           photo_id,
                           fetch_album=True,
                           fetch_scramble_id=True,
                           ) -> JmPhotoDetail:
    """获取指定图片的详细数据及其前置依赖关联信息"""
    photo = await self._fetch_detail_entity(photo_id, JmModuleConfig.photo_class())
    if fetch_album or fetch_scramble_id:
        await self._fetch_photo_additional_field(photo, fetch_album, fetch_scramble_id)
    return photo

get_scramble_id(photo_id, album_id=None) async

获取指定图片的 scramble_id（支持内存级缓存）

Source code in src/jmcomic/jm_async_client.py

async def get_scramble_id(self, photo_id, album_id=None) -> str:
    """获取指定图片的 scramble_id（支持内存级缓存）"""
    cache = JmModuleConfig.SCRAMBLE_CACHE
    if photo_id in cache:
        return cache[photo_id]
    if album_id is not None and album_id in cache:
        return cache[album_id]

    scramble_id = await self.fetch_scramble_id(photo_id)
    cache[photo_id] = scramble_id
    if album_id is not None:
        cache[album_id] = scramble_id
    return scramble_id

login(username, password) async

使用账户密码执行系统登录

Source code in src/jmcomic/jm_async_client.py

async def login(self, username: str, password: str) -> JmApiResp:
    """使用账户密码执行系统登录"""
    resp = await self.req_api('/login', False, data={
        'username': username,
        'password': password,
    })
    cookies = dict(resp.resp.cookies)
    cookies.update({'AVS': resp.res_data['s']})
    # noinspection PyUnresolvedReferences,PyTypeChecker
    self._session.cookies.update(cookies)
    # 同步到 Option 配置，确保 cookies 持久化
    self.option.update_cookies(cookies)
    self._username = username
    return resp

req_api(url, get=True, require_success=True, params=None, **kwargs) async

核心的 API 请求封装方法。 处理参数拼装、请求发送与重试，并返回统一的 JmApiResp 响应对象。

Source code in src/jmcomic/jm_async_client.py

async def req_api(self,
                  url: str,
                  get: bool = True,
                  require_success: bool = True,
                  params: dict | None = None,
                  **kwargs,
                  ) -> JmApiResp:
    """
    核心的 API 请求封装方法。
    处理参数拼装、请求发送与重试，并返回统一的 JmApiResp 响应对象。
    """
    # /setting 是 setup() 内部初始化调用的接口，跳过 setup 防止 asyncio.Lock 不可重入死锁
    if url != '/setting':
        await self.setup()
    else:
        await self._ensure_session()

    # 构建 headers 和时间戳
    headers, ts = self._build_api_headers(url)
    # 合并外部传入的 headers
    ext_headers = kwargs.pop('headers', None)
    if ext_headers:
        headers.update(ext_headers)

    # 构建 URL 路径
    url_path = url
    if params:
        url_path = f'{url}?{urlencode(params)}'

    # 带域名重试的请求（不硬编码 timeout，使用 session 级别的配置）
    resp = await self._request_with_retry(
        url_path, headers, get=get, is_api=True, **kwargs,
    )

    # 封装为 JmApiResp，复用完整的校验链
    api_resp = JmApiResp(resp, ts)
    if require_success:
        self._require_resp_success(api_resp)

    return api_resp

search(search_query, page, main_tag, order_by, time, category, sub_category) async

发起全局搜索请求，提取并包装为搜索结果分页对象。 注意：移动端暂不支持 category 和 sub_category。

Source code in src/jmcomic/jm_async_client.py

async def search(self,
                 search_query: str,
                 page: int,
                 main_tag: int,
                 order_by: str,
                 time: str,
                 category: str,
                 sub_category: str | None,
                 ) -> JmSearchPage:
    """
    发起全局搜索请求，提取并包装为搜索结果分页对象。
    注意：移动端暂不支持 category 和 sub_category。
    """
    # 缓存检查
    cache_key = ('search', search_query, page, main_tag, order_by, time)
    # noinspection PyTypeChecker
    cached: JmSearchPage = self._cache_get(cache_key)
    if cached is not self._SENTINEL:
        return cached

    params = {
        'main_tag': main_tag,
        'search_query': search_query,
        'page': page,
        'o': order_by,
        't': time,
    }
    resp = await self.req_api(self.API_SEARCH, params=params)

    data = resp.model_data
    if data.get('redirect_aid', None) is not None:
        aid = data.redirect_aid
        result = JmSearchPage.wrap_single_album(await self.get_album_detail(aid))
    else:
        result = JmPageTool.parse_api_to_search_page(data)

    self._cache_set(cache_key, result)
    return result

setting() async

获取服务端的环境配置（包含应用版本等参数）

Source code in src/jmcomic/jm_async_client.py

async def setting(self) -> JmApiResp:
    """获取服务端的环境配置（包含应用版本等参数）"""
    resp = await self.req_api('/setting')

    setting_ver = str(resp.model_data.jm3_version)
    if (
            JmModuleConfig.FLAG_USE_VERSION_NEWER_IF_BEHIND
            and JmcomicText.compare_versions(setting_ver, JmMagicConstants.APP_VERSION) == 1
    ):
        jm_log('api.setting',
               f'change APP_VERSION from [{JmMagicConstants.APP_VERSION}] to [{setting_ver}]')
        JmMagicConstants.APP_VERSION = setting_ver

    return resp

setup() async

异步初始化入口，应在使用前调用。 aenter 会自动调用此方法。

Source code in src/jmcomic/jm_async_client.py

async def setup(self):
    """
    异步初始化入口，应在使用前调用。
    __aenter__ 会自动调用此方法。
    """
    if self._has_setup:
        return

    await self._ensure_session()

    cls = self.__class__
    async with cls._setup_lock:
        if not cls._has_setup_domain_and_cookies:
            await self.auto_update_domain()
            if JmModuleConfig.FLAG_API_CLIENT_REQUIRE_COOKIES:
                await self.ensure_have_cookies()
            cls._has_setup_domain_and_cookies = True
        else:
            # 即使已经初始化过域名和 cookie，也需要将已保存的全局 DOMAIN 和 COOKIES 赋值到当前 client
            if JmModuleConfig.DOMAIN_API_UPDATED_LIST:
                self._domain_list = list(JmModuleConfig.DOMAIN_API_UPDATED_LIST)
            if JmModuleConfig.FLAG_API_CLIENT_REQUIRE_COOKIES and JmModuleConfig.APP_COOKIES:
                # noinspection PyUnresolvedReferences
                self._session.cookies.update(JmModuleConfig.APP_COOKIES)

    self._has_setup = True