v2.0.0: 一次大版本更新 (#33)

- 重构JmcomicClient,重新设计抽象层次,以及两种实现:HTMl、API,支持请求重试+域名更换的机制;
- 重构JmOption,更加优雅、简单的配置,支持更多文件类型;
- 优化debug机制,等等。
This commit is contained in:
hect0x7 2023-05-10 22:02:42 +08:00 committed by GitHub
parent abdde83f36
commit 10ef575779
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
22 changed files with 851 additions and 817 deletions

View File

@ -35,13 +35,20 @@ jobs:
pip install commonX -i https://pypi.org/project --upgrade
pip install -e ./
- name: 安装依赖项pull_request
if: ${{ github.event_name == 'pull_request' }}
run: |
python -m pip install --upgrade pip
pip install commonX -i https://pypi.org/project --upgrade
pip install -e ./
- name: 运行下载脚本
env:
JM_USERNAME: ${{ secrets.JM_USERNAME }}
JM_PASSWORD: ${{ secrets.JM_PASSWORD }}
run: |
cd ./usage/
python jmcomic_workflows.py
python jmcomic_workflow_download.py
- name: 压缩下载的漫画
run: |

View File

@ -4,6 +4,9 @@ on:
push:
branches: [ "dev", "master" ]
paths:
- '.github/workflows/*.yml' # 工作流定义
- 'usage/**/*.py' # 工作流脚本
- 'assets/config/*.yml' # option配置文件
- 'src/**/*.py' # 源码
- 'tests/**/*.py' # 测试代码

View File

@ -1,41 +1,24 @@
!!python/object:jmcomic.jm_option.JmOption
dir_tree: !!python/object:jmcomic.jm_option.DownloadDirTree
Bd: D:/jmcomic/download/
flag: Bd_Id_Index_image
version: '2.0'
debug: true
disable_jm_module_debug: false
download_convert_image_suffix: .jpg
download_image_then_decode: true
download_multi_thread_photo_batch_count: 10 # 一次多线程批量下载的数量,太多了禁漫服务器可能顶不住
download_multi_thread_photo_len_limit: 30
download_use_disk_cache: true
filepath: null
dir_rule:
base_dir: ${workspace}/download/
rule: Bd_Ptitle
client_config:
domain: jmcomic1.group
# 请求失败时重试的次数配个正数就行不配默认是5
client:
domain: []
impl: html
postman:
meta_data:
cookies: null
impersonate: chrome110
type: cffi
retry_times: 5
postman_type: cffi
postman_type_list:
- requests
- requests_Session
- cffi
- cffi_Session
meta_data:
proxies: null
allow_redirects: true
cookies: null
headers:
accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9
accept-language: zh-CN,zh;q=0.9
authority: jmcomic1.group
sec-ch-ua: '"Not_A Brand";v="99", "Google Chrome";v="109", "Chromium";v="109"'
sec-ch-ua-mobile: ?0
sec-ch-ua-platform: '"Windows"'
sec-fetch-dest: document
sec-fetch-mode: navigate
sec-fetch-site: none
sec-fetch-user: ?1
upgrade-insecure-requests: '1'
user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML,
like Gecko) Chrome/109.0.0.0 Safari/537.36
download:
cache: true
image:
decode: true
suffix: .jpg
threading:
batch_count: 30

View File

@ -1,27 +1,2 @@
!!python/object:jmcomic.jm_option.JmOption
dir_tree: !!python/object:jmcomic.jm_option.DownloadDirTree
Bd: $(workspace}/download/
flag: Bd_Id_Image
disable_jm_module_debug: false
download_convert_image_suffix: null
download_image_then_decode: true
download_multi_thread_photo_batch_count: 3
download_multi_thread_photo_len_limit: 5
download_use_disk_cache: true
client_config:
# 请求失败时,重试的次数,配个正数就行
retry_times: 10
domain: null
postman_type: cffi
postman_type_list:
- requests
- requests_Session
- cffi
- cffi_Session
meta_data:
proxies: null
allow_redirects: true
cookies: null
# Github Actions 测试用
version: '2.0'

View File

@ -1,56 +0,0 @@
!!python/object:jmcomic.jm_option.JmOption
dir_tree: !!python/object:jmcomic.jm_option.DownloadDirTree
Bd: D:/jmcomic/download/
flag: Bd_Id_Index_image
disable_jm_module_debug: false
download_convert_image_suffix: .png
download_image_then_decode: true
download_multi_thread_photo_batch_count: 10 # 一次多线程批量下载的数量,太多了禁漫服务器可能顶不住
download_multi_thread_photo_len_limit: 30
download_use_disk_cache: true
filepath: null
client_config:
domain: 18comic.vip
# 请求失败时重试的次数配个正数就行不配默认是5
retry_times: 5
postman_type: cffi
postman_type_list:
- requests
- requests_Session
- cffi
- cffi_Session
meta_data:
# 以下proxies的写法都可以:
# proxies: clash
# proxies: v2ray
# proxies: 127.0.0.1:7890
# proxies:
# http: http://127.0.0.1:7890
# https: https://127.0.0.1:7890
proxies: clash
# 允许重定向
allow_redirects: true
headers:
accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7
accept-language: zh-CN,zh;q=0.9
authority: 18comic.vip
cache-control: no-cache
pragma: no-cache
sec-ch-ua: '"Google Chrome";v="111", "Not(A:Brand";v="8", "Chromium";v="111"'
sec-ch-ua-mobile: ?0
sec-ch-ua-platform: '"Windows"'
sec-fetch-dest: document
sec-fetch-mode: navigate
sec-fetch-site: none
sec-fetch-user: ?1
upgrade-insecure-requests: '1'
user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML,
like Gecko) Chrome/111.0.0.0 Safari/537.36
# cookies为null作用是以非登录状态访问JM
cookies: null
# 但如果你要下载的是需要登录才能看的漫画你需要配置cookies的AVS的值如下。
# cookies:
# AVS: 你的AVS

View File

@ -0,0 +1,7 @@
# Github Actions 下载脚本配置
version: '2.0'
dir_tree:
Bd: /home/runner/work/jmcomic/download/
flag: Bd_Aauthor_Atitle_Pindex

View File

@ -1,24 +0,0 @@
!!python/object:jmcomic.jm_option.JmOption
dir_tree: !!python/object:jmcomic.jm_option.DownloadDirTree
Bd: /home/runner/work/jmcomic/download/
flag: Bd_Id_Index_image
disable_jm_module_debug: false
download_convert_image_suffix: null
download_image_then_decode: true
download_multi_thread_photo_batch_count: 15 # 一次多线程批量下载的数量,太多了禁漫服务器可能顶不住
download_multi_thread_photo_len_limit: 35
download_use_disk_cache: true
client_config:
domain: null # null表明动态获取最新可用的禁漫域名
retry_times: 5 # 请求失败时重试的次数配个正数就行不配默认是5
postman_type: cffi
postman_type_list:
- requests
- requests_Session
- cffi
- cffi_Session
meta_data:
cookies: null
impersonate: chrome110

View File

@ -2,6 +2,6 @@
# 被依赖方 <--- 使用方
# config <--- entity <--- toolkit <--- client <--- option
__version__ = '1.9.1'
__version__ = '2.0.0'
from .api import *

View File

