Bot::onCommand("ban {id}", function (Request $request, $id) {
  $request->banChatMember(
    chat_id: chat()->id,
    user_id: $id,
    until_date: now()->addDays(7)->timestamp
  );
})->can('administrator')->hasNotReply();

class Command:
    def execute(self, update, context):
        raise NotImplementedError

class StartCommand(Command):
    def execute(self, update, context):
        context.bot.send_message(chat_id=update.effective_chat.id, text="سلام! خوش آمدید.")

class HelpCommand(Command):
    def execute(self, update, context):
        context.bot.send_message(chat_id=update.effective_chat.id, text="لیست دستورات: /start, /help")

router = {
    '/start': StartCommand(),
    '/help': HelpCommand(),
}

def handle_update(update, context):
    text = update.message.text.split()[0]
    cmd = router.get(text)
    if cmd:
        cmd.execute(update, context)

from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("سلام از بات شما!")

app = ApplicationBuilder().token("YOUR_TOKEN").build()
app.add_handler(CommandHandler("start", start))
app.run_polling()

from flask import Flask, request
from telegram import Update
from telegram.ext import Application, CommandHandler

TOKEN = "YOUR_BOT_TOKEN"
app = Flask(__name__)

app_telegram = Application.builder().token(TOKEN).build()

async def start(update: Update, context):
    await update.message.reply_text("سلام! وب‌هوک فعاله ✅")

app_telegram.add_handler(CommandHandler("start", start))

@app.post("/webhook")
async def webhook():
    data = request.get_json(force=True)
    update = Update.de_json(data, app_telegram.bot)
    await app_telegram.process_update(update)
    return "ok"

if __name__ == "__main__":
    import asyncio
    from telegram import Bot

    bot = Bot(TOKEN)
    url = "https://example.com/webhook"  # آدرس HTTPS معتبر
    asyncio.run(bot.set_webhook(url=url))

    app.run(port=8443)

pip install python-telegram-bot --upgrade

from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters

TOKEN = "YOUR_BOT_TOKEN"

async def start(update: Update, context):
    await update.message.reply_text("سلام، یه پیام بفرست 👋")

async def echo(update: Update, context):
    await update.message.reply_text(update.message.text)

app = ApplicationBuilder().token(TOKEN).build()
app.add_handler(CommandHandler("start", start))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))
app.run_polling()

from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes

TOKEN = "YOUR_BOT_TOKEN"
WEBHOOK_URL = "https://example.com/bot/webhook"  # URL عمومی

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("ربات با Webhook فعاله ✅")

app = ApplicationBuilder().token(TOKEN).build()

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

app.run_webhook(
    listen="0.0.0.0",
    port=8443,
    url_path="bot/webhook",
    webhook_url=WEBHOOK_URL,
)

"""domain/entities.py"""
from dataclasses import dataclass

@dataclass
class User:
    id: int
    username: str | None
    is_admin: bool = False


"""domain/services.py"""
from .entities import User

class AccessService:
    def can_use_admin_panel(self, user: User) -> bool:
        return user.is_admin


"""application/use_cases.py"""
from .dto import MessageDTO
from domain.services import AccessService

class HandleAdminCommand:
    def __init__(self, access_service: AccessService):
        self._access = access_service

    def execute(self, msg: MessageDTO) -> str:
        if not self._access.can_use_admin_panel(msg.user):
            return "دسترسی غیرمجاز ❌"
        return "خوش اومدی ادمین 👋"


"""infrastructure/telegram_handlers.py"""
from aiogram import Router, types
from application.use_cases import HandleAdminCommand
from application.dto import MessageDTO

router = Router()

handle_admin = HandleAdminCommand(access_service=...)

@router.message(commands=["admin"])
async def admin_handler(message: types.Message):
    dto = MessageDTO(
        user_id=message.from_user.id,
        username=message.from_user.username,
        text=message.text,
    )
    reply_text = handle_admin.execute(dto)
    await message.answer(reply_text)

pip install pyTelegramBotAPI

import telebot
from telebot import types
from datetime import datetime

API_TOKEN = "YOUR_BOT_TOKEN_HERE"
bot = telebot.TeleBot(API_TOKEN)

@bot.message_handler(commands=["start"])
def send_welcome(message):
    keyboard = types.InlineKeyboardMarkup()
    btn_time = types.InlineKeyboardButton(
        text="⏰ دیدن زمان الان",
        callback_data="show_time"
    )
    keyboard.add(btn_time)

    bot.send_message(
        message.chat.id,
        "سلام! 👋\nبا دکمه زیر می‌تونی زمان فعلی سرور رو ببینی.",
        reply_markup=keyboard
    )

