توسعه سریع و مدرن ربات و کلاینت در پیام‌رسان بله با پایتون

# نصب مستقیم از سورس یا پکیج
pip install aiobale-py

# یا نصب با ابزار سریع uv
uv pip install aiobale-py

import asyncio
from aiobale import Client, Dispatcher, F
from aiobale.types import Message
from aiobale.filters import IsText

# ساخت دیسپچر و کلاینت
dp = Dispatcher()
client = Client(dp, session_file="my_bot")

@dp.message(F.text == "/start")
async def start_handler(message: Message):
    await message.reply("سلام! به ربات Aiobale خوش آمدید.")

@dp.message(IsText())
async def echo_handler(message: Message):
    await message.reply(f"شما گفتید: {message.text}")

async def main():
    print("در حال اتصال به سرور بله...")
    await client.start()

if __name__ == "__main__":
    asyncio.run(main())

from aiobale import Client

# ۱. پاس دادن مستقیم شماره در کد (پرسش فقط برای کد تایید پیامکی)
client = Client(session_file="my_account", phone_number="09123456789")

# ۲. یا لاگین تعاملی کامل با پرسش شماره و کد در ترمینال
# client = Client(session_file="my_account")

if __name__ == "__main__":
    client.run()

from aiobale import Client

MY_JWT_TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

# راه‌اندازی با توکن مستقیم بدون نیاز به فایل سشن روی دیسک
client = Client(session_file=None, token=MY_JWT_TOKEN)

from aiobale import Router, F
from aiobale.types import Message

admin_router = Router()

@admin_router.message(F.text == "/panel")
async def admin_panel(msg: Message):
    await msg.reply("به پنل مدیریت خوش آمدید.")

from aiobale import Client, Dispatcher
from handlers.admin import admin_router

dp = Dispatcher()
# الحاق روتر ادمین به دیسپچر اصلی
dp.include_router(admin_router)

client = Client(dp, session_file="bot")

from aiobale import F
from aiobale.enums import ChatType

# تطابق دقیق متن
@dp.message(F.text == "ping")
async def on_ping(msg): await msg.reply("pong")

# شروع با پیشوند خاص
@dp.message(F.text.startswith("/calc"))
async def on_calc(msg): ...

# شرط روی نوع چت و فرستنده
@dp.message(F.chat.type == ChatType.GROUP, F.sender_id == 123456)
async def on_admin_in_group(msg): ...

# بررسی وجود سند یا تصویر
@dp.message(F.content.document)
async def on_document(msg): ...

from aiobale.filters import and_f, or_f, invert_f, IsText, IsPrivate

# پیام متنی باشد و در پیوی باشد
@dp.message(and_f(IsText(), IsPrivate()))
async def private_text_only(msg): ...

# پیام غیرمتنی باشد (معکوس کردن)
@dp.message(invert_f(IsText()))
async def media_only(msg): ...

from aiobale.filters.base import Filter
from aiobale.types import Message

class AdminOnly(Filter):
    def __init__(self, admin_ids: list[int]):
        self.admin_ids = admin_ids

    async def __call__(self, message: Message) -> bool:
        return message.sender_id in self.admin_ids

@dp.message(AdminOnly([111222, 333444]))
async def admin_handler(msg: Message):
    await msg.reply("دسترسی ادمین تایید شد.")

from aiobale import Client, Dispatcher, F
from aiobale.fsm import State, StatesGroup, FSMContext
from aiobale.types import Message

class OrderForm(StatesGroup):
    choosing_item = State()
    choosing_quantity = State()

dp = Dispatcher()

@dp.message(F.text == "/order")
async def start_order(msg: Message, state: FSMContext):
    await state.set_state(OrderForm.choosing_item)
    await msg.reply("نام محصول را وارد کنید:")

@dp.message(OrderForm.choosing_item)
async def item_chosen(msg: Message, state: FSMContext):
    await state.update_data(item_name=msg.text)
    await state.set_state(OrderForm.choosing_quantity)
    await msg.reply("تعداد مورد نظر را وارد کنید:")

@dp.message(OrderForm.choosing_quantity)
async def quantity_chosen(msg: Message, state: FSMContext):
    data = await state.get_data()
    item = data.get("item_name")
    await state.clear()
    await msg.reply(f"سفارش شما برای {msg.text} عدد {item} ثبت شد!")

from aiobale import Dispatcher, SQLiteStorage

# ذخیره دائمی وضعیت‌ها در دیتابیس SQLite
storage = SQLiteStorage("fsm_data.db")
dp = Dispatcher(storage=storage)

from aiobale import Dispatcher, BaseMiddleware
from typing import Any, Callable, Dict, Awaitable

