init commit

This commit is contained in:
hect0x7 2023-04-02 22:19:44 +08:00
commit 1df7e64f8c
22 changed files with 2452 additions and 0 deletions

45
.github/workflows/action_workflow.yml vendored Normal file
View File

@ -0,0 +1,45 @@
name: 下载JM本子
on:
push:
branches: [ "workflow" ]
paths:
- 'usage/**/*.py' # 工作流脚本
- 'assets/config/*.yml' # 工作流配置
pull_request:
branches: [ "workflow" ]
permissions:
contents: read
jobs:
build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python 3.10
uses: actions/setup-python@v3
with:
python-version: "3.10"
- name: 安装依赖项
run: |
python -m pip install --upgrade pip
pip install jmcomic -i https://pypi.org/project --upgrade
- name: 运行下载脚本
run: |
cd ./usage/
python jmcomic_workflows.py
- name: 上传结果
uses: actions/upload-artifact@v3.1.2
with:
# Artifact name
name: 下载完成的本子
# A file, directory or wildcard pattern that describes what to upload
path: /home/runner/work/jmcomic/download/
# The desired behavior if no files are found using the provided path.
if-no-files-found: warn
# Duration after which artifact will expire in days. 0 means using default retention.Minimum 1 day. Maximum 90 days unless changed from the repository settings page.
retention-days: 90

View File

@ -0,0 +1,51 @@
# 这个工作流使用workflow_local分支的jmcomic模块。
# 你可以在workflow_local分支修改jmcomic的源码
# 这样工作流就会使用你修改后的代码来下载禁漫本子。
name: 下载JM本子 (Local)
on:
push:
branches: [ "workflow_local" ]
paths:
- 'src/**/*.py' # 源码
- 'usage/**/*.py' # 工作流脚本
- 'assets/config/*.yml' # 工作流配置
pull_request:
branches: [ "workflow_local" ]
permissions:
contents: read
jobs:
build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python 3.10
uses: actions/setup-python@v3
with:
python-version: "3.10"
- name: 安装依赖项
run: |
python -m pip install --upgrade pip
pip install commonX -i https://pypi.org/project --upgrade
pip install -e ./
- name: 运行下载脚本
run: |
cd ./usage/
python jmcomic_workflows.py
- name: 上传结果
uses: actions/upload-artifact@v3.1.2
with:
# Artifact name
name: 下载完成的本子
# A file, directory or wildcard pattern that describes what to upload
path: /home/runner/work/jmcomic/download/
# The desired behavior if no files are found using the provided path.
if-no-files-found: warn
# Duration after which artifact will expire in days. 0 means using default retention.Minimum 1 day. Maximum 90 days unless changed from the repository settings page.
retention-days: 90

25
.github/workflows/publish_pypi.yml vendored Normal file
View File

@ -0,0 +1,25 @@
name: 发布PYPI
on:
push:
branches: [ "releases/**" ]
jobs:
build-and-publish:
name: 构建并发布 jmcomic 到 PyPI
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@main
- name: Set up Python 3.7
uses: actions/setup-python@v3
with:
python-version: "3.7"
- name: 构建模块
run: |
python -m pip install build
python -m build
- name: 发布上传
uses: pypa/gh-action-pypi-publish@release/v1
with:
password:
${{ secrets.PYPI_JMCOMIC }}

164
.gitignore vendored Normal file
View File

@ -0,0 +1,164 @@
# this repo
/assets/download/
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/#use-with-ide
.pdm.toml
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
.idea/

21
LICENSE Normal file
View File

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2023 hect0x7
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

72
README.md Normal file
View File

