好友
阅读权限10
听众
最后登录1970-1-1
|
import requests
import os
import img2pdf
import re
import shutil
import logging
from typing import Dict, List
from concurrent.futures import ThreadPoolExecutor
from tqdm import tqdm # 导入tqdm库
# 配置日志
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
def get_journal_info(journal_id: str) -> Dict:
"""获取杂志详细信息"""
url = "https://api.bookan.com.cn/resource/issueInfoList"
params = {
"instanceId": 12696,
"resourceType": 3,
"issueIds": journal_id,
"isDetail": 1
}
response = requests.get(url, params=params)
if response.status_code != 200 or response.json().get("code") != 0:
raise Exception(f"获取杂志信息失败: {journal_id}")
data = response.json()["data"][0]
# 确保返回的数据中包含 webp 字段
if "webp" not in data:
raise ValueError(f"API 返回的数据中缺少 webp 字段: {journal_id}")
return data
def get_image_hashes(resource_id: str, issue_id: str, page_count: int) -> List[Dict]:
"""获取图片hash列表"""
url = "https://api.bookan.com.cn/resource/getHash"
params = {
"resourceType": 3,
"resourceId": resource_id,
"issueId": issue_id,
"start": 1,
"end": page_count
}
response = requests.get(url, params=params)
if response.status_code != 200 or response.json().get("code") != 0:
raise Exception(f"获取图片hash失败: {resource_id}-{issue_id}")
data = response.json()["data"]
if not data:
raise ValueError(f"未获取到任何图片hash信息: {resource_id}-{issue_id}")
return data
def download_image(item, resource_id, issue_id, output_dir, webp):
"""下载单张图片"""
# 使用 webp 动态构造 img_url
img_url = f"http://img1-qn.bookan.com.cn/jpage{webp}/{resource_id}/{resource_id}-{issue_id}/{item['hash']}_big.jpg"
response = requests.get(img_url)
if response.status_code == 200:
img_path = os.path.join(output_dir, f"page_{item['page']}.jpg")
with open(img_path, "wb") as f:
f.write(response.content)
# logging.info(f"下载成功: {output_dir} 第{item['page']}页") #, URL: {img_url}"
else:
logging.error(f"下载失败: {output_dir} 第{item['page']}页, URL: {img_url}")
def download_images(resource_id: str, issue_id: str, hashes: List[Dict], output_dir: str, webp: str):
"""多线程下载所有图片"""
if not os.path.exists(output_dir):
os.makedirs(output_dir)
total = len(hashes)
with ThreadPoolExecutor(max_workers=5) as executor:
# 使用tqdm显示进度条
with tqdm(total=total, desc="下载进度", unit="页") as pbar:
futures = [
executor.submit(download_image, item, resource_id, issue_id, output_dir, webp)
for item in hashes
]
for future in futures:
future.result()
pbar.update(1) # 更新进度条
def create_pdf(image_dir: str, output_pdf: str):
"""将图片合并为PDF"""
images = []
for img in os.listdir(image_dir):
if re.match(r"page_\d+\.jpg", img):
images.append(os.path.join(image_dir, img))
if not images:
raise ValueError(f"未找到任何图片文件，请检查下载过程: {image_dir}")
logging.info(f"找到 {len(images)} 张图片: {image_dir}")
images.sort(key=lambda x: int(re.search(r"page_(\d+)\.jpg", x).group(1)))
# 使用tqdm显示进度条
with tqdm(total=len(images), desc="生成PDF进度", unit="页") as pbar:
with open(output_pdf, "wb") as f:
f.write(img2pdf.convert(images))
pbar.update(len(images)) # 更新进度条
def process_journal(journal_id: str):
"""处理单本书籍的下载和PDF生成"""
image_dir = f"temp_images_{journal_id}"
try:
# 获取杂志元数据
journal_info = get_journal_info(journal_id)
resource_id = journal_info["resourceId"]
issue_id = journal_info["issueId"]
page_count = int(journal_info["count"])
webp = journal_info["webp"] # 获取 webp 值
# 获取图片hash列表
hashes = get_image_hashes(resource_id, issue_id, page_count)
logging.info(f"hashes for {journal_id}: {hashes}")
# 下载图片
logging.info(f"开始下载图片: {journal_id}")
download_images(resource_id, issue_id, hashes, image_dir, webp)
# 生成PDF文件
pdf_name = f"{journal_info['resourceName']}_{journal_info['issueName']}.pdf".replace("/", "_")
create_pdf(image_dir, pdf_name)
logging.info(f"PDF生成成功: {pdf_name}")
except Exception as e:
logging.error(f"处理过程中发生错误: {journal_id}, {str(e)}")
import traceback
traceback.print_exc()
finally:
# 清理临时文件
if os.path.exists(image_dir):
try:
shutil.rmtree(image_dir)
except Exception as e:
logging.error(f"无法删除目录 {image_dir}: {str(e)}")
def main():
# 输入多个期刊杂志ID，用逗号或空格分隔
journal_ids_input = input("请输入期刊杂志ID（多个ID用逗号或空格分隔）: ").strip()
if not journal_ids_input:
logging.error("期刊杂志ID不能为空")
return
# 分割输入为多个ID
journal_ids = re.split(r"[,\s]+", journal_ids_input)
# 处理每本书籍
for journal_id in journal_ids:
logging.info(f"开始处理期刊: {journal_id}")
process_journal(journal_id)
logging.info(f"期刊处理完成: {journal_id}")
if __name__ == "__main__":
main()你可以尝试用这个下载图书（杂志的需要修改类型，没做到两者合并），使用也很简单，找到需要的图书ID，如：310518779，输入到下面即可，也可以一次下载多本图书，中间使用英文逗号间隔即可，希望可以帮到你 |
|
[复制链接]
发表于 2025-3-14 21:17
|
发表于 2025-3-2 01:28
|楼主
分享到朋友圈
，不是什么大神，也是python初学者，另外代码有一点小改动，可以参考我帖子置顶评论。