class LoggingMiddleware(BaseMiddleware):
    async def __call__(
        self,
        handler: Callable[[Any, Dict[str, Any]], Awaitable[Any]],
        event: Any,
        data: Dict[str, Any]
    ) -> Any:
        print(f"دریافت رویداد: {event}")
        result = await handler(event, data)
        print("پردازش هندلر به اتمام رسید.")
        return result

dp = Dispatcher()
dp.middleware(LoggingMiddleware())

from aiobale.utils.keyboard import InlineKeyboardBuilder, ReplyKeyboardBuilder

# کیبورد شیشه‌ای (Inline)
builder = InlineKeyboardBuilder()
builder.button(text="🌐 وبسایت رسمی", url="https://aiobale.ir")
builder.button(text="دکمه ۱", callback_data="cb_1")
builder.button(text="دکمه ۲", callback_data="cb_2")
inline_kb = builder.as_markup(2, 1)  # سطر اول ۲ تا، سطر دوم ۱ دکمه

# کیبورد معمولی (Reply)
reply = ReplyKeyboardBuilder()
reply.button(text="👤 پروفایل کاربری")
reply.button(text="🛍️ فروشگاه")
reply_kb = reply.as_markup(2)

from aiobale import Command, CommandObject
from aiobale.types import Message

@dp.message(Command("ban", "kick", prefix="/!"))
async def admin_command(msg: Message, command: CommandObject):
    print(command.prefix)     # پیشوند استفاده شده مثلا '!'
    print(command.command)    # نام دستور مثلا 'ban'
    print(command.args)       # رشته کامل آرگومان‌ها
    print(command.args_list)  # لیست آرگومان‌ها به صورت تک‌تک
    await msg.reply(f"دستور {command.command} دریافت شد.")

from aiobale import CallbackData, F

class ProductCB(CallbackData, prefix="product"):
    action: str
    product_id: int

# تولید دیتای دکمه: product:view:99
data_str = ProductCB(action="view", product_id=99).pack()

@dp.message(ProductCB.filter(F.action == "view"))
async def on_view_product(event, callback_data: ProductCB):
    print(f"شناسه محصول: {callback_data.product_id}")

from aiobale import ClientManager, Dispatcher

dp = Dispatcher()
manager = ClientManager(dp)

manager.add_client(session_file="account1", phone_number="09121111111")
manager.add_client(session_file="account2", token="JWT_TOKEN_...")

if __name__ == "__main__":
    manager.run_all()

from aiobale import Client, Dispatcher, F, Conversation
from aiobale.types import Message

dp = Dispatcher()
client = Client(dp, session_file="my_bot")

@dp.message(F.text == "/start")
async def start_wizard(msg: Message, client: Client):
    # شروع یک مکالمه ایزوله برای این چت
    async with client.conversation(msg.chat.id, timeout=120) as conv:
        await conv.send_message("سلام! لطفاً نام و نام خانوادگی خود را وارد کنید:")
        
        # انتظار برای دریافت پاسخ بعدی از کاربر
        name_msg = await conv.get_response(timeout=60)
        
        await conv.send_message(f"خوش آمدی {name_msg.text}! حالا سن خود را وارد کن:")
        age_msg = await conv.get_response(timeout=60)
        
        await conv.send_message(
            f"ثبت‌نام شما تکمیل شد! ✅\nنام: {name_msg.text}\nسن: {age_msg.text}"
        )

if __name__ == "__main__":
    client.run()

from aiobale import Client
from aiobale.enums import ChatType

client = Client(session_file="backup_session")

async def read_chat_history(target_chat_id: int):
    print("در حال خواندن ۱۰۰ پیام اخیر چت...")
    
    async for msg in client.iter_messages(
        chat_id=target_chat_id,
        chat_type=ChatType.PRIVATE,
        limit=100,
        chunk_size=20
    ):
        if msg.text:
            print(f"[{msg.date}] کاربر {msg.sender_id}: {msg.text}")
        elif msg.document:
            print(f"[{msg.date}] فایل پیوست: {msg.document.name} (حجم: {msg.document.size} بایت)")

if __name__ == "__main__":
    import asyncio
    asyncio.run(read_chat_history(123456789))

# ارسال پیام متنی با متد مستقیم کلاینت
sent = await client.send_message(
    chat_id=123456789,
    text="سلام! این یک پیام آزمایشی است."
)

# ارسال عکس با کپشن
await client.send_photo(
    chat_id=123456789,
    file="./images/banner.png",
    caption="بنر جدید کانال"
)

# ویرایش پیام
await client.update_message(
    chat_id=123456789,
    message_id=sent.message_id,
    text="این متن ویرایش شد."
)

# پین کردن پیام
await client.pin_message(
    chat_id=123456789,
    message_id=sent.message_id
)

from aiobale import Client, Dispatcher, F
from aiobale.enums import ChatType, GivingType