@ -13,38 +13,35 @@ def download_album(jm_album_id, option=None):
return download_album_batch(jm_album_id, option)
option, jm_client = build_client(option)
album_detail: JmAlbumDetail = jm_client.get_album_detail(jm_album_id)
album: JmAlbumDetail = jm_client.get_album_detail(jm_album_id)
jm_debug('download_album',
f'获得album_detail成功准备下载。'
f'本子作者是【{album_detail.author}】,一共有{len(album_detail)}集本子')
jm_debug('album',
f'本子获取成功: [{album.id}], '
f'标题: [{album.title}], '
f'作者: [{album.author}], '
f'章节数: [{len(album)}]')
def download_photo(index: int,
photo_detail: JmPhotoDetail,
debug_topic='download_album_photo',
def download_photo(photo: JmPhotoDetail,
debug_topic='photo',
):
jm_client.ensure_photo_can_use(photo_detail)
jm_client.ensure_photo_can_use(photo)
jm_debug(debug_topic,
f"下载第[{index + 1}]集: "
f"图片数为[{len(photo_detail)}]"
f"标题为:({photo_detail.title}) "
f"-- photo {photo_detail.photo_id}")
f'开始下载章节: {photo.id} ({photo.album_id}[{photo.index}/{len(album)}]), '
f'标题: [{photo.title}], '
f'图片数为[{len(photo)}]')
download_by_photo_detail(
photo_detail,
option,
)
download_by_photo_detail(photo, option)
jm_debug(debug_topic,
f"下载完成:({photo_detail.title}) "
f"-- photo {photo_detail.photo_id}")
jm_debug(debug_topic, f'章节下载完成: {photo.id} ({photo.album_id}[{photo.index}/{len(album)}])')
thread_pool_executor(
iter_objs=enumerate(album_detail),
iter_objs=album,
apply_each_obj_func=download_photo,
)
jm_debug('album', f'本子下载完成: [{album.id}]')
def download_album_batch(jm_album_id_iter: Union[Iterable, Generator],
option=None,
@ -88,35 +85,34 @@ def download_by_photo_detail(photo_detail: JmPhotoDetail,
option, jm_client = build_client(option)
# 下载准备
use_cache = option.download_use_disk_cache
decode_image = option.download_image_then_decode
use_cache = option.download_cache
decode_image = option.download_image_decode
jm_client.ensure_photo_can_use(photo_detail)
# 下载每个图片的函数
def download_image(index, img_detail, debug_topic='download_images_of_photo'):
def download_image(index, image: JmImageDetail, debug_topic='image'):
img_save_path = option.decide_image_filepath(photo_detail, index)
# 已下载过,缓存命中
if use_cache is True and file_exists(img_save_path):
jm_debug(debug_topic, f'photo-{img_detail.aid}: '
f'图片{img_detail.filename}已下载过,'
f'命中磁盘缓存({img_save_path}')
jm_debug(debug_topic,
f'图片已存在: {image.aid}[{image.filename}] → {img_save_path}')
return
# 开始下载
jm_client.download_by_image_detail(
img_detail,
image,
img_save_path,
decode_image=decode_image,
)
jm_debug(debug_topic, f'photo-{img_detail.aid}: '
f'图片{img_detail.filename}下载完成:'
f'[{img_detail.img_url}] → [{img_save_path}]')
jm_debug(debug_topic,
f'图片下载完成: {image.aid}/{image.filename}, '
f'[{image.img_url}] → [{img_save_path}]')
length = len(photo_detail)
# 根据图片数,决定下载策略
if length <= option.download_multi_thread_photo_len_limit:
if length <= option.download_threading_batch_count:
# 如果图片数小的话,直接使用多线程下载,一张图一个线程。
multi_thread_launcher(
iter_objs=enumerate(photo_detail),
@ -127,7 +123,7 @@ def download_by_photo_detail(photo_detail: JmPhotoDetail,
multi_task_launcher_batch(
iter_objs=enumerate(photo_detail),
apply_each_obj_func=download_image,
batch_size=option.download_multi_thread_photo_batch_count
batch_size=option.download_threading_batch_count
)
@ -143,5 +139,5 @@ def build_client(option: Optional[JmOption]) -> Tuple[JmOption, JmcomicClient]:
def create_option(filepath: str) -> JmOption:
option = JmOption.create_from_file(filepath)
option = JmOption.from_file(filepath)
return option

View File

@ -1,245 +0,0 @@
from .jm_toolkit import *
class JmcomicClient(PostmanProxy):
retry_postman_type = RetryPostman
def __init__(self, postman: Postman, domain, retry_times=None):
super().__init__(postman.with_retry(retry_times or 5, self.retry_postman_type))
self.domain = domain
def download_image(self,
data_original: str,
img_save_path: str,
scramble_id: str,
photo_id: str = None,
decode_image=True,
):
"""
下载JM的图片
@param data_original: 图片url
@param img_save_path: 图片保存位置
@param scramble_id: 图片所在photo的scramble_id
@param photo_id: 图片所在photo的photo_id可为空
@param decode_image: 要保存的是解密后的图还是原图
"""
# 请求图片
resp = self.jm_get(data_original, is_api=False)
if self.is_empty_image(resp):
raise AssertionError(f'接收到的图片数据为空: {resp.url}')
# gif图无需加解密需要最先判断
if self.img_is_not_need_to_decode(data_original, resp):
JmImageSupport.save_resp_img(resp, img_save_path, False)
return
# 不解密图片,直接返回
if decode_image is False:
# 保存图片
JmImageSupport.save_resp_img(
resp,
img_save_path,
need_convert=suffix_not_equal(data_original, img_save_path),
)
JmImageSupport.save_resp_decoded_img(
resp=resp,
img_detail=JmImageDetail.of(
photo_id or JmcomicText.parse_to_photo_id(data_original),
scramble_id,
data_original,
),
filepath=img_save_path
)
def download_by_image_detail(self,
img_detail: JmImageDetail,
img_save_path,
decode_image=True,
):
self.download_image(
img_detail.img_url,
img_save_path,
img_detail.scramble_id,
photo_id=img_detail.aid,
decode_image=decode_image,
)
# -- get detail返回值是实体类 --
def get_album_detail(self, album_id) -> JmAlbumDetail:
# 参数校验
album_id = JmcomicText.parse_to_photo_id(album_id)
# 请求
resp = self.jm_get(f"/album/{album_id}")
# 用 JmcomicText 解析 html返回实体类
return JmcomicText.analyse_jm_album_html(resp.text)
def get_photo_detail(self, photo_id: str, album=True) -> JmPhotoDetail:
# 参数校验
photo_id = JmcomicText.parse_to_photo_id(photo_id)
# 请求
resp = self.jm_get(f"/photo/{photo_id}")
# 用 JmcomicText 解析 html返回实体类
photo_detail = JmcomicText.analyse_jm_photo_html(resp.text)
# 一并获取该章节的所处本子
if album is True:
photo_detail.from_album = self.get_album_detail(photo_detail.album_id)
return photo_detail
def ensure_photo_can_use(self, photo_detail: JmPhotoDetail):
# 检查 from_album
if photo_detail.from_album is None:
photo_detail.from_album = self.get_album_detail(photo_detail.album_id)
# 检查 page_arr 和 data_original_domain
if photo_detail.page_arr is None or photo_detail.data_original_domain is None:
new = self.get_photo_detail(photo_detail.photo_id, False)
new.from_album = photo_detail.from_album
photo_detail.__dict__.update(new.__dict__)
# -- search --
def search_album(self, search_query, main_tag=0) -> JmSearchPage:
params = {
'main_tag': main_tag,
'search_query': search_query,
}
resp = self.jm_get('/search/photos', params=params)
return JmSearchSupport.analyse_jm_search_html(resp.text)
# -- 对象方法 --
def of_api_url(self, api_path):
return f"{JmModuleConfig.PROT}{self.domain}{api_path}"
def jm_get(self, url, is_api=True, require_200=True, **kwargs):
"""
向禁漫发请求的统一入口
"""
url = self.of_api_url(url) if is_api is True else url
if is_api is True:
jm_debug("api", url)
resp = self.get(url, **kwargs)
if require_200 is True and resp.status_code != 200:
write_text('./resp.html', resp.text)
raise AssertionError(f"请求失败,"
f"响应状态码为{resp.status_code}"
f"URL=[{resp.url}]"
+
(f"响应文本=[{resp.text}]" if len(resp.text) < 50 else
f'响应文本过长(len={len(resp.text)}),不打印')
)
# 图片请求,直接返回
if is_api is False:
return resp
# 检查请求是否成功
self.require_resp_success_else_raise(resp, url)
return resp
# -- 类方法 --
@classmethod
def is_empty_image(cls, resp):
return resp.status_code != 200 or len(resp.content) == 0
@classmethod
def img_is_not_need_to_decode(cls, data_original: str, _resp):
return data_original.endswith('.gif')
# noinspection PyAttributeOutsideInit
def enable_cache(self, debug=False):
def wrap_func_cache(func_name, cache_dict_name):
if hasattr(self, cache_dict_name):
return
cache_dict = {}
setattr(self, cache_dict_name, cache_dict)
# 重载本对象的方法
func = getattr(self, func_name)
cache_hit_msg = f'【缓存命中】{cache_dict_name} ' + '→ [{}]' if debug is True else None
cache_miss_msg = f'【缓存缺失】{cache_dict_name} ' + '← [{}]' if debug is True else None
wrap_func = enable_cache(
cache_dict=cache_dict,
cache_hit_msg=cache_hit_msg,
cache_miss_msg=cache_miss_msg,
)(func)
setattr(self, func_name, wrap_func)
wrap_func_cache('get_photo_detail', 'album_cache_dict')
wrap_func_cache('get_album_detail', 'photo_cache_dict')
def get_jmcomic_url(self, postman=None):
return JmModuleConfig.get_jmcomic_url(postman or self)
def get_jmcomic_url_all(self, postman=None):
return JmModuleConfig.get_jmcomic_url_all(postman or self)
def login(self,
username,
password,
refresh_client_cookies=True,
id_remember='on',
login_remember='on',
):
data = {
'username': username,
'password': password,
'id_remember': id_remember,
'login_remember': login_remember,
'submit_login': '',
}
resp = self \
.get_root_postman() \
.post(self.of_api_url('/login'),
data=data,
allow_redirects=False,
)
if resp.status_code != 301:
raise AssertionError(f'登录失败,状态码为{resp.status_code}')
if refresh_client_cookies is True:
self['cookies'] = resp.cookies
return resp
@classmethod
def require_resp_success_else_raise(cls, resp, url):
# 1. 是否 album_missing
if resp.url.endswith('/error/album_missing'):
raise AssertionError(f'请求的本子不存在!({url})\n'
'原因可能为:\n'
'1. id有误检查你的本子/章节id\n'
'2. 该漫画只对登录用户可见请配置你的cookies\n')
# 2. 是否是特殊html页
cls.check_special_html(resp.text.strip(), url)
@classmethod
def check_special_html(cls, html: str, url=None):
html = html.strip()
error_msg = JmModuleConfig.JM_ERROR_RESPONSE_HTML.get(html, None)
if error_msg is None:
return
raise AssertionError(f'{error_msg}' + f': {url}' if url is not None else '')

View File

@ -0,0 +1,286 @@
from .jm_client_interface import *
class AbstractJmClient(
JmcomicClient,
PostmanProxy,
):
def __init__(self,
postman: Postman,
retry_times: int,
domain=None,
fallback_domain_list=None,
):
super().__init__(postman)
self.retry_times = retry_times
if fallback_domain_list is None:
fallback_domain_list = []
if domain is not None:
fallback_domain_list.insert(0, domain)
self.domain_list = fallback_domain_list
def get(self, url, **kwargs):
return self.request_with_retry(self.postman.get, url, **kwargs)
def post(self, url, **kwargs):
return self.request_with_retry(self.postman.post, url, **kwargs)
def of_api_url(self, api_path, domain):
return f'{JmModuleConfig.PROT}{domain}{api_path}'
def request_with_retry(self,
request,
url,
domain_index=0,
retry_count=0,
**kwargs,
):
if domain_index >= len(self.domain_list):
raise AssertionError("All domains failed.")
domain = self.domain_list[domain_index]
if not url.startswith(JmModuleConfig.PROT):
url = self.of_api_url(url, domain)
jm_debug('api', url)
if domain_index != 0 and retry_count != 0:
jm_debug(
f'请求重试',
', '.join([
f'次数: [{retry_count + 1}/{self.retry_times}]',
f'域名: [{domain} ({domain_index}/{len(self.domain_list)})]',
f'路径: [{url}]',
f'参数: [{kwargs if "login" not in url else "#login_form#"}]'
])
)
try:
return request(url, **kwargs)
except Exception as e:
self.before_retry(e, kwargs, retry_count, url)
if retry_count < self.retry_times:
return self.request_with_retry(request, url, domain_index, retry_count + 1, **kwargs)
else:
return self.request_with_retry(request, url, domain_index + 1, 0, **kwargs)
# noinspection PyMethodMayBeStatic, PyUnusedLocal
def before_retry(self, e, kwargs, retry_count, url):
jm_debug('error', str(e))
def enable_cache(self, debug=False):
def wrap_func_cache(func_name, cache_dict_name):
if hasattr(self, cache_dict_name):
return
cache_dict = {}
setattr(self, cache_dict_name, cache_dict)
# 重载本对象的方法
func = getattr(self, func_name)
cache_hit_msg = f'【缓存命中】{cache_dict_name} ' + '→ [{}]' if debug is True else None
cache_miss_msg = f'【缓存缺失】{cache_dict_name} ' + '← [{}]' if debug is True else None
wrap_func = enable_cache(
cache_dict=cache_dict,
cache_hit_msg=cache_hit_msg,
cache_miss_msg=cache_miss_msg,
)(func)
setattr(self, func_name, wrap_func)
wrap_func_cache('get_photo_detail', 'album_cache_dict')
wrap_func_cache('get_album_detail', 'photo_cache_dict')
wrap_func_cache('search_album', 'search_album_cache_dict')
# 基于网页实现的JmClient
class JmHtmlClient(AbstractJmClient):
def get_album_detail(self, album_id) -> JmAlbumDetail:
# 参数校验
album_id = JmcomicText.parse_to_photo_id(album_id)
# 请求
resp = self.get_jm_html(f"/album/{album_id}")
# 用 JmcomicText 解析 html返回实体类
return JmcomicText.analyse_jm_album_html(resp.text)
def get_photo_detail(self, photo_id: str, album=True) -> JmPhotoDetail:
# 参数校验
photo_id = JmcomicText.parse_to_photo_id(photo_id)
# 请求
resp = self.get_jm_html(f"/photo/{photo_id}")
# 用 JmcomicText 解析 html返回实体类
photo_detail = JmcomicText.analyse_jm_photo_html(resp.text)
# 一并获取该章节的所处本子
if album is True:
photo_detail.from_album = self.get_album_detail(photo_detail.album_id)
return photo_detail
def ensure_photo_can_use(self, photo_detail: JmPhotoDetail):
# 检查 from_album
if photo_detail.from_album is None:
photo_detail.from_album = self.get_album_detail(photo_detail.album_id)
# 检查 page_arr 和 data_original_domain
if photo_detail.page_arr is None or photo_detail.data_original_domain is None:
new = self.get_photo_detail(photo_detail.photo_id, False)
new.from_album = photo_detail.from_album
photo_detail.__dict__.update(new.__dict__)
def search_album(self, search_query, main_tag=0):
params = {
'main_tag': main_tag,
'search_query': search_query,
}
resp = self.get_jm_html('/search/photos', params=params, allow_redirects=True)
# 检查是否发生了重定向
# 因为如果搜索的是禁漫车号,会直接跳转到本子详情页面
if resp.redirect_count != 0 and '/album/' in resp.url:
return JmcomicText.analyse_jm_album_html(resp.text)
else:
return JmSearchSupport.analyse_jm_search_html(resp.text)
# -- 帐号管理 --
def login(self,
username,
password,
refresh_client_cookies=True,
id_remember='on',
login_remember='on',
):
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 != 301:
raise AssertionError(f'登录失败,状态码为{resp.status_code}')
if refresh_client_cookies is True:
self['cookies'] = resp.cookies
return resp
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:
write_text('./resp.html', resp.text)
self.check_special_http_code(resp.status_code, url)
raise AssertionError(f"请求失败,"
f"响应状态码为{resp.status_code}"
f"URL=[{resp.url}]"
+ (f"响应文本=[{resp.text}]" if len(resp.text) < 50 else
f'响应文本过长(len={len(resp.text)}),不打印')
)
# 检查请求是否成功
self.require_resp_success_else_raise(resp)
return resp
def get_jm_image(self, img_url) -> JmImageResp:
return JmImageResp(self.get(img_url))
@classmethod
def require_resp_success_else_raise(cls, resp):
# 1. 是否 album_missing
resp_url = resp.url
if resp_url.endswith('/error/album_missing'):
raise AssertionError(f'请求的本子不存在!({resp_url})\n'
'原因可能为:\n'
'1. id有误检查你的本子/章节id\n'
'2. 该漫画只对登录用户可见请配置你的cookies\n')
# 2. 是否是错误html页
cls.check_error_html(resp.text.strip(), resp_url)
@classmethod
def check_error_html(cls, html: str, url=None):
html = html.strip()
error_msg = JmModuleConfig.JM_ERROR_RESPONSE_HTML.get(html, None)
if error_msg is None:
return
write_text('./resp.html', html)
raise AssertionError(f'{error_msg}'
+ (f': {url}' if url is not None else ''))
@classmethod
def check_special_http_code(cls, code, url=None):
error_msg = JmModuleConfig.JM_ERROR_STATUS_CODE.get(int(code), None)
if error_msg is None:
return
raise AssertionError(f"请求失败,"
f"响应状态码为{code}"
f'原因为: [{error_msg}], '
+ (f'URL=[{url}]' if url is not None else '')
)
class JmApiClient(AbstractJmClient):
API_SEARCH = '/search'
def __init__(self,
base_url,
postman: Postman,
retry_times=5,
):
super().__init__(postman, retry_times)
self.base_url: str = base_url
def search_album(self, search_query: str, main_tag=0) -> JmApiResp:
return self.get(
self.API_SEARCH,
params={
'search_query': search_query,
}
)
def get(self, url, **kwargs) -> JmApiResp:
# set headers
headers, key_ts = self.headers_key_ts
kwargs.setdefault('headers', headers)
resp = super().get(url, **kwargs)
return JmApiResp.wrap(resp, key_ts)
@property
def headers_key_ts(self):
key_ts = time_stamp()
import hashlib
token = hashlib.md5(f"{key_ts}{JmModuleConfig.MAGIC_18COMICAPPCONTENT}".encode()).hexdigest()
return {
"token": token,
"tokenparam": f"{key_ts},1.5.2",
"user-agent": "okhttp/3.12.1",
"accept-encoding": "gzip",
}, key_ts

View File

@ -0,0 +1,212 @@
from .jm_toolkit import *
"""
Response Entity
"""
class JmResp(CommonResp):
@property
def is_success(self) -> bool:
return self.http_code == 200 and len(self.content) != 0
class JmImageResp(JmResp):
def require_success(self):
if self.is_success:
return
msg = f'禁漫图片获取失败: [{self.url}]'
if self.http_code != 200:
msg += f'http状态码={self.http_code}'
if len(self.content) == 0:
msg += f',响应数据为空'
raise AssertionError(msg)
def transfer_to(self,
path,
scramble_id,
decode_image=True,
img_url=None,
):
img_url = img_url or self.url
if decode_image is False:
# 不解密图片,直接返回
JmImageSupport.save_resp_img(
self,
path,
need_convert=suffix_not_equal(img_url, path),
)
else:
# 解密图片,需要 photo_id、scramble_id
JmImageSupport.decode_and_save(
JmImageSupport.get_num_by_url(scramble_id, img_url),
JmImageSupport.open_Image(self.content),
path,
)
class JmApiResp(JmResp):
@classmethod
def wrap(cls, resp, key_ts):
if isinstance(resp, JmApiResp):
raise AssertionError('重复包装')
return cls(resp, key_ts)
def __init__(self, resp, key_ts):
super().__init__(resp)
self.key_ts = key_ts
self.cache_decode_data = None
@staticmethod
def parseData(text, time):
import hashlib
import base64
from Crypto.Cipher import AES
# key为时间+18comicAPPContent的md5结果
key = hashlib.md5(f"{time}18comicAPPContent".encode()).hexdigest().encode()
cipher = AES.new(key, AES.MODE_ECB)
# 先将数据进行base64解码
data = base64.b64decode(text)
# 再进行AES-ECB解密
paddedPlainText = cipher.decrypt(data)
# 将得到的数据进行Utf8解码
res = paddedPlainText.decode('utf-8')
# 得到的数据再末尾有一些乱码
i = len(res) - 1
while i >= 0 and res[i] == '\x0c':
i -= 1
return res[:i + 1]
@property
def decoded_data(self) -> str:
if self.cache_decode_data is None:
self.cache_decode_data = self.parseData(self.encoded_data, self.key_ts)
return self.cache_decode_data
@property
def encoded_data(self) -> str:
return self.json()['data']
@property
def res_data(self) -> Any:
self.require_success()
from json import loads
return loads(self.decoded_data)
def json(self, **kwargs) -> Dict:
return self.resp.json()
def model(self) -> DictModel:
return DictModel(self.json())
@property
def model_data(self) -> DictModel:
self.require_success()
return DictModel(self.res_data)
"""
Client Interface
"""
class JmDetailClient:
def get_album_detail(self, album_id) -> JmAlbumDetail:
raise NotImplementedError
def get_photo_detail(self, photo_id: str, album=True) -> JmPhotoDetail:
raise NotImplementedError
def ensure_photo_can_use(self, photo_detail: JmPhotoDetail):
raise NotImplementedError
def search_album(self, search_query, main_tag=0) -> Union[JmSearchPage, JmAlbumDetail]:
raise NotImplementedError
def of_api_url(self, api_path, domain):
raise NotImplementedError
def enable_cache(self, debug=False):
raise NotImplementedError
class JmUserClient:
def login(self,
username,
password,
refresh_client_cookies=True,
id_remember='on',
login_remember='on',
):
raise NotImplementedError
class JmImageClient:
# -- 下载图片 --
def download_image(self,
img_url: str,
img_save_path: str,
scramble_id: str,
decode_image=True,
):
"""
下载JM的图片
@param img_url: 图片url
@param img_save_path: 图片保存位置
@param scramble_id: 图片所在photo的scramble_id
@param decode_image: 要保存的是解密后的图还是原图
"""
# 请求图片
resp = self.get_jm_image(img_url)
resp.require_success()
# gif图无需加解密需要最先判断
if self.img_is_not_need_to_decode(img_url, resp):
JmImageSupport.save_resp_img(resp, img_save_path, False)
else:
resp.transfer_to(img_save_path, scramble_id, decode_image, img_url)
def download_by_image_detail(self,
img_detail: JmImageDetail,
img_save_path,
decode_image=True,
):
self.download_image(
img_detail.img_url,
img_save_path,
img_detail.scramble_id,
decode_image=decode_image,
)
def get_jm_image(self, img_url) -> JmImageResp:
raise NotImplementedError
@classmethod
def img_is_not_need_to_decode(cls, data_original: str, _resp):
return data_original.endswith('.gif')
class JmcomicClient(
JmImageClient,
JmDetailClient,
JmUserClient,
):
pass

View File

@ -1,3 +1,8 @@
def default_jm_debug(topic: str, msg: str):
from common import format_ts
print(f'{format_ts()}:【{topic}{msg}')
class JmModuleConfig:
# 网站相关
PROT = "https://"
@ -13,11 +18,19 @@ class JmModuleConfig:
"Restricted Access!": "禁漫拒绝你所在ip地区的访问你可以选择: 换域名/换代理",
}
JM_ERROR_STATUS_CODE = {
520: '520: Web server is returning an unknown error (禁漫服务器内部报错)',
524: '524: The origin web server timed out responding to this request. (禁漫服务器处理超时)',
}
# 图片分隔相关
SCRAMBLE_0 = 220980
SCRAMBLE_10 = 268850
SCRAMBLE_NUM_8 = 421926 # 2023-02-08后改了图片切割算法
# API的相关配置
MAGIC_18COMICAPPCONTENT = '18comicAPPContent'
# 下载时的一些默认值
default_author = 'default-author'
default_photo_title = 'default-photo-title'
@ -25,10 +38,7 @@ class JmModuleConfig:
# debug
enable_jm_debug = True
debug_printer = print
# 缓存
jm_client_caches = {}
debug_executor = default_jm_debug
@classmethod
def domain(cls, postman=None):
@ -46,7 +56,7 @@ class JmModuleConfig:
@classmethod
def headers(cls, authority=None):
return {
'authority': authority or cls.domain(),
'authority': authority or '18comic.vip',
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,'
'application/signed-exchange;v=b3;q=0.7',
'accept-language': 'zh-CN,zh;q=0.9',
@ -64,9 +74,9 @@ class JmModuleConfig:
# noinspection PyUnusedLocal
@classmethod
def jm_debug(cls, topic: str, msg: str, from_class=None):
def jm_debug(cls, topic: str, msg: str):
if cls.enable_jm_debug is True:
cls.debug_printer(f'{topic}{msg}')
cls.debug_executor(topic, msg)
@classmethod
def disable_jm_debug(cls):
@ -88,7 +98,7 @@ class JmModuleConfig:
return url
@classmethod
def get_jmcomic_url_all(cls, postman=None):
def get_jmcomic_domain_all(cls, postman=None):
"""
访问禁漫发布页得到所有禁漫的域名
@return['18comic.vip', ..., 'jm365.xyz/ZNPJam'], 最后一个是APP軟件下載

View File

@ -26,9 +26,10 @@ class WorkEntity(JmBaseEntity, SaveableEntity, IterableEntity):
cls_name = self.__class__.__name__
return cls_name[cls_name.index("m") + 1: cls_name.rfind("Detail")].lower()
return '[{}]{}{}'.format(jm_type(), self.get_id(), self.detail_save_file_suffix)
return '[{}]{}{}'.format(jm_type(), self.id, self.detail_save_file_suffix)
def get_id(self) -> str:
@property
def id(self) -> str:
raise NotImplementedError
def __len__(self):
@ -120,6 +121,7 @@ class JmPhotoDetail(WorkEntity):
self.page_arr: List[str] = page_arr
# 图片域名
self.data_original_domain: StrNone = data_original_domain
self.index = self.album_index
@property
def is_single_album(self) -> bool:
@ -191,7 +193,8 @@ class JmPhotoDetail(WorkEntity):
def __getitem__(self, item) -> JmImageDetail:
return self.create_image_detail(item)
def get_id(self):
@property
def id(self):
return self.photo_id
def __len__(self):
@ -263,7 +266,8 @@ class JmAlbumDetail(WorkEntity):
def keywords(self) -> List[str]:
return self._keywords_list
def get_id(self):
@property
def id(self):
return self.album_id
def __len__(self):
@ -292,50 +296,12 @@ class JmAlbumDetail(WorkEntity):
class JmSearchPage(IterableEntity):
def __init__(self, album_info_list) -> None:
def __init__(self, album_info_list):
# (album_id, title, category_none, label_sub_none, tag_list)
self.data: List[Tuple[str, str, StrNone, StrNone, List[str]]] = album_info_list
def album_id_iter(self):
for album_info in self.data:
yield album_info[0]
self.album_info_list: List[Tuple[str, str, StrNone, StrNone, List[str]]] = album_info_list
def __len__(self):
return len(self.data)
return len(self.album_info_list)
def __getitem__(self, item):
return self.data[item]
# cdn爬取请求
class CdnRequest:
SavePathProvider = Callable[[str, str, int, bool], str]
def __init__(self,
photo_id,
scramble_id,
from_index: int,
photo_len: Optional[int],
save_path_provider: SavePathProvider,
):
self.photo_id = photo_id
self.scramble_id = scramble_id
self.from_index = from_index
self.photo_len = photo_len
self.save_path_provider = save_path_provider
@classmethod
def create(cls,
photo_id,
save_path_provider: SavePathProvider,
scramble_id,
from_index=1,
photo_len=None,
):
return CdnRequest(
photo_id,
scramble_id,
from_index,
photo_len,
save_path_provider,
)
return self.album_info_list[item][0:2]

View File

@ -1,4 +1,4 @@
from .jm_client import *
from .jm_client_impl import *
class JmOptionAdvice:
@ -63,188 +63,138 @@ class JmAdviceRegistry:
return cls.advice_registration.setdefault(base, [])
# 目录树的组成
class DownloadDirTree:
# 根目录 / Album作者 / Photo标题 / 图片文件
Bd_Author_Title_Image = 0
# 根目录 / Album作者 / Photo号 / 图片文件
Bd_Author_Id_Image = 1
# 根目录 / Album标题 / Photo标题 / 图片文件
Bd_Title_Title_Image = 2
# 根目录 / Album标题 / Photo号 / 图片文件
Bd_Title_Id_Image = 3
# 根目录 / Photo标题 / 图片文件
Bd_Title_Image = 4
# 根目录 / Photo号 / 图片文件
Bd_Id_Image = 5
# 根目录 / AlbumId / Photo序号 / 图片文件
Bd_Id_Index_image = 6 # 禁漫网站的默认下载方式
AdditionalHandler = Callable[
['DownloadDirTree', Optional[JmAlbumDetail], JmPhotoDetail],
str
class DirRule:
rule_sample = [
# 根目录 / Album-id / Photo-序号 /
'Bd_Aid_Pindex', # 禁漫网站的默认下载方式
# 根目录 / Album-作者 / Album-标题 / Photo-序号 /
'Bd_Aauthor_Atitle_Pindex',
]
additional_tree_flag_handler_mapping: Dict[int, AdditionalHandler] = {}
RuleFunc = Callable[[Union[JmAlbumDetail, JmPhotoDetail, None]], str]
RuleSolver = List[Tuple[int, RuleFunc]]
rule_solver_cache: Dict[Tuple[str, str], RuleSolver] = dict()
def __init__(self, rule, base_dir=None):
self.base_dir = self.parse_dsl(base_dir)
self.rule_dsl = rule
self.get_rule_solver()
def deside_image_save_dir(self,
album: JmAlbumDetail,
photo: JmPhotoDetail,
) -> str:
def choose_param(key):
if key == 0:
return None
if key == 1:
return album
if key == 2:
return photo
solver_ls = self.get_rule_solver()
path_ls = []
for i, (key, func) in enumerate(solver_ls):
try:
param = choose_param(key)
ret = func(param)
path_ls.append(str(ret))
except BaseException as e:
# noinspection PyUnboundLocalVariable
raise AssertionError(f'路径规则"{self.rule_dsl}"的第{i + 1}个解析出错: {e},'
f'param is {param}')
return fix_filepath('/'.join(path_ls), is_dir=True)
def get_rule_solver(self):
key = self.rule_dsl, self.base_dir
solver_ls = self.rule_solver_cache.get(key, None)
if solver_ls is None:
solver_ls = self.solve_rule_dsl(*key)
self.rule_solver_cache[key] = solver_ls
return solver_ls
def solve_rule_dsl(self, rule_dsl: str, base_dir: str) -> RuleSolver:
"""
解析下载路径dsl得到一个路径规则解析列表
"""
if '_' not in rule_dsl:
raise NotImplementedError(f'不支持的dsl: "{rule_dsl}"')
rule_ls = rule_dsl.split('_')
solver_ls = []
for rule in rule_ls:
if rule == 'Bd':
solver_ls.append((0, lambda _: base_dir))
continue
# Axxx or Pyyy
if not rule.startswith(('A', 'P')):
raise NotImplementedError(f'不支持的dsl: "{rule}" in "{rule_dsl}"')
key = 1 if rule[0] == 'A' else 2
solver_ls.append((
key,
lambda entity, ref=rule[1:]: fix_windir_name(str(getattr(entity, ref)))
))
return solver_ls
dsl_support = {
'${workspace}': lambda text: workspace(),
}
def fix_bd(self, Bd):
for dsl_k, func in self.dsl_support.items():
if dsl_k in Bd:
Bd = Bd.replace(dsl_k, func(Bd))
return fix_filepath(os.path.abspath(Bd), True)
def parse_dsl(self, base_dir: str):
for k, func in self.dsl_support.items():
if k in base_dir:
base_dir = base_dir.replace(k, func(base_dir))
return base_dir
class JmOption:
JM_OP_VER = '2.0'
def __init__(self,
Bd: str,
flag: Union[str, int],
dir_rule: Dict,
download: Dict,
client: Dict,
filepath=None,
):
self.Bd = self.fix_bd(Bd)
self.flag = self.get_flag_enum(flag)
def get_flag_enum(self, flag):
if isinstance(flag, int):
return flag
if not isinstance(flag, str):
raise NotImplementedError(flag)
ret = self.__class__.__dict__.get(flag, None)
if ret is None:
raise NotImplementedError(flag)
return ret
def deside_image_save_dir(self,
album: Optional[JmAlbumDetail],
photo: JmPhotoDetail,
) -> str:
if photo is None:
raise AssertionError('章节信息不能为None')
def dirpath(album_dir: Optional[str], photo_dir: str):
"""
@param album_dir: 相册文件夹名
@param photo_dir: 章节文件夹名
"""
from common import fix_windir_name
photo_dir = fix_windir_name(photo_dir.strip())
if album_dir is None:
return f'{self.Bd}{photo_dir}/'
album_dir = fix_windir_name(album_dir.strip())
return f'{self.Bd}{album_dir}/{photo_dir}/'
def photo_dir(flag_for_title):
return photo.title if flag == flag_for_title else photo.photo_id
flag = self.flag
if flag in (0, 1):
# 根目录 / Album作者 / Photo标题 / 图片文件
# 根目录 / Album作者 / Photo号 / 图片文件
author = album.author if album is not None else JmModuleConfig.default_author
return dirpath(author, photo_dir(0))
elif flag in (2, 3):
# 根目录 / Album标题 / Photo标题 / 图片文件
# 根目录 / Album标题 / Photo号 / 图片文件
album_title = album.title if album is not None else photo.title
return dirpath(album_title, photo_dir(2))
elif flag in (4, 5):
# 根目录 / Photo标题 / 图片文件
# 根目录 / Photo号 / 图片文件
return dirpath(None, photo_dir(4))
elif flag == 6:
# 根目录 / AlbumId / Photo序号 / 图片文件
return dirpath(photo.album_id, str(photo.album_index))
else:
if flag in self.additional_tree_flag_handler_mapping:
return self.additional_tree_flag_handler_mapping[flag](self, album, photo)
else:
raise NotImplementedError
@staticmethod
def random_id():
import common
return common.time_stamp()
@staticmethod
def random_title():
import common
return common.format_ts(f_time="%Y-%m-%d-%H-%M-%S")
@classmethod
def accept_flag_dict(cls):
return dict(filter(lambda item: item[0].startswith('Bd_'),
cls.__dict__.items()))
@classmethod
def of(cls,
tree_base_dir: str,
tree_flag: int,
):
for accpet_flag in cls.accept_flag_dict().values():
if accpet_flag is tree_flag:
return DownloadDirTree(tree_base_dir, tree_flag)
raise AssertionError(f'不支持的flag: {tree_flag}'
f'请使用DownloadDirTree类字段中的值或直接实例化DownloadDirTree对象')
class JmOption(SaveableEntity):
_proxies_mapping = {
'clash': ProxyBuilder.clash_proxy,
'v2Ray': ProxyBuilder.v2Ray_proxy
}
when_del_save_file = False
cache_jm_client = True
def __init__(self,
dir_tree: DownloadDirTree,
client_config: dict,
filepath: StrNone = None,
disable_jm_module_debug=False,
download_use_disk_cache=True,
download_convert_image_suffix: StrNone = None,
download_image_then_decode=True,
download_multi_thread_photo_len_limit=30,
download_multi_thread_photo_batch_count=10,
):
# 版本号
self.version = self.JM_OP_VER
# 路径规则配置
self.dir_rule = DirRule(**dir_rule)
# 请求配置
self.client_config = client_config
self.client_config['postman_type_list'] = list(Postmans.postman_impl_class_dict.keys())
self.client = DictModel(client)
# 下载配置
self.dir_tree = dir_tree
self.download_use_disk_cache = download_use_disk_cache
self.download_image_then_decode = download_image_then_decode
self.download_multi_thread_photo_len_limit = download_multi_thread_photo_len_limit
self.download_multi_thread_photo_batch_count = download_multi_thread_photo_batch_count
# suffix的标准形式是 ".xxx"。如果传入的是"xxx",要补成 ".xxx"
if download_convert_image_suffix is not None:
download_convert_image_suffix = fix_suffix(download_convert_image_suffix)
self.download_convert_image_suffix = download_convert_image_suffix
self.download = DictModel(download)
# 其他配置
self.filepath = filepath
self.disable_jm_module_debug = disable_jm_module_debug
if disable_jm_module_debug:
disable_jm_debug()
# 字段
self.jm_client_cache = None
@property
def download_cache(self):
return self.download.cache
@property
def download_image_decode(self):
return self.download.image.decode
@property
def download_threading_batch_count(self):
return self.download.threading.batch_count
@property
def download_image_suffix(self):
return self.download.image.suffix
"""
下面是决定图片保存路径的方法
@ -257,8 +207,8 @@ class JmOption(SaveableEntity):
if save_dir is not None:
return save_dir
# 使用 self.dir_tree 决定 save_dir
save_dir = self.dir_tree.deside_image_save_dir(
# 使用 self.dir_rule 决定 save_dir
save_dir = self.dir_rule.deside_image_save_dir(
photo_detail.from_album,
photo_detail
)
@ -279,7 +229,7 @@ class JmOption(SaveableEntity):
return suffix
# 非动图,以配置为先
return self.download_convert_image_suffix or suffix
return self.download_image_suffix or suffix
def decide_image_filepath(self, photo_detail: JmPhotoDetail, index: int) -> str:
# 先检查advice的回调如果回调支持则优先使用回调
@ -306,161 +256,128 @@ class JmOption(SaveableEntity):
"""
@classmethod
def default(cls) -> 'JmOption':
return JmOption(
DownloadDirTree(
workspace(),
DownloadDirTree.Bd_Title_Image,
),
cls.default_client_config(),
)
def construct(cls, dic: Dict, cover_default=True) -> 'JmOption':
if cover_default:
dic = cls.merge_default_dict(dic)
@classmethod
def create_from_file(cls, filepath: str) -> 'JmOption':
jm_option: JmOption = PackerUtil.unpack(filepath, JmOption)[0]
jm_option.filepath = filepath
return jm_option
version = dic.pop('version', None)
if float(version) != float(cls.JM_OP_VER):
# 版本兼容
raise NotImplementedError('不支持的option版本')
@classmethod
def default_client_config(cls):
debug = dic.pop('debug', True)
if debug is False:
disable_jm_debug()
return cls(**dic)
def deconstruct(self) -> Dict:
return {
'domain': JmModuleConfig.domain(),
'meta_data': {
'cookies': None,
'headers': JmModuleConfig.headers(),
'allow_redirects': True,
'version': self.version,
'debug': JmModuleConfig.enable_jm_debug,
'dir_rule': {
'rule': self.dir_rule.rule_dsl,
'base_dir': self.dir_rule.base_dir,
},
'postman_type_list': [
'requests',
'requests_Session',
'cffi',
'cffi_Session',
]
'download': self.download.src_dict,
'client': self.client.src_dict,
}
def save_base_dir(self):
return of_dir_path(self.filepath)
@classmethod
def from_file(cls, filepath: str) -> 'JmOption':
dic: dict = PackerUtil.unpack(filepath)[0]
return cls.construct(dic)
def save_file_name(self) -> str:
return of_file_name(self.filepath)
def save_to_file(self, filepath=None):
def to_file(self, filepath=None):
if filepath is None:
filepath = self.filepath
if filepath is None:
raise AssertionError("未指定JmOption的保存路径")
super().save_to_file(filepath)
PackerUtil.pack(self.deconstruct(), filepath)
"""
下面是 build 方法
"""
# 缓存
cache_jm_client = True
jm_client_impl_mapping: Dict[str, Type[AbstractJmClient]] = {
'html': JmHtmlClient,
'api': JmApiClient,
}
def build_jm_client(self) -> JmcomicClient:
if self.cache_jm_client is not True:
return self.new_jm_client()
client = self.jm_client_cache
if client is None:
client = self.new_jm_client()
else:
key = self
client = JmModuleConfig.jm_client_caches.get(key, None)
if client is None:
client = self.new_jm_client()
JmModuleConfig.jm_client_caches.setdefault(key, client)
self.jm_client_cache = client
return client
def new_jm_client(self) -> JmcomicClient:
meta_data = self.client_config['meta_data']
postman_clazz = Postmans.get_impl_clazz(self.client_config.get('postman_type', 'cffi'))
proxies = None
domain = None
postman: Optional[Postman] = None
# postman
postman = Postmans.create(data=self.client.postman)
def decide_proxies(key='proxies'):
nonlocal proxies
proxies = meta_data.get(key, None)
# domain_list
domain_list = self.client.domain
if len(domain_list) == 0:
domain_list = JmModuleConfig.get_jmcomic_domain_all(postman)[:-1]
# 无代理,或代理已配置好好的
if proxies is None or isinstance(proxies, dict):
return
# 有代理
if proxies in self._proxies_mapping:
proxies = self._proxies_mapping[proxies]()
else:
proxies = ProxyBuilder.build_proxy(proxies)
meta_data[key] = proxies
def decide_domain(key='domain') -> str:
nonlocal domain
domain = self.client_config.get(key, None)
if domain is None:
temp_postman = postman_clazz.create(
headers=JmModuleConfig.headers(JmModuleConfig.JM_REDIRECT_URL),
proxies=proxies,
)
domain = JmModuleConfig.domain(temp_postman)
domain = JmcomicText.parse_to_jm_domain(domain)
self.client_config[key] = domain
return domain
def handle_headers(key='headers'):
headers = meta_data.get(key, None)
if headers is None or (not isinstance(headers, dict)) or len(headers) == 0:
# 未配置headers使用默认headers
headers = JmModuleConfig.headers(domain)
meta_data[key] = headers
def handle_postman():
nonlocal postman
postman = postman_clazz(meta_data)
# 1. 决定 代理
decide_proxies()
# 2. 指定 JM域名
decide_domain()
# 3. 处理 headers
handle_headers()
# 4. 创建 postman
handle_postman()
jm_debug('创建JmcomicClient', f'使用域名: {domain}使用Postman实现: {postman_clazz}')
# 创建 JmcomicClient 对象
client = JmcomicClient(
postman=postman,
domain=domain,
retry_times=self.client_config.get('retry_times', None)
# client
client = self.jm_client_impl_mapping[self.client.impl](
postman,
self.client.retry_times,
fallback_domain_list=domain_list,
)
return client
@classmethod
def default_dict(cls) -> Dict:
return {
'version': '2.0',
'debug': True,
'dir_rule': {'rule': 'Bd_Ptitle', 'base_dir': workspace()},
'download': {
'cache': True,
'image': {'decode': True, 'suffix': None},
'threading': {'batch_count': 30},
},
'client': {
'domain': [],
'postman': {
'type': 'cffi',
'meta_data': {
'impersonate': 'chrome110',
'cookies': None,
'headers': JmModuleConfig.headers(),
}
},
'impl': 'html',
'retry_times': 5
}
}
def _register_yaml_constructor():
from yaml import add_constructor, Loader, Node
@classmethod
def default(cls):
return cls.construct({})
tag_mapping = {
'tag:yaml.org,2002:python/object:jmcomic.jm_option.JmOption': JmOption,
'tag:yaml.org,2002:python/object:jmcomic.jm_option.DownloadDirTree': DownloadDirTree,
}
@classmethod
def merge_default_dict(cls, user_dict, default_dict=None):
"""
深度合并两个字典
"""
if default_dict is None:
default_dict = cls.default_dict()
def constructor(loader: Loader, node: Node):
for tag, clazz in tag_mapping.items():
if node.tag == tag:
state = loader.construct_mapping(node)
try:
obj = clazz(**state)
except TypeError as e:
raise AssertionError(f"构造函数不匹配: {clazz.__name__}\nTypeError: {e}")
# obj.__dict__.update(state)
return obj
for tag in tag_mapping.keys():
add_constructor(tag, constructor)
_register_yaml_constructor()
for key, value in user_dict.items():
if isinstance(value, dict) and isinstance(default_dict.get(key), dict):
default_dict[key] = cls.merge_default_dict(value, default_dict[key])
else:
default_dict[key] = value
return default_dict

View File

@ -207,7 +207,7 @@ class JmImageSupport:
filepath: str
) -> None:
cls.decode_and_save(
cls.calculate_segmentation_num(img_detail),
cls.get_num_by_detail(img_detail),
cls.open_Image(resp.content),
filepath
)
@ -219,7 +219,7 @@ class JmImageSupport:
decoded_save_path: str
) -> None:
cls.decode_and_save(
cls.calculate_segmentation_num(img_detail),
cls.get_num_by_detail(img_detail),
cls.open_Image(img_filepath),
decoded_save_path
)
@ -275,12 +275,13 @@ class JmImageSupport:
return Image.open(fp)
@classmethod
def calculate_segmentation_num(cls, detail: JmImageDetail) -> int:
def get_num(cls, scramble_id, aid, filename: str) -> int:
"""
获得图片分割数
"""
scramble_id = int(detail.scramble_id)
aid = int(detail.aid)
scramble_id = int(scramble_id)
aid = int(aid)
if aid < scramble_id:
return 0
@ -289,10 +290,28 @@ class JmImageSupport:
else:
import hashlib
x = 10 if aid < JmModuleConfig.SCRAMBLE_NUM_8 else 8
s = f"{aid}{detail.img_file_name}" # 拼接
s = f"{aid}{filename}" # 拼接
s = s.encode()
s = hashlib.md5(s).hexdigest()
num = ord(s[-1])
num %= x
num = num * 2 + 2
return num
@classmethod
def get_num_by_url(cls, scramble_id, url) -> int:
"""
获得图片分割数
"""
return cls.get_num(
scramble_id,
aid=JmcomicText.parse_to_photo_id(url),
filename=of_file_name(url, True),
)
@classmethod
def get_num_by_detail(cls, detail: JmImageDetail) -> int:
"""
获得图片分割数
"""
return cls.get_num(detail.scramble_id, detail.aid, detail.img_file_name)

View File

@ -1,4 +1,3 @@
import sys
import io
import platform
import unittest
@ -63,12 +62,7 @@ class JmTestConfigurable(unittest.TestCase):
@classmethod
def adapt_win(cls):
# 尝试更新 cookies
cookies = ChromePluginCookieParser({'remember', 'comic'}) \
.apply(when_valid_message="更新jmcomic-option成功")
if cookies is not None:
cls.option.client_config['meta_data']['cookies'] = cookies
cls.option.save_to_file()
pass
@classmethod
def adapt_linux(cls):

View File

@ -115,6 +115,6 @@ class Test_Api(JmTestConfigurable):
)
def test_get_jmcomic_url(self):
print(JmModuleConfig.get_jmcomic_url_all())
print(self.client.get_jmcomic_url())
print(self.client.get_jmcomic_url_all())
print(JmModuleConfig.get_jmcomic_domain_all())
print(JmModuleConfig.get_jmcomic_url(self.client))
print(JmModuleConfig.get_jmcomic_domain_all(self.client))

View File

@ -40,9 +40,9 @@ class Test_Client(JmTestConfigurable):
print(f'本子: [{album_detail.title}] 一共有{album_detail.page_count}页图')
def test_search(self):
jm_search_page = self.client.search_album('MANA')
for album_id in jm_search_page.album_id_iter():
print(album_id)
jm_search_page: JmSearchPage = self.client.search_album('MANA')
for album_id, title in reversed(jm_search_page):
print(album_id, title)
def test_gt_300_photo(self):
photo_id = '147643'

View File

@ -1,16 +0,0 @@
from test_jmcomic import *
class Test_Search(JmTestConfigurable):
def test_analyse_search_html(self):
search_album: JmSearchPage = self.client.search_album('MANA')
for i, info in enumerate(search_album):
# info: (album_id, title, category_none, label_sub_none, tag_list)
print(f'index: {i}--------------------------------------------------------\n'
f'id: {info[0]}\n'
f'title: {info[1]}\n'
f'category_none: {info[2]}\n'
f'label_sub_none: {info[3]}\n'
f'tag_list: {info[4]}\n\n'
)

View File

@ -9,7 +9,7 @@ jmcomic.download_album('422866') # 传入要下载的album的id即可下载
# 上面的这行代码还有一个可选参数option: JmOption表示配置项
# 配置项的作用是告诉程序下载时候的一些选择,
# 比如,要下载到哪个文件夹,使用怎样的路径组织方式(比如[/作者/本子id/图片] 或者 [/作者/本子名称/图片].
# 如果没有配置,则会使用 JmOption.default(),下载的路径是[当前工作文件夹/本子名称/图片].
# 如果没有配置,则会使用 JmOption.default(),下载的路径是[当前工作文件夹/本子章节名称/图片].
"""
@ -48,10 +48,10 @@ print(url)
# 先获取默认的JmOption对象
jm_option = jmcomic.JmOption.default()
# 可以把对象保存为yml文件方便日后修改。保存格式会根据后缀名智能转换。
jm_option.save_to_file('./禁漫下载默认配置.yml') # yml格式推荐
jm_option.save_to_file('./禁漫下载默认配置.json') # json格式
# 可以把对象保存为文件,方便编辑
jm_option.to_file('./你的配置文件路名称.yml') # yml格式
jm_option.to_file('./你的配置文件路名称.json') # json格式
# 如果你修改了默认配置,现在想用你修改后的配置来下载,使用如下代码
jm_option = jmcomic.create_option('./禁漫下载默认配置.yml')
jm_option = jmcomic.create_option('./你的配置文件路名称.yml')
jmcomic.download_album('23333', jm_option)

View File

@ -12,7 +12,7 @@ def main():
def get_option():
# 读取 option 配置文件
option = create_option('../assets/config/workflow_option.yml')
option = create_option('../assets/config/option_workflow_download.yml')
# 启用 client 的缓存
client = option.build_jm_client()