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__) COMMON_STOPWORDS = { "a", "an", "the", "in", "on", "at", "to", "for", "of", "with", "by", "from", "up", "about", "into", "over", "after", "beneath", "under", "above", "is", "am", "are", "was", "were", "be", "been", "being", "have", "has", "had", "do", "does", "did", "and", "but", "if", "or", "because", "as", "until", "while", "this", "that", "these", "those", "i", "you", "he", "she", "it", "we", "they", "them", "him", "her", "us", "me", "my", "your", "his", "its", "our", "their", "not", "no", "nor", "can", "could", "will", "would", "shall", "should", "may", "might", "must", "there", "here", "when", "where", "why", "how", "all", "any", "both", "each", "few", "more", "most", "other", "some", "such", "than", "too", "very", "so", "s", "t", "d", "ll", "ve", "m" } def clean_dict_translation(raw: str) -> str: """Extracts first 1-2 clean Cyrillic meanings, stripping phonetic / Latin noise.""" if not raw: return "" from app.dictionary import clean_russian_translation cleaned = clean_russian_translation(raw) parts = [p.strip() for p in cleaned.split(",") if p.strip()] return ", ".join(parts[:2]) if parts else raw[:50].strip() 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 clean_dict_translation(row["translation"]) except Exception: pass return "" def get_sentence_dictionary_hints(sentence: str, target_word: str = "") -> str: """Extracts verified offline dictionary translations for key words and phrases in sentence.""" if not sentence: return "" words = re.findall(r"\b[a-zA-Z\'-]+\b", sentence) if not words: return "" candidates = set() # Check 2-word contiguous phrases (collocations/idioms) for i in range(len(words) - 1): phrase = f"{words[i]} {words[i+1]}".lower() candidates.add(phrase) # Check content words for w in words: wl = w.lower() if len(wl) > 2 and wl not in COMMON_STOPWORDS: candidates.add(wl) if not candidates: return "" hints = [] special_guidance = { "owl": "owl — это сова/неясыть (ни в коем случае НЕ орёл!)", "tawny owl": "tawny owl — это сова/неясыть (птица семейства совиных, ни в коем случае НЕ орёл!)", } try: with get_db() as conn: placeholders = ",".join(["?"] * len(candidates)) query = f"SELECT term, translation FROM offline_dictionary WHERE term IN ({placeholders}) COLLATE NOCASE" rows = conn.execute(query, list(candidates)).fetchall() matched_dict = {} for r in rows: term_lower = r["term"].lower() clean_trans = clean_dict_translation(r["translation"]) if clean_trans: matched_dict[term_lower] = clean_trans seen_terms = set() for cand in sorted(matched_dict.keys(), key=lambda x: (0 if " " in x else 1, 0 if x == target_word.lower() else 1)): if cand in seen_terms: continue seen_terms.add(cand) trans = matched_dict[cand] guidance = special_guidance.get(cand, "") if guidance: hints.append(f"- {cand}: {trans} (ВНИМАНИЕ: {guidance})") else: hints.append(f"- {cand}: {trans}") if len(hints) >= 8: break except Exception as e: logger.warning(f"Error extracting sentence dictionary hints: {e}") return "\n".join(hints) def build_translation_prompt( sentence: str, target_word: str = "", context: str = "", dict_hint: str = "", sentence_hints: 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" hints_section = "" if sentence_hints: hints_section = f"Словарные подсказки по ключевым словам предложения:\n{sentence_hints}\n\n" return ( "Ты — профессиональный литературный переводчик художественной и научно-популярной литературы на русский язык.\n" "Твоя задача — перевести указанное английское предложение на красивый, естественный, богатый и грамматически безупречный русский язык, идеально вписывающийся в повествование книги.\n\n" f"{context_section}" f'Предложение для перевода:\n"{clean_sentence}"\n\n' f"{target_section}" f"{hints_section}" "Обязательные правила перевода:\n" "1. ПОЛНОТА И СВЯЗНОСТЬ: Переводи абсолютно ВСЕ смысловые части, обороты и определения без пропусков! Обращай особое внимание на цепочки прилагательных и определений (например: 'a large, tawny owl' -> 'крупная неясыть' или 'большая рыжевато-бурая сова' — ни в коем случае не пропускай 'large'!). Также переводи бессоюзные определительные обороты (zero relative clauses, например: 'Tuesday our story starts' -> 'вторник, с которого начинается наша история'). Ничего не выбрасывай!\n" "2. БЕЗ ГАЛЛЮЦИНАЦИЙ И ТОЧНОСТЬ ЗНАЧЕНИЙ: Фауна, флора и объекты должны переводиться строго точно по смыслу и словарю (например: 'owl' — это сова/неясыть, ни в коем случае НЕ орёл!). Никаких смысловых подмен.\n" "3. БЕЗ ДОСЛОВНОГО КАЛЬКИРОВАНИЯ: Никаких топорных калек предлогов ('on Tuesday' -> 'во вторник', а не 'на вторнике'). Парные эпитеты ('dull, gray') переводи богатыми русскими синонимами ('пасмурный, серый'), не повторяя одно и то же слово.\n" "4. ИМЕНА И ТИТУЛЫ: Имена персонажей и фамилии передавай полностью на русском языке (кириллицей), без латинских букв (Mr. and Mrs. Dursley -> мистер и миссис Дурсли). Сохраняй титулы: Mr. -> мистер, Mrs. -> миссис, Dr. -> доктор, Prof. -> профессор.\n" "5. РУССКИЙ СИНТАКСИС: Сохраняй естественный русский порядок слов (обстоятельства времени и места ставь перед сказуемым: 'вскоре по всей стране начнут происходить...').\n" f'6. В поле "word_meaning" укажи только точный перевод слова "{clean_word}" на русский язык в контексте этого предложения (выбрав лучший вариант из подсказок).\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 hints to steer LLM towards authentic meanings dict_hint = get_dictionary_hint(clean_word) if clean_word else "" sentence_hints = get_sentence_dictionary_hints(clean_sentence, clean_word) # 3. Local AI Generation (Ollama) prompt = build_translation_prompt( sentence=clean_sentence, target_word=clean_word, context=clean_context, dict_hint=dict_hint, sentence_hints=sentence_hints, ) payload = { "model": config.OLLAMA_MODEL, "prompt": prompt, "stream": False, "format": "json", "keep_alive": "24h", "options": { "temperature": 0.2, "num_predict": 300, }, } sentence_ru = "" word_meaning = "" source = f"ollama:{config.OLLAMA_MODEL}" try: async with httpx.AsyncClient(timeout=60.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: err_msg = f"{type(e).__name__}: {e}" if str(e) else type(e).__name__ logger.warning(f"Ollama connection error ({err_msg})") # 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, } async def warmup_ollama_model() -> bool: """Preloads the configured Ollama model into RAM with keep_alive to eliminate cold-start latency.""" try: async with httpx.AsyncClient(timeout=60.0) as client: resp = await client.post( f"{config.OLLAMA_BASE_URL}/api/generate", json={"model": config.OLLAMA_MODEL, "keep_alive": "24h"}, ) if resp.status_code == 200: logger.info(f"Ollama model '{config.OLLAMA_MODEL}' warmed up successfully (keep_alive=24h).") return True else: logger.warning(f"Ollama warmup returned HTTP {resp.status_code}: {resp.text}") except Exception as e: err_msg = f"{type(e).__name__}: {e}" if str(e) else type(e).__name__ logger.warning(f"Ollama warmup deferred ({err_msg})") return False def generate_deterministic_debrief( session_data: Dict[str, Any], history_data: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """ Fast rule-based analytical debrief engine with Murphy pedagogic insight. Ensures zero server delay and 100% reliable fallback if Ollama is offline. """ history_data = history_data or {} total_score = int(session_data.get("total_score", 0)) errors = session_data.get("errors", []) steps = session_data.get("steps", []) if total_score >= 90: grade = "A+" headline = "Превосходная сессия! Отличная концентрация и высокая точность." elif total_score >= 75: grade = "A" headline = "Отличная работа! Большинство заданий выполнено уверенно." elif total_score >= 60: grade = "B" headline = "Хороший результат! Есть отдельные нюансы для закрепления." else: grade = "C" headline = "Урок завершён! Рекомендуем уделить внимание повторению разобранных тем." # Identify strengths strengths = [] for s in steps: st_type = s.get("type", "") st_score = s.get("score", 100) if st_score >= 80: if st_type == "vocab": strengths.append(f"Уверенное воспроизведение словарного запаса ({s.get('correct', 0)} из {s.get('total', 0)} слов).") elif st_type == "grammar_review": strengths.append("Успешное закрепление ранее изученных грамматических правил.") elif st_type == "grammar_new": strengths.append("Быстрое освоение новой грамматической темы Мерфи.") elif st_type == "listening": strengths.append("Отличное распознавание английской речи и ключевых деталей на слух.") elif st_type == "reading": words_saved = s.get("words_saved", 0) strengths.append(f"Качественная словарная работа с текстом (+{words_saved} новых слов в словарь).") if not strengths: strengths.append("Вы успешно завершили все запланированные этапы комплексного урока.") # Identify error patterns error_patterns = [] chronic_vocab = set(history_data.get("chronic_vocab_errors", [])) past_grammar = set(history_data.get("past_grammar_errors", [])) for err in errors[:5]: err_type = err.get("type", "") if err_type == "grammar": title = err.get("unit_title", "Грамматика") unit_num = err.get("unit_num", 0) formula = err.get("formula", "") exp = err.get("explanation", "") is_repeating = unit_num in past_grammar repeat_tag = " [Повторяющаяся ошибка]" if is_repeating else "" msg = f"{title}{repeat_tag}: {exp}" if exp else f"{title}{repeat_tag}: обратите внимание на формулу {formula}." if msg not in error_patterns: error_patterns.append(msg) elif err_type == "vocab": term = err.get("term", "") trans = err.get("translation", "") is_chronic = term.lower() in chronic_vocab tag = " [Трудное слово — уже встречалась ошибка]" if is_chronic else "" error_patterns.append(f"Слово «{term}» ({trans}){tag}: требует повторного прогона в тренажёре.") elif err_type == "listening": q = err.get("question", "") ideal = err.get("ideal_answer", "") error_patterns.append(f"Аудирование («{q}»): диктор обращал внимание на ответ «{ideal}».") if not error_patterns: error_patterns.append("Ошибок в этой сессии не зафиксировано! Все тестовые задания решены верно.") # Tomorrow focus tomorrow_focus = [] grammar_errors = [e for e in errors if e.get("type") == "grammar"] vocab_errors = [e for e in errors if e.get("type") == "vocab"] if grammar_errors: u_name = grammar_errors[0].get("unit_title", "пройденную тему") tomorrow_focus.append(f"Закрепить правило: {u_name}.") if vocab_errors: terms_str = ", ".join([f"«{e.get('term')}»" for e in vocab_errors[:3]]) tomorrow_focus.append(f"Повторить проблемные слова: {terms_str}.") if not tomorrow_focus: tomorrow_focus.append("Продолжить движение по учебнику Мерфи и расширять словарный запас.") return { "grade": grade, "score": total_score, "headline": headline, "strengths": strengths[:3], "error_patterns": error_patterns[:4], "tomorrow_focus": tomorrow_focus[:3], "source": "fallback:deterministic", } async def generate_lesson_ai_debrief( session_data: Dict[str, Any], history_data: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """ Generates a personalized pedagogic debrief using local LLM (Qwen 2.5 3B). Blends deterministic fact verification with natural narrative mentorship. Gracefully falls back to deterministic analysis if LLM is offline or delayed. """ history_data = history_data or {} total_score = int(session_data.get("total_score", 0)) preset_name = session_data.get("preset_name", "Комплексный урок") duration_min = round(session_data.get("duration_seconds", 0) / 60) errors = session_data.get("errors", []) steps = session_data.get("steps", []) # Format steps summary steps_lines = [] for s in steps: steps_lines.append(f"- {s.get('title', s.get('type', 'Блок'))}: результат {s.get('score', 100)}%") steps_summary_text = "\n".join(steps_lines) if steps_lines else "Все блоки пройдены" # Format errors summary errors_lines = [] for e in errors: e_type = e.get("type", "") if e_type == "grammar": errors_lines.append( f"- [Грамматика {e.get('unit_title', '')}]: ответ студента: '{e.get('user_answer', '')}', " f"правильно: '{e.get('correct_answer', '')}'. Формула: {e.get('formula', '')}. {e.get('explanation', '')}" ) elif e_type == "vocab": errors_lines.append( f"- [Слово]: '{e.get('term', '')}' ({e.get('translation', '')}) — ошибка в карточке." ) elif e_type == "listening": errors_lines.append( f"- [Аудирование]: Вопрос '{e.get('question', '')}', ответ студента: '{e.get('user_answer', '')}', " f"правильный ответ: '{e.get('ideal_answer', '')}'" ) errors_text = "\n".join(errors_lines) if errors_lines else "Ошибок нет, 100% правильных ответов." # Format historical context chronic_vocab = history_data.get("chronic_vocab_errors", []) past_grammar = history_data.get("past_grammar_errors", []) history_lines = [] if chronic_vocab: history_lines.append(f"- Слова с частыми ошибками в прошлом: {', '.join(chronic_vocab[:5])}") if past_grammar: history_lines.append(f"- Юниты грамматики с низкой точностью в прошлом: {', '.join(map(str, past_grammar[:3]))}") history_text = "\n".join(history_lines) if history_lines else "История чистая или ошибок ранее не зафиксировано." prompt = ( "Ты — чуткий, высококвалифицированный преподаватель английского языка по методике Raymond Murphy.\n" "Студент только что завершил комплексный урок. Проанализируй его результаты, свяжи ошибки с грамматическими формулами " "и составь лаконичный, мотивирующий и практически полезный педагогический разбор на русском языке.\n\n" f"Урок: {preset_name}\n" f"Длительность: {duration_min} мин\n" f"Итоговый балл: {total_score}%\n\n" f"Пройденные блоки:\n{steps_summary_text}\n\n" f"Ошибки в этой сессии:\n{errors_text}\n\n" f"История прошлых трудностей:\n{history_text}\n\n" "Требования к ответу:\n" "1. Ответь ИСКЛЮЧИТЕЛЬНО в формате JSON.\n" "2. Поле 'headline': краткий, воодушевляющий заголовок (1 предложение).\n" "3. Поле 'strengths': список из 1-3 пунктов, что получилось лучше всего.\n" "4. Поле 'error_patterns': список из 1-3 пунктов с объяснением допущенных ошибок (если были) через контраст и формулы Мерфи. Если ошибок не было — похвали за точность.\n" "5. Поле 'tomorrow_focus': список из 2-3 конкретных рекомендаций на завтра.\n\n" "Формат JSON:\n" "{\n" f' "grade": "{"A+" if total_score >= 90 else ("A" if total_score >= 75 else ("B" if total_score >= 60 else "C"))}",\n' f' "score": {total_score},\n' ' "headline": "заголовок",\n' ' "strengths": ["пункт 1", "пункт 2"],\n' ' "error_patterns": ["разбор ошибки 1", "разбор ошибки 2"],\n' ' "tomorrow_focus": ["совет 1", "совет 2"]\n' "}" ) payload = { "model": config.OLLAMA_MODEL, "prompt": prompt, "stream": False, "format": "json", "keep_alive": "24h", "options": { "temperature": 0.3, "num_predict": 450, }, } try: async with httpx.AsyncClient(timeout=25.0) as client: resp = await client.post(f"{config.OLLAMA_BASE_URL}/api/generate", json=payload) if resp.status_code == 200: raw_response = resp.json().get("response", "{}") # Parse 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] data = json.loads(clean_raw.strip()) return { "grade": data.get("grade", "A" if total_score >= 75 else "B"), "score": total_score, "headline": data.get("headline", "Отличная сессия!"), "strengths": data.get("strengths", ["Хорошая работа на уроке"]), "error_patterns": data.get("error_patterns", []), "tomorrow_focus": data.get("tomorrow_focus", ["Продолжайте ежедневные занятия"]), "source": f"ollama:{config.OLLAMA_MODEL}", } except Exception as e: logger.warning(f"Ollama debrief generation deferred to deterministic engine: {e}") # Fallback to deterministic debrief return generate_deterministic_debrief(session_data, history_data)