"""Сервис генерации слова дня""" 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. Использует batch-генерацию: 1 запрос на язык вместо 1 запроса на уровень. 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 # Определяем какие уровни ещё не сгенерированы levels_to_generate = [] for level in levels: existing = await self._get_word_for_date(session, today, lang, level) if not existing: levels_to_generate.append(level) else: logger.debug(f"Слово дня уже существует: {lang}/{level}") if not levels_to_generate: logger.info(f"Все слова для {lang} уже сгенерированы") continue # Собираем исключения для каждого уровня excluded_words = {} for level in levels_to_generate: excluded_words[level] = await self._get_recent_words( session, lang, level, days=30 ) # Генерируем все слова одним запросом try: batch_result = await ai_service.generate_words_of_day_batch( language=lang, levels=levels_to_generate, translation_lang="ru", excluded_words=excluded_words ) if not batch_result: results["errors"] += len(levels_to_generate) logger.error(f"Не удалось сгенерировать слова для {lang}") continue # Сохраняем каждое слово for level in levels_to_generate: word_data = batch_result.get(level) if word_data and word_data.get("word"): 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) results[lang] += 1 logger.info( f"Сгенерировано слово дня: {word_data.get('word')} " f"({lang}/{level})" ) else: results["errors"] += 1 logger.warning(f"Нет данных для {lang}/{level} в ответе AI") await session.commit() except Exception as e: results["errors"] += len(levels_to_generate) logger.error(f"Ошибка batch-генерации для {lang}: {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()