mamonov.ep
16a7df0343
- Добавлена поддержка персональных AI моделей для каждого пользователя - Оптимизация создания заданий: батч-запрос к AI вместо N запросов - Фильтрация слов по языку изучения (source_lang) в словаре - Удалены неиспользуемые колонки examples и category из vocabulary - Миграции для ai_model_id и удаления колонок 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
371 lines
13 KiB
Python
371 lines
13 KiB
Python
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from database.models import Vocabulary, WordSource, LanguageLevel, WordTranslation
|
||
from typing import List, Optional, Dict
|
||
import re
|
||
|
||
|
||
class VocabularyService:
|
||
"""Сервис для работы со словарным запасом"""
|
||
|
||
@staticmethod
|
||
async def add_word(
|
||
session: AsyncSession,
|
||
user_id: int,
|
||
word_original: str,
|
||
word_translation: str,
|
||
source_lang: Optional[str] = None,
|
||
translation_lang: Optional[str] = None,
|
||
transcription: Optional[str] = None,
|
||
difficulty_level: Optional[str] = None,
|
||
source: WordSource = WordSource.MANUAL,
|
||
notes: Optional[str] = None
|
||
) -> Vocabulary:
|
||
"""
|
||
Добавить слово в словарь пользователя
|
||
|
||
Args:
|
||
session: Сессия базы данных
|
||
user_id: ID пользователя
|
||
word_original: Оригинальное слово
|
||
word_translation: Перевод
|
||
transcription: Транскрипция
|
||
difficulty_level: Уровень сложности
|
||
source: Источник добавления
|
||
notes: Заметки пользователя
|
||
|
||
Returns:
|
||
Созданный объект слова
|
||
"""
|
||
# Преобразование difficulty_level в enum
|
||
difficulty_enum = None
|
||
if difficulty_level:
|
||
try:
|
||
difficulty_enum = LanguageLevel[difficulty_level]
|
||
except KeyError:
|
||
difficulty_enum = None
|
||
|
||
new_word = Vocabulary(
|
||
user_id=user_id,
|
||
word_original=word_original,
|
||
word_translation=word_translation,
|
||
source_lang=source_lang,
|
||
translation_lang=translation_lang,
|
||
transcription=transcription,
|
||
difficulty_level=difficulty_enum,
|
||
source=source,
|
||
notes=notes
|
||
)
|
||
|
||
session.add(new_word)
|
||
await session.commit()
|
||
await session.refresh(new_word)
|
||
|
||
return new_word
|
||
|
||
@staticmethod
|
||
@staticmethod
|
||
def _is_japanese(text: str) -> bool:
|
||
if not text:
|
||
return False
|
||
return re.search(r"[\u3040-\u30FF\u3400-\u4DBF\u4E00-\u9FFF]", text) is not None
|
||
|
||
@staticmethod
|
||
def _filter_by_learning_lang(words: List[Vocabulary], learning_lang: Optional[str]) -> List[Vocabulary]:
|
||
if not learning_lang:
|
||
return words
|
||
# Если в БД указан source_lang – фильтруем по нему.
|
||
with_lang = [w for w in words if getattr(w, 'source_lang', None)]
|
||
if with_lang:
|
||
return [w for w in words if (w.source_lang or '').lower() == learning_lang.lower()]
|
||
# Фолбэк-эвристика для японского, если язык не сохранён
|
||
if learning_lang.lower() == 'ja':
|
||
return [w for w in words if VocabularyService._is_japanese(w.word_original)]
|
||
return [w for w in words if not VocabularyService._is_japanese(w.word_original)]
|
||
|
||
@staticmethod
|
||
async def get_user_words(
|
||
session: AsyncSession,
|
||
user_id: int,
|
||
limit: int = 50,
|
||
offset: int = 0,
|
||
learning_lang: Optional[str] = None
|
||
) -> List[Vocabulary]:
|
||
"""
|
||
Получить слова пользователя с пагинацией
|
||
|
||
Args:
|
||
session: Сессия базы данных
|
||
user_id: ID пользователя
|
||
limit: Максимальное количество слов
|
||
offset: Смещение для пагинации
|
||
|
||
Returns:
|
||
Список слов пользователя
|
||
"""
|
||
result = await session.execute(
|
||
select(Vocabulary)
|
||
.where(Vocabulary.user_id == user_id)
|
||
.order_by(Vocabulary.created_at.desc())
|
||
)
|
||
words = list(result.scalars().all())
|
||
words = VocabularyService._filter_by_learning_lang(words, learning_lang)
|
||
return words[offset:offset + limit]
|
||
|
||
@staticmethod
|
||
async def get_words_count(session: AsyncSession, user_id: int, learning_lang: Optional[str] = None) -> int:
|
||
"""
|
||
Получить количество слов в словаре пользователя
|
||
|
||
Args:
|
||
session: Сессия базы данных
|
||
user_id: ID пользователя
|
||
|
||
Returns:
|
||
Количество слов
|
||
"""
|
||
result = await session.execute(
|
||
select(Vocabulary).where(Vocabulary.user_id == user_id)
|
||
)
|
||
words = list(result.scalars().all())
|
||
words = VocabularyService._filter_by_learning_lang(words, learning_lang)
|
||
return len(words)
|
||
|
||
@staticmethod
|
||
async def find_word(
|
||
session: AsyncSession,
|
||
user_id: int,
|
||
word: str,
|
||
source_lang: Optional[str] = None
|
||
) -> Optional[Vocabulary]:
|
||
"""
|
||
Найти слово в словаре пользователя
|
||
|
||
Args:
|
||
session: Сессия базы данных
|
||
user_id: ID пользователя
|
||
word: Слово для поиска
|
||
source_lang: Язык изучения для фильтрации (если указан)
|
||
|
||
Returns:
|
||
Объект слова или None
|
||
"""
|
||
query = (
|
||
select(Vocabulary)
|
||
.where(Vocabulary.user_id == user_id)
|
||
.where(Vocabulary.word_original.ilike(f"%{word}%"))
|
||
)
|
||
if source_lang:
|
||
query = query.where(Vocabulary.source_lang == source_lang.lower())
|
||
result = await session.execute(query)
|
||
return result.scalar_one_or_none()
|
||
|
||
@staticmethod
|
||
async def get_word_by_original(
|
||
session: AsyncSession,
|
||
user_id: int,
|
||
word: str,
|
||
source_lang: Optional[str] = None
|
||
) -> Optional[Vocabulary]:
|
||
"""
|
||
Получить слово по точному совпадению
|
||
|
||
Args:
|
||
session: Сессия базы данных
|
||
user_id: ID пользователя
|
||
word: Слово для поиска (точное совпадение)
|
||
source_lang: Язык изучения для фильтрации (если указан)
|
||
|
||
Returns:
|
||
Объект слова или None
|
||
"""
|
||
query = (
|
||
select(Vocabulary)
|
||
.where(Vocabulary.user_id == user_id)
|
||
.where(Vocabulary.word_original == word.lower())
|
||
)
|
||
if source_lang:
|
||
query = query.where(Vocabulary.source_lang == source_lang.lower())
|
||
result = await session.execute(query)
|
||
return result.scalar_one_or_none()
|
||
|
||
@staticmethod
|
||
async def get_all_user_word_strings(
|
||
session: AsyncSession,
|
||
user_id: int,
|
||
learning_lang: Optional[str] = None
|
||
) -> List[str]:
|
||
"""
|
||
Получить список всех слов пользователя (только строки)
|
||
|
||
Args:
|
||
session: Сессия базы данных
|
||
user_id: ID пользователя
|
||
learning_lang: Язык изучения для фильтрации
|
||
|
||
Returns:
|
||
Список строк — оригинальных слов
|
||
"""
|
||
result = await session.execute(
|
||
select(Vocabulary)
|
||
.where(Vocabulary.user_id == user_id)
|
||
)
|
||
words = list(result.scalars().all())
|
||
words = VocabularyService._filter_by_learning_lang(words, learning_lang)
|
||
return [w.word_original.lower() for w in words]
|
||
|
||
# === Методы для работы с переводами ===
|
||
|
||
@staticmethod
|
||
async def add_translation(
|
||
session: AsyncSession,
|
||
vocabulary_id: int,
|
||
translation: str,
|
||
context: Optional[str] = None,
|
||
context_translation: Optional[str] = None,
|
||
is_primary: bool = False
|
||
) -> WordTranslation:
|
||
"""
|
||
Добавить перевод к слову
|
||
|
||
Args:
|
||
session: Сессия базы данных
|
||
vocabulary_id: ID слова в словаре
|
||
translation: Перевод
|
||
context: Пример предложения на языке изучения
|
||
context_translation: Перевод примера
|
||
is_primary: Является ли основным переводом
|
||
|
||
Returns:
|
||
Созданный объект перевода
|
||
"""
|
||
# Если это основной перевод, снимаем флаг с других
|
||
if is_primary:
|
||
result = await session.execute(
|
||
select(WordTranslation)
|
||
.where(WordTranslation.vocabulary_id == vocabulary_id)
|
||
.where(WordTranslation.is_primary == True)
|
||
)
|
||
for existing in result.scalars().all():
|
||
existing.is_primary = False
|
||
|
||
new_translation = WordTranslation(
|
||
vocabulary_id=vocabulary_id,
|
||
translation=translation,
|
||
context=context,
|
||
context_translation=context_translation,
|
||
is_primary=is_primary
|
||
)
|
||
|
||
session.add(new_translation)
|
||
await session.commit()
|
||
await session.refresh(new_translation)
|
||
|
||
return new_translation
|
||
|
||
@staticmethod
|
||
async def add_translations_bulk(
|
||
session: AsyncSession,
|
||
vocabulary_id: int,
|
||
translations: List[Dict]
|
||
) -> List[WordTranslation]:
|
||
"""
|
||
Добавить несколько переводов к слову
|
||
|
||
Args:
|
||
session: Сессия базы данных
|
||
vocabulary_id: ID слова
|
||
translations: Список словарей с переводами
|
||
[{"translation": "...", "context": "...", "context_translation": "...", "is_primary": bool}]
|
||
|
||
Returns:
|
||
Список созданных переводов
|
||
"""
|
||
created = []
|
||
for i, tr_data in enumerate(translations):
|
||
new_translation = WordTranslation(
|
||
vocabulary_id=vocabulary_id,
|
||
translation=tr_data.get('translation', ''),
|
||
context=tr_data.get('context'),
|
||
context_translation=tr_data.get('context_translation'),
|
||
is_primary=tr_data.get('is_primary', i == 0) # Первый по умолчанию основной
|
||
)
|
||
session.add(new_translation)
|
||
created.append(new_translation)
|
||
|
||
await session.commit()
|
||
for tr in created:
|
||
await session.refresh(tr)
|
||
|
||
return created
|
||
|
||
@staticmethod
|
||
async def get_word_translations(
|
||
session: AsyncSession,
|
||
vocabulary_id: int
|
||
) -> List[WordTranslation]:
|
||
"""
|
||
Получить все переводы слова
|
||
|
||
Args:
|
||
session: Сессия базы данных
|
||
vocabulary_id: ID слова
|
||
|
||
Returns:
|
||
Список переводов
|
||
"""
|
||
result = await session.execute(
|
||
select(WordTranslation)
|
||
.where(WordTranslation.vocabulary_id == vocabulary_id)
|
||
.order_by(WordTranslation.is_primary.desc(), WordTranslation.created_at)
|
||
)
|
||
return list(result.scalars().all())
|
||
|
||
@staticmethod
|
||
async def get_primary_translation(
|
||
session: AsyncSession,
|
||
vocabulary_id: int
|
||
) -> Optional[WordTranslation]:
|
||
"""
|
||
Получить основной перевод слова
|
||
|
||
Args:
|
||
session: Сессия базы данных
|
||
vocabulary_id: ID слова
|
||
|
||
Returns:
|
||
Основной перевод или None
|
||
"""
|
||
result = await session.execute(
|
||
select(WordTranslation)
|
||
.where(WordTranslation.vocabulary_id == vocabulary_id)
|
||
.where(WordTranslation.is_primary == True)
|
||
)
|
||
return result.scalar_one_or_none()
|
||
|
||
@staticmethod
|
||
async def delete_translation(
|
||
session: AsyncSession,
|
||
translation_id: int
|
||
) -> bool:
|
||
"""
|
||
Удалить перевод
|
||
|
||
Args:
|
||
session: Сессия базы данных
|
||
translation_id: ID перевода
|
||
|
||
Returns:
|
||
True если удалено, False если не найдено
|
||
"""
|
||
result = await session.execute(
|
||
select(WordTranslation).where(WordTranslation.id == translation_id)
|
||
)
|
||
translation = result.scalar_one_or_none()
|
||
|
||
if translation:
|
||
await session.delete(translation)
|
||
await session.commit()
|
||
return True
|
||
return False
|