297 lines
10 KiB
Python
297 lines
10 KiB
Python
from io import BytesIO
|
||
from typing import Optional, Union
|
||
|
||
from aiogram import Bot, Dispatcher, F, Router
|
||
from aiogram.client.default import DefaultBotProperties
|
||
from aiogram.enums import ChatAction, ChatType
|
||
from aiogram.exceptions import TelegramForbiddenError
|
||
from aiogram.filters import Command
|
||
from aiogram.types import (
|
||
BufferedInputFile,
|
||
CallbackQuery,
|
||
ChosenInlineResult,
|
||
InlineKeyboardButton,
|
||
InlineKeyboardMarkup,
|
||
InlineQuery,
|
||
InlineQueryResultArticle,
|
||
InputTextMessageContent,
|
||
Message,
|
||
User,
|
||
)
|
||
from aiogram.utils.deep_linking import create_start_link
|
||
|
||
from collections import Counter
|
||
|
||
from colors import *
|
||
from config import *
|
||
from statistics import *
|
||
from supergenerator import *
|
||
|
||
router = Router()
|
||
|
||
commands = ["/stats", "/broadcast", "/ben", "/reporn", "/start", "/prompt", "/config", "/settings", "/clear"]
|
||
|
||
@router.message(Command("stats"))
|
||
async def stats_handler(message: Message):
|
||
users = get_all_users()
|
||
|
||
total_users = len(users)
|
||
|
||
models = Counter()
|
||
streams = Counter()
|
||
calls = Counter()
|
||
|
||
for data in users.values():
|
||
if isinstance(data, dict):
|
||
models[data.get("model")] += 1
|
||
streams[data.get("stream")] += 1
|
||
calls[data.get("call")] += 1
|
||
|
||
top_models = models.most_common(3)
|
||
top_streams = streams.most_common(3)
|
||
|
||
text = [
|
||
"✅ *общая статистика*",
|
||
f"☝️пользователей: {total_users}",
|
||
f"📞всего генераций: {get_stats('stats/gens')}",
|
||
f"👻ген в гостевом: {get_stats('stats/guest')}",
|
||
f"🙏ген в инлайне: {get_stats('stats/inline')}",
|
||
f"🤝ben: {get_stats('stats/ben')}",
|
||
f"🤝reporn: {get_stats('stats/reporn')}",
|
||
"🥺топ моделей:",
|
||
*(f"- `{m}`: {c}" for m, c in top_models),
|
||
"✍️топ режимов стрима:",
|
||
*(f"- `{s}`: {c}" for s, c in top_streams),
|
||
"👀топ режимов вызова:",
|
||
*(f"- `{c}`: {n}" for c, n in calls.most_common()),
|
||
]
|
||
|
||
await message.reply("\n".join(text))
|
||
|
||
|
||
|
||
@router.message(Command("broadcast"))
|
||
async def broadcast_handler(message: Message):
|
||
if message.from_user.id != ADMIN_ID:
|
||
return await message.reply("🚫 pashol naxxuy.")
|
||
|
||
text = message.text.removeprefix("/broadcast").strip()
|
||
if not text:
|
||
return await message.reply("🚫 напиши текст после команды ало.")
|
||
|
||
users = get_all_user_ids()
|
||
|
||
success = 0
|
||
failed = 0
|
||
|
||
for user_id in users:
|
||
try:
|
||
if user_id > 0:
|
||
await message.bot.send_message(user_id, text)
|
||
success += 1
|
||
except TelegramForbiddenError:
|
||
failed += 1
|
||
except Exception as e:
|
||
print(f"{RED} -- broadcast err - {e}{RESET}")
|
||
failed += 1
|
||
|
||
await message.reply(
|
||
f"✅ *готово врат.*\n- успешно: {success}\n- не удалось: {failed}"
|
||
)
|
||
|
||
|
||
@router.message(Command("ben"))
|
||
async def start_handler(message: Message):
|
||
if message.chat.type in [ChatType.GROUP, ChatType.SUPERGROUP]:
|
||
if message.chat.id in BEN_IDS:
|
||
replied = message.reply_to_message or None
|
||
if replied:
|
||
name = replied.from_user.first_name
|
||
ben_id = replied.from_user.id
|
||
else:
|
||
name = message.from_user.first_name
|
||
ben_id = message.from_user.id
|
||
|
||
await message.reply(
|
||
f"{name} ({ben_id}) заbenен за Botting, spamming, and coordinated inauthentic behavior.😡✅\n"
|
||
f"админ: Ben (958011829)✅"
|
||
)
|
||
add_one_to("stats/ben")
|
||
return
|
||
|
||
|
||
|
||
@router.message(Command("reporn"))
|
||
async def start_handler(message: Message):
|
||
user_name = message.from_user.first_name or message.sender_chat.title or "вы"
|
||
if message.chat.type in [ChatType.GROUP, ChatType.SUPERGROUP]:
|
||
if message.chat.id in REPORN_IDS:
|
||
replied = message.reply_to_message or None
|
||
if replied:
|
||
name = replied.from_user.first_name
|
||
ben_id = replied.from_user.id
|
||
else:
|
||
name = message.from_user.first_name
|
||
ben_id = message.from_user.id
|
||
|
||
await message.reply(
|
||
f"отправлен репорт на {name} ({ben_id})!📞"
|
||
)
|
||
add_one_to("stats/reporn")
|
||
return
|
||
|
||
|
||
@router.message(Command("start"))
|
||
async def start_handler(message: Message):
|
||
if message.chat.type in [ChatType.GROUP, ChatType.SUPERGROUP]:
|
||
await message.reply(
|
||
"привет, запусти эту команду в ЛиС (личные сообщения) бота✅"
|
||
)
|
||
return
|
||
|
||
if message.sender_chat:
|
||
user_id = message.sender_chat.id
|
||
else:
|
||
user_id = message.from_user.id
|
||
|
||
invite_url = f"https://t.me/{BOT_USERNAME}?startgroup=start"
|
||
|
||
keyboard = InlineKeyboardMarkup(
|
||
inline_keyboard=[
|
||
[InlineKeyboardButton(text="команды", callback_data="start")],
|
||
[
|
||
InlineKeyboardButton(
|
||
text="настройки", callback_data=f"settings%{user_id}"
|
||
)
|
||
],
|
||
[InlineKeyboardButton(text="добавить в группу", url=invite_url)],
|
||
]
|
||
)
|
||
|
||
try:
|
||
await message.reply(
|
||
"Привет, я УзбекГПТ✅ национальный бот Узбикестан и я готов не помочь. Напиши любае сообщение или отправь файл чтоб я быстра ответил!1!1 Или нажми кнопку ниже чтоба посмотрет мои команды..",
|
||
reply_markup=keyboard,
|
||
)
|
||
except Exception as e:
|
||
print(f"{RED} -- {e}{RESET}")
|
||
|
||
|
||
@router.message(Command("prompt"))
|
||
async def prompt_handler(message: Message):
|
||
if PROMPT_COMMAND:
|
||
await message.reply(
|
||
f"🤝 *текущий системный промпт узбекГПТ*: ```prompt\n{SYSTEM_PROMPT}```*для фото*: ```prompt\n{IMAGE_PROMPT}```изменить системный промпт узбекгпт пока что нельзя."
|
||
)
|
||
else:
|
||
if SHOW_ACC_MSG:
|
||
await message.reply(
|
||
"🚫 *данный команда отключен в этом боте.* попросите владельца включить🤝"
|
||
)
|
||
|
||
|
||
@router.message(Command("config"))
|
||
async def config_handler(message: Message):
|
||
if CONFIG_COMMAND:
|
||
await message.reply(
|
||
f"🤝 *конфиг узбекгпт*:\n`MAX_CONTEXT` = `{MAX_CONTEXT}`\n`MAX_PROMPT` = `{MAX_PROMPT}`\n"
|
||
f"`DEFAULT_MODEL` = `{DEFAULT_MODEL}`\n`IMAGE_MODEL` = `{IMAGE_MODEL}`\n"
|
||
f"`DEFAULT_STREAM` = `{DEFAULT_STREAM}`\n"
|
||
"если нужно что-то изменить здесь, то обратитесь к владельцу бота🥺"
|
||
)
|
||
else:
|
||
if SHOW_ACC_MSG:
|
||
await message.reply(
|
||
"🚫 *данный команда отключен в этом боте.* попросите владельца включить🤝"
|
||
)
|
||
|
||
|
||
async def settings_handler(event: Union[Message, CallbackQuery]):
|
||
user_id = (
|
||
event.sender_chat.id
|
||
if isinstance(event, Message) and event.sender_chat
|
||
else event.from_user.id
|
||
)
|
||
|
||
config = get_user_config(user_id)
|
||
models_count = sum(len(models) for models in MODELS.values())
|
||
|
||
if config["stream"] == "native":
|
||
stream_status = "натив"
|
||
stream_cb = "edit"
|
||
elif config["stream"] == "edit":
|
||
stream_status = "редакт"
|
||
stream_cb = "none"
|
||
else:
|
||
stream_status = "net"
|
||
stream_cb = "native"
|
||
|
||
if config["call"] == "da":
|
||
call_status = "da"
|
||
call_cb = "net"
|
||
else:
|
||
call_status = "net"
|
||
call_cb = "da"
|
||
|
||
keyboard = InlineKeyboardMarkup(
|
||
inline_keyboard=[
|
||
[
|
||
InlineKeyboardButton(
|
||
text=f"модели ({models_count})", callback_data=f"models%{user_id}"
|
||
)
|
||
],
|
||
[
|
||
InlineKeyboardButton(
|
||
text=f"стрим: {stream_status}",
|
||
callback_data=f"set_stream%{user_id}%{stream_cb}",
|
||
)
|
||
],
|
||
[
|
||
InlineKeyboardButton(
|
||
text=f"отклик: {call_status}",
|
||
callback_data=f"set_call%{user_id}%{call_cb}",
|
||
)
|
||
],
|
||
[InlineKeyboardButton(text="донат", callback_data=f"donate%{user_id}")],
|
||
[InlineKeyboardButton(text="о боте", callback_data=f"about%{user_id}")],
|
||
]
|
||
)
|
||
|
||
text = f"⚡ *алло узбекгпт настройки*\nтекущая модель: `{config.get('model')}`"
|
||
|
||
if isinstance(event, Message):
|
||
await event.reply(text, reply_markup=keyboard)
|
||
else:
|
||
await event.message.edit_text(text, reply_markup=keyboard)
|
||
await event.answer()
|
||
|
||
|
||
@router.message(Command("settings"))
|
||
async def settings_command(message: Message):
|
||
if not SETTINGS_COMMAND:
|
||
if SHOW_ACC_MSG:
|
||
await message.reply(
|
||
"🚫 *данный команда отключен в этом боте.* попросите владельца включить🤝"
|
||
)
|
||
return
|
||
await settings_handler(message)
|
||
|
||
|
||
@router.message(Command("clear"))
|
||
async def clear_handler(message: Message):
|
||
if not CLEAR_COMMAND:
|
||
if SHOW_ACC_MSG:
|
||
await message.reply(
|
||
"🚫 *данный команда отключен в этом боте.* попросите владельца включить🤝"
|
||
)
|
||
return
|
||
user_id = message.sender_chat.id if message.sender_chat else message.from_user.id
|
||
|
||
user_contexts[user_id] = []
|
||
try:
|
||
await message.reply("контекст очищен. ✅")
|
||
except TelegramForbiddenError as e:
|
||
user_contexts[user_id] = []
|
||
except Exception as e:
|
||
print(f"{RED} -- {e}{RESET}")
|