CD
3618387b8e
- Blogspot Atom Feed API로 전체 에피소드 자막 URL 발견 - AniList prequel 체인 기반 시즌 에피소드 오프셋 자동 감지 - Nyaa S-tag 감지 → 절대/시즌 번호 체계 자동 판별 - 기존 자막 에피소드 스킵 (URL 페치 전 pre-skip) - 오프셋 적용 자막 리네임 (시즌번호→절대번호 매칭) - ASW HEVC 토렌트 우선 정렬 (truncation 방지) - 토렌트 완료 대기 → 자동 삭제 라이프사이클 - 중복 자막 자동 삭제 - .smi 자막 확장자 지원
398 lines
14 KiB
Python
398 lines
14 KiB
Python
"""제목 매칭 + NAS 폴더명 생성.
|
|
|
|
Anissia(한글) ↔ Nyaa(영어/로마자) 제목을 매칭하고,
|
|
NAS 저장 폴더명을 [yy_q분기]제목 형식으로 생성합니다.
|
|
"""
|
|
|
|
import re
|
|
import logging
|
|
import unicodedata
|
|
from difflib import SequenceMatcher
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
|
|
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로 일본어 원제의 영어/로마자 제목 조회.
|
|
|
|
Returns:
|
|
{"default": "Sousou no Frieren 2nd Season",
|
|
"english": "Frieren: Beyond Journey's End Season 2",
|
|
"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": query, "limit": 5},
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
items = data.get("data", [])
|
|
if not items:
|
|
return {}
|
|
|
|
# 검색어와 가장 잘 매칭되는 항목 선택
|
|
# 한글 검색이면 title_japanese와 비교, 그 외에는 title과 비교
|
|
best = None
|
|
best_score = 0.0
|
|
for item in items:
|
|
# 여러 제목 필드와 비교하여 최고 유사도 채택
|
|
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.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", []):
|
|
if t["type"] == "Synonym":
|
|
result["synonyms"].append(t["title"])
|
|
|
|
logger.info(
|
|
f"Jikan 제목 조회: '{query}' → "
|
|
f"default={result['default']}, jp={result['japanese']}"
|
|
)
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Jikan API 조회 실패: {e}")
|
|
return {}
|
|
|
|
|
|
# ──────────────────────────────────────────────
|
|
# 일어 → 로마자 변환 테이블 (히라가나/카타카나)
|
|
# ──────────────────────────────────────────────
|
|
|
|
_KANA_ROMAJI = {
|
|
# 히라가나
|
|
'あ': 'a', 'い': 'i', 'う': 'u', 'え': 'e', 'お': 'o',
|
|
'か': 'ka', 'き': 'ki', 'く': 'ku', 'け': 'ke', 'こ': 'ko',
|
|
'さ': 'sa', 'し': 'shi', 'す': 'su', 'せ': 'se', 'そ': 'so',
|
|
'た': 'ta', 'ち': 'chi', 'つ': 'tsu', 'て': 'te', 'と': 'to',
|
|
'な': 'na', 'に': 'ni', 'ぬ': 'nu', 'ね': 'ne', 'の': 'no',
|
|
'は': 'ha', 'ひ': 'hi', 'ふ': 'fu', 'へ': 'he', 'ほ': 'ho',
|
|
'ま': 'ma', 'み': 'mi', 'む': 'mu', 'め': 'me', 'も': 'mo',
|
|
'や': 'ya', 'ゆ': 'yu', 'よ': 'yo',
|
|
'ら': 'ra', 'り': 'ri', 'る': 'ru', 'れ': 're', 'ろ': 'ro',
|
|
'わ': 'wa', 'を': 'wo', 'ん': 'n',
|
|
'が': 'ga', 'ぎ': 'gi', 'ぐ': 'gu', 'げ': 'ge', 'ご': 'go',
|
|
'ざ': 'za', 'じ': 'ji', 'ず': 'zu', 'ぜ': 'ze', 'ぞ': 'zo',
|
|
'だ': 'da', 'ぢ': 'di', 'づ': 'du', 'で': 'de', 'ど': 'do',
|
|
'ば': 'ba', 'び': 'bi', 'ぶ': 'bu', 'べ': 'be', 'ぼ': 'bo',
|
|
'ぱ': 'pa', 'ぴ': 'pi', 'ぷ': 'pu', 'ぺ': 'pe', 'ぽ': 'po',
|
|
'きゃ': 'kya', 'きゅ': 'kyu', 'きょ': 'kyo',
|
|
'しゃ': 'sha', 'しゅ': 'shu', 'しょ': 'sho',
|
|
'ちゃ': 'cha', 'ちゅ': 'chu', 'ちょ': 'cho',
|
|
'にゃ': 'nya', 'にゅ': 'nyu', 'にょ': 'nyo',
|
|
'ひゃ': 'hya', 'ひゅ': 'hyu', 'ひょ': 'hyo',
|
|
'みゃ': 'mya', 'みゅ': 'myu', 'みょ': 'myo',
|
|
'りゃ': 'rya', 'りゅ': 'ryu', 'りょ': 'ryo',
|
|
'ぎゃ': 'gya', 'ぎゅ': 'gyu', 'ぎょ': 'gyo',
|
|
'じゃ': 'ja', 'じゅ': 'ju', 'じょ': 'jo',
|
|
'びゃ': 'bya', 'びゅ': 'byu', 'びょ': 'byo',
|
|
'ぴゃ': 'pya', 'ぴゅ': 'pyu', 'ぴょ': 'pyo',
|
|
'っ': '', # 촉음 (다음 자음 반복)
|
|
}
|
|
|
|
# 카타카나 → 히라가나 변환 오프셋
|
|
_KATA_OFFSET = ord('ア') - ord('あ')
|
|
|
|
|
|
def _kata_to_hira(text: str) -> str:
|
|
"""카타카나를 히라가나로 변환 (ー 장음기호는 유지)."""
|
|
result = []
|
|
for ch in text:
|
|
cp = ord(ch)
|
|
# 카타카나 범위이되, ー(U+30FC), ・(U+30FB) 등 기호는 제외
|
|
if 0x30A1 <= cp <= 0x30F6: # ア~ヶ (실제 카타카나 문자만)
|
|
result.append(chr(cp - _KATA_OFFSET))
|
|
else:
|
|
result.append(ch)
|
|
return "".join(result)
|
|
|
|
|
|
def japanese_to_romaji(text: str) -> str:
|
|
"""일본어 텍스트를 로마자로 변환 (pykakasi 기반, fallback: 카나 테이블)."""
|
|
try:
|
|
import pykakasi
|
|
kks = pykakasi.kakasi()
|
|
result_items = kks.convert(text)
|
|
romaji = " ".join(item["hepburn"] for item in result_items)
|
|
# 연속 공백 정리
|
|
romaji = re.sub(r'\s+', ' ', romaji).strip()
|
|
return romaji
|
|
except ImportError:
|
|
logger.warning("pykakasi 미설치 — 카나 테이블 fallback 사용")
|
|
return _japanese_to_romaji_fallback(text)
|
|
|
|
|
|
def _japanese_to_romaji_fallback(text: str) -> str:
|
|
"""일본어→로마자 fallback (카나만 변환, 한자는 그대로)."""
|
|
text = _kata_to_hira(text)
|
|
|
|
result = []
|
|
i = 0
|
|
while i < len(text):
|
|
if text[i] == '\u30FC': # ー
|
|
i += 1
|
|
continue
|
|
if i + 1 < len(text) and text[i:i+2] in _KANA_ROMAJI:
|
|
result.append(_KANA_ROMAJI[text[i:i+2]])
|
|
i += 2
|
|
elif text[i] in _KANA_ROMAJI:
|
|
romaji = _KANA_ROMAJI[text[i]]
|
|
if text[i] == 'っ' and i + 1 < len(text):
|
|
next_romaji = _KANA_ROMAJI.get(text[i+1], "")
|
|
if next_romaji:
|
|
result.append(next_romaji[0])
|
|
else:
|
|
result.append(romaji)
|
|
i += 1
|
|
else:
|
|
result.append(text[i])
|
|
i += 1
|
|
|
|
return "".join(result)
|
|
|
|
|
|
|
|
def normalize_title(title: str) -> str:
|
|
"""제목 비교용 정규화: 소문자 + 특수문자 제거 + 공백 정리."""
|
|
title = title.lower().strip()
|
|
# 기수 표기 정규화: 2nd → 2, S2 → 2
|
|
title = re.sub(r'\b(\d+)(?:st|nd|rd|th)\b', r'\1', title)
|
|
title = re.sub(r'\bs(\d+)\b', r'\1', title)
|
|
title = re.sub(r'season\s*(\d+)', r'\1', title)
|
|
title = re.sub(r'(\d+)\s*기', r'\1', title)
|
|
# 특수문자 제거
|
|
title = re.sub(r'[^\w\s]', ' ', title)
|
|
title = re.sub(r'\s+', ' ', title).strip()
|
|
return title
|
|
|
|
|
|
def title_similarity(title_a: str, title_b: str) -> float:
|
|
"""두 제목 간 유사도 (0.0 ~ 1.0)."""
|
|
a = normalize_title(title_a)
|
|
b = normalize_title(title_b)
|
|
return SequenceMatcher(None, a, b).ratio()
|
|
|
|
|
|
def match_titles(
|
|
korean_title: str,
|
|
original_title: str,
|
|
nyaa_results: list,
|
|
threshold: float = 0.4,
|
|
) -> list:
|
|
"""Anissia 제목과 Nyaa 검색 결과 매칭.
|
|
|
|
Args:
|
|
korean_title: 한글 제목 (Anissia subject)
|
|
original_title: 원어 제목 (Anissia originalSubject)
|
|
nyaa_results: TorrentResult 리스트
|
|
threshold: 최소 유사도
|
|
|
|
Returns:
|
|
매칭된 TorrentResult 리스트 (유사도 내림차순)
|
|
"""
|
|
# 원제의 로마자 변환
|
|
romaji = japanese_to_romaji(original_title)
|
|
|
|
scored = []
|
|
for result in nyaa_results:
|
|
# Nyaa 제목에서 그룹태그 제거: [ASW] Title - 07 [...] → Title
|
|
clean_title = re.sub(r'\[[^\]]*\]', '', result.title).strip()
|
|
clean_title = re.sub(r'\s*-\s*\d+.*$', '', clean_title).strip()
|
|
|
|
# 유사도 계산 (로마자 vs Nyaa 제목)
|
|
sim_romaji = title_similarity(romaji, clean_title)
|
|
# 한글 vs Nyaa (일부 자막 포함 릴리스인 경우)
|
|
sim_korean = title_similarity(korean_title, clean_title)
|
|
# 원제 그대로 vs Nyaa
|
|
sim_original = title_similarity(original_title, clean_title)
|
|
|
|
best_sim = max(sim_romaji, sim_korean, sim_original)
|
|
|
|
# ASW HEVC 릴리스는 threshold 면제 (약칭으로 올라와도 포함)
|
|
title_upper = result.title.upper()
|
|
is_preferred = "[ASW]" in result.title and ("HEVC" in title_upper or "X265" in title_upper)
|
|
|
|
if best_sim >= threshold or is_preferred:
|
|
# ASW HEVC 릴리스는 유사도 보너스 (+0.5) → 정렬 시 상위 배치
|
|
effective_sim = best_sim + 0.5 if is_preferred else best_sim
|
|
scored.append((effective_sim, result))
|
|
|
|
# 유사도 내림차순 정렬
|
|
scored.sort(key=lambda x: x[0], reverse=True)
|
|
return [r for _, r in scored]
|
|
|
|
|
|
# ──────────────────────────────────────────────
|
|
# NAS 폴더명 생성
|
|
# ──────────────────────────────────────────────
|
|
|
|
def get_quarter(date_str: str) -> tuple[int, int]:
|
|
"""날짜 문자열에서 연도와 분기 추출.
|
|
|
|
Args:
|
|
date_str: "2026-01-11" 형식
|
|
|
|
Returns:
|
|
(year, quarter): (26, 1)
|
|
"""
|
|
if not date_str:
|
|
from datetime import date
|
|
today = date.today()
|
|
year = today.year % 100
|
|
quarter = (today.month - 1) // 3 + 1
|
|
return year, quarter
|
|
|
|
parts = date_str.split("-")
|
|
year = int(parts[0]) % 100
|
|
month = int(parts[1])
|
|
quarter = (month - 1) // 3 + 1
|
|
return year, quarter
|
|
|
|
|
|
def make_nas_folder_name(title: str, start_date: str = "") -> str:
|
|
"""NAS 저장 폴더명 생성.
|
|
|
|
예: [26_1분기]장송의프리렌2기
|
|
"""
|
|
year, quarter = get_quarter(start_date)
|
|
|
|
# 제목에서 폴더명에 쓸 수 없는 문자 제거
|
|
safe_title = re.sub(r'[<>:"/\\|?*]', '', title)
|
|
safe_title = safe_title.strip()
|
|
|
|
return f"[{year:02d}_{quarter}분기]{safe_title}"
|
|
|
|
|
|
def rename_subtitle_to_video(
|
|
video_filename: str,
|
|
subtitle_ext: str = ".ass",
|
|
) -> str:
|
|
"""영상 파일명에 맞게 자막 파일명 생성.
|
|
|
|
예: [ASW] Sousou no Frieren S2 - 07.mkv → [ASW] Sousou no Frieren S2 - 07.ass
|
|
"""
|
|
stem = re.sub(r'\.[^.]+$', '', video_filename)
|
|
return f"{stem}{subtitle_ext}"
|