@ -0,0 +1,72 @@
# Python API For JMComic (禁漫天堂)
封装了一套可用于爬取JM的Python API.
简单来说就是可以通过简单的几行Python代码实现下载JM上的本子到本地并且是处理好的图片.
**友情提示珍爱JM为了减轻JM的服务器压力请不要一次性爬取太多本子西门🙏🙏🙏**.
## 安装教程
* 通过pip官方源安装推荐
```shell
pip install jmcomic -i https://pypi.org/project --upgrade
```
* 本地安装
```shell
cd ./modules/core/
pip install -e ./
```
## 快速上手
使用下面的两行代码即可实现功能把某个本子集album里的所有本子photo下载到本地
```python
import jmcomic # 导入此模块,需要先安装.
jmcomic.download_album('422866') # 传入要下载的album的id即可下载整个album到本地.
# 上面的这行代码还有一个可选参数option: JmOption表示配置项
# 配置项的作用是告诉程序下载时候的一些选择,
# 比如,要下载到哪个文件夹,使用怎样的路径组织方式(比如[/作者/本子id/图片] 或者 [/作者/本子名称/图片].
# 如果没有配置,则会使用 JmOption.default(),下载的路径是[当前工作文件夹/本子名称/图片].
```
进一步的使用可以参考usage文件夹下的示例代码: `jmcomic_getting_started.py` `jmcomic_usage.py`
## 项目特点
- **绕过Cloudflare的反爬虫**
- 支持使用**Github Action**下载漫画,不会编程都能用([教程使用Github Actions下载禁漫本子](./assets/docs/教程使用Github%20Actions下载禁漫本子.md)
- 可配置性强
- 不配置也能使用,十分方便
- 配置可以从**配置文件**生成无需写Python代码
- 配置点有:`是否使用磁盘缓存` `是否使用代理` `图片类型转换` `本子下载路径` `请求元信息headers,cookies,重试次数)等 `
- 多线程下载(可细化到一图一线程,效率极高)
- 跟进了JM最新的图片分割算法2023-02-08
## 使用小说明
* Python >= 3.7
* 项目只有代码注释没有API文档。因此想深入高级地使用自行看源码和改造代码叭 ^^_
* JM不是前后端分离的网站因此本api是通过正则表达式解析HTML网页的信息详见`JmcomicText`),进而实现的下载图片。
## 项目文件夹介绍
* assets存放一些非代码的资源文件
* config存放配置文件
* docs项目文档
* src存放源代码
* jmcomic`jmcomic`模块
* tests测试目录存放测试代码使用unittest
* usage用法目录存放示例/使用代码
## 感谢以下项目
### 图片分割算法的来源
[![Readme Card](https://github-readme-stats.vercel.app/api/pin/?username=tonquer&repo=JMComic-qt)](https://github.com/tonquer/JMComic-qt)

41
assets/config/option.yml Normal file
View File

@ -0,0 +1,41 @@
!!python/object:jmcomic.jm_option.JmOption
dir_tree: !!python/object:jmcomic.jm_option.DownloadDirTree
Bd: D:/jmcomic/download/
flag: Bd_Title_Image
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
client_config:
domain: jmcomic1.group
# 请求失败时重试的次数配个正数就行不配默认是5
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.rocks
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

View File

@ -0,0 +1,56 @@
!!python/object:jmcomic.jm_option.JmOption
dir_tree: !!python/object:jmcomic.jm_option.DownloadDirTree
Bd: D:/jmcomic/download/
flag: Bd_Title_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,39 @@
!!python/object:jmcomic.jm_option.JmOption
dir_tree: !!python/object:jmcomic.jm_option.DownloadDirTree
Bd: /home/runner/work/jmcomic/download/
flag: Bd_Author_Title_Image
disable_jm_module_debug: false
download_convert_image_suffix: null
download_image_then_decode: true
download_multi_thread_photo_batch_count: 10 # 一次多线程批量下载的数量,太多了禁漫服务器可能顶不住
download_multi_thread_photo_len_limit: 30
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:
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.rocks
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

4
requirements-dev.txt Normal file
View File

@ -0,0 +1,4 @@
commonX
curl_cffi
PyYAML
Pillow

48
setup.py Normal file
View File

@ -0,0 +1,48 @@
from setuptools import setup, find_packages
with open("README.md", encoding='utf-8') as f:
long_description = f.read()
version = None
with open('./src/jmcomic/__init__.py', encoding='utf-8') as f:
for line in f:
if '__version__' in line:
version = line[line.index("'") + 1: line.rindex("'")]
break
if version is None:
print('Set version first!')
exit(1)
setup(
name='jmcomic',
version=version,
description='Python API For JMComic (禁漫天堂)',
long_description_content_type="text/markdown",
long_description=long_description,
url='https://github.com/hect0x7/JMComic-Crawler-Python',
author='hect0x7',
author_email='93357912+hect0x7@users.noreply.github.com',
packages=find_packages("src"),
package_dir={"": "src"},
python_requires=">=3.7",
install_requires=[
'commonX',
'curl_cffi',
'PyYAML',
'Pillow',
],
keywords=['python', 'jmcomic', '18comic', '禁漫天堂', 'NSFW'],
classifiers=[
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Operating System :: MacOS",
"Operating System :: POSIX :: Linux",
"Operating System :: Microsoft :: Windows",
],
)

7
src/jmcomic/__init__.py Normal file
View File

@ -0,0 +1,7 @@
# 模块依赖关系如下:
# 被依赖方 <--- 使用方
# config <--- entity <--- toolkit <--- client <--- service <--- option
__version__ = '1.0.0'
from .api import *

180
src/jmcomic/api.py Normal file
View File

@ -0,0 +1,180 @@
from .jm_option import *
def download_album(jm_album_id, option=None):
"""
下载一个本子集入口api
@param jm_album_id: 禁漫的本子的id类型可以是str/int/iterable[str]
如果是iterable[str]则会调用批量下载方法 download_album_batch
@param option: 下载选项为空默认是 JmOption.default()
"""
if not isinstance(jm_album_id, (str, int)):
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)
jm_debug('download_album',
f'获得album_detail成功准备下载。'
f'本子作者是【{album_detail.author}】,一共有{len(album_detail)}集本子')
def download_photo(index, photo_detail: JmPhotoDetail, debug_topic='download_album_photo'):
jm_client.fill_page_arr(photo_detail)
jm_debug(debug_topic,
f"下载第[{index + 1}]集: "
f"图片数为[{len(photo_detail)}]"
f"标题为:({photo_detail.title}) "
f"-- photo {photo_detail.photo_id}")
download_by_photo_detail(
photo_detail,
option,
)
jm_debug(debug_topic,
f"下载完成:({photo_detail.title}) "
f"-- photo {photo_detail.photo_id}")
multi_thread_launcher(
iter_objs=enumerate(album_detail),
apply_each_obj_func=download_photo,
wait_finish=True,
)
def download_album_batch(jm_album_id_iter: Union[Iterable, Generator],
option=None,
wait_finish=True) -> List[Thread]:
"""
批量下载album每个album一个线程使用的是同一个option
@param jm_album_id_iter: album_id的可迭代对象
@param option: 下载选项为空默认是 JmOption.default()
@param wait_finish: 是否要等待这些下载线程全部完成
@return 返回值是List[Thread]里面是每个下载漫画的线程
"""
if option is None:
option = JmOption.default()
return multi_thread_launcher(
iter_objs=((album_id, option) for album_id in jm_album_id_iter),
apply_each_obj_func=download_album,
wait_finish=wait_finish,
)
def download_photo(jm_photo_id, option=None):
"""
下载一个本子的一章入口api
"""
option, jm_client = build_client(option)
photo_detail = jm_client.get_photo_detail(jm_photo_id)
download_by_photo_detail(photo_detail, option)
def download_by_photo_detail(photo_detail: JmPhotoDetail,
option=None,
):
"""
下载一个本子的一章根据 photo_detail
@param photo_detail: 本子章节信息
@param option: 选项
"""
option, jm_client = build_client(option)
# 下载准备
use_cache = option.download_use_disk_cache
decode_image = option.download_image_then_decode
if photo_detail.from_album is None:
jm_client.fill_from_album(photo_detail)
if photo_detail.page_arr is None:
jm_client.fill_page_arr(photo_detail)
# 下载每个图片的函数
def download_image(index, debug_topic='download_images_of_photo'):
img_detail = photo_detail[index]
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_detail.img_url}')
return
# 开始下载
jm_client.download_by_image_detail(
img_detail,
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}]')
length = len(photo_detail)
# 根据图片数,决定下载策略
if length <= option.download_multi_thread_photo_len_limit:
# 如果图片数小的话,直接使用多线程下载,一张图一个线程。
multi_thread_launcher(
iter_objs=range(len(photo_detail)),
apply_each_obj_func=download_image,
wait_finish=True
)
else:
# 如果图片数多的话,还是分批下载。
batch_count = option.download_multi_thread_photo_batch_count
batch_times = length // batch_count
for i in range(batch_times):
begin = i * batch_count
multi_thread_launcher(
iter_objs=range(begin, begin + batch_count),
apply_each_obj_func=download_image,
)
multi_thread_launcher(
iter_objs=range(batch_times * batch_count,
length),
apply_each_obj_func=download_image,
)
def renew_jm_default_domain():
"""
由于禁漫的域名经常变化调用此方法可以获取一个当前可用的最新的域名 domain
并且设置把 domain 设置为禁漫模块的默认域名
这样一来配置文件也不用配置域名了一切都在运行时动态获取
"""
domain = JmcomicText.parse_to_jm_domain(JmModuleConfig.get_jmcomic_url())
JmModuleConfig.DOMAIN = domain
return domain
def build_client(option: Optional[JmOption]) -> Tuple[JmOption, JmcomicClient]:
"""
处理option的判空并且创建jm_client
"""
if option is None:
option = JmOption.default()
option.client_config['domain'] = renew_jm_default_domain()
jm_client = option.build_jm_client()
return option, jm_client
def create_option(filepath: str) -> JmOption:
"""
创建 JmOption同时检查域名是否配置未配置则补上配置
@param filepath:
@return:
"""
option = JmOption.create_from_file(filepath)
client_config = option.client_config
key = 'domain'
if client_config.get(key, None) is None or client_config[key] is None:
client_config[key] = renew_jm_default_domain()
return option

