Blame for utils.py

import re
import logging
import unicodedata
import httpx
from typing import Optional, List, Tuple

logger = logging.getLogger("utils")

def format_size(bytes_size: int) -> str:
    if not bytes_size:
        return "0 B"
    units = ["B", "KB", "MB", "GB", "TB"]
    i = 0
    while bytes_size >= 1024 and i < len(units) - 1:
        bytes_size /= 1024.0
        i += 1
    return f"{bytes_size:.2f} {units[i]}"

def normalize_release_name(name: str) -> str:
    """Normalize release filename for deduplication (lowercasing, removing noise/tags/brackets/extensions)."""
    if not name:
        return ""
    # Strip file extension
    s = re.sub(r'\.[a-z0-9]{2,5}$', '', name.lower())
    # Strip common channel tag patterns like @channel, [channel], (channel), {channel}
    s = re.sub(r'\[.*?\]|\(.*?\)|@\S+[\s:\-_|]*|\{.*?\}', ' ', s)
    # Strip non-alphanumeric characters
    s = re.sub(r'[^a-z0-9]', '', s)
    return s.strip()

# Normalize common numbers and terminology for reliable matching
_NORM_MAP = {
    "one": "1", "two": "2", "three": "3", "four": "4", "five": "5",
    "six": "6", "seven": "7", "eight": "8", "nine": "9", "ten": "10",
    "uno": "1", "dos": "2", "tres": "3", "cuatro": "4", "cinco": "5",
    "seis": "6", "siete": "7", "ocho": "8", "nueve": "9", "diez": "10",
    "temporada": "season", "temp": "season", "capitulo": "episode", 
    "capítulo": "episode", "cap": "episode", "ep": "episode", "ch": "episode", "chapter": "episode"
}

def _normalize_filename(text: str) -> str:
    if not text:
        return ""
    t = text.lower()
    t = re.sub(r'\.[a-z0-9]{2,5}$', '', t)  # strip file extension
    t = re.sub(r'[._\-]', ' ', t)
    
    words = t.split()
    new_words = []
    for w in words:
        if w in _NORM_MAP:
            w = _NORM_MAP[w]
        new_words.append(w)
    return " ".join(new_words)

# Regex lists for identifying seasons/episodes in both orders
_SEASON_EPISODE_PATTERNS = [
    re.compile(r'\bs\s*(?P<s>\d{1,2})\s*[.\-_ ]?\s*e\s*(?P<e>\d{1,3})\b', re.IGNORECASE),
    re.compile(r'\bseason\s*(?P<s>\d{1,2})\D{0,10}?episode\s*(?P<e>\d{1,3})\b', re.IGNORECASE),
    re.compile(r'\bt\s*(?P<s>\d{1,2})\s*[.\-_ ]?\s*c\s*(?P<e>\d{1,3})\b', re.IGNORECASE),
    re.compile(r'(?<!\d)(?P<s>\d{1,2})\s*[xX]\s*(?P<e>\d{1,3})(?!\d)'),
]

_EPISODE_SEASON_PATTERNS = [
    re.compile(r'\be\s*(?P<e>\d{1,3})\s*[.\-_ xX]?\s*s\s*(?P<s>\d{1,2})\b', re.IGNORECASE),
    re.compile(r'\bepisode\s*(?P<e>\d{1,3})\D{0,10}?season\s*(?P<s>\d{1,2})\b', re.IGNORECASE),
    re.compile(r'\bc\s*(?P<e>\d{1,3})\s*[.\-_ ]?\s*t\s*(?P<s>\d{1,2})\b', re.IGNORECASE),
]

_STANDALONE_EPISODE_PATTERNS = [
    re.compile(r'\bep(?:isode)?\s*[.\-_ ]?\s*(?P<e>\d{1,3})\b', re.IGNORECASE),
    re.compile(r'\bcap(?:itulo|ítulo)?\s*[.\-_ ]?\s*(?P<e>\d{1,3})\b', re.IGNORECASE),
    re.compile(r'\[(?P<e>\d{2,3})\]'),
    re.compile(r'(?:^|[\s\-_])[-–]\s*(?P<e>\d{2,3})\s*(?:[-–]|$|\.)'),
]

