Files
notes-qpid-pro/app/ai.py
T

332 lines
15 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
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",
"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,
}