186
src/jmcomic/jm_client.py Normal file
View File

@ -0,0 +1,186 @@
from .jm_toolkit import *
class JmcomicClient(PostmanProxy):
debug_from_class = 'jm-client'
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, jm_photo_id: str) -> JmPhotoDetail:
# 参数校验
photo_id = JmcomicText.parse_to_photo_id(jm_photo_id)
# 请求
resp = self.jm_get(f"/photo/{photo_id}")
# 用 JmcomicText 解析 html返回实体类
return JmcomicText.analyse_jm_photo_html(resp.text)
def fill_from_album(self, photo_detail: JmPhotoDetail) -> JmAlbumDetail:
"""
获取 photo_detail 所在的 album
并把 album_detail 赋值给 photo_detail.from_album
"""
album_detail = self.get_album_detail(photo_detail.album_id)
photo_detail.from_album = album_detail
return album_detail
def fill_page_arr(self, photo_detail: JmPhotoDetail):
new = self.get_photo_detail(photo_detail.photo_id)
photo_detail.page_arr = new.page_arr
photo_detail.data_original_domain = new.data_original_domain
return new
# -- 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.HTTP}{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:
self.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 True and resp.text.strip() == JmModuleConfig.JM_SERVER_ERROR_HTML:
raise AssertionError("【JM异常】Could not connect to mysql! Please check your database settings!")
return resp
# -- 类方法 --
@classmethod
def is_empty_image(cls, resp):
return resp.status_code != 200 or len(resp.content) == 0
@staticmethod
def debug(topic: str, *args, sep='', end='\n', file=None, from_class=debug_from_class):
jm_debug(topic, *args, sep=sep, end=end, file=file, from_class=from_class)
@classmethod
def img_is_not_need_to_decode(cls, data_original: str, _resp):
return data_original.endswith('.gif')
# 爬取策略
class FetchStrategy:
def __init__(self,
from_index,
photo_len,
resp_getter,
resp_consumer,
):
self.from_index = from_index
self.photo_len = photo_len
self.resp_getter = resp_getter
self.resp_consumer = resp_consumer
def do_fetch(self):
raise NotImplementedError
def args(self):
return (self.from_index,
self.photo_len,
self.resp_getter,
self.resp_consumer,
)

76
src/jmcomic/jm_config.py Normal file
View File

@ -0,0 +1,76 @@
class JmModuleConfig:
# 网站相关
HTTP = "https://"
DOMAIN = "jmcomic1.group" # jmcomic默认域名
JM_REDIRECT_URL = f'{HTTP}jm365.xyz/3YeBdF' # 永久網域,怕走失的小伙伴收藏起来↓
JM_PUB_URL = f'{HTTP}jmcomic1.bet'
JM_CDN_IMAGE_URL_TEMPLATE = HTTP + 'cdn-msp.{domain}/media/photos/{photo_id}/{index:05}{suffix}' # index 从1开始
JM_SERVER_ERROR_HTML = "Could not connect to mysql! Please check your database settings!"
JM_IMAGE_SUFFIX = ['.jpg', '.webp', '.png', '.gif']
# 图片分隔相关
SCRAMBLE_0 = 220980
SCRAMBLE_10 = 268850
SCRAMBLE_NUM_8 = 421926 # 2023-02-08后改了图片切割算法
# 下载时的一些默认值
default_author = 'default-author'
default_photo_title = 'default-photo-title'
default_photo_id = 'default-photo-id'
# debug
enable_jm_debug = True
debug_printer = print
# 缓存
jm_client_caches = {}
@classmethod
def default_headers(cls):
return {
'authority': cls.DOMAIN,
'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',
'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',
}
# noinspection PyUnusedLocal
@classmethod
def jm_debug(cls, topic: str, *args, sep='', end='\n', file=None, from_class='api'):
if cls.enable_jm_debug is True:
cls.debug_printer(f'{topic}', *args, sep=sep, end=end, file=file)
@classmethod
def disable_jm_debug(cls):
cls.enable_jm_debug = False
@classmethod
def get_jmcomic_url(cls):
"""
访问禁漫的永久网域从而得到一个可用的禁漫网址
"""
from common import Postmans
domain = Postmans \
.get_impl_clazz('cffi') \
.create(headers=cls.default_headers()) \
.with_wrap_resp() \
.get(cls.JM_REDIRECT_URL, allow_redirects=False) \
.redirect_url
cls.jm_debug('获取禁漫地址', f'获取成功,最新可用的禁漫地址: {domain}')
return domain
jm_debug = JmModuleConfig.jm_debug
disable_jm_debug = JmModuleConfig.disable_jm_debug

319
src/jmcomic/jm_entity.py Normal file
View File

