feat(anime): 자막/토렌트 파이프라인 대폭 개선
- Blogspot Atom Feed API로 전체 에피소드 자막 URL 발견 - AniList prequel 체인 기반 시즌 에피소드 오프셋 자동 감지 - Nyaa S-tag 감지 → 절대/시즌 번호 체계 자동 판별 - 기존 자막 에피소드 스킵 (URL 페치 전 pre-skip) - 오프셋 적용 자막 리네임 (시즌번호→절대번호 매칭) - ASW HEVC 토렌트 우선 정렬 (truncation 방지) - 토렌트 완료 대기 → 자동 삭제 라이프사이클 - 중복 자막 자동 삭제 - .smi 자막 확장자 지원
This commit is contained in:
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user