225 lines
7.9 KiB
Python
225 lines
7.9 KiB
Python
"""MCP 서버 — 애니메이션 도구.
|
|
|
|
Gemini CLI에서 MCP로 연결하여 애니 검색/다운로드/편성표 조회를 수행합니다.
|
|
stdio 트랜스포트를 사용합니다.
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# 프로젝트 루트를 sys.path에 추가 (tools/, config 접근용)
|
|
PROJECT_ROOT = Path(__file__).parent.parent
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
import config # noqa: E402 — .env 로드
|
|
|
|
from mcp.server.fastmcp import FastMCP # noqa: E402
|
|
|
|
mcp = FastMCP("anime")
|
|
|
|
|
|
# ──────────────────────────────────────────
|
|
# 애니 검색
|
|
# ──────────────────────────────────────────
|
|
|
|
@mcp.tool()
|
|
async def anime_search(title: str) -> str:
|
|
"""애니메이션을 검색합니다.
|
|
|
|
제목으로 Anissia에서 애니를 검색하고, 자막 제작자 정보와
|
|
Nyaa 토렌트 검색 결과를 함께 반환합니다.
|
|
|
|
Args:
|
|
title: 검색할 애니 제목 (한글 또는 일본어)
|
|
"""
|
|
from tools.anime_pipeline import AnimePipeline
|
|
|
|
pipeline = AnimePipeline()
|
|
result = await pipeline.search(title)
|
|
|
|
if not result.success:
|
|
errors = "; ".join(result.errors) if result.errors else ""
|
|
return f"검색 실패: {result.message} {errors}"
|
|
|
|
anime = result.anime
|
|
lines = [
|
|
f"## {anime.subject} ({anime.original_subject})",
|
|
f"- 장르: {anime.genres}",
|
|
]
|
|
|
|
week_names = ['일', '월', '화', '수', '목', '금', '토', '기타']
|
|
if anime.week is not None:
|
|
lines.append(f"- 편성: {week_names[anime.week]}요일 {anime.time}")
|
|
lines.append(f"- NAS 폴더: {result.nas_folder}")
|
|
|
|
if result.captions:
|
|
lines.append(f"\n### 자막 ({len(result.captions)}명)")
|
|
for c in result.captions[:5]:
|
|
url = f" ({c.website})" if c.website else ""
|
|
lines.append(f" - {c.name} — {c.episode}화{url}")
|
|
|
|
if result.torrents:
|
|
lines.append(f"\n### 토렌트 ({len(result.torrents)}건)")
|
|
for t in result.torrents[:5]:
|
|
ep = f"{t.episode}화 " if t.episode else ""
|
|
lines.append(f" - [{t.group}] {ep}{t.size} (시드: {t.seeders})")
|
|
|
|
if result.errors:
|
|
lines.append(f"\n### 오류")
|
|
for e in result.errors:
|
|
lines.append(f" - {e}")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ──────────────────────────────────────────
|
|
# 애니 다운로드
|
|
# ──────────────────────────────────────────
|
|
|
|
@mcp.tool()
|
|
async def anime_download(
|
|
title: str,
|
|
mode: str = "auto",
|
|
episode: int | None = None,
|
|
) -> str:
|
|
"""애니메이션을 다운로드합니다 (자막+영상).
|
|
|
|
Anissia에서 검색 후 자막과 토렌트를 NAS에 다운로드합니다.
|
|
|
|
Args:
|
|
title: 다운로드할 애니 제목
|
|
mode: "auto" (자막+영상), "sub_only" (자막만), "video_only" (영상만)
|
|
episode: 특정 에피소드 번호 (None이면 최신)
|
|
"""
|
|
from tools.anime_pipeline import AnimePipeline
|
|
|
|
pipeline = AnimePipeline()
|
|
result = await pipeline.download(title, mode=mode, episode=episode)
|
|
return result.message
|
|
|
|
|
|
# ──────────────────────────────────────────
|
|
# 편성표 조회
|
|
# ──────────────────────────────────────────
|
|
|
|
@mcp.tool()
|
|
async def anime_schedule(weekday: int | None = None, sub_only: bool = False) -> str:
|
|
"""애니 편성표를 조회합니다.
|
|
|
|
Args:
|
|
weekday: 요일 번호 (0=일, 1=월, ..., 6=토). None이면 전체 방영 중 목록.
|
|
sub_only: True이면 자막 있는 애니만 표시
|
|
"""
|
|
from tools.anissia_client import AnissiaClient
|
|
|
|
client = AnissiaClient()
|
|
|
|
if weekday is not None:
|
|
schedule = await client.get_schedule(weekday)
|
|
week_names = ['일', '월', '화', '수', '목', '금', '토']
|
|
title = f"{week_names[weekday]}요일 편성표"
|
|
else:
|
|
schedule = await client.get_all_schedule()
|
|
schedule = [a for a in schedule if a.status == "ON"]
|
|
title = "현재 방영 중인 애니"
|
|
|
|
if sub_only:
|
|
schedule = [a for a in schedule if a.caption_count > 0]
|
|
title += " (자막 있음)"
|
|
|
|
lines = [f"## {title} ({len(schedule)}개)"]
|
|
for a in schedule[:30]:
|
|
sub = f" 📝{a.caption_count}" if a.caption_count > 0 else ""
|
|
lines.append(f"- {a.subject} — {a.time}{sub}")
|
|
|
|
if len(schedule) > 30:
|
|
lines.append(f"... 외 {len(schedule) - 30}개")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ──────────────────────────────────────────
|
|
# 다운로드 상태
|
|
# ──────────────────────────────────────────
|
|
|
|
@mcp.tool()
|
|
async def anime_download_status() -> str:
|
|
"""qBittorrent의 현재 다운로드 상태를 확인합니다."""
|
|
from tools.anime_pipeline import AnimePipeline
|
|
|
|
pipeline = AnimePipeline()
|
|
conn = await pipeline.qbit.test_connection()
|
|
|
|
if not conn.get("connected"):
|
|
return f"qBittorrent 연결 실패: {conn.get('error', '알 수 없는 오류')}"
|
|
|
|
torrents = await pipeline.get_status()
|
|
|
|
if not torrents:
|
|
return "다운로드 중인 항목이 없습니다."
|
|
|
|
lines = [f"## 다운로드 큐 ({len(torrents)}건)"]
|
|
for t in torrents[:15]:
|
|
icon = "✅" if t["progress"] == "100.0%" else "⏳"
|
|
lines.append(
|
|
f"- {icon} {t['name'][:50]} — {t['progress']} "
|
|
f"({t['speed']}, ETA: {t['eta']})"
|
|
)
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ──────────────────────────────────────────
|
|
# NAS 애니 목록
|
|
# ──────────────────────────────────────────
|
|
|
|
@mcp.tool()
|
|
async def anime_nas_list(current_quarter: bool = False, keyword: str = "") -> str:
|
|
"""NAS에 다운로드된 애니 목록을 조회합니다.
|
|
|
|
Args:
|
|
current_quarter: True이면 이번 분기 애니만 표시
|
|
keyword: 검색 키워드 (빈 문자열이면 전체)
|
|
"""
|
|
from tools.nas_scanner import NasScanner
|
|
|
|
scanner = NasScanner()
|
|
if not scanner.is_accessible():
|
|
return f"NAS 접근 불가: {scanner.base_path}"
|
|
|
|
if keyword:
|
|
folders = scanner.search(keyword)
|
|
elif current_quarter:
|
|
folders = scanner.get_current_quarter_anime()
|
|
else:
|
|
folders = scanner.list_anime_folders()
|
|
|
|
if not folders:
|
|
return "조건에 맞는 다운로드된 애니가 없습니다."
|
|
|
|
total_vids = sum(f.video_count for f in folders)
|
|
total_subs = sum(f.subtitle_count for f in folders)
|
|
total_size = sum(f.total_size_gb for f in folders)
|
|
|
|
lines = [f"## NAS 애니 목록 ({len(folders)}개)"]
|
|
for f in folders[:25]:
|
|
sub = f" 📝{f.subtitle_count}" if f.subtitle_count > 0 else ""
|
|
lines.append(
|
|
f"- {f.title} — 🎬{f.video_count}화{sub} ({f.total_size_gb:.1f}GB)"
|
|
)
|
|
|
|
if len(folders) > 25:
|
|
lines.append(f"... 외 {len(folders) - 25}개")
|
|
|
|
lines.append(f"\n총 {total_vids}개 영상 | {total_subs}개 자막 | {total_size:.1f}GB")
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ──────────────────────────────────────────
|
|
# 실행
|
|
# ──────────────────────────────────────────
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run(transport="stdio")
|