@ -0,0 +1,319 @@
from common import *
from .jm_config import *
class JmBaseEntity:
pass
class WorkEntity(JmBaseEntity, SaveableEntity, IterableEntity):
when_del_save_file = False
after_save_print_info = True
attr_char = '_'
cache_getitem_result = True
cache_field_name = '__cache_items_dict__'
detail_save_base_dir = workspace()
detail_save_file_suffix = '.yml'
def save_base_dir(self):
return self.detail_save_base_dir
def save_file_name(self) -> str:
return f"{self.get_id_prefix_of_filename()}{self.get_id()}{self.get_title()}{self.detail_save_file_suffix}"
def get_id_prefix_of_filename(self):
# "JmAlbumDetail" -> "album"
cls_name = self.__class__.__name__
id_prefix = cls_name[cls_name.index("m") + 1: cls_name.rfind("Detail")]
return f'{id_prefix}-'
def get_id(self) -> str:
raise NotImplementedError
def get_title(self) -> str:
raise NotImplementedError
def __len__(self):
raise NotImplementedError
def __getitem__(self, item):
raise NotImplementedError
class JmImageDetail(JmBaseEntity):
def __init__(self,
aid,
scramble_id,
img_url,
img_file_name,
img_file_suffix,
from_photo=None,
) -> None:
self.aid: str = aid
self.scramble_id: str = scramble_id
self.img_url: str = img_url
self.img_file_name: str = img_file_name
self.img_file_suffix: str = img_file_suffix
self.from_photo: Optional[JmPhotoDetail] = from_photo
@property
def filename(self):
return self.img_file_name + self.img_file_suffix
@classmethod
def of(cls,
photo_id: str,
scramble_id: str,
data_original: str,
from_photo=None,
) -> 'JmImageDetail':
"""
该方法用于创建 JmImageDetail 对象
"""
# /xxx.yyy
# ↑ ↑
# x y
x = data_original.rfind('/')
y = data_original.rfind('.')
return JmImageDetail(
aid=photo_id,
scramble_id=scramble_id,
img_url=data_original,
img_file_name=data_original[x + 1:y],
img_file_suffix=data_original[y:],
from_photo=from_photo,
)
class JmPhotoDetail(WorkEntity):
def __init__(self,
photo_id,
scramble_id,
title,
keywords,
series_id,
page_arr=None,
data_original_domain=None,
author=None,
from_album=None,
):
self.photo_id: str = photo_id
self.scramble_id: str = scramble_id
self.title: str = title
self._keywords: str = keywords
self._series_id: int = int(series_id)
self._author: StrNone = author
self.from_album: Optional[JmAlbumDetail] = from_album
# 下面的属性和图片url有关
if isinstance(page_arr, str):
import json
page_arr = json.loads(page_arr)
# 该photo的所有图片名 img_name
self.page_arr: List[str] = page_arr
self.data_original_domain = data_original_domain
def is_single_album(self) -> bool:
return self._series_id == 0
@property
def keyword_list(self) -> List[str]:
return self._keywords.split(',')
@property
def album_id(self) -> str:
return self.photo_id if self.is_single_album() else self._series_id
@property
def author(self) -> str:
# self._author 不为空字符串
if self._author is not None and self._author != '':
return self._author.strip()
# self._author 为空,先向上找
if self.from_album is not None:
return self.from_album.author
# 无向上元素,使用默认
return JmModuleConfig.default_author
def create_image_detail(self, index) -> JmImageDetail:
# 校验参数
length = len(self.page_arr)
if index >= length:
raise AssertionError(f'创建JmImageDetail失败{index} >= {length}')
data_original = self.get_img_data_original(self.page_arr[index])
return JmImageDetail.of(
self.photo_id,
self.scramble_id,
data_original,
from_photo=self
)
def get_img_data_original(self, img_name: str) -> str:
"""
根据图片名生成图片的完整请求路径 URL
例如img_name = 01111.webp
返回https://cdn-msp2.18comic.org/media/photos/147643/01111.webp
"""
return f'https://{self.data_original_domain}/media/photos/{self.photo_id}/{img_name}'
def __getitem__(self, item) -> JmImageDetail:
return self.create_image_detail(item)
def get_id(self):
return self.photo_id
def get_title(self):
return self.title
def __len__(self):
return len(self.page_arr)
class JmAlbumDetail(WorkEntity):
def __init__(self,
album_id,
scramble_id,
title,
episode_list,
page_count,
author_list,
pub_date,
):
self.album_id: str = album_id
self.scramble_id: str = scramble_id
self.title: str = title
self.page_count = int(page_count)
self._author_list: List[str] = author_list
self.pub_date = pub_date
# 有的 album 没有章节,则自成一章。
if len(episode_list) == 0:
# photo_id, photo_index_of_album, photo_title, photo_pub_date
episode_list = [(album_id, 0, title, pub_date)]
self.episode_list: List[Tuple] = self.distinct_episode(episode_list)
def create_photo_detail(self, index) -> Tuple[JmPhotoDetail, Tuple]:
# 校验参数
length = len(self.episode_list)
if index >= length:
raise AssertionError(f'创建JmPhotoDetail失败{index} >= {length}')
# episode_info: ('212214', '81', '94 突然打來', '2020-08-29')
episode_info: tuple = self.episode_list[index]
photo_id, photo_index_of_album, photo_title, photo_pub_date = episode_info
photo_detail = JmPhotoDetail(
photo_id=photo_id,
scramble_id=self.scramble_id,
title=photo_title,
keywords='',
series_id=self.album_id,
author=self.author,
from_album=self,
page_arr=None,
data_original_domain=None
)
return photo_detail, episode_info
@property
def author(self):
if len(self._author_list) >= 1:
return self._author_list[0]
return JmModuleConfig.default_author
def get_id(self):
return self.album_id
def get_title(self):
return self.title
def __len__(self):
return len(self.episode_list)
def __getitem__(self, item) -> JmPhotoDetail:
return self.create_photo_detail(item)[0]
@staticmethod
def distinct_episode(episode_list):
ret = []
def not_exist(episode):
photo_id = episode[0]
for each in ret:
if each[0] == photo_id:
return False
return True
for episode in episode_list:
if not_exist(episode):
ret.append(episode)
return ret
class JmSearchPage(IterableEntity):
def __init__(self, album_info_list) -> None:
# (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]
def __len__(self):
return len(self.data)
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,
)

480
src/jmcomic/jm_option.py Normal file
View File