def parse_season_episode(filename: str) -> tuple:
    if not filename:
        return None, None
    
    fn = _normalize_filename(filename)
    
    for pat in _SEASON_EPISODE_PATTERNS:
        m = pat.search(fn)
        if m:
            try:
                s = int(m.group('s'))
                e = int(m.group('e'))
                return s, e
            except (ValueError, KeyError, IndexError):
                pass
                
    for pat in _EPISODE_SEASON_PATTERNS:
        m = pat.search(fn)
        if m:
            try:
                s = int(m.group('s'))
                e = int(m.group('e'))
                return s, e
            except (ValueError, KeyError, IndexError):
                pass
                
    for pat in _STANDALONE_EPISODE_PATTERNS:
        m = pat.search(fn)
        if m:
            try:
                e = int(m.group('e'))
                return 1, e
            except (ValueError, KeyError, IndexError):
                pass
                
    return None, None

def matches_episode(filename: str, season: int, episode: int) -> bool:
    if season is None or episode is None:
        return True
        
    f_season, f_episode = parse_season_episode(filename)
    if f_season == season and f_episode == episode:
        return True
        
    fn = _normalize_filename(filename)
    
    patterns = [
        rf'\bs\s*{season:02d}\s*[.\-_ ]?\s*e\s*{episode:02d}\b',
        rf'\bs\s*{season}\s*[.\-_ ]?\s*e\s*{episode:02d}\b',
        rf'(?<!\d){season}[xX]{episode:02d}(?!\d)',
        rf'(?<!\d){season}[xX]{episode}(?!\d)',
        rf'\[season\s*0*{season}\].*?\[episode\s*0*{episode}\]',
        rf'season\s*0*{season}\D{{0,20}}?episode\s*0*{episode}(?!\d)',
        rf'\bt\s*{season:02d}\s*c\s*{episode:02d}\b',
        rf'\bt\s*{season}\s*c\s*{episode}\b',
        rf'(?<!\d){season}{episode:02d}(?!\d)',
    ]
    
    # Allow fallback standalone episode checks for Season 1
    has_explicit_season = any(re.search(p, fn, re.IGNORECASE) for p in [r'\bs\d', r'\bseason\s*\d', r'\bt\d', r'\d+[xX]'])
    if season == 1 and not has_explicit_season:
        patterns += [
            rf'\bepisode\s*0*{episode}\b',
            rf'\bcap\s*0*{episode}\b',
            rf'\[0*{episode}\]',
            rf'[-–]\s*0*{episode:02d}\s*(?:[-–]|$)',
        ]
    
    for pat in patterns:
        if re.search(pat, fn, re.IGNORECASE):
            return True
            
    return False

def matches_subtitle(video_filename: str, sub_filename: str) -> bool:
    if not video_filename or not sub_filename:
        return False
        
    v_fn = video_filename.lower()
    s_fn = sub_filename.lower()
    
    v_base = v_fn.rsplit('.', 1)[0]
    s_base = s_fn.rsplit('.', 1)[0]
    
    s_base_clean = re.sub(r'\.(eng|en|english|sub|subtitle|srt|vtt)$', '', s_base)
    
    if s_base_clean in v_base or v_base in s_base_clean:
        return True
        
    return False

def get_search_query_from_filename(filename: str) -> str:
    if not filename:
        return ""
    name = filename.lower()
    name = name.rsplit('.', 1)[0]
    name = re.sub(r'[._\-]', ' ', name)
    
    terms = r'\b(2160p|1080p|720p|480p|360p|4k|8k|10bit|h264|x264|h265|x265|hevc|web[- ]?rip|bluray|brrip|hdrip)\b'
    match = re.search(terms, name)
    if match:
        name = name[:match.start()]
    
    name = re.sub(r'\s+', ' ', name).strip()
    return name

def parse_split_info(filename: str) -> tuple:
    if not filename:
        return None, None
        
    # Match suffix .001, .002 etc.
    m1 = re.search(r'\.(\d{3,4})$', filename)
    if m1:
        part = int(m1.group(1))
        base = filename[:m1.start()]
        return base, part
        
    # Match part1, part01, part_1 etc.
    m2 = re.search(r'[._\- ]part_?(\d+)(?:\.([^.]+))?$', filename, re.IGNORECASE)
    if m2:
        part = int(m2.group(1))
        ext = m2.group(2) or ""
        base = filename[:m2.start()]
        if ext:
            base += f".{ext}"
        return base, part
        
    return None, None

