uzbekgpt/commands.py

from io import BytesIO
from typing import Optional, Union

from aiogram import Bot, Dispatcher, F, Router
from aiogram.methods import SendRichMessage
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ChatAction, ChatType
from aiogram.exceptions import TelegramForbiddenError
from aiogram.filters import Command, CommandStart, CommandObject
from aiogram.types import (
    BufferedInputFile,
    CallbackQuery,
    ChosenInlineResult,
    InlineKeyboardButton,
    InlineKeyboardMarkup,
    InlineQuery,
    InlineQueryResultArticle,
    InputRichMessage,
    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", "/stop", "/markpidor"]

@router.message(Command("stats"))
async def stats_handler(message: Message):
    users = get_all_users()

    total_users = len(users)

    models = Counter()
    streams = Counter()
    calls = Counter()
    prompts = 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
            prompts[data.get("prompt")] += 1

    top_models = models.most_common(5)
    top_streams = streams.most_common(2)
    top_prompts = prompts.most_common(5)

    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')}",
        "🥺топ 5 моделей:",
        *(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()),
        "📝топ 5 промптов:",
        *(f"- `{c}`: {n}" for c, n in prompts.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(only_notify=True)

    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("stop"))
async def stop_handler(message: Message):
    if message.sender_chat:
        user_id = message.sender_chat.id
    else:
        user_id = message.from_user.id
        
    config = get_user_config(user_id)
    set_user_config(
        user_id, 
        config.get("model"), 
        config.get("stream"), 
        config.get("call"), 
        config.get("prompt"), 
        "net", 
        config.get("markdown")
    )
    await message.reply("✅ *рассылка успешно отключена!* чтобы включить её обратно:\n/settings --> рассылка")

@router.message(Command("ben"))
async def ben_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 reporn_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(CommandStart())
async def start_handler(message: Message, command: CommandObject):
    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
    
    config = get_user_config(user_id)
    
    payload = command.args
    if payload == "new_md":
        set_user_config(
            user_id, 
            config.get("model"), 
            config.get("stream"), 
            config.get("call"), 
            config.get("prompt"), 
            config.get("notify"), 
            "new"
        )
        await message.answer("✅ установлен *новый* стиль выделения сообщений!")
    elif payload == "old_md":
        set_user_config(
            user_id, 
            config.get("model"), 
            config.get("stream"), 
            config.get("call"), 
            config.get("prompt"), 
            config.get("notify"), 
            "old"
        )
        await message.answer("✅ установлен *старый* стиль выделения сообщений!")

    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 message.sender_chat:
        user_id = message.sender_chat.id
    else:
        user_id = message.from_user.id
    config = get_user_config(user_id)
    system_prompt_name = config["prompt"]
    if system_prompt_name in SYSTEM_PROMPTS:
        system_prompt = SYSTEM_PROMPTS[system_prompt_name]
    else:
        await message.reply(
            f"🤝 произошла ошибка при отображении промпта `{system_prompt_name}`\n*для фото*: ```prompt\n{IMAGE_PROMPT}```изменить системный промпт узбекгпт пока что нельзя."
        )
        return
    
    if PROMPT_COMMAND:
        await message.reply(
            f"🤝 *ваш выбранный промпт* `{system_prompt_name}`: ```prompt\n{system_prompt}```\n*для фото*: ```prompt\n{IMAGE_PROMPT}```\nизменить системный промпт узбекгпт пока что нельзя."
        )
    else:
        if SHOW_ACC_MSG:
            await message.reply(
                "🚫 *данный команда отключен в этом боте.* попросите владельца включить🤝"
            )
    return


@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"
            "/settings для изменения настроек!"
        )
    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())
    prompts_count = len(SYSTEM_PROMPTS)

    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"

    if config["notify"] == "da":
        notify_status = "da"
        notify_cb = "net"
    else:
        notify_status = "net"
        notify_cb = "da"

    keyboard = InlineKeyboardMarkup(
        inline_keyboard=[
            [
                InlineKeyboardButton(
                    text=f"модели ({models_count})", callback_data=f"models%{user_id}"
                )
            ],
            [
                InlineKeyboardButton(
                    text=f"промпты ({prompts_count})", callback_data=f"prompts%{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=f"рассылка: {notify_status}",
                    callback_data=f"set_notify%{user_id}%{notify_cb}",
                )
            ],
            [InlineKeyboardButton(text="донат", callback_data=f"donate%{user_id}")],
            [InlineKeyboardButton(text="о боте", callback_data=f"about%{user_id}")],
        ]
    )

    text = f"⚡ *алло узбекгпт настройки*\nтекущая модель: `{config.get('model')}`\n" \
        f"текущий промпт: `{config.get('prompt')}`"

    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}")

        
@router.message(Command("markpidor"))
async def alo(message: Message):
    await message.reply_rich(InputRichMessage(markdown="# ало"))
    return