@ -0,0 +1,480 @@
from .jm_service import *
class JmOptionAdvice:
def decide_image_save_dir(self,
option: 'JmOption',
album_detail: Optional[JmAlbumDetail],
photo_detail: JmPhotoDetail,
) -> StrNone:
"""
决定一个本子图片的下载文件夹
@param option: JmOption对象
@param album_detail: 本子集实体类
@param photo_detail: 本子章节实体类
@return: 下载文件夹为空表示不处理
"""
pass
def decide_image_filepath(self,
option: 'JmOption',
photo_detail: JmPhotoDetail,
index: int,
) -> StrNone:
"""
决定一个本子图片的绝对路径
@param option: JmOption对象
@param photo_detail: 本子章节实体类
@param index: 本子章节里的第几章图片
@return: 下载绝对路径为空表示不处理
"""
pass
def decide_image_suffix(self,
option: 'JmOption',
img_detail: JmImageDetail,
) -> StrNone:
"""
决定一个图片的保存后缀名
@param option: JmOption对象
@param img_detail: 禁漫图片实体类
@return: 保存后缀名为空表示不处理
"""
pass
class JmAdviceRegistry:
advice_registration: Dict[Any, List[JmOptionAdvice]] = {}
@classmethod
def register_advice(cls, base, *advice):
advice_ls = cls.advice_registration.get(base, None)
if advice_ls is None:
advice_ls = list(advice)
cls.advice_registration[base] = advice_ls
else:
for e in advice:
advice_ls.append(e)
return advice
@classmethod
def get_advice(cls, base) -> List[JmOptionAdvice]:
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
AdditionalHandler = Callable[[Optional[JmAlbumDetail], JmPhotoDetail], str]
additional_tree_flag_handler_mapping: Dict[int, AdditionalHandler] = {}
def __init__(self,
Bd: str,
flag: Union[str, int],
):
self.Bd = fix_filepath(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)
if album_dir is None:
return f'{self.Bd}{photo_dir}/'
album_dir = fix_windir_name(album_dir)
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))
else:
if flag in self.additional_tree_flag_handler_mapping:
return self.additional_tree_flag_handler_mapping[flag](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.client_config = client_config
self.client_config['postman_type_list'] = list(Postmans.postman_impl_class_dict.keys())
# 下载配置
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.filepath = filepath
self.disable_jm_module_debug = disable_jm_module_debug
if disable_jm_module_debug:
disable_jm_debug()
"""
下面是决定图片保存路径的方法
"""
def decide_image_save_dir(self, album_detail, photo_detail) -> str:
if album_detail is None and photo_detail is None:
raise AssertionError('album和photo不能同时为None')
# 先检查advice的回调如果回调支持则优先使用回调
for advice in JmAdviceRegistry.get_advice(self):
save_dir = advice.decide_image_save_dir(self, album_detail, photo_detail)
if save_dir is not None:
return save_dir
# 使用 self.dir_tree 决定 save_dir
save_dir = self.dir_tree.deside_image_save_dir(
album_detail or photo_detail.from_album,
photo_detail
)
mkdir_if_not_exists(save_dir)
return save_dir
def decide_image_suffix(self, img_detail: JmImageDetail):
# 先检查advice的回调如果回调支持则优先使用回调
for advice in JmAdviceRegistry.get_advice(self):
suffix = advice.decide_image_suffix(self, img_detail)
if suffix is not None:
return suffix
# 动图则使用原后缀
suffix = img_detail.img_file_suffix
if suffix.endswith("gif"):
return suffix
# 非动图,以配置为先
return self.download_convert_image_suffix or suffix
def decide_image_filepath(self, photo_detail: JmPhotoDetail, index: int) -> str:
# 先检查advice的回调如果回调支持则优先使用回调
for advice in JmAdviceRegistry.get_advice(self):
filepath = advice.decide_image_filepath(self, photo_detail, index)
if filepath is not None:
return filepath
# 通过拼接生成绝对路径
save_dir = self.decide_image_save_dir(photo_detail.from_album, photo_detail)
image: JmImageDetail = photo_detail[index]
suffix = self.decide_image_suffix(image)
return save_dir + image.img_file_name + suffix
"""
下面是对Advice的支持
"""
def register_advice(self, *advice):
JmAdviceRegistry.register_advice(self, *advice)
"""
下面是创建对象相关方法
"""
@classmethod
def default(cls) -> 'JmOption':
return JmOption(
DownloadDirTree(
workspace(),
DownloadDirTree.Bd_Title_Image,
),
cls.default_client_config(),
)
@classmethod
def create_from_file(cls, filepath: str) -> 'JmOption':
jm_option: JmOption = PackerUtil.unpack(filepath, JmOption)[0]
jm_option.filepath = filepath
return jm_option
@classmethod
def default_client_config(cls):
return {
'domain': JmModuleConfig.DOMAIN,
'meta_data': {
'cookies': None,
'headers': JmModuleConfig.default_headers(),
'allow_redirects': True,
},
'postman_type_list': [
'requests',
'requests_Session',
'cffi',
'cffi_Session',
]
}
def save_base_dir(self):
return of_dir_path(self.filepath)
def save_file_name(self) -> str:
return of_file_name(self.filepath)
def save_to_file(self, filepath=None):
if filepath is None:
filepath = self.filepath
if filepath is None:
raise AssertionError("未指定JmOption的保存路径")
super().save_to_file(filepath)
"""
下面是 build 方法
"""
def build_jm_client(self) -> JmcomicClient:
if self.cache_jm_client is not True:
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)
return client
def new_jm_client(self) -> JmcomicClient:
meta_data = self.client_config['meta_data']
# 处理代理
def handle_proxies(key='proxies'):
proxies = meta_data.get(key, None)
# 无代理,或代理已配置好好的
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
# 处理 headers
def handle_headers(key='headers'):
headers = meta_data.get(key, None)
if headers is None or (not isinstance(headers, dict)) or len(headers) == 0:
meta_data[key] = JmModuleConfig.default_headers()
# 处理【特殊配置项】
handle_proxies()
handle_headers()
# 决定Postman的实现类根据配置项 client_config.postman
postman_clazz = Postmans.get_impl_clazz(self.client_config.get('postman_type', 'cffi'))
# 决定域名
domain = self.client_config.get('domain', JmModuleConfig.DOMAIN)
jm_debug('创建JmcomicClient', f'使用域名: {domain}使用Postman实现: {postman_clazz}')
# 创建 JmcomicClient 实例
client = JmcomicClient(
postman=postman_clazz(meta_data),
domain=domain,
retry_times=self.client_config.get('retry_times', None)
)
return client
"""
下面的方法是对CdnOption的支持已弃用
"""
def build_cdn_option(self, use_multi_thread_strategy=True):
return CdnConfig.create(
cdn_domain=self.client_config.get('domain', JmModuleConfig.DOMAIN),
fetch_strategy=MultiThreadFetch if use_multi_thread_strategy else InOrderFetch,
cdn_image_suffix=None,
use_cache=self.download_use_disk_cache,
decode_image=self.download_image_then_decode
)
def build_cdn_crawler(self, use_multi_thread_strategy=True):
return CdnFetchService(self.build_cdn_option(use_multi_thread_strategy),
self.build_jm_client())
def build_cdn_request(self,
photo_id: str,
scramble_id=str(JmModuleConfig.SCRAMBLE_10),
from_index=1,
photo_len=None,
use_default=False
) -> CdnRequest:
return CdnRequest.create(
photo_id,
self.build_save_path_provider(use_default, None if use_default else photo_id),
scramble_id,
from_index,
photo_len,
)
def build_save_path_provider(self,
use_all_default_save_path,
photo_id=None,
) -> CdnRequest.SavePathProvider:
if use_all_default_save_path is True:
# 不通过请求获取 photo 的信息,相当于使用【空本子】和【空集】
photo_detail, album_detail = None, None
else:
# 通过请求获得 photo 的本子信息
client = self.build_jm_client()
photo_detail = client.get_photo_detail(photo_id)
album_detail = client.fill_from_album(photo_detail)
def save_path_provider(url, _suffix: str, _index, _is_decode):
return '{0}{1}{2}'.format(self.decide_image_save_dir(album_detail, photo_detail),
of_file_name(url, trim_suffix=True),
self.download_convert_image_suffix or _suffix)
return save_path_provider
def _register_yaml_constructor():
from yaml import add_constructor, Loader, Node
tag_mapping = {
'tag:yaml.org,2002:python/object:jmcomic.jm_option.JmOption': JmOption,
'tag:yaml.org,2002:python/object:jmcomic.jm_option.DownloadDirTree': DownloadDirTree,
}
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()