@bot.callback_query_handler(func=lambda c: c.data == "show_time")
def process_callback_show_time(callback_query):
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    bot.answer_callback_query(callback_query.id)
    bot.send_message(
        callback_query.message.chat.id,
        f"⏰ زمان فعلی: {now}"
    )

if __name__ == "__main__":
    print("Bot is running...")
    bot.infinity_polling()

from abc import ABC, abstractmethod

# Strategy interface
class CommandStrategy(ABC):
    @abstractmethod
    def handle(self, update, context):
        pass

# Concrete strategies
class StartCommand(CommandStrategy):
    def handle(self, update, context):
        update.message.reply_text("سلام 👋 خوش اومدی!")

class HelpCommand(CommandStrategy):
    def handle(self, update, context):
        update.message.reply_text("/start - شروع
/help - راهنما")

class UnknownCommand(CommandStrategy):
    def handle(self, update, context):
        update.message.reply_text("این دستور رو نمیشناسم 😅")

# Context / Dispatcher
class CommandDispatcher:
    def __init__(self):
        self._strategies = {
            "start": StartCommand(),
            "help": HelpCommand(),
        }
        self._fallback = UnknownCommand()

    def dispatch(self, update, context):
        cmd = update.message.text.split()[0].replace("/", "")
        strategy = self._strategies.get(cmd, self._fallback)
        strategy.handle(update, context)

pip install python-telegram-bot --upgrade

from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes

TOKEN = "YOUR_BOT_TOKEN"

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("سلام! 👋 ربات فعاله.")

app = ApplicationBuilder().token(TOKEN).build()
app.add_handler(CommandHandler("start", start))
app.run_polling()

pip install python-telegram-bot --upgrade

from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import (
    ApplicationBuilder,
    CommandHandler,
    CallbackQueryHandler,
    ContextTypes,
)

BOT_TOKEN = "YOUR_BOT_TOKEN_HERE"

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    keyboard = [
        [
            InlineKeyboardButton("📊 وضعیت سرور", callback_data="status"),
            InlineKeyboardButton("ℹ️ درباره ربات", callback_data="about"),
        ]
    ]
    reply_markup = InlineKeyboardMarkup(keyboard)

    await update.message.reply_text(
        "سلام 👋\nیکی از گزینه‌های زیر رو انتخاب کن:",
        reply_markup=reply_markup,
    )

async def button_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()

    if query.data == "status":
        text = "سرور الان آنلاینه ✅\nCPU: 23% | RAM: 58%"
    elif query.data == "about":
        text = "این یه ربات نمونه با inline keyboard هست 🤖"
    else:
        text = "گزینه نامعتبر"  # حالت پیش‌فرض

    await query.edit_message_text(text)

if __name__ == "__main__":
    app = ApplicationBuilder().token(BOT_TOKEN).build()

    app.add_handler(CommandHandler("start", start))
    app.add_handler(CallbackQueryHandler(button_handler))

    app.run_polling()

from flask import Flask, request
from telegram import Update
from telegram.ext import (
    Application,
    CommandHandler,
)

TOKEN = "YOUR_BOT_TOKEN"
app = Flask(__name__)

application = Application.builder().token(TOKEN).build()

async def start(update: Update, context):
    await update.message.reply_text("سلام 👋")

application.add_handler(CommandHandler("start", start))

@app.post("/webhook")
def webhook():
    update = Update.de_json(request.get_json(force=True), application.bot)
    application.update_queue.put_nowait(update)
    return "ok"

if __name__ == "__main__":
    # فقط یک بار اجرا کن (یا با curl) برای ست کردن webhook
    import requests
    url = "https://your-domain.com/webhook"
    requests.get(
        f"https://api.telegram.org/bot{TOKEN}/setWebhook",
        params={"url": url},
    )
    app.run(port=8443)

from flask import Flask, request
from telegram import Update
from telegram.ext import (
    Application,
    CommandHandler,
)

TOKEN = "YOUR_BOT_TOKEN"
WEBHOOK_URL = "https://example.com/telegram-webhook"

app = Flask(__name__)
application = Application.builder().token(TOKEN).build()

async def start(update: Update, context):
    await update.message.reply_text("سلام 👋")

application.add_handler(CommandHandler("start", start))

@app.post("/telegram-webhook")
def webhook():
    update = Update.de_json(request.get_json(force=True),
                            application.bot)
    application.update_queue.put_nowait(update)
    return "ok"

if __name__ == "__main__":
    import asyncio
    async def main():
        await application.bot.set_webhook(WEBHOOK_URL)
        app.run(host="0.0.0.0", port=8000)
    asyncio.run(main())

class DomainUser:
    def __init__(self, id, full_name, is_vip):
        self.id = id
        self.full_name = full_name
        self.is_vip = is_vip


# Gateway سیستم قدیمی (مثال REST خام)
class LegacyCrmClient:
    def get_user(self, chat_id: int) -> dict:
        # در عمل اینجا requests.get و ... دارید
        return {
            "contact_id": chat_id,
            "fname": "Ali",
            "lname": "Ahmadi",
            "segment": "gold",  # یا basic, silver
        }


# Anti-Corruption Layer
class CrmAcl:
    def __init__(self, client: LegacyCrmClient):
        self._client = client

    def get_domain_user(self, chat_id: int) -> DomainUser:
        raw = self._client.get_user(chat_id)

        full_name = f"{raw['fname']} {raw['lname']}"
        is_vip = raw["segment"] in ["gold", "platinum"]

        return DomainUser(
            id=raw["contact_id"],
            full_name=full_name,
            is_vip=is_vip,
        )


# استفاده در Handler ربات
crm_acl = CrmAcl(LegacyCrmClient())


def handle_start(update, context):
    user = crm_acl.get_domain_user(update.effective_chat.id)
    if user.is_vip:
        context.bot.send_message(
            chat_id=user.id,
            text=f"خوش اومدی {user.full_name} 🌟"
        )
    else:
        context.bot.send_message(
            chat_id=user.id,
            text=f"سلام {user.full_name}!"
        )

pip install python-telegram-bot==21.6

from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("سلام، ربات آنلاینه ✌️")

app = ApplicationBuilder().token("BOT_TOKEN").build()
app.add_handler(CommandHandler("start", start))
app.run_polling()

./run-web.sh

run-web.bat

import time
import requests

class CircuitBreaker:
    def __init__(self, failure_threshold=3, reset_timeout=10):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failures = 0
        self.state = 'CLOSED'
        self.last_failure_time = 0

    def call(self, func, fallback, *args, **kwargs):
        if self.state == 'OPEN':
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = 'HALF_OPEN'
            else:
                return fallback()

        try:
            result = func(*args, **kwargs)
            self.failures = 0
            self.state = 'CLOSED'
            return result
        except Exception:
            self.failures += 1
            self.last_failure_time = time.time()
            if self.failures >= self.failure_threshold:
                self.state = 'OPEN'
            return fallback()

cb = CircuitBreaker()

def call_ai_service(text):
    return requests.post('https://api.example.com/ai', json={'q': text}).json()

def ai_fallback():
    return {'answer': 'سرویس هوش مصنوعی موقتاً در دسترس نیست 😔'}

# داخل هندلر ربات
response = cb.call(call_ai_service, ai_fallback, message.text)
bot.send_message(chat_id, response['answer'])

pip install pyTelegramBotAPI

import telebot
from telebot import types

API_TOKEN = "YOUR_BOT_TOKEN_HERE"
bot = telebot.TeleBot(API_TOKEN)

# یک سوال نمونه
QUESTION = "کدام زبان برای توسعه ربات تلگرام در این مثال استفاده شده است؟"
OPTIONS = ["Java", "Python", "PHP", "Go"]
CORRECT_INDEX = 1  # Python

@bot.message_handler(commands=["start"])
def send_welcome(message):
    bot.reply_to(message, "سلام! برای شروع آزمون دستور /quiz رو بفرست ✅")

@bot.message_handler(commands=["quiz"])
def send_quiz(message):
    keyboard = types.InlineKeyboardMarkup()
    for i, option in enumerate(OPTIONS):
        btn = types.InlineKeyboardButton(
            text=option,
            callback_data=f"quiz_{i}"
        )
        keyboard.add(btn)

    bot.send_message(
        message.chat.id,
        QUESTION,
        reply_markup=keyboard
    )

@bot.callback_query_handler(func=lambda call: call.data.startswith("quiz_"))
def handle_quiz_answer(call):
    index = int(call.data.split("_")[1])

    if index == CORRECT_INDEX:
        text = "آفرین 🎉 جواب درست بود (Python)."
    else:
        correct_option = OPTIONS[CORRECT_INDEX]
        text = f"❌ نه! جواب درست: {correct_option}"

    # پاسخ به همون پیام
    bot.answer_callback_query(call.id, text, show_alert=True)

if __name__ == "__main__":
    print("Quiz bot is running...")
    bot.infinity_polling()