_metadata_cache = {}

async def get_metadata_from_cinemeta(meta_type: str, imdb_id: str) -> dict:
    cache_key = f"{meta_type}:{imdb_id}"
    if cache_key in _metadata_cache:
        return _metadata_cache[cache_key]

    url = f"https://v3-cinemeta.strem.io/meta/{meta_type}/{imdb_id}.json"
    logger.info(f"Fetching metadata from Cinemeta: {url}")
    
    try:
        async with httpx.AsyncClient(timeout=8.0, follow_redirects=True) as client:
            resp = await client.get(url)
            if resp.status_code == 200:
                data = resp.json()
                meta = data.get("meta", {})
                if meta:
                    result = {
                        "name": meta.get("name"),
                        "year": meta.get("year"),
                        "genres": meta.get("genres", []),
                        "poster": meta.get("poster")
                    }
                    _metadata_cache[cache_key] = result
                    return result
    except Exception as e:
        logger.error(f"Cinemeta metadata lookup failed: {e}")
        
    return {}

VIDEO_EXTENSIONS = ('.mp4', '.mkv', '.avi', '.mov', '.wmv', '.flv', '.webm', '.ts', '.m4v')

def is_video_file(filename: str) -> bool:
    return filename.lower().endswith(VIDEO_EXTENSIONS)

def normalize_title(title: str) -> str:
    if not title:
        return ""
    t = "".join(c for c in unicodedata.normalize('NFD', title) if unicodedata.category(c) != 'Mn')
    t = t.lower()
    t = re.sub(r'[^a-z0-9\s]', ' ', t)
    t = re.sub(r'\bii\b', '2', t)
    t = re.sub(r'\biii\b', '3', t)
    t = re.sub(r'\biv\b', '4', t)
    t = re.sub(r'\bv\b', '5', t)
    t = re.sub(r'\bvi\b', '6', t)
    t = re.sub(r'\bvii\b', '7', t)
    t = re.sub(r'\bviii\b', '8', t)
    t = re.sub(r'\bix\b', '9', t)
    t = re.sub(r'\bx\b', '10', t)
    t = re.sub(r'\s+', ' ', t).strip()
    return t

def _clean_title_prefix(filename: str) -> str:
    if not filename:
        return ""
    fn_lower = filename.lower()
    first_match_idx = len(filename)
    
    # Locate season/episode split point
    all_patterns = _SEASON_EPISODE_PATTERNS + _EPISODE_SEASON_PATTERNS + _STANDALONE_EPISODE_PATTERNS
    for pat in all_patterns:
        m = pat.search(fn_lower)
        if m:
            first_match_idx = min(first_match_idx, m.start())
            
    # Locate year split point
    year_match = re.search(r'\b(19\d{2}|20[0-2]\d)\b', filename)
    if year_match:
        first_match_idx = min(first_match_idx, year_match.start())
        
    prefix = filename[:first_match_idx]
    return prefix.strip()

def matches_title(filename: str, title: str) -> bool:
    if not title:
        return True
        
    norm_title = normalize_title(title)
    prefix = _clean_title_prefix(filename)
    norm_prefix = normalize_title(prefix)
    
    if not norm_prefix:
        norm_prefix = normalize_title(filename)
        
    if norm_title in norm_prefix:
        return True
        
    # Check if all major words of the title are in the prefix
    words = [w for w in norm_title.split() if w not in ('a', 'an', 'the', 'and', 'or', 'of', 'in', 'to', 'for', 'with')]
    if not words:
        words = norm_title.split()
        
    return all(word in norm_prefix for word in words)

def get_mime_type(filename: str, default_mime: str = None) -> str:
    if default_mime and default_mime not in ("application/octet-stream", "application/zip", "binary/octet-stream"):
        return default_mime
    fn = str(filename).lower()
    if fn.endswith(".mp4"):
        return "video/mp4"
    if fn.endswith(".mkv"):
        return "video/x-matroska"
    if fn.endswith(".webm"):
        return "video/webm"
    if fn.endswith(".avi"):
        return "video/x-msvideo"
    if fn.endswith(".mov"):
        return "video/quicktime"
    if fn.endswith(".ts"):
        return "video/mp2t"
    if fn.endswith(".m4v"):
        return "video/x-m4v"
    return default_mime or "video/mp4"