248
src/jmcomic/jm_service.py Normal file
View File

@ -0,0 +1,248 @@
from .jm_client import *
class CdnConfig:
def __init__(self,
cdn_domain: str,
cdn_image_suffix: str,
fetch_strategy: Type[FetchStrategy],
use_cache,
decode_image,
):
self.cdn_domain = cdn_domain
self.cdn_image_suffix = cdn_image_suffix
self.fetch_strategy = fetch_strategy
self.use_cache = use_cache
self.decode_image = decode_image
def get_cdn_image_url(self, photo_id: str, index: int) -> str:
return JmModuleConfig.JM_CDN_IMAGE_URL_TEMPLATE.format(
domain=self.cdn_domain,
photo_id=photo_id,
index=index,
suffix=self.cdn_image_suffix
)
def check_request_is_valid(self, req: CdnRequest):
from common import is_function
if req.photo_len is not None and not isinstance(req.photo_len, int):
raise AssertionError('传参错误photo_len要么给整数要么给None')
if not is_function(req.save_path_provider):
raise AssertionError('传参错误save_path_provider应该是函数')
if self.decode_image is True and req.scramble_id is None:
raise AssertionError('传参缺失指定decode_image=True就必须提供scramble_id')
if req.from_index <= 0:
raise AssertionError(f'传参错误from_index必须大于1: {req.from_index} < 1')
@classmethod
def create(cls,
cdn_domain,
fetch_strategy,
cdn_image_suffix=None,
use_cache=False,
decode_image=True,
):
return CdnConfig(
cdn_domain,
cdn_image_suffix or '.webp',
fetch_strategy,
use_cache,
decode_image,
)
class InOrderFetch(FetchStrategy):
def do_fetch(self):
index, photo_len, resp_getter, resp_consumer = self.args()
while True:
if photo_len is not None and index > photo_len:
break
resp_info = resp_getter(index)
if resp_info[0] is None:
break
resp_consumer(resp_info, index)
index += 1
class MultiThreadFetch(FetchStrategy):
fetch_batch_count = 5
def do_fetch(self):
from common import multi_thread_launcher
begin, photo_len, resp_getter, resp_consumer = self.args()
if photo_len is not None:
# 确定要爬多少页
def accpet_index(index):
resp_info = resp_getter(index)
if resp_info[0] is None:
return
resp_consumer(resp_info, index)
multi_thread_launcher(
iter_objs=range(begin, photo_len + 1),
apply_each_obj_func=accpet_index,
wait_finish=True,
)
else:
# 不确定要爬多少页
# 批量单位
batch_count = self.fetch_batch_count
# 标记位
fetch_suffer_none_resp = False
def accpet_index(index):
resp_info = resp_getter(index)
if resp_info is None:
nonlocal fetch_suffer_none_resp
fetch_suffer_none_resp = True
return
resp_consumer(resp_info, index)
while True:
# 5次5次爬每5次中但凡有1次出现了失败就停止
multi_thread_launcher(
iter_objs=range(begin, begin + batch_count),
apply_each_obj_func=accpet_index,
wait_finish=True
)
# 检查标记
if fetch_suffer_none_resp is False:
begin += batch_count
else:
break
class CdnFetchService:
def __init__(self,
config: CdnConfig,
client: 'JmcomicClient',
):
self.client = client
self.config = config
self.debug = self.client.debug
def download_photo_from_cdn_directly(self,
req: CdnRequest,
):
# 校验参数
self.config.check_request_is_valid(req)
# 基本信息
photo_id = req.photo_id
scramble_id = req.scramble_id
use_cache = self.config.use_cache
decode_image = self.config.decode_image
# 获得响应
def get_resp(index: int) -> Tuple[Optional[Any], str, str]:
url = self.config.get_cdn_image_url(photo_id, index)
suffix = self.config.cdn_image_suffix
pre_try_save_path = req.save_path_provider(url, suffix, index, decode_image)
# 判断是否命中缓存(预先尝试)
if use_cache is True and file_exists(pre_try_save_path):
# 命中,返回特殊值
return None, suffix, pre_try_save_path
else:
return self.try_get_cdn_image_resp(
self.client,
url,
suffix,
photo_id,
index,
)
# 保存响应
def save_resp(resp_info: Tuple[Optional[Any], str, str], index: int):
resp, suffix, img_url = resp_info
# 1. 判断是不是特殊值
if resp_info[0] is None:
# 是,表示命中缓存,直接返回
save_path = resp_info[2]
self.debug('图片下载成功',
f'photo-{photo_id}: {index}{suffix}命中磁盘缓存 ({save_path})')
return
# 2. 判断是否命中缓存
save_path = req.save_path_provider(img_url, suffix, index, decode_image)
if use_cache is True and file_exists(save_path):
# 命中,直接返回
self.debug('图片下载成功',
f'photo-{photo_id}: {index}{suffix}命中磁盘缓存 ({save_path})')
return
# 3. 保存图片
if decode_image is False:
JmImageSupport.save_resp_img(resp, save_path)
else:
JmImageSupport.save_resp_decoded_img(
resp,
JmImageDetail.of(photo_id, scramble_id, data_original=img_url),
save_path,
)
# 4. debug 消息
self.debug('图片下载成功',
f'photo-{photo_id}: {index}{suffix}下载完成 ('
+ ('已解码' if decode_image is True else '未解码') +
f') [{img_url}] → [{save_path}]')
# 调用爬虫策略
self.config.fetch_strategy(
req.from_index,
req.photo_len,
resp_getter=get_resp,
resp_consumer=save_resp,
).do_fetch()
# 准备提供给爬虫策略的函数
@staticmethod
def try_get_cdn_image_resp(client: JmcomicClient,
url: str,
suffix: str,
photo_id,
index,
) -> Optional[Tuple[Any, str, str]]:
resp = client.jm_get(url, False, False)
# 第一次校验,不空则直接返回
if not client.is_empty_image(resp):
return resp, suffix, url
# 下面进行重试
client.debug(
'图片获取重试',
f'photo-{photo_id},图片序号为{index}url=[{url}]'
)
# 重试点1是否文件后缀名不对
for alter_suffix in JmModuleConfig.JM_IMAGE_SUFFIX:
url = change_file_suffix(url, alter_suffix)
resp = client.jm_get(url, False, False)
if not client.is_empty_image(resp):
client.debug(
'图片获取重试 → 成功√',
f'更改请求后缀({suffix} -> {alter_suffix}url={url}'
)
return resp, alter_suffix, url
# 结论:可能是图片到头了
client.debug(
'图片获取重试 ← 失败×',
f'更换后缀名不成功,停止爬取。'
f'(推断本子长度<={index - 1}。当前图片序号为{index},已经到达尽头,'
f'photo-{photo_id})'
)

