import json import logging import re from typing import Optional, Dict, Any import httpx from app import config from app.database import get_db, get_cached_sentence_translation, save_sentence_translation logger = logging.getLogger(__name__) def get_dictionary_hint(term: str) -> str: """Retrieves verified offline dictionary translation hints for term.""" clean = term.strip() if not clean: return "" try: with get_db() as conn: row = conn.execute( "SELECT translation FROM offline_dictionary WHERE term = ? COLLATE NOCASE", (clean,), ).fetchone() if row and row["translation"]: return row["translation"].strip() except Exception: pass return "" def build_translation_prompt( sentence: str, target_word: str = "", context: str = "", dict_hint: str = "", ) -> str: clean_sentence = sentence.strip() clean_word = target_word.strip() clean_context = context.strip() context_section = "" if clean_context and clean_context != clean_sentence: short_ctx = clean_context[:500].strip() context_section = f'Контекст абзаца:\n"{short_ctx}"\n\n' target_section = "" if clean_word: target_section = f'Целевое слово/выражение: "{clean_word}"\n' if dict_hint: target_section += f'Возможные словарные значения: {dict_hint}\n' target_section += "\n" return ( "Ты — профессиональный литературный переводчик художественной и научно-популярной литературы на русский язык.\n" "Твоя задача — перевести указанное английское предложение на красивый, естественный и грамотный русский литературный язык, идеально вписывающийся в повествование книги.\n\n" f"{context_section}" f'Предложение для перевода:\n"{clean_sentence}"\n\n' f"{target_section}" "Обязательные правила перевода:\n" "1. Перевод должен быть живым и художественным (НЕ дословным, без топорного калькирования).\n" "2. Переводи всё предложение ПОЛНОСТЬЮ от начала до конца, не обрывая на точках, обращениях или знаках препинания.\n" "3. Обязательно сохраняй обращения и титулы: Mr. -> мистер, Mrs. -> миссис, Dr. -> доктор, Prof. -> профессор, St. -> святой/Сент-. Не теряй их!\n" "4. В поле \"word_meaning\" укажи только точный перевод выделенного слова на русский язык в контексте данного предложения (выбрав самый подходящий вариант из словарных подсказок).\n\n" "Ответь ИСКЛЮЧИТЕЛЬНО в формате JSON:\n" "{\n" ' "sentence_ru": "полный литературный перевод всего предложения",\n' f' "word_meaning": "{"русский перевод слова в данном контексте" if clean_word else ""}"\n' "}" ) async def translate_sentence_ai( sentence: str, target_word: str = "", context: str = "" ) -> Dict[str, Any]: """ Translates a sentence using local AI (Ollama), with SQLite instant-cache lookup. Incorporates paragraph context and offline dictionary hints for maximum naturalness. Returns: { "original_sentence": "...", "sentence_ru": "...", "target_word": "...", "word_meaning": "...", "source": "ollama:model" | "cache" | "fallback", "cached": bool } """ clean_sentence = sentence.strip() clean_word = target_word.strip() clean_context = context.strip() if not clean_sentence: return { "original_sentence": "", "sentence_ru": "", "target_word": clean_word, "word_meaning": "", "source": "none", "cached": False, } # 1. Fast SQLite Cache Lookup (0.01 ms) cached = get_cached_sentence_translation(clean_sentence, clean_word) if cached: return { "original_sentence": clean_sentence, "sentence_ru": cached["translated_text"], "target_word": clean_word, "word_meaning": cached.get("word_meaning") or "", "source": cached.get("source") or "cache", "cached": True, } # 2. Retrieve offline dictionary hint to steer LLM towards authentic meanings dict_hint = get_dictionary_hint(clean_word) if clean_word else "" # 3. Local AI Generation (Ollama) prompt = build_translation_prompt( sentence=clean_sentence, target_word=clean_word, context=clean_context, dict_hint=dict_hint, ) payload = { "model": config.OLLAMA_MODEL, "prompt": prompt, "stream": False, "format": "json", "options": { "temperature": 0.2, "num_predict": 300, }, } sentence_ru = "" word_meaning = "" source = f"ollama:{config.OLLAMA_MODEL}" try: async with httpx.AsyncClient(timeout=35.0) as client: resp = await client.post(f"{config.OLLAMA_BASE_URL}/api/generate", json=payload) if resp.status_code == 200: data = resp.json() raw_response = data.get("response", "{}") try: parsed = json.loads(raw_response) except Exception: # In case LLM emitted markdown wrapper ```json ... ``` clean_raw = raw_response.strip() if clean_raw.startswith("```json"): clean_raw = clean_raw[7:] if clean_raw.startswith("```"): clean_raw = clean_raw[3:] if clean_raw.endswith("```"): clean_raw = clean_raw[:-3] try: parsed = json.loads(clean_raw.strip()) except Exception: parsed = {} sentence_ru = str(parsed.get("sentence_ru", "")).strip() # Extract word meaning flexibly raw_meaning = ( parsed.get("word_meaning") or parsed.get(f"{clean_word}_meaning") or parsed.get("meaning") or "" ) raw_meaning = str(raw_meaning).strip() # Clean up format like "briefcase: портфель" or "портфель (сущ.)" if ":" in raw_meaning: raw_meaning = raw_meaning.split(":", 1)[1].strip() # If model returned the English term back, fallback to dictionary hint if raw_meaning.lower() == clean_word.lower() or not raw_meaning: if dict_hint: raw_meaning = dict_hint.split(",")[0].strip() else: raw_meaning = "" word_meaning = raw_meaning else: logger.warning(f"Ollama returned HTTP {resp.status_code}: {resp.text}") except Exception as e: logger.warning(f"Ollama connection error: {e}") # 4. Graceful Fallback if Ollama is unreachable if not sentence_ru: sentence_ru = "(Локальный ИИ временно недоступен)" source = "fallback" # Save to SQLite Cache if valid translation obtained if sentence_ru and not sentence_ru.startswith("("): save_sentence_translation( sentence=clean_sentence, translated_text=sentence_ru, target_word=clean_word, word_meaning=word_meaning, source=source, ) return { "original_sentence": clean_sentence, "sentence_ru": sentence_ru, "target_word": clean_word, "word_meaning": word_meaning, "source": source, "cached": False, } async def get_ollama_status() -> Dict[str, Any]: """Checks Ollama server connectivity and lists installed models.""" try: async with httpx.AsyncClient(timeout=3.0) as client: resp = await client.get(f"{config.OLLAMA_BASE_URL}/api/tags") if resp.status_code == 200: data = resp.json() models = [m.get("name") for m in data.get("models", [])] return { "online": True, "models": models, "active_model": config.OLLAMA_MODEL, } except Exception: pass return { "online": False, "models": [], "active_model": config.OLLAMA_MODEL, }