import base64 import io import re from typing import Optional import requests # type: ignore[import-untyped] from PIL import Image SMART_PUNCT_TRANSLATION = str.maketrans( { "\u2018": "'", "\u2019": "'", "\u201c": '"', "\u201d": '"', } ) SPEECHABLE_PATTERN = re.compile( r"[^\w\s.,!?;:'\"\-()\/\\@#%&*+=$€£¥₹₽¢\[\]{}<>~`^|…—–\n\r\t]", flags=re.UNICODE, ) # Pattern per rilevare blocchi CJK (Cinese, Giapponese, Coreano) + altre lingue non ammesse CJK_PATTERN = re.compile( r"[\u4e00-\u9fff\u3400-\u4dbf\uf900-\ufaff\ufe30-\ufe4f" # Han (Unified + Extension A + Compatibility Ideographs + Compatibility Forms) r"\u3040-\u309f\u30a0-\u30ff\u31f0-\u31ff" # Hiragana + Katakana (giapponese) r"\uac00-\ud7af\u1100-\u11ff\u3130-\u318f" # Hangul (coreano) r"\u0600-\u06ff\u0750-\u077f\ufb50-\ufdff\ufe70-\ufeff]" # Arabo ) # Lettere accentate spagnole + base latina per spagnolo SPANISH_ALLOWED_CHARS = set( "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ" "áéíóúüñÁÉÍÓÚÜÑ" "0123456789" " ,.!?;:'\"-()/@#%&*+=$€£[]{}<>~`^|…—–\n\r\t" ) def remove_unspeechable(text: str) -> str: """Keep only speechable characters: letters, digits, punctuation, whitespace. support unicode characters (english, arabic, chinese, japanese, korean, etc.) """ text = text.translate(SMART_PUNCT_TRANSLATION) return SPEECHABLE_PATTERN.sub("", text) def sanitize_language_output( text: str, target_language: str = "es", strict: bool = False, ) -> tuple[str, int]: """Ultimo strato di difesa: rimuove testo non appartenente alla lingua target. Rileva e rimuove/tzona blocchi di lingue non ammesse (CJK, Arabo, ecc.). Se strict=True, tronca il testo al PRIMO blocco non ammesso di lunghezza > soglia (per bloccare meta-commenti allucinati come quelli di Qwen in cinese). Returns: (testo_pulito, numero_caratteri_rimossi) """ if not text: return text, 0 original_len = len(text) removed_log = [] # Step 1: Rimuovi blocchi CJK + Arabo (sempre, indipendentemente da strict) def _remove_cjk(match: re.Match[str]) -> str: removed_log.append(f"CJK chars: {match.group(0)[:30]}...") return "" text_clean = CJK_PATTERN.sub(_remove_cjk, text) # Step 2: Se strict, trova il primo punto dove c'e' una sequenza non spagnola # di > 15 caratteri e tronca li' (Qwen inizia il cinese di colpo senza transizione) if strict and target_language == "es": bad_run_start = -1 bad_run_len = 0 for i, ch in enumerate(text_clean): is_ok = ch in SPANISH_ALLOWED_CHARS or ch.isspace() if not is_ok: if bad_run_start == -1: bad_run_start = i bad_run_len += 1 if bad_run_len > 15: # Tronca prima del run non ammesso removed_log.append( f"Troncato a pos {bad_run_start}: run di {bad_run_len} chars non spagnoli" ) text_clean = text_clean[:bad_run_start].rstrip() break else: bad_run_start = -1 bad_run_len = 0 removed_count = original_len - len(text_clean) if removed_count > 0: import logging as _log _log.getLogger(__name__).warning( "sanitize_language_output: rimossi %d caratteri non-%s. Motivi: %s", removed_count, target_language, "; ".join(removed_log[:3]) if removed_log else "sconosciuto", ) return text_clean, removed_count WHISPER_LANGUAGE_TO_LLM_LANGUAGE = { "en": "english", "fr": "french", "es": "spanish", "zh": "chinese", "ja": "japanese", "ko": "korean", "hi": "hindi", "de": "german", "pt": "portuguese", "pl": "polish", "it": "italian", "nl": "dutch", } def resolve_auto_language(language_code: Optional[str]) -> tuple[Optional[str], Optional[str]]: """Strip the ``-auto`` suffix and resolve the human-readable language name. Returns ``(clean_code, language_name)``. ``language_name`` is non-None when the code (with or without ``-auto``) maps to a known language. """ if not language_code: return language_code, None if language_code.endswith("-auto"): language_code = language_code[:-5] if language_code not in WHISPER_LANGUAGE_TO_LLM_LANGUAGE: return language_code, None return language_code, WHISPER_LANGUAGE_TO_LLM_LANGUAGE.get(language_code) def image_url_to_pil(image_url: str) -> Image.Image: """Convert an image URL or base64 data URI to a PIL Image. Accepts: - 'data:image/...;base64,' data URIs - 'https://...`` or ``http://...' URLs (fetched with a 10s timeout) """ if image_url.startswith("data:"): _, b64_data = image_url.split(",", 1) return Image.open(io.BytesIO(base64.b64decode(b64_data))) resp = requests.get(image_url, timeout=10) resp.raise_for_status() return Image.open(io.BytesIO(resp.content))