281
src/jmcomic/jm_toolkit.py Normal file
View File

@ -0,0 +1,281 @@
from PIL import Image
from .jm_entity import *
class JmcomicText:
pattern_jm_domain = compile('https://([\w.]+)')
pattern_jm_pa_id = compile('/(photos?|album)/(\d+)')
pattern_html_photo_photo_id = compile('<meta property="og:url" content=".*?/photo/(\d+)/?.*?">')
pattern_html_photo_scramble_id = compile('var scramble_id = (\d+);')
pattern_html_photo_title = compile('<title>(.*?)\|.*</title>')
# pattern_html_photo_data_original_list = compile('data-original="(.*?)" id="album_photo_.+?"')
pattern_html_photo_data_original_domain = compile('src="https://(.*?)/media/albums/blank')
pattern_html_photo_keywords = compile('<meta name="keywords" content="(.*?)" />')
pattern_html_photo_series_id = compile('var series_id = (\d+);')
pattern_html_photo_page_arr = compile('var page_arr = (.*?);')
pattern_html_album_album_id = compile('<span class="number">.*?JM(\d+)</span>')
pattern_html_album_scramble_id = compile('var scramble_id = (\d+);')
pattern_html_album_title = compile('panel-heading[\s\S]*?<h1>(.*?)</h1>')
pattern_html_album_episode_list = compile('<a href=".*?" data-album="(\d+)">\n'
'<li.*>\n第(\d+)話\n(.*)\n'
'<[\s\S]*?>(\d+-\d+-\d+).*?')
pattern_html_album_page_count = compile('<span class="pagecount">.*?:(\d+)</span>')
pattern_html_album_pub_date = compile('itemprop="datePublished" content=".*?">更新日期 : (.*?)</span>')
# album 作者
pattern_html_album_author_list = [
compile('作者: <span itemprop="author" data-type="author">(\s*<a.*?>(.*?)</a>)*\s*</span>'),
compile("<a.*?>(.*?)</a>"),
]
@classmethod
def parse_to_jm_domain(cls, text: str):
if text.startswith("https"):
return cls.pattern_jm_domain.search(text)[1]
return text
@classmethod
def parse_to_photo_id(cls, text) -> str:
if isinstance(text, int):
return str(text)
if not isinstance(text, str):
raise AssertionError(f"无法解析jm车号, 参数类型为: {type(text)}")
if len(text) <= 2:
raise AssertionError(f"无法解析jm车号, 文本为: {text}")
# text: JM12341
c0 = text[0]
c1 = text[1]
if (c0 == 'J' or c0 == 'j') and (c1 == 'M' or c1 == 'm'):
# JM123456
return text[2:]
elif text.isdigit():
# 123456
return str(text)
else:
# https://xxx/photo/412038
match = cls.pattern_jm_pa_id.search(text)
if match is None:
raise AssertionError(f"无法解析jm车号, 文本为: {text}")
return match[2]
@classmethod
def parse_to_album_id(cls, text) -> str:
return cls.parse_to_photo_id(text)
@classmethod
def analyse_jm_photo_html(cls, html: str) -> JmPhotoDetail:
return cls.reflect_new_instance(
html,
"pattern_html_photo_",
JmPhotoDetail
)
@classmethod
def analyse_jm_album_html(cls, html: str) -> JmAlbumDetail:
return cls.reflect_new_instance(
html,
"pattern_html_album_",
JmAlbumDetail
)
@classmethod
def reflect_new_instance(cls, html: str, cls_field_prefix: str, clazz: type):
def match_field(field_key: str, pattern: Union[Pattern, List[Pattern]], text):
if isinstance(pattern, list):
# 如果是 pattern 是 List[re.Pattern]
# 取最后一个 pattern 用于 match field
# 其他的 pattern 用来给文本缩小范围(相当于多次正则匹配)
last_pattern = pattern[len(pattern) - 1]
# 缩小文本
for i in range(0, len(pattern) - 1):
match = pattern[i].search(text)
if match is None:
return None
text = match[0]
pattern = last_pattern
if field_key.endswith("_list"):
return pattern.findall(text)
else:
match = pattern.search(text)
if match is not None:
return match[1]
return None
field_dict = {}
pattern_name: str
for pattern_name, pattern_value in cls.__dict__.items():
if not pattern_name.startswith(cls_field_prefix):
continue
# 获取字段名和值
field_name = pattern_name[pattern_name.index(cls_field_prefix) + len(cls_field_prefix):]
field_value = match_field(field_name, pattern_value, html)
if field_value is None:
raise AssertionError(f"文本没有匹配上字段:字段名为{field_name}pattern为cls.{pattern_name}")
# 保存字段
field_dict[field_name] = field_value
return clazz(**field_dict)
class JmSearchSupport:
# 用来缩减html的长度
pattern_html_search_shorten_for = compile('<div class="well well-sm">([\s\S]*)'
'<div class="row">[\s\S]*'
'<div class="bot-per visible-xs visible-sm">')
# 用来提取搜索页面的的album的信息
pattern_html_search_album_info_list = compile(
'<a href="/album/(\d+)/.+"[\s\S]*?'
'title="(.*?)"[\s\S]*?'
'(<div class="label-category" style="">'
'\n(.*)\n</div>\n<div class="label-sub" style=" ">'
'(.*?)\n<[\s\S]*?)?'
'<div class="title-truncate tags .*>\n'
'(<a[\s\S]*?) </div>'
)
# 用来查找tag列表
pattern_html_search_tag_list = compile('<a href=".*?">(.*?)</a>')
@classmethod
def analyse_jm_search_html(cls, html: str) -> JmSearchPage:
html = cls.pattern_html_search_shorten_for.search(html)[0]
album_info_list = cls.pattern_html_search_album_info_list.findall(html)
for i, (album_id, title, *args) in enumerate(album_info_list):
_, category_none, label_sub_none, tag_text = args
tag_list = cls.pattern_html_search_tag_list.findall(tag_text)
album_info_list[i] = (album_id, title, category_none, label_sub_none, tag_list)
return JmSearchPage(album_info_list)
class JmImageSupport:
@classmethod
def save_resp_img(cls, resp: Any, filepath: str, need_convert=True):
"""
保存图片的响应数据
@param resp: Response对象
@param filepath: 响应数据保存的绝对路径
@param need_convert: True 使用PIL打开图片再保存; False 直接保存resp.content;
如果需要改变图片的格式比如 .jpg .png则需要neet_convert=True
如果不需要改变文件的格式使用need_convert=False可以跳过PIL解析图片效率更高
"""
if need_convert is True:
cls.open_Image(resp.content).save(filepath)
else:
save_resp_content(resp, filepath)
@classmethod
def save_resp_decoded_img(cls,
resp: Any,
img_detail: JmImageDetail,
filepath: str
) -> None:
cls.decode_and_save(
cls.calculate_segmentation_num(img_detail),
cls.open_Image(resp.content),
filepath
)
@classmethod
def decode_disk_img(cls,
img_detail: JmImageDetail,
img_filepath: str,
decoded_save_path: str
) -> None:
cls.decode_and_save(
cls.calculate_segmentation_num(img_detail),
cls.open_Image(img_filepath),
decoded_save_path
)
@classmethod
def decode_and_save(cls,
num: int,
img_src: Image,
decoded_save_path: str
) -> None:
"""
解密图片并保存
@param num: 分割数可以用 cls.calculate_segmentation_num 计算
@param img_src: 原始图片
@param decoded_save_path: 解密图片的保存路径
"""
# 无需解密,直接保存
if num == 0:
img_src.save(decoded_save_path)
return
import math
w, h = img_src.size
# 创建新的解密图片
img_decode = Image.new("RGB", (w, h))
remainder = h % num
copyW = w
for i in range(num):
copyH = math.floor(h / num)
py = copyH * i
y = h - (copyH * (i + 1)) - remainder
if i == 0:
copyH += remainder
else:
py += remainder
img_decode.paste(
img_src.crop((0, y, copyW, y + copyH)),
(0, py, copyW, py + copyH)
)
# 保存到新的解密文件
img_decode.save(decoded_save_path)
@classmethod
def open_Image(cls, fp: Union[str, bytes]):
from PIL import Image
from io import BytesIO
fp = fp if isinstance(fp, str) else BytesIO(fp)
return Image.open(fp)
@classmethod
def calculate_segmentation_num(cls, detail: JmImageDetail) -> int:
"""
获得图片分割数
"""
scramble_id = int(detail.scramble_id)
aid = int(detail.aid)
if aid < scramble_id:
return 0
elif aid < JmModuleConfig.SCRAMBLE_10:
return 10
else:
import hashlib
x = 10 if aid < JmModuleConfig.SCRAMBLE_NUM_8 else 8
s = f"{aid}{detail.img_file_name}" # 拼接
s = s.encode()
s = hashlib.md5(s).hexdigest()
num = ord(s[-1])
num %= x
num = num * 2 + 2
return num