_SITE_AND_GROUP_PATTERNS = [
    re.compile(r'@\S+', re.IGNORECASE),
    re.compile(r'\b(?:1tamilmv|tamilmv|themoviesboss|udevmy|skymovieshd|moviesmod|bollyflix|vegamovies|katmoviehd|cinemaluxe|yify|yts|rarbg|galaxyrg|qxr|ion10|psa)\b', re.IGNORECASE),
]

_NOISE_STRIP_PATTERNS = [
    re.compile(r'\b(?:hi10p?|high10|hevc|h264|h265|x264|x265|ddp5\.1|dd5\.1|dts-hd|truehd|atmos|aac2\.0|aac|ac3|eac3|dts|mp3|opus|flac)\b', re.IGNORECASE),
    re.compile(r'\b(?:sci-fi|hi-fi|hi-res|hi-def)\b', re.IGNORECASE),
    re.compile(r'\b(?:2160p|1080p|720p|480p|360p|4k|8k|uhd|fhd|hd|sd)\b', re.IGNORECASE),
    re.compile(r'\b(?:web-?dl|web-?rip|bluray|brrip|bdrip|hdrip|dvdrip|hdtc|hdts|camrip|cam|screener|scr|telesync)\b', re.IGNORECASE),
    re.compile(r'\b(?:proper|repack|unrated|extended|imax|remastered|uncut|directors\.cut)\b', re.IGNORECASE),
    re.compile(r'\b(?:10bit|8bit|hdr10\+?|hdr|dv|dolby\.vision|sdr)\b', re.IGNORECASE),
    re.compile(r'\b(?:clean|cleaned|org|original)\b', re.IGNORECASE),
]

_SUB_STRIP_PATTERNS = [
    re.compile(r'\b[a-z]{2,10}[\s._\-]*(?:subs?|subtitles?)\b', re.IGNORECASE),
    re.compile(r'\b(?:e|m|soft|hard|vob|pgs|idx)?subs?(?:title)?s?\b', re.IGNORECASE),
    re.compile(r'\[[^\]]*\b(?:sub|subs|subtitles?)\b[^\]]*\]', re.IGNORECASE),
    re.compile(r'\([^\)]*\b(?:sub|subs|subtitles?)\b[^\)]*\)', re.IGNORECASE),
]

_MULTI_EXPLICIT_PATTERN = re.compile(
    r'\b(?:multi(?:[\s._\-]*(?:audio|dub|subs?|lang|language))?|multiple[\s._\-]*audio|tri[\s._\-]*audio|quad[\s._\-]*audio|dual[\s._\-]*audio)\b',
    re.IGNORECASE
)

_LANG_DEFINITIONS = [
    ("hi", ["hindi", "hind"], ["hin", "hi"]),
    ("eng", ["english"], ["eng", "en"]),
    ("tam", ["tamil"], ["tam", "ta"]),
    ("tel", ["telugu"], ["tel", "te"]),
    ("mal", ["malayalam"], ["mal", "ml"]),
    ("kan", ["kannada"], ["kan", "kn"]),
    ("ben", ["bengali", "bangla"], ["ben", "bng", "bn"]),
    ("pun", ["punjabi"], ["pun", "pan", "pa"]),
    ("mar", ["marathi"], ["mar", "mr"]),
    ("guj", ["gujarati"], ["guj", "gu"]),
    ("urd", ["urdu"], ["urd", "ur"]),
    ("ori", ["odia", "oriya"], ["ori"]),
    ("asm", ["assamese"], ["asm"]),
    ("nep", ["nepali"], ["nep"]),
    ("sin", ["sinhala", "sinhalese"], ["sin"]),
    ("tag", ["tagalog", "filipino"], ["tag", "fil"]),
    ("spa", ["spanish", "castellano", "espanol", "español", "latino"], ["spa", "esp", "es"]),
    ("fre", ["french", "francais", "français"], ["fre", "fra", "fr"]),
    ("ger", ["german", "deutsch"], ["ger", "deu", "de"]),
    ("ita", ["italian", "italiano"], ["ita", "it"]),
    ("por", ["portuguese", "portugues", "português"], ["por", "pt", "pt-br"]),
    ("rus", ["russian"], ["rus", "ru"]),
    ("jap", ["japanese"], ["jap", "jpn", "ja"]),
    ("kor", ["korean"], ["kor", "ko"]),
    ("chi", ["chinese", "mandarin", "cantonese"], ["chi", "zho", "zh"]),
    ("ara", ["arabic"], ["ara", "ar"]),
    ("tur", ["turkish"], ["tur", "tr"]),
    ("tha", ["thai"], ["tha", "th"]),
    ("vie", ["vietnamese"], ["vie", "vi"]),
    ("ind", ["indonesian"], ["ind", "id"]),
    ("heb", ["hebrew"], ["heb", "he"]),
    ("pol", ["polish"], ["pol", "pl"]),
    ("dut", ["dutch"], ["dut", "nld", "nl"]),
    ("swe", ["swedish"], ["swe", "sv"]),
    ("ukr", ["ukrainian"], ["ukr", "uk"]),
]

