"""제목 매칭 + 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"[^>]*>([^<]+)', html ) snippet_matches = _re.findall( r'class="result__snippet"[^>]*>(.*?)', 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}"