# src/bot/handlers/state_handlers/survey.py # All comments in English. import asyncio import contextlib import re import os import io import mimetypes import logging from typing import Dict, Any, Tuple, Optional import httpx from aiogram import Router, F from aiogram.enums import ChatAction from aiogram.filters import CommandStart, Command from aiogram.types import Message, CallbackQuery, InlineKeyboardMarkup, InlineKeyboardButton from aiogram.fsm.context import FSMContext from aiogram.fsm.state import StatesGroup, State from sqlalchemy import delete from sqlalchemy.ext.asyncio import AsyncSession from src.bot.keyboards.main import kb_confirm_wipe, kb_strategy_edit_confirm, kb_strategy_overwrite_confirm from src.bot.utils.prompts import strategy_prompt from src.database.models import StrategyDayState, Strategy from src.database.session import SessionLocal from src.database.db_methods.users import ( is_user_registered, register_user, get_user_by_tg, delete_user_by_tg, ) from src.database.enums import ( Gender, PracticePlace, ClientsExperience, MassageTechnique, SocialSkill, CommunicationEase, StrategyStatus, ) from src.bot.states.survey import SurveyFSM from src.bot.keyboards.survey import ( kb_gender, kb_place, kb_clients, kb_technique, kb_social, kb_comm, ) from src.bot.keyboards.daily import kb_daily_step from src.database.db_methods.strategy import ( create_strategy, ensure_today_issued, _ensure_day_state, _max_day, _get_strategy, get_active_strategy ) from src.llm.openai_client import ask_json router = Router(name="survey_router") logger = logging.getLogger(__name__) # ───────── Review FSM ───────── class ReviewFSM(StatesGroup): await_feedback = State() # waiting for user's corrections (text or voice) # ───────── helpers: session, typing ───────── @contextlib.asynccontextmanager async def session_scope() -> AsyncSession: session = SessionLocal() try: yield session await session.commit() except Exception: await session.rollback() raise finally: await session.close() async def _typing_loop(bot, chat_id: int, stop_event: asyncio.Event, action: ChatAction = ChatAction.TYPING): try: while not stop_event.is_set(): await bot.send_chat_action(chat_id, action) await asyncio.sleep(4) except Exception: pass # ───────── UI: review keyboard ───────── def kb_review() -> InlineKeyboardMarkup: return InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="✅ Подтверждаю", callback_data="review:accept")], [InlineKeyboardButton(text="✍️ Хочу внести правки", callback_data="review:edit")], ]) # ───────── text utils ───────── def _safe_fill(template: str, mapping: Dict[str, Any]) -> str: out = template for k, v in mapping.items(): out = out.replace("{" + k + "}", str(v) if v is not None else "") return out def _strategy_without_day_blocks(text: str) -> str: """Trim possible '<Шаг N: ...>' blocks if model inserted steps into strategy.""" if not text: return text m = re.search(r"(?:^|\n)\s*Шаг\s*\d+\s*[:\-—]", text, flags=re.IGNORECASE) if m: return text[:m.start()].rstrip() return text def _extract_day_html(steps: Dict[str, Any], day: int) -> str: if not isinstance(steps, dict): return "" key = str(day) val = steps.get(key) if isinstance(val, dict): return val.get("html", "") or "" if isinstance(val, str): return val return "" # ───────── survey-aware constraints ───────── def _uses_social_networks(survey: Dict[str, Any]) -> bool: """ True — socials allowed; False — user doesn't use socials (ban). Supports various wordings. """ social_skill = survey.get("social_skill") # Handle enum values if social_skill == "dont use": return False # Handle text values (legacy) raw = str(social_skill or "").strip().lower() if not raw: return True deny = [ "dont use", "don't use", "do not use", "no social", "no_social", "нет", "не польз", "не корист", "не користуюсь", "не користуєшся", "без соц", "never", "none" ] return not any(m in raw for m in deny) def _is_sports_technique(survey: Dict[str, Any]) -> bool: tech = str(survey.get("massage_technique") or "").strip().lower() return "sport" in tech or "спорт" in tech def _social_constraints_block(survey: Dict[str, Any]) -> str: social_skill = survey.get("social_skill") # Если пользователь не пользуется соцсетями if not _uses_social_networks(survey): return ( "\nОГРАНИЧЕНИЕ ВАЖНО: Пользователь НЕ пользуется соцсетями. " "ЗАПРЕЩЕНО предлагать любые действия в соцсетях (Instagram, Facebook, TikTok, VK, Telegram-каналы, YouTube, Threads и т.п.). " "Разрешены только офлайн/несоцсетевые каналы: визитки/листівки, объявления/каталоги, сарафанное радио, партнёрства, " "стенды в локациях, Google Business Profile, сайт/лендинг, коллаборации с бизнесами, мероприятия, рекомендации, " "Avito/Юла, районные объявления, доски объявлений." ) # Если пользователь пользуется соцсетями - даем рекомендации по уровню навыков if social_skill in ["active", "user"]: # Опытные пользователи соцсетей return ( "\nРЕКОМЕНДАЦИИ ПО КАНАЛАМ: Пользователь активно пользуется соцсетями. " "ПРИОРИТЕТНЫЕ каналы: VK (местные группы), Telegram (каналы/чаты), локальные чаты (чат дома, чат подъезда, чат района, чат класса, чат школы, чат группы, чат потока, чат родителей, чат отдела, чат смены, чат бригады, чат филиала). " "СТРОГО ЗАПРЕЩЕНО упоминать Instagram и Facebook. " "Дополнительно: сарафанное радио, партнёрства, Google Business Profile." ) elif social_skill in ["watcher", "registered"]: # Новички в соцсетях return ( "\nРЕКОМЕНДАЦИИ ПО КАНАЛАМ: Пользователь новичок в соцсетях. " "ПРИОРИТЕТНЫЕ каналы: VK (простые группы), Telegram (личные сообщения), локальные чаты (чат дома, чат подъезда, чат района, чат класса, чат школы, чат группы, чат потока, чат родителей, чат отдела, чат смены, чат бригады, чат филиала). " "СТРОГО ЗАПРЕЩЕНО упоминать Instagram и Facebook. " "Дополнительно: офлайн каналы (визитки, объявления), сарафанное радио, Avito/Юла." ) # По умолчанию - базовые рекомендации return ( "\nРЕКОМЕНДАЦИИ ПО КАНАЛАМ: " "ПРИОРИТЕТНЫЕ каналы: VK (местные группы), Telegram (каналы/чаты), локальные чаты (чат дома, чат подъезда, чат района, чат класса, чат школы, чат группы, чат потока, чат родителей, чат отдела, чат смены, чат бригады, чат филиала). " "СТРОГО ЗАПРЕЩЕНО упоминать Instagram, Facebook, TikTok, YouTube, Threads. " "Дополнительно: сарафанное радио, партнёрства, офлайн каналы." ) # ───────── prompts (СТРАТЕГІЯ окремо, КРОКИ окремо) ───────── def _render_strategy_user_prompt_from_survey(survey: Dict[str, Any], bot_settings: Dict[str, Any] = None) -> str: payload = { "gender": survey.get("gender") or "", "age": survey.get("age") or "", "city": survey.get("city") or "", "practice_place": survey.get("practice_place") or "", "clients_experience": survey.get("clients_experience") or "", "massage_technique": survey.get("massage_technique") or "", "social_skill": survey.get("social_skill") or "", "communication_ease": survey.get("communication_ease") or "", "extra_notes": "", } # Add bot gender information to payload if bot_settings: payload["bot_voice_gender"] = bot_settings.get("bot_voice_gender") or "" payload["bot_voice_name"] = bot_settings.get("bot_voice_name") or "" else: payload["bot_voice_gender"] = "" payload["bot_voice_name"] = "" base = _safe_fill(strategy_prompt, payload) hard_rules = [ "СТРОГО СЛЕДУЙ ДАННЫМ АНКЕТЫ. Никаких предположений вне данных.", "Не описывай ежедневные задания — только стратегический каркас.", "Формат Телеграм: только и , переносы строк через \\n.", ] if _is_sports_technique(survey): hard_rules.append("Если техника массажа спортивная — целевая аудитория ТОЛЬКО спортсмены и активные люди. Укажи виды спорта и точки контакта (клубы, залы, секции, турниры).") constraints = _social_constraints_block(survey) contract = ( "\n\nПРАВИЛА:\n- " + "\n- ".join(hard_rules) + "\n\nВерни СТРОГО ОДИН JSON-объект без Markdown и без комментариев:\n" "{\n" ' "strategy": "полный текст стратегии (только , , переносы \\n). Без ежедневных шагов."\n' "}\n" ) return base + constraints + contract def _render_steps_user_prompt(survey: Dict[str, Any], strategy_text: str) -> str: base = ( "Ты бизнес-коуч. Отвечай строго на русском. Составь ЕЖЕДНЕВНЫЕ задания (15–20 минут) для массажиста.\n" "Формат Телеграма: только и , переносы '\\n', без
//Markdown.\n" "Структура шага:\n" "Шаг {номер дня}: Название\\n" "2–3 предложения сути (конкретно, без общих слов).\\n" "Пример: одно предложение мини-кейса\\n" "Подшаги:\n" "• пункт 1\\n• пункт 2\\n• пункт 3\n" "Номер шага ОБЯЗАТЕЛЕН и равен номеру дня (ключу в steps).\\n" "Запрещены бессмысленные шаги типа «Определение целевой аудитории», «Создай соцсеть», «Подумай». " "Каждый шаг — практическое действие с измеримым результатом (контакт, договорённость, запись, отзыв и т.п.).\n" ) ctx = { "survey": { "gender": survey.get("gender"), "city": survey.get("city"), "practice_place": survey.get("place"), "clients_experience": survey.get("clients"), "massage_technique": survey.get("technique"), "social_skill": survey.get("social"), "communication_ease": survey.get("comm"), }, "strategy": strategy_text[:6000], } hard_rules = [ "СТРОГО СЛЕДУЙ анкете и стратегии. Никаких противоречий.", "Запрещены шаги «Определи ЦА», «Сформулируй УТП» и подобные — это уже сделано в стратегии.", "Каждый шаг даёт осязаемый результат: контакт, договорённость, запись, визитка, листовка, звонок, встреча, отзыв, повторная запись.", "Шаги адаптированы под 15–20 минут. Если действие больше — дай микро-часть.", "Только и , переносы строк '\\n'.", ] if _is_sports_technique(survey): hard_rules.append("Фокус на спортсменов/активных: клубы, тренеры, турниры, фитнес-залы, секции. Никаких «офисных» ЦА.") if not _uses_social_networks(survey): hard_rules.append("Никаких соцсетей. Только офлайн/несоцсетевые каналы (объявления, партнёрства, листовки, клубы, тренеры, Google Business Profile, сайт/лендинг).") contract = ( "\nДАНО:\n" + str(ctx) + "\nПРАВИЛА:\n- " + "\n- ".join(hard_rules) + "\nВерни СТРОГО ОДИН JSON-объект:\n" "{\n" ' "steps": {\n' ' "1": "Шаг 1: ...\\nОписание...\\nПример: ...\\nПодшаги:\\n• ...",\n' ' "2": "Шаг 2: ...\\n...",\n' ' "...": "... до 30"\n' " }\n" "}\n" ) return base + contract def _render_revision_prompts(survey: Dict[str, Any], prev_strategy: str, feedback: str) -> Tuple[str, str]: constraints = _social_constraints_block(survey) strat_user = ( _render_strategy_user_prompt_from_survey(survey) + "\nДополнение от пользователя (учти строго при правке стратегии):\n" + feedback[:2000] + "\nЕсли часть текущей стратегии уместна — сохрани, но скорректируй под правки.\n" + constraints ) steps_user = ( _render_steps_user_prompt(survey, prev_strategy) + "\nДополнение от пользователя (учти строго при правке шагов):\n" + feedback[:2000] + "\nСогласуй стиль и содержание с обновленной стратегией.\n" + constraints ) return strat_user, steps_user # ───────── steps sanitization ───────── BANNED_STEP_PATTERNS = [ r"определ(и|ение)\s+целев(ой|ої)\s+аудитории", r"определи\s+ца", r"сформулируй\s+утп", r"создай\s+(аккаунт|страницу)\s+в\s+соцсет", r"проанализируй\s+конкурентов(?!.*конкретн)", ] def _default_step1_for_sport_no_social(city: str) -> str: return ( "Шаг 1: Быстрые контакты с тренерами\\n" "Найди и запиши 5 телефонов/чатов тренеров из 2 ближайших спортзалов. Свяжись с двумя и предложи пробный массаж для их спортсмена.\\n" "Пример: «Здравствуйте, я спортивный массажист рядом с {city}. Дам 1 пробный сеанс вашему атлету после тренировки, чтобы показать восстановление»\\n" "Подшаги:\\n" "• Открой сайты/Google-карты двух залов и выпиши контакты тренеров (5+)\\n" "• Напиши 2 коротких сообщения/позвони 2 тренерам\\n" "• Зафиксируй 1 договорённость (встреча/пробник)" ).replace("{city}", city or "") def _sanitize_steps(raw_steps: Dict[str, Any], survey: Dict[str, Any]) -> Dict[str, Any]: import re as _re steps = dict(raw_steps or {}) def _bad(text: str) -> bool: t = (text or "").lower() return any(_re.search(p, t) for p in BANNED_STEP_PATTERNS) first = steps.get("1", "") if isinstance(first, dict): first_text = first.get("html", "") or "" else: first_text = str(first or "") if _bad(first_text): if _is_sports_technique(survey) and not _uses_social_networks(survey): steps["1"] = _default_step1_for_sport_no_social(survey.get("city") or "") else: steps["1"] = ( "Шаг 1: Быстрая точка контакта\\n" "Сделай 5 коротких касаний с потенциальными партнёрами/клиентами из твоего канала привлечения.\\n" "Пример: «Здравствуйте, я массажист рядом. Готов провести 1 пробный сеанс для вашего клиента/сотрудника»\\n" "Подшаги:\\n" "• Выпиши 5 контактов\\n" "• Сделай 2 звонка/2 сообщения\\n" "• Зафиксируй 1 договорённость" ) return steps def _normalize_partial_keys(steps: Dict[str, Any], start: int, end: int) -> Dict[str, Any]: if not isinstance(steps, dict): return {} target_len = end - start + 1 numeric_keys = [] for k in steps.keys(): try: numeric_keys.append(int(str(k).strip())) except Exception: pass if len(numeric_keys) == target_len and min(numeric_keys, default=1) == 1 and max(numeric_keys, default=target_len) == target_len: ordered_items = [v for _, v in sorted(((int(str(k)), v) for k, v in steps.items()), key=lambda x: x[0])] remapped: Dict[str, Any] = {} cur = start for v in ordered_items: remapped[str(cur)] = v cur += 1 return remapped expected_keys = {str(i) for i in range(start, end + 1)} if set(map(str, steps.keys())).issuperset(expected_keys) or set(map(str, steps.keys())) == expected_keys: return {str(k): steps[str(k)] if str(k) in steps else steps[k] for k in range(start, end + 1) if (str(k) in steps) or (k in steps)} items = list(steps.values()) remapped: Dict[str, Any] = {} for idx in range(min(target_len, len(items))): remapped[str(start + idx)] = items[idx] return remapped # ───────── LLM builders ───────── async def _ask_json_strict(system: str, user: str, *, max_tokens: int, temperature: float, timeout_s: int): try: return await ask_json(system, user, temperature=temperature, max_tokens=max_tokens, timeout_s=timeout_s) except Exception: # First fallback: strict JSON instructions user2 = user + ( "\n\nОТВЕЧАЙ СТРОГО ОДНИМ JSON-ОБЪЕКТОМ без пояснений, без Markdown, без ```." " Никакого текста до или после. Формально валидный JSON." ) try: return await ask_json(system, user2, temperature=0.0, max_tokens=max_tokens, timeout_s=timeout_s) except Exception: # Second fallback: even more strict user3 = user + ( "\n\nВерни ТОЛЬКО JSON объект. Никакого текста. Никаких комментариев. " "Только валидный JSON. Пример: {\"strategy\": \"текст\"}" ) system2 = system + " ВАЖНО: Верни ТОЛЬКО JSON. Никакого другого текста." try: return await ask_json(system2, user3, temperature=0.0, max_tokens=max_tokens, timeout_s=timeout_s) except Exception: # Final fallback: return default response logger.error("All JSON parsing attempts failed, returning default response") return {"strategy": "Ошибка генерации стратегии. Попробуйте еще раз."} async def _build_strategy_text(survey: Dict[str, Any]) -> str: system = ( "Ты создаешь стратегии для массажистов. " "Отвечай ТОЛЬКО в формате JSON без дополнительного текста. " "Пример: {\"strategy\": \"🎯 Целевая аудитория: Офисные сотрудники 25-45 лет с болями в спине и шее\\n😰 Боль аудитории: Постоянные боли в спине и шее от сидячей работы\\n📱 Источник трафика: VK (местные группы района) + Telegram (каналы района) + локальные чаты\\n🎁 Лид-магнит: Бесплатная консультация \\\"5 простых упражнений от боли в шее за 5 минут\\\"\\n💰 Продукт: Массаж спины и шеи на дому у массажиста (60 минут)\\n⭐ УТП: \\\"Массаж спины и шеи в уютной домашней обстановке без очередей и спешки\\\"\"} " "НЕ используй Instagram и Facebook. " "Используй только VK, Telegram, локальные чаты." ) user = _render_strategy_user_prompt_from_survey(survey) resp = await _ask_json_strict(system, user, temperature=0.2, max_tokens=1800, timeout_s=45) return (resp.get("strategy") or "").strip() async def _build_steps_partial_only(survey: Dict[str, Any], strategy_text: str, start: int, end: int) -> Dict[str, Any]: system = ( f"Ты создаёшь подробный план ежедневных заданий для массажиста на дни с {start} по {end}.\n" "Отвечай СТРОГО в формате JSON без какого-либо текста вне JSON, без комментариев, без пояснений.\n" "Структура ответа ДОЛЖНА быть строго такой:\n" "{\n" " \"steps\": {\n" + ''.join([f' \"{i}\": \"...описание задания для Дня {i}...\",\n' for i in range(start, end + 1)]) + " }\n" "}\n" "\n" "ВНИМАТЕЛЬНО: ключи внутри \"steps\" должны начинаться СТРОГО с \"" + str(start) + "\" и идти ПО ПОРЯДКУ до \"" + str(end) + "\". НЕЛЬЗЯ использовать нумерацию с 1 для этой части.\n" "Пример ключей: \"" + str(start) + "\": \"...\", \"" + str(start + 1) + "\": \"...\" (и так далее до \"" + str(end) + "\").\n" "\n" "ТРЕБОВАНИЯ К КАЖДОМУ ДНЮ:\n" "1. Каждый день — ОТДЕЛЬНОЕ УНИКАЛЬНОЕ задание, практическое и выполнимое за один день. КРИТИЧЕСКИ ВАЖНО: каждое задание должно быть РАЗНЫМ от всех остальных. НЕ ДУБЛИРУЙ задания, даже если они похожи по теме.\n" "2. Минимум 3 полноценные связанные фразы (не пункты-огрызки типа «сделай это»), то есть небольшой связный мини-план.\n" "3. В каждом дне должен быть конкретный пример сообщения/текста/фразы, которую массажист может сказать или написать потенциальному клиенту.\n" "4. Используй только разрешённые каналы привлечения клиентов: VK (личные сообщения ...), Telegram (...), устные личные контакты, сарафанное радио, визитки, офлайн-объявления, партнёрства с локальными бизнесами, Avito/Юла.\n" "5. СТРОГО ЗАПРЕЩЕНО упоминать Instagram и Facebook в любом виде.\n" "6. Никаких ссылок, хэштегов, HTML-тегов (, и т.п.) и никакого форматирования — чистый обычный текст.\n" "7. Не используй списки с «-», «•», нумерацию внутри дня и т.п. Пиши цельным текстом абзацем.\n" "8. Каждый день должен быть самодостаточным. Не ссылайся на «вчерашнее задание», «как мы делали ранее», «продолжи предыдущий шаг». Пиши так, будто это первая задача, которую он видит.\n" "9. ВАЖНО: Разнообразь задания! Используй разные каналы, разные подходы, разные форматы. Не повторяй одно и то же задание для разных дней.\n" "\n" "ПРО ПОЛЬЗОВАТЕЛЯ:\n" "1. Ты можешь использовать только те факты о массажисте, которые явно указаны во входных данных (анкета и стратегия). Если чего-то нет во входных данных, НЕ ВЫДУМЫВАЙ.\n" "2. Если каких-то данных (например, город, специализация, цена) нет во входных данных, используй нейтральные формулировки без конкретики, типа «в вашем районе», «ваши услуги массажа», «ваш контакт».\n" "\n" "ЕЩЁ РАЗ О ФОРМАТЕ:\n" "- Ключ верхнего уровня: только \"steps\".\n" f'- Внутри \"steps\" должны быть РОВНО ключи-строки от \"{start}\" до \"{end}\" включительно.\n' "- Значение каждого ключа — это строка с описанием задания.\n" "- Никаких других ключей добавлять нельзя.\n" "- Ответ должен быть корректным JSON.\n" ) user = _render_steps_user_prompt(survey, strategy_text) resp = await _ask_json_strict( system, user, temperature=0.2, max_tokens=4096, timeout_s=60 ) steps = resp.get("steps") or {} return steps if isinstance(steps, dict) else {} def _detect_duplicate_steps(steps: Dict[str, Any]) -> Dict[int, int]: """ Detect duplicate step texts and return mapping of duplicate day -> original day. Returns: Dict mapping duplicate day number to original day number """ duplicates = {} step_texts = {} for day_str, step_text in steps.items(): try: day = int(day_str) except (ValueError, TypeError): continue # Normalize step text for comparison (remove extra whitespace, lowercase) if isinstance(step_text, dict): normalized = (step_text.get("html", "") or step_text.get("text", "")).strip().lower() else: normalized = str(step_text).strip().lower() # Remove very short texts from comparison (likely placeholders) if len(normalized) < 20: continue # Check for similar texts (exact match or very similar) for existing_day, existing_text in step_texts.items(): # Exact match if normalized == existing_text: duplicates[day] = existing_day break # Very similar (80%+ word overlap) if len(normalized) > 50 and len(existing_text) > 50: words1 = set(normalized.split()) words2 = set(existing_text.split()) if len(words1) > 0 and len(words2) > 0: similarity = len(words1 & words2) / max(len(words1), len(words2)) if similarity > 0.8: duplicates[day] = existing_day break if day not in duplicates: step_texts[day] = normalized return duplicates async def _build_steps_only(survey: Dict[str, Any], strategy_text: str) -> Dict[str, Any]: steps_1_15 = await _build_steps_partial_only(survey, strategy_text, 1, 15) steps_16_30 = await _build_steps_partial_only(survey, strategy_text, 16, 30) steps = {**steps_1_15, **steps_16_30} # Проверка на дубликаты duplicates = _detect_duplicate_steps(steps) if duplicates: logger.warning(f"Found duplicate steps: {duplicates}") # Перегенерируем дубликаты for dup_day, orig_day in duplicates.items(): # Генерируем новое задание для дубликата new_partial = await _build_steps_partial_only(survey, strategy_text, dup_day, dup_day) if str(dup_day) in new_partial: steps[str(dup_day)] = new_partial[str(dup_day)] # Повторная проверка после регенерации duplicates_after = _detect_duplicate_steps(steps) if duplicates_after: logger.warning(f"Still have duplicates after regeneration: {duplicates_after}") return steps async def _build_strategy_and_steps(survey: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]: strategy_text = await _build_strategy_text(survey) steps = await _build_steps_only(survey, strategy_text) return strategy_text, steps async def _build_steps_partial_with_feedback( survey: Dict[str, Any], steps_user_base: str, start: int, end: int, ) -> Dict[str, Any]: system = ( f"Ты создаёшь подробный план ежедневных заданий для массажиста только для дней с {start} по {end}.\n" "Отвечай СТРОГО в формате JSON без какого-либо текста вне JSON, без комментариев, без пояснений.\n" "Структура ответа ДОЛЖНА быть строго такой:\n" "{\n" " \"steps\": {\n" + ''.join([f' \"{i}\": \"...описание задания для Дня {i}...\",\n' for i in range(start, end + 1)]) + " }\n" "}\n" "\n" "ВНИМАТЕЛЬНО: ключи внутри \"steps\" должны начинаться СТРОГО с \"" + str(start) + "\" и идти ПО ПОРЯДКУ до \"" + str(end) + "\". НЕЛЬЗЯ использовать нумерацию с 1 для этой части.\n" "Пример ключей: \"" + str(start) + "\": \"...\", \"" + str(start + 1) + "\": \"...\" (и так далее до \"" + str(end) + "\").\n" "\n" "ТРЕБОВАНИЯ К КАЖДОМУ ДНЮ:\n" "1. Каждый день — ОТДЕЛЬНОЕ УНИКАЛЬНОЕ задание, практическое и выполнимое за один день. КРИТИЧЕСКИ ВАЖНО: каждое задание должно быть РАЗНЫМ от всех остальных. НЕ ДУБЛИРУЙ задания, даже если они похожи по теме.\n" "2. Минимум 3 полноценные связанные фразы (не пункты-огрызки типа «сделай это»), то есть небольшой связный мини-план.\n" "3. В каждом дне должен быть конкретный пример сообщения/текста/фразы, которую массажист может сказать или написать потенциальному клиенту.\n" "4. Используй только разрешённые каналы привлечения клиентов: VK (личные сообщения ...), Telegram (...), устные личные контакты, сарафанное радио, визитки, офлайн-объявления, партнёрства с локальными бизнесами, Avito/Юла.\n" "5. СТРОГО ЗАПРЕЩЕНО упоминать Instagram и Facebook в любом виде.\n" "6. Никаких ссылок, хэштегов, HTML-тегов (, и т.п.) и никакого форматирования — чистый обычный текст.\n" "7. Не используй списки с «-», «•», нумерацию внутри дня и т.п. Пиши цельным текстом абзацем.\n" "8. Каждый день должен быть самодостаточным. Не ссылайся на «вчерашнее задание», «как мы делали ранее», «продолжи предыдущий шаг». Пиши так, будто это первая задача, которую он видит.\n" "9. ВАЖНО: Разнообразь задания! Используй разные каналы, разные подходы, разные форматы. Не повторяй одно и то же задание для разных дней.\n" ) user = ( steps_user_base + f"\nГЕНЕРИРУЙ ТОЛЬКО ДНИ {start}–{end}. Ключи строго \"{start}\"..\"{end}\"." ) resp = await _ask_json_strict( system, user, temperature=0.2, max_tokens=4096, timeout_s=60 ) steps = resp.get("steps") or {} return steps if isinstance(steps, dict) else {} async def _rebuild_strategy_and_steps_with_feedback( survey: Dict[str, Any], prev_strategy: str, feedback: str ) -> Tuple[str, Dict[str, Any]]: strat_system = ( "Ты создаешь стратегии для массажистов. " "Отвечай ТОЛЬКО в формате JSON без дополнительного текста. " "Пример: {'strategy': '...'} " "НЕ используй Instagram и Facebook. " "Используй только VK, Telegram, локальные чаты." ) steps_system = ( "Ты создаешь задания для массажистов. " "Отвечай ТОЛЬКО в формате JSON без дополнительного текста. " "Пример: {\"steps\": {\"1\": \"...\"}} " "НЕ используй Instagram и Facebook. " "Используй только VK, Telegram, локальные чаты." ) strat_user, steps_user = _render_revision_prompts(survey, prev_strategy, feedback) strat_resp = await _ask_json_strict(strat_system, strat_user, temperature=0.2, max_tokens=1800, timeout_s=60) new_strategy = (strat_resp.get("strategy") or "").strip() # Шаги с учетом feedback двумя частями steps_1_15 = await _build_steps_partial_with_feedback(survey, steps_user, 1, 15) steps_16_30 = await _build_steps_partial_with_feedback(survey, steps_user, 16, 30) new_steps = {**steps_1_15, **steps_16_30} if not isinstance(new_steps, dict): new_steps = {} # Проверка на дубликаты duplicates = _detect_duplicate_steps(new_steps) if duplicates: logger.warning(f"Found duplicate steps in rebuild: {duplicates}") # Перегенерируем дубликаты for dup_day, orig_day in duplicates.items(): new_partial = await _build_steps_partial_with_feedback(survey, steps_user, dup_day, dup_day) if str(dup_day) in new_partial: new_steps[str(dup_day)] = new_partial[str(dup_day)] new_steps = _sanitize_steps(new_steps, survey) return new_strategy, new_steps # ───────── /start flow ───────── @router.message(CommandStart()) async def start(message: Message, state: FSMContext): tg_id = str(message.from_user.id) async with session_scope() as session: registered = await is_user_registered(session, tg_id) if registered: await message.answer( "Ты уже проходил анкету. Хочешь удалить старые данные и начать заново?", reply_markup=kb_confirm_wipe() ) return await state.clear() await state.set_state(SurveyFSM.name) await message.answer("Как тебя зовут ? Впиши полное имя.") # ───────── name → gender ───────── @router.message(SurveyFSM.name) async def q_gender(message: Message, state: FSMContext): name = (message.text or "").strip() if len(name) < 2: await message.answer("Похоже, что ты ошибся, впиши полное имя") return await state.update_data(name=name) await state.set_state(SurveyFSM.gender) await message.answer("Твой пол:", reply_markup=kb_gender()) # ───────── gender → city ───────── @router.callback_query(SurveyFSM.gender, F.data.startswith("gender:")) async def gender_chosen(cb: CallbackQuery, state: FSMContext): try: value = cb.data.split(":", 1)[1] gender = Gender(value) except Exception: await cb.answer("Неверное значение", show_alert=True) return await state.update_data(gender=gender.value) await state.set_state(SurveyFSM.city) await cb.message.edit_text("В каком городе ты живёшь? Введи в формате Город/область") await cb.answer() # ───────── city → place ───────── @router.message(SurveyFSM.city) async def q_place(message: Message, state: FSMContext): city = (message.text or "").strip() if not city: await message.answer("Введи корректный город.") return await state.update_data(city=city) await state.set_state(SurveyFSM.place) await message.answer("Где вы сейчас практикуете массаж?", reply_markup=kb_place()) # ───────── place → clients ───────── @router.callback_query(SurveyFSM.place, F.data.startswith("place:")) async def place_chosen(cb: CallbackQuery, state: FSMContext): try: value = cb.data.split(":", 1)[1] place = PracticePlace(value) except Exception: await cb.answer("Неверное значение", show_alert=True) return await state.update_data(place=place.value) await state.set_state(SurveyFSM.clients) await cb.message.edit_text("У тебя уже были клиенты?") await cb.message.edit_reply_markup(reply_markup=kb_clients()) await cb.answer() # ───────── clients → technique ───────── @router.callback_query(SurveyFSM.clients, F.data.startswith("clients:")) async def clients_chosen(cb: CallbackQuery, state: FSMContext): try: value = cb.data.split(":", 1)[1] clients = ClientsExperience(value) except Exception: await cb.answer("Неверное значение", show_alert=True) return await state.update_data(clients=clients.value) await state.set_state(SurveyFSM.technique) await cb.message.edit_text("Какую технику массажа ты практикуешь чаще всего?") await cb.message.edit_reply_markup(reply_markup=kb_technique()) await cb.answer() # ───────── technique → social ───────── @router.callback_query(SurveyFSM.technique, F.data.startswith("tech:")) async def tech_chosen(cb: CallbackQuery, state: FSMContext): try: value = cb.data.split(":", 1)[1] tech = MassageTechnique(value) except Exception: await cb.answer("Неверное значение", show_alert=True) return await state.update_data(technique=tech.value) await state.set_state(SurveyFSM.social) await cb.message.edit_text("Насколько уверенно ты пользуешься соцсетями?") await cb.message.edit_reply_markup(reply_markup=kb_social()) await cb.answer() # ───────── social → comm ───────── @router.callback_query(SurveyFSM.social, F.data.startswith("social:")) async def social_chosen(cb: CallbackQuery, state: FSMContext): try: value = cb.data.split(":", 1)[1] social = SocialSkill(value) except Exception: await cb.answer("Неверное значение", show_alert=True) return await state.update_data(social=social.value) await state.set_state(SurveyFSM.comm) await cb.message.edit_text("Насколько легко ты взаимодействуешь с незнакомыми людьми?") await cb.message.edit_reply_markup(reply_markup=kb_comm()) await cb.answer() # ───────── comm → persist user → build(strategy+steps) → REVIEW (NO TASK YET) ───────── @router.callback_query(SurveyFSM.comm, F.data.startswith("comm:")) async def comm_chosen(cb: CallbackQuery, state: FSMContext): try: value = cb.data.split(":", 1)[1] comm = CommunicationEase(value) except Exception: await cb.answer("Неверное значение", show_alert=True) return await state.update_data(comm=comm.value) data = await state.get_data() tg_id = str(cb.from_user.id) await cb.answer() await cb.message.edit_text(f"{data['name']}, анкета сохранена. Формирую персональную стратегию...") stop_typing = asyncio.Event() typing_task = asyncio.create_task(_typing_loop(cb.message.bot, cb.message.chat.id, stop_typing)) try: async with session_scope() as session: await register_user( session, tg_id=tg_id, user_name=data["name"], user_gender=data["gender"], city=data["city"], where_practicing=data["place"], have_clients=data["clients"], massage_technique=data["technique"], social_skill=data["social"], communication_ease=data["comm"], raw_json=data, ) strategy_text, steps = await _build_strategy_and_steps(data) await state.update_data( draft_strategy_text=strategy_text, draft_steps=steps, survey_payload=data, user_tg_id=tg_id, edit_mode=None, ) finally: stop_typing.set() with contextlib.suppress(Exception): await typing_task clean_strategy = _strategy_without_day_blocks(strategy_text) await cb.message.answer( clean_strategy or "Стратегия пустая.", reply_markup=kb_review() ) # ───────── REVIEW: accept / edit ───────── @router.callback_query(F.data == "review:accept") async def review_accept(cb: CallbackQuery, state: FSMContext): data = await state.get_data() strategy_text: Optional[str] = data.get("draft_strategy_text") steps: Dict[str, Any] = data.get("draft_steps") or {} if not strategy_text: await cb.answer("Стратегия ещё не сгенерирована. Нажми /start и пройди анкету заново.", show_alert=True) return await cb.answer("Сохраняю…") async with session_scope() as session: user = await get_user_by_tg(session, str(cb.from_user.id)) if not user: await cb.message.answer("Анкета не найдена. Нажми /start и пройди короткую анкету.") return strat = await create_strategy( session, user_id=user.id, strategy_text=strategy_text, steps=steps, make_active=True, raise_if_active_exists=False, ) await ensure_today_issued(session, strat.id) with contextlib.suppress(Exception): await cb.message.edit_reply_markup(reply_markup=None) await cb.message.answer("Готово. Стратегия сохранена и активирована.") # Immediately issue first task for the new strategy first_day_html = _extract_day_html(steps, 1) await cb.message.answer(first_day_html or "Сегодняшний шаг недоступен.", reply_markup=kb_daily_step()) await state.update_data(draft_strategy_text=None, draft_steps=None) # Review edit handler moved to strategy_management.py to avoid conflicts # ───────── /strategy command: show & toggle edit mode ───────── def _survey_from_user(user) -> Dict[str, Any]: """Extract survey fields from User ORM to a simple dict (for LLM context).""" if not user: return {} # Get profile data if exists profile = getattr(user, "profile", None) return { "gender": getattr(user, "user_gender", None), "age": getattr(user, "age", None), "city": getattr(user, "city", None), "practice_place": getattr(profile, "where_practicing", None) if profile else None, "clients_experience": getattr(profile, "have_clients", None) if profile else None, "massage_technique": getattr(profile, "massage_technique", None) if profile else None, "social_skill": getattr(profile, "social_skill", None) if profile else None, "communication_ease": getattr(profile, "communication_ease", None) if profile else None, } # Command /strategy moved to strategy_management.py to avoid conflicts # Callback handlers moved to strategy_management.py to avoid conflicts # Review feedback handler moved to strategy_management.py to avoid conflicts # ───────── Voice/audio/video_note feedback (TYPING only) ───────── async def _download_file_bytes(message: Message, file_id: str) -> bytes: buf = io.BytesIO() try: f = await message.bot.get_file(file_id) if getattr(f, "file_path", None): await message.bot.download_file(f.file_path, destination=buf) # aiogram v2 else: await message.bot.download(file_id, destination=buf) # aiogram v3 except Exception as e1: logger.warning("[survey] primary download path failed: %s, trying fallback", e1) try: await message.bot.download(file_id, destination=buf) except Exception as e2: logger.exception("[survey] download failed: %s", e2) raise return buf.getvalue() async def _transcribe_bytes_oa(audio_bytes: bytes, filename: str) -> str: api_key = os.getenv("OPENAI_API_KEY") if not api_key: logger.error("[survey] OPENAI_API_KEY is not set") raise RuntimeError("OPENAI_API_KEY is not set") mime = mimetypes.guess_type(filename)[0] or "audio/mpeg" files = {"file": (filename, audio_bytes, mime)} data = {"model": "whisper-1"} logger.info("[survey] transcribe start: filename=%s mime=%s size=%d", filename, mime, len(audio_bytes)) async with httpx.AsyncClient(timeout=httpx.Timeout(120.0)) as client: for attempt in range(3): try: r = await client.post( "https://api.openai.com/v1/audio/transcriptions", headers={"Authorization": f"Bearer {api_key}"}, files=files, data=data, ) r.raise_for_status() j = r.json() text = (j.get("text") or "").strip() logger.info("[survey] transcribe done: len=%d preview=%r", len(text), text[:120]) return text except httpx.HTTPStatusError as e: status = e.response.status_code if status in (429, 500, 502, 503, 504) and attempt < 2: await asyncio.sleep(1.2 * (attempt + 1)) continue logger.exception("[survey] transcribe failed: status=%s body=%r", status, e.response.text[:200]) raise # Voice feedback handler moved to strategy_management.py to avoid conflicts # Voice feedback handler moved to strategy_management.py to avoid conflicts # Voice feedback handler moved to strategy_management.py to avoid conflicts # ───────── Overwrite strategy helper ───────── async def overwrite_strategy( session: AsyncSession, *, strategy_id: int, new_strategy_text: str, new_steps: Dict[str, Any], reset_to_day: int = 1, drop_day_states: bool = True, ) -> Strategy: strat = await _get_strategy(session, strategy_id) if not strat: raise ValueError("стратегия не найдена") if drop_day_states: await session.execute( delete(StrategyDayState).where(StrategyDayState.strategy_id == strat.id) ) strat.strategy = new_strategy_text strat.steps = new_steps or {} strat.status = StrategyStatus.active strat.current_day = reset_to_day await session.flush() if _max_day(strat.steps) >= reset_to_day: await _ensure_day_state(session, strat.id, reset_to_day) return strat # Overwrite confirmation callbacks moved to strategy_management.py to avoid conflicts # ───────── Wipe flow ───────── @router.callback_query(F.data == "wipe:yes") async def confirm_wipe(cb: CallbackQuery, state: FSMContext): tg_id = str(cb.from_user.id) async with session_scope() as session: await delete_user_by_tg(session, tg_id) await state.clear() await cb.message.edit_text("🧹 Всё удалено. Начнём заново.\nКак тебя зовут? Впиши полное имя.") await state.set_state(SurveyFSM.name) await cb.answer() @router.callback_query(F.data == "wipe:no") async def cancel_wipe(cb: CallbackQuery, state: FSMContext): await cb.message.edit_text("Ок, старая анкета сохранена. Можешь продолжать работу со стратегией 👍") await cb.answer()