_COMPOUND_SEP = r'[\s+\-_/&,.:|]'

def _clean_text_for_lang_detection(text: str, title: Optional[str] = None) -> str:
    if not text:
        return ""
    t = text.replace("_", " ")

    for p in _SITE_AND_GROUP_PATTERNS:
        t = p.sub(' ', t)

    if title:
        title_words = re.findall(r'\b[a-zA-Z0-9]+\b', title)
        for tw in title_words:
            if len(tw) >= 2:
                t = re.sub(rf'\b{re.escape(tw)}\b', ' ', t, flags=re.IGNORECASE)

    for p in _NOISE_STRIP_PATTERNS:
        t = p.sub(' ', t)

    for p in _SUB_STRIP_PATTERNS:
        t = p.sub(' ', t)

    return t

def _extract_caption_audio_section(caption: str) -> Optional[str]:
    if not caption:
        return None
    m = re.search(r'(?:^|\n)\s*(?:[🔊🎧🗣️🎙️ℹ️\s]*)(?:audio(?:\s*language)?|language|languages|audios)\s*[:\-]\s*([^\n\r]+)', caption, re.IGNORECASE)
    if m:
        return m.group(1).strip()
    return None

def parse_audio_languages(
    filename: str = "",
    caption: str = "",
    title: Optional[str] = None,
    default: str = "Telegram File"
) -> str:
    """
    Parses audio language(s) from a Telegram release filename and message caption.
    Returns:
      - 'multi' if explicit multi-audio or >= 3 languages are detected
      - Comma-separated list like 'hi,eng' if 2 languages are detected
      - Single language code like 'hi' or 'tam' if 1 language is detected
      - `default` if no audio language could be recognized
    """
    raw_combined = f"{filename or ''} {caption or ''}".strip()
    if not raw_combined:
        return default

    caption_audio = _extract_caption_audio_section(caption)
    text_to_check = f"{filename or ''} {caption_audio or caption or ''}"

    is_explicit_multi = bool(_MULTI_EXPLICIT_PATTERN.search(text_to_check.replace("_", " ")))

    cleaned = _clean_text_for_lang_detection(text_to_check, title=title)

    detected = []
    seen = set()

    for code, full_names, abbrs in _LANG_DEFINITIONS:
        matched = False
        match_pos = 999999

        for fn in full_names:
            m = re.search(rf'\b{re.escape(fn)}\b', cleaned, re.IGNORECASE)
            if m:
                matched = True
                match_pos = min(match_pos, m.start())

        if not matched:
            for ab in abbrs:
                if len(ab) >= 3:
                    m = re.search(rf'\b{re.escape(ab)}\b', cleaned, re.IGNORECASE)
                else:
                    m = re.search(rf'(?<=[+.\-/\s\[\(]{_COMPOUND_SEP}){re.escape(ab)}(?=[+.\-/\s\]\)]{_COMPOUND_SEP}|$)', cleaned, re.IGNORECASE)
                    if not m:
                        m = re.search(rf'\[[^\]]*\b{re.escape(ab)}\b[^\]]*\]', cleaned, re.IGNORECASE)

                if m:
                    matched = True
                    match_pos = min(match_pos, m.start())

        if matched and code not in seen:
            seen.add(code)
            detected.append((match_pos, code))

    detected.sort(key=lambda x: x[0])
    detected_codes = [c for _, c in detected]

    if detected_codes:
        return ",".join(detected_codes)

    if is_explicit_multi:
        return "multi"

    return default