NewComicDownloader/komga.py

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