dp = Dispatcher()
client = Client(dp, session_file="wallet_bot")

@dp.message(F.text == "/balance")
async def check_balance(msg):
    wallet_resp = await client.get_wallet()
    wallet = wallet_resp.wallet
    await msg.reply(
        f"💰 موجودی کیف‌پول شما: {wallet.balance:,} ریال\n"
        f"💳 شماره کارت متصل: {wallet.pan or 'نامشخص'}\n"
        f"🏦 شماره حساب: {wallet.account or 'نامشخص'}"
    )

@dp.message(F.text == "/send_gift")
async def send_gift_pack(msg):
    # ارسال عیدی ۱۰,۰۰۰ تومانی بین ۵ نفر به صورت تصادفی/شانسی
    await client.send_gift(
        chat_id=msg.chat.id,
        chat_type=msg.chat.type,
        amount=100000,
        message="عیدی اعضای گروه 🎁",
        gift_count=5,
        giving_type=GivingType.RANDOM
    )
    await msg.reply("بسته هدیه با موفقیت ارسال شد!")

@dp.message()
async def handler(msg: Message):
    # ریپلای کردن روی همین پیام
    await msg.reply("پاسخ در ریپلای")

    # ارسال پیام در همین چت بدون ریپلای
    await msg.answer("پیام مستقیم")

    # ارسال عکس در همین چت
    await msg.answer_photo("./photo.jpg", caption="توضیحات")

    # حذف همین پیام
    await msg.delete()

    # ویرایش متن پیام (در صورتی که توسط خود کلاینت ارسال شده باشد)
    await msg.edit_text("متن جدید")

    # پین کردن پیام
    await msg.pin()

import re
from aiobale import Client, Dispatcher, F
from aiobale.types import Message
from aiobale.enums import ChatType

dp = Dispatcher()
client = Client(dp, session_file="mod_bot")

LINK_PATTERN = re.compile(r"(https?://|ble\.ir/|t\.me/|eitaa\.com/)")

# حذف پیام‌های حاوی لینک تبلیغاتی در گروه‌ها
@dp.message(F.chat.type == ChatType.GROUP)
async def anti_link_guard(msg: Message, client: Client):
    if msg.text and LINK_PATTERN.search(msg.text):
        # حذف پیام متخلف
        await msg.delete()
        print(f"پیام حاوی لینک از کاربر {msg.sender_id} حذف شد.")

@dp.message(F.text.startswith("/ban"), F.chat.type == ChatType.GROUP)
async def ban_command(msg: Message, client: Client):
    if msg.replied_to:
        target_user = msg.replied_to.sender_id
        await client.kick_user(msg.chat.id, target_user)
        await msg.answer(f"کاربر {target_user} از گروه اخراج شد.")

if __name__ == "__main__":
    client.run()

from aiobale import Client, Dispatcher, F
from aiobale.types import Message

dp = Dispatcher()
client = Client(dp, session_file="media_bot")

@dp.message(F.content.document)
async def download_doc_handler(msg: Message, client: Client):
    doc = msg.content.document
    await msg.reply(f"در حال دریافت فایل {doc.name}...")
    
    # دانلود مستقیم فایل
    file_bytes = await client.download_file(
        file_id=doc.file_id,
        access_hash=doc.access_hash
    )
    
    await msg.reply(f"فایل با حجم {len(file_bytes)} بایت با موفقیت ذخیره شد.")

if __name__ == "__main__":
    client.run()

import asyncio
from aiobale import Client

CHANNELS = [100001, 100002, 100003]

async def broadcast_news(text: str):
    client = Client(session_file="sender")
    await client.start()

    for ch_id in CHANNELS:
        try:
            await client.send_message(chat_id=ch_id, text=text)
            print(f"پیام با موفقیت به {ch_id} ارسال شد.")
            await asyncio.sleep(1) # جلوگیری از ریت‌لیمیت
        except Exception as e:
            print(f"خطا در ارسال به {ch_id}: {e}")

    await client.stop()

if __name__ == "__main__":
    asyncio.run(broadcast_news("اطلاعیه رسمی جدید در کانال‌ها منتشر شد."))

from aiobale import Client, Dispatcher, Router, F
from aiobale.types import Message

user_router = Router()
admin_router = Router()

@user_router.message(F.text == "/help")
async def user_help(msg: Message):
    await msg.reply("راهنمای کاربر: دستورات مجاز /start و /about")

@admin_router.message(F.text == "/stats")
async def admin_stats(msg: Message, client: Client):
    dialogs = await client.load_dialogs(limit=50)
    await msg.reply(f"تعداد چت‌های فعال: {len(dialogs.chats)}")

dp = Dispatcher()
dp.include_router(user_router)
dp.include_router(admin_router)

client = Client(dp, session_file="modular_bot")

if __name__ == "__main__":
    client.run()