diff --git a/.gitignore b/.gitignore index 8033124..b83693a 100644 --- a/.gitignore +++ b/.gitignore @@ -5,4 +5,5 @@ CBZ/* output/* downloads/* -/**/__pycache__ \ No newline at end of file +/**/__pycache__ +komga*.json \ No newline at end of file diff --git a/komga.py b/komga.py new file mode 100644 index 0000000..8aa51d6 --- /dev/null +++ b/komga.py @@ -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 \ No newline at end of file diff --git a/run.py b/run.py index 2f563cb..00953d0 100644 --- a/run.py +++ b/run.py @@ -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__": diff --git a/src/common/extractor.py b/src/common/extractor.py index 3895ded..b3bae25 100644 --- a/src/common/extractor.py +++ b/src/common/extractor.py @@ -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: diff --git a/src/common/item.py b/src/common/item.py index a202517..6469f1a 100644 --- a/src/common/item.py +++ b/src/common/item.py @@ -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 = [] diff --git a/src/common/naming.py b/src/common/naming.py index fa44806..18dd381 100644 --- a/src/common/naming.py +++ b/src/common/naming.py @@ -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 \ No newline at end of file + 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) \ No newline at end of file diff --git a/src/common/utils.py b/src/common/utils.py index 263db22..091a634 100644 --- a/src/common/utils.py +++ b/src/common/utils.py @@ -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: diff --git a/src/config.py b/src/config.py index 84dff3e..d2516ab 100644 --- a/src/config.py +++ b/src/config.py @@ -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' diff --git a/src/sites/configs/rouman.py b/src/sites/configs/rouman.py index a2d913c..da433f3 100644 --- a/src/sites/configs/rouman.py +++ b/src/sites/configs/rouman.py @@ -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 链接 diff --git a/src/sites/manager.py b/src/sites/manager.py index fe851b0..160e6a7 100644 --- a/src/sites/manager.py +++ b/src/sites/manager.py @@ -17,7 +17,7 @@ class MangaManager: SITE_MAP: Dict[str, Type[BaseSite]] = { # 'manhuagui.com': ManhuaguiSite, - 'roum20.xyz': RoumanSite, + 'roum26.xyz': RoumanSite, 'rouman5.com': RoumanSite, # 在这里添加更多网站支持 } diff --git a/test.py b/test.py index 4fb51a0..972a539 100644 --- a/test.py +++ b/test.py @@ -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) \ No newline at end of file + #print(items) + #test().clean_old_cbz() + test().format_path() + #file_name = FirstLetterClassifier().format_name("2025年的") + #print(file_name) \ No newline at end of file