"""Сервис генерации слова дня""" import logging from datetime import datetime, date from typing import Optional, Dict, List from sqlalchemy import select, and_ from sqlalchemy.ext.asyncio import AsyncSession from database.db import async_session_maker from database.models import WordOfDay, LanguageLevel, JLPTLevel, JLPT_LANGUAGES from services.ai_service import ai_service logger = logging.getLogger(__name__) # Уровни для каждого языка CEFR_LEVELS = [level.value for level in LanguageLevel] # A1-C2 JLPT_LEVELS = [level.value for level in JLPTLevel] # N5-N1 # Языки для генерации LEARNING_LANGUAGES = ["en", "ja"] class WordOfDayService: """Сервис для генерации и получения слова дня""" async def generate_all_words_for_today(self) -> Dict[str, int]: """ Генерация слов дня для всех языков и уровней. Вызывается в 00:00 UTC. Returns: Dict с количеством сгенерированных слов по языкам """ today = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0) results = {"en": 0, "ja": 0, "errors": 0} async with async_session_maker() as session: for lang in LEARNING_LANGUAGES: levels = JLPT_LEVELS if lang in JLPT_LANGUAGES else CEFR_LEVELS for level in levels: try: # Проверяем, не сгенерировано ли уже existing = await self._get_word_for_date( session, today, lang, level ) if existing: logger.debug( f"Слово дня уже существует: {lang}/{level}" ) continue # Получаем список недавних слов для исключения excluded = await self._get_recent_words(session, lang, level, days=30) # Генерируем слово word_data = await ai_service.generate_word_of_day( level=level, learning_lang=lang, translation_lang="ru", # Базовый перевод на русский excluded_words=excluded ) if word_data: word_of_day = WordOfDay( word=word_data.get("word", ""), transcription=word_data.get("transcription"), translation=word_data.get("translation", ""), examples=word_data.get("examples"), synonyms=word_data.get("synonyms"), etymology=word_data.get("etymology"), learning_lang=lang, level=level, date=today ) session.add(word_of_day) await session.commit() results[lang] += 1 logger.info( f"Сгенерировано слово дня: {word_data.get('word')} " f"({lang}/{level})" ) else: results["errors"] += 1 logger.warning( f"Не удалось сгенерировать слово для {lang}/{level}" ) except Exception as e: results["errors"] += 1 logger.error( f"Ошибка генерации слова для {lang}/{level}: {e}" ) total = results["en"] + results["ja"] logger.info( f"Генерация слов дня завершена: всего={total}, " f"en={results['en']}, ja={results['ja']}, ошибок={results['errors']}" ) return results async def get_word_of_day( self, learning_lang: str, level: str, target_date: Optional[datetime] = None ) -> Optional[WordOfDay]: """ Получить слово дня для языка и уровня. Args: learning_lang: Язык изучения (en/ja) level: Уровень (A1-C2 или N5-N1) target_date: Дата (по умолчанию сегодня) Returns: WordOfDay или None """ if target_date is None: target_date = datetime.utcnow().replace( hour=0, minute=0, second=0, microsecond=0 ) else: target_date = target_date.replace( hour=0, minute=0, second=0, microsecond=0 ) async with async_session_maker() as session: return await self._get_word_for_date( session, target_date, learning_lang, level ) async def _get_word_for_date( self, session: AsyncSession, target_date: datetime, learning_lang: str, level: str ) -> Optional[WordOfDay]: """Получить слово из БД для конкретной даты""" result = await session.execute( select(WordOfDay).where( and_( WordOfDay.date == target_date, WordOfDay.learning_lang == learning_lang, WordOfDay.level == level ) ) ) return result.scalar_one_or_none() async def _get_recent_words( self, session: AsyncSession, learning_lang: str, level: str, days: int = 30 ) -> List[str]: """Получить список недавних слов для исключения""" from datetime import timedelta cutoff_date = datetime.utcnow() - timedelta(days=days) result = await session.execute( select(WordOfDay.word).where( and_( WordOfDay.learning_lang == learning_lang, WordOfDay.level == level, WordOfDay.date >= cutoff_date ) ) ) return [row[0] for row in result.fetchall()] async def format_word_for_user( self, word: WordOfDay, translation_lang: str = "ru", ui_lang: str = None ) -> str: """ Форматировать слово дня для отображения пользователю. Args: word: Объект WordOfDay translation_lang: Язык перевода для пользователя ui_lang: Язык интерфейса (для локализации заголовков) Returns: Отформатированная строка """ from utils.i18n import t lang = ui_lang or translation_lang or "ru" lines = [] # Заголовок со словом if word.transcription: lines.append(f"📚 {word.word} [{word.transcription}]") else: lines.append(f"📚 {word.word}") # Перевод lines.append(f"📝 {word.translation}") # Синонимы if word.synonyms: lines.append(f"\n🔄 {t(lang, 'wod.synonyms')}: {word.synonyms}") # Примеры if word.examples: lines.append(f"\n📖 {t(lang, 'wod.examples')}:") for i, example in enumerate(word.examples[:3], 1): sentence = example.get("sentence", "") translation = example.get("translation", "") lines.append(f" {i}. {sentence}") if translation: lines.append(f" {translation}") # Этимология/интересный факт if word.etymology: lines.append(f"\n💡 {word.etymology}") return "\n".join(lines) # Глобальный экземпляр сервиса wordofday_service = WordOfDayService()