Files
variet-agent/tools/anissia_client.py

212 lines
7.4 KiB
Python

"""Anissia API 클라이언트 — 애니 편성표 + 자막 정보 조회.
API Base: https://api.anissia.net
"""
import httpx
import logging
from dataclasses import dataclass, field
from typing import Optional
logger = logging.getLogger("variet.tools.anissia")
BASE_URL = "https://api.anissia.net"
WEEK_NAMES = {
0: "", 1: "", 2: "", 3: "",
4: "", 5: "", 6: "", 7: "기타",
}
@dataclass
class CaptionInfo:
"""자막 제작 정보."""
episode: str
name: str # 제작자 이름
website: str # 제작자 사이트 URL
updated: str # 업데이트 시각
@dataclass
class AnimeInfo:
"""애니메이션 정보."""
anime_no: int
subject: str # 한글 제목
original_subject: str # 원어 제목 (일어)
genres: str
week: int
time: str
status: str # ON / OFF
caption_count: int
start_date: str
end_date: str
website: str
twitter: str
class AnissiaClient:
"""Anissia REST API 클라이언트."""
def __init__(self, timeout: float = 15.0):
self._timeout = timeout
async def get_schedule(self, week: int) -> list[AnimeInfo]:
"""요일별 편성표 조회 (week: 0=일 ~ 6=토, 7=기타)."""
async with httpx.AsyncClient(timeout=self._timeout) as client:
resp = await client.get(f"{BASE_URL}/anime/schedule/{week}")
resp.raise_for_status()
data = resp.json()
if data.get("code") != "ok":
raise RuntimeError(f"Anissia API 오류: {data}")
return [
AnimeInfo(
anime_no=item["animeNo"],
subject=item["subject"],
original_subject=item.get("originalSubject", ""),
genres=item.get("genres", ""),
week=item.get("week", week) if isinstance(item.get("week"), int) else int(item.get("week", week)),
time=item.get("time", ""),
status=item.get("status", ""),
caption_count=item.get("captionCount", 0),
start_date=item.get("startDate", ""),
end_date=item.get("endDate", ""),
website=item.get("website", ""),
twitter=item.get("twitter", ""),
)
for item in data["data"]
]
async def get_all_schedule(self) -> list[AnimeInfo]:
"""전체 요일 편성표 조회 (0~7)."""
all_anime = []
for week in range(8):
try:
schedule = await self.get_schedule(week)
all_anime.extend(schedule)
except Exception as e:
logger.warning(f"편성표 조회 실패 (week={week}): {e}")
return all_anime
async def get_captions(self, anime_no: int) -> list[CaptionInfo]:
"""특정 애니 자막 목록 조회."""
async with httpx.AsyncClient(timeout=self._timeout) as client:
resp = await client.get(f"{BASE_URL}/anime/caption/animeNo/{anime_no}")
resp.raise_for_status()
data = resp.json()
if data.get("code") != "ok":
raise RuntimeError(f"Anissia caption API 오류: {data}")
return [
CaptionInfo(
episode=item.get("episode", ""),
name=item.get("name", ""),
website=item.get("website", ""),
updated=item.get("updDt", ""),
)
for item in data["data"]
]
async def search_anime(self, keyword: str) -> list[AnimeInfo]:
"""키워드로 전체 편성표에서 검색 (한글/일어/영문 fuzzy 매칭)."""
import re as _re
all_anime = await self.get_all_schedule()
keyword_lower = keyword.lower()
# 특수문자 제거 버전 (따옴표, 괄호 등)
keyword_norm = _re.sub(r'[^\w\s]', '', keyword_lower)
try:
from tools.title_matcher import japanese_to_romaji, title_similarity
use_romaji = True
except ImportError:
use_romaji = False
results = []
fuzzy_candidates = []
for a in all_anime:
subj_lower = a.subject.lower()
orig_lower = a.original_subject.lower()
# 특수문자 제거 버전
subj_norm = _re.sub(r'[^\w\s]', '', subj_lower)
orig_norm = _re.sub(r'[^\w\s]', '', orig_lower)
# 1차: substring 매칭 (원본 + 정규화)
if (keyword_lower in subj_lower or keyword_norm in subj_norm):
results.append(a)
elif (keyword_lower in orig_lower or keyword_norm in orig_norm):
results.append(a)
elif use_romaji:
romaji = japanese_to_romaji(a.original_subject).lower()
# 2차: romaji substring
if keyword_lower in romaji:
results.append(a)
else:
# 3차: 단어 단위 fuzzy — 검색어와 romaji 개별 단어 비교
words = romaji.split()
best_word_sim = max(
(title_similarity(keyword, w) for w in words),
default=0.0,
)
# 전체 문자열 유사도도 참고
full_sim = title_similarity(keyword, romaji)
best_sim = max(best_word_sim, full_sim)
if best_sim >= 0.6:
fuzzy_candidates.append((best_sim, a))
# exact 결과가 없을 때만 fuzzy 결과 사용
if not results and fuzzy_candidates:
fuzzy_candidates.sort(key=lambda x: x[0], reverse=True)
results = [a for _, a in fuzzy_candidates[:10]]
return results
# ── CLI 진입점 ──
if __name__ == "__main__":
import sys
import asyncio
client = AnissiaClient()
args = sys.argv[1:]
async def main():
if not args:
print("사용법: python tools/anissia_client.py [schedule|search|captions] [인자]")
return
if args[0] == "schedule":
# python tools/anissia_client.py schedule 3 (수요일)
week = int(args[1]) if len(args) > 1 else 0
anime_list = await client.get_schedule(week)
day = WEEK_NAMES.get(week, "?")
print(f"📺 {day}요일 편성표 ({len(anime_list)}개):")
for a in anime_list:
cap = f"자막 {a.caption_count}" if a.caption_count else "자막 없음"
print(f" {a.time} {a.subject} ({a.original_subject}) [{cap}]")
elif args[0] == "search" and len(args) > 1:
# python tools/anissia_client.py search "프리렌"
keyword = " ".join(args[1:])
results = await client.search_anime(keyword)
print(f"🔍 '{keyword}' 검색 결과 ({len(results)}개):")
for a in results:
print(f" [{a.anime_no}] {a.subject} ({a.original_subject}) | {WEEK_NAMES.get(a.week, '?')} {a.time}")
elif args[0] == "captions" and len(args) > 1:
# python tools/anissia_client.py captions 12345
anime_no = int(args[1])
captions = await client.get_captions(anime_no)
print(f"📝 자막 목록 ({len(captions)}건):")
for c in captions:
print(f" {c.episode}화 | {c.name} | {c.website} | {c.updated}")
else:
print("사용법: python tools/anissia_client.py [schedule|search|captions] [인자]")
asyncio.run(main())