| import os
|
| import threading
|
| import hashlib
|
| import asyncio
|
| import io
|
| import aiohttp
|
| import random
|
| import uuid
|
| from flask import Flask, request
|
| from vkbottle import Keyboard, KeyboardButtonColor, Text, OpenLink, PhotoMessageUploader, API
|
| from vkbottle.bot import Bot, Message
|
| from supabase import create_client, Client
|
| from loguru import logger
|
| from dotenv import load_dotenv
|
| from PIL import Image
|
|
|
| load_dotenv()
|
|
|
|
|
| import prompts
|
| import generator
|
|
|
|
|
| TOKEN = os.environ.get("VK_TOKEN")
|
| SECOND_GROUP_TOKEN = os.environ.get("SECOND_GROUP_TOKEN")
|
| YOOMONEY_SECRET = os.environ.get("YOOMONEY_SECRET")
|
| SUPABASE_URL = os.environ.get("SUPABASE_URL")
|
| SUPABASE_KEY = os.environ.get("SUPABASE_KEY")
|
| YOOMONEY_WALLET = "4100119452661463"
|
|
|
| GROUP_ID = 226495679
|
| ADDITIONAL_GROUP_ID = 218575968
|
|
|
| SUPPORT_URL = "https://vk.me/olga_luchnaya"
|
| BUCKET_NAME = "photos"
|
|
|
|
|
| bot = Bot(token=TOKEN)
|
|
|
| second_bot_api = API(token=SECOND_GROUP_TOKEN)
|
|
|
| uploader = PhotoMessageUploader(bot.api)
|
| supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY)
|
| app = Flask(__name__)
|
| loop = None
|
|
|
| logger.add("bot_log.log", rotation="10 MB", level="INFO")
|
|
|
|
|
|
|
|
|
| async def upload_to_bucket(url, user_id, prefix="temp"):
|
| """Скачивает фото и временно сохраняет в Supabase Storage. Возвращает путь и публичную ссылку."""
|
| try:
|
| async with aiohttp.ClientSession() as session:
|
| async with session.get(url) as resp:
|
| if resp.status == 200:
|
| img_data = await resp.read()
|
| file_name = f"{user_id}/{prefix}_{uuid.uuid4()}.png"
|
| supabase.storage.from_(BUCKET_NAME).upload(
|
| path=file_name,
|
| file=img_data,
|
| file_options={"content-type": "image/png"}
|
| )
|
|
|
| res = supabase.storage.from_(BUCKET_NAME).get_public_url(file_name)
|
| return file_name, res
|
| except Exception as e:
|
| logger.error(f"Ошибка загрузки в бакет ({prefix}): {e}")
|
| return None, None
|
|
|
|
|
| async def download_file(url):
|
| """Скачивает файл по URL и возвращает байты для загрузки в ВК"""
|
| try:
|
| async with aiohttp.ClientSession() as session:
|
| async with session.get(url) as resp:
|
| if resp.status == 200:
|
| return await resp.read()
|
| except Exception as e:
|
| logger.error(f"Ошибка скачивания файла для ВК: {e}")
|
| return None
|
|
|
|
|
| async def upload_photo_to_vk(image_bytes, peer_id, file_name="image.png"):
|
| """Загружает фото в сообщения ВК вручную через upload server."""
|
|
|
|
|
| try:
|
| with Image.open(io.BytesIO(image_bytes)) as img:
|
| if img.mode != "RGB":
|
| img = img.convert("RGB")
|
|
|
| normalized_buffer = io.BytesIO()
|
| img.save(
|
| normalized_buffer,
|
| format="JPEG",
|
| quality=95,
|
| optimize=True
|
| )
|
| normalized_bytes = normalized_buffer.getvalue()
|
| except Exception as e:
|
| raise RuntimeError(f"Не удалось нормализовать изображение для ВК: {e}")
|
|
|
| logger.info(
|
| f"VK image normalized: original={len(image_bytes)} bytes, "
|
| f"jpeg={len(normalized_bytes)} bytes"
|
| )
|
|
|
| last_error = None
|
|
|
|
|
|
|
| for attempt in range(1, 5):
|
| try:
|
| upload_server = await bot.api.photos.get_messages_upload_server(peer_id=peer_id)
|
|
|
| form = aiohttp.FormData()
|
| form.add_field(
|
| "photo",
|
| normalized_bytes,
|
| filename="image.jpg",
|
| content_type="image/jpeg"
|
| )
|
|
|
| async with aiohttp.ClientSession() as session:
|
| async with session.post(upload_server.upload_url, data=form) as resp:
|
| upload_text = await resp.text()
|
|
|
| logger.info(
|
| f"VK upload server response (attempt {attempt}/4): {upload_text}"
|
| )
|
|
|
| if resp.status != 200:
|
| raise RuntimeError(
|
| f"VK upload server HTTP {resp.status}: {upload_text}"
|
| )
|
|
|
| try:
|
| upload_data = await resp.json(content_type=None)
|
| except Exception:
|
| raise RuntimeError(
|
| f"VK upload server returned invalid JSON: {upload_text}"
|
| )
|
|
|
| photo_value = upload_data.get("photo")
|
| server_value = upload_data.get("server")
|
| hash_value = upload_data.get("hash")
|
|
|
| if not photo_value:
|
| raise RuntimeError(
|
| f"VK upload server returned empty photo: {upload_data}"
|
| )
|
|
|
| saved_photos = await bot.api.photos.save_messages_photo(
|
| server=server_value,
|
| photo=photo_value,
|
| hash=hash_value
|
| )
|
|
|
| if not saved_photos:
|
| raise RuntimeError("VK photos.saveMessagesPhoto returned empty result")
|
|
|
| saved_photo = saved_photos[0]
|
|
|
| attachment = f"photo{saved_photo.owner_id}_{saved_photo.id}"
|
| if getattr(saved_photo, "access_key", None):
|
| attachment += f"_{saved_photo.access_key}"
|
|
|
| return attachment
|
|
|
| except Exception as e:
|
| last_error = e
|
| logger.error(f"VK upload attempt {attempt}/4 failed: {e}")
|
|
|
| if attempt < 4:
|
| await asyncio.sleep(2)
|
|
|
| raise RuntimeError(f"VK upload failed after 4 attempts: {last_error}")
|
|
|
| def get_user(user_id):
|
| res = supabase.table("users").select("*").eq("id", user_id).execute()
|
| if not res.data:
|
| new_user = {
|
| "id": user_id, "balance": 0, "free_used": False, "status": "main",
|
| "gender": None, "primary_photo": None, "last_style": None, "temp_promo": None
|
| }
|
| supabase.table("users").insert(new_user).execute()
|
| return new_user
|
| return res.data[0]
|
|
|
|
|
| async def is_subscribed(user_id):
|
| try:
|
| res = await bot.api.groups.is_member(group_id=GROUP_ID, user_id=user_id)
|
| return bool(res)
|
| except Exception as e:
|
| logger.error(f"Ошибка проверки подписки: {e}")
|
| return True
|
|
|
|
|
| async def is_subscribed_both(user_id):
|
| """Проверка подписки на оба сообщества для промокода"""
|
| try:
|
| res1 = await bot.api.groups.is_member(group_id=GROUP_ID, user_id=user_id)
|
| res2 = await second_bot_api.groups.is_member(group_id=ADDITIONAL_GROUP_ID, user_id=user_id)
|
| return bool(res1) and bool(res2)
|
| except Exception as e:
|
| logger.error(f"Ошибка проверки двойной подписки: {e}")
|
| return False
|
|
|
|
|
|
|
|
|
| def get_main_menu(free_used=False):
|
| kb = Keyboard(one_time=False)
|
| if not free_used:
|
| kb.add(Text("🎁 Попробовать бесплатно"), color=KeyboardButtonColor.POSITIVE).row()
|
|
|
| kb.add(Text("📸 Выбрать стиль фото"), color=KeyboardButtonColor.PRIMARY).row()
|
| kb.add(Text("🎨 Режим творца"), color=KeyboardButtonColor.PRIMARY).row()
|
| kb.add(Text("🔄 Сменить пол"), color=KeyboardButtonColor.SECONDARY)
|
| kb.add(Text("🔄 Сменить фото"), color=KeyboardButtonColor.SECONDARY).row()
|
| kb.add(Text("💰 Баланс"), color=KeyboardButtonColor.SECONDARY)
|
| kb.add(Text("❓ Как это работает?"), color=KeyboardButtonColor.SECONDARY).row()
|
| kb.add(Text("🎟 Промокод"), color=KeyboardButtonColor.PRIMARY)
|
| kb.add(OpenLink(SUPPORT_URL, "🆘 Поддержка"))
|
| return kb.get_json()
|
|
|
|
|
| def get_promo_check_kb():
|
| """Клавиатура с кнопками подписки и проверки"""
|
| kb = Keyboard(inline=True)
|
| kb.add(OpenLink(f"https://vk.com/club{GROUP_ID}", "📢 Подписаться 1"))
|
| kb.add(OpenLink(f"https://vk.com/club{ADDITIONAL_GROUP_ID}", "📢 Подписаться 2"))
|
| kb.row()
|
| kb.add(Text("✅ Я подписан(а)"), color=KeyboardButtonColor.POSITIVE)
|
| return kb.get_json()
|
|
|
|
|
| def get_fast_packs_menu():
|
| kb = Keyboard(inline=True)
|
| packs = supabase.table("settings").select("*").eq("is_premium", False).order("price").execute().data
|
| for i, pack in enumerate(packs):
|
| kb.add(Text(f"📸 {pack['name']} - {int(pack['price'])} р"))
|
| if (i + 1) % 2 == 0:
|
| kb.row()
|
| if len(packs) % 2 != 0:
|
| kb.row()
|
| kb.add(Text("⬅️ В главное меню"), color=KeyboardButtonColor.SECONDARY)
|
| return kb.get_json()
|
|
|
|
|
| def get_gender_menu():
|
| return (Keyboard(inline=True)
|
| .add(Text("🙋♂️ Мужской"), color=KeyboardButtonColor.PRIMARY)
|
| .add(Text("🙋♀️ Женский"), color=KeyboardButtonColor.PRIMARY)
|
| ).get_json()
|
|
|
|
|
| def get_styles_menu(gender):
|
| kb = Keyboard(one_time=False, inline=False)
|
| g = gender if gender else "female"
|
|
|
| if g in prompts.FACE_SWAP_STYLES:
|
| styles_list = [s for s in prompts.FACE_SWAP_STYLES[g].keys() if s != "🎁 Попробовать бесплатно"]
|
| for i, style in enumerate(styles_list):
|
| kb.add(Text(style))
|
| if (i + 1) % 2 == 0:
|
| kb.row()
|
| if len(styles_list) % 2 != 0:
|
| kb.row()
|
|
|
| kb.add(Text("⬅️ В главное меню"), color=KeyboardButtonColor.SECONDARY)
|
| return kb.get_json()
|
|
|
|
|
|
|
|
|
| @bot.on.message(text=["Начать", "начать", "Меню", "start", "⬅️ В главное меню"])
|
| async def start_handler(message: Message):
|
| user = get_user(message.from_id)
|
| supabase.table("users").update({"status": "main"}).eq("id", message.from_id).execute()
|
|
|
| if not user['free_used']:
|
| msg = ("Привет 👋\n"
|
| "Добро пожаловать в твоего персонального нейрофотографа!\n\n"
|
| "🎁Для тебя есть подарок — 1 бесплатная генерация.\n"
|
| "Нажми кнопку «Попробовать бесплатно»👇\n\n")
|
| else:
|
| msg = ("Выбирай режим в меню ниже, чтобы создать новые шедевры ✨")
|
|
|
| await message.answer(msg, keyboard=get_main_menu(user['free_used']))
|
|
|
|
|
| @bot.on.message(text="❓ Как это работает?")
|
| async def help_handler(message: Message):
|
| user = get_user(message.from_id)
|
| help_text = (
|
| "⚙️ ИНСТРУКЦИЯ ПО ИСПОЛЬЗОВАНИЮ\n\n"
|
| "🎁 1. Бесплатный старт\n"
|
| "Для знакомства с сервисом тебе доступна 1 тестовая генерация.\n"
|
| "Она поможет оценить качество и стиль фотографий.\n\n"
|
| "⚡ 2. Выбор стиля фото\n"
|
| "Нейросеть переносит твоё лицо с одного селфи на готовый шаблон.\n"
|
| "Идеально, если нужен быстрый и эффектный результат.\n\n"
|
| "🎨 3. Режим творца\n"
|
| "Опиши любой образ или идею — нейросеть создаст фото по твоему сценарию.\n"
|
| "Полная свобода для фантазии и экспериментов.\n\n"
|
| "⚠️ ВАЖНО\n"
|
| "Присылай чёткие фото лица: без очков, головных уборов и посторонних предметов."
|
| )
|
| await message.answer(help_text, keyboard=get_main_menu(user['free_used']))
|
|
|
|
|
| @bot.on.message(text="🎟 Промокод")
|
| async def promo_handler(message: Message):
|
| supabase.table("users").update({"status": "wait_promo"}).eq("id", message.from_id).execute()
|
| await message.answer("Введите промокод:")
|
|
|
|
|
| @bot.on.message(text="✅ Я подписан(а)")
|
| async def promo_check_after_sub(message: Message):
|
| user = get_user(message.from_id)
|
| if user['status'] != "check_subs_for_promo" or not user.get('temp_promo'):
|
| return await start_handler(message)
|
|
|
| if await is_subscribed_both(message.from_id):
|
| return await apply_promo_logic(message, user['temp_promo'], user)
|
| else:
|
| await message.answer("Вы всё еще не подписаны на оба сообщества! 📢", keyboard=get_promo_check_kb())
|
|
|
|
|
| async def apply_promo_logic(message: Message, code: str, user: dict):
|
| """Вспомогательная функция для начисления промокода"""
|
| res = supabase.table("promocodes").select("*").eq("code", code).execute()
|
| if not res.data: return
|
|
|
| promo = res.data[0]
|
|
|
| supabase.rpc("increment_balance", {"user_id": message.from_id, "amount": promo['amount']}).execute()
|
|
|
| supabase.table("promocodes").update({"uses_left": promo['uses_left'] - 1}).eq("code", code).execute()
|
|
|
| supabase.table("promocode_history").insert({"user_id": message.from_id, "code": code}).execute()
|
|
|
| supabase.table("users").update({"status": "main", "temp_promo": None}).eq("id", message.from_id).execute()
|
|
|
| await message.answer(f"✅ Промокод активирован! Начислено генераций: {promo['amount']}",
|
| keyboard=get_main_menu(user['free_used']))
|
|
|
|
|
| @bot.on.message(text=["🎁 Попробовать бесплатно", "✅ Я подписался"])
|
| async def free_trial_handler(message: Message):
|
| user = get_user(message.from_id)
|
|
|
| if user['free_used']:
|
| return await message.answer("Вы уже использовали бесплатную попытку!", keyboard=get_main_menu(True))
|
|
|
| if not user['gender']:
|
| supabase.table("users").update({"status": "wait_gender_free"}).eq("id", message.from_id).execute()
|
| return await message.answer("Сначала выберите ваш пол:", keyboard=get_gender_menu())
|
|
|
| if not user['primary_photo']:
|
| supabase.table("users").update({"status": "wait_primary_photo"}).eq("id", message.from_id).execute()
|
| return await message.answer("📸 Загрузи 1 фото для бесплатной генерации\n\n"
|
| "Пожалуйста, обрати внимание на требования:\n"
|
| "• в кадре только ты\n"
|
| "• лицо крупное и чёткое\n"
|
| "• без очков, головных уборов и фильтров\n"
|
| "• без других людей в кадре\n\n"
|
| "Я создам для тебя стильное нейро-фото ✨\n\n"
|
| "Загрузи фотографию прямо сейчас ⬇️")
|
|
|
| if not await is_subscribed(message.from_id):
|
| subscribe_kb = (
|
| Keyboard(inline=True)
|
| .add(OpenLink(f"https://vk.com/club{GROUP_ID}", "📢 Подписаться"))
|
| .row()
|
| .add(Text("✅ Я подписался"), color=KeyboardButtonColor.POSITIVE)
|
| ).get_json()
|
|
|
| return await message.answer(
|
| "🎁 Для бесплатной генерации подпишись на наше сообщество!\n\n"
|
| "После подписки нажмите кнопку «✅ Я подписался» ниже 👇",
|
| keyboard=subscribe_kb
|
| )
|
|
|
| await message.answer("⏳ Создаю фотографию... Это займет около 30 секунд.")
|
|
|
| gender = user['gender']
|
| prompt = prompts.FACE_SWAP_STYLES[gender]["🎁 Попробовать бесплатно"]
|
|
|
|
|
| paths_to_clean = []
|
| bucket_path, temporary_url = await upload_to_bucket(user['primary_photo'], message.from_id, "src")
|
| if bucket_path: paths_to_clean.append(bucket_path)
|
|
|
| res_url = await generator.generate_image(
|
| prompt=prompt,
|
| gender=gender,
|
| photo_url=temporary_url if temporary_url else user['primary_photo'],
|
| lora_model=None
|
| )
|
|
|
| if res_url:
|
|
|
| res_path, res_url_bucket = await upload_to_bucket(res_url, message.from_id, "res")
|
| if res_path: paths_to_clean.append(res_path)
|
|
|
| if res_url_bucket:
|
| try:
|
|
|
| image_bytes = await download_file(res_url_bucket)
|
| if image_bytes:
|
| photo_vk = await upload_photo_to_vk(image_bytes, message.peer_id, file_name="result.png")
|
| await message.answer("Ваше фото готово! ✨", attachment=photo_vk)
|
|
|
| promo_text = (
|
| "Теперь ты можешь выбрать режим:\n\n"
|
| "📸Выбор стиля фото\n"
|
| "Создание профессионального нейро-фото по одной твоей фотографии лица. "
|
| "Идеально, если нужен быстрый и эффектный результат.\n\n"
|
| "🎨Режим творца\n"
|
| "Создавай любые сюжеты для фото. Полная свобода для твоей фантазии 🎭\n\n"
|
| "Попробуем?\n"
|
| "Выбирай режим в меню ниже 📸💫"
|
| )
|
| await message.answer(promo_text, keyboard=get_main_menu(True))
|
|
|
| supabase.table("users").update({"free_used": True}).eq("id", message.from_id).execute()
|
| else:
|
| await message.answer("Ошибка при обработке результата.")
|
| except Exception as e:
|
| logger.error(f"Ошибка в free_trial VK upload: {e}")
|
| await message.answer("Техническая ошибка при отправке фото.")
|
| else:
|
| await message.answer("Ошибка при обработке результата.")
|
| else:
|
| await message.answer("Произошла ошибка при генерации. Попробуйте другое фото.")
|
|
|
|
|
| if paths_to_clean:
|
| supabase.storage.from_(BUCKET_NAME).remove(paths_to_clean)
|
|
|
|
|
| @bot.on.message(text=["💰 Баланс", "💰 Баланс / Профиль"])
|
| async def balance_handler(message: Message):
|
| user = get_user(message.from_id)
|
| photo_st = "Есть" if user['primary_photo'] else "Нет"
|
|
|
| gender_ru = "Не выбран"
|
| if user['gender'] == "male":
|
| gender_ru = "Мужской"
|
| elif user['gender'] == "female":
|
| gender_ru = "Женский"
|
|
|
| kb = Keyboard()
|
| kb.add(Text("🛒 Купить генерации")).row()
|
| kb.add(Text("⬅️ В главное меню"))
|
|
|
| await message.answer(
|
| f"🪙 Баланс: {user['balance']} генераций\n"
|
| f"👤 Пол: {gender_ru}\n"
|
| f"🖼 Закрепленная фотография: {photo_st}",
|
| keyboard=kb.get_json()
|
| )
|
|
|
|
|
| @bot.on.message(text="🛒 Купить генерации")
|
| async def shop_handler(message: Message):
|
| await message.answer("Выберите количество генераций:", keyboard=get_fast_packs_menu())
|
|
|
|
|
| @bot.on.message(text="📸 Выбрать стиль фото")
|
| async def styles_handler(message: Message):
|
| user = get_user(message.from_id)
|
| if not user['gender']: return await message.answer("Сначала выберите пол:", keyboard=get_gender_menu())
|
|
|
| await message.answer("Здесь ты можешь создать стильное фото за 30 секунд.\n\n"
|
| "В каждом стиле доступно 10 разных вариантов фотографий.\n"
|
| "При каждой генерации варианты выбираются случайно.\n\n"
|
| "✨ Важно знать\n"
|
| "Даже если выбрать один и тот же стиль несколько раз,\n"
|
| "ты каждый раз будешь получать разные фотографии.\n\n"
|
| "🎭 Выбери стиль\n", keyboard=get_styles_menu(user['gender']))
|
|
|
|
|
| @bot.on.message(text="🎨 Режим творца")
|
| async def creator_mode(message: Message):
|
| user = get_user(message.from_id)
|
| if user['balance'] <= 0:
|
| return await message.answer("🪙 Недостаточно генераций для режима творца. Пополните баланс:",
|
| keyboard=get_fast_packs_menu())
|
|
|
| supabase.table("users").update({"status": "wait_custom_prompt"}).eq("id", message.from_id).execute()
|
| kb = Keyboard(inline=True).add(Text("⬅️ В главное меню"), color=KeyboardButtonColor.SECONDARY).get_json()
|
| await message.answer(
|
| "Здесь вы можете создавать образы по своему запросу: стиль, одежда, фон, настроение. Напишите запрос ниже👇",
|
| keyboard=kb)
|
|
|
|
|
|
|
| @bot.on.message()
|
| async def main_handler(message: Message):
|
| user = get_user(message.from_id)
|
| text = message.text
|
|
|
| if text == "⬅️ В главное меню":
|
| supabase.table("users").update({"status": "main"}).eq("id", message.from_id).execute()
|
| return await start_handler(message)
|
|
|
|
|
| if user['status'] == "wait_promo":
|
| code = text.strip().upper()
|
|
|
| res = supabase.table("promocodes").select("*").eq("code", code).eq("active", True).execute()
|
| if not res.data or res.data[0]['uses_left'] <= 0:
|
| supabase.table("users").update({"status": "main"}).eq("id", message.from_id).execute()
|
| return await message.answer("❌ Промокод не существует или устарел",
|
| keyboard=get_main_menu(user['free_used']))
|
|
|
|
|
| hist = supabase.table("promocode_history").select("*").eq("user_id", message.from_id).eq("code", code).execute()
|
| if hist.data:
|
| supabase.table("users").update({"status": "main"}).eq("id", message.from_id).execute()
|
| return await message.answer("⛔ Вы уже использовали этот промокод!",
|
| keyboard=get_main_menu(user['free_used']))
|
|
|
|
|
| if await is_subscribed_both(message.from_id):
|
| return await apply_promo_logic(message, code, user)
|
| else:
|
|
|
| supabase.table("users").update({"status": "check_subs_for_promo", "temp_promo": code}).eq("id",
|
| message.from_id).execute()
|
| return await message.answer("Для активации промокода нужно подписаться на два наших сообщества! 👇",
|
| keyboard=get_promo_check_kb())
|
|
|
|
|
| packs = supabase.table("settings").select("*").execute().data
|
| for pack in packs:
|
| if pack['name'] in text and str(int(pack['price'])) in text:
|
| pay_url = (
|
| f"https://yoomoney.ru/quickpay/confirm.xml?"
|
| f"receiver={YOOMONEY_WALLET}&"
|
| f"quickpay-form=button&"
|
| f"sum={pack['price']}&"
|
| f"label={message.from_id}&"
|
| f"targets=Оплата+пакета+{pack['name'].replace(' ', '+')}"
|
| )
|
| kb = Keyboard(inline=True).add(OpenLink(pay_url, "💳 Перейти к оплате")).get_json()
|
| return await message.answer(f"Вы выбрали: {pack['name']}\nК оплате: {int(pack['price'])} руб.", keyboard=kb)
|
|
|
|
|
| if text in ["🙋♂️ Мужской", "🙋♀️ Женский"]:
|
| g = "male" if "Мужской" in text else "female"
|
| supabase.table("users").update({"gender": g}).eq("id", message.from_id).execute()
|
|
|
| if user['status'] == "wait_gender_change":
|
| supabase.table("users").update({"status": "main"}).eq("id", message.from_id).execute()
|
| return await message.answer("✅ Изменения сохранены!", keyboard=get_main_menu(user['free_used']))
|
|
|
| if not user['primary_photo']:
|
| supabase.table("users").update({"status": "wait_primary_photo"}).eq("id", message.from_id).execute()
|
| return await message.answer(
|
| "📸 Загрузи 1 фото для бесплатной генерации\n\n"
|
| "Загрузи фотографию прямо сейчас ⬇️"
|
| )
|
| else:
|
| supabase.table("users").update({"status": "main"}).eq("id", message.from_id).execute()
|
| return await message.answer(f"✅ Пол сохранен!", keyboard=get_main_menu(user['free_used']))
|
|
|
| if text == "🔄 Сменить пол":
|
| supabase.table("users").update({"status": "wait_gender_change"}).eq("id", message.from_id).execute()
|
| return await message.answer("Выберите ваш пол:", keyboard=get_gender_menu())
|
|
|
| if text == "🔄 Сменить фото":
|
| supabase.table("users").update({"status": "wait_primary_photo"}).eq("id", message.from_id).execute()
|
| return await message.answer("Пришлите новое четкое фото лица:")
|
|
|
|
|
| if user['status'] == "wait_primary_photo" or (message.attachments and not user['primary_photo']):
|
| if message.attachments:
|
| photo_sizes = message.attachments[0].photo.sizes
|
| photo_url = sorted(photo_sizes, key=lambda x: x.width)[-1].url
|
| supabase.table("users").update({"primary_photo": photo_url, "status": "main"}).eq("id",
|
| message.from_id).execute()
|
|
|
| if not user['free_used']:
|
| await message.answer("⏳ Создаю фотографию... Это займет около 30 секунд.")
|
| gender = user['gender'] or "female"
|
| prompt = prompts.FACE_SWAP_STYLES[gender]["🎁 Попробовать бесплатно"]
|
|
|
|
|
| paths_to_clean = []
|
| bucket_path, temporary_url = await upload_to_bucket(photo_url, message.from_id, "src")
|
| if bucket_path: paths_to_clean.append(bucket_path)
|
|
|
| res_url = await generator.generate_image(prompt=prompt, gender=gender,
|
| photo_url=temporary_url if temporary_url else photo_url)
|
|
|
| if res_url:
|
| res_path, res_url_bucket = await upload_to_bucket(res_url, message.from_id, "res")
|
| if res_path: paths_to_clean.append(res_path)
|
|
|
| if res_url_bucket:
|
| try:
|
|
|
| image_bytes = await download_file(res_url_bucket)
|
| if image_bytes:
|
| photo_vk = await upload_photo_to_vk(image_bytes, message.peer_id, file_name="result.png")
|
| await message.answer("Ваше фото готово! ✨", attachment=photo_vk)
|
|
|
| supabase.table("users").update({"free_used": True}).eq("id", message.from_id).execute()
|
| if paths_to_clean: supabase.storage.from_(BUCKET_NAME).remove(paths_to_clean)
|
| return await message.answer("Выбирай режим в меню ниже 📸💫", keyboard=get_main_menu(True))
|
| except Exception as e:
|
| logger.error(f"VK Error: {e}")
|
|
|
| if paths_to_clean: supabase.storage.from_(BUCKET_NAME).remove(paths_to_clean)
|
| return await message.answer("Произошла ошибка при генерации.")
|
|
|
| return await message.answer("✅ Фото лица принято!", keyboard=get_main_menu(user['free_used']))
|
|
|
|
|
| gender = user['gender'] or "female"
|
| styles_dict = prompts.FACE_SWAP_STYLES.get(gender, {})
|
|
|
| if text in styles_dict or user['status'] == "wait_custom_prompt":
|
| if user['balance'] <= 0:
|
| return await message.answer("🪙 Недостаточно генераций. Пополните баланс:", keyboard=get_fast_packs_menu())
|
|
|
| if not user['primary_photo']:
|
| supabase.table("users").update({"status": "wait_primary_photo"}).eq("id", message.from_id).execute()
|
| return await message.answer("Сначала отправьте фото вашего лица")
|
|
|
| if text in styles_dict:
|
| options = styles_dict[text]
|
| last_val = user.get('last_style') or ""
|
| last_idx = -1
|
| if "|" in last_val and last_val.split("|")[0] == text:
|
| try:
|
| last_idx = int(last_val.split("|")[1])
|
| except:
|
| last_idx = -1
|
| available_indices = [i for i in range(len(options)) if i != last_idx]
|
| new_idx = random.choice(available_indices)
|
| prompt = options[new_idx]
|
| new_last_style = f"{text}|{new_idx}"
|
| else:
|
| prompt = text
|
| new_last_style = "custom_prompt|0"
|
|
|
| await message.answer("🎨 Генерирую образ... Пожалуйста, подождите.")
|
|
|
|
|
| paths_to_clean = []
|
| bucket_path, temporary_url = await upload_to_bucket(user['primary_photo'], message.from_id, "src")
|
| if bucket_path: paths_to_clean.append(bucket_path)
|
|
|
| res_url = await generator.generate_image(
|
| prompt=prompt,
|
| gender=gender,
|
| photo_url=temporary_url if temporary_url else user['primary_photo']
|
| )
|
|
|
| if res_url:
|
| res_path, res_url_bucket = await upload_to_bucket(res_url, message.from_id, "res")
|
| if res_path: paths_to_clean.append(res_path)
|
|
|
| if res_url_bucket:
|
| try:
|
|
|
| image_bytes = await download_file(res_url_bucket)
|
| if image_bytes:
|
| photo_vk = await upload_photo_to_vk(image_bytes, message.peer_id, file_name="style.png")
|
| await message.answer("Готово! ✨", attachment=photo_vk)
|
| supabase.rpc("increment_balance", {"user_id": message.from_id, "amount": -1}).execute()
|
|
|
| supabase.table("users").update(
|
| {"last_style": new_last_style, "status": "main"}).eq("id",
|
| message.from_id).execute()
|
| else:
|
| await message.answer("ВК не принял фото. Баланс сохранен.")
|
| except Exception as e:
|
| logger.error(f"VK Style Error: {e}")
|
| await message.answer("ВК не принял фото. Баланс сохранен.")
|
| else:
|
| await message.answer("Ошибка нейросети. Попробуйте другой запрос.")
|
|
|
| if paths_to_clean: supabase.storage.from_(BUCKET_NAME).remove(paths_to_clean)
|
| supabase.table("users").update({"status": "main"}).eq("id", message.from_id).execute()
|
| return
|
|
|
| await message.answer("Воспользуйтесь меню 👇", keyboard=get_main_menu(user['free_used']))
|
|
|
|
|
| @app.route('/yoomoney', methods=['POST'])
|
| def yoomoney_webhook():
|
| d = request.form
|
| check = f"{d.get('notification_type')}&{d.get('operation_id')}&{d.get('amount')}&{d.get('currency')}&{d.get('datetime')}&{d.get('sender')}&{d.get('codepro')}&{YOOMONEY_SECRET}&{d.get('label')}"
|
| my_hash = hashlib.sha1(check.encode('utf-8')).hexdigest()
|
| if my_hash == d.get('sha1_hash'):
|
| try:
|
| u_id = int(d.get('label'))
|
| received_amount = float(d.get('withdraw_amount') or d.get('amount'))
|
| packs = supabase.table("settings").select("*").execute().data
|
| for pack in packs:
|
| if abs(received_amount - float(pack['price'])) / float(pack['price']) < 0.05:
|
| supabase.rpc("increment_balance", {"user_id": u_id, "amount": pack['amount']}).execute()
|
| if loop:
|
| asyncio.run_coroutine_threadsafe(bot.api.messages.send(peer_id=u_id, random_id=0,
|
| message=f"✅ Оплата прошла успешно! Пакет «{pack['name']}» активирован."),
|
| loop)
|
| break
|
| except Exception as e:
|
| logger.error(f"Webhook error: {e}")
|
| return "OK", 200
|
|
|
|
|
| @app.route('/')
|
| def index():
|
| return "<h1>VK Bot is Running</h1>", 200
|
|
|
|
|
| if __name__ == "__main__":
|
| loop = asyncio.get_event_loop()
|
| threading.Thread(target=lambda: app.run(host="0.0.0.0", port=7860, use_reloader=False), daemon=True).start()
|
| bot.run_forever() |
