From 3618387b8eede37bffa1984bfe9f2448e1765cf1 Mon Sep 17 00:00:00 2001 From: CD Date: Sun, 15 Mar 2026 18:23:57 +0900 Subject: [PATCH] =?UTF-8?q?feat(anime):=20=EC=9E=90=EB=A7=89/=ED=86=A0?= =?UTF-8?q?=EB=A0=8C=ED=8A=B8=20=ED=8C=8C=EC=9D=B4=ED=94=84=EB=9D=BC?= =?UTF-8?q?=EC=9D=B8=20=EB=8C=80=ED=8F=AD=20=EA=B0=9C=EC=84=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Blogspot Atom Feed API로 전체 에피소드 자막 URL 발견 - AniList prequel 체인 기반 시즌 에피소드 오프셋 자동 감지 - Nyaa S-tag 감지 → 절대/시즌 번호 체계 자동 판별 - 기존 자막 에피소드 스킵 (URL 페치 전 pre-skip) - 오프셋 적용 자막 리네임 (시즌번호→절대번호 매칭) - ASW HEVC 토렌트 우선 정렬 (truncation 방지) - 토렌트 완료 대기 → 자동 삭제 라이프사이클 - 중복 자막 자동 삭제 - .smi 자막 확장자 지원 --- api/discord_bot.py | 356 +++++++-- config.py | 5 + prompts/agent.md | 28 +- prompts/operator.md | 6 + tools/anime_pipeline.py | 1380 +++++++++++++++++++++++----------- tools/anissia_client.py | 20 +- tools/subtitle_downloader.py | 6 +- tools/title_matcher.py | 117 ++- 8 files changed, 1386 insertions(+), 532 deletions(-) diff --git a/api/discord_bot.py b/api/discord_bot.py index ff075e1..87e0fdd 100644 --- a/api/discord_bot.py +++ b/api/discord_bot.py @@ -142,7 +142,7 @@ async def _agent_call(text: str, history: str, project_path: str) -> str: ) response = await gemini.call_agent( - "agent", context, cwd=project_path, timeout=300, + "agent", context, cwd=project_path, timeout=1200, ) return response @@ -291,6 +291,39 @@ async def on_message(message: discord.Message): ) async with message.channel.typing(): + # 1단계: NLU 빠른 분류 (Gemini text 모드, ~5초) + intent = await _classify_intent(user_text, ws.path) + logger.info(f"NLU 분류: '{user_text[:50]}' → {intent}") + + # 2단계: anime 의도면 직접 실행 (Gemini agent 우회) + if intent and intent.get("mode") == "anime": + if progress_msg: + await progress_msg.edit(embed=discord.Embed( + title="📦 애니 작업 실행 중...", + description=f"```{user_text[:200]}```", + color=0x3498DB, + )) + await _handle_anime(message, intent) + if progress_msg: + try: await progress_msg.delete() + except Exception: pass + return + + # 2-2단계: research 의도면 조사+위키 등록 + if intent and intent.get("mode") == "research": + if progress_msg: + await progress_msg.edit(embed=discord.Embed( + title="🔍 리서치 진행 중...", + description=f"주제: **{intent.get('topic', user_text[:50])}**", + color=0x9B59B6, + )) + await _handle_research(message, intent, ws.path) + if progress_msg: + try: await progress_msg.delete() + except Exception: pass + return + + # 3단계: 일반 → Gemini agent 모드 history = await _get_channel_history(message.channel, limit=10) response = await _agent_call(user_text, history, ws.path) @@ -337,7 +370,47 @@ async def on_message(message: discord.Message): # ────────────────────────────────────────────── -# Anime 핸들러 (AI가 분류한 의도 실행) +# NLU 빠른 분류 (Gemini text 모드) +# ────────────────────────────────────────────── + +_NLU_PROMPT = """\ +사용자 메시지를 분류하여 JSON으로 응답하세요. 반드시 JSON만 출력. + +분류: +- 애니메이션 다운로드/검색/상태 → mode: "anime" +- 조사+위키 등록/정리 요청 → mode: "research" +- 그 외 → mode: "chat" + +anime 필드: action ("batch"|"download"|"search"|"status"|"schedule"|"list"), title, episode, filter + +research 필드: +- action: "research" (조사+등록), "organize" (기존 정리), "update" (페이지 수정) +- topic: 주제 + +예시: +"이번분기 애니 자막 다운받아줘" → {"mode":"anime","action":"batch","filter":"자막"} +"FX Forward 조사해서 위키에 정리해줘" → {"mode":"research","action":"research","topic":"FX Forward"} +"위키에 서버 관련 내용 정리해줘" → {"mode":"research","action":"organize","topic":"서버"} +"GraphQL이 뭐야?" → {"mode":"chat"} +""" + + +async def _classify_intent(text: str, project_path: str) -> dict | None: + """Gemini text 모드로 빠른 의도 분류 (~5초).""" + try: + gemini = GeminiCaller(project_path) + raw = await gemini.call("unified", f"{_NLU_PROMPT}\n\n사용자: {text}", timeout=30) + import json as _json + m = re.search(r'\{[^}]+\}', raw, re.DOTALL) + if m: + return _json.loads(m.group(0)) + except Exception as e: + logger.warning(f"NLU 분류 실패 (agent fallback): {e}") + return None + + +# ────────────────────────────────────────────── +# Anime 핸들러 (NLU 분류 결과로 직접 실행) # ────────────────────────────────────────────── async def _handle_anime(message: discord.Message, parsed: dict): @@ -362,6 +435,9 @@ async def _handle_anime(message: discord.Message, parsed: dict): elif action == "list": await _anime_list(message, pipeline, filter_str) + elif action == "batch": + await _anime_batch_direct(message, filter_str or "자막 배치") + elif action in ("download", "sub_only", "video_only"): # 필터에 batch 조건이 있으면 복수 다운로드 if not title and filter_str: @@ -437,66 +513,236 @@ async def _anime_download(message, pipeline, title, mode, episode): async def _anime_batch(message, pipeline, action, filter_str): - """필터 기반 복수 애니 다운로드 (이번분기 자막있는것 등).""" - embed = discord.Embed(title="⏳ 편성표 분석 중...", description="조건에 맞는 애니 검색", color=0xF39C12) + """필터 기반 복수 애니 다운로드 — batch_download() 사용.""" + await _anime_batch_direct(message, f"{filter_str} {action}") + + +async def _anime_batch_direct(message, user_text: str): + """애니 배치 다운로드 — Gemini 없이 직접 실행.""" + from tools.anime_pipeline import AnimePipeline + + t = user_text.lower() + mode = "auto" + if "자막만" in t or "sub_only" in t: + mode = "sub_only" + elif "영상만" in t or "video_only" in t: + mode = "video_only" + + sub_filter = "자막" in t or "sub:yes" in t # 자막 언급 시 자막 필터 활성 + + embed = discord.Embed( + title="📦 배치 다운로드 시작", + description=f"모드: `{mode}` | 자막 필터: `{'ON' if sub_filter else 'OFF'}`\n⏳ NAS 스캔 + Anissia 로딩 중...", + color=0xF39C12, + ) status_msg = await message.channel.send(embed=embed) - # 전체 편성표 로드 - all_anime = await pipeline.anissia.get_all_schedule() + pipeline = AnimePipeline() - # 필터 적용 - filtered = all_anime - if "sub:yes" in filter_str or "자막" in filter_str: - filtered = [a for a in filtered if a.caption_count > 0] - if "quarter:current" in filter_str or "이번" in filter_str: - from datetime import date - today = date.today() - current_q = (today.month - 1) // 3 + 1 - current_year = today.year - def _in_current_quarter(a): - if not a.start_date: - return False - parts = a.start_date.split("-") - y, m = int(parts[0]), int(parts[1]) - q = (m - 1) // 3 + 1 - return y == current_year and q == current_q - filtered = [a for a in filtered if _in_current_quarter(a)] - if "status:on" in filter_str: - filtered = [a for a in filtered if a.status == "ON"] - else: - # 기본: ON 상태만 - filtered = [a for a in filtered if a.status == "ON"] - - embed.title = f"📋 조건 매칭: {len(filtered)}개" - embed.description = "\n".join(f"• {a.subject} (자막 {a.caption_count}명)" for a in filtered[:15]) - if len(filtered) > 15: - embed.description += f"\n... 외 {len(filtered)-15}개" - embed.color = 0x3498DB - await status_msg.edit(embed=embed) - - if not filtered: + try: + results = await pipeline.batch_download(mode=mode, sub_filter=sub_filter) + except Exception as e: + logger.error(f"배치 다운로드 오류: {e}", exc_info=True) + embed.title = "❌ 배치 다운로드 실패" + embed.description = str(e)[:500] + embed.color = 0xE74C3C + await status_msg.edit(embed=embed) return - # 다운로드 실행 - success_count = 0 - fail_count = 0 - for anime in filtered: - try: - result = await pipeline.download(anime.subject, mode=action) - if result.torrent_added or result.subtitles: - success_count += 1 - else: - fail_count += 1 - except Exception as e: - logger.error(f"배치 다운로드 오류 ({anime.subject}): {e}") - fail_count += 1 + # 결과 정리 + success = [r for r in results if r.success] + lines = [] + for r in results: + icon = "✅" if r.success else "❌" + title = r.anime.subject if r.anime else "?" + detail = "" + if r.torrent_added: + detail += " 🎬토렌트추가" + if r.subtitles: + detail += f" 📝자막{len(r.subtitles)}" + if r.errors: + detail += f" ⚠️{r.errors[0][:50]}" + lines.append(f"{icon} {title}{detail}") - result_embed = discord.Embed( - title=f"📊 배치 다운로드 결과", - description=f"✅ 성공: {success_count}개\n⚠️ 실패/보류: {fail_count}개", - color=0x2ECC71 if success_count > 0 else 0xF39C12, + embed.title = f"📊 배치 결과: {len(success)}/{len(results)}건 성공" + embed.description = "\n".join(lines) or "처리할 애니가 없습니다." + embed.color = 0x2ECC71 if success else 0xF39C12 + await status_msg.edit(embed=embed) + + +# ────────────────────────────────────────────── +# Research 핸들러 (조사+위키 등록 / 기존 정리) +# ────────────────────────────────────────────── + +async def _handle_research(message: discord.Message, parsed: dict, project_path: str): + """리서치 요청 처리 — 조사 후 위키 등록 또는 기존 위키 정리.""" + from tools.wiki_client import WikiClient + + action = parsed.get("action", "research") + topic = parsed.get("topic", "") + wiki = WikiClient() + + if not topic: + await message.reply("🔍 어떤 주제를 조사할까요?") + return + + try: + if action == "research": + await _research_and_publish(message, topic, wiki, project_path) + elif action == "organize": + await _organize_wiki(message, topic, wiki, project_path) + elif action == "update": + await _research_and_publish(message, topic, wiki, project_path) + else: + await message.reply(f"❓ 알 수 없는 리서치 액션: {action}") + except Exception as e: + logger.error(f"리서치 핸들러 오류: {e}", exc_info=True) + await message.reply(f"❌ 리서치 오류: {str(e)[:300]}") + + +async def _research_and_publish( + message: discord.Message, topic: str, + wiki, project_path: str, +): + """주제 조사 → 위키 페이지 등록.""" + from tools.wiki_client import WikiClient + + status_msg = await message.channel.send( + embed=discord.Embed( + title="🔍 조사 중...", + description=f"**{topic}**\n\nGemini가 웹 검색 + 자료를 수집하고 있습니다.", + color=0x9B59B6, + ) ) - await message.channel.send(embed=result_embed) + + # Gemini agent로 조사 (google_web_search 자동 사용) + gemini = GeminiCaller(project_path) + research_prompt = ( + f"다음 주제에 대해 깊이 있게 조사하고, 위키 페이지용 마크다운으로 정리하세요.\n\n" + f"주제: {topic}\n\n" + f"요구사항:\n" + f"1. 반드시 웹 검색을 통해 최신 정보를 확인하세요\n" + f"2. 핵심 개념, 장단점, 실무 활용법을 포함하세요\n" + f"3. 출처 URL을 반드시 포함하세요\n" + f"4. 마크다운 형식으로 작성하세요 (# 제목부터)\n" + f"5. 한국어로 작성하세요\n" + ) + + try: + content = await gemini.call_agent( + "agent", research_prompt, cwd=project_path, timeout=300, + ) + except GeminiCallError as e: + await status_msg.edit(embed=discord.Embed( + title="❌ 조사 실패", description=str(e)[:500], color=0xE74C3C, + )) + return + + if not content or len(content) < 50: + await status_msg.edit(embed=discord.Embed( + title="⚠️ 결과 부족", description="충분한 정보를 수집하지 못했습니다.", + color=0xF39C12, + )) + return + + # 위키 등록 + slug = WikiClient.slugify(topic) + path = f"research/{slug}" + page = await wiki.upsert_page(path, topic, content, description=f"리서치: {topic}") + + # 대시보드 갱신 + await wiki.update_dashboard() + + await status_msg.edit(embed=discord.Embed( + title=f"✅ 위키 등록 완료: {topic}", + description=( + f"📄 [{path}](https://wiki.variet.net/{path})\n" + f"📊 대시보드 갱신됨\n" + f"📝 {len(content)}자 작성" + ), + color=0x2ECC71, + )) + + +async def _organize_wiki( + message: discord.Message, topic: str, + wiki, project_path: str, +): + """기존 위키 페이지들을 주제별로 재정리.""" + status_msg = await message.channel.send( + embed=discord.Embed( + title="📋 위키 분석 중...", + description=f"**{topic}** 관련 페이지를 수집하고 있습니다.", + color=0x9B59B6, + ) + ) + + # 관련 페이지 수집 + all_pages = await wiki.list_pages() + topic_lower = topic.lower() + related = [p for p in all_pages if topic_lower in p.title.lower() or topic_lower in p.path.lower()] + + if not related: + await status_msg.edit(embed=discord.Embed( + title="⚠️ 관련 페이지 없음", + description=f"'{topic}' 관련 위키 페이지를 찾지 못했습니다.", + color=0xF39C12, + )) + return + + # 각 페이지 내용 수집 + contents = [] + for p in related: + full_page = await wiki.get_page(p.id) + contents.append(f"## 페이지: {full_page.title} (/{full_page.path})\n{full_page.content}\n") + + combined = "\n---\n".join(contents) + + await status_msg.edit(embed=discord.Embed( + title="🔄 재정리 중...", + description=f"{len(related)}개 페이지를 Gemini가 통합 정리합니다.", + color=0x9B59B6, + )) + + # Gemini로 통합 정리 + gemini = GeminiCaller(project_path) + organize_prompt = ( + f"다음은 '{topic}' 관련 기존 위키 페이지들입니다.\n" + f"이 내용을 하나의 통합된 위키 페이지로 재정리하세요.\n\n" + f"요구사항:\n" + f"1. 중복 제거, 논리적 구조화\n" + f"2. 누락된 정보가 있으면 웹 검색으로 보완\n" + f"3. 마크다운 형식, 한국어\n\n" + f"=== 기존 페이지들 ===\n{combined[:15000]}" + ) + + try: + content = await gemini.call_agent( + "agent", organize_prompt, cwd=project_path, timeout=300, + ) + except GeminiCallError as e: + await status_msg.edit(embed=discord.Embed( + title="❌ 정리 실패", description=str(e)[:500], color=0xE74C3C, + )) + return + + # 통합 페이지 등록 + from tools.wiki_client import WikiClient + slug = WikiClient.slugify(topic) + path = f"research/{slug}" + page = await wiki.upsert_page(path, f"{topic} (통합)", content, description=f"통합 정리: {topic}") + + await wiki.update_dashboard() + + await status_msg.edit(embed=discord.Embed( + title=f"✅ 위키 정리 완료: {topic}", + description=( + f"📄 [{path}](https://wiki.variet.net/{path})\n" + f"🔗 원본 {len(related)}개 → 통합 1개\n" + f"📝 {len(content)}자" + ), + color=0x2ECC71, + )) async def _anime_schedule(message, pipeline, filter_str): diff --git a/config.py b/config.py index bcbf38e..3b1bd38 100644 --- a/config.py +++ b/config.py @@ -59,3 +59,8 @@ QBIT_PASSWORD: str = os.getenv("QBIT_PASSWORD", "") NAS_ANIME_PATH: str = os.getenv( "NAS_ANIME_PATH", r"\\192.168.10.10\NasData\Video\Animation" ) + +# === Wiki.js === +WIKI_URL: str = os.getenv("WIKI_URL", "https://wiki.variet.net") +WIKI_API_KEY: str = os.getenv("WIKI_API_KEY", "") + diff --git a/prompts/agent.md b/prompts/agent.md index da1d054..6c4ebe6 100644 --- a/prompts/agent.md +++ b/prompts/agent.md @@ -29,12 +29,30 @@ - `gitea_commits`, `gitea_prs`, `gitea_issues`, `gitea_branches` — Git 관리 - `vikunja_tasks`, `vikunja_create_task`, `vikunja_complete_task` — 태스크 관리 -## 복수 작품 처리 방법 +## ⚠️ 복수 작품 처리 — 반드시 전부 완료할 것 -사용자가 "여러 작품 다운로드" 등 복수 작업을 요청하면: -1. 먼저 `anime_nas_list`로 대상 목록을 확인하세요 -2. 각 작품마다 `anime_download`를 **개별 호출**하세요 -3. 진행 상황과 결과를 정리하여 보고하세요 +사용자가 "이번 분기 애니 다운받아줘" 등 **복수 작업**을 요청하면: + +1. `anime_nas_list(current_quarter=True)`로 이번 분기 애니 **전체 목록** 확인 +2. 목록의 **모든 작품**에 대해 `anime_download`를 **하나씩 순서대로 호출** +3. **1개만 하고 멈추지 마세요** — 목록 끝까지 전부 처리해야 합니다 +4. 도중에 개별 실패가 있어도 **다음 작품으로 넘어가세요** +5. 전부 완료한 뒤 결과를 정리하여 보고하세요 + +### 예시 흐름 + +``` +→ anime_nas_list(current_quarter=True) + "5개 애니 확인: A, B, C, D, E" + +→ anime_download("A") → 결과 기록 +→ anime_download("B") → 결과 기록 +→ anime_download("C") → 결과 기록 +→ anime_download("D") → 결과 기록 +→ anime_download("E") → 결과 기록 + +→ 최종 보고: "5개 중 3개 성공, 2개 보류" +``` ## 응답 규칙 diff --git a/prompts/operator.md b/prompts/operator.md index 6bd23d1..027dfda 100644 --- a/prompts/operator.md +++ b/prompts/operator.md @@ -39,6 +39,12 @@ C:\ProgramData\miniforge3\envs\agent_chat\python.exe tools/nyaa_client.py search C:\ProgramData\miniforge3\envs\agent_chat\python.exe tools/qbit_client.py status C:\ProgramData\miniforge3\envs\agent_chat\python.exe tools/qbit_client.py add "magnet:..." --path "경로" C:\ProgramData\miniforge3\envs\agent_chat\python.exe tools/qbit_client.py delete + +# Wiki.js 도구 +C:\ProgramData\miniforge3\envs\agent_chat\python.exe tools/wiki_client.py list [prefix] +C:\ProgramData\miniforge3\envs\agent_chat\python.exe tools/wiki_client.py get +C:\ProgramData\miniforge3\envs\agent_chat\python.exe tools/wiki_client.py create [content] +C:\ProgramData\miniforge3\envs\agent_chat\python.exe tools/wiki_client.py dashboard ``` ## 실행 패턴 예시 diff --git a/tools/anime_pipeline.py b/tools/anime_pipeline.py index c7ce768..06ae448 100644 --- a/tools/anime_pipeline.py +++ b/tools/anime_pipeline.py @@ -1,14 +1,16 @@ """애니메이션 자동화 파이프라인. 전체 흐름: -1. Anissia에서 애니 검색 → 자막 정보 확인 -2. Nyaa.si에서 토렌트 검색 → 제목 매칭 -3. qBittorrent에 magnet 추가 → NAS 경로 지정 -4. 자막 다운로드 → 파일명 매칭 +1. resolve() — 4요소 세트 완성 (NAS폴더, 파일현황, Anissia명, Nyaa명) +2. download() — 완성된 WorkUnit으로 다운로드 실행 +3. batch_download() — resolve() 반복 + download() 실행 """ import asyncio import logging +import math +import re +from collections import Counter from dataclasses import dataclass, field from pathlib import Path from typing import Optional @@ -20,11 +22,43 @@ 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, - fetch_english_title, + fetch_english_title, fetch_title_via_jikan, web_search_anime_title, ) logger = logging.getLogger("variet.tools.pipeline") +VIDEO_EXTS = {".mkv", ".mp4", ".avi", ".webm", ".m4v", ".ts"} +SUB_EXTS = {".ass", ".srt", ".ssa", ".sub", ".smi"} + + +# ────────────────────────────────────────── +# 데이터 모델 +# ────────────────────────────────────────── + +@dataclass +class AnimeWorkUnit: + """다운로드 작업 1건의 완전한 정보 세트. + + resolve()가 이 세트를 완성한 뒤에만 download()가 실행됨. + """ + # NAS 정보 + nas_folder: str = "" # "[26_1분기]장송의프리렌2기" + nas_path: Path = field(default_factory=Path) + existing_videos: list[str] = field(default_factory=list) + existing_subs: list[str] = field(default_factory=list) + existing_eps: set[int] = field(default_factory=set) + release_group: str = "" # "ASW" + release_name: str = "" # "Sousou no Frieren S2" + episode_offset: int = 0 # 절대번호→시즌번호 변환 오프셋 (예: 12) + + # Anissia 정보 + anime: Optional[AnimeInfo] = None + captions: list[CaptionInfo] = field(default_factory=list) + + # Nyaa 검색 정보 + nyaa_keywords: list[str] = field(default_factory=list) + torrents: list[TorrentResult] = field(default_factory=list) + @dataclass class DownloadResult: @@ -36,10 +70,15 @@ class DownloadResult: subtitles: list[SubtitleFile] = field(default_factory=list) nas_folder: str = "" torrent_added: bool = False + torrent_hashes: list[str] = field(default_factory=list) message: str = "" errors: list[str] = field(default_factory=list) +# ────────────────────────────────────────── +# 파이프라인 +# ────────────────────────────────────────── + class AnimePipeline: """애니메이션 다운로드 자동화 파이프라인.""" @@ -54,96 +93,383 @@ class AnimePipeline: from tools.nas_scanner import NasScanner self.nas = NasScanner(self.nas_base) - async def search(self, title: str) -> DownloadResult: - """애니 검색 — 정보 + 자막 + 토렌트 현황 표시. + # 캐시 (세션당 1회 로드) + self._schedule_cache: list[AnimeInfo] | None = None + self._nas_folder_cache: list | None = None - 실제 다운로드 없이 검색 결과만 반환. + # ────────────────────────────────────── + # 1단계: resolve — 4요소 세트 완성 + # ────────────────────────────────────── + + async def resolve(self, title: str) -> AnimeWorkUnit | None: + """애니 제목에서 4요소 세트(NAS/Anissia/Nyaa/파일현황)를 완성합니다. + + Fallback 체인: + 1. Anissia 직접 검색 (compact 매칭 포함) + 2. 웹 검색 — DuckDuckGo 한글+"애니" → 후보 → Anissia 재검색 + 3. NAS 파일명(영문) → Jikan(일본어) → Anissia 재검색 + + Returns: + 완성된 AnimeWorkUnit, 또는 매칭 실패 시 None """ - result = DownloadResult(success=False) + unit = AnimeWorkUnit() - # 1. Anissia에서 검색 + # ── Step 1: NAS 기존 폴더 스캔 (1회) ── + nas_folder_obj = self._find_existing_nas_folder(title) + if nas_folder_obj: + unit.nas_folder = nas_folder_obj.folder_name + unit.nas_path = Path(nas_folder_obj.full_path) + unit.existing_videos = list(nas_folder_obj.video_files) + unit.existing_subs = list(nas_folder_obj.subtitle_files) + # 에피소드 번호 추출 + for vf in unit.existing_videos: + ep = self._extract_episode(vf) + if ep is not None: + unit.existing_eps.add(ep) + # 릴리스 그룹 + 이름 추출 + if unit.existing_videos: + unit.release_name = self._extract_release_name(unit.existing_videos[0]) + groups = [] + for vf in unit.existing_videos: + m = re.match(r'\[([^\]]+)\]', vf) + if m: + groups.append(m.group(1)) + if groups: + unit.release_group = Counter(groups).most_common(1)[0][0] + + if unit.existing_eps: + logger.info( + f"NAS 기존 폴더: {unit.nas_folder} | " + f"영상:{len(unit.existing_videos)} 자막:{len(unit.existing_subs)} " + f"그룹:[{unit.release_group}]" + ) + + # ── Step 2: Anissia 매칭 (fallback 체인) ── + anime = await self._resolve_anissia(title, unit) + if not anime: + logger.warning(f"Anissia 매칭 실패: '{title}'") + return None + unit.anime = anime + + # NAS 폴더 재검증 — start_date로 정확한 시즌 매칭 + if anime.start_date: + correct_folder = self._find_existing_nas_folder( + anime.subject, start_date=anime.start_date, + ) + if correct_folder and correct_folder.folder_name != unit.nas_folder: + logger.info( + f"NAS 폴더 교정: {unit.nas_folder} → {correct_folder.folder_name}" + ) + unit.nas_folder = correct_folder.folder_name + unit.nas_path = Path(correct_folder.full_path) + unit.existing_videos = list(correct_folder.video_files) + unit.existing_subs = list(correct_folder.subtitle_files) + unit.existing_eps = set() + for vf in unit.existing_videos: + ep = self._extract_episode(vf) + if ep is not None: + unit.existing_eps.add(ep) + if unit.existing_videos: + unit.release_name = self._extract_release_name( + unit.existing_videos[0] + ) + groups = [] + for v in unit.existing_videos: + m = re.match(r'\[([^\]]+)\]', v) + if m: + groups.append(m.group(1)) + if groups: + from collections import Counter as _Counter + unit.release_group = _Counter(groups).most_common(1)[0][0] + + # NAS 폴더가 없으면 Anissia 정보로 생성 + if not unit.nas_folder: + unit.nas_folder = make_nas_folder_name(anime.subject, anime.start_date) + unit.nas_path = Path(self.nas_base) / unit.nas_folder + + # ── Step 3: 자막 정보 ── try: - anime_list = await self.anissia.search_anime(title) + unit.captions = await self.anissia.get_captions(anime.anime_no) except Exception as e: - result.errors.append(f"Anissia 검색 오류: {e}") - return result + logger.warning(f"자막 조회 오류: {e}") - if not anime_list: - result.message = f"'{title}' 검색 결과가 없습니다." - return result + # ── Step 4: Nyaa 키워드 + 토렌트 확보 ── + await self._resolve_nyaa(unit) - anime = anime_list[0] # 첫 번째 결과 사용 - result.anime = anime + # ── Step 5: 에피소드 번호 오프셋 감지 ── + self._detect_episode_offset(unit) - # 2. 자막 정보 + logger.info( + f"WorkUnit 완성: {anime.subject} | " + f"NAS:{unit.nas_folder} | sub:{len(unit.captions)} | " + f"tor:{len(unit.torrents)} | offset:{unit.episode_offset}" + ) + return unit + + async def _resolve_anissia( + self, title: str, unit: AnimeWorkUnit + ) -> AnimeInfo | None: + """다단계 fallback으로 Anissia 애니 매칭.""" + + # 1차: Anissia 직접 검색 (compact 매칭 포함) + anime_list = await self._search_anissia(title) + if anime_list: + return anime_list[0] + + # 2차: 웹 검색 — 한글+"애니" → 후보 → Anissia 재검색 + web_candidates = await web_search_anime_title(title) + for candidate in web_candidates: + logger.info(f"웹 검색 후보 → Anissia: '{candidate}'") + anime_list = await self._search_anissia(candidate) + if anime_list: + return anime_list[0] + + # 3차: NAS 파일명(영문) → Jikan(일본어) → Anissia + if unit.release_name: + logger.info(f"NAS 파일명 fallback: '{unit.release_name}'") + jikan = await fetch_title_via_jikan(unit.release_name) + if jikan: + for key in ("japanese", "default"): + alt_title = jikan.get(key, "") + if not alt_title: + continue + logger.info(f"Jikan {key}: {alt_title}") + anime_list = await self._search_anissia(alt_title) + if anime_list: + return anime_list[0] + + return None + + async def _search_anissia(self, keyword: str) -> list[AnimeInfo]: + """Anissia 검색 (에러 무시).""" try: - captions = await self.anissia.get_captions(anime.anime_no) - result.captions = captions - except Exception as e: - result.errors.append(f"자막 조회 오류: {e}") + return await self.anissia.search_anime(keyword) + except Exception: + return [] - # 3. NAS 기존 폴더 확인 → 검색 전략 결정 (방영 시점 기반) - nas_existing = self._find_existing_nas_folder(anime.subject, anime.start_date) - - # 3. Nyaa 토렌트 검색 + async def _resolve_nyaa(self, unit: AnimeWorkUnit): + """Nyaa 토렌트 검색 — 기존 릴리스명 또는 Jikan 제목 사용.""" try: - if nas_existing and nas_existing.video_files: - # ── 기존 파일명에서 릴리스명 추출 → Nyaa 검색 (안전) ── - release_name = self._extract_release_name(nas_existing.video_files[0]) - if release_name: - logger.info(f"NAS 기존 릴리스명: '{release_name}'") - found = await self.nyaa.search(release_name, use_default_suffix=False) - matched = [t for t in found - if self._title_contains_keyword(t.title, [release_name.lower()])] - result.torrents = matched[:30] - if matched: - logger.info(f"NAS 릴리스명 검색 → {len(found)}건 중 {len(matched)}건 매칭") + # 전략 1: 기존 릴리스명이 있으면 그대로 검색 (가장 정확) + if unit.release_name: + logger.info(f"Nyaa 검색 (릴리스명): '{unit.release_name}'") + found = await self.nyaa.search(unit.release_name, use_default_suffix=False) + matched = [t for t in found + if self._title_contains_keyword(t.title, [unit.release_name.lower()])] + if matched: + # ASW/HEVC 우선 정렬 → 최대 50건 + matched.sort(key=lambda t: ( + 0 if '[ASW]' in t.title else 1, + 0 if 'HEVC' in t.title.upper() else 1, + )) + unit.torrents = matched[:50] + unit.nyaa_keywords = [unit.release_name.lower()] + logger.info(f"Nyaa 릴리스명 → {len(found)}건 중 {len(matched)}건 매칭") + return - if not result.torrents: - # ── 신규 애니: Jikan API + ASW HEVC 전략 ── - eng_titles = await fetch_english_title(anime.original_subject) + # 전략 2: Jikan 영어 제목 + ASW HEVC 검색 + if unit.anime: + eng_titles = await fetch_english_title(unit.anime.original_subject) eng_default = eng_titles.get("default", "") eng_english = eng_titles.get("english", "") synonyms = eng_titles.get("synonyms", []) keywords = self._build_match_keywords( - eng_default, eng_english, synonyms, anime.original_subject, + eng_default, eng_english, synonyms, unit.anime.original_subject, ) + unit.nyaa_keywords = keywords logger.info(f"매칭 키워드: {keywords}") - # STEP 1: "ASW HEVC"로 검색 → 키워드로 필터 asw_results = await self.nyaa.search("ASW HEVC", use_default_suffix=False) matched = [t for t in asw_results if self._title_contains_keyword(t.title, keywords)] if matched: - logger.info(f"ASW HEVC 검색 → {len(asw_results)}건 중 {len(matched)}건 매칭") - else: - # ASW 릴리스 없음 — 사용자에게 안내 - result.errors.append( - f"⚠️ ASW HEVC 릴리스가 없습니다.\n" - f"영어 제목: {eng_default or '(조회 실패)'}\n" - f"Nyaa에서 직접 검색해주세요." - ) + logger.info(f"ASW HEVC → {len(asw_results)}건 중 {len(matched)}건 매칭") + unit.torrents = matched[:30] - result.torrents = matched[:30] except Exception as e: - result.errors.append(f"Nyaa 검색 오류: {e}") + logger.warning(f"Nyaa 검색 오류: {e}") - # NAS 폴더: 기존 폴더 있으면 재사용, 없으면 새로 생성 - if nas_existing: - result.nas_folder = nas_existing.folder_name - else: - result.nas_folder = make_nas_folder_name(anime.subject, anime.start_date) + def _detect_episode_offset(self, unit: AnimeWorkUnit): + """Nyaa 토렌트의 에피소드 번호 체계 감지 → 오프셋 설정. - 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}`" + - S2/S3 태그가 있으면 → 시즌 상대 번호 (오프셋 불필요) + - 없으면 → AniList prequel 체인으로 오프셋 계산 + - Nyaa min_ep > offset이면 절대 번호 확정 + """ + if not unit.torrents or not unit.anime: + return + + # ASW 토렌트 중심으로 S-tag 감지 + asw_torrents = [t for t in unit.torrents if '[ASW]' in t.title] + check_torrents = asw_torrents[:10] if asw_torrents else unit.torrents[:10] + + has_season_tag = any( + re.search(r'\bS\d+\b|Season\s*\d+', t.title, re.IGNORECASE) + for t in check_torrents ) + + if has_season_tag: + logger.info("에피소드 번호: 시즌 상대 번호 (S태그 감지)") + return # offset = 0 + + # S태그 없음 → AniList에서 prequel 체인으로 offset 계산 + try: + offset = self._get_anilist_offset(unit.release_name or + unit.anime.original_subject) + if offset <= 0: + return + + # Nyaa ep 번호가 실제로 offset보다 큰지 확인 (절대 번호 확정) + eps = [self._extract_episode(t.title) for t in check_torrents] + eps = [e for e in eps if e is not None] + if not eps: + return + + min_ep = min(eps) + if min_ep > offset: + unit.episode_offset = offset + logger.info( + f"에피소드 번호: 절대 번호 감지 (offset={offset}, " + f"Nyaa min_ep={min_ep})" + ) + else: + logger.info( + f"에피소드 번호: S태그 없지만 시즌 상대 번호 " + f"(offset={offset}, Nyaa min_ep={min_ep})" + ) + except Exception as e: + logger.warning(f"에피소드 오프셋 감지 실패: {e}") + + def _get_anilist_offset(self, title: str) -> int: + """AniList prequel 체인으로 시즌 에피소드 오프셋 계산.""" + import httpx + + search_q = ''' + query ($search: String) { + Page(page: 1, perPage: 10) { + media(search: $search, type: ANIME, format_in: [TV, TV_SHORT]) { + id + title { romaji } + episodes + relations { + edges { + relationType + node { id format } + } + } + } + } + } + ''' + detail_q = ''' + query ($id: Int) { + Media(id: $id, type: ANIME) { + id + title { romaji } + episodes + format + relations { + edges { + relationType + node { id format } + } + } + } + } + ''' + + resp = httpx.post( + 'https://graphql.anilist.co', + json={'query': search_q, 'variables': {'search': title}}, + timeout=15, + ) + results = resp.json().get('data', {}).get('Page', {}).get('media', []) + if not results: + return 0 + + # prequel이 있는 결과 우선 (2기 이상) + media = None + for r in results: + edges = r.get('relations', {}).get('edges', []) + has_prequel = any( + e['relationType'] == 'PREQUEL' + and e['node'].get('format') in ('TV', 'TV_SHORT', None) + for e in edges + ) + if has_prequel: + media = r + break + if not media: + return 0 # prequel 없음 = 1기 또는 단일 시즌 + + # prequel 체인 역추적 + chain_eps = [] + current = media + + while True: + edges = current.get('relations', {}).get('edges', []) + prequels = [ + e for e in edges + if e['relationType'] == 'PREQUEL' + and e['node'].get('format') in ('TV', 'TV_SHORT', None) + ] + if not prequels: + break + + prequel_id = prequels[0]['node']['id'] + resp2 = httpx.post( + 'https://graphql.anilist.co', + json={'query': detail_q, 'variables': {'id': prequel_id}}, + timeout=15, + ) + current = resp2.json().get('data', {}).get('Media') + if not current: + break + chain_eps.append(current.get('episodes', 0) or 0) + + return sum(chain_eps) + + # ────────────────────────────────────── + # 2단계: search — 검색 결과 표시용 (기존 호환) + # ────────────────────────────────────── + + async def search(self, title: str) -> DownloadResult: + """애니 검색 — 정보 + 자막 + 토렌트 현황 표시. + + 실제 다운로드 없이 검색 결과만 반환. + """ + unit = await self.resolve(title) + if not unit: + return DownloadResult( + success=False, + message=f"'{title}' 검색 결과가 없습니다.", + ) + + result = DownloadResult( + success=True, + anime=unit.anime, + captions=unit.captions, + torrents=unit.torrents, + nas_folder=unit.nas_folder, + ) + result.message = ( + f"**{unit.anime.subject}** ({unit.anime.original_subject})\n" + f"자막 제작자: {len(unit.captions)}명 | " + f"토렌트: {len(unit.torrents)}건\n" + f"NAS 폴더: `{unit.nas_folder}`" + ) + if not unit.torrents: + result.errors.append( + f"⚠️ ASW HEVC 릴리스가 없습니다.\n" + f"Nyaa에서 직접 검색해주세요." + ) return result + # ────────────────────────────────────── + # 3단계: download — WorkUnit 기반 다운로드 + # ────────────────────────────────────── + async def download( self, title: str, @@ -154,116 +480,408 @@ class AnimePipeline: Args: title: 한글 제목 - mode: - "auto" — 영상+자막 무조건 다운 (기본) - "sub_required" — 자막 있는 에피소드만 영상 다운 - "sub_only" — 자막만 - "video_only" — 영상만 - episode: 특정 에피소드만 (None이면 빠진 것 전부) + mode: "auto" | "sub_required" | "sub_only" | "video_only" + episode: 특정 에피소드 (None이면 빠진 것 전부) """ - # 먼저 검색 - result = await self.search(title) - if not result.success: - return result + # resolve로 WorkUnit 완성 + unit = await self.resolve(title) + if not unit: + return DownloadResult( + success=False, + message=f"'{title}' 검색 결과가 없습니다.", + ) - anime = result.anime - nas_folder = Path(self.nas_base) / result.nas_folder + return await self._execute_download(unit, mode, episode) + + async def _execute_download( + self, + unit: AnimeWorkUnit, + mode: str = "auto", + episode: Optional[int] = None, + ) -> DownloadResult: + """완성된 WorkUnit으로 다운로드 실행 (추가 검색/스캔 없음).""" + result = DownloadResult( + success=True, + anime=unit.anime, + captions=unit.captions, + torrents=unit.torrents, + nas_folder=unit.nas_folder, + ) + result.message = ( + f"**{unit.anime.subject}** ({unit.anime.original_subject})\n" + f"자막: {len(unit.captions)}명 | 토렌트: {len(unit.torrents)}건" + ) + + nas_folder = unit.nas_path # ── 자막 다운로드 ── if mode in ("auto", "sub_only", "sub_required"): - await self._download_subtitles(result, nas_folder, episode) + await self._download_subtitles(result, unit, episode) # ── 영상 토렌트 추가 ── if mode in ("auto", "video_only"): - await self._add_torrents(result, nas_folder, episode) + await self._add_torrents(result, unit, episode) elif mode == "sub_required": - # 자막이 실제로 다운됐을 때만 영상 추가 if result.subtitles: - await self._add_torrents(result, nas_folder, episode) + await self._add_torrents(result, unit, episode) else: result.errors.append("자막이 없어 영상 다운로드를 보류합니다.") - # 결과 메시지 구성 + # ── 토렌트 완료 대기 + 자막 리네임 + 정리 ── + if result.torrent_hashes: + await self._wait_and_cleanup_torrents(result, nas_folder) + + # ── 항상: 기존+신규 자막을 영상 파일명에 맞게 리네임 ── + try: + self._rename_subtitles_to_match_videos(nas_folder, result, offset=unit.episode_offset) + except Exception as e: + logger.warning(f"자막 리네임 오류: {e}") + + # 결과 메시지 parts = [result.message] if result.subtitles: parts.append(f"\n📝 자막 {len(result.subtitles)}건 다운로드 완료") if result.torrent_added: - parts.append(f"\n🎬 토렌트 추가 완료 → `{nas_folder}`") + parts.append(f"\n🎬 토렌트 다운로드 완료 → `{nas_folder}`") if result.errors: parts.append(f"\n⚠️ 오류: " + "; ".join(result.errors)) - result.message = "\n".join(parts) return result + # ────────────────────────────────────── + # 4단계: batch_download — resolve + 일괄 실행 + # ────────────────────────────────────── + + async def batch_download( + self, + mode: str = "auto", + sub_filter: bool = True, + ) -> list[DownloadResult]: + """이번 분기 애니 일괄 다운로드. + + Args: + mode: "auto" | "sub_only" | "video_only" + sub_filter: True면 자막 없는 애니는 영상만 다운 (video_only 모드) + """ + # 1. NAS 이번 분기 폴더 스캔 + current_folders = self.nas.get_current_quarter_anime() + if not current_folders: + logger.warning("이번 분기 NAS 폴더 없음") + return [] + + logger.info(f"이번 분기 NAS 폴더: {len(current_folders)}개") + + # 2. 캐시 미리 로드 (NAS + Anissia, 병렬 resolve 전) + self._nas_folder_cache = await asyncio.get_event_loop().run_in_executor( + None, self.nas.list_anime_folders + ) + await self.anissia.search_anime("_cache_warmup_") + logger.info(f"캐시 로드 완료: NAS {len(self._nas_folder_cache)}개, Anissia {len(self.anissia._schedule_cache or [])}개") + + # 3. 전체 resolve — 병렬 실행 + async def _safe_resolve(folder): + try: + return await self.resolve(folder.title) + except Exception as e: + logger.error(f"resolve 오류 ({folder.title}): {e}") + return None + + resolve_tasks = [_safe_resolve(f) for f in current_folders] + resolved = await asyncio.gather(*resolve_tasks) + + # 4. 유효한 WorkUnit 필터 + 모드 결정 + units: list[tuple[AnimeWorkUnit, str]] = [] + for folder, unit in zip(current_folders, resolved): + if not unit: + logger.info(f" Anissia 매칭 실패: {folder.title}") + continue + + effective_mode = mode + if sub_filter and not unit.captions: + effective_mode = "video_only" + logger.info(f" 자막 없음 → 영상만: {unit.anime.subject}") + + units.append((unit, effective_mode)) + + logger.info(f"resolve 완료: {len(units)}/{len(current_folders)}개 유효") + + # 5. 일괄 다운로드 + results = [] + for unit, effective_mode in units: + try: + result = await self._execute_download(unit, mode=effective_mode) + results.append(result) + status = "✅" if result.success else "❌" + logger.info(f" {status} {unit.anime.subject}: {result.message[:80]}") + except Exception as e: + logger.error(f" 오류 ({unit.anime.subject}): {e}") + results.append(DownloadResult( + success=False, + anime=unit.anime, + message=f"{unit.anime.subject}: 오류 - {e}", + errors=[str(e)], + )) + + return results + + # ────────────────────────────────────── + # 토렌트 완료 대기 + 정리 + # ────────────────────────────────────── + + async def _wait_and_cleanup_torrents( + self, + result: DownloadResult, + nas_folder: Path, + timeout: int = 600, + poll_interval: int = 10, + ): + """토렌트 다운로드 완료 대기 → 자동 삭제. + + Args: + timeout: 최대 대기 시간 (초, 기본 10분) + poll_interval: 폴링 간격 (초) + """ + if not result.torrent_hashes: + return + + pending = set(result.torrent_hashes) + completed = set() + failed = set() + elapsed = 0 + + logger.info( + f"토렌트 완료 대기 시작: {len(pending)}건, 타임아웃 {timeout}초" + ) + + while pending and elapsed < timeout: + await asyncio.sleep(poll_interval) + elapsed += poll_interval + + for h in list(pending): + try: + status = await self.qbit.get_torrent_status(h) + if status is None: + # 토렌트가 이미 없음 (수동 삭제 등) + pending.discard(h) + continue + + if status.progress >= 1.0: + completed.add(h) + pending.discard(h) + logger.info( + f"토렌트 완료: {status.name[:50]} " + f"({status.size / (1024**2):.0f}MB)" + ) + + elif status.state in ("error", "missingFiles"): + failed.add(h) + pending.discard(h) + result.errors.append( + f"토렌트 실패 ({status.state}): {status.name[:50]}" + ) + except Exception as e: + logger.warning(f"토렌트 상태 확인 실패 ({h[:8]}): {e}") + + if pending: + logger.info( + f"토렌트 대기중: {len(pending)}건 남음 " + f"({elapsed}/{timeout}초)" + ) + + # 타임아웃된 토렌트 + for h in pending: + result.errors.append(f"토렌트 타임아웃: {h[:16]}...") + failed.add(h) + + # 완료된 토렌트 삭제 (파일은 남김) + for h in completed: + try: + await self.qbit.delete_torrent(h, delete_files=False) + logger.info(f"토렌트 삭제: {h[:16]}") + except Exception as e: + logger.warning(f"토렌트 삭제 실패 ({h[:8]}): {e}") + + logger.info( + f"토렌트 정리 완료: {len(completed)}건 성공, " + f"{len(failed)}건 실패" + ) + + # ────────────────────────────────────── + # 자막 다운로드 + # ────────────────────────────────────── + async def _download_subtitles( self, result: DownloadResult, - nas_folder: Path, + unit: AnimeWorkUnit, episode: Optional[int], ): - """자막 다운로드 → 영상 폴더에 직접 저장 + 영상명 매칭 리네임. + """자막 다운로드 → 영상 폴더에 직접 저장. - 기존 자막이 있는 에피소드는 건너뜀 (수동 자막 보호). + Anissia는 최신 에피소드 URL만 제공하므로, + URL 패턴을 분석해 이전 에피소드 URL을 자동 생성. """ - # 영상 폴더에 직접 저장 (subtitles/ 하위 아님) + nas_folder = unit.nas_path nas_folder.mkdir(parents=True, exist_ok=True) - # 기존 자막 파일이 있는 에피소드 스캔 → 스킵 대상 + # WorkUnit에서 기존 자막 에피소드 가져옴 (재스캔 없음) existing_sub_eps = set() - sub_exts = {".ass", ".srt", ".ssa", ".sub", ".smi"} - if nas_folder.exists(): - for f in nas_folder.iterdir(): - if f.suffix.lower() in sub_exts: - ep = self._extract_episode(f.stem) - if ep is not None: - existing_sub_eps.add(ep) + offset = unit.episode_offset + for sf in unit.existing_subs: + ep = self._extract_episode(sf) + if ep is not None: + existing_sub_eps.add(ep) + # offset이 있으면 시즌 상대 번호도 추가 + if offset > 0 and ep > offset: + existing_sub_eps.add(ep - offset) if existing_sub_eps: logger.info(f"기존 자막 에피소드 (스킵): {sorted(existing_sub_eps)}") - for caption in result.captions: + for caption in unit.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 + # 캡션 URL에서 에피소드 패턴 분석 → 이전 에피소드 URL 생성 + urls_to_check = self._discover_episode_urls( + caption.website, caption.episode, episode, + ) - # 기존 자막이 있는 에피소드 스킵 - if sub.episode is not None and sub.episode in existing_sub_eps: - logger.info(f"자막 스킵 (기존 존재): {sub.episode}화 - {sub.filename}") - continue + for url, discovered_ep in urls_to_check: + # 이미 자막 있는 에피소드 → URL 페치 전에 스킵 + if discovered_ep is not None and discovered_ep in existing_sub_eps: + logger.info(f"자막 URL 스킵 (기존 존재): ep{discovered_ep}") + continue - try: - await self.sub_downloader.download_file(sub, str(nas_folder)) - 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}") + try: + subs = await self.sub_downloader.find_subtitles(url) + for sub in subs: + # sub.episode이 None이면 파일명에서 추출 시도 + if sub.episode is None: + sub.episode = self._extract_episode(sub.filename) + # discovered_ep fallback + if sub.episode is None: + sub.episode = discovered_ep - # 다운로드 후: 기존 영상 파일과 매칭하여 자막 리네임 - self._rename_subtitles_to_match_videos(nas_folder, result) + if episode is not None and sub.episode is not None and sub.episode != episode: + continue + if sub.episode is not None and sub.episode in existing_sub_eps: + logger.info(f"자막 스킵 (기존 존재): {sub.episode}화 - {sub.filename}") + continue + + try: + await self.sub_downloader.download_file(sub, str(nas_folder)) + result.subtitles.append(sub) + if sub.episode is not None: + existing_sub_eps.add(sub.episode) + except Exception as e: + result.errors.append(f"자막 다운로드 실패 ({sub.filename}): {e}") + except Exception as e: + # URL이 404일 수 있으므로 debug만 + logger.debug(f"자막 URL 접근 실패: {url} - {e}") + + # 자막 리네임 + self._rename_subtitles_to_match_videos(nas_folder, result, offset=unit.episode_offset) + + def _discover_episode_urls( + self, + base_url: str, + caption_episode: str, + target_episode: Optional[int], + ) -> list[tuple[str, Optional[int]]]: + """캡션 URL에서 같은 애니의 모든 에피소드 자막 URL 자동 검색. + + Returns: [(url, episode_number), ...] + """ + try: + ep_num = int(caption_episode) if caption_episode else None + except (ValueError, TypeError): + ep_num = None + + if target_episode is not None: + return [(base_url, ep_num)] + + # Blogspot인 경우 Atom Feed 사용 + if 'blogspot.com' in base_url: + feed_urls = self._discover_blogspot_episodes(base_url) + if feed_urls: + return feed_urls + + return [(base_url, ep_num)] + + def _discover_blogspot_episodes(self, known_url: str) -> list[tuple[str, Optional[int]]]: + """Blogspot Atom Feed에서 같은 애니의 모든 에피소드 URL 검색. + + Returns: [(url, episode_number), ...] 오래된 에피소드부터 + """ + import httpx + + m = re.match(r'(https?://[^/]+\.blogspot\.com)', known_url) + if not m: + return [] + + blog_base = m.group(1) + feed_url = f"{blog_base}/feeds/posts/default" + + try: + resp = httpx.get( + feed_url, + params={'alt': 'json', 'max-results': 100}, + timeout=15, + ) + if resp.status_code != 200: + return [] + + data = resp.json() + entries = data.get('feed', {}).get('entry', []) + + # known_url 포스트 찾기 → 애니 이름 추출 + anime_name = None + for entry in entries: + links = [l for l in entry.get('link', []) if l.get('rel') == 'alternate'] + entry_url = links[0].get('href', '') if links else '' + if entry_url == known_url: + title = entry.get('title', {}).get('$t', '') + anime_name = re.sub(r'\s*\d+화.*$', '', title).strip() + break + + if not anime_name: + logger.info(f"Atom Feed에서 원본 포스트를 찾지 못함: {known_url}") + return [] + + # 같은 애니 이름의 모든 포스트 (url, episode) 수집 + episode_urls = [] + for entry in entries: + title = entry.get('title', {}).get('$t', '') + if anime_name not in title: + continue + links = [l for l in entry.get('link', []) if l.get('rel') == 'alternate'] + entry_url = links[0].get('href', '') if links else '' + if entry_url: + ep = self._extract_episode(title) + episode_urls.append((entry_url, ep)) + + episode_urls.reverse() # 오래된 순으로 + + logger.info( + f"Blogspot Feed 검색: '{anime_name}' → {len(episode_urls)}건 발견" + ) + return episode_urls + + except Exception as e: + logger.warning(f"Blogspot Feed 검색 실패: {e}") + return [] def _rename_subtitles_to_match_videos( - self, folder: Path, result: DownloadResult + self, folder: Path, result: DownloadResult, + offset: int = 0, ): - """폴더 내 자막 파일을 영상 파일명에 맞게 리네임. - - 예: [ASW] Sousou no Frieren S2 - 03.mkv - → 3화.ass 를 [ASW] Sousou no Frieren S2 - 03.ass 로 변경 - """ - import re as _re - - # 영상 파일 목록 (에피소드 → 파일명) - video_exts = {".mkv", ".mp4", ".avi", ".webm"} + """폴더 내 자막 파일을 영상 파일명에 맞게 리네임.""" videos = {} # episode_num -> video_path for f in folder.iterdir(): - if f.suffix.lower() in video_exts: + if f.suffix.lower() in VIDEO_EXTS: ep = self._extract_episode(f.stem) if ep is not None: videos[ep] = f @@ -271,250 +889,93 @@ class AnimePipeline: if not videos: return - # 자막 파일 리네임 - sub_exts = {".ass", ".srt", ".ssa", ".sub"} for f in folder.iterdir(): - if f.suffix.lower() not in sub_exts: + if f.suffix.lower() not in SUB_EXTS: continue ep = self._extract_episode(f.stem) - if ep is not None and ep in videos: - video_stem = videos[ep].stem - new_name = f"{video_stem}{f.suffix}" - new_path = folder / new_name - if new_path != f and not new_path.exists(): - try: - f.rename(new_path) - logger.info(f"자막 리네임: {f.name} → {new_name}") - except Exception as e: - logger.warning(f"자막 리네임 실패: {e}") - - @staticmethod - def _extract_episode(text: str) -> Optional[int]: - """텍스트에서 에피소드 번호 추출.""" - import re as _re - # 패턴 1: S01E03, S02E07 (SxxExx — 시즌+에피소드, 가장 먼저 체크) - m = _re.search(r'[Ss]\d{1,2}[Ee](\d{1,4})', text) - if m: - return int(m.group(1)) - # 패턴 2: "- 03", "- 06", "- 10v2" (torrent 파일명, v2 등 version suffix 허용) - m = _re.search(r'[-–]\s*(\d{1,4})(?:v\d)?(?:\s|$|\.|\[|\()', text) - if m: - return int(m.group(1)) - # 패턴 3: "3화", "03화" - m = _re.search(r'(\d{1,4})\s*화', text) - if m: - return int(m.group(1)) - # 패턴 4: "EP03", "Episode 3" - m = _re.search(r'(?:EP|Episode)\s*(\d{1,4})', text, _re.IGNORECASE) - if m: - return int(m.group(1)) - return None - - def _find_existing_nas_folder(self, korean_title: str, start_date: str = ""): - """NAS에서 기존 폴더 찾기 — 제목 + 방영 시점(year/quarter) 기반. - - 같은 애니라도 방영 분기가 다르면 다른 시즌 → 매칭하지 않음. - 예: '최애의 아이 3기'(26_1분기) ≠ [23_2분기]최애의아이 (1기) - """ - import re as _re - from tools.title_matcher import get_quarter - - title_norm = _re.sub(r'[^\w]', '', korean_title.lower()) - if len(title_norm) < 2: - return None - - # 방영 분기 계산 - anime_year, anime_quarter = get_quarter(start_date) - - try: - all_folders = self.nas.list_anime_folders() - except Exception as e: - logger.warning(f"NAS 폴더 검색 실패: {e}") - return None - - candidates = [] - for folder in all_folders: - folder_norm = _re.sub(r'[^\w]', '', folder.title.lower()) - # 제목 부분 매칭 (양방향) - if not (title_norm in folder_norm or folder_norm in title_norm): + if ep is None: continue - # 방영 분기 일치 확인 - if anime_year and folder.year != anime_year: + + # 직접 매칭 또는 offset 적용 매칭 + video_ep = None + if ep in videos: + video_ep = ep + elif offset > 0 and (ep + offset) in videos: + video_ep = ep + offset + + if video_ep is None: continue - if anime_quarter and folder.quarter != anime_quarter: - continue - candidates.append(folder) - if not candidates: - return None + video_stem = videos[video_ep].stem + new_name = f"{video_stem}{f.suffix}" + new_path = folder / new_name + if new_path == f: + continue # 이미 정확한 이름 + if new_path.exists(): + # 이미 리네임된 자막 존재 → 중복 원본 삭제 + try: + f.unlink() + logger.info(f"중복 자막 삭제: {f.name} (→ {new_name} 이미 존재)") + except Exception as e: + logger.warning(f"중복 자막 삭제 실패: {e}") + else: + try: + f.rename(new_path) + logger.info(f"자막 리네임: {f.name} → {new_name}") + except Exception as e: + logger.warning(f"자막 리네임 실패: {e}") - best = candidates[0] - logger.info(f"NAS 기존 폴더 발견: {best.folder_name}") - return best - - @staticmethod - def _extract_release_name(filename: str) -> str: - """영상 파일명에서 릴리스 이름 추출. - - [ASW] Hime-sama Goumon no Jikan desu - 21 [1080p HEVC].mkv - → 'Hime-sama Goumon no Jikan desu' - """ - import re as _re - # 확장자 제거 - name = _re.sub(r'\.[^.]+$', '', filename) - # [그룹태그] 제거 - name = _re.sub(r'^\[[^\]]*\]\s*', '', name) - # 에피소드 번호 이후 제거: " - 21 [...]" - name = _re.sub(r'\s*[-–]\s*\d+.*$', '', name).strip() - # S02E09 패턴 제거 - name = _re.sub(r'\s*S\d+E\d+.*$', '', name, flags=_re.IGNORECASE).strip() - return name - - @staticmethod - def _build_match_keywords( - eng_default: str, eng_english: str, - synonyms: list[str], original_title: str, - ) -> list[str]: - """Jikan 제목들에서 매칭용 키워드 추출. - - 예: "Sousou no Frieren 2nd Season" → ["Frieren", "Sousou no Frieren"] - synonyms: ["Omagoto"] → ["Omagoto"] - """ - import re as _re - - keywords = [] - - # synonyms 중 짧은 것 (Omagoto 같은 약칭) - for syn in synonyms: - cleaned = syn.strip() - if 3 <= len(cleaned) <= 30: - keywords.append(cleaned.lower()) - - # eng_default에서 키워드 추출 (시즌 표기 제거) - if eng_default: - clean = _re.sub(r'\s*(2nd|3rd|\d+th)\s*Season.*$', '', eng_default, flags=_re.IGNORECASE).strip() - clean = _re.sub(r'\s*S\d+$', '', clean).strip() - if len(clean) >= 3: - keywords.append(clean.lower()) - - # eng_english에서 콜론 앞 핵심 단어 - if eng_english: - short = eng_english.split(":")[0].strip() - if len(short) >= 3: - keywords.append(short.lower()) - - # 원제 (일본어) — 정규화 없이 원본으로 비교 - if original_title and len(original_title) >= 2: - import re as _re2 - clean = _re2.sub(r'\s*第\d+期$', '', original_title).strip() - if clean: - keywords.append(clean.lower()) - - # 중복 제거 - seen = set() - unique = [] - for k in keywords: - if k not in seen: - seen.add(k) - unique.append(k) - return unique - - @staticmethod - def _title_contains_keyword(nyaa_title: str, keywords: list[str]) -> bool: - """Nyaa 토렌트 제목에 키워드 중 하나라도 포함되는지 체크. - - 영문 키워드: 특수문자(하이픈, 따옴표) 제거 후 비교. - 일본어 키워드: 원본 그대로 비교. - """ - import re as _re - title_lower = nyaa_title.lower() - # 영문 정규화 버전 - title_norm = _re.sub(r'[^a-z0-9\s]', '', title_lower) - - for kw in keywords: - if not kw or len(kw) < 2: - continue - # ASCII만 포함된 키워드 → 정규화 비교 - kw_norm = _re.sub(r'[^a-z0-9\s]', '', kw) - if kw_norm and len(kw_norm) >= 3 and kw_norm in title_norm: - return True - # 비ASCII(일본어 등) → 원본 비교 - if not kw.isascii() and kw in title_lower: - return True - return False + # ────────────────────────────────────── + # 토렌트 추가 + # ────────────────────────────────────── async def _add_torrents( self, result: DownloadResult, - nas_folder: Path, + unit: AnimeWorkUnit, episode: Optional[int], ): - """토렌트 추가 — 빠진 에피소드 전부 다운로드. - - 릴리스 그룹 일관성: NAS 기존 파일의 릴리스 그룹(예: ASW)이 있으면 - 같은 그룹의 토렌트만 추가. 매칭 없으면 스킵. - """ - if not result.torrents: + """토렌트 추가 — WorkUnit의 기존 데이터 사용 (재스캔 없음).""" + if not unit.torrents: result.errors.append("매칭되는 토렌트가 없습니다.") return - import math - import re as _re - - # NAS 기존 에피소드 + 릴리스 그룹 스캔 - existing_eps = set() - existing_groups = [] # 기존 파일들의 릴리스 그룹 - if nas_folder.exists(): - video_exts = {".mkv", ".mp4", ".avi", ".webm", ".ts"} - for f in nas_folder.iterdir(): - if f.suffix.lower() in video_exts: - ep = self._extract_episode(f.stem) - if ep is not None: - existing_eps.add(ep) - # 릴리스 그룹 추출: [ASW], [SubsPlease] 등 - m = _re.match(r'\[([^\]]+)\]', f.name) - if m: - existing_groups.append(m.group(1)) + # WorkUnit에서 기존 정보 사용 + existing_eps = unit.existing_eps + required_group = unit.release_group if len([ + v for v in unit.existing_videos + if re.match(r'\[' + re.escape(unit.release_group) + r'\]', v) + ]) >= 2 else None if unit.release_group else None if existing_eps: logger.info(f"NAS 기존 에피소드: {sorted(existing_eps)}") + if required_group: + logger.info(f"NAS 릴리스 그룹: [{required_group}]") - # 기존 릴리스 그룹 결정 (가장 많이 등장하는 그룹) - required_group = None - if existing_groups: - from collections import Counter - group_counts = Counter(existing_groups) - dominant_group, count = group_counts.most_common(1)[0] - if count >= 2: # 2개 이상 파일에서 동일 그룹이면 확정 - required_group = dominant_group - logger.info(f"NAS 릴리스 그룹: [{required_group}] ({count}개 파일)") - - # 에피소드별 최고 점수 토렌트 그룹핑 - ep_best: dict[int, tuple[int, object]] = {} # ep → (score, torrent) - for t in result.torrents: - ep = self._extract_episode(t.title) - if ep is None: + # 에피소드별 최고 점수 토렌트 + offset = unit.episode_offset + ep_best: dict[int, tuple[int, TorrentResult]] = {} + for t in unit.torrents: + raw_ep = self._extract_episode(t.title) + if raw_ep is None: + continue + # 절대번호 → 시즌번호 변환 (표시용) + season_ep = raw_ep - offset if offset > 0 and raw_ep > offset else raw_ep + if season_ep <= 0: + continue # 이전 시즌 에피소드 + if episode is not None and season_ep != episode: + continue + # existing_eps는 절대번호이므로 raw_ep으로 비교 + if raw_ep in existing_eps: continue - # 특정 에피소드 요청 시 해당 에피소드만 - if episode is not None and ep != episode: - continue - - # NAS 중복 스킵 - if ep in existing_eps: - continue - - # VOSTFR 제외 title_upper = t.title.upper() if "VOSTFR" in title_upper or "VOSTA" in title_upper: continue - # 릴리스 그룹 일관성 필터: NAS에 특정 그룹이 있으면 같은 그룹만 허용 - if required_group: - if f"[{required_group}]" not in t.title: - continue + if required_group and f"[{required_group}]" not in t.title: + continue - # 스코어링 score = 0 if "[ASW]" in t.title: score += 100 @@ -525,8 +986,8 @@ class AnimePipeline: if t.seeders > 0: score += int(math.log(t.seeders) * 5) - if ep not in ep_best or score > ep_best[ep][0]: - ep_best[ep] = (score, t) + if season_ep not in ep_best or score > ep_best[season_ep][0]: + ep_best[season_ep] = (score, t) if not ep_best: if episode is not None: @@ -534,10 +995,10 @@ class AnimePipeline: elif required_group: result.errors.append( f"새로 다운로드할 에피소드가 없습니다 " - f"([{required_group}] 릴리스 기준, 모두 NAS에 존재하거나 미출시)." + f"([{required_group}] 릴리스 기준)." ) else: - result.errors.append("새로 다운로드할 에피소드가 없습니다 (모두 NAS에 존재).") + result.errors.append("새로 다운로드할 에피소드가 없습니다.") return # 에피소드 순서대로 추가 @@ -547,12 +1008,16 @@ class AnimePipeline: try: success = await self.qbit.add_torrent( magnet_or_url=torrent.magnet_link, - save_path=str(nas_folder), + save_path=str(unit.nas_path), category="anime", - tags=result.anime.subject if result.anime else "", + tags=unit.anime.subject if unit.anime else "", ) if success: added_count += 1 + # magnet에서 hash 추출 + h_match = re.search(r'btih:([a-fA-F0-9]{40})', torrent.magnet_link) + if h_match: + result.torrent_hashes.append(h_match.group(1).lower()) logger.info(f"토렌트 추가: ep{ep} - {torrent.title[:50]}") else: result.errors.append(f"ep{ep} 토렌트 추가 실패") @@ -564,47 +1029,9 @@ class AnimePipeline: new_eps = sorted(ep_best.keys()) result.message += f"\n📥 {added_count}개 에피소드 추가: {new_eps}" - @staticmethod - def _select_best_torrent(candidates: list, existing_eps: set = None): - """ASW HEVC 우선으로 최적 토렌트 선택 (검색 결과 표시용). - - 스코어링: - +100 [ASW] 그룹 - +50 HEVC / x265 코덱 - +20 1080p 해상도 - +log(시더수) * 5 (최대 ~30) - VOSTFR / non-English 릴리스는 완전 제외. - 기존 에피소드는 스킵. - """ - import re as _re - import math - - if existing_eps is None: - existing_eps = set() - - scored = [] - for t in candidates: - title_upper = t.title.upper() - if "VOSTFR" in title_upper or "VOSTA" in title_upper: - continue - - score = 0 - if "[ASW]" in t.title: - score += 100 - if "HEVC" in title_upper or "X265" in title_upper: - score += 50 - if "1080P" in title_upper: - score += 20 - if t.seeders > 0: - score += int(math.log(t.seeders) * 5) - - scored.append((score, t)) - - if not scored: - return None - - scored.sort(key=lambda x: x[0], reverse=True) - return scored[0][1] + # ────────────────────────────────────── + # 유틸리티 + # ────────────────────────────────────── async def get_status(self) -> list[dict]: """현재 다운로드 큐 상태.""" @@ -626,87 +1053,129 @@ class AnimePipeline: logger.error(f"qBittorrent 상태 조회 오류: {e}") return [] - async def batch_download( - self, - mode: str = "auto", - sub_filter: bool = True, - ) -> list[DownloadResult]: - """이번 분기 애니 일괄 다운로드. + @staticmethod + def _extract_episode(text: str) -> Optional[int]: + """텍스트에서 에피소드 번호 추출.""" + # SxxExx + m = re.search(r'[Ss]\d{1,2}[Ee](\d{1,4})', text) + if m: + return int(m.group(1)) + # "- 03", "- 06v2" + m = re.search(r'[-–]\s*(\d{1,4})(?:v\d)?(?:\s|$|\.|\[|\()', text) + if m: + return int(m.group(1)) + # "3화" + m = re.search(r'(\d{1,4})\s*화', text) + if m: + return int(m.group(1)) + # EP03 + m = re.search(r'(?:EP|Episode)\s*(\d{1,4})', text, re.IGNORECASE) + if m: + return int(m.group(1)) + return None - Args: - mode: 다운로드 모드 ("auto", "sub_only", "video_only") - sub_filter: True면 Anissia에 자막이 등록된 애니만 처리 + def _find_existing_nas_folder(self, korean_title: str, start_date: str = ""): + """NAS에서 기존 폴더 찾기 (캐시 사용).""" + from tools.title_matcher import get_quarter - Returns: - 각 애니별 DownloadResult 리스트 + title_norm = re.sub(r'[^\w]', '', korean_title.lower()) + if len(title_norm) < 2: + return None + + anime_year, anime_quarter = get_quarter(start_date) + + # 캐시가 있으면 사용, 없으면 직접 로드 + try: + all_folders = self._nas_folder_cache or self.nas.list_anime_folders() + except Exception as e: + logger.warning(f"NAS 폴더 검색 실패: {e}") + return None + + for folder in all_folders: + folder_norm = re.sub(r'[^\w]', '', folder.title.lower()) + if not (title_norm in folder_norm or folder_norm in title_norm): + continue + if anime_year and folder.year != anime_year: + continue + if anime_quarter and folder.quarter != anime_quarter: + continue + logger.info(f"NAS 기존 폴더 발견: {folder.folder_name}") + return folder + + return None + + @staticmethod + def _extract_release_name(filename: str) -> str: + """영상 파일명에서 릴리스 이름 추출. + + [ASW] Sousou no Frieren S2 - 07 [1080p HEVC].mkv → 'Sousou no Frieren S2' """ - # 1. NAS에서 이번 분기 애니 폴더 스캔 - current_folders = self.nas.get_current_quarter_anime() - if not current_folders: - logger.warning("이번 분기 NAS 폴더 없음") - return [] + name = re.sub(r'\.[^.]+$', '', filename) + name = re.sub(r'^\[[^\]]*\]\s*', '', name) + name = re.sub(r'\s*[-–]\s*\d+.*$', '', name).strip() + name = re.sub(r'\s*S\d+E\d+.*$', '', name, flags=re.IGNORECASE).strip() + return name - logger.info(f"이번 분기 NAS 폴더: {len(current_folders)}개") - results = [] + @staticmethod + def _build_match_keywords( + eng_default: str, eng_english: str, + synonyms: list[str], original_title: str, + ) -> list[str]: + """Jikan 제목들에서 매칭용 키워드 추출.""" + keywords = [] - for folder in current_folders: - title = folder.title - logger.info(f"\n{'='*40}") - logger.info(f"처리 중: {folder.folder_name}") + for syn in synonyms: + cleaned = syn.strip() + if 3 <= len(cleaned) <= 30: + keywords.append(cleaned.lower()) - try: - # 2. Anissia에서 검색 → 자막 정보 확인 - anime_list = await self.anissia.search_anime(title) - if not anime_list: - logger.info(f" Anissia 검색 결과 없음 → 건너뜀") - continue + if eng_default: + clean = re.sub(r'\s*(2nd|3rd|\d+th)\s*Season.*$', '', eng_default, flags=re.IGNORECASE).strip() + clean = re.sub(r'\s*S\d+$', '', clean).strip() + if len(clean) >= 3: + keywords.append(clean.lower()) - anime = anime_list[0] + if eng_english: + short = eng_english.split(":")[0].strip() + if len(short) >= 3: + keywords.append(short.lower()) - # 3. 자막 필터: Anissia에 자막 제작자가 있는지 확인 - if sub_filter: - captions = await self.anissia.get_captions(anime.anime_no) - if not captions: - logger.info(f" 자막 없음 → 건너뜀") - continue - logger.info(f" 자막 {len(captions)}건 발견 → 다운로드 진행") + if original_title and len(original_title) >= 2: + clean = re.sub(r'\s*第\d+期$', '', original_title).strip() + if clean: + keywords.append(clean.lower()) - # 4. 기존 에피소드 확인 - from pathlib import Path as _Path - nas_path = _Path(self.nas_base) / folder.folder_name - existing_eps = set() - video_exts = {".mkv", ".mp4", ".avi", ".webm", ".ts"} - if nas_path.exists(): - for f in nas_path.iterdir(): - if f.suffix.lower() in video_exts: - ep = self._extract_episode(f.stem) - if ep is not None: - existing_eps.add(ep) + seen = set() + return [k for k in keywords if k not in seen and not seen.add(k)] - # 5. 다운로드 실행 - result = await self.download(title, mode=mode) - results.append(result) + @staticmethod + def _title_contains_keyword(nyaa_title: str, keywords: list[str]) -> bool: + """Nyaa 토렌트 제목에 키워드 중 하나라도 포함되는지 체크.""" + title_lower = nyaa_title.lower() + title_norm = re.sub(r'[^a-z0-9\s]', '', title_lower) - status = "✅" if result.success else "❌" - logger.info(f" {status} {result.message[:100]}") - - except Exception as e: - logger.error(f" 오류 ({folder.folder_name}): {e}") - err_result = DownloadResult( - success=False, - message=f"{folder.folder_name}: 오류 - {e}", - errors=[str(e)], - ) - results.append(err_result) - - return results + for kw in keywords: + if not kw or len(kw) < 2: + continue + kw_norm = re.sub(r'[^a-z0-9\s]', '', kw) + if kw_norm and len(kw_norm) >= 3 and kw_norm in title_norm: + return True + if not kw.isascii() and kw in title_lower: + return True + return False # ── CLI 진입점 ── if __name__ == "__main__": import sys import asyncio - import json + import logging + + # 로그 출력 설정 + logging.basicConfig( + level=logging.INFO, + format='%(asctime)s [%(levelname)s] %(name)s: %(message)s' + ) args = sys.argv[1:] pipeline = AnimePipeline() @@ -717,7 +1186,6 @@ if __name__ == "__main__": return if args[0] == "search" and len(args) > 1: - # python tools/anime_pipeline.py search "프리렌" title = " ".join(args[1:]) result = await pipeline.search(title) print(result.message) @@ -725,7 +1193,6 @@ if __name__ == "__main__": print(f"⚠️ 오류: {'; '.join(result.errors)}") elif args[0] == "download" and len(args) > 1: - # python tools/anime_pipeline.py download "프리렌" [--mode auto] [--episode 10] title_parts = [] mode = "auto" episode = None @@ -746,7 +1213,6 @@ if __name__ == "__main__": print(result.message) elif args[0] == "batch": - # python tools/anime_pipeline.py batch [--no-sub-filter] [--mode auto] mode = "auto" sub_filter = True i = 1 @@ -772,7 +1238,6 @@ if __name__ == "__main__": print(f" {icon} {title}: {r.message[:80]}") elif args[0] == "status": - # python tools/anime_pipeline.py status status = await pipeline.get_status() if not status: print("🎬 다운로드 중인 항목 없음") @@ -784,4 +1249,3 @@ if __name__ == "__main__": print("사용법: python tools/anime_pipeline.py [search|download|batch|status] [옵션]") asyncio.run(main()) - diff --git a/tools/anissia_client.py b/tools/anissia_client.py index f725929..b1cfade 100644 --- a/tools/anissia_client.py +++ b/tools/anissia_client.py @@ -49,6 +49,7 @@ class AnissiaClient: def __init__(self, timeout: float = 15.0): self._timeout = timeout + self._schedule_cache: list[AnimeInfo] | None = None async def get_schedule(self, week: int) -> list[AnimeInfo]: """요일별 편성표 조회 (week: 0=일 ~ 6=토, 7=기타).""" @@ -110,10 +111,17 @@ class AnissiaClient: ] async def search_anime(self, keyword: str) -> list[AnimeInfo]: - """키워드로 전체 편성표에서 검색 (한글/일어/영문 fuzzy 매칭).""" + """키워드로 전체 편성표에서 검색 (한글/일어/영문 fuzzy 매칭). + + 스케줄은 세션당 1회만 API 호출, 이후 캐시 사용. + """ import re as _re - all_anime = await self.get_all_schedule() + # 캐시 사용 + if self._schedule_cache is None: + self._schedule_cache = await self.get_all_schedule() + logger.info(f"스케줄 캐시 로드: {len(self._schedule_cache)}개") + all_anime = self._schedule_cache keyword_lower = keyword.lower() # 특수문자 제거 버전 (따옴표, 괄호 등) keyword_norm = _re.sub(r'[^\w\s]', '', keyword_lower) @@ -133,9 +141,13 @@ class AnissiaClient: # 특수문자 제거 버전 subj_norm = _re.sub(r'[^\w\s]', '', subj_lower) orig_norm = _re.sub(r'[^\w\s]', '', orig_lower) + # 공백까지 제거 버전 (NAS 폴더명→Anissia 매칭용) + subj_compact = _re.sub(r'\s+', '', subj_norm) + keyword_compact = _re.sub(r'\s+', '', keyword_norm) - # 1차: substring 매칭 (원본 + 정규화) - if (keyword_lower in subj_lower or keyword_norm in subj_norm): + # 1차: substring 매칭 (원본 + 정규화 + 공백제거) + if (keyword_lower in subj_lower or keyword_norm in subj_norm + or keyword_compact in subj_compact): results.append(a) elif (keyword_lower in orig_lower or keyword_norm in orig_norm): results.append(a) diff --git a/tools/subtitle_downloader.py b/tools/subtitle_downloader.py index 6cc99a8..e39aa35 100644 --- a/tools/subtitle_downloader.py +++ b/tools/subtitle_downloader.py @@ -218,7 +218,7 @@ class SubtitleDownloader: seen_urls = {r.download_url for r in results} # HTML <a href="..."> - for gurl in re.findall(r'href="([^"]+\.(?:ass|srt|ssa|sub|zip|7z)(?:\?[^"]*)?)"', html, re.IGNORECASE): + for gurl in re.findall(r'href="([^"]+\.(?:ass|srt|ssa|sub|smi|zip|7z)(?:\?[^"]*)?)"', html, re.IGNORECASE): if gurl not in seen_urls: seen_urls.add(gurl) filename = unquote(gurl.split("/")[-1].split("?")[0]) @@ -230,7 +230,7 @@ class SubtitleDownloader: )) # 마크다운 [텍스트](url) — Blogspot 등 - for text, gurl in re.findall(r'\[([^\]]+)\]\((https?://[^)]+\.(?:ass|srt|ssa|sub|zip|7z)[^)]*)\)', html, re.IGNORECASE): + for text, gurl in re.findall(r'\[([^\]]+)\]\((https?://[^)]+\.(?:ass|srt|ssa|sub|smi|zip|7z)[^)]*)\)', html, re.IGNORECASE): if gurl not in seen_urls: seen_urls.add(gurl) results.append(SubtitleFile( @@ -307,7 +307,7 @@ class SubtitleDownloader: if name.endswith("/"): continue ext = Path(name).suffix.lower() - if ext in (".ass", ".srt", ".ssa", ".sub"): + if ext in (".ass", ".srt", ".ssa", ".sub", ".smi"): # 중첩 폴더 무시, 파일만 추출 out_name = Path(name).name out_path = target_dir / out_name diff --git a/tools/title_matcher.py b/tools/title_matcher.py index a99c478..ae318d1 100644 --- a/tools/title_matcher.py +++ b/tools/title_matcher.py @@ -19,6 +19,79 @@ logger = logging.getLogger("variet.tools.matcher") # 영어 제목 조회 (Jikan API / MyAnimeList) # ────────────────────────────────────────────── +async def web_search_anime_title(query: str) -> list[str]: + """웹 검색으로 애니 제목 후보를 찾습니다. + + 압축된 한글 제목(예: '너따위가마왕을이길수있다고생각하지마')으로 + 검색하여 정확한 애니 제목 후보들을 반환합니다. + + Returns: + 검색 결과에서 추출한 제목 후보 리스트 (최대 5개) + """ + import re as _re + + search_query = f"{query} 애니" + candidates = [] + + try: + # DuckDuckGo HTML 검색 (API 키 불필요) + async with httpx.AsyncClient(timeout=10, follow_redirects=True) as client: + resp = await client.get( + "https://html.duckduckgo.com/html/", + params={"q": search_query}, + headers={"User-Agent": "Mozilla/5.0"}, + ) + resp.raise_for_status() + html = resp.text + + from html import unescape as _unescape + + # 검색 결과 제목/스니펫에서 한글 애니 제목 추출 + title_matches = _re.findall( + r'class="result__a"[^>]*>([^<]+)</a>', html + ) + snippet_matches = _re.findall( + r'class="result__snippet"[^>]*>(.*?)</a>', html, _re.DOTALL + ) + + all_text = " ".join(title_matches + snippet_matches) + all_text = _re.sub(r'<[^>]+>', ' ', all_text) + all_text = _unescape(all_text) # " → " 등 HTML 엔티티 변환 + + def _clean_candidate(text: str) -> str: + """후보 텍스트 정리: HTML 엔티티, 말줄임, 부가 정보 제거.""" + text = _unescape(text) + text = _re.sub(r'<[^>]+>', '', text) + text = _re.sub(r'\s*\.{2,}$', '', text) # 말줄임 제거 + text = _re.sub(r'\s*\d+월\s*\d+일.*$', '', text) # 날짜 이후 제거 + text = _re.sub(r'\s*\d+화.*$', '', text) # "N화" 이후 제거 + text = _re.sub(r'\s*[-–—]\s*(나무위키|위키백과|namu\.wiki|Wikipedia|onnada).*$', '', text) + text = text.strip().strip('"\'「」『』') + return text.strip() + + # 「」『』"" 안의 제목 추출 + quoted = _re.findall(r'[「『"]([\w\s~·!?,가-힣]+?)[」』"]', all_text) + for q in quoted: + clean = _clean_candidate(q) + if len(clean) >= 4 and any('\uAC00' <= c <= '\uD7A3' for c in clean): + if clean not in candidates: + candidates.append(clean) + + # 검색 결과 제목에서 사이트명 제거 + for t in title_matches[:5]: + clean = _clean_candidate(t) + if len(clean) >= 4 and any('\uAC00' <= c <= '\uD7A3' for c in clean): + if clean not in candidates: + candidates.append(clean) + + logger.info(f"웹 검색 '{search_query}' → {len(candidates)}개 후보: {candidates[:3]}") + return candidates[:5] + + except Exception as e: + logger.warning(f"웹 검색 실패: {e}") + return [] + + async def fetch_english_title(japanese_title: str) -> dict[str, str]: """Jikan API로 일본어 원제의 영어/로마자 제목 조회. @@ -28,11 +101,27 @@ async def fetch_english_title(japanese_title: str) -> dict[str, str]: "synonyms": ["Frieren at the Funeral Season 2"]} 실패 시 빈 dict. """ + return await fetch_title_via_jikan(japanese_title) + + +async def fetch_title_via_jikan(query: str) -> dict[str, str]: + """Jikan API로 제목 조회 — 한글/일본어/영어 어떤 검색어든 가능. + + NAS 폴더명(한글 압축) → Jikan 검색 → 정확한 제목들을 반환. + Anissia 직접 검색 실패 시 fallback으로 사용. + + Returns: + {"default": "로마자/영어 제목", + "english": "영어 제목", + "japanese": "일본어 제목", + "synonyms": ["동의어1", ...]} + 실패 시 빈 dict. + """ try: async with httpx.AsyncClient(timeout=10) as client: resp = await client.get( "https://api.jikan.moe/v4/anime", - params={"q": japanese_title, "limit": 5}, + params={"q": query, "limit": 5}, ) resp.raise_for_status() data = resp.json() @@ -41,22 +130,36 @@ async def fetch_english_title(japanese_title: str) -> dict[str, str]: if not items: return {} - # 원제와 가장 잘 매칭되는 항목 선택 + # 검색어와 가장 잘 매칭되는 항목 선택 + # 한글 검색이면 title_japanese와 비교, 그 외에는 title과 비교 best = None best_score = 0.0 for item in items: - jp = item.get("title_japanese", "") - score = SequenceMatcher(None, japanese_title, jp).ratio() + # 여러 제목 필드와 비교하여 최고 유사도 채택 + candidates = [ + item.get("title_japanese", ""), + item.get("title", ""), + item.get("title_english", "") or "", + ] + for t in item.get("titles", []): + candidates.append(t.get("title", "")) + + score = max( + SequenceMatcher(None, query, c).ratio() + for c in candidates if c + ) if candidates else 0.0 + if score > best_score: best_score = score best = item - if not best or best_score < 0.5: + if not best or best_score < 0.3: return {} result = { "default": best.get("title", ""), "english": best.get("title_english") or "", + "japanese": best.get("title_japanese") or "", "synonyms": [], } for t in best.get("titles", []): @@ -64,8 +167,8 @@ async def fetch_english_title(japanese_title: str) -> dict[str, str]: result["synonyms"].append(t["title"]) logger.info( - f"Jikan 영어 제목 조회: {japanese_title} → " - f"default={result['default']}, english={result['english']}" + f"Jikan 제목 조회: '{query}' → " + f"default={result['default']}, jp={result['japanese']}" ) return result