In Python, building AI-powered Telegram bots unlocks massive potential for image generation, processing, and automation—master this to create viral tools and ace full-stack interviews! 🤖

# Basic Bot Setup - The foundation (PTB v20+ Async)
from telegram.ext import Application, CommandHandler, MessageHandler, filters

async def start(update, context):
    await update.message.reply_text(
        "✨ AI Image Bot Active!\n"
        "/generate - Create images from text\n"
        "/enhance - Improve photo quality\n"
        "/help - Full command list"
    )

app = Application.builder().token("YOUR_BOT_TOKEN").build()
app.add_handler(CommandHandler("start", start))
app.run_polling()

# Image Generation - DALL-E Integration (OpenAI)
import openai
from telegram.ext import ContextTypes

openai.api_key = os.getenv("OPENAI_API_KEY")

async def generate(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if not context.args:
        await update.message.reply_text("❌ Usage: /generate cute robot astronaut")
        return
    
    prompt = " ".join(context.args)
    try:
        response = openai.Image.create(
            prompt=prompt,
            n=1,
            size="1024x1024"
        )
        await update.message.reply_photo(
            photo=response['data'][0]['url'],
            caption=f"🎨 Generated: *{prompt}*",
            parse_mode="Markdown"
        )
    except Exception as e:
        await update.message.reply_text(f"🔥 Error: {str(e)}")

app.add_handler(CommandHandler("generate", generate))

# Stable Diffusion - Self-Hosted Alternative
from stability_sdk import client
import stability_sdk.interfaces.gooseai.generation.generation_pb2 as generation

async def stable_diffusion(update, context):
    stability_api = client.StabilityInference(
        key=os.getenv('STABILITY_KEY'),
        verbose=True
    )
    
    answers = stability_api.generate(
        prompt=" ".join(context.args),
        seed=42,
        steps=50,
        cfg_scale=8.0,
        width=512, 
        height=512
    )
    
    for resp in answers:
        for artifact in resp.artifacts:
            if artifact.finish_reason == generation.FILTER:
                await update.message.reply_text("⚠️ NSFW content filtered")
            if artifact.type == generation.ARTIFACT_IMAGE:
                image = Image.open(io.BytesIO(artifact.binary))
                bio = io.BytesIO()
                image.save(bio, 'PNG')
                bio.seek(0)
                await update.message.reply_photo(photo=bio)

app.add_handler(CommandHandler("sd", stable_diffusion))

# Image Enhancement - Super Resolution
from realesrgan import RealESRGANer
from basicsr.archs.rrdbnet_arch import RRDBNet

async def enhance(update, context):
    photo_file = await update.message.photo[-1].get_file()
    await photo_file.download('input.jpg')
    
    # Initialize model (run once at startup)
    model = RRDBNet(num_in_ch=3, num_out_ch=3, num_feat=64, num_block=23, num_grow_ch=32, scale=4)
    upsampler = RealESRGANer(
        scale=4,
        model_path='realesrgan-x4plus.pth',
        model=model
    )
    
    # Process image
    output, _ = upsampler.enhance('input.jpg', outscale=4)
    cv2.imwrite('enhanced.png', output)
    
    await update.message.reply_photo(
        photo=open('enhanced.png', 'rb'),
        caption="🚀 4x Resolution Boost!"
    )

# Style Transfer - Artistic Transformation
import torch
import torchvision.transforms as transforms

async def style_transfer(update, context):
    # Download content image
    content_file = await update.message.photo[-1].get_file()
    await content_file.download('content.jpg')
    
    # Load models (initialized at startup)
    vgg = models.vgg19(pretrained=True).features.eval()
    style_model = StyleTransferNet()
    style_model.load_state_dict(torch.load('mosaic.pth'))
    
    # Process
    content = load_image('content.jpg')
    output = style_model(content)
    save_image('styled.jpg', output[0])
    
    await update.message.reply_photo(
        photo=open('styled.jpg', 'rb'),
        caption="🎨 Van Gogh Style Applied!"
    )

# Background Removal - Instant Cutout
from rembg import remove

async def remove_bg(update, context):
    photo = await update.message.photo[-1].get_file()
    await photo.download('input.png')
    
    with open('input.png', 'rb') as i:
        with open('output.png', 'wb') as o:
            o.write(remove(i.read()))
    
    await update.message.reply_document(
        document=open('output.png', 'rb'),
        caption="✂️ Transparent Background Ready!"
    )

# Text-to-Image with ControlNet - Precise Generation
from controlnet_aux import OpenposeDetector

async def pose_to_image(update, context):
    # 1. Get user's pose photo
    photo = await update.message.photo[-1].get_file()
    await photo.download('pose.jpg')
    
    # 2. Generate pose keypoints
    openpose = OpenposeDetector.from_pretrained()
    pose_image = openpose('pose.jpg')
    
    # 3. Generate image with pose control
    pipe = StableDiffusionControlNetPipeline.from_pretrained(
        "lllyasviel/sd-controlnet-openpose"
    )
    image = pipe(
        "cyberpunk cityscape", 
        image=pose_image
    ).images[0]
    
    image.save('result.png')
    await update.message.reply_photo(photo=open('result.png', 'rb'))

# Rate Limiting - Production Essential
from functools import wraps
from collections import defaultdict
import time

user_limits = defaultdict(list)

def rate_limit(calls=5, period=60):
    def decorator(func):
        @wraps(func)
        async def wrapper(update, context):
            now = time.time()
            user_id = update.effective_user.id
            
            # Clean old requests
            user_limits[user_id] = [t for t in user_limits[user_id] if now - t < period]
            
            if len(user_limits[user_id]) >= calls:
                await update.message.reply_text(
                    f"⏳ Too many requests! Try again in {period} seconds"
                )
                return
            
            user_limits[user_id].append(now)
            return await func(update, context)
        return wrapper
    return decorator

@rate_limit(calls=3, period=30)
async def generate(update, context):
    # Generation logic here
    pass

# Cloud Integration - Serverless Processing
import boto3
from telegram.ext import CallbackQueryHandler

async def cloud_generate(update, context):
    # Offload to AWS Lambda
    lambda_client = boto3.client('lambda')
    response = lambda_client.invoke(
        FunctionName='image-generator',
        Payload=json.dumps({"prompt": " ".join(context.args)})
    )
    
    # Get S3 URL from response
    payload = json.load(response['Payload'])
    await update.message.reply_photo(
        photo=payload['image_url'],
        reply_markup=InlineKeyboardMarkup([[
            InlineKeyboardButton("Download HD", url=payload['hd_url'])
        ]])
    )

app.add_handler(CommandHandler("cloud", cloud_generate))

# Async Queue System - Handle Traffic Spikes
import asyncio
from collections import deque

task_queue = deque()
active_tasks = 0
MAX_CONCURRENCY = 3

async def process_queue():
    global active_tasks
    while True:
        if task_queue and active_tasks < MAX_CONCURRENCY:
            update, context, func = task_queue.popleft()
            active_tasks += 1
            asyncio.create_task(execute_task(update, context, func))
        await asyncio.sleep(0.1)

async def execute_task(update, context, func):
    try:
        await func(update, context)
    finally:
        global active_tasks
        active_tasks -= 1

# Usage in command handlers
async def generate(update, context):
    task_queue.append((update, context, _generate_impl))

async def _generate_impl(update, context):
    # Actual generation logic
    pass

# Start queue processor
app.job_queue.run_repeating(lambda _: process_queue(), interval=0.1)

# Payment Integration - Monetize Your Bot
from telegram import LabeledPrice

async def subscribe(update, context):
    await context.bot.send_invoice(
        chat_id=update.effective_chat.id,
        title="AI Image Pro Plan",
        description="500 credits for image generation",
        payload="subscription_payload",
        provider_token=os.getenv("PAYMENT_TOKEN"),
        currency="USD",
        prices=[LabeledPrice("500 credits", 990)],  # $9.90
        start_parameter="credit_purchase"
    )

async def precheckout_check(update, context):
    query = update.pre_checkout_query
    if query.total_amount != 990:
        await query.answer(ok=False, error_message="Wrong amount!")
    else:
        await query.answer(ok=True)

app.add_handler(PreCheckoutQueryHandler(precheckout_check))

# User Analytics - Track Engagement
import firebase_admin
from firebase_admin import firestore

# Initialize at startup
firebase_admin.initialize_app()
db = firestore.client()

async def track_command(update, context):
    db.collection('commands').add({
        'user_id': update.effective_user.id,
        'command': context.args[0] if context.args else 'start',
        'timestamp': firestore.SERVER_TIMESTAMP
    })

app.add_handler(MessageHandler(filters.COMMAND, track_command))

# Docker Deployment - Production Ready
# Dockerfile snippet:
# FROM python:3.10-slim
# RUN pip install python-telegram-bot openai stability-sdk
# COPY bot.py /app/
# ENV BOT_TOKEN=${BOT_TOKEN}
# CMD ["python", "/app/bot.py"]

# docker-compose.yml
# version: '3'
# services:
#   bot:
#     build: .
#     environment:
#       - BOT_TOKEN=your_token
#       - OPENAI_API_KEY=${OPENAI_KEY}
#     restart: always

# Error Monitoring - Critical for reliability
import sentry_sdk
from sentry_sdk.integrations.telegram import TelegramIntegration

sentry_sdk.init(
    dsn=os.getenv("SENTRY_DSN"),
    integrations=[TelegramIntegration()],
    traces_sample_rate=1.0
)

# Auto-capture all exceptions
async def error_handler(update, context):
    await update.message.reply_text(
        "🚨 System error! Our engineers are notified."
    )
    return

app.add_error_handler(error_handler)

# Media Group Processing - Handle albums
async def process_album(update, context):
    media_group = update.message.media_group_id
    if not media_group:
        return
    
    # Collect all photos in group
    if media_group not in context.bot_data:
        context.bot_data[media_group] = []
    
    context.bot_data[media_group].append(update.message.photo[-1])
    
    # Process when complete (simplified)
    if len(context.bot_data[media_group]) == update.message.media_group_length:
        photos = context.bot_data.pop(media_group)
        # Process all photos together
        await merge_photos(photos, update)

app.add_handler(MessageHandler(filters.PHOTO, process_album))

# Advanced Feature: Image-to-Prompt
from transformers import AutoProcessor, BlipForConditionalGeneration

async def describe_image(update, context):
    photo = await update.message.photo[-1].get_file()
    await photo.download('input.jpg')
    
    processor = AutoProcessor.from_pretrained("Salesforce/blip-image-captioning-base")
    model = BlipForConditionalGeneration.from_pretrained("Salesforce/blip-image-captioning-base")
    
    raw_image = Image.open('input.jpg').convert('RGB')
    inputs = processor(images=raw_image, return_tensors="pt")
    outputs = model.generate(**inputs, max_new_tokens=50)
    caption = processor.decode(outputs[0], skip_special_tokens=True)
    
    await update.message.reply_text(
        f"📝 Prompt suggestion:\n`{caption}`",
        parse_mode="Markdown"
    )

# Interview Power Move: Real-time Progress Updates
async def long_generation(update, context):
    msg = await update.message.reply_text("⏳ Starting generation...")
    
    # Simulate long process
    for i in range(1, 6):
        await asyncio.sleep(2)
        await msg.edit_text(f"⏳ Processing... {i*20}%")
    
    # Send final image
    await msg.edit_text("✅ Generation complete!", reply_markup=...)

# Pro Tip: Optimize Image Delivery
async def optimized_send(update, context):
    # Generate WebP instead of PNG
    image.save('output.webp', 'WEBP', quality=85)
    
    # Compress before sending
    await update.message.reply_document(
        document=open('output.webp', 'rb'),
        filename='result.webp',
        caption="⚡ 70% smaller file size!"
    )

# Real-World Case Study: E-commerce Product Visualizer
async def product_visualize(update, context):
    """Generate product mockups from user photos"""
    # 1. Get product photo
    photo = await update.message.photo[-1].get_file()
    await photo.download('product.jpg')
    
    # 2. Remove background
    rembg.remove_file('product.jpg', 'product.png')
    
    # 3. Generate mockups
    mockups = [
        generate_mockup('product.png', 't-shirt'),
        generate_mockup('product.png', 'mug'),
        generate_mockup('product.png', 'phone_case')
    ]
    
    # 4. Send as album
    media = [InputMediaPhoto(media=open(m, 'rb')) for m in mockups]
    await update.message.reply_media_group(media=media)
    
    # 5. Offer download link
    await update.message.reply_text(
        "📥 Download high-res versions: [link]",
        reply_markup=InlineKeyboardMarkup([[
            InlineKeyboardButton("Download ZIP", url=generate_zip(mockups))
        ]])
    )

By: @husseinsheikho 🤯

#Python #TelegramBot #AI #ImageGeneration #StableDiffusion #OpenAI #MachineLearning #CodingInterview #FullStack #Chatbots #DeepLearning #ComputerVision #Programming #TechJobs #DeveloperTips #CareerGrowth #CloudComputing #Docker #APIs #Python3 #Productivity #TechTips