"""多章节下载引擎 在 basic.py 的单章节下载基础上，增加： - 多章节批量下载编排 - SQLite 状态追踪（去重、断点续传） - 章节范围选择 """ from __future__ import annotations from pathlib import Path from typing import TYPE_CHECKING if TYPE_CHECKING: from kobo_manga.sources.base import BaseSource from kobo_manga.config import AppConfig from kobo_manga.db.database import Database from kobo_manga.db.queries import ( are_all_pages_downloaded, get_downloaded_chapter_ids, get_pending_pages, mark_page_downloaded, mark_page_failed, set_chapter_status, upsert_chapters, upsert_manga, upsert_pages, ) from kobo_manga.downloader.basic import download_chapter as _download_chapter_basic from kobo_manga.models import ( Chapter, ChapterResult, DownloadResult, MangaInfo, ) class DownloadEngine: """多章节下载编排器，带 SQLite 状态追踪。""" def __init__( self, db: Database, source: "BaseSource", # 漫画源适配器 config: AppConfig, base_dir: Path = Path("downloads"), ): self.db = db self.source = source self.config = config self.base_dir = base_dir async def download_manga( self, manga_id_or_url: str, chapter_range: tuple[float, float] | None = None, chapter_ids: list[str] | None = None, chapter_type: str | None = None, ) -> DownloadResult: """下载漫画的多个章节。 Args: manga_id_or_url: 漫画 URL 或 ID chapter_range: (起始章节号, 结束章节号)，闭区间 chapter_ids: 指定章节 ID 列表（与 chapter_range 二选一） chapter_type: 章节类型筛选 (volume/chapter/extra) Returns: DownloadResult 汇总结果 """ # 1. 获取漫画信息和章节列表 print(f"获取漫画信息: {manga_id_or_url}") manga = await self.source.get_manga_info(manga_id_or_url) print(f" {manga.title} - 共 {len(manga.chapters)} 个章节") # 2. 持久化到数据库 upsert_manga(self.db, manga) upsert_chapters(self.db, manga.id, manga.source, manga.chapters) # 3. 去重：获取已下载的章节 done_ids = get_downloaded_chapter_ids(self.db, manga.id, manga.source) # 4. 筛选要下载的章节 selected = self._select_chapters( manga.chapters, chapter_range, chapter_ids, chapter_type ) to_download = [ch for ch in selected if ch.id not in done_ids] skipped = len(selected) - len(to_download) print( f" 选中 {len(selected)} 个章节，" f"跳过 {skipped} 个已下载，" f"待下载 {len(to_download)} 个" ) # 5. 逐章下载 result = DownloadResult( manga=manga, chapters_total=len(selected), chapters_skipped=skipped, ) for i, chapter in enumerate(to_download, 1): print(f"\n[{i}/{len(to_download)}] {chapter.title}") ch_result = await self.download_chapter(manga, chapter) result.chapter_results.append(ch_result) if ch_result.status == "downloaded": result.chapters_downloaded += 1 else: result.chapters_failed += 1 # 汇总 print(f"\n{'='*50}") print( f"完成: {result.chapters_downloaded} 下载 / " f"{result.chapters_skipped} 跳过 / " f"{result.chapters_failed} 失败" ) return result async def download_chapter( self, manga: MangaInfo, chapter: Chapter ) -> ChapterResult: """下载单个章节，带状态追踪。""" source_name = manga.source # 标记为下载中 set_chapter_status(self.db, chapter.id, source_name, "downloading") try: # 获取图片列表 images = await self.source.get_chapter_images(chapter) print(f" 共 {len(images)} 页") # 记录页面到数据库 upsert_pages(self.db, chapter.id, source_name, images) # 断点续传：只下载未完成的页 pending = get_pending_pages(self.db, chapter.id, source_name) total_pages = len(images) if not pending: # 所有页面已下载 output_dir = self._chapter_dir(manga, chapter) set_chapter_status( self.db, chapter.id, source_name, "downloaded", str(output_dir), ) print(f" 所有页面已存在，跳过") return ChapterResult( chapter=chapter, status="downloaded", pages_total=total_pages, pages_downloaded=total_pages, download_path=output_dir, ) already_done = total_pages - len(pending) if already_done > 0: print(f" 续传: {already_done} 页已存在，下载剩余 {len(pending)} 页") # 调用基础下载器 output_dir = self._chapter_dir(manga, chapter) downloaded = await _download_chapter_basic( pending, output_dir, chapter.url, self.config ) # 更新页面状态 downloaded_nums = set() for page in downloaded: if page.local_path: mark_page_downloaded( self.db, chapter.id, source_name, page.page_number, page.local_path, ) downloaded_nums.add(page.page_number) # 标记失败的页面 for page in pending: if page.page_number not in downloaded_nums: mark_page_failed( self.db, chapter.id, source_name, page.page_number ) # 检查是否全部完成 all_done = are_all_pages_downloaded( self.db, chapter.id, source_name ) pages_downloaded = already_done + len(downloaded_nums) if all_done: set_chapter_status( self.db, chapter.id, source_name, "downloaded", str(output_dir), ) status = "downloaded" print(f" 完成: {pages_downloaded}/{total_pages} 页") else: set_chapter_status( self.db, chapter.id, source_name, "failed" ) status = "partial" print( f" 部分失败: {pages_downloaded}/{total_pages} 页" ) return ChapterResult( chapter=chapter, status=status, pages_total=total_pages, pages_downloaded=pages_downloaded, download_path=output_dir, ) except Exception as e: set_chapter_status( self.db, chapter.id, source_name, "failed" ) print(f" 下载失败: {e}") return ChapterResult( chapter=chapter, status="failed", pages_total=0, pages_downloaded=0, ) async def resume_incomplete(self, manga_id_or_url: str) -> DownloadResult: """恢复未完成的下载。重新运行 download_manga 即可，去重逻辑会自动跳过已完成的。""" return await self.download_manga(manga_id_or_url) def _select_chapters( self, chapters: list[Chapter], chapter_range: tuple[float, float] | None, chapter_ids: list[str] | None, chapter_type: str | None = None, ) -> list[Chapter]: """按范围、ID 或类型筛选章节。""" # 先按类型过滤 if chapter_type is not None: chapters = [ch for ch in chapters if ch.chapter_type == chapter_type] if chapter_ids is not None: id_set = set(chapter_ids) return [ch for ch in chapters if ch.id in id_set] if chapter_range is not None: start, end = chapter_range return [ ch for ch in chapters if start <= ch.chapter_number <= end ] # 未指定范围则选择全部（已按类型过滤） return list(chapters) def _chapter_dir(self, manga: MangaInfo, chapter: Chapter) -> Path: """计算章节下载目录。""" safe_manga = _sanitize_filename(manga.title) safe_chapter = _sanitize_filename(chapter.title) return self.base_dir / safe_manga / safe_chapter def _sanitize_filename(name: str) -> str: """清理文件名中的非法字符。""" return "".join( c if c.isalnum() or c in " _-()（）【】" else "_" for c in name )