Initial commit: Telegram Sticker Bot Manager
Features: - Create and manage sticker packs - Add existing sticker sets to database - Automatic image optimization (512px, WebP conversion) - GIF to animated video sticker conversion (WebM, VP9) - Background removal (local rembg + optional Remove.bg API) - Hybrid API with retry logic and usage tracking - Premium status detection - SQLite database for persistence - Docker containerization - Complete OPENCODE.md documentation Technical Stack: - python-telegram-bot 22.7 - Pillow for image processing - FFmpeg for video conversion - rembg for local background removal - aiosqlite for async database operations
This commit is contained in:
coding
310
sticker_manager.py
Normal file
310
sticker_manager.py
Normal file
@@ -0,0 +1,310 @@
|
||||
"""Sticker management utilities for Telegram bot."""
|
||||
import logging
|
||||
from typing import List, Optional, Tuple, Dict, Any
|
||||
from telegram import InlineKeyboardMarkup, InlineKeyboardButton, Update
|
||||
from telegram.ext import ContextTypes
|
||||
from telegram.constants import StickerFormat
|
||||
|
||||
from database import db
|
||||
from image_processor import process_image_for_sticker, validate_image
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def check_owner(update: Update) -> bool:
|
||||
"""Check if user is the bot owner."""
|
||||
from config import OWNER_ID
|
||||
user_id = update.effective_user.id
|
||||
return user_id == OWNER_ID
|
||||
|
||||
|
||||
async def check_premium_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool:
|
||||
"""Check if user has Telegram Premium."""
|
||||
user_id = update.effective_user.id
|
||||
|
||||
# Get current user info from Telegram
|
||||
try:
|
||||
chat = await context.bot.get_chat(user_id)
|
||||
is_premium = bool(getattr(chat, 'is_premium', False))
|
||||
|
||||
# Update database
|
||||
await db.update_premium_status(user_id, is_premium)
|
||||
|
||||
return is_premium
|
||||
except Exception as e:
|
||||
logger.error(f"Error checking premium status: {e}")
|
||||
# Fallback to cached status
|
||||
cached = await db.get_premium_status(user_id)
|
||||
return cached if cached is not None else False
|
||||
|
||||
|
||||
def get_main_menu_keyboard() -> InlineKeyboardMarkup:
|
||||
"""Generate main menu keyboard."""
|
||||
keyboard = [
|
||||
[InlineKeyboardButton("➕ Neues Stickerpaket", callback_data="create_set")],
|
||||
[InlineKeyboardButton("📥 Existierendes hinzufügen", callback_data="add_existing")],
|
||||
[InlineKeyboardButton("📦 Meine Stickerpakete", callback_data="list_sets")],
|
||||
[InlineKeyboardButton("❓ Hilfe", callback_data="help")],
|
||||
]
|
||||
return InlineKeyboardMarkup(keyboard)
|
||||
|
||||
|
||||
def get_set_action_keyboard(set_name: str) -> InlineKeyboardMarkup:
|
||||
"""Generate keyboard for set actions."""
|
||||
keyboard = [
|
||||
[InlineKeyboardButton("➕ Sticker hinzufügen", callback_data=f"add_sticker:{set_name}")],
|
||||
[InlineKeyboardButton("🗑 Sticker löschen", callback_data=f"delete_sticker:{set_name}")],
|
||||
[InlineKeyboardButton("🏷 Emoji ändern", callback_data=f"change_emoji:{set_name}")],
|
||||
[InlineKeyboardButton("✏️ Paket umbenennen", callback_data=f"rename_set:{set_name}")],
|
||||
[InlineKeyboardButton("🗑 Paket löschen", callback_data=f"delete_set:{set_name}")],
|
||||
[InlineKeyboardButton("⬅️ Zurück", callback_data="back_to_menu")],
|
||||
]
|
||||
return InlineKeyboardMarkup(keyboard)
|
||||
|
||||
|
||||
async def get_user_sets_keyboard(user_id: int) -> Tuple[Optional[InlineKeyboardMarkup], str]:
|
||||
"""Generate keyboard with user's sticker sets."""
|
||||
sets = await db.get_user_sticker_sets(user_id)
|
||||
|
||||
if not sets:
|
||||
return None, "📭 Du hast noch keine Stickerpakete."
|
||||
|
||||
keyboard = []
|
||||
for s in sets:
|
||||
display_name = s['set_title'][:30] if s['set_title'] else s['set_name']
|
||||
keyboard.append([InlineKeyboardButton(
|
||||
f"📦 {display_name}",
|
||||
callback_data=f"select_set:{s['set_name']}"
|
||||
)])
|
||||
|
||||
keyboard.append([InlineKeyboardButton("⬅️ Zurück", callback_data="back_to_menu")])
|
||||
|
||||
text = f"📦 **Deine Stickerpakete** ({len(sets)} total)\\n\\nWähle ein Paket:"
|
||||
return InlineKeyboardMarkup(keyboard), text
|
||||
|
||||
|
||||
async def create_sticker_set(
|
||||
update: Update,
|
||||
context: ContextTypes.DEFAULT_TYPE,
|
||||
name: str,
|
||||
title: str,
|
||||
sticker_file,
|
||||
emoji: str = "🤖"
|
||||
) -> Tuple[bool, str]:
|
||||
"""Create a new sticker set with automatic image processing."""
|
||||
try:
|
||||
user_id = update.effective_user.id
|
||||
bot_username = context.bot.username
|
||||
set_name = f"{name}_by_{bot_username}"
|
||||
|
||||
# Download the file
|
||||
file_obj = await sticker_file.download_as_bytearray()
|
||||
file_bytes = bytes(file_obj)
|
||||
|
||||
# Validate image
|
||||
is_valid, error_msg = await validate_image(file_bytes)
|
||||
if not is_valid:
|
||||
return False, f"Invalid image: {error_msg}"
|
||||
|
||||
# Process image for sticker format
|
||||
processed_image = await process_image_for_sticker(file_bytes)
|
||||
if not processed_image:
|
||||
return False, "Failed to process image. Please ensure it's a valid image file (PNG, JPG, WebP) and under 10MB."
|
||||
|
||||
# Create sticker input with processed image
|
||||
from telegram import InputSticker
|
||||
sticker = InputSticker(
|
||||
sticker=processed_image,
|
||||
format=StickerFormat.STATIC,
|
||||
emoji_list=[emoji]
|
||||
)
|
||||
|
||||
# Create sticker set
|
||||
await context.bot.create_new_sticker_set(
|
||||
user_id=user_id,
|
||||
name=set_name,
|
||||
title=title,
|
||||
stickers=[sticker]
|
||||
)
|
||||
|
||||
# Track in database
|
||||
await db.add_sticker_set(user_id, set_name, title)
|
||||
|
||||
return True, set_name
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating sticker set: {e}")
|
||||
return False, str(e)
|
||||
|
||||
|
||||
async def add_sticker_to_set(
|
||||
update: Update,
|
||||
context: ContextTypes.DEFAULT_TYPE,
|
||||
set_name: str,
|
||||
sticker_file,
|
||||
emoji: str = "🤖"
|
||||
) -> bool:
|
||||
"""Add a sticker to existing set with automatic image processing."""
|
||||
try:
|
||||
# Download and process the file
|
||||
file_obj = await sticker_file.download_as_bytearray()
|
||||
file_bytes = bytes(file_obj)
|
||||
|
||||
# Validate image
|
||||
is_valid, error_msg = await validate_image(file_bytes)
|
||||
if not is_valid:
|
||||
logger.error(f"Invalid image for sticker: {error_msg}")
|
||||
return False
|
||||
|
||||
# Process image for sticker format
|
||||
processed_image = await process_image_for_sticker(file_bytes)
|
||||
if not processed_image:
|
||||
logger.error("Failed to process image for sticker")
|
||||
return False
|
||||
|
||||
from telegram import InputSticker
|
||||
sticker = InputSticker(
|
||||
sticker=processed_image,
|
||||
format=StickerFormat.STATIC,
|
||||
emoji_list=[emoji]
|
||||
)
|
||||
|
||||
await context.bot.add_sticker_to_set(
|
||||
user_id=update.effective_user.id,
|
||||
name=set_name,
|
||||
sticker=sticker
|
||||
)
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Error adding sticker: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def add_video_sticker_to_set(
|
||||
update: Update,
|
||||
context: ContextTypes.DEFAULT_TYPE,
|
||||
set_name: str,
|
||||
webm_bytes: bytes,
|
||||
emoji: str = "🤖"
|
||||
) -> bool:
|
||||
"""Add a video sticker (WebM) to existing set."""
|
||||
try:
|
||||
from telegram import InputSticker
|
||||
sticker = InputSticker(
|
||||
sticker=webm_bytes,
|
||||
format=StickerFormat.VIDEO,
|
||||
emoji_list=[emoji]
|
||||
)
|
||||
|
||||
await context.bot.add_sticker_to_set(
|
||||
user_id=update.effective_user.id,
|
||||
name=set_name,
|
||||
sticker=sticker
|
||||
)
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Error adding video sticker: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def delete_sticker_from_set(update: Update, context: ContextTypes.DEFAULT_TYPE,
|
||||
sticker_file_id: str) -> bool:
|
||||
"""Delete a sticker from a set."""
|
||||
try:
|
||||
await context.bot.delete_sticker_from_set(sticker_file_id)
|
||||
await db.delete_sticker(sticker_file_id)
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting sticker: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def delete_sticker_set(update: Update, context: ContextTypes.DEFAULT_TYPE,
|
||||
set_name: str) -> bool:
|
||||
"""Delete a sticker set."""
|
||||
try:
|
||||
await context.bot.delete_sticker_set(set_name)
|
||||
await db.delete_sticker_set(set_name)
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting sticker set: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def rename_sticker_set(update: Update, context: ContextTypes.DEFAULT_TYPE,
|
||||
set_name: str, new_title: str) -> bool:
|
||||
"""Rename a sticker set title."""
|
||||
try:
|
||||
await context.bot.set_sticker_set_title(set_name, new_title)
|
||||
await db.rename_sticker_set(set_name, new_title)
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Error renaming sticker set: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def change_sticker_emoji(update: Update, context: ContextTypes.DEFAULT_TYPE,
|
||||
sticker_file_id: str, new_emoji: str) -> bool:
|
||||
"""Change emoji of a sticker."""
|
||||
try:
|
||||
await context.bot.set_sticker_emoji_list(sticker_file_id, [new_emoji])
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Error changing sticker emoji: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def add_existing_sticker_set(
|
||||
update: Update,
|
||||
context: ContextTypes.DEFAULT_TYPE,
|
||||
set_name: str
|
||||
) -> Tuple[bool, str]:
|
||||
"""
|
||||
Add an existing sticker set to the database.
|
||||
|
||||
Args:
|
||||
set_name: Name of the sticker set (e.g., "my_stickers_by_mybot")
|
||||
|
||||
Returns:
|
||||
Tuple of (success, message)
|
||||
"""
|
||||
try:
|
||||
# Try to get sticker set from Telegram
|
||||
sticker_set = await context.bot.get_sticker_set(set_name)
|
||||
|
||||
if not sticker_set:
|
||||
return False, "Stickerpaket nicht gefunden."
|
||||
|
||||
# Get set info
|
||||
title = sticker_set.title
|
||||
user_id = update.effective_user.id
|
||||
|
||||
# Check if already in database
|
||||
existing_sets = await db.get_user_sticker_sets(user_id)
|
||||
if any(s['set_name'] == set_name for s in existing_sets):
|
||||
return False, "Dieses Stickerpaket ist bereits in deiner Liste."
|
||||
|
||||
# Add to database
|
||||
await db.add_sticker_set(user_id, set_name, title)
|
||||
|
||||
logger.info(f"Added existing sticker set '{set_name}' to database for user {user_id}")
|
||||
return True, f"Stickerpaket '{title}' wurde hinzugefügt!"
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error adding existing sticker set: {e}")
|
||||
return False, f"Fehler: {str(e)}\n\nPrüfe, ob der Name korrekt ist und du Zugriff auf das Paket hast."
|
||||
|
||||
|
||||
async def schedule_premium_check(context: ContextTypes.DEFAULT_TYPE):
|
||||
"""Scheduled job to check premium status periodically."""
|
||||
from config import OWNER_ID, logger
|
||||
try:
|
||||
# Check owner's premium status
|
||||
chat = await context.bot.get_chat(OWNER_ID)
|
||||
is_premium = bool(getattr(chat, 'is_premium', False))
|
||||
await db.update_premium_status(OWNER_ID, is_premium)
|
||||
|
||||
logger.info(f"Premium check completed. Owner premium status: {is_premium}")
|
||||
except Exception as e:
|
||||
logger.error(f"Error in scheduled premium check: {e}")
|
||||
Reference in New Issue
Block a user