cwx/NewComicDownloader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update20260815

Browse Source
This commit is contained in:
cwx
2026-08-15 22:35:15 +08:00
parent b124547f24
commit 01bc2522a3
11 changed files with 691 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2 -1
View File
@@ -5,4 +5,5 @@
CBZ/*
output/*
downloads/*
/**/__pycache__
/**/__pycache__
komga*.json
komga.py
+502
View File
@@ -0,0 +1,502 @@
import requests
from requests.auth import HTTPBasicAuth
import json
from typing import Dict, List, Optional, Set, Any
from tqdm import tqdm
import argparse
import os
from datetime import datetime
class KomgaChecker:
def __init__(self, base_url: str = None, username: str = None, password: str = None):
self.base_url = base_url.rstrip('/') if base_url else None
if base_url and username and password:
self.auth = HTTPBasicAuth(username, password)
self.session = requests.Session()
self.session.auth = self.auth
else:
self.session = None
def get_libraries(self) -> List[Dict]:
"""获取所有漫画库"""
if not self.session:
raise ValueError("未初始化Komga连接")
response = self.session.get(f"{self.base_url}/api/v1/libraries")
response.raise_for_status()
return response.json()
def get_series(self, library_id: str) -> List[Dict]:
"""获取指定库中的所有系列"""
if not self.session:
raise ValueError("未初始化Komga连接")
series = []
page = 0
size = 100
while True:
response = self.session.get(
f"{self.base_url}/api/v1/series",
params={"library_id": library_id, "page": page, "size": size}
)
response.raise_for_status()
data = response.json()
if not data["content"]:
break
series.extend(data["content"])
page += 1
return series
def get_books(self, series_id: str) -> List[Dict]:
"""获取指定系列中的所有书籍"""
if not self.session:
raise ValueError("未初始化Komga连接")
books = []
page = 0
size = 100
while True:
response = self.session.get(
f"{self.base_url}/api/v1/series/{series_id}/books",
params={"page": page, "size": size}
)
response.raise_for_status()
data = response.json()
if not data["content"]:
break
books.extend(data["content"])
page += 1
return books
def extract_volume_number(self, metadata: Dict) -> Optional[float]:
"""从元数据中提取卷号"""
# 尝试从volume字段获取
volume = metadata.get("number")
if volume is not None and volume != "":
try:
return float(str(volume))
except (ValueError, TypeError):
pass
# 尝试从标题中提取卷号
title = metadata.get("title", "").lower()
# 匹配 "vol. 1", "volume 1", "第1卷" 等格式
import re
patterns = [
r'vol\.?\s*(\d+(?:\.\d+)?)',
r'volume\s*(\d+(?:\.\d+)?)',
r'第\s*(\d+(?:\.\d+)?)\s*卷',
r'#\s*(\d+(?:\.\d+)?)'
]
for pattern in patterns:
match = re.search(pattern, title)
if match:
try:
return float(match.group(1))
except (ValueError, TypeError):
continue
return None
def check_missing_volumes(self, books: List[Dict]) -> Dict:
"""检查缺失的卷"""
volumes = []
for book in books:
metadata = book.get("metadata", {})
volume_num = self.extract_volume_number(metadata)
if volume_num is not None:
volumes.append({
'number': volume_num,
'title': metadata.get('title', 'Unknown'),
'id': book['id']
})
# 按卷号排序
volumes.sort(key=lambda x: x['number'])
if not volumes:
return {'has_volumes': False, 'volumes': [], 'missing': []}
# 找出缺失的卷号
volume_numbers = [v['number'] for v in volumes]
min_vol = int(min(volume_numbers))
max_vol = int(max(volume_numbers))
missing_volumes = []
for vol in range(min_vol, max_vol + 1):
if vol not in volume_numbers:
# 检查是否有接近的卷号（如1.0, 1.5等）
has_close_match = any(abs(v - vol) < 0.1 for v in volume_numbers)
if not has_close_match:
missing_volumes.append(vol)
return {
'has_volumes': True,
'volumes': volumes,
'missing': missing_volumes,
'total_found': len(volumes),
'expected_range': f"{min_vol}-{max_vol}"
}
def analyze_series(self, library_name: str, series: Dict) -> Optional[Dict]:
"""分析单个系列"""
try:
series_id = series['id']
series_name = series['metadata']['title']
print(f"正在检查系列: {series_name}")
books = self.get_books(series_id)
if not books:
return None
result = self.check_missing_volumes(books)
result.update({
'series_name': series_name,
'series_id': series_id,
'library_name': library_name,
'total_books': len(books)
})
return result
except Exception as e:
print(f"分析系列 {series.get('metadata', {}).get('title', 'Unknown')} 时出错: {e}")
return None
def run_check(self) -> List[Dict]:
"""运行完整的检查"""
if not self.session:
raise ValueError("未初始化Komga连接")
print("开始检查Komga服务器...")
try:
libraries = self.get_libraries()
print(f"找到 {len(libraries)} 个库")
except Exception as e:
print(f"获取库列表失败: {e}")
return []
all_results = []
for library in libraries:
library_id = library['id']
library_name = library['name']
print(f"\n正在处理库: {library_name}")
try:
series_list = self.get_series(library_id)
print(f"找到 {len(series_list)} 个系列")
for series in tqdm(series_list, desc=f"检查 {library_name}"):
result = self.analyze_series(library_name, series)
if result:
all_results.append(result)
except Exception as e:
print(f"处理库 {library_name} 时出错: {e}")
return all_results
def save_missing_to_json(self, results: List[Dict], filename: str = None):
"""将缺失卷的系列保存到JSON文件"""
if filename is None:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"komga_missing_volumes_{timestamp}.json"
# 筛选有缺失卷的系列
series_with_missing = [r for r in results if r['has_volumes'] and r['missing']]
if not series_with_missing:
print("没有找到缺失卷的系列，不生成JSON文件。")
return
# 准备保存的数据
output_data = {
'generated_at': datetime.now().isoformat(),
'komga_server': self.base_url,
'total_series_checked': len(results),
'series_with_missing_volumes': len(series_with_missing),
'missing_series': []
}
for result in series_with_missing:
series_data = {
'series_name': result['series_name'],
'series_id': result['series_id'],
'library_name': result['library_name'],
'total_books': result['total_books'],
'volumes_found': result['total_found'],
'expected_range': result['expected_range'],
'missing_volumes': result['missing'],
'existing_volumes': [
{
'number': vol['number'],
'title': vol['title'],
'book_id': vol['id']
} for vol in result['volumes']
]
}
output_data['missing_series'].append(series_data)
# 保存到文件
try:
with open(filename, 'w', encoding='utf-8') as f:
json.dump(output_data, f, ensure_ascii=False, indent=2)
print(f"\n✅ 缺失卷信息已保存到: {filename}")
except Exception as e:
print(f"❌ 保存JSON文件失败: {e}")
return filename
def load_from_json(self, filename: str) -> Dict[str, Any]:
"""从JSON文件加载检查结果"""
try:
with open(filename, 'r', encoding='utf-8') as f:
data = json.load(f)
print(f"✅ 已从 {filename} 加载检查结果")
return data
except Exception as e:
print(f"❌ 加载JSON文件失败: {e}")
return {}
def compare_with_previous(self, current_results: List[Dict], previous_filename: str):
"""与之前的检查结果进行比较"""
previous_data = self.load_from_json(previous_filename)
if not previous_data:
return
previous_missing = {s['series_name']: s['missing_volumes'] for s in previous_data.get('missing_series', [])}
current_missing = {r['series_name']: r['missing'] for r in current_results if r['has_volumes'] and r['missing']}
print("\n" + "="*80)
print("与之前检查结果比较")
print("="*80)
improved_series = []
worsened_series = []
unchanged_series = []
new_series_with_missing = []
fixed_series = []
for series_name, current_missing_vols in current_missing.items():
if series_name in previous_missing:
previous_missing_vols = previous_missing[series_name]
if set(current_missing_vols) == set(previous_missing_vols):
unchanged_series.append(series_name)
elif len(current_missing_vols) < len(previous_missing_vols):
improved_series.append({
'name': series_name,
'previous': previous_missing_vols,
'current': current_missing_vols,
'fixed': list(set(previous_missing_vols) - set(current_missing_vols))
})
else:
worsened_series.append({
'name': series_name,
'previous': previous_missing_vols,
'current': current_missing_vols,
'added': list(set(current_missing_vols) - set(previous_missing_vols))
})
else:
new_series_with_missing.append(series_name)
# 检查之前有缺失但现在完整的系列
for series_name in previous_missing:
if series_name not in current_missing:
fixed_series.append(series_name)
# 输出比较结果
if fixed_series:
print(f"\n🎉 已修复的系列 ({len(fixed_series)}):")
for series in fixed_series:
print(f" ✅ {series}")
if improved_series:
print(f"\n📈 改善的系列 ({len(improved_series)}):")
for series in improved_series:
print(f" 📚 {series['name']}")
print(f" 之前缺失: {series['previous']}")
print(f" 现在缺失: {series['current']}")
print(f" 已补全: {series['fixed']}")
if worsened_series:
print(f"\n📉 恶化的系列 ({len(worsened_series)}):")
for series in worsened_series:
print(f" 📚 {series['name']}")
print(f" 之前缺失: {series['previous']}")
print(f" 现在缺失: {series['current']}")
print(f" 新增缺失: {series['added']}")
if new_series_with_missing:
print(f"\n🆕 新发现的缺失系列 ({len(new_series_with_missing)}):")
for series in new_series_with_missing:
print(f" 📚 {series} - 缺失卷: {current_missing[series]}")
if unchanged_series:
print(f"\n➡️ 未变化的系列 ({len(unchanged_series)}):")
for i, series in enumerate(unchanged_series[:10]): # 只显示前10个
print(f" 📚 {series} - 缺失卷: {current_missing[series]}")
if len(unchanged_series) > 10:
print(f" ... 还有 {len(unchanged_series) - 10} 个系列")
# 总体统计
print(f"\n📊 总体比较:")
print(f" 已修复的系列: {len(fixed_series)}")
print(f" 改善的系列: {len(improved_series)}")
print(f" 恶化的系列: {len(worsened_series)}")
print(f" 新发现的缺失系列: {len(new_series_with_missing)}")
print(f" 未变化的系列: {len(unchanged_series)}")
def generate_report_from_json(self, filename: str):
"""从JSON文件生成报告"""
data = self.load_from_json(filename)
if not data:
return
print("\n" + "="*80)
print(f"Komga漫画缺集检查报告 (来自: {filename})")
print("="*80)
print(f"生成时间: {data.get('generated_at', '未知')}")
print(f"Komga服务器: {data.get('komga_server', '未知')}")
print(f"检查的系列总数: {data.get('total_series_checked', '未知')}")
print(f"有缺失卷的系列数: {data.get('series_with_missing_volumes', '未知')}")
missing_series = data.get('missing_series', [])
if missing_series:
print(f"\n{'='*50}")
print("有缺失卷的系列:")
print(f"{'='*50}")
for series in missing_series:
print(f"\n📚 系列: {series['series_name']}")
print(f" 库: {series['library_name']}")
print(f" 找到卷数: {series['volumes_found']}")
print(f" 预期范围: {series['expected_range']}")
print(f" ❌ 缺失卷号: {series['missing_volumes']}")
else:
print("\n🎉 没有发现缺失卷的系列!")
def generate_report(self, results: List[Dict], save_json: bool = True):
"""生成检查报告"""
print("\n" + "="*80)
print("Komga漫画缺集检查报告")
print("="*80)
series_with_missing = [r for r in results if r['has_volumes'] and r['missing']]
series_complete = [r for r in results if r['has_volumes'] and not r['missing']]
series_no_volumes = [r for r in results if not r['has_volumes']]
print(f"\n总计检查 {len(results)} 个系列")
print(f"🔍 有缺失卷的系列: {len(series_with_missing)}")
print(f"✅ 完整的系列: {len(series_complete)}")
print(f"❓ 无法识别卷号的系列: {len(series_no_volumes)}")
# 保存JSON文件
if save_json and series_with_missing:
json_filename = self.save_missing_to_json(results)
if series_with_missing:
print(f"\n{'='*50}")
print("有缺失卷的系列:")
print(f"{'='*50}")
for result in series_with_missing:
print(f"\n📚 系列: {result['series_name']}")
print(f" 库: {result['library_name']}")
print(f" 找到卷数: {result['total_found']}")
print(f" 预期范围: {result['expected_range']}")
print(f" ❌ 缺失卷号: {result['missing']}")
# 显示现有的卷号范围
volumes = result['volumes']
if len(volumes) <= 10:
existing = [v['number'] for v in volumes]
print(f" 📖 现有卷号: {existing}")
if series_no_volumes:
print(f"\n{'='*50}")
print("无法识别卷号的系列:")
print(f"{'='*50}")
for result in series_no_volumes[:10]: # 只显示前10个
print(f"📚 {result['series_name']} (库: {result['library_name']})")
if len(series_no_volumes) > 10:
print(f"... 还有 {len(series_no_volumes) - 10} 个系列")
# 返回JSON文件名用于后续比较
return json_filename if save_json and series_with_missing else None
def main():
parser = argparse.ArgumentParser(description='检查Komga服务器中的漫画缺集情况')
parser.add_argument('--url', help='Komga服务器地址 (例如: http://localhost:8080)')
parser.add_argument('--username', help='Komga用户名')
parser.add_argument('--password', help='Komga密码')
parser.add_argument('--output', '-o', help='输出JSON文件名 (可选)')
parser.add_argument('--no-json', action='store_true', help='不生成JSON文件')
parser.add_argument('--load-json', help='从JSON文件加载并显示报告')
parser.add_argument('--compare', help='与之前的JSON检查结果进行比较')
args = parser.parse_args()
# 创建检查器实例
if args.load_json:
# 仅加载模式，不需要Komga连接
checker = KomgaChecker()
checker.generate_report_from_json(args.load_json)
elif args.url and args.username and args.password:
# 正常检查模式
checker = KomgaChecker(args.url, args.username, args.password)
# 运行检查
results = checker.run_check()
# 生成报告
json_filename = checker.generate_report(results, save_json=not args.no_json)
# 如果指定了输出文件名，额外保存一份
if args.output and results:
checker.save_missing_to_json(results, args.output)
json_filename = args.output
# 如果指定了比较文件，进行比较
if args.compare and json_filename:
checker.compare_with_previous(results, args.compare)
else:
print("错误: 需要提供Komga服务器连接信息或指定--load-json参数")
parser.print_help()
if __name__ == "__main__":
#main()
# 直接在你的代码中使用
checker = KomgaChecker(
base_url="https://komga.caiwenxiu.cn",
username="caiwenxiu0806@163.com",
password="cwx@komga"
)
# 运行检查
results = checker.run_check()
# 生成报告
json_filename = checker.generate_report(results, save_json=True)
# 如果指定了输出文件名，额外保存一份
output="komga.json"
if output and results:
checker.save_missing_to_json(results, output)
json_filename = output
run.py
+4 -4
View File
@@ -6,12 +6,12 @@ logger = setup_logging(__name__)
async def main():
# 配置下载参数
# manga_url = "https://rouman5.com/books/4355ff5c-98d7-45c0-95a9-b4023b8eb812"
manga_list_url = "https://rouman5.com/books?continued=undefined"
#manga_url = "https://rouman5.com/books/5f70fd06-5554-4ac5-b3a8-5b82749a4a7a"
manga_list_url = "https://rouman5.com/books?continued=true"
# 开始下载
# await MangaManager().download_manga(manga_url)
for i in range(0,80):
#await MangaManager().download_manga(manga_url)
for i in range(0,30):
await MangaManager().download_list_manga(f"{manga_list_url}&page={i}")
if __name__ == "__main__":
src/common/extractor.py
+4 -1
View File
@@ -20,7 +20,10 @@ class SelectorProcessor:
elif len_elements == 1:
return elements[0]
elif len_elements > 1 and index > -1:
return elements[index]
if index > len_elements:
return ""
else:
return elements[index]
else:
return elements
except Exception as e:
src/common/item.py
+2
View File
@@ -146,6 +146,8 @@ class MangaInfo(BaseModel):
cover_info = {}
if isinstance(v, str) and not v.startswith('http'):
cover_info['url'] = HttpUrl(cls.base_url + v)
else:
cover_info['url'] = HttpUrl(v)
return CoverItem(**cover_info)
tags: str = []
src/common/naming.py
+66 -10
View File
@@ -1,12 +1,13 @@
from pathlib import Path
from datetime import datetime
from typing import Callable
import base64,hashlib,os,re
import base64,hashlib,os,re,shutil
from src.config import BASE_IMAGES_DIR,CBZ_DIR,OLD_CBZ_DIR
from src.common.item import MangaInfo,MangaItem
from typing import Generator, Union, List, Optional
from datetime import datetime
from opencc import OpenCC
from pypinyin import pinyin, Style
PREFIX_SCRAMBLE = "scramble="
@@ -16,23 +17,30 @@ class DirectoryNaming:
"""确保目录存在"""
directory.mkdir(parents=True, exist_ok=True)
@classmethod
def date_path(cls) -> Path:
"""按日期组织的路径生成器"""
today = datetime.now()
return os.path.join(str(today.year),f"{today.month:02d}",f"{today.day:02d}")
@classmethod
def chapter_images_dir(cls, manga_info: MangaInfo, chapter: str, filename: str = None) -> Path:
"""生成章节目录"""
if filename:
return Path(BASE_IMAGES_DIR,f"{manga_info.project}","images",f"{manga_info.title}",chapter.title, filename)
return Path(BASE_IMAGES_DIR,f"{manga_info.project}","images",f"{cls.date_path()}",f"{manga_info.title}",chapter.title, filename)
else:
return Path(BASE_IMAGES_DIR,f"{manga_info.project}","images",f"{manga_info.title}",chapter.title)
return Path(BASE_IMAGES_DIR,f"{manga_info.project}","images",f"{cls.date_path()}",f"{manga_info.title}",chapter.title)
@classmethod
def chapter_cbz_dir(cls, manga_info: MangaInfo) -> Path:
"""生成章节CBZ文件目录"""
return Path(CBZ_DIR,f"{manga_info.project}",f"{manga_info.title}")
#return Path(CBZ_DIR,f"{manga_info.project}",f"{manga_info.title}")
return Path(CBZ_DIR,f"{manga_info.project}",f"{FirstLetterClassifier().format_title(manga_info.title)}")
@classmethod
def manga_cover_dir(cls, manga_item: MangaItem) -> Path:
"""生成漫画封面目录"""
return Path(BASE_IMAGES_DIR,f"{manga_item.info.project}","icons",f"{manga_item.info.title}",f"{manga_item.info.title}.jpg")
return Path(BASE_IMAGES_DIR,f"{manga_item.info.project}","icons",f"{FirstLetterClassifier().format_title(manga_item.info.title)}",f"{manga_item.info.title}.jpg")
@classmethod
def manga_cover_dir(cls, manga_info: MangaInfo, cache: bool = True, is_dir: bool = False) -> Path:
@@ -41,7 +49,7 @@ class DirectoryNaming:
if cache:
path = Path(BASE_IMAGES_DIR,f"{manga_info.project}","icons",".cache")
else:
path = Path(BASE_IMAGES_DIR,f"{manga_info.project}","icons",f"{manga_info.title}")
path = Path(BASE_IMAGES_DIR,f"{manga_info.project}","icons",f"{FirstLetterClassifier().format_title(manga_info.title)}")
if not is_dir:
path = os.path.join(path, f"{manga_info.title}.jpg")
return Path(path)
@@ -74,8 +82,9 @@ class FileNaming:
@classmethod
def chapter_cbz(cls, manga_info: MangaInfo, chapter: str) -> Path:
"""生成章节CBZ文件目录"""
return Path(CBZ_DIR,f"{manga_info.project}",f"{manga_info.title}",f"{chapter.title}.CBZ")
#return Path(CBZ_DIR,f"{manga_info.project}",f"{manga_info.title}",f"{chapter.title}.CBZ")
return Path(CBZ_DIR,f"{manga_info.project}",f"{FirstLetterClassifier().format_title(manga_info.title)}",f"{chapter.title}.CBZ")
@classmethod
def old_chapter_cbz(cls, manga_info: MangaInfo, chapter: str) -> Path:
"""生成章节CBZ文件目录"""
@@ -320,4 +329,51 @@ class NamingStrategy:
from ..utils import get_file_extension
ext = get_file_extension(url)
return f"{prefix}_{idx:0{digits}d}{ext}"
return filename_generator
return filename_generator
class FirstLetterClassifier:
"""首字母分类器 - 根据名称返回首字大写的拼音首字母"""
@classmethod
def get_first_letter(cls,name):
"""
获取名称的首字母分类
Args:
name (str): 文件夹名或任何字符串
Returns:
str: 首字母大写
"""
if not name:
return "OTHER"
first_char = name[0]
# 汉字转拼音首字母
if '\u4e00' <= first_char <= '\u9fff':
try:
pinyin_list = pinyin(first_char, style=Style.FIRST_LETTER)
if pinyin_list and pinyin_list[0]:
return pinyin_list[0][0].upper()
except:
pass
# 英文字母转大写
if first_char.isalpha():
return first_char.upper()
#result = re.sub(r'[^a-zA-Z0-9]', '#', first_char)
# 其他字符直接返回
return first_char
@classmethod
def format_name(cls, name):
char_name = cls.get_first_letter(name)
first_name = re.sub(r'[^a-zA-Z0-9]', '#', char_name)
if first_name in [ '0', '1', '2', '3', '4', '5', '6', '7', '8', '9']:
return os.path.join(first_name, str(name)[-1])
return os.path.join(first_name, name[0])
@classmethod
def format_title(cls, title):
return os.path.join(cls.format_name(title), title)
src/common/utils.py
+3
View File
@@ -211,6 +211,9 @@ class MangaDownloader:
# logger.info(f"文件已存在，跳过下载: {file_path}")
# return True
# 从缓存中获取图片
if len(url) < 5:
logger.error(f"url={url}, 非法错误")
exit
cached_images = self.cache.get_image(url)
if cached_images:
with open(save_path, 'wb') as f:
src/config.py
+4 -3
View File
@@ -3,8 +3,9 @@ from pathlib import Path
from fake_useragent import UserAgent
# 基础配置
BASE_DIR = ""
# BASE_DIR = Path("/mnt/Comics")
#BASE_DIR = Path("/config/tempComic")
# BASE_DIR = ""
BASE_DIR = Path("/mnt/Comics")
BASE_IMAGES_DIR = Path(BASE_DIR,"output")
CACHE_DIR = Path(BASE_DIR, ".cache")
CACHE_IMAGE_DIR = CACHE_DIR / "images"
@@ -22,7 +23,7 @@ COMIC_INFO_NAME = "ComicInfo.xml"
XSD_FILE = "src/assets/ComicInfo_2.1.xsd"
# 代理配置
# PROXY_URL = "http://47.98.225.49:9890"
PROXY_URL = ""
PROXY_URL = "http://10.0.10.1:7899"
# 日志配置
LOG_FORMAT = '%(asctime)s - %(levelname)s - %(message)s'
src/sites/configs/rouman.py
+34 -3
View File
@@ -23,10 +23,41 @@ class RoumanSite(BaseSite):
html = await self._get(chapter_url)
tree = etree.HTML(html)
image_urls_str = []
json_all=""
for data_json in tree.xpath('//script/text()'):
data_json = data_json.replace('\\', '')
if "imageUrl" in data_json:
image_urls_str = re.findall(r'"imageUrl":"(https?://[^"]+)"', data_json)
if str(data_json).startswith('self.__next_f.push([1,'):
data_json = data_json.replace('self.__next_f.push([1,', '')
data_json = data_json.replace('])', '')
data_json = data_json.replace('\\n', '')
data_json = data_json.replace('\\', '')
if str(data_json).startswith('"'):
data_json=data_json[1:]
if str(data_json).endswith('"'):
data_json=data_json[0:-1]
#if str(data_json).endswith('"') and not str(data_json).endswith('.jpg"') and not str(data_json).endswith('.webp"') and not str(data_json).endswith('ind\"'):
#if str(data_json).endswith('.jpg"') and not str(data_json).endswith('.webp"') and not str(data_json).endswith('ind\"'):
# data_json=data_json[0:-2]
json_all=json_all+data_json
#if "imageUrl" in json_all:
# image_urls_str = re.findall(r'"imageUrl":"(https?://[^"]+)"', json_all)
# 方法3：使用更精确的正则表达式，确保imageUrl和ind在同一对象中
pattern_advanced = r'\{[^{}]*"imageUrl":"([^"]+)"[^{}]*"ind":(\d+)[^{}]*\}'
matches_advanced = re.findall(pattern_advanced, json_all)
# 检查图片数量是否完整
if len(matches_advanced)-1 != int(matches_advanced[-1][1]) or len(matches_advanced) == 0:
print(f"{chapter_url} \n图片数量不完整或为0")
exit()
ver_count = 0
image_urls_str = []
for image_url, ind in matches_advanced:
if str(image_url).startswith("http") and image_url.split(".")[-1] in [ "jpg" , "png" , "webp"]:
#print(f"imageUrl: {image_url}, ind: {ind}")
image_urls_str.append(image_url)
ver_count += 1
if ver_count != len(matches_advanced):
print(f"{matches_advanced} \n图片数量不完整,第二次校验")
exit()
# 再次通过获取的XPATH数据解析并保存到ci(ComicItem)中
# 正则表达式匹配 .jpg 链接
# 打印提取的 .jpg 链接
src/sites/manager.py
+1 -1
View File
@@ -17,7 +17,7 @@ class MangaManager:
SITE_MAP: Dict[str, Type[BaseSite]] = {
# 'manhuagui.com': ManhuaguiSite,
'roum20.xyz': RoumanSite,
'roum26.xyz': RoumanSite,
'rouman5.com': RoumanSite,
# 在这里添加更多网站支持
}
test.py
+69 -9
View File
@@ -14,7 +14,9 @@ from src.common.ComicInfo import ComicInfoXml
from lxml import etree
from collections import defaultdict
from typing import List, Dict, Tuple
from pathlib import Path
from src.common.naming import FirstLetterClassifier
class test:
def clean_min_cbz(self):
@@ -36,7 +38,7 @@ class test:
m_time = datetime.fromtimestamp(os.path.getmtime(cbz_path))
str_strftime = '%Y%m%d'
zip_time = m_time.strftime(str_strftime)
file_date_time = ""
with ZipFile(cbz_path, 'r') as zip_ref:
old_img = 0
for file_info in zip_ref.infolist():
@@ -47,22 +49,76 @@ class test:
# 格式化输出日期时间，例如：YYYY-MM-DD HH:MM:SS
file_date_time = dt.strftime(str_strftime)
# 一周内的图片跳过
if int(zip_time) - int(file_date_time) > 7:
#if int(zip_time) - int(file_date_time) > 7:
#print(f"Clear Filename: {file_info.filename}, zip: {cbz_path}")
if int(file_date_time) < 20250910:
break
if int(file_date_time) > 20250910:
old_img += 1
break
if old_img > 0:
#os.remove(cbz_path)
print(f"remove cbz {cbz_path}")
os.remove(cbz_path)
print(f"remove cbz {cbz_path}. date={file_date_time}")
def _bydate_clean_old_cbz(self, cbz_path):
creation_time = Path(cbz_path).stat().st_ctime
#m_time = datetime.fromtimestamp(os.path.getmtime(cbz_path))
m_time = datetime.fromtimestamp(creation_time)
str_strftime = '%Y%m%d'
zip_time = m_time.strftime(str_strftime)
# 删除这个时间之后的文件
remove_time = 20250907
if int(zip_time) > remove_time:
os.remove(cbz_path)
print(f"remove cbz {cbz_path}, date= {zip_time}")
def clean_old_cbz(self):
dir_path = "/mnt/Comics/CBZ/rm_comic"
for dir in os.listdir(dir_path):
c_dir = os.path.join(dir_path, dir)
if os.path.isdir(c_dir):
m_time = datetime.fromtimestamp(os.path.getmtime(c_dir))
str_strftime = '%Y%m%d'
zip_time = m_time.strftime(str_strftime)
# 删除这个时间之后的文件
remove_time = 20250907
if int(zip_time) > remove_time:
print(c_dir)
files = list(FileNaming.get_filenames_optimized(c_dir, ext_filter=['.CBZ']))
for file in files:
self._clean_old_cbz(file)
def bydate_clean_old_cbz(self):
dir_path = "/mnt/Comics/CBZ/rm_comic"
for dir in os.listdir(dir_path):
c_dir = os.path.join(dir_path, dir)
if os.path.isdir(c_dir):
files = list(FileNaming.get_filenames_optimized(c_dir, ext_filter=['.CBZ']))
for file in files:
self._clean_old_cbz(file)
self._bydate_clean_old_cbz(file)
def format_path(self):
#dir_path = "/config/tempComic/CBZ/rm_comic"
#dir_path = "/mnt/Comics/CBZ/rm_comic"
# icons
#dir_path = "/mnt/Comics/output/rm_comic/icons"
# json
dir_path = "/mnt/Comics/output/rm_comic/json"
for dir in os.listdir(dir_path):
c_dir = os.path.join(dir_path, dir)
if os.path.isdir(c_dir):
files = list(FileNaming.get_filenames_optimized(c_dir, ext_filter=['.CBZ', '.jpg', '.json']))
#for file in files:
# self._bydate_clean_old_cbz(file)
if len(files) > 0 and not str(dir).startswith("."):
format_name = FirstLetterClassifier.format_name(dir)
format_dir = os.path.join(dir_path, format_name)
if not os.path.exists(format_dir):
os.makedirs(format_dir)
format_path = os.path.join(format_dir, dir)
print(f"{c_dir} ===> {format_path}")
shutil.move(c_dir, format_path)
class comicInfo:
@@ -434,7 +490,11 @@ class comicInfo:
if __name__ == "__main__":
print("开始处理")
# ComicInfoXml()._xml_file_to_comicinfo("/Users/cc/Documents/Dev/WorkSpace/VSCodeProjects/NewComicDownloader/CBZ/rm_comic/和朋友的妈妈做朋友/第37话 37.CBZ")
xml_path = ComicInfoXml().update_comicinfo_count_or_number(count=37,cbz_path="/Users/cc/Documents/Dev/WorkSpace/VSCodeProjects/NewComicDownloader/CBZ/rm_comic/和朋友的妈妈做朋友/第37话 37.CBZ")
comicInfo().update_cbz_with_new_xml("/Users/cc/Documents/Dev/WorkSpace/VSCodeProjects/NewComicDownloader/CBZ/rm_comic/和朋友的妈妈做朋友/第37话 37.CBZ", xml_path.read_text(encoding="utf-8"))
#xml_path = ComicInfoXml().update_comicinfo_count_or_number(count=37,cbz_path="/Users/cc/Documents/Dev/WorkSpace/VSCodeProjects/NewComicDownloader/CBZ/rm_comic/和朋友的妈妈做朋友/第37话 37.CBZ")
#comicInfo().update_cbz_with_new_xml("/Users/cc/Documents/Dev/WorkSpace/VSCodeProjects/NewComicDownloader/CBZ/rm_comic/和朋友的妈妈做朋友/第37话 37.CBZ", xml_path.read_text(encoding="utf-8"))
#items = ci().__dict__.keys()
#print(items)
#print(items)
#test().clean_old_cbz()
test().format_path()
#file_name = FirstLetterClassifier().format_name("2025年的")
#print(file_name)