import os
import logging
import asyncio
import random
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from pyrogram import filters
from pyrogram.client import Client
from pyrogram.types import Message
from pyrogram.enums import ParseMode
from dotenv import load_dotenv
import aiohttp
import requests
# Load environment variables
load_dotenv()
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=\[
logging.FileHandler('bot.log'),
logging.StreamHandler()
\]
)
logger = logging.getLogger(_name_)
@dataclass
class EffectInfo:
"""Data class for message effect information."""
id: str
name: str
emoji: str
description: str
category: str = "general"
class Config:
"""Configuration management for the bot."""
def \__init_\_(self):
"""Initialize configuration from environment variables."""
self.BOT_TOKEN = os.getenv('BOT_TOKEN')
self.API_ID = os.getenv('API_ID')
self.API_HASH = os.getenv('API_HASH')
\# Feature flags
self.ENABLE_EFFECTS = os.getenv('ENABLE_EFFECTS', 'true').lower() == 'true'
self.FALLBACK_ON_ERROR = os.getenv('FALLBACK_ON_ERROR', 'true').lower() == 'true'
\# Rate limiting
self.RATE_LIMIT_MESSAGES = int(os.getenv('RATE_LIMIT_MESSAGES', '30'))
self.RATE_LIMIT_WINDOW = int(os.getenv('RATE_LIMIT_WINDOW', '60'))
self.\_validate()
def \_validate(self):
"""Validate required configuration."""
required_vars = \['BOT_TOKEN', 'API_ID', 'API_HASH'\]
missing_vars = \[var for var in required_vars if not getattr(self, var)\]
if missing_vars:
raise ValueError(f"Missing required environment variables: {', '.join(missing_vars)}")
if self.API_ID:
try:
self.API_ID = int(self.API_ID)
except (ValueError, TypeError):
raise ValueError("API_ID must be a valid integer")
logger.info("ā
Configuration validated successfully")
class MessageEffects:
"""Manager for Telegram message effects."""
def \__init_\_(self):
"""Initialize message effects with predefined effect IDs."""
self.effects = {
'fire': EffectInfo(
id='5104841245755180586',
name='Fire',
emoji='š„',
description='Blazing fire effect',
category='energy'
),
'party': EffectInfo(
id='5046509860389126442',
name='Party',
emoji='š',
description='Celebration confetti',
category='celebration'
),
'heart': EffectInfo(
id='5044134455711629726',
name='Heart',
emoji='ā¤ļø',
description='Loving hearts effect',
category='emotion'
),
'thumbs_up': EffectInfo(
id='5107584321108051014',
name='Thumbs Up',
emoji='š',
description='Positive thumbs up',
category='reaction'
),
'thumbs_down': EffectInfo(
id='5104858069142078462',
name='Thumbs Down',
emoji='š',
description='Negative thumbs down',
category='reaction'
),
'poop': EffectInfo(
id='5046589136895476101',
name='Poop',
emoji='š©',
description='Funny poop effect',
category='humor'
),
'hearts_shower': EffectInfo(
id='5159385139981059251',
name='Hearts Shower',
emoji='ā¤ļøā¤ļøā¤ļø',
description='Shower of hearts',
category='emotion'
)
}
logger.info(f"Initialized {len(self.effects)} message effects")
def get_effect_id(self, effect_name: str) -\> Optional\[str\]:
"""Get effect ID by name."""
effect = self.effects.get(effect_name.lower())
return effect.id if effect else None
def get_effect_info_by_id(self, effect_id: str) -\> Optional\[Dict\[str, Any\]\]:
"""Get effect info by ID."""
for effect in self.effects.values():
if effect.id == effect_id:
return {
'name': effect.name,
'emoji': effect.emoji,
'description': effect.description,
'category': effect.category
}
return None
def get_random_effect(self) -\> str:
"""Get a random effect name."""
return random.choice(list(self.effects.keys()))
def get_all_effects(self) -\> List\[EffectInfo\]:
"""Get all available effects."""
return list(self.effects.values())
class TelegramEffectsBot:
"""Main bot class handling all Telegram interactions and message effects."""
def \__init_\_(self):
"""Initialize the bot with configuration and effects."""
self.config = Config()
self.effects = MessageEffects()
\# Initialize Pyrogram client
self.app = Client(
"single_file_bot",
api_id=int(self.config.API_ID),
api_hash=str(self.config.API_HASH),
bot_token=str(self.config.BOT_TOKEN),
workdir="sessions"
)
self.\_register_handlers()
def \_register_handlers(self):
"""Register all message handlers."""
@self.app.on_message(filters.command("start") & filters.private)
async def handle_start(client, message: Message):
"""Handle /start command with welcome message and effects."""
try:
logger.info(f"Received /start command from user {message.from_user.id}")
user_name = message.from_user.first_name or "Friend"
logger.info(f"Start command from user: {user_name} (@{message.from_user.username}, ID: {message.from_user.id})")
\# Send welcome message with random effect
welcome_text = f"""š \*\*Hi {user_name}!\*\*
š **Greetings!**
You've just discovered a bot with superpowers - I can add magical effects to my messages!"""
effect_name = self.effects.get_random_effect()
success = await self.send_message_with_effect(
message.chat.id,
welcome_text,
effect_name
)
if success:
logger.info(f"Welcome message sent successfully with {effect_name} effect")
\# Send follow-up tips with another effect
await asyncio.sleep(2)
tips_text = """š” \*\*Quick Tips:\*\*
ā¢ Effects work best in private chats
ā¢ Try `/effects` to see all available effects
ā¢ Use `/help` for more information
Ready to explore? š"""
tips_effect = self.effects.get_random_effect()
await self.send_message_with_effect(
message.chat.id,
tips_text,
tips_effect
)
except Exception as e:
logger.error(f"Error handling /start command: {e}")
await self.\_send_fallback_message(message.chat.id, "Welcome! Something went wrong, but I'm still here to help!")
@self.app.on_message(filters.command("help") & filters.private)
async def handle_help(client, message: Message):
"""Handle /help command."""
try:
help_text = """š¤ \*\*Effects Bot Help\*\*
**Available Commands:**
ā¢ `/start` - Get welcome message with effects
ā¢ `/help` - Show this help message
ā¢ `/effects` - Demo all available effects
**About Message Effects:**
This bot sends real Telegram animated message effects! When I send messages, you'll see beautiful animations like fire, confetti, hearts, and more.
**How it works:**
ā¢ Effects are built into Telegram
ā¢ You'll see animations in supported clients
ā¢ Works best in private chats
Ready to see some magic? Try `/effects`! āØ"""
await self.\_send_fallback_message(message.chat.id, help_text)
\# Send a follow-up with effect
await asyncio.sleep(1)
await self.send_message_with_effect(
message.chat.id,
"Hope this helps! š",
"party"
)
except Exception as e:
logger.error(f"Error handling /help command: {e}")
await self.\_send_fallback_message(message.chat.id, "Help information is temporarily unavailable.")
@self.app.on_message(filters.command("effects") & filters.private)
async def handle_effects(client, message: Message):
"""Handle /effects command to demonstrate all effects."""
try:
logger.info(f"Received /effects command from user {message.from_user.id}")
await self.demo_effects(message.chat.id)
except Exception as e:
logger.error(f"Error handling /effects command: {e}")
await self.\_send_fallback_message(message.chat.id, "Sorry, something went wrong! Please try again.")
\# Log all private messages for debugging
@self.app.on_message(filters.private)
async def handle_any_message(client, message: Message):
"""Log all private messages for debugging."""
try:
logger.info(f"Received message from user {message.from_user.id}: {message.text}")
except Exception as e:
logger.error(f"Error logging message: {e}")
async def send_message_with_effect(self, chat_id: int, text: str, effect_name: str = "fire", fallback: bool = True) -\> bool:
"""
Send a message with the specified effect.
Args:
chat_id: Target chat ID
text: Message text
effect_name: Name of the effect to use
fallback: Whether to send without effect if effect fails
Returns:
bool: True if message was sent successfully
"""
effect_id = self.effects.get_effect_id(effect_name)
if not effect_id:
logger.warning(f"Unknown effect: {effect_name}")
if fallback:
return await self.\_send_fallback_message(chat_id, text)
return False
try:
logger.info(f"Sending message with {effect_name} effect using raw API...")
return await self.\_send_with_real_effects(chat_id, text, effect_id, fallback)
except Exception as e:
logger.error(f"Error sending message with effect: {e}")
if fallback:
return await self.\_send_fallback_message(chat_id, text)
return False
async def \_send_with_real_effects(self, chat_id: int, text: str, effect_id: str, fallback: bool = True) -\> bool:
"""Send message with real Telegram message effects using HTTP API."""
try:
\# Get effect info
effect_info = self.effects.get_effect_info_by_id(effect_id)
effect_name = effect_info\['name'\] if effect_info else "Unknown"
\# Prepare the API request for real message effects
url = f"https://api.telegram.org/bot{self.config.BOT_TOKEN}/sendMessage"
payload = {
'chat_id': chat_id,
'text': text,
'message_effect_id': effect_id
}
\# Try with aiohttp first
try:
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload) as response:
result = await response.json()
if result.get('ok'):
logger.info(f"ā
Sent real message effect: {effect_name}")
return True
else:
error_msg = result.get('description', 'Unknown error')
logger.warning(f"ā ļø Message effect failed: {error_msg}")
if fallback:
return await self.\_send_fallback_message(chat_id, text)
return False
except Exception as aiohttp_error:
logger.warning(f"aiohttp failed, trying requests: {aiohttp_error}")
\# Fallback to requests
response = requests.post(url, json=payload, timeout=10)
result = response.json()
if result.get('ok'):
logger.info(f"ā
Sent real message effect via requests: {effect_name}")
return True
else:
error_msg = result.get('description', 'Unknown error')
logger.warning(f"ā ļø Message effect failed via requests: {error_msg}")
if fallback:
return await self.\_send_fallback_message(chat_id, text)
return False
except Exception as e:
logger.error(f"Error sending message with real effects: {e}")
if fallback:
return await self.\_send_fallback_message(chat_id, text)
return False
async def \_send_fallback_message(self, chat_id: int, text: str) -\> bool:
"""Send message without effects as fallback."""
try:
await self.app.send_message(
chat_id=chat_id,
text=text,
parse_mode=ParseMode.MARKDOWN
)
logger.info("š¤ Sent fallback message without effects")
return True
except Exception as e:
logger.error(f"Error sending fallback message: {e}")
return False
async def demo_effects(self, chat_id: int):
"""Demonstrate all available message effects."""
try:
\# Send intro message
intro_text = """šØ \*\*Message Effects Demo\*\*
Watch as I demonstrate all available effects! Each message will have a different animation:"""
await self.\_send_fallback_message(chat_id, intro_text)
\# Demo each effect with a small delay
for effect_name, effect_info in self.effects.effects.items():
await asyncio.sleep(1.5) # Pause between effects
demo_text = f"{effect_info.emoji} Testing {effect_info.name} effect!"
await self.send_message_with_effect(chat_id, demo_text, effect_name)
\# Send conclusion
await asyncio.sleep(2)
conclusion_text = """āØ \*\*Demo Complete!\*\*
Pretty cool, right? These are real Telegram message effects that work in all supported clients.
Try `/start` to experience a random effect, or just send me any message to chat!"""
await self.\_send_fallback_message(chat_id, conclusion_text)
except Exception as e:
logger.error(f"Error in demo_effects: {e}")
await self.\_send_fallback_message(chat_id, "Demo failed, but the bot is still working!")
async def start(self):
"""Start the bot."""
try:
await self.app.start()
logger.info("š Bot started successfully!")
\# Get bot info
bot_info = await self.app.get_me()
logger.info(f"Bot info: @{bot_info.username} ({bot_info.first_name})")
print(f"Bot @{bot_info.username} is now running! Send /start in private chat to test effects.")
\# Keep the bot running
await asyncio.Event().wait()
except Exception as e:
logger.error(f"Error starting bot: {e}")
raise
async def stop(self):
"""Stop the bot."""
try:
await self.app.stop()
logger.info("š Bot stopped successfully!")
except Exception as e:
logger.error(f"Error stopping bot: {e}")
async def main():
"""Main function to run the bot."""
bot = TelegramEffectsBot()
try:
await bot.start()
except KeyboardInterrupt:
logger.info("š Received interrupt signal, stopping bot...")
await bot.stop()
except Exception as e:
logger.error(f"ā Fatal error: {e}")
raise
if _name_ == "_main_":
print("š¤ Starting Telegram Effects Bot...")
print("š Make sure you have set BOT_TOKEN, API_ID, and API_HASH environment variables")
print("š„ This bot sends real animated message effects!")
print()
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\\nš Bot stopped by user")
except Exception as e:
print(f"\\nā Error: {e}")
exit(1)