Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 61 additions & 1 deletion python/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
import requests

from modules import (
BetterBotBaseDataService, Commands
BetterBotBaseDataService, Commands, VkKeyboard, TopPaginationKeyboard
)
from tokens import BOT_TOKEN
from userbot import UserBot
Expand Down Expand Up @@ -112,6 +112,26 @@ def message_new(
except Exception as e:
print(e)

def message_event(
self,
event: Dict[str, Any]
) -> NoReturn:
"""Handling callback button events.
"""
event_data = event["object"]
peer_id = event_data["peer_id"]
from_id = event_data["user_id"]
payload = event_data.get("payload", {})
event_id = event_data["event_id"]

# Get user for callback handling
user = self.data.get_user(from_id, self) if from_id > 0 else None

try:
self.commands.process_callback(
payload, peer_id, from_id, event_id, user)
except Exception as e:
print(f"Error handling callback: {e}")

def delete_message(
self,
Expand Down Expand Up @@ -168,6 +188,46 @@ def send_msg(
message=msg, peer_id=peer_id,
disable_mentions=1, random_id=0))

def send_msg_with_keyboard(
self,
msg: str,
peer_id: int,
keyboard: str
) -> NoReturn:
"""Sends message with keyboard to chat with {peer_id}.

:param msg: message text
:param peer_id: chat ID
:param keyboard: JSON keyboard string
"""
self.call_method(
'messages.send',
dict(
message=msg, peer_id=peer_id,
keyboard=keyboard,
disable_mentions=1, random_id=0))

def send_callback_answer(
self,
event_id: str,
peer_id: int,
event_data: Dict[str, Any] = None
) -> NoReturn:
"""Send answer to callback query.

:param event_id: Event ID from callback
:param peer_id: chat ID
:param event_data: Optional event data (for snackbar, etc.)
"""
params = {
"event_id": event_id,
"peer_id": peer_id
}
if event_data:
params["event_data"] = event_data

self.call_method('messages.sendMessageEventAnswer', params)

def get_user_name(
self,
uid: int,
Expand Down
1 change: 1 addition & 0 deletions python/modules/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
from .data_service import BetterBotBaseDataService
from .data_builder import DataBuilder
from .vk_instance import VkInstance
from .keyboard_utils import VkKeyboard, TopPaginationKeyboard
from .utils import (
get_default_programming_language,
contains_string,
Expand Down
139 changes: 127 additions & 12 deletions python/modules/commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from datetime import datetime
from time import time
import os
import json

from regex import Pattern, Match, split, match, search, IGNORECASE, sub
from requests import post
Expand All @@ -13,6 +14,7 @@
from .commands_builder import CommandsBuilder
from .data_service import BetterBotBaseDataService
from .data_builder import DataBuilder
from .keyboard_utils import TopPaginationKeyboard
from .utils import (
get_default_programming_language,
contains_all_strings,
Expand Down Expand Up @@ -124,22 +126,28 @@ def top(
self,
reverse: bool = False
) -> NoReturn:
"""Sends users top."""
"""Sends users top with pagination."""
if self.peer_id < 2e9:
return
maximum_users = self.matched.group("maximum_users")
maximum_users = int(maximum_users) if maximum_users else -1
users = DataBuilder.get_users_sorted_by_karma(
self.vk_instance, self.data_service, self.peer_id)
users = [i for i in users if
(i["karma"] != 0) or
("programming_languages" in i and len(i["programming_languages"]) > 0)
]
self.vk_instance.send_msg(
CommandsBuilder.build_top_users(

# If specific maximum_users is requested, use old behavior without pagination
if maximum_users:
maximum_users = int(maximum_users)
users = DataBuilder.get_users_sorted_by_karma(
self.vk_instance, self.data_service, self.peer_id)
users = [i for i in users if
(i["karma"] != 0) or
("programming_languages" in i and len(i["programming_languages"]) > 0)
]
message = CommandsBuilder.build_top_users(
users, self.data_service, reverse,
self.karma_enabled, maximum_users),
self.peer_id)
self.karma_enabled, maximum_users)
if message:
self.vk_instance.send_msg(message, self.peer_id)
else:
# Use new paginated version
self.top_paginated(reverse=reverse, page=0)

def top_langs(
self,
Expand Down Expand Up @@ -436,3 +444,110 @@ def process(
if self.matched:
action()
return

def process_callback(
self,
payload: Dict[str, Any],
peer_id: int,
from_id: int,
event_id: str,
user: BetterUser
) -> NoReturn:
"""Process callback button events

:param payload: Callback payload data
:param peer_id: chat ID
:param from_id: user ID
:param event_id: callback event ID
:param user: user object
"""
self.peer_id = peer_id
self.from_id = from_id
self.karma_enabled = peer_id in config.CHATS_KARMA_WHITELIST
self.current_user = user

if from_id < 0:
return

try:
# Parse payload if it's a string
if isinstance(payload, str):
payload = json.loads(payload)

action = payload.get("action")

if action == "paginate":
self.handle_top_pagination(payload, event_id)
elif action == "page_info":
# Just acknowledge the page info button click
self.vk_instance.send_callback_answer(event_id, peer_id)

except Exception as e:
print(f"Error processing callback: {e}")
# Send empty callback answer to acknowledge the button press
self.vk_instance.send_callback_answer(event_id, peer_id)

def handle_top_pagination(
self,
payload: Dict[str, Any],
event_id: str
) -> NoReturn:
"""Handle pagination for top command"""
command = payload.get("command", "top")
page = payload.get("page", 0)
reverse = payload.get("reverse", False)

# Execute the appropriate top command with pagination
if command == "top":
self.top_paginated(reverse=reverse, page=page)
elif command == "bottom":
self.top_paginated(reverse=True, page=page)
elif command == "people":
self.top_paginated(reverse=reverse, page=page)

# Acknowledge the callback
self.vk_instance.send_callback_answer(event_id, self.peer_id)

def top_paginated(
self,
reverse: bool = False,
page: int = 0
) -> NoReturn:
"""Sends paginated users top with callback buttons."""
if self.peer_id < 2e9:
return

users = DataBuilder.get_users_sorted_by_karma(
self.vk_instance, self.data_service, self.peer_id)
users = [i for i in users if
(i["karma"] != 0) or
("programming_languages" in i and len(i["programming_languages"]) > 0)
]

if reverse:
users = list(reversed(users))

# Calculate pagination
pagination_info = TopPaginationKeyboard.calculate_pagination(len(users))
total_pages = pagination_info["total_pages"]

# Ensure page is within valid range
page = max(0, min(page, total_pages - 1))

# Get users for current page
page_users = TopPaginationKeyboard.get_page_users(users, page)

# Build message
command_type = "bottom" if reverse else "top"
message = CommandsBuilder.build_top_users(
page_users, self.data_service, False,
self.karma_enabled, -1)

if message:
# Create pagination keyboard
keyboard = TopPaginationKeyboard.create_pagination_keyboard(
page, total_pages, command_type, reverse)

self.vk_instance.send_msg_with_keyboard(message, self.peer_id, keyboard)
else:
self.vk_instance.send_msg("Пользователи не найдены.", self.peer_id)
Loading
Loading