View File

@ -0,0 +1,40 @@
"""
--------------------
API快速上手
--------------------
"""
import jmcomic # 导入此模块,需要先安装.
jmcomic.download_album('422866') # 传入要下载的album的id即可下载整个album到本地.
# 上面的这行代码还有一个可选参数option: JmOption表示配置项
# 配置项的作用是告诉程序下载时候的一些选择,
# 比如,要下载到哪个文件夹,使用怎样的路径组织方式(比如[/作者/本子id/图片] 或者 [/作者/本子名称/图片].
# 如果没有配置,则会使用 JmOption.default(),下载的路径是[当前工作文件夹/本子名称/图片].
"""
--------------------
批量下载介绍
--------------------
"""
# 如果你想要批量下载,可以使用 list/set/tuple/生成器 作为第一个参数。
# 第二个参数依然是可选的JmOption对象
jmcomic.download_album(['422866', '1', '2', '3']) # list
jmcomic.download_album({'422866', '1', '2', '3'}) # set
jmcomic.download_album(('422866', '1', '2', '3')) # tuple
jmcomic.download_album(aid for aid in ('422866', '1', '2', '3')) # 生成器
"""
--------------------
配置文件介绍
--------------------
"""
# 先获取默认的JmOption对象
jm_option = jmcomic.JmOption.default()
# 可以把对象保存为yml文件方便日后修改。保存格式会根据后缀名智能转换。
jm_option.save_to_file('./禁漫下载默认配置.yml') # yml格式推荐
jm_option.save_to_file('./禁漫下载默认配置.json') # json格式
# 如果你修改了默认配置,现在想用你修改后的配置来下载,使用如下代码
jm_option = jmcomic.create_option('./禁漫下载默认配置.yml')
jmcomic.download_album('23333', jm_option)

52
usage/jmcomic_usage.py Normal file
View File

@ -0,0 +1,52 @@
from jmcomic import *
jm_option = create_option(
f'你的配置文件路径,例如: D:/a/b/c/jmcomic/config.yml'
)
@timeit('下载本子集: ')
def download_jm_album():
ls = str_to_list('''
JM435413
''')
download_album(ls, jm_option) # 效果同下面的代码
# download_album_batch(ls, jm_option)
@timeit('获取实体类: ')
def get_album_photo_detail():
client = jm_option.build_jm_client()
album: JmAlbumDetail = client.get_album_detail('427413')
def show(p):
p: JmPhotoDetail = client.get_photo_detail(p.photo_id)
for img in p:
img: JmImageDetail
print(img.img_url)
multi_thread_launcher(
iter_objs=album,
apply_each_obj_func=show
)
@timeit('搜索本子: ')
def search_jm_album():
client = jm_option.build_jm_client()
search_album: JmSearchPage = client.search_album(search_query='+MANA +无修正')
for album_id, title, *_args in search_album:
print(f'[{album_id}]{title}')
def main():
search_jm_album()
download_jm_album()
get_album_photo_detail()
if __name__ == '__main__':
main()

View File

@ -0,0 +1,17 @@
from jmcomic import *
def get_option():
option = create_option('../assets/config/workflow_option.yml')
return option
# 下方填入你要下载的本子的id一行一个。
# 每行的首尾可以有空白字符
jm_albums = str_to_list('''
422866
''')
# 调用jmcomic的download_album方法下载漫画
download_album(jm_albums, option=get_option())