""" Strategy generation service using LLM. """ import logging from typing import Dict, Any, Tuple from src.llm.openai_client import ask_json from src.bot.utils.text_utils import safe_fill from src.bot.utils.survey_constraints import ( uses_social_networks, is_sports_technique, get_social_constraints_block ) from src.bot.utils.prompts import strategy_prompt logger = logging.getLogger(__name__) class StrategyService: """Service for generating business strategies using LLM.""" def __init__(self): self.logger = logger async def _ask_json_strict( self, system: str, user: str, *, max_tokens: int, temperature: float, timeout_s: int ) -> Dict[str, Any]: """ Ask LLM for JSON response with strict formatting. Args: system: System prompt user: User prompt max_tokens: Maximum tokens to generate temperature: Temperature for generation timeout_s: Timeout in seconds Returns: JSON response from LLM """ 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": "Ошибка генерации стратегии. Попробуйте еще раз."} def _render_strategy_user_prompt_from_survey(self, survey: Dict[str, Any], bot_settings: Dict[str, Any] = None) -> str: """ Render user prompt for strategy generation from survey data. Args: survey: Survey data dictionary bot_settings: Bot settings dictionary with bot_voice_gender, bot_voice_name Returns: Formatted user prompt """ 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 = get_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(self, survey: Dict[str, Any], strategy_text: str) -> str: """ Render user prompt for steps generation. Args: survey: Survey data dictionary strategy_text: Generated strategy text Returns: Formatted user prompt for steps """ 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( self, survey: Dict[str, Any], prev_strategy: str, feedback: str ) -> Tuple[str, str]: """ Render prompts for strategy revision with feedback. Args: survey: Survey data dictionary prev_strategy: Previous strategy text feedback: User feedback Returns: Tuple of (strategy_prompt, steps_prompt) """ constraints = get_social_constraints_block(survey) strat_user = ( self._render_strategy_user_prompt_from_survey(survey) + "\nДополнение от пользователя (учти строго при правке стратегии):\n" + feedback[:2000] + "\nЕсли часть текущей стратегии уместна — сохрани, но скорректируй под правки.\n" + constraints ) steps_user = ( self._render_steps_user_prompt(survey, prev_strategy) + "\nДополнение от пользователя (учти строго при правке шагов):\n" + feedback[:2000] + "\nСогласуй стиль и содержание с обновленной стратегией.\n" + constraints ) return strat_user, steps_user async def build_strategy_text(self, survey: Dict[str, Any], bot_settings: Dict[str, Any] = None) -> str: """ Build strategy text from survey data. Args: survey: Survey data dictionary bot_settings: Bot settings dictionary Returns: Generated strategy text """ system = ( "Ты создаешь стратегии для массажистов. " "Отвечай ТОЛЬКО в формате JSON без дополнительного текста. " "Пример: {\"strategy\": \"🎯 Целевая аудитория: Офисные сотрудники 25-45 лет с болями в спине и шее\\n😰 Боль аудитории: Постоянные боли в спине и шее от сидячей работы\\n📱 Источник трафика: VK (местные группы района) + Telegram (каналы района) + локальные чаты\\n🎁 Лид-магнит: Бесплатная консультация \\\"5 простых упражнений от боли в шее за 5 минут\\\"\\n💰 Продукт: Массаж спины и шеи на дому у массажиста (60 минут)\\n⭐ УТП: \\\"Массаж спины и шеи в уютной домашней обстановке без очередей и спешки\\\"\"} " "НЕ используй Instagram и Facebook. " "Используй только VK, Telegram, локальные чаты." ) user = self._render_strategy_user_prompt_from_survey(survey, bot_settings) resp = await self._ask_json_strict( system, user, temperature=0.2, max_tokens=1800, timeout_s=45 ) return (resp.get("strategy") or "").strip() def _normalize_partial_keys(self, steps: Dict[str, Any], start: int, end: int) -> Dict[str, Any]: if not isinstance(steps, dict): return {} target_len = end - start + 1 # Попытка понять, что модель вернула 1..N вместо start..end numeric_keys = [] for k in steps.keys(): try: numeric_keys.append(int(str(k).strip())) except Exception: pass # Если ключи идут с 1 и их ровно столько, сколько нужно — перенумеруем на нужный диапазон 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[k] if str(k) in steps else steps[str(k)] for k in range(start, end + 1) if (str(k) in steps)} # Фоллбек: попытаемся просто разложить первые target_len элементов по нужному диапазону 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 async def _build_steps_partial_only(self, 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 = self._render_steps_user_prompt(survey, strategy_text) resp = await self._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 _build_steps_partial_with_feedback( self, 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 self._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(self, 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 (90%+ similarity) if len(normalized) > 50 and len(existing_text) > 50: # Simple similarity check: common words ratio 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: # 80%+ word overlap duplicates[day] = existing_day break if day not in duplicates: step_texts[day] = normalized return duplicates async def build_steps_only(self, survey: Dict[str, Any], strategy_text: str) -> Dict[str, Any]: steps_1_15 = await self._build_steps_partial_only(survey, strategy_text, 1, 15) steps_16_30 = await self._build_steps_partial_only(survey, strategy_text, 16, 30) steps = {**steps_1_15, **steps_16_30} # Проверка на дубликаты duplicates = self._detect_duplicate_steps(steps) if duplicates: self.logger.warning(f"Found duplicate steps: {duplicates}") # Перегенерируем дубликаты for dup_day, orig_day in duplicates.items(): # Генерируем новое задание для дубликата if dup_day <= 15: # Регенерируем только этот день из первого батча new_partial = await self._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)] else: # Регенерируем только этот день из второго батча new_partial = await self._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 = self._detect_duplicate_steps(steps) if duplicates_after: self.logger.warning(f"Still have duplicates after regeneration: {duplicates_after}") return steps async def build_strategy_and_steps(self, survey: Dict[str, Any], bot_settings: Dict[str, Any] = None) -> Tuple[str, Dict[str, Any]]: strategy_text = await self.build_strategy_text(survey, bot_settings) steps = await self.build_steps_only(survey, strategy_text) return strategy_text, steps async def rebuild_strategy_and_steps_with_feedback( self, survey: Dict[str, Any], prev_strategy: str, feedback: str ) -> Tuple[str, Dict[str, Any]]: """ Rebuild strategy and steps with user feedback. Args: survey: Survey data dictionary prev_strategy: Previous strategy text feedback: User feedback Returns: Tuple of (new_strategy_text, new_steps_dict) """ strat_system = ( "Ты создаешь стратегии для массажистов. " "Отвечай ТОЛЬКО в формате JSON без дополнительного текста. " "Пример: {'strategy': '...'} " "НЕ используй Instagram и Facebook. " "Используй только VK, Telegram, локальные чаты." ) strat_user, steps_user = self._render_revision_prompts(survey, prev_strategy, feedback) strat_resp = await self._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 self._build_steps_partial_with_feedback(survey, steps_user, 1, 15) steps_16_30 = await self._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 = self._detect_duplicate_steps(new_steps) if duplicates: self.logger.warning(f"Found duplicate steps in rebuild: {duplicates}") # Перегенерируем дубликаты for dup_day, orig_day in duplicates.items(): new_partial = await self._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)] return new_strategy, new_steps