""" Improved daily handlers with enhanced hint system. This module provides better hint functionality that: 1. Tracks hint count per day 2. Provides different levels of hints (basic -> detailed -> microsteps) 3. Reformulates responses when no more detail is possible """ import asyncio import contextlib import logging from typing import Optional, List, Dict, Any from aiogram import Router, F from aiogram.types import CallbackQuery from aiogram.enums import ChatAction from aiogram.fsm.context import FSMContext from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from src.bot.keyboards.daily import kb_daily_step, kb_fail_reason from src.bot.utils.badges import get_badge_message from src.database.session import SessionLocal from src.database.db_methods.strategy import ( get_active_strategy_with_steps_by_tg, complete_today, snooze_today, ask_hint_today, fail_today, ensure_today_issued, get_due_for_daily_issue, ) from src.database.models import User, Strategy, StrategyDayState from src.database.enums import FailReason, DayStatus from src.llm.openai_client import ask_json logger = logging.getLogger(__name__) router = Router(name="daily_improved") async def _typing(bot, chat_id: int, seconds: int = 2): """Show typing indicator for specified duration.""" await bot.send_chat_action(chat_id, ChatAction.TYPING) await asyncio.sleep(seconds) async def _get_hint_count(session: AsyncSession, strategy_id: int, day: int) -> int: """ Get the number of hints requested for this day. Since we don't have a hint_count field, we'll use a simple approach: count how many times the status was set to 'hint' for this day. """ # For now, we'll use a simple counter stored in the session # In a real implementation, you might want to add a hint_count field to StrategyDayState return getattr(session, '_hint_count', {}).get(f"{strategy_id}_{day}", 0) async def _increment_hint_count(session: AsyncSession, strategy_id: int, day: int) -> int: """Increment and return the hint count for this day.""" if not hasattr(session, '_hint_count'): session._hint_count = {} key = f"{strategy_id}_{day}" current_count = session._hint_count.get(key, 0) session._hint_count[key] = current_count + 1 return session._hint_count[key] def _get_hint_prompt_level(hint_count: int) -> Dict[str, Any]: """ Get prompt configuration based on hint count. Args: hint_count: Number of hints already requested (0-based) Returns: Dictionary with prompt configuration """ if hint_count == 0: # First hint: Basic step-by-step plan return { "system": ( "Ты строгий практичный методист. Отвечай строго на русском. " "Не меняй формулировку задания и не добавляй новое. " "Верни короткий ПОШАГОВЫЙ план выполнения текущего задания на сегодня. " "Каждый шаг начинается глаголом, конкретно и проверяемо. " "Запрещены мотивационные фразы, вода, «поддержка», оценочные суждения." ), "user_template": ( "Нужно помочь выполнить текущее задание. " "Верни СТРОГО один JSON без пояснений:\n" "{\n" ' "plan": ["Шаг 1 — ...", "Шаг 2 — ...", "Шаг 3 — ...", "Шаг 4 — ..."]\n' "}\n" "ТРЕБУЕТСЯ: 3–6 конкретных шагов, без микрошагов, без изменения задачи.\n" "ТЕКУЩЕЕ ЗАДАНИЕ (НЕ МЕНЯТЬ):\n" "{payload}" ), "temperature": 0.1, "max_tokens": 500 } elif hint_count == 1: # Second hint: More detailed steps with micro-actions return { "system": ( "Ты детальный практический консультант. Отвечай строго на русском. " "Разбей задание на МЕЛКИЕ конкретные действия. " "Каждый шаг должен быть выполним за 5-15 минут. " "Добавь конкретные примеры и подсказки. " "Не меняй суть задания, но сделай его максимально понятным." ), "user_template": ( "Пользователь просит подсказку повторно. Нужно РАЗЖЕВАТЬ задание. " "Верни СТРОГО один JSON:\n" "{\n" ' "detailed_plan": [\n' ' {\n' ' "step": "Шаг 1 — ...",\n' ' "microsteps": ["Микро-действие 1", "Микро-действие 2"],\n' ' "example": "Пример: ..."\n' ' }\n' ' ]\n' "}\n" "ТРЕБУЕТСЯ: 2-4 детальных шага с микро-действиями и примерами.\n" "ТЕКУЩЕЕ ЗАДАНИЕ:\n" "{payload}" ), "temperature": 0.2, "max_tokens": 800 } elif hint_count == 2: # Third hint: Ultra-detailed with alternatives and troubleshooting return { "system": ( "Ты эксперт-наставник. Отвечай строго на русском. " "Дай МАКСИМАЛЬНО детальный план с альтернативами. " "Включи возможные проблемы и их решения. " "Каждое действие должно быть кристально ясным." ), "user_template": ( "Пользователь просит подсказку в третий раз. Нужно МАКСИМАЛЬНО разжевать. " "Верни СТРОГО один JSON:\n" "{\n" ' "ultra_detailed": [\n' ' {\n' ' "main_action": "Главное действие",\n' ' "step_by_step": ["Действие 1", "Действие 2", "Действие 3"],\n' ' "alternatives": ["Альтернатива 1", "Альтернатива 2"],\n' ' "troubleshooting": "Что делать если не получается: ...",\n' ' "example": "Конкретный пример: ..."\n' ' }\n' ' ]\n' "}\n" "ТЕКУЩЕЕ ЗАДАНИЕ:\n" "{payload}" ), "temperature": 0.3, "max_tokens": 1200 } else: # Fourth+ hint: Reformulate and encourage return { "system": ( "Ты поддерживающий коуч. Отвечай строго на русском. " "Переформулируй задание другими словами, но сохрани суть. " "Добавь мотивацию и покажи, что задание выполнимо. " "Используй другой подход к объяснению." ), "user_template": ( "Пользователь просит подсказку уже {hint_count} раз. " "Переформулируй задание по-другому, но сохрани суть. " "Верни СТРОГО один JSON:\n" "{\n" ' "reformulated": [\n' ' {\n' ' "new_approach": "Новый подход к заданию",\n' ' "simple_steps": ["Простой шаг 1", "Простой шаг 2"],\n' ' "motivation": "Почему это важно и выполнимо",\n' ' "encouragement": "Поддерживающие слова"\n' ' }\n' ' ]\n' "}\n" "ТЕКУЩЕЕ ЗАДАНИЕ:\n" "{payload}" ), "temperature": 0.4, "max_tokens": 600 } def _format_hint_response(hint_json: Dict[str, Any], hint_count: int) -> str: """ Format the hint response based on the hint count and JSON structure. Args: hint_json: JSON response from LLM hint_count: Number of hints requested (0-based) Returns: Formatted hint text """ if hint_count == 0: # Basic plan format plan = hint_json.get("plan", []) if not isinstance(plan, list): plan = [] plan = [str(t).strip() for t in plan if str(t).strip()][:6] if plan: lines = ["План выполнения текущего шага:"] for i, p in enumerate(plan, 1): lines.append(f"{i}. {p}") return "\n".join(lines) else: return "Сконцентрируйся на одном действии и последовательно выполни его по описанию задания." elif hint_count == 1: # Detailed plan format detailed_plan = hint_json.get("detailed_plan", []) if not isinstance(detailed_plan, list): detailed_plan = [] if detailed_plan: lines = ["Детальный план выполнения:"] for i, step_data in enumerate(detailed_plan, 1): if isinstance(step_data, dict): step = step_data.get("step", "") microsteps = step_data.get("microsteps", []) example = step_data.get("example", "") lines.append(f"\n{i}. {step}") if microsteps: for micro in microsteps: lines.append(f" • {micro}") if example: lines.append(f" 💡 {example}") else: lines.append(f"{i}. {str(step_data)}") return "\n".join(lines) else: return "Разбей задание на мелкие шаги и выполняй их по очереди." elif hint_count == 2: # Ultra-detailed format ultra_detailed = hint_json.get("ultra_detailed", []) if not isinstance(ultra_detailed, list): ultra_detailed = [] if ultra_detailed: lines = ["Максимально детальный план:"] for i, detail_data in enumerate(ultra_detailed, 1): if isinstance(detail_data, dict): main_action = detail_data.get("main_action", "") step_by_step = detail_data.get("step_by_step", []) alternatives = detail_data.get("alternatives", []) troubleshooting = detail_data.get("troubleshooting", "") example = detail_data.get("example", "") lines.append(f"\n{i}. {main_action}") if step_by_step: lines.append(" Пошагово:") for step in step_by_step: lines.append(f" • {step}") if alternatives: lines.append(" Альтернативы:") for alt in alternatives: lines.append(f" • {alt}") if troubleshooting: lines.append(f" 🔧 Если не получается: {troubleshooting}") if example: lines.append(f" 💡 Пример: {example}") else: lines.append(f"{i}. {str(detail_data)}") return "\n".join(lines) else: return "Выполняй задание пошагово, не спеши. Каждый шаг важен." else: # Reformulated format reformulated = hint_json.get("reformulated", []) if not isinstance(reformulated, list): reformulated = [] if reformulated: lines = ["Новый взгляд на задание:"] for i, reform_data in enumerate(reformulated, 1): if isinstance(reform_data, dict): new_approach = reform_data.get("new_approach", "") simple_steps = reform_data.get("simple_steps", []) motivation = reform_data.get("motivation", "") encouragement = reform_data.get("encouragement", "") lines.append(f"\n{i}. {new_approach}") if simple_steps: lines.append(" Простые шаги:") for step in simple_steps: lines.append(f" • {step}") if motivation: lines.append(f" 💪 {motivation}") if encouragement: lines.append(f" 🌟 {encouragement}") else: lines.append(f"{i}. {str(reform_data)}") return "\n".join(lines) else: return "Попробуй подойти к заданию с другой стороны. Ты справишься! 💪" HINT_FLAG_KEY = "hint_busy_day" @router.callback_query(F.data == "step:hint") async def step_hint_improved(cb: CallbackQuery, state: FSMContext): """ Защита от повторных нажатий: - первый клик проходит - последующие клики до завершения не делают ничего """ tg_id = str(cb.from_user.id) # грузим активный день заранее чтобы знать ключ блокировки async with SessionLocal() as session: strat, steps, day = await get_active_strategy_with_steps_by_tg(session, tg_id) # если стратегии вообще нет то просто ответим и выйдем if not strat: await cb.answer("Активная стратегия не найдена. Нажми /start и пройди анкету.", show_alert=True) return # читаем из FSM флаг занятости именно для этого дня data = await state.get_data() busy_info = data.get(HINT_FLAG_KEY, {}) # dict вида { "12": True } if busy_info.get(str(day)): # уже обрабатывается. просто гасим "часики" и уходим. await cb.answer() return # ставим флаг "занят" busy_info[str(day)] = True await state.update_data({HINT_FLAG_KEY: busy_info}) # сразу отвечаем на callback чтобы убрать часы await cb.answer() # можно попытаться убрать клавиатуру, но даже если не сработает это не критично # важный момент: мы не будем падать из-за этого try: original_text = cb.message.html_text or cb.message.text or "" await cb.message.bot.edit_message_text( chat_id=cb.message.chat.id, message_id=cb.message.message_id, text=original_text, reply_markup=None, parse_mode="HTML" ) except Exception as e: logger.debug("edit_message_text failed (keyboard may still be visible): %r", e) # typing await _typing(cb.message.bot, cb.message.chat.id, 2) # дальше вся логика генерации подсказки async with SessionLocal() as session: # strat и day у нас уже есть но нам еще нужно steps актуально из БД # и надо инкремент хинтов и т.д., поэтому повторим нормально strat, steps, day = await get_active_strategy_with_steps_by_tg(session, tg_id) step_html = steps.get(str(day)) if steps else "" if isinstance(step_html, dict): step_html = step_html.get("html", "") or step_html.get("text", "") hint_count = await _get_hint_count(session, strat.id, day) new_hint_count = await _increment_hint_count(session, strat.id, day) await ask_hint_today(session, strat.id) await session.commit() prompt_config = _get_hint_prompt_level(hint_count) payload = { "strategy": (strat.strategy or "")[:3500], "today_step_html": (step_html or "")[:2000], } try: hint_json = await ask_json( system_prompt=prompt_config["system"], user_prompt=prompt_config["user_template"].format( payload=payload, hint_count=new_hint_count, ), temperature=prompt_config["temperature"], max_tokens=prompt_config["max_tokens"], timeout_s=30, ) except Exception as e: logger.exception("Error getting hint from LLM: %s", e) if hint_count == 0: fallback = ( "План выполнения текущего шага:\n" "1. Прочитай задание внимательно\n" "2. Определи первое действие\n" "3. Выполни его пошагово" ) elif hint_count == 1: fallback = ( "Детальный план:\n" "1. Разбей задание на мелкие части\n" "2. Выполняй по одной части\n" "3. Не спеши, качество важнее скорости" ) elif hint_count == 2: fallback = ( "Максимально детально:\n" "1. Представь, что объясняешь ребенку\n" "2. Каждое действие опиши подробно\n" "3. Если что-то не получается - попробуй по-другому" ) else: fallback = ( "Новый подход:\n" "1. Посмотри на задание свежим взглядом\n" "2. Попробуй другой способ\n" "3. Помни: ты справишься!" ) hint_json = {"fallback": fallback} text = _format_hint_response(hint_json, hint_count) # отправляем ответ с новой клавой (клавиатура для следующего шага/действия) await cb.message.answer( text, reply_markup=kb_daily_step() ) # снимаем блокировку после завершения генерации # чтобы он мог запросить новый хинт позже ещё раз data = await state.get_data() busy_info = data.get(HINT_FLAG_KEY, {}) busy_info[str(day)] = False await state.update_data({HINT_FLAG_KEY: busy_info}) # Export the improved router __all__ = ['router']