"""全流程编排器 串联 搜索 → 下载 → 图片处理 → KEPUB 打包 的完整流水线。 """ from pathlib import Path from kobo_manga.config import AppConfig from kobo_manga.converter.kepub import KepubBuilder from kobo_manga.db.database import Database from kobo_manga.db.queries import ( get_downloaded_chapter_ids, get_manga, get_manga_chapter_stats, update_subscription_checked, upsert_chapters, upsert_manga, ) from kobo_manga.downloader.engine import DownloadEngine from kobo_manga.models import Chapter, DownloadResult, MangaInfo from kobo_manga.processor.pipeline import ImageProcessor from kobo_manga.sources import get_source class MangaPipeline: """全流程编排：搜索 → 下载 → 处理 → 打包。""" def __init__( self, config: AppConfig, db: Database, base_dir: Path = Path("."), ): self.config = config self.db = db self.downloads_dir = base_dir / "downloads" self.output_dir = base_dir / "output" def _source_name(self) -> str: """获取当前配置的源名称。""" return self.config.sources[0] if self.config.sources else "manhuagui" async def search( self, keyword: str, source_name: str | None = None ) -> list[MangaInfo]: """搜索漫画。source_name=None 时搜索所有已配置的源。""" if source_name: async with get_source(source_name) as source: return await source.search(keyword) # 多源聚合搜索 results: list[MangaInfo] = [] for name in self.config.sources or ["manhuagui"]: try: async with get_source(name) as source: results.extend(await source.search(keyword)) except Exception as e: print(f" [!] {name} 搜索失败: {e}") return results async def get_manga_info( self, manga_url: str, source_name: str | None = None ) -> MangaInfo: """获取漫画详情（含章节列表）。""" name = source_name or self._source_name() async with get_source(name) as source: manga = await source.get_manga_info(manga_url) upsert_manga(self.db, manga) return manga async def download_and_convert( self, manga_url: str, source_name: str | None = None, chapter_range: tuple[float, float] | None = None, chapter_ids: list[str] | None = None, chapter_type: str | None = None, ) -> list[Path]: """全流程：下载 → 图片处理 → KEPUB 打包。 Args: manga_url: 漫画 URL source_name: 源名称，None 则用默认源 chapter_range: 章节号范围 (start, end)，闭区间 chapter_ids: 指定章节 ID 列表 chapter_type: 章节类型筛选 (volume/chapter/extra) Returns: 生成的 .kepub.epub 文件路径列表 """ name = source_name or self._source_name() async with get_source(name) as source: engine = DownloadEngine( db=self.db, source=source, config=self.config, base_dir=self.downloads_dir, ) result = await engine.download_manga( manga_url, chapter_range=chapter_range, chapter_ids=chapter_ids, chapter_type=chapter_type, ) return self._process_results(result) def _process_results(self, result: DownloadResult) -> list[Path]: """对下载成功的章节执行图片处理和 KEPUB 打包。""" kepub_paths = [] processor = ImageProcessor( self.config.processing, self.config.device ) successful = [ cr for cr in result.chapter_results if cr.status == "downloaded" and cr.download_path ] if not successful: return kepub_paths manga_output = self.output_dir / _sanitize_filename(result.manga.title) for i, cr in enumerate(successful, 1): # 检查是否已有 KEPUB（用章节 ID 避免同名冲突） kepub_name = ( f"{_sanitize_filename(result.manga.title)} - " f"{cr.chapter.id} {_sanitize_filename(cr.chapter.title)}.kepub.epub" ) kepub_path = manga_output / kepub_name if kepub_path.exists(): print(f" [SKIP] 已存在: {cr.chapter.title}") kepub_paths.append(kepub_path) continue print( f"\n[{i}/{len(successful)}] " f"处理+打包: {cr.chapter.title}" ) try: path = self._process_and_convert( result.manga, cr.chapter, cr.download_path ) kepub_paths.append(path) print(f" [OK] {path.name}") except Exception as e: print(f" [!] 处理失败: {e}") return kepub_paths def _process_and_convert( self, manga: MangaInfo, chapter: Chapter, download_path: Path, ) -> Path: """处理图片 + 打包 KEPUB（单章节）。""" processor = ImageProcessor( self.config.processing, self.config.device ) # 收集下载的图片文件 image_paths = sorted( p for p in download_path.iterdir() if p.suffix.lower() in (".jpg", ".jpeg", ".png", ".webp") ) if not image_paths: raise ValueError(f"目录中无图片文件: {download_path}") # 图片处理 processed_dir = ( self.downloads_dir.parent / "processed" / _sanitize_filename(manga.title) / _sanitize_filename(chapter.title) ) processed_paths = processor.process_chapter(image_paths, processed_dir) print(f" 处理: {len(image_paths)} -> {len(processed_paths)} 张") # KEPUB 打包 output_dir = self.output_dir / _sanitize_filename(manga.title) builder = KepubBuilder(manga, chapter, self.config.device) kepub_path = builder.build(processed_paths, output_dir) print(f" 打包: {kepub_path.name}") return kepub_path async def check_and_download_updates( self, manga_id: str, source_name: str ) -> list[Path]: """检查订阅漫画的新章节并下载转换。 Returns: 新生成的 .kepub.epub 路径列表 """ # 1. 从 DB 获取漫画记录 manga = get_manga(self.db, manga_id, source_name) if not manga: print(f" [!] 漫画不存在: {manga_id} ({source_name})") return [] # 2. 拉取最新章节列表 async with get_source(source_name) as source: fresh_manga = await source.get_manga_info(manga.url) # 3. 更新 DB upsert_manga(self.db, fresh_manga) upsert_chapters( self.db, fresh_manga.id, fresh_manga.source, fresh_manga.chapters ) # 4. 找出新章节（未下载的） downloaded_ids = get_downloaded_chapter_ids( self.db, fresh_manga.id, fresh_manga.source ) new_chapters = [ ch for ch in fresh_manga.chapters if ch.id not in downloaded_ids ] if not new_chapters: print(f" [OK] {fresh_manga.title}: 无新章节") update_subscription_checked(self.db, manga_id, source_name) return [] print(f" 发现 {len(new_chapters)} 个新章节") new_chapter_ids = [ch.id for ch in new_chapters] # 5. 下载新章节 async with get_source(source_name) as source: engine = DownloadEngine( db=self.db, source=source, config=self.config, base_dir=self.downloads_dir, ) result = await engine.download_manga( fresh_manga.url, chapter_ids=new_chapter_ids, ) # 6. 处理+打包 kepub_paths = self._process_results(result) # 7. 更新订阅状态 last_ch_id = new_chapters[-1].id if new_chapters else None update_subscription_checked( self.db, manga_id, source_name, last_chapter_id=last_ch_id ) return kepub_paths def get_stats(self, manga: MangaInfo) -> dict: """获取漫画的章节统计。""" return get_manga_chapter_stats(self.db, manga.id, manga.source) def _sanitize_filename(name: str) -> str: """清理文件名中的非法字符。""" return "".join( c if c.isalnum() or c in " _-()（）【】" else "_" for c in name )