CD
c92433b0b1
- tools/anissia_client.py: Anissia API 클라이언트 (편성표/자막) - tools/nyaa_client.py: Nyaa.si RSS 토렌트 검색 - tools/qbit_client.py: qBittorrent Web API 클라이언트 - tools/subtitle_downloader.py: Google Drive/Tistory/Naver 자막 파서 - tools/title_matcher.py: 제목 매칭 + NAS 폴더명 생성 - tools/anime_pipeline.py: 전체 파이프라인 오케스트레이터 - tools/nas_scanner.py: NAS 폴더/파일 스캔 - prompts/unified.md: anime 모드 추가 (AI 평문 의도 분류) - api/discord_bot.py: AI 평문 anime 핸들러 + /anime 슬래시 커맨드 - config.py: qBittorrent/NAS 설정 추가 - .agents/: agent_guide 워크플로우 통합 - docs/devlog: 세션 기록
245 lines
8.7 KiB
Python
245 lines
8.7 KiB
Python
"""애니메이션 자동화 파이프라인.
|
|
|
|
전체 흐름:
|
|
1. Anissia에서 애니 검색 → 자막 정보 확인
|
|
2. Nyaa.si에서 토렌트 검색 → 제목 매칭
|
|
3. qBittorrent에 magnet 추가 → NAS 경로 지정
|
|
4. 자막 다운로드 → 파일명 매칭
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import config
|
|
from tools.anissia_client import AnissiaClient, AnimeInfo, CaptionInfo
|
|
from tools.nyaa_client import NyaaClient, TorrentResult
|
|
from tools.qbit_client import QBitClient
|
|
from tools.subtitle_downloader import SubtitleDownloader, SubtitleFile
|
|
from tools.title_matcher import (
|
|
match_titles, make_nas_folder_name, rename_subtitle_to_video,
|
|
)
|
|
|
|
logger = logging.getLogger("variet.tools.pipeline")
|
|
|
|
|
|
@dataclass
|
|
class DownloadResult:
|
|
"""파이프라인 실행 결과."""
|
|
success: bool
|
|
anime: Optional[AnimeInfo] = None
|
|
captions: list[CaptionInfo] = field(default_factory=list)
|
|
torrents: list[TorrentResult] = field(default_factory=list)
|
|
subtitles: list[SubtitleFile] = field(default_factory=list)
|
|
nas_folder: str = ""
|
|
torrent_added: bool = False
|
|
message: str = ""
|
|
errors: list[str] = field(default_factory=list)
|
|
|
|
|
|
class AnimePipeline:
|
|
"""애니메이션 다운로드 자동화 파이프라인."""
|
|
|
|
def __init__(self):
|
|
self.anissia = AnissiaClient()
|
|
self.nyaa = NyaaClient()
|
|
self.qbit = QBitClient()
|
|
self.sub_downloader = SubtitleDownloader()
|
|
self.nas_base = getattr(config, "NAS_ANIME_PATH",
|
|
r"\\192.168.10.10\NasData\Video\Animation")
|
|
|
|
from tools.nas_scanner import NasScanner
|
|
self.nas = NasScanner(self.nas_base)
|
|
|
|
async def search(self, title: str) -> DownloadResult:
|
|
"""애니 검색 — 정보 + 자막 + 토렌트 현황 표시.
|
|
|
|
실제 다운로드 없이 검색 결과만 반환.
|
|
"""
|
|
result = DownloadResult(success=False)
|
|
|
|
# 1. Anissia에서 검색
|
|
try:
|
|
anime_list = await self.anissia.search_anime(title)
|
|
except Exception as e:
|
|
result.errors.append(f"Anissia 검색 오류: {e}")
|
|
return result
|
|
|
|
if not anime_list:
|
|
result.message = f"'{title}' 검색 결과가 없습니다."
|
|
return result
|
|
|
|
anime = anime_list[0] # 첫 번째 결과 사용
|
|
result.anime = anime
|
|
|
|
# 2. 자막 정보
|
|
try:
|
|
captions = await self.anissia.get_captions(anime.anime_no)
|
|
result.captions = captions
|
|
except Exception as e:
|
|
result.errors.append(f"자막 조회 오류: {e}")
|
|
|
|
# 3. Nyaa 토렌트 검색 (원제 로마자로)
|
|
try:
|
|
from tools.title_matcher import japanese_to_romaji
|
|
romaji_title = japanese_to_romaji(anime.original_subject)
|
|
|
|
# 먼저 로마자로 검색
|
|
torrents = await self.nyaa.search(romaji_title)
|
|
if not torrents:
|
|
# 원제 그대로 검색
|
|
torrents = await self.nyaa.search(anime.original_subject)
|
|
|
|
# 제목 매칭 필터링
|
|
matched = match_titles(
|
|
anime.subject, anime.original_subject, torrents, threshold=0.3
|
|
)
|
|
result.torrents = matched[:20] # 상위 20개
|
|
except Exception as e:
|
|
result.errors.append(f"Nyaa 검색 오류: {e}")
|
|
|
|
# NAS 폴더명 생성
|
|
result.nas_folder = make_nas_folder_name(anime.subject, anime.start_date)
|
|
|
|
result.success = True
|
|
result.message = (
|
|
f"**{anime.subject}** ({anime.original_subject})\n"
|
|
f"자막 제작자: {len(result.captions)}명 | "
|
|
f"토렌트: {len(result.torrents)}건\n"
|
|
f"NAS 폴더: `{result.nas_folder}`"
|
|
)
|
|
return result
|
|
|
|
async def download(
|
|
self,
|
|
title: str,
|
|
mode: str = "auto",
|
|
episode: Optional[int] = None,
|
|
) -> DownloadResult:
|
|
"""애니 다운로드 실행.
|
|
|
|
Args:
|
|
title: 한글 제목
|
|
mode: "auto" (자막+영상), "sub_only" (자막만), "video_only" (영상만)
|
|
episode: 특정 에피소드만 (None이면 최신)
|
|
"""
|
|
# 먼저 검색
|
|
result = await self.search(title)
|
|
if not result.success:
|
|
return result
|
|
|
|
anime = result.anime
|
|
nas_folder = Path(self.nas_base) / result.nas_folder
|
|
|
|
# ── 자막 다운로드 ──
|
|
if mode in ("auto", "sub_only"):
|
|
await self._download_subtitles(result, nas_folder, episode)
|
|
|
|
# ── 영상 토렌트 추가 ──
|
|
if mode in ("auto", "video_only"):
|
|
force = (mode == "video_only")
|
|
await self._add_torrents(result, nas_folder, episode, force=force)
|
|
|
|
# 결과 메시지 구성
|
|
parts = [result.message]
|
|
if result.subtitles:
|
|
parts.append(f"\n📝 자막 {len(result.subtitles)}건 다운로드 완료")
|
|
if result.torrent_added:
|
|
parts.append(f"\n🎬 토렌트 추가 완료 → `{nas_folder}`")
|
|
if result.errors:
|
|
parts.append(f"\n⚠️ 오류: " + "; ".join(result.errors))
|
|
|
|
result.message = "\n".join(parts)
|
|
return result
|
|
|
|
async def _download_subtitles(
|
|
self,
|
|
result: DownloadResult,
|
|
nas_folder: Path,
|
|
episode: Optional[int],
|
|
):
|
|
"""자막 다운로드 처리."""
|
|
sub_dir = nas_folder / "subtitles"
|
|
|
|
for caption in result.captions:
|
|
if not caption.website:
|
|
continue
|
|
if episode is not None and caption.episode != str(episode):
|
|
continue
|
|
|
|
try:
|
|
subs = await self.sub_downloader.find_subtitles(caption.website)
|
|
for sub in subs:
|
|
if episode is not None and sub.episode is not None and sub.episode != episode:
|
|
continue
|
|
try:
|
|
await self.sub_downloader.download_file(sub, str(sub_dir))
|
|
result.subtitles.append(sub)
|
|
except Exception as e:
|
|
result.errors.append(f"자막 다운로드 실패 ({sub.filename}): {e}")
|
|
except Exception as e:
|
|
result.errors.append(f"자막 사이트 접근 실패 ({caption.name}): {e}")
|
|
|
|
async def _add_torrents(
|
|
self,
|
|
result: DownloadResult,
|
|
nas_folder: Path,
|
|
episode: Optional[int],
|
|
force: bool = False,
|
|
):
|
|
"""토렌트 추가 처리."""
|
|
if not result.torrents:
|
|
result.errors.append("매칭되는 토렌트가 없습니다.")
|
|
return
|
|
|
|
# 에피소드 필터링
|
|
candidates = result.torrents
|
|
if episode is not None:
|
|
candidates = [t for t in candidates if t.episode == episode]
|
|
if not candidates:
|
|
result.errors.append(f"{episode}화 토렌트를 찾지 못했습니다.")
|
|
return
|
|
|
|
# auto 모드 기본 조건: 자막이 있어야 영상 다운로드 (force면 무시)
|
|
if not force and not result.captions and not result.subtitles:
|
|
# 자막이 없으면 사용자에게 안내만
|
|
result.errors.append("자막이 없어 영상 다운로드를 보류합니다. /anime video로 강제 다운로드 가능")
|
|
return
|
|
|
|
# 최상위 1개 (가장 시더 많은) 추가
|
|
best = candidates[0]
|
|
try:
|
|
success = await self.qbit.add_torrent(
|
|
magnet_or_url=best.magnet_link,
|
|
save_path=str(nas_folder),
|
|
category="anime",
|
|
tags=result.anime.subject if result.anime else "",
|
|
)
|
|
result.torrent_added = success
|
|
if not success:
|
|
result.errors.append("qBittorrent 토렌트 추가 실패")
|
|
except Exception as e:
|
|
result.errors.append(f"qBittorrent 오류: {e}")
|
|
|
|
async def get_status(self) -> list[dict]:
|
|
"""현재 다운로드 큐 상태."""
|
|
try:
|
|
torrents = await self.qbit.list_torrents(category="anime")
|
|
return [
|
|
{
|
|
"name": t.name,
|
|
"progress": f"{t.progress * 100:.1f}%",
|
|
"state": t.state,
|
|
"size": f"{t.size / (1024**3):.2f} GB" if t.size > 0 else "?",
|
|
"speed": f"{t.download_speed / (1024**2):.1f} MB/s" if t.download_speed > 0 else "0",
|
|
"eta": f"{t.eta // 60}분" if t.eta > 0 else "∞",
|
|
"path": t.save_path,
|
|
}
|
|
for t in torrents
|
|
]
|
|
except Exception as e:
|
|
logger.error(f"qBittorrent 상태 조회 오류: {e}")
|
|
return []
|