From f634b884eca2bde1f831a2f0c8df0320bd987111 Mon Sep 17 00:00:00 2001 From: coding Date: Wed, 25 Mar 2026 08:59:38 +0000 Subject: [PATCH] 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 --- .env.example | 7 + Dockerfile | 31 ++ OPENCODE.md | 346 ++++++++++++ background_remover.py | 183 +++++++ bot.py | 1158 +++++++++++++++++++++++++++++++++++++++++ config.py | 38 ++ database.py | 174 +++++++ docker-compose.yml | 19 + gif_processor.py | 267 ++++++++++ image_processor.py | 169 ++++++ requirements.txt | 6 + sticker_manager.py | 310 +++++++++++ 12 files changed, 2708 insertions(+) create mode 100644 .env.example create mode 100644 Dockerfile create mode 100644 OPENCODE.md create mode 100644 background_remover.py create mode 100644 bot.py create mode 100644 config.py create mode 100644 database.py create mode 100644 docker-compose.yml create mode 100644 gif_processor.py create mode 100644 image_processor.py create mode 100644 requirements.txt create mode 100644 sticker_manager.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..3247b3f --- /dev/null +++ b/.env.example @@ -0,0 +1,7 @@ +# Telegram Sticker Bot Configuration +BOT_TOKEN=your_bot_token_here +OWNER_ID=your_telegram_user_id_here + +# Optional: Remove.bg API Key for background removal +# If not set, only local processing is available (50 free calls/month) +REMOVE_BG_API_KEY=your_remove_bg_api_key_here \ No newline at end of file diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..8137bf2 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,31 @@ +FROM python:3.11-alpine + +# Set environment variables +ENV PYTHONDONTWRITEBYTECODE=1 +ENV PYTHONUNBUFFERED=1 + +# Install system dependencies +RUN apk add --no-cache gcc musl-dev ffmpeg + +# Set work directory +WORKDIR /app + +# Copy requirements first for better caching +COPY requirements.txt . + +# Install Python dependencies +RUN pip install --no-cache-dir -r requirements.txt + +# Copy project +COPY . . + +# Create non-root user +RUN adduser -D -u 1000 appuser && chown -R appuser:appuser /app +USER appuser + +# Health check +HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ + CMD python -c "import sys; sys.exit(0)" + +# Run the bot +CMD ["python", "bot.py"] \ No newline at end of file diff --git a/OPENCODE.md b/OPENCODE.md new file mode 100644 index 0000000..27c58a8 --- /dev/null +++ b/OPENCODE.md @@ -0,0 +1,346 @@ +# OPENCODE.md - Telegram Sticker Bot + +## Overview +This is a Telegram bot for managing sticker packs. Built with python-telegram-bot v22.7, it allows the bot owner to create, manage, and organize sticker sets through an interactive conversation interface. + +## Architecture + +### Technology Stack +- **Python**: 3.11+ +- **Library**: python-telegram-bot v22.7 (supports Bot API 9.5) +- **Database**: SQLite with aiosqlite for async operations +- **Image Processing**: Pillow (PIL) for automatic image conversion +- **Video Processing**: FFmpeg for GIF to WebM conversion +- **Background Removal**: rembg (local) + remove.bg API (optional) +- **Container**: Docker + Docker Compose +- **Environment**: python-dotenv for configuration + +### Project Structure +``` +stickerbot/ +├── bot.py # Main bot with ConversationHandler +├── config.py # Configuration and environment loading +├── database.py # SQLite database operations +├── sticker_manager.py # Sticker management utilities +├── image_processor.py # Automatic image conversion +├── gif_processor.py # GIF to WebM conversion +├── background_remover.py # Background removal (local + API) +├── requirements.txt # Python dependencies +├── Dockerfile # Docker image configuration +├── docker-compose.yml # Docker Compose configuration +├── .env.example # Environment template +└── data/ # Persistent data directory +``` + +### Database Schema +- **sticker_sets**: Tracks created sticker sets (user_id, set_name, set_title, timestamps) +- **user_premium**: Caches user premium status with last check timestamp +- **stickers**: Tracks individual stickers in sets (file_id, emoji, set_name) +- **api_usage**: Tracks API usage for external services (remove.bg) + +## Commands + +### Available Commands +- `/start` - Start the bot and show main menu +- `/hilfe` - Display detailed help message +- `/cancel` - Cancel current conversation + +## Build/Lint/Test Commands + +### Docker +```bash +# Build and run +sudo docker-compose up -d --build + +# View logs +sudo docker-compose logs -f + +# Stop +sudo docker-compose down + +# Rebuild after changes +sudo docker-compose up -d --build --force-recreate +``` + +### Local Development (without Docker) +```bash +# Install dependencies +pip install -r requirements.txt + +# Run bot +python bot.py + +# Run with environment file +python -c "from bot import main; main()" +``` + +### Testing +- No test suite implemented yet +- Manual testing via Telegram bot interface +- Use `/cancel` to abort any conversation + +### Linting +```bash +# Optional: Add ruff or flake8 +pip install ruff +ruff check . +ruff format . +``` + +## Code Style Guidelines + +### Imports +- Group imports: stdlib, third-party, local +- Use absolute imports +- Example: +```python +import logging +from telegram import Update +from telegram.ext import Application +from config import BOT_TOKEN +``` + +### Async/Await +- All handlers are async (python-telegram-bot v22.x requirement) +- Database operations use aiosqlite (async) +- Always await coroutines +- Example: +```python +async def handler(update: Update, context: ContextTypes.DEFAULT_TYPE): + await update.message.reply_text("Hello") +``` + +### Types +- Use type hints where possible +- Follow telegram library types: Update, ContextTypes, InlineKeyboardMarkup +- Database functions return Optional[bool], List[Dict[str, Any]] + +### Naming Conventions +- Functions: snake_case +- Constants: UPPER_CASE +- Classes: PascalCase +- Private: _prefix + +### Error Handling +- Use try-except blocks for API calls +- Log errors with logger.error() +- Return success/failure tuples or booleans +- Always clean up user_data on errors + +### Conversation States +- Defined as constants at module level +- Use range() for sequential states +- Clear user_data when conversation ends +- State order: MENU → CREATE_SET_* → SELECT_SET → SET_ACTION → specific actions + +### Security +- OWNER_ID check on every /start +- No group chat support (private only) +- SQLite for local persistence +- No secrets in code (use .env) + +## Key Features + +### Premium Detection +- Check via `chat.is_premium` from Telegram API +- Cached in database with 24h refresh interval +- Scheduled job using JobQueue +- Displayed in main menu + +### Sticker Operations + +#### 1. Create New Sticker Set +**Flow:** Name → Title → Emoji → Image +- 4-step wizard with input validation +- Automatic image optimization +- Optional background removal + +#### 2. Add Existing Sticker Set +**Flow:** Select "📥 Existierendes hinzufügen" → Enter set name +- User provides existing sticker set name +- Bot verifies via `getStickerSet()` API +- Auto-appends `_by_` if not present +- Duplicate check prevents double entries + +#### 3. Add Sticker to Set +**Supported formats:** +- **Static images**: PNG, JPG, WebP (automatic conversion to WebP) +- **GIFs**: Converted to animated video stickers (WebM, max 3 seconds) + +**Features:** +- Automatic resizing to 512px +- Background removal option (if REMOVE_BG_API_KEY configured) +- GIF trimming for animations > 3 seconds + +#### 4. Delete Sticker +- Forward sticker to bot +- Confirmation before deletion + +#### 5. Change Emoji +- Forward sticker → Enter new emoji + +#### 6. Rename Set +- Enter new title (max 64 characters) + +#### 7. Delete Set +- Confirmation required +- Cannot be undone + +### Automatic Image Processing +All uploaded images are automatically converted to Telegram sticker format: +- **Resize**: One dimension set to 512px (other ≤512px), maintaining aspect ratio +- **Format conversion**: Converted to WebP with transparency (RGBA) +- **Compression**: Quality adjusted automatically (60-95) to stay under 3MB +- **Supported formats**: PNG, JPG, JPEG, WebP, GIF, BMP, TIFF +- **Quality**: Uses LANCZOS resampling for best results + +### Background Removal (Hybrid) + +#### Local Processing (rembg) +- **Cost**: Free, runs on server +- **Method**: U²Net deep learning model +- **First run**: Downloads model (~180MB) +- **Speed**: Depends on image size + +#### Remove.bg API (Optional) +- **Cost**: 50 free calls/month, then paid +- **Configuration**: Set REMOVE_BG_API_KEY in .env +- **Retry logic**: 3 attempts with delays (2s, 5s, 10s) +- **Limit tracking**: Stored in database, resets monthly +- **Fallback**: Automatically falls back to local on API errors + +#### User Interface +``` +Without API key: +Image → [Auto: Local processing] → Emoji → Done + +With API key: +Image → Selection: + ├─ [🤖 Local (free)] + ├─ [☁️ API (45/50)] or [☁️ API (50/50) ⚠️ Limit] + └─ [⏭️ Skip removal] + +If API fails/limit reached: + ❌ Remove.bg API limit reached (50/50) + + [🤖 Process locally instead] + [⏭️ Without background removal] + [❌ Cancel] +``` + +### GIF to Video Sticker Conversion +- **Format**: WebM with VP9 codec +- **Resolution**: 512px (one side), other ≤512px +- **FPS**: 30 frames per second +- **Duration**: Maximum 3 seconds +- **File size**: Maximum 256KB +- **Compression**: Automatic quality adjustment +- **Trimming**: User confirmation for GIFs > 3s + +## Environment Variables + +```bash +# Required +BOT_TOKEN=your_telegram_bot_token +OWNER_ID=your_telegram_user_id + +# Optional +DATABASE_PATH=/app/data/sticker_bot.db +PREMIUM_CHECK_INTERVAL=24 # Hours, default: 24 + +# Optional: Remove.bg API for background removal +# 50 free calls/month +REMOVE_BG_API_KEY=your_remove_bg_api_key +``` + +## Useful Commands + +```bash +# Check bot is running +sudo docker ps | grep stickerbot + +# View logs +sudo docker-compose logs -f + +# Database inspection +sqlite3 data/sticker_bot.db ".tables" +sqlite3 data/sticker_bot.db "SELECT * FROM sticker_sets;" +sqlite3 data/sticker_bot.db "SELECT * FROM api_usage;" + +# Reset database +rm data/sticker_bot.db + +# Rebuild container +sudo docker-compose up -d --build --force-recreate +``` + +## Important Notes + +### Sticker Requirements +- **Static stickers**: 512x512px, WebP format, max 3MB +- **Animated stickers**: WebM (VP9), 512px one side, max 3s, max 256KB +- **Emoji**: One per sticker (can be changed later) + +### Set Names +- Must end with `_by_` (handled automatically) +- 1-64 characters +- Only alphanumeric and underscores for short name + +### Limits +- **Regular sets**: Max 120 stickers +- **Emoji sets**: Max 200 stickers +- **API calls**: 50/month for remove.bg (free tier) + +### Best Practices +- Send images as file (not compressed photo) for best quality +- Use /start to return to main menu anytime +- Use /cancel to abort current operation +- Database persists in `./data/` volume + +## API Methods Used + +### Telegram Bot API +- `create_new_sticker_set()` - Create sticker sets +- `add_sticker_to_set()` - Add stickers +- `delete_sticker_from_set()` - Remove stickers +- `set_sticker_emoji_list()` - Change emoji +- `set_sticker_set_title()` - Rename set +- `delete_sticker_set()` - Delete entire set +- `get_sticker_set()` - Retrieve set info (for existing sets) +- `get_chat()` - Check premium status + +### External APIs +- **Remove.bg API**: `POST https://api.remove.bg/v1.0/removebg` - Background removal + +## Common Issues + +1. **Import errors in IDE**: Install requirements.txt (`pip install -r requirements.txt`) +2. **Database locked**: Only one process should access SQLite at a time +3. **Sticker creation fails**: Check image format (must be valid image file) +4. **Premium check fails**: User must have started bot first +5. **API limit reached**: Check current usage with `/hilfe` or in database +6. **Background removal slow**: First run downloads ML model (~180MB) +7. **GIF conversion fails**: Ensure GIF is under 10MB before upload + +## Troubleshooting + +### Bot not responding +```bash +# Check if container is running +sudo docker ps | grep stickerbot + +# Check logs for errors +sudo docker-compose logs -f +``` + +### Database issues +```bash +# Reset database (removes all data) +rm data/sticker_bot.db +sudo docker-compose restart +``` + +### API issues +- Verify REMOVE_BG_API_KEY is correct +- Check API usage in database: `sqlite3 data/sticker_bot.db "SELECT * FROM api_usage;"` +- API automatically falls back to local processing on errors diff --git a/background_remover.py b/background_remover.py new file mode 100644 index 0000000..23fea90 --- /dev/null +++ b/background_remover.py @@ -0,0 +1,183 @@ +"""Background removal module supporting local (rembg) and API (remove.bg) methods.""" +import io +import logging +import time +from typing import Optional, Tuple, Union +import requests +from PIL import Image + +logger = logging.getLogger(__name__) + +# Remove.bg API endpoint +REMOVE_BG_API_URL = "https://api.remove.bg/v1.0/removebg" +MAX_API_CALLS_PER_MONTH = 50 + +# Retry configuration +MAX_RETRIES = 3 +RETRY_DELAYS = [2, 5, 10] # seconds + + +async def remove_background_local(image_bytes: bytes) -> Optional[bytes]: + """ + Remove background using local rembg library (U²Net model). + + Args: + image_bytes: Raw image bytes + + Returns: + PNG image bytes with transparent background, or None if failed + """ + try: + from rembg import remove + + # Open image + input_image = Image.open(io.BytesIO(image_bytes)) + + # Remove background + output_image = remove(input_image) + + # Save to bytes + output_buffer = io.BytesIO() + output_image.save(output_buffer, format='PNG') + output_buffer.seek(0) + + logger.info("Background removed successfully (local)") + return output_buffer.getvalue() + + except Exception as e: + logger.error(f"Error removing background locally: {e}") + return None + + +async def remove_background_api( + image_bytes: bytes, + api_key: str +) -> Tuple[Union[bytes, str], str]: + """ + Remove background using remove.bg API with retry logic. + + Args: + image_bytes: Raw image bytes + api_key: Remove.bg API key + + Returns: + Tuple of (result, message) + - Success: (image_bytes, "success") + - Rate limit: ("LIMIT_REACHED", error_message) + - Other error: ("API_ERROR", error_message) + """ + headers = { + "X-Api-Key": api_key + } + + last_error = None + + # Retry loop + for attempt in range(MAX_RETRIES): + try: + # Prepare multipart form data + files = { + 'image_file': ('image.png', io.BytesIO(image_bytes), 'image/png') + } + + response = requests.post( + REMOVE_BG_API_URL, + headers=headers, + files=files, + timeout=30 + ) + + # Check response + if response.status_code == 200: + # Success - return image + logger.info("Background removed successfully (API)") + return response.content, "success" + + elif response.status_code in (402, 429): + # Rate limit / Payment required (limit reached) + error_data = response.json() + error_msg = error_data.get('errors', [{}])[0].get('title', 'Rate limit exceeded') + logger.warning(f"Remove.bg API limit reached: {error_msg}") + return "LIMIT_REACHED", error_msg + + else: + # Other API error + error_data = response.json() if response.text else {} + error_msg = error_data.get('errors', [{}])[0].get('title', f'HTTP {response.status_code}') + logger.error(f"Remove.bg API error: {error_msg}") + last_error = error_msg + + except requests.exceptions.Timeout: + last_error = "Timeout - API antwortet nicht" + logger.warning(f"API timeout (attempt {attempt + 1}/{MAX_RETRIES})") + + except requests.exceptions.ConnectionError: + last_error = "Verbindungsfehler - API nicht erreichbar" + logger.warning(f"API connection error (attempt {attempt + 1}/{MAX_RETRIES})") + + except Exception as e: + last_error = str(e) + logger.error(f"API call error (attempt {attempt + 1}/{MAX_RETRIES}): {e}") + + # Wait before retry (except for last attempt) + if attempt < MAX_RETRIES - 1: + delay = RETRY_DELAYS[attempt] + logger.info(f"Retrying in {delay} seconds...") + time.sleep(delay) + + # All retries failed + logger.error(f"All API retries failed: {last_error}") + return "API_ERROR", f"API nicht erreichbar nach {MAX_RETRIES} Versuchen: {last_error}" + + +async def should_show_api_limit_warning(api_key: str, current_usage: int) -> bool: + """ + Check if we should show warning on the API button. + + Args: + api_key: API key (if None, no warning) + current_usage: Current month usage count + + Returns: + True if limit reached or very close + """ + if not api_key: + return False + + # Show warning when at or above limit + return current_usage >= MAX_API_CALLS_PER_MONTH + + +async def validate_image_for_bg_removal(image_bytes: bytes) -> Tuple[bool, str]: + """ + Validate if image is suitable for background removal. + + Args: + image_bytes: Raw image bytes + + Returns: + Tuple of (is_valid, error_message) + """ + try: + image = Image.open(io.BytesIO(image_bytes)) + + # Check format + if image.format not in ('PNG', 'JPEG', 'JPG', 'WEBP', 'BMP'): + return False, f"Format '{image.format}' nicht unterstützt. Nutze PNG, JPG oder WebP." + + # Check size + width, height = image.size + if width < 64 or height < 64: + return False, "Bild zu klein (min. 64x64 Pixel)" + + if width > 4096 or height > 4096: + return False, "Bild zu groß (max. 4096x4096 Pixel)" + + # Check file size + if len(image_bytes) > 10 * 1024 * 1024: # 10MB + return False, "Datei zu groß (max. 10MB)" + + return True, "" + + except Exception as e: + return False, f"Ungültiges Bild: {str(e)}" diff --git a/bot.py b/bot.py new file mode 100644 index 0000000..b92f15f --- /dev/null +++ b/bot.py @@ -0,0 +1,1158 @@ +"""Telegram bot for managing sticker packs.""" +import logging +from telegram import Update, InlineKeyboardMarkup, InlineKeyboardButton +from telegram.ext import ( + Application, + CommandHandler, + MessageHandler, + CallbackQueryHandler, + ConversationHandler, + ContextTypes, + filters, +) + +from config import BOT_TOKEN, OWNER_ID, PREMIUM_CHECK_INTERVAL, REMOVE_BG_API_KEY, logger +from database import db +from sticker_manager import ( + check_owner, + check_premium_status, + get_main_menu_keyboard, + get_set_action_keyboard, + get_user_sets_keyboard, + create_sticker_set, + add_sticker_to_set, + add_video_sticker_to_set, + delete_sticker_from_set, + delete_sticker_set, + rename_sticker_set, + change_sticker_emoji, + schedule_premium_check, + add_existing_sticker_set, +) +from gif_processor import ( + get_gif_info, + validate_gif_duration, + convert_gif_to_webm, +) +from background_remover import ( + remove_background_local, + remove_background_api, + should_show_api_limit_warning, + validate_image_for_bg_removal, + MAX_API_CALLS_PER_MONTH, +) + +# Conversation states +(MENU, CREATE_SET_NAME, CREATE_SET_TITLE, CREATE_SET_EMOJI, CREATE_SET_STICKER, +SELECT_SET, SET_ACTION, ADD_STICKER, ADD_STICKER_EMOJI, ADD_ANIMATION_EMOJI, +DELETE_STICKER, CHANGE_EMOJI, RENAME_SET, DELETE_CONFIRM, +GIF_DURATION_CONFIRM, GIF_CONVERTING, REMOVE_BG_CHOICE, REMOVE_BG_API_ERROR, +ADD_EXISTING_SET_NAME) = range(19) + +# Data keys +CURRENT_SET = "current_set" +SET_NAME = "set_name" +SET_TITLE = "set_title" +STICKER_TO_CHANGE = "sticker_to_change" +PENDING_ANIMATION = "pending_animation" + + +async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Send welcome message and show main menu.""" + if not await check_owner(update): + await update.message.reply_text("⛔️ Dieser Bot ist nur für den Besitzer verfügbar.") + return ConversationHandler.END + + # Check premium status on start + is_premium = await check_premium_status(update, context) + premium_text = "💎 Premium aktiv" if is_premium else "👤 Standard" + + welcome_text = ( + f"🎨 **Willkommen beim Sticker Bot Manager!**\n\n" + f"Status: {premium_text}\n\n" + "Mit diesem Bot kannst du deine Telegram Stickerpakete verwalten:\n\n" + "✅ **Neue Pakete erstellen**\n" + "✅ **Sticker hinzufügen** (Bilder & GIFs)\n" + "✅ **Sticker löschen**\n" + "✅ **Emoji ändern**\n" + "✅ **Pakete umbenennen**\n" + "✅ **Pakete löschen**\n\n" + "🎨 **Extras:**\n" + "• Automatische Bildoptimierung\n" + "• GIF → animierte Video-Sticker\n" + "• Hintergrundentfernung (optional)\n\n" + "Nutze /hilfe für Details.\n\n" + "Wähle eine Aktion:" + ) + + await update.message.reply_text( + welcome_text, + reply_markup=get_main_menu_keyboard(), + parse_mode="Markdown" + ) + return MENU + + +async def menu_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Handle menu callbacks.""" + query = update.callback_query + await query.answer() + + data = query.data + + if data == "create_set": + await query.edit_message_text( + "📝 **Neues Stickerpaket erstellen**\n\n" + "Schritt 1/4: Gib einen kurzen Namen ein\n" + "(nur Kleinbuchstaben, Unterstriche, Zahlen):" + ) + return CREATE_SET_NAME + + elif data == "list_sets": + keyboard, text = await get_user_sets_keyboard(update.effective_user.id) + await query.edit_message_text( + text, + reply_markup=keyboard, + parse_mode="Markdown" + ) + return SELECT_SET + + elif data == "help": + await query.edit_message_text( + "❓ **Hilfe**\n\n" + "**Funktionen:**\n" + "• Neues Stickerpaket erstellen\n" + "• Sticker zu Paket hinzufügen (Bilder & GIFs)\n" + "• Sticker aus Paket löschen\n" + "• Emoji eines Stickers ändern\n" + "• Paket umbenennen\n" + "• Paket löschen\n\n" + "**Tipps:**\n" + "• Bilder: 512x512 PNG/WebP (automatisch konvertiert)\n" + "• GIFs: Werden zu animierten Video-Stickern\n" + "• Max. 3 Sekunden für Animationen\n" + "• Nutze /start um zum Hauptmenü zurückzukehren", + reply_markup=InlineKeyboardMarkup([ + [InlineKeyboardButton("⬅️ Zurück", callback_data="back_to_menu")] + ]) + ) + return MENU + + elif data == "add_existing": + await query.edit_message_text( + "📥 **Existierendes Stickerpaket hinzufügen**\n\n" + "Gib den Namen des Sticker-Pakets ein:\n\n" + "**Formate:**\n" + "• Kurzname: `meine_sticker` (wird automatisch ergänzt)\n" + "• Vollständig: `meine_sticker_by_botname`\n\n" + "Du findest den Namen im Sticker-Paket unter 'Info'." + ) + return ADD_EXISTING_SET_NAME + + elif data == "back_to_menu": + is_premium = await check_premium_status(update, context) + premium_text = "💎 Premium aktiv" if is_premium else "👤 Standard" + + await query.edit_message_text( + f"🎨 **Sticker Bot Manager**\n" + f"Status: {premium_text}\n\n" + "Wähle eine Aktion:", + reply_markup=get_main_menu_keyboard(), + parse_mode="Markdown" + ) + return MENU + + return MENU + + +async def create_set_name(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Get sticker set name from user.""" + name = update.message.text.strip().lower().replace(" ", "_") + + # Validate name + if not all(c.isalnum() or c == "_" for c in name): + await update.message.reply_text( + "⚠️ Ungültiger Name. Nur Kleinbuchstaben, Zahlen und Unterstriche erlaubt.\n" + "Versuche es erneut:" + ) + return CREATE_SET_NAME + + if len(name) > 60: + await update.message.reply_text( + "⚠️ Name zu lang (max. 60 Zeichen). Versuche es erneut:" + ) + return CREATE_SET_NAME + + context.user_data[SET_NAME] = name + await update.message.reply_text( + "📝 **Schritt 2/4**\n\n" + "Gib nun einen Titel für das Stickerpaket ein:" + ) + return CREATE_SET_TITLE + + +async def create_set_title(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Get sticker set title from user.""" + title = update.message.text.strip() + + if len(title) > 64: + await update.message.reply_text( + "⚠️ Titel zu lang (max. 64 Zeichen). Versuche es erneut:" + ) + return CREATE_SET_TITLE + + context.user_data[SET_TITLE] = title + await update.message.reply_text( + "🏷 **Schritt 3/4**\n\n" + "Gib ein Emoji für das erste Sticker ein (z.B. 🤖):" + ) + return CREATE_SET_EMOJI + + +async def create_set_emoji(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Get emoji for first sticker.""" + emoji = update.message.text.strip() + + if len(emoji) > 2: + await update.message.reply_text( + "⚠️ Bitte gib nur ein Emoji ein. Versuche es erneut:" + ) + return CREATE_SET_EMOJI + + context.user_data["emoji"] = emoji + await update.message.reply_text( + "🖼 **Schritt 4/4**\n\n" + "Sende nun das erste Sticker-Bild:\n" + "(PNG oder WebP, 512x512 Pixel empfohlen)" + ) + return CREATE_SET_STICKER + + +async def create_set_sticker(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Get first sticker and create the set with optional background removal.""" + if not update.message.photo and not update.message.document: + await update.message.reply_text( + "⚠️ Bitte sende ein Bild (Foto oder Dokument)." + ) + return CREATE_SET_STICKER + + # Get file + if update.message.photo: + file = await update.message.photo[-1].get_file() + else: + file = await update.message.document.get_file() + + name = context.user_data.get(SET_NAME) + title = context.user_data.get(SET_TITLE) + emoji = context.user_data.get("emoji", "🤖") + + # Check if API key is configured + if REMOVE_BG_API_KEY: + # Get current API usage + current_usage = await db.get_monthly_api_usage("remove_bg") + is_limit_reached = await should_show_api_limit_warning(REMOVE_BG_API_KEY, current_usage) + + # Show background removal options + button_text = f"☁️ Remove.bg API ({current_usage}/{MAX_API_CALLS_PER_MONTH})" + if is_limit_reached: + button_text += " ⚠️ Limit erreicht" + + keyboard = InlineKeyboardMarkup([ + [InlineKeyboardButton("🤖 Lokal entfernen (kostenlos)", callback_data="bg_local")], + [InlineKeyboardButton(button_text, callback_data="bg_api")], + [InlineKeyboardButton("⏭️ Ohne Entfernung", callback_data="bg_skip")], + ]) + + await update.message.reply_text( + "🎨 **Hintergrund entfernen?**\n\n" + "Wähle eine Option:", + reply_markup=keyboard, + parse_mode="Markdown" + ) + + # Store file for later processing + context.user_data["pending_file"] = file + context.user_data["pending_name"] = name + context.user_data["pending_title"] = title + context.user_data["pending_emoji"] = emoji + + return REMOVE_BG_CHOICE + else: + # No API key - process directly + await update.message.reply_text("⏳ Erstelle Stickerpaket...") + + success, result = await create_sticker_set( + update, context, name, title, file, emoji + ) + + if success: + await update.message.reply_text( + f"✅ **Stickerpaket erstellt!**\n\n" + f"📦 Titel: {title}\n" + f"🔗 Name: `{result}`\n\n" + f"Nutze /start um weitere Aktionen durchzuführen.", + parse_mode="Markdown" + ) + else: + await update.message.reply_text( + f"❌ **Fehler beim Erstellen:**\n{result}\n\n" + f"Mögliche Ursachen:\n" + f"• Name bereits vergeben\n" + f"• Ungültiges Bildformat\n" + f"• Bild zu groß/klein" + ) + + # Cleanup + context.user_data.pop(SET_NAME, None) + context.user_data.pop(SET_TITLE, None) + context.user_data.pop("emoji", None) + + return ConversationHandler.END + + +async def bg_choice_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Handle background removal choice.""" + query = update.callback_query + await query.answer() + + data = query.data + file = context.user_data.get("pending_file") + name = context.user_data.get("pending_name") + title = context.user_data.get("pending_title") + emoji = context.user_data.get("pending_emoji") + + if not file: + await query.edit_message_text("❌ Fehler: Kein Bild gefunden. Starte neu mit /start") + return ConversationHandler.END + + if data == "bg_local": + # Local background removal + await query.edit_message_text("⏳ Entferne Hintergrund (lokal)...") + + try: + # Download file + file_obj = await file.download_as_bytearray() + file_bytes = bytes(file_obj) + + # Validate image + is_valid, error_msg = await validate_image_for_bg_removal(file_bytes) + if not is_valid: + await query.edit_message_text(f"❌ **Fehler:**\n{error_msg}\n\nVersuche es mit einem anderen Bild.") + return ConversationHandler.END + + # Remove background + result = await remove_background_local(file_bytes) + + if result: + await query.edit_message_text("⏳ Erstelle Stickerpaket...") + + # Create sticker set with processed image + from image_processor import process_image_for_sticker + processed = await process_image_for_sticker(result) + + if processed: + from telegram import InputSticker + from telegram.constants import StickerFormat + + sticker = InputSticker( + sticker=processed, + format=StickerFormat.STATIC, + emoji_list=[emoji] + ) + + set_name = f"{name}_by_{context.bot.username}" + await context.bot.create_new_sticker_set( + user_id=update.effective_user.id, + name=set_name, + title=title, + stickers=[sticker] + ) + + await db.add_sticker_set(update.effective_user.id, set_name, title) + + await query.edit_message_text( + f"✅ **Stickerpaket erstellt!**\n\n" + f"📦 Titel: {title}\n" + f"🔗 Name: `{set_name}`\n\n" + f"Nutze /start um weitere Aktionen durchzuführen.", + parse_mode="Markdown" + ) + else: + await query.edit_message_text("❌ Fehler bei der Bildverarbeitung.") + else: + await query.edit_message_text( + "❌ **Hintergrundentfernung fehlgeschlagen**\n\n" + "Nutze /start um neu zu beginnen." + ) + except Exception as e: + logger.error(f"Error in local bg removal: {e}") + await query.edit_message_text(f"❌ Fehler: {str(e)}") + + return ConversationHandler.END + + elif data == "bg_api": + # API background removal + await query.edit_message_text("⏳ Verbinde mit Remove.bg API...") + + try: + # Download file + file_obj = await file.download_as_bytearray() + file_bytes = bytes(file_obj) + + # Validate image + is_valid, error_msg = await validate_image_for_bg_removal(file_bytes) + if not is_valid: + await query.edit_message_text(f"❌ **Fehler:**\n{error_msg}\n\nVersuche es mit einem anderen Bild.") + return ConversationHandler.END + + # Call API with retry logic + result, message = await remove_background_api(file_bytes, REMOVE_BG_API_KEY) + + if result == "LIMIT_REACHED" or result == "API_ERROR": + # Show error with fallback options + current_usage = await db.get_monthly_api_usage("remove_bg") + keyboard = InlineKeyboardMarkup([ + [InlineKeyboardButton("🤖 Stattdessen lokal entfernen", callback_data="bg_local")], + [InlineKeyboardButton("⏭️ Ohne Entfernung", callback_data="bg_skip")], + [InlineKeyboardButton("❌ Abbrechen", callback_data="bg_cancel")], + ]) + + warning_text = "" + if result == "LIMIT_REACHED": + warning_text = f"⚠️ **API-Limit erreicht** ({current_usage}/{MAX_API_CALLS_PER_MONTH})\n\n" + else: + warning_text = f"⚠️ **API-Fehler:** {message}\n\n" + + await query.edit_message_text( + warning_text + + "Was möchtest du stattdessen tun?", + reply_markup=keyboard, + parse_mode="Markdown" + ) + + return REMOVE_BG_API_ERROR + + # Success - track API call + await db.track_api_call("remove_bg") + await query.edit_message_text("⏳ Erstelle Stickerpaket...") + + # Process result + from image_processor import process_image_for_sticker + processed = await process_image_for_sticker(result) + + if processed: + from telegram import InputSticker + from telegram.constants import StickerFormat + + set_name = f"{name}_by_{context.bot.username}" + sticker = InputSticker( + sticker=processed, + format=StickerFormat.STATIC, + emoji_list=[emoji] + ) + + await context.bot.create_new_sticker_set( + user_id=update.effective_user.id, + name=set_name, + title=title, + stickers=[sticker] + ) + + await db.add_sticker_set(update.effective_user.id, set_name, title) + + await query.edit_message_text( + f"✅ **Stickerpaket erstellt!**\n\n" + f"📦 Titel: {title}\n" + f"🔗 Name: `{set_name}`\n\n" + f"Hintergrund entfernt mit Remove.bg API\n" + f"Nutze /start um weitere Aktionen durchzuführen.", + parse_mode="Markdown" + ) + else: + await query.edit_message_text("❌ Fehler bei der Bildverarbeitung.") + + except Exception as e: + logger.error(f"Error in API bg removal: {e}") + await query.edit_message_text(f"❌ Fehler: {str(e)}") + + return ConversationHandler.END + + elif data == "bg_skip": + # Skip background removal + await query.edit_message_text("⏳ Erstelle Stickerpaket...") + + try: + from telegram import InputSticker + from telegram.constants import StickerFormat + + file_obj = await file.download_as_bytearray() + file_bytes = bytes(file_obj) + + from image_processor import process_image_for_sticker + processed = await process_image_for_sticker(file_bytes) + + if processed: + set_name = f"{name}_by_{context.bot.username}" + sticker = InputSticker( + sticker=processed, + format=StickerFormat.STATIC, + emoji_list=[emoji] + ) + + await context.bot.create_new_sticker_set( + user_id=update.effective_user.id, + name=set_name, + title=title, + stickers=[sticker] + ) + + await db.add_sticker_set(update.effective_user.id, set_name, title) + + await query.edit_message_text( + f"✅ **Stickerpaket erstellt!**\n\n" + f"📦 Titel: {title}\n" + f"🔗 Name: `{set_name}`\n\n" + f"Nutze /start um weitere Aktionen durchzuführen.", + parse_mode="Markdown" + ) + else: + await query.edit_message_text("❌ Fehler bei der Bildverarbeitung.") + except Exception as e: + logger.error(f"Error creating sticker set: {e}") + await query.edit_message_text(f"❌ Fehler: {str(e)}") + + return ConversationHandler.END + + elif data == "bg_cancel": + await query.edit_message_text("❌ Abgebrochen. Nutze /start zum Neustart.") + return ConversationHandler.END + + return ConversationHandler.END + + +async def select_set_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Handle sticker set selection.""" + query = update.callback_query + await query.answer() + + data = query.data + + if data.startswith("select_set:"): + set_name = data.split(":", 1)[1] + context.user_data[CURRENT_SET] = set_name + + # Get set info from database + sets = await db.get_user_sticker_sets(update.effective_user.id) + set_info = next((s for s in sets if s['set_name'] == set_name), None) + title = set_info['set_title'] if set_info else set_name + + await query.edit_message_text( + f"📦 **Stickerpaket**\n" + f"📝 Titel: {title}\n" + f"🔗 Name: `{set_name}`\n\n" + f"Was möchtest du tun?", + reply_markup=get_set_action_keyboard(set_name), + parse_mode="Markdown" + ) + return SET_ACTION + + elif data == "back_to_menu": + return await menu_callback(update, context) + + return SELECT_SET + + +async def set_action_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Handle set action selection.""" + query = update.callback_query + await query.answer() + + data = query.data + set_name = context.user_data.get(CURRENT_SET) + + if data.startswith("add_sticker:"): + await query.edit_message_text( + "🖼 Sende ein neues Sticker-Bild:\n\n" + "(PNG oder WebP, 512x512 Pixel empfohlen)" + ) + return ADD_STICKER + + elif data.startswith("delete_sticker:"): + await query.edit_message_text( + "🗑 Sende den Sticker, den du löschen möchtest:\n\n" + "(Klicke auf einen Sticker und wähle 'Weiterleiten' zu diesem Bot)" + ) + return DELETE_STICKER + + elif data.startswith("change_emoji:"): + await query.edit_message_text( + "🗑 Sende den Sticker, dessen Emoji du ändern möchtest:\n\n" + "(Klicke auf einen Sticker und wähle 'Weiterleiten')" + ) + return CHANGE_EMOJI + + elif data.startswith("rename_set:"): + await query.edit_message_text( + "✏️ Gib den neuen Titel für das Paket ein:" + ) + return RENAME_SET + + elif data.startswith("delete_set:"): + keyboard = InlineKeyboardMarkup([ + [InlineKeyboardButton("✅ Ja, löschen", callback_data="confirm_delete")], + [InlineKeyboardButton("❌ Abbrechen", callback_data="cancel_delete")], + ]) + await query.edit_message_text( + f"⚠️ **Bist du sicher?**\n\n" + f"Das Stickerpaket wird unwiderruflich gelöscht!", + reply_markup=keyboard, + parse_mode="Markdown" + ) + return DELETE_CONFIRM + + elif data == "back_to_menu": + return await menu_callback(update, context) + + return SET_ACTION + + +async def add_sticker_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Handle new sticker addition - including GIFs.""" + set_name = context.user_data.get(CURRENT_SET) + + # Check for GIF/Animation + if update.message.animation: + # It's a GIF/animation + animation = update.message.animation + + # Download the file + file = await animation.get_file() + file_obj = await file.download_as_bytearray() + gif_bytes = bytes(file_obj) + + # Get GIF info + gif_info = await get_gif_info(gif_bytes) + duration = gif_info.get('duration', 0) + + # Check if duration exceeds limit + is_valid, error_msg = await validate_gif_duration(duration) + + if not is_valid: + # GIF too long - ask user + context.user_data[PENDING_ANIMATION] = { + 'bytes': gif_bytes, + 'duration': duration + } + + keyboard = InlineKeyboardMarkup([ + [InlineKeyboardButton("✂️ Auf 3 Sekunden kürzen", callback_data="trim_gif")], + [InlineKeyboardButton("❌ Abbrechen", callback_data="cancel_gif")], + ]) + + await update.message.reply_text( + f"🎬 **GIF zu lang**\n\n" + f"Dauer: {duration:.1f} Sekunden\n" + f"Maximal erlaubt: 3 Sekunden\n\n" + f"Was möchtest du tun?", + reply_markup=keyboard, + parse_mode="Markdown" + ) + return GIF_DURATION_CONFIRM + + # GIF is within limits - proceed with conversion + context.user_data[PENDING_ANIMATION] = {'bytes': gif_bytes, 'duration': duration} + await update.message.reply_text("🏷 Gib ein Emoji für den animierten Sticker ein (z.B. 😊):") + return ADD_ANIMATION_EMOJI + + # Handle regular images + if not update.message.photo and not update.message.document: + await update.message.reply_text("⚠️ Bitte sende ein Bild oder GIF.") + return ADD_STICKER + + # Get file + if update.message.photo: + file = await update.message.photo[-1].get_file() + else: + file = await update.message.document.get_file() + + # Store file for next step (emoji selection) + context.user_data["pending_sticker"] = file + await update.message.reply_text( + "🏷 Gib ein Emoji für den neuen Sticker ein (z.B. 😊):" + ) + return ADD_STICKER_EMOJI + + +async def gif_duration_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Handle GIF duration confirmation callback.""" + query = update.callback_query + await query.answer() + + if query.data == "trim_gif": + await query.edit_message_text( + "✂️ GIF wird auf 3 Sekunden gekürzt...\n\n" + "🏷 Gib ein Emoji für den animierten Sticker ein (z.B. 😊):" + ) + return ADD_ANIMATION_EMOJI + + elif query.data == "cancel_gif": + await query.edit_message_text("❌ GIF-Hinzufügen abgebrochen.") + context.user_data.pop(PENDING_ANIMATION, None) + return ConversationHandler.END + + return GIF_DURATION_CONFIRM + + +async def add_animation_emoji(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Get emoji for animated sticker and process GIF.""" + emoji = update.message.text.strip() + set_name = context.user_data.get(CURRENT_SET) + anim_data = context.user_data.get(PENDING_ANIMATION) + + if not anim_data: + await update.message.reply_text("❌ Fehler: Kein GIF gefunden. Starte neu mit /start") + return ConversationHandler.END + + gif_bytes = anim_data['bytes'] + duration = anim_data['duration'] + + # Determine max duration + max_duration = 3.0 if duration > 3.0 else None + + await update.message.reply_text("⏳ Konvertiere GIF zu Video Sticker...") + + # Convert GIF to WebM + webm_bytes = await convert_gif_to_webm(gif_bytes, max_duration) + + if not webm_bytes: + await update.message.reply_text( + "❌ **Fehler bei der Konvertierung**\n\n" + "Das GIF konnte nicht konvertiert werden.\n" + "Mögliche Ursachen:\n" + "• Datei zu groß\n" + "• Ungültiges Format\n" + "• Komprimierung fehlgeschlagen" + ) + context.user_data.pop(PENDING_ANIMATION, None) + return ConversationHandler.END + + await update.message.reply_text("⏳ Füge animierten Sticker hinzu...") + + # Add as video sticker + success = await add_video_sticker_to_set(update, context, set_name, webm_bytes, emoji) + + if success: + await update.message.reply_text(f"✅ Animierter Sticker hinzugefügt mit Emoji {emoji}!") + else: + await update.message.reply_text("❌ Fehler beim Hinzufügen des animierten Stickers.") + + context.user_data.pop(PENDING_ANIMATION, None) + return ConversationHandler.END + + +async def add_existing_set_name(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Handle existing sticker set name input.""" + set_name_input = update.message.text.strip() + + # Auto-append _by_ if not present + if "_by_" not in set_name_input: + bot_username = context.bot.username + set_name = f"{set_name_input}_by_{bot_username}" + else: + set_name = set_name_input + + await update.message.reply_text(f"⏳ Suche Stickerpaket `{set_name}`...", parse_mode="Markdown") + + # Try to add existing set + success, message = await add_existing_sticker_set(update, context, set_name) + + if success: + await update.message.reply_text( + f"✅ {message}\n\n" + f"Nutze /start um das Paket zu verwalten.", + parse_mode="Markdown" + ) + else: + await update.message.reply_text( + f"❌ **Fehler:**\n{message}\n\n" + f"**Tipps:**\n" + f"• Prüfe den Namen (z.B. `meine_sticker` oder `meine_sticker_by_botname`)\n" + f"• Das Paket muss öffentlich sein\n" + f"• Du musst Besitzer des Pakets sein\n\n" + f"Versuche es erneut oder nutze /start.", + parse_mode="Markdown" + ) + + return ConversationHandler.END + + +async def add_sticker_emoji(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Get emoji for new sticker and add it.""" + emoji = update.message.text.strip() + set_name = context.user_data.get(CURRENT_SET) + file = context.user_data.get("pending_sticker") + + if not file: + await update.message.reply_text("❌ Fehler: Kein Sticker gefunden. Starte neu mit /start") + return ConversationHandler.END + + await update.message.reply_text("⏳ Füge Sticker hinzu...") + + success = await add_sticker_to_set(update, context, set_name, file, emoji) + + if success: + await update.message.reply_text(f"✅ Sticker hinzugefügt mit Emoji {emoji}!") + else: + await update.message.reply_text("❌ Fehler beim Hinzufügen des Stickers.") + + context.user_data.pop("pending_sticker", None) + return ConversationHandler.END + + +async def delete_sticker_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Handle sticker deletion.""" + if not update.message.sticker: + await update.message.reply_text("⚠️ Bitte sende einen Sticker (nicht ein Bild).") + return DELETE_STICKER + + sticker_file_id = update.message.sticker.file_id + + await update.message.reply_text("⏳ Lösche Sticker...") + + success = await delete_sticker_from_set(update, context, sticker_file_id) + + if success: + await update.message.reply_text("✅ Sticker gelöscht!") + else: + await update.message.reply_text("❌ Fehler beim Löschen des Stickers.") + + return ConversationHandler.END + + +async def change_emoji_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Handle sticker emoji change - step 1: get sticker.""" + if not update.message.sticker: + await update.message.reply_text("⚠️ Bitte sende einen Sticker.") + return CHANGE_EMOJI + + sticker_file_id = update.message.sticker.file_id + context.user_data[STICKER_TO_CHANGE] = sticker_file_id + + await update.message.reply_text( + "🏷 Gib das neue Emoji für den Sticker ein (z.B. 😊):" + ) + return CHANGE_EMOJI + + +async def change_emoji_input(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Handle new emoji input.""" + new_emoji = update.message.text.strip() + sticker_file_id = context.user_data.get(STICKER_TO_CHANGE) + + if not sticker_file_id: + await update.message.reply_text("❌ Fehler. Starte neu mit /start") + return ConversationHandler.END + + await update.message.reply_text("⏳ Ändere Emoji...") + + success = await change_sticker_emoji(update, context, sticker_file_id, new_emoji) + + if success: + await update.message.reply_text(f"✅ Emoji geändert zu {new_emoji}!") + else: + await update.message.reply_text("❌ Fehler beim Ändern des Emojis.") + + context.user_data.pop(STICKER_TO_CHANGE, None) + return ConversationHandler.END + + +async def rename_set_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Handle sticker set renaming.""" + set_name = context.user_data.get(CURRENT_SET) + new_title = update.message.text.strip() + + if len(new_title) > 64: + await update.message.reply_text("⚠️ Titel zu lang (max. 64 Zeichen). Versuche es erneut:") + return RENAME_SET + + await update.message.reply_text("⏳ Benenne Paket um...") + + success = await rename_sticker_set(update, context, set_name, new_title) + + if success: + await update.message.reply_text(f"✅ Paket umbenannt in: {new_title}") + else: + await update.message.reply_text("❌ Fehler beim Umbenennen des Pakets.") + + return ConversationHandler.END + + +async def delete_confirm_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Handle set deletion confirmation.""" + query = update.callback_query + await query.answer() + + if query.data == "confirm_delete": + set_name = context.user_data.get(CURRENT_SET) + + await query.edit_message_text("⏳ Lösche Paket...") + + success = await delete_sticker_set(update, context, set_name) + + if success: + await query.edit_message_text("✅ Stickerpaket gelöscht!") + else: + await query.edit_message_text("❌ Fehler beim Löschen des Pakets.") + + context.user_data.pop(CURRENT_SET, None) + return ConversationHandler.END + + elif query.data == "cancel_delete": + set_name = context.user_data.get(CURRENT_SET) + sets = await db.get_user_sticker_sets(update.effective_user.id) + set_info = next((s for s in sets if s['set_name'] == set_name), None) + title = set_info['set_title'] if set_info else set_name + + await query.edit_message_text( + f"📦 **Stickerpaket**\n" + f"📝 Titel: {title}\n" + f"🔗 Name: `{set_name}`\n\n" + f"Was möchtest du tun?", + reply_markup=get_set_action_keyboard(set_name), + parse_mode="Markdown" + ) + return SET_ACTION + + return ConversationHandler.END + + +async def cancel(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: + """Cancel conversation.""" + await update.message.reply_text( + "❌ Aktion abgebrochen. Nutze /start zum Neustart." + ) + # Cleanup + keys_to_remove = [SET_NAME, SET_TITLE, CURRENT_SET, STICKER_TO_CHANGE, PENDING_ANIMATION, + "pending_file", "pending_name", "pending_title", "pending_emoji", "emoji", "pending_sticker"] + for key in keys_to_remove: + context.user_data.pop(key, None) + return ConversationHandler.END + + +async def post_init(application: Application): + """Initialize database and schedule jobs after bot startup.""" + await db.init() + + # Schedule premium check job + if application.job_queue: + application.job_queue.run_repeating( + schedule_premium_check, + interval=PREMIUM_CHECK_INTERVAL * 3600, # Convert to seconds + first=10 # First check after 10 seconds + ) + logger.info(f"Scheduled premium check every {PREMIUM_CHECK_INTERVAL} hours") + + +def main(): + """Run the bot.""" + application = Application.builder().token(BOT_TOKEN).build() + + # Add post_init callback + application.post_init = post_init + + conv_handler = ConversationHandler( + entry_points=[CommandHandler("start", start)], + states={ + MENU: [ + CallbackQueryHandler(menu_callback, pattern="^(create_set|list_sets|help|back_to_menu)$"), + ], + CREATE_SET_NAME: [ + MessageHandler(filters.TEXT & ~filters.COMMAND, create_set_name), + ], + CREATE_SET_TITLE: [ + MessageHandler(filters.TEXT & ~filters.COMMAND, create_set_title), + ], + CREATE_SET_EMOJI: [ + MessageHandler(filters.TEXT & ~filters.COMMAND, create_set_emoji), + ], + CREATE_SET_STICKER: [ + MessageHandler(filters.PHOTO | filters.Document.ALL, create_set_sticker), + ], + SELECT_SET: [ + CallbackQueryHandler(select_set_callback, pattern="^(select_set:|back_to_menu)$"), + ], + SET_ACTION: [ + CallbackQueryHandler(set_action_callback, pattern="^(add_sticker:|delete_sticker:|change_emoji:|rename_set:|delete_set:|back_to_menu)$"), + ], + ADD_STICKER: [ + MessageHandler(filters.PHOTO | filters.Document.ALL | filters.ANIMATION, add_sticker_handler), + ], + ADD_STICKER_EMOJI: [ + MessageHandler(filters.TEXT & ~filters.COMMAND, add_sticker_emoji), + ], + GIF_DURATION_CONFIRM: [ + CallbackQueryHandler(gif_duration_callback, pattern="^(trim_gif|cancel_gif)$"), + ], + ADD_ANIMATION_EMOJI: [ + MessageHandler(filters.TEXT & ~filters.COMMAND, add_animation_emoji), + ], + DELETE_STICKER: [ + MessageHandler(filters.Sticker.ALL, delete_sticker_handler), + ], + CHANGE_EMOJI: [ + MessageHandler(filters.Sticker.ALL, change_emoji_handler), + ], + RENAME_SET: [ + MessageHandler(filters.TEXT & ~filters.COMMAND, rename_set_handler), + ], + DELETE_CONFIRM: [ + CallbackQueryHandler(delete_confirm_callback, pattern="^(confirm_delete|cancel_delete)$"), + ], + REMOVE_BG_CHOICE: [ + CallbackQueryHandler(bg_choice_callback, pattern="^(bg_local|bg_api|bg_skip|bg_cancel)$"), + ], + REMOVE_BG_API_ERROR: [ + CallbackQueryHandler(bg_choice_callback, pattern="^(bg_local|bg_skip|bg_cancel)$"), + ], + }, + fallbacks=[CommandHandler("cancel", cancel)], + ) + + application.add_handler(conv_handler) + + logger.info("Starting sticker bot...") + application.run_polling(allowed_updates=Update.ALL_TYPES) + + +if __name__ == "__main__": + main() + + +async def hilfe_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + """Send help message.""" + if not await check_owner(update): + await update.message.reply_text("⛔️ Dieser Bot ist nur für den Besitzer verfügbar.") + return + + help_text = ( + "❓ **Hilfe - Sticker Bot Manager**\n\n" + + "**📱 Befehle:**\n" + "• /start - Bot starten\n" + "• /hilfe - Diese Hilfe anzeigen\n" + "• /cancel - Aktuelle Aktion abbrechen\n\n" + + "**🎨 Funktionen:**\n" + "1️⃣ **Neues Stickerpaket**\n" + " - Name → Titel → Emoji → Bild\n" + " - Bilder werden automatisch optimiert\n\n" + + "2️⃣ **Sticker hinzufügen**\n" + " - Bilder: PNG, JPG, WebP (automatische Konvertierung)\n" + " - GIFs: Werden zu animierten Video-Stickern (max. 3s)\n" + " - Optional: Hintergrund automatisch entfernen\n\n" + + "3️⃣ **Sticker löschen**\n" + " - Sende den Sticker direkt an den Bot\n\n" + + "4️⃣ **Emoji ändern**\n" + " - Sticker senden → Neuen Emoji eingeben\n\n" + + "5️⃣ **Paket umbenennen**\n" + " - Neuen Titel eingeben (max. 64 Zeichen)\n\n" + + "6️⃣ **Paket löschen**\n" + " - ⚠️ Kann nicht rückgängig gemacht werden!\n\n" + + "**💡 Tipps:**\n" + "• Bilder als Datei senden für beste Qualität\n" + "• Ideale Größe: 512x512 Pixel\n" + "• Max. 120 Sticker pro reguläres Paket\n" + "• Max. 200 Sticker pro Emoji-Paket\n\n" + + "**🎨 Hintergrundentfernung:**\n" + "• Lokal: Kostenlos, direkt auf dem Server\n" + "• Remove.bg API: 50 Aufrufe/Monat gratis\n\n" + + "Nutze /start zum Öffnen des Hauptmenüs." + ) + + await update.message.reply_text(help_text, parse_mode="Markdown") + + +def main(): + """Run the bot.""" + application = Application.builder().token(BOT_TOKEN).build() + + # Add post_init callback + application.post_init = post_init + + # Add standalone command handlers + application.add_handler(CommandHandler("hilfe", hilfe_command)) + + conv_handler = ConversationHandler( + entry_points=[CommandHandler("start", start)], + states={ + MENU: [ + CallbackQueryHandler(menu_callback, pattern="^(create_set|list_sets|help|back_to_menu)$"), + ], + CREATE_SET_NAME: [ + MessageHandler(filters.TEXT & ~filters.COMMAND, create_set_name), + ], + CREATE_SET_TITLE: [ + MessageHandler(filters.TEXT & ~filters.COMMAND, create_set_title), + ], + CREATE_SET_EMOJI: [ + MessageHandler(filters.TEXT & ~filters.COMMAND, create_set_emoji), + ], + CREATE_SET_STICKER: [ + MessageHandler(filters.PHOTO | filters.Document.ALL, create_set_sticker), + ], + SELECT_SET: [ + CallbackQueryHandler(select_set_callback, pattern="^(select_set:|back_to_menu)$"), + ], + SET_ACTION: [ + CallbackQueryHandler(set_action_callback, pattern="^(add_sticker:|delete_sticker:|change_emoji:|rename_set:|delete_set:|back_to_menu)$"), + ], + ADD_STICKER: [ + MessageHandler(filters.PHOTO | filters.Document.ALL | filters.ANIMATION, add_sticker_handler), + ], + ADD_STICKER_EMOJI: [ + MessageHandler(filters.TEXT & ~filters.COMMAND, add_sticker_emoji), + ], + GIF_DURATION_CONFIRM: [ + CallbackQueryHandler(gif_duration_callback, pattern="^(trim_gif|cancel_gif)$"), + ], + ADD_ANIMATION_EMOJI: [ + MessageHandler(filters.TEXT & ~filters.COMMAND, add_animation_emoji), + ], + DELETE_STICKER: [ + MessageHandler(filters.Sticker.ALL, delete_sticker_handler), + ], + CHANGE_EMOJI: [ + MessageHandler(filters.Sticker.ALL, change_emoji_handler), + ], + RENAME_SET: [ + MessageHandler(filters.TEXT & ~filters.COMMAND, rename_set_handler), + ], + DELETE_CONFIRM: [ + CallbackQueryHandler(delete_confirm_callback, pattern="^(confirm_delete|cancel_delete)$"), + ], + REMOVE_BG_CHOICE: [ + CallbackQueryHandler(bg_choice_callback, pattern="^(bg_local|bg_api|bg_skip|bg_cancel)$"), + ], + REMOVE_BG_API_ERROR: [ + CallbackQueryHandler(bg_choice_callback, pattern="^(bg_local|bg_skip|bg_cancel)$"), + ], + }, + fallbacks=[CommandHandler("cancel", cancel)], + ) + + application.add_handler(conv_handler) + + logger.info("Starting sticker bot...") + application.run_polling(allowed_updates=Update.ALL_TYPES) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/config.py b/config.py new file mode 100644 index 0000000..dfd371c --- /dev/null +++ b/config.py @@ -0,0 +1,38 @@ +"""Configuration module for sticker bot.""" +import os +import logging +from dotenv import load_dotenv + +load_dotenv() + +# Logging setup +logging.basicConfig( + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", + level=logging.INFO +) +logger = logging.getLogger(__name__) + +# Bot Configuration +BOT_TOKEN = os.getenv("BOT_TOKEN") +OWNER_ID = os.getenv("OWNER_ID") + +if not BOT_TOKEN: + raise ValueError("BOT_TOKEN must be set in .env file") + +if not OWNER_ID: + raise ValueError("OWNER_ID must be set in .env file") + +try: + OWNER_ID = int(OWNER_ID) +except ValueError: + raise ValueError("OWNER_ID must be a valid integer") + +# Database Configuration +DATABASE_PATH = os.getenv("DATABASE_PATH", "sticker_bot.db") + +# Premium Check Interval (in hours) +PREMIUM_CHECK_INTERVAL = int(os.getenv("PREMIUM_CHECK_INTERVAL", "24")) + +# Optional: Remove.bg API Key for background removal +# If not set, only local processing is available +REMOVE_BG_API_KEY = os.getenv("REMOVE_BG_API_KEY") \ No newline at end of file diff --git a/database.py b/database.py new file mode 100644 index 0000000..f2048ee --- /dev/null +++ b/database.py @@ -0,0 +1,174 @@ +"""Database module for sticker bot.""" +import aiosqlite +from datetime import datetime +from typing import List, Optional, Dict, Any +from config import DATABASE_PATH, logger + + +class Database: + """SQLite database handler for sticker bot.""" + + def __init__(self, db_path: str = DATABASE_PATH): + self.db_path = db_path + + async def init(self): + """Initialize database tables.""" + async with aiosqlite.connect(self.db_path) as db: + # Sticker sets table + await db.execute(""" + CREATE TABLE IF NOT EXISTS sticker_sets ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id INTEGER NOT NULL, + set_name TEXT NOT NULL UNIQUE, + set_title TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + + # User premium status table + await db.execute(""" + CREATE TABLE IF NOT EXISTS user_premium ( + user_id INTEGER PRIMARY KEY, + is_premium BOOLEAN DEFAULT FALSE, + last_checked TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + + # Track stickers for each set + await db.execute(""" + CREATE TABLE IF NOT EXISTS stickers ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + set_name TEXT NOT NULL, + file_id TEXT NOT NULL, + emoji TEXT, + added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY (set_name) REFERENCES sticker_sets(set_name) ON DELETE CASCADE + ) + """) + + # API usage tracking (for remove.bg and other external services) + await db.execute(""" + CREATE TABLE IF NOT EXISTS api_usage ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + service TEXT NOT NULL, + year_month TEXT NOT NULL, + usage_count INTEGER DEFAULT 0, + last_used TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + UNIQUE(service, year_month) + ) + """) + + await db.commit() + logger.info("Database initialized successfully") + + async def add_sticker_set(self, user_id: int, set_name: str, set_title: str): + """Add a new sticker set to tracking.""" + async with aiosqlite.connect(self.db_path) as db: + await db.execute( + """INSERT OR REPLACE INTO sticker_sets (user_id, set_name, set_title, updated_at) + VALUES (?, ?, ?, CURRENT_TIMESTAMP)""", + (user_id, set_name, set_title) + ) + await db.commit() + + async def get_user_sticker_sets(self, user_id: int) -> List[Dict[str, Any]]: + """Get all sticker sets for a user.""" + async with aiosqlite.connect(self.db_path) as db: + db.row_factory = aiosqlite.Row + async with db.execute( + "SELECT * FROM sticker_sets WHERE user_id = ? ORDER BY created_at DESC", + (user_id,) + ) as cursor: + rows = await cursor.fetchall() + return [dict(row) for row in rows] + + async def delete_sticker_set(self, set_name: str): + """Delete a sticker set from tracking.""" + async with aiosqlite.connect(self.db_path) as db: + await db.execute("DELETE FROM sticker_sets WHERE set_name = ?", (set_name,)) + await db.commit() + + async def rename_sticker_set(self, set_name: str, new_title: str): + """Update sticker set title.""" + async with aiosqlite.connect(self.db_path) as db: + await db.execute( + """UPDATE sticker_sets SET set_title = ?, updated_at = CURRENT_TIMESTAMP + WHERE set_name = ?""", + (new_title, set_name) + ) + await db.commit() + + async def update_premium_status(self, user_id: int, is_premium: bool): + """Update user's premium status.""" + async with aiosqlite.connect(self.db_path) as db: + await db.execute( + """INSERT OR REPLACE INTO user_premium (user_id, is_premium, last_checked) + VALUES (?, ?, CURRENT_TIMESTAMP)""", + (user_id, is_premium) + ) + await db.commit() + + async def get_premium_status(self, user_id: int) -> Optional[bool]: + """Get user's premium status.""" + async with aiosqlite.connect(self.db_path) as db: + async with db.execute( + "SELECT is_premium FROM user_premium WHERE user_id = ?", + (user_id,) + ) as cursor: + row = await cursor.fetchone() + return row[0] if row else None + + async def add_sticker(self, set_name: str, file_id: str, emoji: Optional[str] = None): + """Track a sticker in the database.""" + async with aiosqlite.connect(self.db_path) as db: + await db.execute( + "INSERT INTO stickers (set_name, file_id, emoji) VALUES (?, ?, ?)", + (set_name, file_id, emoji) + ) + await db.commit() + + async def delete_sticker(self, file_id: str): + """Remove a sticker from tracking.""" + async with aiosqlite.connect(self.db_path) as db: + await db.execute("DELETE FROM stickers WHERE file_id = ?", (file_id,)) + await db.commit() + + async def get_stickers_in_set(self, set_name: str) -> List[Dict[str, Any]]: + """Get all stickers in a set.""" + async with aiosqlite.connect(self.db_path) as db: + db.row_factory = aiosqlite.Row + async with db.execute( + "SELECT * FROM stickers WHERE set_name = ?", + (set_name,) + ) as cursor: + rows = await cursor.fetchall() + return [dict(row) for row in rows] + + async def track_api_call(self, service: str): + """Track an API call for usage statistics.""" + year_month = datetime.now().strftime("%Y-%m") + async with aiosqlite.connect(self.db_path) as db: + await db.execute( + """INSERT INTO api_usage (service, year_month, usage_count, last_used) + VALUES (?, ?, 1, CURRENT_TIMESTAMP) + ON CONFLICT(service, year_month) + DO UPDATE SET usage_count = usage_count + 1, last_used = CURRENT_TIMESTAMP""", + (service, year_month) + ) + await db.commit() + + async def get_monthly_api_usage(self, service: str) -> int: + """Get current month's API usage count.""" + year_month = datetime.now().strftime("%Y-%m") + async with aiosqlite.connect(self.db_path) as db: + async with db.execute( + "SELECT usage_count FROM api_usage WHERE service = ? AND year_month = ?", + (service, year_month) + ) as cursor: + row = await cursor.fetchone() + return row[0] if row else 0 + + +# Global database instance +db = Database() \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..58a0f11 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,19 @@ +version: '3.8' + +services: + stickerbot: + build: . + container_name: telegram-sticker-bot + restart: unless-stopped + env_file: + - .env + volumes: + - ./data:/app/data + - ./sticker_bot.db:/app/sticker_bot.db + environment: + - DATABASE_PATH=/app/data/sticker_bot.db + logging: + driver: "json-file" + options: + max-size: "10m" + max-file: "3" \ No newline at end of file diff --git a/gif_processor.py b/gif_processor.py new file mode 100644 index 0000000..27f08ac --- /dev/null +++ b/gif_processor.py @@ -0,0 +1,267 @@ +"""GIF to WebM Video Sticker converter using ffmpeg.""" +import io +import logging +import subprocess +import tempfile +from typing import Optional, Tuple, Dict, Any +from PIL import Image + +logger = logging.getLogger(__name__) + +# Telegram Video Sticker requirements +MAX_DURATION_SECONDS = 3 +MAX_FILE_SIZE_KB = 256 +MAX_FILE_SIZE_BYTES = MAX_FILE_SIZE_KB * 1024 +STICKER_SIZE = 512 +TARGET_FPS = 30 + + +async def get_gif_info(gif_bytes: bytes) -> Dict[str, Any]: + """ + Get GIF metadata using ffprobe. + + Returns dict with duration, width, height, fps + """ + try: + with tempfile.NamedTemporaryFile(suffix='.gif', delete=False) as tmp_in: + tmp_in.write(gif_bytes) + tmp_in.flush() + + # Use ffprobe to get duration + result = subprocess.run( + [ + 'ffprobe', '-v', 'error', '-show_entries', + 'format=duration', '-show_entries', + 'stream=width,height,r_frame_rate', + '-of', 'default=noprint_wrappers=1', + tmp_in.name + ], + capture_output=True, + text=True + ) + + info = {'duration': 0, 'width': 0, 'height': 0, 'fps': 0} + + for line in result.stdout.strip().split('\n'): + if '=' in line: + key, value = line.split('=', 1) + if key == 'duration': + info['duration'] = float(value) + elif key == 'width': + info['width'] = int(value) + elif key == 'height': + info['height'] = int(value) + elif key == 'r_frame_rate': + # Parse fraction like "30/1" + if '/' in value: + num, den = value.split('/') + info['fps'] = int(num) / int(den) if int(den) != 0 else 0 + + return info + + except Exception as e: + logger.error(f"Error getting GIF info: {e}") + return {'duration': 0, 'width': 0, 'height': 0, 'fps': 0} + + +async def validate_gif_duration(duration: float) -> Tuple[bool, str]: + """ + Check if GIF duration is within limits. + + Returns: + Tuple of (is_valid, message) + """ + if duration > MAX_DURATION_SECONDS: + return False, f"GIF ist {duration:.1f}s lang. Maximal erlaubt: {MAX_DURATION_SECONDS}s" + return True, "" + + +async def convert_gif_to_webm( + gif_bytes: bytes, + max_duration: Optional[float] = None +) -> Optional[bytes]: + """ + Convert GIF to WebM Video Sticker format using ffmpeg. + + Args: + gif_bytes: Raw GIF file bytes + max_duration: Maximum duration in seconds (None = no limit, will be truncated) + + Returns: + WebM video bytes or None if conversion fails + """ + try: + # Get GIF info + gif_info = await get_gif_info(gif_bytes) + duration = gif_info.get('duration', 0) + width = gif_info.get('width', 512) + height = gif_info.get('height', 512) + + logger.info(f"Converting GIF: {width}x{height}, {duration}s, " + f"{gif_info.get('fps', 0)}fps") + + # Calculate dimensions (one side must be exactly 512px) + if width >= height: + new_width = STICKER_SIZE + new_height = int((height / width) * STICKER_SIZE) + else: + new_height = STICKER_SIZE + new_width = int((width / height) * STICKER_SIZE) + + # Ensure neither exceeds 512 + new_width = min(new_width, STICKER_SIZE) + new_height = min(new_height, STICKER_SIZE) + + # Determine duration limit + duration_limit = min(duration, max_duration or duration) + if duration > MAX_DURATION_SECONDS: + duration_limit = MAX_DURATION_SECONDS + logger.info(f"GIF too long ({duration}s), truncating to {duration_limit}s") + + # Create temporary files + with tempfile.NamedTemporaryFile(suffix='.gif', delete=False) as tmp_in: + tmp_in.write(gif_bytes) + tmp_in_path = tmp_in.name + + with tempfile.NamedTemporaryFile(suffix='.webm', delete=False) as tmp_out: + tmp_out_path = tmp_out.name + + # Build ffmpeg command + # Video sticker requirements: WebM, VP9, 512px, 30fps, no audio, max 3s, max 256KB + cmd = [ + 'ffmpeg', + '-i', tmp_in_path, + '-t', str(duration_limit), # Duration limit + '-vf', f'scale={new_width}:{new_height}:flags=lanczos,fps={TARGET_FPS}', # Resize + FPS + '-c:v', 'libvpx-vp9', # VP9 codec + '-pix_fmt', 'yuva420p', # With alpha channel (transparency) + '-b:v', '0', # Use constant quality + '-crf', '40', # Quality (lower = better, 0-63) + '-deadline', 'good', # Encoding speed/quality tradeoff + '-cpu-used', '2', # Encoding speed + '-row-mt', '1', # Multi-threading + '-an', # No audio + '-f', 'webm', # Output format + '-y', # Overwrite output + tmp_out_path + ] + + # Run ffmpeg + result = subprocess.run( + cmd, + capture_output=True, + text=True + ) + + if result.returncode != 0: + logger.error(f"ffmpeg failed: {result.stderr}") + return None + + # Read output file + with open(tmp_out_path, 'rb') as f: + webm_bytes = f.read() + + file_size_kb = len(webm_bytes) / 1024 + logger.info(f"Initial conversion: {file_size_kb:.1f}KB") + + # If too large, try with lower quality + if len(webm_bytes) > MAX_FILE_SIZE_BYTES: + webm_bytes = await _compress_webm(tmp_in_path, tmp_out_path, duration_limit, new_width, new_height) + + if webm_bytes and len(webm_bytes) <= MAX_FILE_SIZE_BYTES: + final_size_kb = len(webm_bytes) / 1024 + logger.info(f"Final WebM: {final_size_kb:.1f}KB, " + f"{new_width}x{new_height}") + return webm_bytes + else: + logger.error("Could not compress to required size") + return None + + except Exception as e: + logger.error(f"Error converting GIF to WebM: {e}") + return None + + +async def _compress_webm( + input_path: str, + output_path: str, + duration: float, + width: int, + height: int +) -> Optional[bytes]: + """ + Try to compress WebM to meet size requirements. + + Tries progressively lower quality settings. + """ + # Try different CRF values (lower = better quality, higher = smaller file) + crf_values = [45, 50, 55, 60] + + for crf in crf_values: + try: + logger.info(f"Trying compression with CRF={crf}") + + cmd = [ + 'ffmpeg', + '-i', input_path, + '-t', str(duration), + '-vf', f'scale={width}:{height}:flags=lanczos,fps={TARGET_FPS}', + '-c:v', 'libvpx-vp9', + '-pix_fmt', 'yuva420p', + '-b:v', '0', + '-crf', str(crf), + '-deadline', 'good', + '-cpu-used', '3', # Faster encoding for smaller files + '-row-mt', '1', + '-an', + '-f', 'webm', + '-y', + output_path + ] + + result = subprocess.run(cmd, capture_output=True, text=True) + + if result.returncode == 0: + with open(output_path, 'rb') as f: + webm_bytes = f.read() + + file_size_kb = len(webm_bytes) / 1024 + logger.info(f"Compression CRF={crf}: {file_size_kb:.1f}KB") + + if len(webm_bytes) <= MAX_FILE_SIZE_BYTES: + return webm_bytes + + except Exception as e: + logger.error(f"Compression attempt failed: {e}") + continue + + return None + + +async def is_animated_image(file_bytes: bytes) -> bool: + """ + Check if image file is animated (GIF with multiple frames). + + Args: + file_bytes: Raw file bytes + + Returns: + True if animated, False otherwise + """ + try: + image = Image.open(io.BytesIO(file_bytes)) + + # Check if it's a GIF + if image.format != 'GIF': + return False + + # Check if it has multiple frames + try: + image.seek(1) # Try to go to second frame + return True + except EOFError: + return False + + except Exception as e: + logger.error(f"Error checking if image is animated: {e}") + return False diff --git a/image_processor.py b/image_processor.py new file mode 100644 index 0000000..885d918 --- /dev/null +++ b/image_processor.py @@ -0,0 +1,169 @@ +"""Image processing module for converting images to Telegram sticker format.""" +import io +import logging +from typing import Optional, Tuple +from PIL import Image + +logger = logging.getLogger(__name__) + +# Telegram sticker requirements +STICKER_SIZE = 512 # One side must be exactly 512px +MAX_FILE_SIZE_MB = 3 # Maximum file size in MB +MAX_FILE_SIZE_BYTES = MAX_FILE_SIZE_MB * 1024 * 1024 + + +async def process_image_for_sticker(file_data: bytes, filename: Optional[str] = None) -> Optional[bytes]: + """ + Process an image file to meet Telegram sticker requirements. + + Args: + file_data: Raw image bytes from Telegram + filename: Original filename (optional, for format detection) + + Returns: + Processed image bytes in WebP format, or None if processing fails + """ + try: + # Open image from bytes + image = Image.open(io.BytesIO(file_data)) + + # Convert to RGBA if necessary (Telegram requires transparency support) + if image.mode != 'RGBA': + if image.mode == 'RGB': + # Add alpha channel + image = image.convert('RGBA') + elif image.mode in ('L', 'LA', 'P'): + # Grayscale or palette - convert to RGBA + image = image.convert('RGBA') + else: + # Any other mode - convert to RGBA + image = image.convert('RGBA') + + # Resize to Telegram sticker format (512px on one side) + image = resize_for_sticker(image) + + # Convert to WebP format with optimization + output = io.BytesIO() + + # Try different quality levels to meet file size requirements + quality = 95 + while quality >= 60: + output.seek(0) + output.truncate(0) + + image.save( + output, + format='WEBP', + quality=quality, + method=6, # Best compression method + exact=True # Preserve transparency + ) + + file_size = output.tell() + + if file_size <= MAX_FILE_SIZE_BYTES: + logger.info(f"Image processed: {image.size}, quality={quality}, size={file_size/1024:.1f}KB") + output.seek(0) + return output.getvalue() + + # Reduce quality and try again + quality -= 5 + logger.debug(f"File too large ({file_size/1024:.1f}KB), reducing quality to {quality}") + + # If we get here, even lowest quality is too large + logger.error(f"Could not compress image below {MAX_FILE_SIZE_MB}MB") + return None + + except Exception as e: + logger.error(f"Error processing image: {e}") + return None + + +def resize_for_sticker(image: Image.Image) -> Image.Image: + """ + Resize image to meet Telegram sticker requirements. + One side must be exactly 512px, other side <= 512px + + Args: + image: PIL Image object + + Returns: + Resized image + """ + width, height = image.size + + # Determine which side is longer + if width >= height: + # Landscape or square - set width to 512 + new_width = STICKER_SIZE + new_height = int((height / width) * STICKER_SIZE) + else: + # Portrait - set height to 512 + new_height = STICKER_SIZE + new_width = int((width / height) * STICKER_SIZE) + + # Ensure neither dimension exceeds 512 + new_width = min(new_width, STICKER_SIZE) + new_height = min(new_height, STICKER_SIZE) + + # Resize using LANCZOS for best quality + resized = image.resize((new_width, new_height), Image.Resampling.LANCZOS) + + logger.debug(f"Resized image from {width}x{height} to {new_width}x{new_height}") + return resized + + +async def validate_image(file_data: bytes) -> Tuple[bool, str]: + """ + Validate if file data is a valid image. + + Args: + file_data: Raw file bytes + + Returns: + Tuple of (is_valid, error_message) + """ + try: + image = Image.open(io.BytesIO(file_data)) + + # Check format + if image.format not in ('PNG', 'JPEG', 'JPG', 'WEBP', 'GIF', 'BMP', 'TIFF'): + return False, f"Unsupported image format: {image.format}" + + # Check if image can be processed + width, height = image.size + if width < 64 or height < 64: + return False, "Image too small (minimum 64x64 pixels)" + + # Check file size + if len(file_data) > 10 * 1024 * 1024: # 10MB limit for input + return False, "Image file too large (max 10MB for upload)" + + return True, "" + + except Exception as e: + return False, f"Invalid image file: {str(e)}" + + +async def get_image_info(file_data: bytes) -> dict: + """ + Get information about an image. + + Args: + file_data: Raw image bytes + + Returns: + Dictionary with image info + """ + try: + image = Image.open(io.BytesIO(file_data)) + return { + "format": image.format, + "mode": image.mode, + "size": image.size, + "width": image.width, + "height": image.height, + } + except Exception as e: + logger.error(f"Error getting image info: {e}") + return {} diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..260d5b8 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,6 @@ +python-telegram-bot==22.7 +python-dotenv==1.0.1 +aiosqlite==0.20.0 +Pillow==11.1.0 +rembg==2.0.60 +requests==2.32.3 \ No newline at end of file diff --git a/sticker_manager.py b/sticker_manager.py new file mode 100644 index 0000000..f9f8c05 --- /dev/null +++ b/sticker_manager.py @@ -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}") \ No newline at end of file