# script was programmed by: Kodo
# For any inquiries: @DevKodo
# Data source and operation: @SourceKodo
import asyncio
import logging
import aiohttp
from datetime import datetime, timedelta
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from aiogram import Bot, Dispatcher, Router, F
from aiogram.types import (
Message, CallbackQuery, Update,
InlineKeyboardMarkup, InlineKeyboardButton,
ChatMemberUpdated,
)
from aiogram.filters import CommandStart
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
from aiohttp import web
from sqlalchemy import select
from urllib.parse import quote
import sys, os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from config import (
FACTORY_TOKEN, WEBHOOK_URL, WEBHOOK_PATH, WEBHOOK_SECRET,
WEBAPP_HOST, WEBAPP_PORT, BOT_TYPES, DEVELOPER_IDS
)
from information.database import (
init_db, get_or_create_user, get_user, get_settings,
get_help_text,
get_start_template_with_entities, get_token_template_with_entities,
update_settings, get_session, get_all_buttons, get_button,
get_mandatory_channels, get_funded_channels,
get_payment_packages, get_payment_settings,
get_user_balance, add_user_balance, deduct_user_balance,
is_payment_processed, record_payment,
get_bot_subscription, create_bot_subscription, expire_bot_subscription,
get_user_paid_bots, get_user_unpaid_bots, is_bot_paid,
get_expiring_subscriptions, get_bot_by_id,
get_factory_cobot_mandatory_channels,
get_factory_cobot_funded_channels, record_factory_cobot_funded_sub,
get_factory_cobot_funded_count, mark_factory_cobot_funded_complete,
mark_factory_funded_pending, is_factory_funded_pending,
clear_factory_funded_pending, clear_all_factory_funded_pending,
)
from sqlalchemy import select, func
from information.models import (
Bot as BotModel, BotTransfer, TransferStatus, BotType, User,
MandatoryChannel, FundedChannel, FundedSubscription,
CobotMandatoryChannel, CobotFundedChannel, CobotFundedSubscription,
ChannelDisplayType, BotGroup, ButtonSetting, CobotUser,
BotSubscription,
)
from dev import dev_router, notify_devs, is_dev_or_admin_db, btn_label, make_inline_button
from aiogram.fsm.storage.memory import MemoryStorage as CobotMemoryStorage
cobot_storages: dict[str, CobotMemoryStorage] = {}
cobot_dispatchers: dict[str, Dispatcher] = {}
def get_cobot_storage(bot_username: str) -> CobotMemoryStorage:
if bot_username not in cobot_storages:
cobot_storages[bot_username] = CobotMemoryStorage()
return cobot_storages[bot_username]
def get_cobot_dispatcher(bot_username: str) -> Dispatcher:
if bot_username not in cobot_dispatchers:
import sys
import os
factory_dir = os.path.dirname(os.path.abspath(__file__))
if factory_dir not in sys.path:
sys.path.insert(0, factory_dir)
from cobots.communication.com import router as com_router
from cobots.communication.com import dev_router as com_dev_router
storage = get_cobot_storage(bot_username)
dp = Dispatcher(storage=storage)
dp.include_router(com_dev_router)
dp.include_router(com_router)
cobot_dispatchers[bot_username] = dp
return cobot_dispatchers[bot_username]
_com_dispatcher: Dispatcher | None = None
def get_com_dispatcher() -> Dispatcher:
global _com_dispatcher
if _com_dispatcher is None:
from cobots.communication.com import router as com_router
from cobots.communication.com import dev_router as com_dev_router
storage = MemoryStorage()
dp = Dispatcher(storage=storage)
dp.include_router(com_dev_router)
dp.include_router(com_router)
_com_dispatcher = dp
return _com_dispatcher
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger(__name__)
# قاموس أقفال لكل نوع بوت - يمنع تداخل معالجة بوتات النوع الواحد
# (لأنها تتشارك نفس الموديول ومتغيراته العامة)
_BOT_TYPE_LOCKS: dict = {}
router = Router()
class CreateBotStates(StatesGroup):
waiting_token = State()
class TransferStates(StatesGroup):
waiting_receiver_id = State()
class EditBotStates(StatesGroup):
waiting_new_name = State()
waiting_new_photo = State()
class PaymentStates(StatesGroup):
waiting_subscription_bot = State()
class BotOwnerStates(StatesGroup):
waiting_custom_owner_id = State()
def is_developer(user_id: int) -> bool:
return user_id in DEVELOPER_IDS
async def validate_token(token: str) -> dict | None:
try:
async with aiohttp.ClientSession() as session:
async with session.get(
f"https://api.telegram.org/bot{token}/getMe",
timeout=aiohttp.ClientTimeout(total=10)
) as resp:
data = await resp.json()
if data.get("ok"):
return data["result"]
except Exception:
pass
return None
async def create_bot_database(token: str, username: str, bot_type: str, owner_id: int):
try:
if bot_type == "communication":
from cobots.communication.db import init_db as com_init_db
await com_init_db(username.lower())
logger.info(f"✅ تم إنشاء قاعدة بيانات بوت التواصل: {username}")
elif bot_type == "download":
from cobots.download.db import init_db as down_init_db
await down_init_db(username.lower())
logger.info(f"✅ تم إنشاء قاعدة بيانات بوت التحميل: {username}")
elif bot_type == "decor":
from cobots.decor.db import init_db as decor_init_db
await decor_init_db(username.lower())
logger.info(f"✅ تم إنشاء قاعدة بيانات بوت الزخرفة: {username}")
elif bot_type == "factory":
from cobots.factory.db import init_db as factory_init_db
await factory_init_db(username.lower())
logger.info(f"✅ تم إنشاء قاعدة بيانات بوت المصنع: {username}")
elif bot_type == "transcoding":
from cobots.transcoding.db import init_db as trans_init_db
await trans_init_db(username.lower())
logger.info(f"✅ تم إنشاء قاعدة بيانات بوت تحويل الصيغ: {username}")
elif bot_type == "translation":
from cobots.translation.db import init_db as trans_init_db
await trans_init_db(username.lower())
logger.info(f"✅ تم إنشاء قاعدة بيانات بوت الترجمة: {username}")
elif bot_type == "ai":
from cobots.ai.db import init_db as ai_init_db
await ai_init_db(username.lower())
logger.info(f"✅ تم إنشاء قاعدة بيانات بوت الذكاء الاصطناعي: {username}")
elif bot_type == "buttons":
from cobots.buttons.db import init_db as buttons_init_db
await buttons_init_db(username.lower())
logger.info(f"✅ تم إنشاء قاعدة بيانات بوت الأزرار: {username}")
except Exception as e:
logger.error(f"❌ خطأ في إنشاء قاعدة البيانات: {e}")
async def set_bot_webhook(token: str, bot_username: str) -> bool:
webhook_url = f"https://mahde.publicvm.com/kodo/bots/{bot_username.lower()}"
secret = f"kodo_{bot_username.lower()}_secret"
try:
async with aiohttp.ClientSession() as session:
async with session.post(
f"https://api.telegram.org/bot{token}/setWebhook",
json={
"url": webhook_url,
"secret_token": secret,
"allowed_updates": ["message", "callback_query", "edited_message", "my_chat_member", "managed_bot"],
},
timeout=aiohttp.ClientTimeout(total=10)
) as resp:
data = await resp.json()
return data.get("ok", False)
except Exception:
return False
async def delete_bot_webhook(token: str) -> bool:
try:
async with aiohttp.ClientSession() as session:
async with session.post(
f"https://api.telegram.org/bot{token}/deleteWebhook",
timeout=aiohttp.ClientTimeout(total=10)
) as resp:
data = await resp.json()
return data.get("ok", False)
except Exception:
return False
async def check_user_subscriptions(bot: Bot, user_id: int):
channels = await get_mandatory_channels()
not_sub_buttons = []
not_sub_text = []
for ch in channels:
try:
member = await bot.get_chat_member(ch.channel_id, user_id)
if member.status in ("left", "kicked", "banned"):
raise Exception
except Exception:
if ch.display_type == ChannelDisplayType.buttons:
not_sub_buttons.append(ch)
else:
not_sub_text.append(ch)
return not_sub_buttons, not_sub_text
async def check_funded_subscriptions(bot: Bot, user_id: int) -> list:
channels = await get_funded_channels()
not_subbed = []
for ch in channels:
try:
member = await bot.get_chat_member(ch.channel_id, user_id)
if member.status in ("left", "kicked", "banned"):
raise Exception
except Exception:
not_subbed.append(ch)
return not_subbed
async def record_funded_subscription(bot: Bot, user_id: int, channel: FundedChannel):
async with get_session() as session:
result = await session.execute(
select(FundedSubscription).where(
FundedSubscription.funded_channel_id == channel.id,
FundedSubscription.user_id == user_id,
)
)
if result.scalar_one_or_none():
return
session.add(FundedSubscription(
funded_channel_id = channel.id,
user_id = user_id,
))
ch_result = await session.execute(
select(FundedChannel).where(FundedChannel.id == channel.id)
)
ch = ch_result.scalar_one_or_none()
if ch:
ch.current_count += 1
settings = await get_settings()
if settings.notify_funded_sub:
user_result = await session.execute(select(User).where(User.id == user_id))
user = user_result.scalar_one_or_none()
name = ""
if user:
if user.first_name:
name = user.first_name
if user.username:
name = f"@{user.username}"
name = name or str(user_id)
link = ch.invite_link or (
f"https://t.me/{ch.channel_username}" if ch.channel_username else str(ch.channel_id)
)
await notify_devs(
bot,
f"🔔 اشعار اشتراك جديد في قناة ممولة:\n"
f"الحساب: {name}\n"
f"القناة الممولة: {ch.channel_title}\n"
f"عدد المشتركين عبر البوت حالياً: {ch.current_count} عضو\n"
f"عدد الأعضاء المطلوب: {ch.target_count}"
)
if ch.current_count >= ch.target_count:
ch.is_active = False
ch.completed_at = datetime.utcnow()
link = ch.invite_link or (
f"https://t.me/{ch.channel_username}" if ch.channel_username else str(ch.channel_id)
)
await notify_devs(
bot,
f"✅ اكتملت قناة ممولة!\n"
f"القناة: {ch.channel_title}\n"
f"تم الوصول إلى {ch.target_count} عضو وتم إزالتها تلقائياً."
)
async def handle_subscription_check(bot: Bot, user_id: int, msg_or_cb) -> bool:
is_cb = isinstance(msg_or_cb, CallbackQuery)
send = msg_or_cb.message.answer if is_cb else msg_or_cb.answer
not_btn, not_txt = await check_user_subscriptions(bot, user_id)
if not_btn:
ch = not_btn[0]
link = ch.invite_link or (f"https://t.me/{ch.channel_username}" if ch.channel_username else "#")
await send(
"عزيزي المستخدم اشترك في القنوات التالية قبل استخدام البوت:",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text=f"@{ch.channel_username or ch.channel_title}", url=link)],
[InlineKeyboardButton(text="✅ تحقق من الاشتراك", callback_data="check_sub")],
])
)
return False
if not_txt:
ch = not_txt[0]
link = ch.invite_link or (f"https://t.me/{ch.channel_username}" if ch.channel_username else "#")
await send(
f"عزيزي المستخدم اشترك في القناة التالية قبل استخدام البوت:\n"
f"- {ch.channel_title}\n\nاشترك ثم أرسل /start",
parse_mode="HTML", disable_web_page_preview=True)
return False
not_funded = await check_funded_subscriptions(bot, user_id)
if not_funded:
ch = not_funded[0]
link = ch.invite_link or (f"https://t.me/{ch.channel_username}" if ch.channel_username else "#")
if ch.display_type == ChannelDisplayType.buttons:
await send(
"عزيزي المستخدم اشترك في القناة التالية قبل استخدام البوت:",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text=f"@{ch.channel_username or ch.channel_title}", url=link)],
[InlineKeyboardButton(text="✅ تحقق من الاشتراك", callback_data="check_sub")],
])
)
else:
await send(
f"عزيزي المستخدم اشترك في القناة التالية قبل استخدام البوت:\n"
f"- {ch.channel_title}\n\nاشترك ثم أرسل /start",
parse_mode="HTML", disable_web_page_preview=True)
return False
channels = await get_funded_channels()
for ch in channels:
try:
member = await bot.get_chat_member(ch.channel_id, user_id)
if member.status not in ("left", "kicked", "banned"):
await record_funded_subscription(bot, user_id, ch)
except Exception:
pass
return True
async def _dyn_btn(key: str, cb: str, fallback_text: str = None) -> InlineKeyboardButton:
from dev import btn_label, make_inline_button
buttons = await get_all_buttons()
btn_map = {b.button_key: b for b in buttons}
b = btn_map.get(key)
if b and b.is_visible:
return make_inline_button(text=btn_label(b), callback_data=cb, style=b.color, emoji_id=b.custom_emoji_id)
return InlineKeyboardButton(text=fallback_text or key, callback_data=cb)
async def _back_btn(cb: str = "menu:back") -> InlineKeyboardButton:
return await _dyn_btn("back", cb, "رجوع")
async def build_main_keyboard() -> InlineKeyboardMarkup:
settings = await get_settings()
buttons = await get_all_buttons()
btn_map = {b.button_key: b for b in buttons}
def make_btn(key: str, callback: str = None, url_override: str = None):
b = btn_map.get(key)
if not b or not b.is_visible:
return None
label = btn_label(b)
url = b.url or url_override
return make_inline_button(
text = label,
callback_data = None if url else (callback or f"menu:{key}"),
url = url,
style = b.color,
emoji_id = b.custom_emoji_id,
)
rows = []
r1 = [x for x in [make_btn("create", "menu:create")] if x]
if r1: rows.append(r1)
r2 = [x for x in [make_btn("mybots", "menu:mybots"), make_btn("help", "menu:help")] if x]
if r2: rows.append(r2)
r3 = [x for x in [make_btn("paid_create", "menu:paid_create")] if x]
if r3: rows.append(r3)
r3b = [x for x in [make_btn("subscription", "menu:subscription")] if x]
if r3b: rows.append(r3b)
r4 = [
x for x in [
make_btn("bots_channel", url_override=settings.bots_channel_url),
make_btn("source_channel", url_override=settings.source_channel_url),
] if x
]
if r4: rows.append(r4)
r5 = [x for x in [make_btn("developer", url_override=settings.developer_url)] if x]
if r5: rows.append(r5)
return InlineKeyboardMarkup(inline_keyboard=rows)
async def build_bot_types_keyboard(prefix: str = "create") -> InlineKeyboardMarkup:
from config import BOT_TYPES
buttons = await get_all_buttons()
btn_map = {b.button_key: b for b in buttons}
rows = []
for bot_key, default_name in BOT_TYPES.items():
b = btn_map.get(bot_key)
if b and not b.is_visible:
continue
label = btn_label(b) if b else default_name
style = b.color if b else "default"
emoji_id = b.custom_emoji_id if b else None
rows.append([make_inline_button(
text = label,
callback_data = f"{prefix}:type:{bot_key}",
style = style,
emoji_id = emoji_id,
)])
back_btn = btn_map.get("back")
if back_btn and back_btn.is_visible:
rows.append([make_inline_button(
text = btn_label(back_btn),
callback_data = "menu:back",
style = back_btn.color,
emoji_id = back_btn.custom_emoji_id,
)])
else:
rows.append([await _back_btn("menu:back")])
return InlineKeyboardMarkup(inline_keyboard=rows)
async def kb_cancel(back_cb: str = "menu:back") -> InlineKeyboardMarkup:
buttons = await get_all_buttons()
btn_map = {b.button_key: b for b in buttons}
back_b = await build_dynamic_button("back", btn_map, callback_data=back_cb)
return InlineKeyboardMarkup(inline_keyboard=[
[back_b or InlineKeyboardButton(text="إلغاء", callback_data=back_cb)]
])
async def manager_bot_username(bot: Bot) -> str:
try:
me = await bot.me()
if me and me.username:
return me.username
except Exception as e:
logger.warning(f"تعذّر جلب اسم البوت من getMe: {e}")
return (getattr(bot, "_me", None) and bot._me.username) or ""
def managed_bot_url(manager_username: str, suggested_username: str = "", suggested_name: str = "") -> str:
url = f"https://t.me/newbot/{manager_username}"
if suggested_username:
url = f"{url}/{suggested_username}"
if suggested_name:
return f"{url}?name={quote(suggested_name)}"
return url
async def kb_token_request(back_cb: str = "menu:back", manager_username: str = "",
suggested_username: str = "", suggested_name: str = "") -> InlineKeyboardMarkup:
cancel_kb = await kb_cancel(back_cb)
rows = [[InlineKeyboardButton(
text = "إنشاء بوت فوري",
url = managed_bot_url(manager_username, suggested_username, suggested_name),
style = "success",
)]]
rows.extend(cancel_kb.inline_keyboard)
return InlineKeyboardMarkup(inline_keyboard=rows)
async def build_mybots_keyboard(bots: list, btn_map: dict) -> InlineKeyboardMarkup:
buttons = []
for bot in bots:
b = btn_map.get("bot_url")
style = b.color if b else "default"
emoji_id = b.custom_emoji_id if b else None
bot_url_btn = make_inline_button(
text = f"@{bot.username}",
url = f"https://t.me/{bot.username}",
style = style,
emoji_id = emoji_id,
)
manage_btn = await build_dynamic_button(
"bot_manage", btn_map,
callback_data=f"bot:manage:{bot.id}"
)
row = [x for x in [bot_url_btn, manage_btn] if x]
if not row:
row = [
InlineKeyboardButton(text=f"@{bot.username}", url=f"https://t.me/{bot.username}"),
await _dyn_btn("bot_manage", f"bot:manage:{bot.id}", "إدارة"),
]
buttons.append(row)
back_b = await build_dynamic_button("back", btn_map, callback_data="menu:back")
buttons.append([back_b or InlineKeyboardButton(text="رجوع", callback_data="menu:back")])
return InlineKeyboardMarkup(inline_keyboard=buttons)
async def build_manage_bot_keyboard(bot_id: int, btn_map: dict) -> InlineKeyboardMarkup:
del_btn = await build_dynamic_button(
"bot_delete", btn_map,
callback_data=f"bot:delete_ask:{bot_id}"
)
back_b = await build_dynamic_button("back", btn_map, callback_data="menu:mybots")
transfer_b = await build_dynamic_button("bot_transfer", btn_map, callback_data=f"bot:transfer:{bot_id}")
edit_photo = await build_dynamic_button("bot_edit_photo", btn_map, callback_data=f"bot:edit_photo:{bot_id}")
edit_name = await build_dynamic_button("bot_edit_name", btn_map, callback_data=f"bot:edit_name:{bot_id}")
change_type= await build_dynamic_button("bot_change_type",btn_map, callback_data=f"bot:change_type:{bot_id}")
return InlineKeyboardMarkup(inline_keyboard=[
[transfer_b or InlineKeyboardButton(text="نقل البوت", callback_data=f"bot:transfer:{bot_id}")],
[del_btn or InlineKeyboardButton(text="حذف البوت", callback_data=f"bot:delete_ask:{bot_id}")],
[
edit_photo or InlineKeyboardButton(text="تغيير الصورة", callback_data=f"bot:edit_photo:{bot_id}"),
edit_name or InlineKeyboardButton(text="تغيير الاسم", callback_data=f"bot:edit_name:{bot_id}"),
],
[change_type or InlineKeyboardButton(text="تغيير نوع البوت", callback_data=f"bot:change_type:{bot_id}")],
[back_b or InlineKeyboardButton(text="رجوع", callback_data="menu:mybots")],
])
async def kb_transfer_receiver(transfer_id: int) -> InlineKeyboardMarkup:
buttons = await get_all_buttons()
btn_map = {b.button_key: b for b in buttons}
accept_b = await build_dynamic_button("transfer_accept", btn_map, callback_data=f"transfer:accept:{transfer_id}")
reject_b = await build_dynamic_button("transfer_reject", btn_map, callback_data=f"transfer:reject:{transfer_id}")
return InlineKeyboardMarkup(inline_keyboard=[
[accept_b or await _dyn_btn("transfer_accept", f"transfer:accept:{transfer_id}", "موافقة")],
[reject_b or await _dyn_btn("transfer_reject", f"transfer:reject:{transfer_id}", "رفض")],
])
async def kb_transfer_confirm(transfer_id: int) -> InlineKeyboardMarkup:
buttons = await get_all_buttons()
btn_map = {b.button_key: b for b in buttons}
confirm_b = await build_dynamic_button("transfer_confirm", btn_map, callback_data=f"transfer:confirm:{transfer_id}")
cancel_b = await build_dynamic_button("back", btn_map, callback_data=f"transfer:cancel:{transfer_id}")
return InlineKeyboardMarkup(inline_keyboard=[
[confirm_b or await _dyn_btn("transfer_confirm", f"transfer:confirm:{transfer_id}", "تأكيد")],
[cancel_b or await _back_btn(f"transfer:cancel:{transfer_id}")],
])
async def build_start_text(user) -> tuple:
import json as _json
settings = await get_settings()
template = settings.start_template
entities_json = getattr(settings, "start_template_entities", None)
if template:
text = template.replace("{first_name}", user.first_name)
text = text.replace("{username}", f"@{user.username}" if user.username else user.first_name)
entities = None
if entities_json:
try:
entities = _json.loads(entities_json)
except Exception:
entities = None
return text, entities
return (
f"مرحباً {user.first_name} 👋\n"
"أهلاً بك في KODO Factory — مصنع البوتات الاحترافي!\n\n"
"اختر من القائمة أدناه:"
), None
@router.message(CommandStart())
async def cmd_start(message: Message, state: FSMContext, bot: Bot):
await state.clear()
args = message.text.split(None, 1)[1] if message.text and len(message.text.split()) > 1 else ""
if args == "subscription":
user = message.from_user
await get_or_create_user(user_id=user.id, username=user.username, first_name=user.first_name)
balance = await get_user_balance(user.id)
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
await message.answer(
f"💳 قائمة الاشتراك\n\nرصيد حسابك: {_fmt_usd(balance)}",
parse_mode="HTML",
disable_web_page_preview=True,
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text=f"💰 رصيد حسابك: {_fmt_usd(balance)}", callback_data="pay:noop")],
[await _dyn_btn("pay_topup", "pay:topup", "إضافة رصيد 💳"),
await _dyn_btn("pay_subscribe", "pay:subscribe", "اشتراك مدفوع ⭐")],
[await _dyn_btn("pay_paid_bots", "pay:paid_bots", "البوتات المدفوعة 💎")],
[InlineKeyboardButton(text="تواصل مع الدعم 📞", url="https://t.me/n_u_7")],
[await _back_btn("menu:back")],
])
)
return
user = message.from_user
db_user = await get_or_create_user(
user_id = user.id,
username = user.username,
first_name = user.first_name,
)
settings = await get_settings()
if settings.notify_new_users:
from datetime import timedelta
if (datetime.utcnow() - db_user.joined_at).total_seconds() < 10:
name = f"@{user.username}" if user.username else user.first_name
await notify_devs(
bot,
f"👤 مستخدم جديد دخل المصنع!\n"
f"الاسم: {user.first_name}\n"
f"المعرف: {name}\n"
f"الـ ID: {user.id}",
)
if db_user.is_banned:
await message.answer("⛔ أنت محظور من استخدام هذا البوت.")
return
if not settings.is_open and not (db_user.is_dev or db_user.is_admin):
msg = settings.closed_message or "🔴 البوت مقفول حالياً، يرجى المحاولة لاحقاً."
await message.answer(msg)
return
all_subbed = await handle_subscription_check(bot, user.id, message)
if not all_subbed:
return
start_text, start_entities = await build_start_text(user)
keyboard = await build_main_keyboard()
if db_user.is_dev or db_user.is_admin or is_developer(user.id):
from dev import kb_dev_panel
dev_keyboard = await kb_dev_panel()
await message.answer(
"» هذه هي لوحة التحكم وادارة مصنع البوتات الخاص بك.\n\n» لوحة تحكم متكاملة تتيح لك الإشراف على صلاحيات البوت، متابعة الإحصائيات وإدارة الاشتراكات وكل شيء من مكان واحد.\n\n» قم بأختيار الاجراء المطلوب من القائمة ادناه:", parse_mode="HTML",
reply_markup=dev_keyboard
)
import json as _json
from aiogram.types import MessageEntity
send_kwargs = dict(text=start_text, reply_markup=keyboard)
if start_entities:
try:
ents = [MessageEntity(type=e["type"], offset=e["offset"], length=e["length"],
custom_emoji_id=e.get("custom_emoji_id")) for e in start_entities]
send_kwargs["entities"] = ents
except Exception:
send_kwargs["parse_mode"] = "HTML"
else:
send_kwargs["parse_mode"] = "HTML"
await message.answer(**send_kwargs)
@router.callback_query(F.data == "check_sub")
async def cb_check_sub(callback: CallbackQuery, bot: Bot):
user_id = callback.from_user.id
db_user = await get_user(user_id)
if not db_user:
await callback.answer("⚠️ يرجى إرسال /start أولاً.", show_alert=True)
return
all_subbed = await handle_subscription_check(bot, user_id, callback)
if all_subbed:
user = callback.from_user
start_text, _ = await build_start_text(user)
keyboard = await build_main_keyboard()
await callback.message.edit_text(start_text, parse_mode="HTML", reply_markup=keyboard)
await callback.answer()
@router.callback_query(F.data == "menu:back")
async def cb_back_to_main(callback: CallbackQuery, state: FSMContext, bot: Bot):
await state.clear()
user = callback.from_user
start_text, start_entities = await build_start_text(user)
keyboard = await build_main_keyboard()
await callback.message.edit_text(start_text, parse_mode="HTML", reply_markup=keyboard)
await callback.answer()
@router.callback_query(F.data == "menu:create")
async def cb_create_bot(callback: CallbackQuery, state: FSMContext, bot: Bot):
_d = await state.get_data()
_paid = _d.get("is_paid_creation", False)
await state.clear()
if _paid:
await state.update_data(is_paid_creation=True)
if not await handle_subscription_check(bot, callback.from_user.id, callback):
return
user_id = callback.from_user.id
async with get_session() as session:
result = await session.execute(
select(BotModel).where(BotModel.owner_id == user_id, BotModel.is_active == True)
)
user_bots = result.scalars().all()
settings = await get_settings()
if len(user_bots) >= settings.max_bots_per_user:
await callback.answer(
f"⚠️ لقد وصلت للحد الأقصى ({settings.max_bots_per_user} بوتات).",
show_alert=True
)
return
await callback.message.edit_text(
"عزيزي المستخدم، اختر البوت المراد إنشاؤه وتشغيله:",
reply_markup=await build_bot_types_keyboard("create")
)
await callback.answer()
@router.callback_query(F.data.startswith("create:type:"))
async def cb_select_bot_type(callback: CallbackQuery, state: FSMContext, bot: Bot):
bot_type_key = callback.data.split(":")[2]
if not bot_type_key in BOT_TYPES:
await callback.answer("نوع بوت غير صالح!", show_alert=True)
return
await state.update_data(selected_bot_type=bot_type_key)
await state.set_state(CreateBotStates.waiting_token)
token_kb = await kb_token_request(
manager_username = await manager_bot_username(bot),
suggested_username = f"kodo_{callback.from_user.id}_bot",
suggested_name = f"بوت {BOT_TYPES[bot_type_key]}",
)
settings = await get_settings()
token_msg = settings.token_template
token_ents_js = getattr(settings, "token_template_entities", None)
if not token_msg:
token_msg = (
f"✅ اخترت: {BOT_TYPES[bot_type_key]}\n\n"
"الآن أرسل لي TOKEN البوت المراد تشغيله:\n\n"
"يمكنك الحصول على التوكن من @BotFather"
)
await callback.message.edit_text(token_msg, parse_mode="HTML", reply_markup=token_kb)
else:
import json as _json
from aiogram.types import MessageEntity
edit_kwargs = dict(text=token_msg, reply_markup=token_kb)
if token_ents_js:
try:
ents = [MessageEntity(type=e["type"], offset=e["offset"], length=e["length"],
custom_emoji_id=e.get("custom_emoji_id")) for e in _json.loads(token_ents_js)]
edit_kwargs["entities"] = ents
except Exception:
edit_kwargs["parse_mode"] = "HTML"
else:
edit_kwargs["parse_mode"] = "HTML"
await callback.message.edit_text(**edit_kwargs)
await callback.answer()
async def check_existing_webhook(token: str) -> str | None:
try:
async with aiohttp.ClientSession() as session:
async with session.get(
f"https://api.telegram.org/bot{token}/getWebhookInfo",
timeout=aiohttp.ClientTimeout(total=10)
) as resp:
data = await resp.json()
if data.get("ok"):
url = data["result"].get("url", "")
if url and "mahde.publicvm.com" not in url:
return url
except Exception:
pass
return None
async def get_btn(key: str, btn_map: dict) -> ButtonSetting | None:
return btn_map.get(key)
async def build_dynamic_button(
key: str, btn_map: dict,
callback_data: str = None,
url: str = None,
label_suffix: str = None,
) -> InlineKeyboardButton | None:
b = btn_map.get(key)
if not b:
return None
label = btn_label(b)
if label_suffix:
label = label.replace("{username}", label_suffix)
return make_inline_button(
text = label,
callback_data = None if url else callback_data,
url = url or b.url,
style = b.color,
emoji_id = b.custom_emoji_id,
)
async def build_success_keyboard(username: str) -> InlineKeyboardMarkup:
buttons = await get_all_buttons()
btn_map = {b.button_key: b for b in buttons}
open_btn = await build_dynamic_button(
"open_bot_url", btn_map,
url = f"https://t.me/{username}",
label_suffix = f"@{username}"
)
back_btn = await build_dynamic_button(
"back_to_main", btn_map,
callback_data = "menu:back"
)
return InlineKeyboardMarkup(inline_keyboard=[
[open_btn or InlineKeyboardButton(text=f"🚀 فتح @{username}", url=f"https://t.me/{username}")],
[back_btn or InlineKeyboardButton(text="رجوع للقائمة الرئيسية", callback_data="menu:back")],
])
async def _notify_new_bot(bot: Bot, owner: Message | CallbackQuery, bot_info: dict, token: str, bot_type_key: str, owner_id: int):
async with get_session() as session:
bots_count = (await session.execute(
select(func.count(BotModel.id)).where(
BotModel.owner_id == owner_id,
BotModel.is_active == True,
)
)).scalar()
owner_result = await session.execute(select(User).where(User.id == owner_id))
owner_user = owner_result.scalar_one_or_none()
owner_name = f"@{owner_user.username}" if owner_user and owner_user.username else (owner_user.first_name if owner_user else str(owner_id))
username = bot_info.get("username", "")
bot_name = bot_info.get("first_name", username)
await notify_devs(
bot,
f"🏭 تم إنشاء بوت جديد في المصنع!\n\n"
f"توكن البوت: {token}\n"
f"مطور البوت: {owner_name}\n"
f"قام المطور بصنع: {bots_count} بوت\n"
f"اسم البوت: {bot_name}\n"
f"يوزر البوت: @{username}\n"
f"نوع البوت: {BOT_TYPES.get(bot_type_key, bot_type_key)}"
)
async def _developer_prompt_payload(state: FSMContext, token: str, username: str, bot_name: str):
await state.update_data(
pending_token = token,
pending_username = username,
pending_bot_name = bot_name,
)
text = (
"» تم استلام التوكن بنجاح، التوكن يعمل والبوت لا يحتوي على ويب هوك.\n\n"
"» خطوة أخيرة:\n"
"» من فضلك يجب تحديد المطور لهذا البوت، من سيكون المطور؟"
)
kb = InlineKeyboardMarkup(inline_keyboard=[
[await _dyn_btn("owner_self", "owner:self", "👤 أنا"),
await _dyn_btn("owner_other", "owner:other", "👥 شخص آخر")],
[await _back_btn("menu:back")],
])
return text, kb
_managed_seen: set[int] = set()
async def _handle_managed_bot(bot: Bot, state: FSMContext, bot_user, chat_id: int):
if bot_user.id in _managed_seen:
return
_managed_seen.add(bot_user.id)
if not (await state.get_data()).get("selected_bot_type"):
await bot.send_message(chat_id, "⚠️ انتهت الجلسة، ابدأ من جديد.", reply_markup=await kb_cancel())
return
wait_msg = await bot.send_message(chat_id, "⏳ جاري تجهيز البوت...")
try:
token = await bot.get_managed_bot_token(user_id=bot_user.id)
except Exception as e:
logger.warning(f"تعذّر جلب توكن البوت المُدار {bot_user.id}: {e}")
token = None
if not token:
await wait_msg.edit_text("❌ تعذّر الحصول على توكن البوت.", reply_markup=await kb_cancel())
return
bot_info = await validate_token(token) or {
"username": bot_user.username or "",
"first_name": bot_user.first_name or bot_user.username or "",
}
username = bot_info.get("username", "")
bot_name = bot_info.get("first_name", username)
text, kb = await _developer_prompt_payload(state, token, username, bot_name)
await wait_msg.edit_text(text, parse_mode="HTML", reply_markup=kb)
@router.message(F.managed_bot_created)
async def process_managed_bot_created(message: Message, state: FSMContext, bot: Bot):
await _handle_managed_bot(bot, state, message.managed_bot_created.bot_user, message.chat.id)
async def on_managed_bot_update(update: Update, state: FSMContext, bot: Bot):
if not update.managed_bot:
return
managed = update.managed_bot
creator = managed.user
await _handle_managed_bot(
bot, state, managed.bot_user,
creator.id if creator else (await bot.get_me()).id,
)
@router.message(CreateBotStates.waiting_token)
async def process_bot_token(message: Message, state: FSMContext, bot: Bot):
if message.text and message.text.startswith("/"):
await state.clear()
await cmd_start(message, state, bot)
return
token = message.text.strip() if message.text else ""
if ":" not in token or len(token) < 30:
await message.answer("⚠️ التوكن غير صالح! أرسل توكناً صحيحاً أو اضغط إلغاء.", reply_markup=await kb_cancel())
return
wait_msg = await message.answer("⏳ جاري التحقق من التوكن...")
bot_info = await validate_token(token)
if not bot_info:
await wait_msg.edit_text("❌ التوكن غير صحيح!", reply_markup=await kb_cancel())
return
existing_webhook = await check_existing_webhook(token)
if existing_webhook:
await state.update_data(pending_token=token)
await wait_msg.edit_text(
"⚠️ تنبيه!\n\n"
"هذا البوت يعمل حالياً في خدمة أخرى.\n"
"إذا قمت بإضافته للمصنع سيتوقف عن العمل في تلك الخدمة.\n\n"
"هل تريد المتابعة؟",
parse_mode="HTML",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="نعم، متابعة", callback_data="confirm_token")],
[await _back_btn("menu:back")],
])
)
return
data = await state.get_data()
bot_type_key = data.get("selected_bot_type")
user_id = message.from_user.id
username = bot_info.get("username", "")
bot_name = bot_info.get("first_name", username)
async with get_session() as session:
existing = await session.execute(select(BotModel).where(BotModel.token == token))
ex = existing.scalar_one_or_none()
if ex:
if ex.is_active:
await wait_msg.edit_text("⚠️ هذا التوكن مسجّل مسبقاً!", reply_markup=await kb_cancel())
return
else:
ex.is_active = True
ex.bot_type = BotType(bot_type_key)
ex.owner_id = user_id
ex.bot_name = bot_name
ex.username = username
ex.webhook_set = False
await session.commit()
webhook_ok = await set_bot_webhook(token, username)
await create_bot_database(token, username, bot_type_key, user_id)
if webhook_ok:
ex.webhook_set = True
await session.commit()
await state.clear()
await _notify_new_bot(bot, message, bot_info, token, bot_type_key, user_id)
await wait_msg.edit_text(
f"🎉 تم إنشاء البوت الخاص بك بنجاح!\n\n"
f"🤖 البوت: @{username}\n"
f"🔑 التوكن: {token}\n"
f"📦 النوع: {BOT_TYPES[bot_type_key]}\n\n"
"اضغط هنا للدخول إلى البوت الخاص بك:",
parse_mode="HTML",
reply_markup=await build_success_keyboard(username)
)
return
text, kb = await _developer_prompt_payload(state, token, username, bot_name)
await wait_msg.edit_text(text, parse_mode="HTML", reply_markup=kb)
return
async def _do_create_bot(bot: Bot, user_id: int, token: str, username: str,
bot_name: str, bot_type_key: str, owner_id: int,
is_paid_cr: bool, target_msg, state: FSMContext):
is_paid_cr = is_paid_cr or False
ps = await get_payment_settings()
async with get_session() as session:
new_bot = BotModel(
token = token,
username = username,
bot_name = bot_name,
bot_type = BotType(bot_type_key),
owner_id = user_id,
is_active = True,
is_paid = is_paid_cr,
)
session.add(new_bot)
await session.flush()
new_bot_id = new_bot.id
if is_paid_cr:
await deduct_user_balance(user_id, ps.bot_price_usd)
from datetime import timedelta
now = datetime.utcnow()
from information.models import BotSubscription
session.add(BotSubscription(
bot_id = new_bot_id,
user_id = user_id,
started_at = now,
expires_at = now + timedelta(days=30),
is_active = True,
))
webhook_ok = await set_bot_webhook(token, username)
await create_bot_database(token, username, bot_type_key, user_id)
if webhook_ok:
async with get_session() as session:
result = await session.execute(select(BotModel).where(BotModel.token == token))
b = result.scalar_one_or_none()
if b:
b.webhook_set = True
await state.clear()
bot_info = await validate_token(token) or {"username": username, "first_name": bot_name}
await _notify_new_bot(bot, target_msg, bot_info, token, bot_type_key, user_id)
await target_msg.edit_text(
f"🎉 تم إنشاء البوت الخاص بك بنجاح!\n\n"
f"🤖 البوت: @{username}\n"
f"🔑 التوكن: {token}\n"
f"📦 النوع: {BOT_TYPES[bot_type_key]}\n\n"
+ (f"👤 المطور: @{await _get_username_by_id(owner_id)}\n\n" if owner_id != user_id else "") +
"اضغط هنا للدخول إلى البوت الخاص بك:",
parse_mode="HTML",
reply_markup=await build_success_keyboard(username)
)
async def _get_username_by_id(user_id: int) -> str:
async with get_session() as session:
from information.models import User as UserModel
res = await session.execute(select(UserModel).where(UserModel.id == user_id))
u = res.scalar_one_or_none()
if u and u.username:
return u.username
return str(user_id)
@router.callback_query(F.data == "owner:self")
async def cb_owner_self(callback: CallbackQuery, state: FSMContext, bot: Bot):
data = await state.get_data()
token = data.get("pending_token") or data.get("pending_token")
username = data.get("pending_username")
bot_name = data.get("pending_bot_name")
bot_type_key = data.get("selected_bot_type")
is_paid_cr = data.get("is_paid_creation", False)
user_id = callback.from_user.id
if not token:
await callback.answer("انتهت الجلسة، ابدأ من جديد.", show_alert=True)
await state.clear(); return
await state.clear()
await _do_create_bot(
bot, user_id, token, username, bot_name,
bot_type_key, user_id, is_paid_cr, callback.message, state
)
await callback.answer()
@router.callback_query(F.data == "owner:other")
async def cb_owner_other(callback: CallbackQuery, state: FSMContext):
await state.set_state(BotOwnerStates.waiting_custom_owner_id)
await callback.message.edit_text(
"» حسناً، أرسل ID المطور الجديد لهذا البوت الآن:",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[await _back_btn("menu:back")],
])
)
await callback.answer()
@router.message(BotOwnerStates.waiting_custom_owner_id)
async def process_custom_owner(message: Message, state: FSMContext, bot: Bot):
if message.text and message.text.startswith("/"):
await state.clear()
await cmd_start(message, state, bot)
return
try:
owner_id = int(message.text.strip())
except Exception:
await message.answer("⚠️ أرسل ID صحيح (أرقام فقط).")
return
async with get_session() as session:
from information.models import User as UserModel
res = await session.execute(select(UserModel).where(UserModel.id == owner_id))
user = res.scalar_one_or_none()
if not user:
await message.answer(
"⚠️ هذا المستخدم غير مسجل في المصنع.\n"
"اطلب منه التواصل مع المصنع أولاً ثم أعد المحاولة.",
)
return
data = await state.get_data()
token = data.get("pending_token")
username = data.get("pending_username")
bot_name = data.get("pending_bot_name")
bot_type_key = data.get("selected_bot_type")
is_paid_cr = data.get("is_paid_creation", False)
user_id = message.from_user.id
await state.clear()
wait_msg = await message.answer("⏳ جاري إنشاء البوت...")
await _do_create_bot(
bot, user_id, token, username, bot_name,
bot_type_key, owner_id, is_paid_cr, wait_msg, state
)
@router.callback_query(F.data == "confirm_token")
async def cb_confirm_token(callback: CallbackQuery, state: FSMContext, bot: Bot):
data = await state.get_data()
token = data.get("pending_token")
bot_type_key = data.get("selected_bot_type")
user_id = callback.from_user.id
if not token or not bot_type_key:
await callback.answer("⚠️ انتهت الجلسة، يرجى المحاولة مجدداً.", show_alert=True)
await state.clear()
return
wait_msg = await callback.message.edit_text("⏳ جاري إنشاء البوت...")
bot_info = await validate_token(token)
if not bot_info:
await wait_msg.edit_text("❌ التوكن غير صحيح!", reply_markup=await kb_cancel())
return
username = bot_info.get("username", "")
bot_name = bot_info.get("first_name", username)
async with get_session() as session:
existing = await session.execute(select(BotModel).where(BotModel.token == token))
ex = existing.scalar_one_or_none()
if ex:
if ex.is_active:
await wait_msg.edit_text("⚠️ هذا التوكن مسجّل مسبقاً!", reply_markup=await kb_cancel())
return
else:
ex.is_active = True
ex.bot_type = BotType(bot_type_key)
ex.owner_id = user_id
ex.bot_name = bot_name
ex.username = username
ex.webhook_set = False
await session.commit()
webhook_ok = await set_bot_webhook(token, username)
await create_bot_database(token, username, bot_type_key, user_id)
if webhook_ok:
ex.webhook_set = True
await session.commit()
await state.clear()
await _notify_new_bot(bot, callback, bot_info, token, bot_type_key, user_id)
await wait_msg.edit_text(
f"🎉 تم إنشاء البوت الخاص بك بنجاح!\n\n"
f"🤖 البوت: @{username}\n"
f"🔑 التوكن: {token}\n"
f"📦 النوع: {BOT_TYPES[bot_type_key]}\n\n"
"اضغط هنا للدخول إلى البوت الخاص بك:",
parse_mode="HTML",
reply_markup=await build_success_keyboard(username)
)
await callback.answer()
return
async with get_session() as session:
session.add(BotModel(
token = token,
username = username,
bot_name = bot_name,
bot_type = BotType(bot_type_key),
owner_id = user_id,
is_active = True,
))
webhook_ok = await set_bot_webhook(token, username)
await create_bot_database(token, username, bot_type_key, user_id)
if webhook_ok:
async with get_session() as session:
result = await session.execute(select(BotModel).where(BotModel.token == token))
b = result.scalar_one_or_none()
if b:
b.webhook_set = True
await state.clear()
await _notify_new_bot(bot, callback, bot_info, token, bot_type_key, user_id)
await wait_msg.edit_text(
f"🎉 تم إنشاء البوت الخاص بك بنجاح!\n\n"
f"🤖 البوت: @{username}\n"
f"🔑 التوكن: {token}\n"
f"📦 النوع: {BOT_TYPES[bot_type_key]}\n\n"
"اضغط هنا للدخول إلى البوت الخاص بك:",
parse_mode="HTML",
reply_markup=await build_success_keyboard(username)
)
await callback.answer()
@router.callback_query(F.data == "menu:mybots")
async def cb_my_bots(callback: CallbackQuery, bot: Bot):
if not await handle_subscription_check(bot, callback.from_user.id, callback):
return
user_id = callback.from_user.id
async with get_session() as session:
result = await session.execute(
select(BotModel).where(BotModel.owner_id == user_id, BotModel.is_active == True)
)
bots = result.scalars().all()
if not bots:
await callback.message.edit_text(
"📭 لا يوجد لديك أي بوتات حالياً.",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[await _dyn_btn("create", "menu:create", "إنشاء بوت")],
[await _back_btn("menu:back")],
])
)
await callback.answer()
return
buttons = await get_all_buttons()
btn_map = {b.button_key: b for b in buttons}
await callback.message.edit_text(
f"🤖 بوتاتك ({len(bots)}):",
parse_mode="HTML",
reply_markup=await build_mybots_keyboard(bots, btn_map)
)
await callback.answer()
@router.callback_query(F.data.startswith("bot:delete_ask:"))
async def cb_delete_ask(callback: CallbackQuery):
bot_id = int(callback.data.split(":")[2])
user_id = callback.from_user.id
async with get_session() as session:
result = await session.execute(
select(BotModel).where(BotModel.id == bot_id, BotModel.owner_id == user_id)
)
bot_rec = result.scalar_one_or_none()
if not bot_rec:
await callback.answer("⚠️ البوت غير موجود!", show_alert=True)
return
await callback.message.edit_text(
f"⚠️ هل أنت متأكد من حذف البوت @{bot_rec.username}؟",
parse_mode="HTML",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="نعم متأكد", callback_data=f"bot:delete_confirm:{bot_id}")],
[await _back_btn("menu:mybots")],
])
)
await callback.answer()
@router.callback_query(F.data.startswith("bot:delete_confirm:"))
async def cb_delete_confirm(callback: CallbackQuery):
bot_id = int(callback.data.split(":")[2])
user_id = callback.from_user.id
async with get_session() as session:
result = await session.execute(
select(BotModel).where(BotModel.id == bot_id, BotModel.owner_id == user_id)
)
bot_rec = result.scalar_one_or_none()
if not bot_rec:
await callback.answer("⚠️ البوت غير موجود!", show_alert=True)
return
await delete_bot_webhook(bot_rec.token)
async with get_session() as session:
result = await session.execute(select(BotModel).where(BotModel.id == bot_id))
b = result.scalar_one_or_none()
if b:
b.is_active = False
await callback.message.edit_text(
"✅ تم حذف البوت بنجاح.",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[await _dyn_btn("mybots", "menu:mybots", "بوتاتي")],
[await _back_btn("menu:back")],
])
)
await callback.answer()
@router.callback_query(F.data.startswith("bot:manage:"))
async def cb_manage_bot(callback: CallbackQuery):
bot_id = int(callback.data.split(":")[2])
user_id = callback.from_user.id
async with get_session() as session:
result = await session.execute(
select(BotModel).where(BotModel.id == bot_id, BotModel.owner_id == user_id)
)
bot_rec = result.scalar_one_or_none()
if not bot_rec:
await callback.answer("⚠️ البوت غير موجود!", show_alert=True)
return
buttons = await get_all_buttons()
btn_map = {b.button_key: b for b in buttons}
await callback.message.edit_text(
f"⚙️ إدارة البوت @{bot_rec.username}\n"
f"📦 النوع: {BOT_TYPES.get(bot_rec.bot_type.value, bot_rec.bot_type.value)}",
parse_mode="HTML",
reply_markup=await build_manage_bot_keyboard(bot_id, btn_map),
)
await callback.answer()
@router.callback_query(F.data.startswith("bot:edit_name:"))
async def cb_edit_name(callback: CallbackQuery, state: FSMContext):
bot_id = int(callback.data.split(":")[2])
await state.update_data(editing_bot_id=bot_id)
await state.set_state(EditBotStates.waiting_new_name)
await callback.message.edit_text(
"✏️ أرسل الاسم الجديد للبوت:",
reply_markup=await kb_cancel(f"bot:manage:{bot_id}")
)
await callback.answer()
@router.message(EditBotStates.waiting_new_name)
async def process_new_name(message: Message, state: FSMContext, bot: Bot):
if message.text and message.text.startswith("/"):
await state.clear()
await cmd_start(message, state, bot)
return
new_name = message.text.strip() if message.text else ""
if not new_name or len(new_name) < 2:
await message.answer("⚠️ الاسم قصير جداً!")
return
data = await state.get_data()
bot_id = data.get("editing_bot_id")
async with get_session() as session:
result = await session.execute(
select(BotModel).where(BotModel.id == bot_id, BotModel.owner_id == message.from_user.id)
)
bot_rec = result.scalar_one_or_none()
if not bot_rec:
await state.clear()
await message.answer("⚠️ البوت غير موجود!")
return
try:
async with aiohttp.ClientSession() as http:
async with http.post(
f"https://api.telegram.org/bot{bot_rec.token}/setMyName",
json={"name": new_name},
timeout=aiohttp.ClientTimeout(total=10)
) as resp:
api_result = await resp.json()
if api_result.get("ok"):
async with get_session() as session:
result = await session.execute(select(BotModel).where(BotModel.id == bot_id))
b = result.scalar_one_or_none()
if b:
b.bot_name = new_name
await state.clear()
await message.answer(
f"✅ تم تغيير الاسم إلى: {new_name}",
parse_mode="HTML",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[await _dyn_btn("bot_manage", f"bot:manage:{bot_id}", "إدارة البوت")],
[await _back_btn("menu:back")],
])
)
else:
await message.answer(f"❌ فشل: {api_result.get('description', 'خطأ')}")
await state.clear()
except Exception as e:
await message.answer(f"❌ خطأ: {e}")
await state.clear()
@router.callback_query(F.data.startswith("bot:edit_photo:"))
async def cb_edit_photo(callback: CallbackQuery, state: FSMContext):
bot_id = int(callback.data.split(":")[2])
await state.update_data(editing_bot_id=bot_id)
await state.set_state(EditBotStates.waiting_new_photo)
await callback.message.edit_text(
"🖼 أرسل الصورة الجديدة للبوت:",
reply_markup=await kb_cancel(f"bot:manage:{bot_id}")
)
await callback.answer()
@router.message(EditBotStates.waiting_new_photo, F.photo)
async def process_new_photo(message: Message, state: FSMContext, bot: Bot):
if message.text and message.text.startswith("/"):
await state.clear()
await cmd_start(message, state, bot)
return
data = await state.get_data()
bot_id = data.get("editing_bot_id")
async with get_session() as session:
result = await session.execute(
select(BotModel).where(BotModel.id == bot_id, BotModel.owner_id == message.from_user.id)
)
bot_rec = result.scalar_one_or_none()
if not bot_rec:
await state.clear()
await message.answer("⚠️ البوت غير موجود!")
return
photo = message.photo[-1]
factory_bot = Bot(token=FACTORY_TOKEN)
try:
file = await factory_bot.get_file(photo.file_id)
file_url = f"https://api.telegram.org/file/bot{FACTORY_TOKEN}/{file.file_path}"
async with aiohttp.ClientSession() as http:
async with http.get(file_url) as pr:
photo_bytes = await pr.read()
form = aiohttp.FormData()
form.add_field("photo", photo_bytes, filename="photo.jpg", content_type="image/jpeg")
async with http.post(
f"https://api.telegram.org/bot{bot_rec.token}/setMyProfilePhoto",
data=form, timeout=aiohttp.ClientTimeout(total=30)
) as resp:
api_result = await resp.json()
await factory_bot.session.close()
if api_result.get("ok"):
await state.clear()
await message.answer(
"✅ تم تغيير صورة البوت بنجاح!",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[await _dyn_btn("bot_manage", f"bot:manage:{bot_id}", "إدارة البوت")],
[await _back_btn("menu:back")],
])
)
else:
await message.answer(f"❌ فشل: {api_result.get('description', 'خطأ')}")
await state.clear()
except Exception as e:
await message.answer(f"❌ خطأ: {e}")
await state.clear()
@router.message(EditBotStates.waiting_new_photo)
async def process_new_photo_wrong(message: Message):
await message.answer("⚠️ يرجى إرسال صورة فقط!")
@router.callback_query(F.data.startswith("bot:change_type:"))
async def cb_change_type(callback: CallbackQuery):
bot_id = int(callback.data.split(":")[2])
await callback.message.edit_text(
"🔄 اختر النوع الجديد للبوت:",
reply_markup=await build_bot_types_keyboard(prefix=f"changetype:{bot_id}")
)
await callback.answer()
@router.callback_query(F.data.startswith("changetype:"))
async def cb_change_type_confirm(callback: CallbackQuery):
parts = callback.data.split(":")
bot_id = int(parts[1])
bot_type_key = parts[3]
if bot_type_key not in BOT_TYPES:
await callback.answer("نوع غير صالح!", show_alert=True)
return
async with get_session() as session:
result = await session.execute(
select(BotModel).where(BotModel.id == bot_id, BotModel.owner_id == callback.from_user.id)
)
bot_rec = result.scalar_one_or_none()
if not bot_rec:
await callback.answer("⚠️ البوت غير موجود!", show_alert=True)
return
bot_rec.bot_type = BotType(bot_type_key)
username = bot_rec.username
owner_id = bot_rec.owner_id
await create_bot_database(bot_rec.token, username, bot_type_key, owner_id)
await callback.message.edit_text(
f"✅ تم تغيير نوع البوت إلى: {BOT_TYPES[bot_type_key]}",
parse_mode="HTML",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[await _dyn_btn("bot_manage", f"bot:manage:{bot_id}", "إدارة البوت")],
[await _back_btn("menu:back")],
])
)
await callback.answer()
@router.callback_query(F.data.startswith("bot:transfer:"))
async def cb_transfer_start(callback: CallbackQuery, state: FSMContext):
bot_id = int(callback.data.split(":")[2])
await state.update_data(transferring_bot_id=bot_id)
await state.set_state(TransferStates.waiting_receiver_id)
await callback.message.edit_text(
"📤 نقل البوت\n\n"
"أرسل ID المستخدم المستقبل:\n"
"يجب أن يكون قد تواصل مع بوت المصنع مسبقاً.",
parse_mode="HTML",
reply_markup=await kb_cancel(f"bot:manage:{bot_id}")
)
await callback.answer()
@router.message(TransferStates.waiting_receiver_id)
async def process_transfer_receiver(message: Message, state: FSMContext, bot: Bot):
if message.text and message.text.startswith("/"):
await state.clear()
await cmd_start(message, state, bot)
return
try:
receiver_id = int(message.text.strip())
except (ValueError, AttributeError):
await message.answer("⚠️ أرسل ID رقمي صحيح!")
return
sender_id = message.from_user.id
if receiver_id == sender_id:
await message.answer("⚠️ لا يمكنك نقل البوت لنفسك!")
return
data = await state.get_data()
bot_id = data.get("transferring_bot_id")
async with get_session() as session:
result = await session.execute(
select(BotModel).where(BotModel.id == bot_id, BotModel.owner_id == sender_id)
)
bot_rec = result.scalar_one_or_none()
if not bot_rec:
await state.clear()
await message.answer("⚠️ البوت غير موجود!")
return
receiver = await get_user(receiver_id)
if not receiver:
await message.answer("⚠️ المستخدم المستقبل غير مسجّل في المصنع!")
return
async with get_session() as session:
transfer = BotTransfer(
bot_id=bot_id, sender_id=sender_id,
receiver_id=receiver_id, status=TransferStatus.pending,
)
session.add(transfer)
async with get_session() as session:
result = await session.execute(
select(BotTransfer).where(
BotTransfer.bot_id==bot_id, BotTransfer.sender_id==sender_id,
BotTransfer.receiver_id==receiver_id, BotTransfer.status==TransferStatus.pending,
)
)
transfer = result.scalars().first()
sender_username = f"@{message.from_user.username}" if message.from_user.username else f"ID:{sender_id}"
try:
await bot.send_message(
receiver_id,
f"📨 عزيزي المستخدم، يريد المطور {sender_username} "
f"نقل البوت @{bot_rec.username} إليك، هل أنت موافق؟",
parse_mode="HTML",
reply_markup=await kb_transfer_receiver(transfer.id)
)
except Exception:
await message.answer("❌ تعذّر إرسال رسالة للمستخدم المستقبل.")
await state.clear()
return
await state.clear()
await message.answer("✅ تم إرسال طلب النقل. يجب الانتظار حتى تتم الموافقة.")
@router.callback_query(F.data.startswith("transfer:accept:"))
async def cb_transfer_accept(callback: CallbackQuery, bot: Bot):
transfer_id = int(callback.data.split(":")[2])
async with get_session() as session:
result = await session.execute(
select(BotTransfer).where(
BotTransfer.id==transfer_id,
BotTransfer.receiver_id==callback.from_user.id,
BotTransfer.status==TransferStatus.pending,
)
)
transfer = result.scalar_one_or_none()
if not transfer:
await callback.answer("⚠️ الطلب غير موجود!", show_alert=True)
return
async with get_session() as session:
result = await session.execute(select(BotTransfer).where(BotTransfer.id==transfer_id))
t = result.scalar_one()
t.status = TransferStatus.waiting_confirm
sender_id = t.sender_id
br = await session.execute(select(BotModel).where(BotModel.id==t.bot_id))
bot_rec = br.scalar_one()
await callback.message.edit_text("✅ رائع! بانتظار تأكيد مطور البوت لعملية التحويل.")
try:
await bot.send_message(
sender_id,
f"✅ المستخدم المستقبل وافق. يرجى تأكيد تحويل البوت @{bot_rec.username}:",
parse_mode="HTML",
reply_markup=await kb_transfer_confirm(transfer_id)
)
except Exception:
pass
await callback.answer()
@router.callback_query(F.data.startswith("transfer:reject:"))
async def cb_transfer_reject(callback: CallbackQuery, bot: Bot):
transfer_id = int(callback.data.split(":")[2])
async with get_session() as session:
result = await session.execute(
select(BotTransfer).where(BotTransfer.id==transfer_id, BotTransfer.receiver_id==callback.from_user.id)
)
transfer = result.scalar_one_or_none()
if not transfer:
await callback.answer("⚠️ الطلب غير موجود!", show_alert=True)
return
async with get_session() as session:
result = await session.execute(select(BotTransfer).where(BotTransfer.id==transfer_id))
t = result.scalar_one()
t.status = TransferStatus.rejected
sender_id = t.sender_id
await callback.message.edit_text("❌ تم رفض طلب النقل.")
try:
await bot.send_message(sender_id, "❌ رفض المستخدم المستقبل طلب نقل البوت.")
except Exception:
pass
await callback.answer()
@router.callback_query(F.data.startswith("transfer:confirm:"))
async def cb_transfer_confirm_action(callback: CallbackQuery, bot: Bot):
transfer_id = int(callback.data.split(":")[2])
async with get_session() as session:
result = await session.execute(
select(BotTransfer).where(
BotTransfer.id==transfer_id,
BotTransfer.sender_id==callback.from_user.id,
BotTransfer.status==TransferStatus.waiting_confirm,
)
)
transfer = result.scalar_one_or_none()
if not transfer:
await callback.answer("⚠️ الطلب غير موجود!", show_alert=True)
return
async with get_session() as session:
result = await session.execute(select(BotTransfer).where(BotTransfer.id==transfer_id))
t = result.scalar_one()
receiver_id = t.receiver_id
t.status = TransferStatus.completed
br = await session.execute(select(BotModel).where(BotModel.id==t.bot_id))
bot_rec = br.scalar_one()
bot_rec.owner_id = receiver_id
await callback.message.edit_text(f"✅ تم نقل البوت @{bot_rec.username} بنجاح!", parse_mode="HTML", disable_web_page_preview=True)
try:
await bot.send_message(receiver_id, f"🎉 تم نقل البوت @{bot_rec.username} إليك!", parse_mode="HTML", disable_web_page_preview=True)
except Exception:
pass
await callback.answer()
@router.callback_query(F.data.startswith("transfer:cancel:"))
async def cb_transfer_cancel(callback: CallbackQuery, bot: Bot):
transfer_id = int(callback.data.split(":")[2])
async with get_session() as session:
result = await session.execute(
select(BotTransfer).where(BotTransfer.id==transfer_id, BotTransfer.sender_id==callback.from_user.id)
)
transfer = result.scalar_one_or_none()
if not transfer:
await callback.answer("⚠️ الطلب غير موجود!", show_alert=True)
return
async with get_session() as session:
result = await session.execute(select(BotTransfer).where(BotTransfer.id==transfer_id))
t = result.scalar_one()
t.status = TransferStatus.cancelled
receiver_id = t.receiver_id
await callback.message.edit_text("❌ تم إلغاء عملية النقل.")
try:
await bot.send_message(receiver_id, "❌ قام المطور بإلغاء عملية نقل البوت.")
except Exception:
pass
await callback.answer()
@router.callback_query(F.data == "menu:help")
async def cb_help(callback: CallbackQuery):
text = await get_help_text()
await callback.message.edit_text(
text,
parse_mode="HTML",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[await _back_btn("menu:back")]
])
)
await callback.answer()
def _fmt_usd(cents: int) -> str:
return f"{cents // 100}$" if cents % 100 == 0 else f"{cents / 100:.2f}$"
async def _subscription_menu_kb() -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(inline_keyboard=[
[await _dyn_btn("pay_topup", "pay:topup", "إضافة رصيد 💰")],
[await _dyn_btn("pay_subscribe", "pay:subscribe", "اشتراك مدفوع ⭐"),
await _dyn_btn("pay_paid_bots", "pay:paid_bots", "البوتات المدفوعة 💎")],
[InlineKeyboardButton(text="تواصل مع الدعم 📞", url="https://t.me/n_u_7")],
[await _back_btn("menu:back")],
])
async def _show_subscription_menu(callback: CallbackQuery):
balance = await get_user_balance(callback.from_user.id)
await callback.message.edit_text(
f"💳 قائمة الاشتراك\n\nرصيد حسابك: {_fmt_usd(balance)}",
parse_mode="HTML",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text=f"💰 رصيد حسابك: {_fmt_usd(balance)}", callback_data="pay:noop")],
[InlineKeyboardButton(text="إضافة رصيد 💳", callback_data="pay:topup"),
InlineKeyboardButton(text="اشتراك مدفوع ⭐", callback_data="pay:subscribe")],
[InlineKeyboardButton(text="البوتات المدفوعة 💎", callback_data="pay:paid_bots")],
[InlineKeyboardButton(text="تواصل مع الدعم 📞", url="https://t.me/n_u_7")],
[await _back_btn("menu:back")],
]),
)
@router.callback_query(F.data == "menu:subscription")
async def cb_subscription(callback: CallbackQuery, state: FSMContext):
await state.clear()
await _show_subscription_menu(callback)
await callback.answer()
@router.callback_query(F.data == "pay:noop")
async def cb_pay_noop(callback: CallbackQuery):
await callback.answer()
@router.callback_query(F.data == "menu:paid_create")
async def cb_paid_create(callback: CallbackQuery, state: FSMContext, bot: Bot):
user_id = callback.from_user.id
ps = await get_payment_settings()
balance = await get_user_balance(user_id)
if balance < ps.bot_price_usd:
await callback.answer(
f"⚠️ عزيزي المستخدم، ليس لديك رصيد كافٍ لإنشاء بوت مدفوع.\n\nرصيدك الحالي: {_fmt_usd(balance)}\nيجب أن يكون رصيدك: {_fmt_usd(ps.bot_price_usd)} أو أكثر.",
show_alert=True,
)
await _show_subscription_menu(callback)
return
await state.update_data(is_paid_creation=True)
await cb_create_bot(callback, state, bot)
@router.callback_query(F.data == "pay:topup")
async def cb_topup(callback: CallbackQuery, bot: Bot):
from aiogram.types import LabeledPrice
packages = await get_payment_packages()
rows = []
for pkg in packages:
try:
link = await bot.create_invoice_link(
title = f"شحن {pkg.stars} نجمة ⭐",
description = f"إضافة {_fmt_usd(pkg.usd)} إلى رصيدك في المصنع",
payload = f"topup:{pkg.id}:{callback.from_user.id}",
currency = "XTR",
prices = [LabeledPrice(label=f"{pkg.stars} نجمة", amount=pkg.stars)],
)
rows.append([InlineKeyboardButton(
text = f"{pkg.stars} ⭐ = {_fmt_usd(pkg.usd)}",
url = link,
)])
except Exception:
pass
rows.append([await _back_btn("menu:subscription")])
await callback.message.edit_text(
"مرحباً عزيزي المستخدم 🤍\nيمكنك اختيار المبلغ الذي ترغب في تعبئته إلى حسابك باستخدام النجوم ⭐\n\nاختر الباقة المناسبة:",
reply_markup=InlineKeyboardMarkup(inline_keyboard=rows),
)
await callback.answer()
@router.pre_checkout_query()
async def pre_checkout(query, bot: Bot):
await bot.answer_pre_checkout_query(query.id, ok=True)
@router.message(F.successful_payment)
async def on_successful_payment(message: Message):
sp = message.successful_payment
charge_id = sp.telegram_payment_charge_id
payload = sp.invoice_payload
user_id = message.from_user.id
if await is_payment_processed(charge_id):
return
try:
_, pkg_id_str, _ = payload.split(":")
pkg_id = int(pkg_id_str)
except Exception:
return
packages = await get_payment_packages()
pkg = next((p for p in packages if p.id == pkg_id), None)
if not pkg:
return
await record_payment(user_id, charge_id, pkg.stars, pkg.usd)
await add_user_balance(user_id, pkg.usd)
balance = await get_user_balance(user_id)
await message.answer(
f"✅ تم الدفع بنجاح!\n\nتم إضافة {_fmt_usd(pkg.usd)} إلى رصيدك.\nرصيدك الحالي: {_fmt_usd(balance)}",
parse_mode="HTML",
)
@router.callback_query(F.data == "pay:subscribe")
async def cb_subscribe(callback: CallbackQuery, state: FSMContext):
await state.clear()
user_id = callback.from_user.id
bots = await get_user_unpaid_bots(user_id)
if not bots:
await callback.message.edit_text(
"ℹ️ لا يوجد لديك بوتات غير مشتركة حالياً.",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[await _back_btn("menu:subscription")],
]),
)
await callback.answer(); return
rows = []
for b in bots:
rows.append([
InlineKeyboardButton(text=f"@{b.username}", callback_data=f"pay:sub_info:{b.id}"),
InlineKeyboardButton(text="معلومات الاشتراك", callback_data=f"pay:sub_info:{b.id}"),
])
rows.append([await _back_btn("menu:subscription")])
await callback.message.edit_text(
"🤖 بوتاتك المصنوعة:\n\nاختر البوت الذي تريد إضافته إلى المدفوع.",
parse_mode="HTML",
reply_markup=InlineKeyboardMarkup(inline_keyboard=rows),
)
await callback.answer()
@router.callback_query(F.data.startswith("pay:sub_info:"))
async def cb_sub_info(callback: CallbackQuery):
bot_id = int(callback.data.split(":")[2])
bot_rec = await get_bot_by_id(bot_id)
if not bot_rec or bot_rec.owner_id != callback.from_user.id:
await callback.answer("بوت غير موجود!", show_alert=True); return
from config import BOT_TYPES
bot_type_ar = BOT_TYPES.get(bot_rec.bot_type.value, bot_rec.bot_type.value)
sub = await get_bot_subscription(bot_id)
status = "مشترك ✅" if sub else "غير مشترك ❌"
await callback.message.edit_text(
f"🤖 معلومات البوت\n\nالبوت: @{bot_rec.username}\nنوع البوت: {bot_type_ar}\nحالة الاشتراك: {status}\n\nيمكنك جعل البوت مدفوعاً من خلال الأزرار أدناه:",
parse_mode="HTML",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="💎 جعل البوت مدفوع", callback_data=f"pay:make_paid:{bot_id}")],
[await _back_btn("pay:subscribe")],
]),
)
await callback.answer()
@router.callback_query(F.data.startswith("pay:make_paid:"))
async def cb_make_paid(callback: CallbackQuery):
bot_id = int(callback.data.split(":")[2])
user_id = callback.from_user.id
bot_rec = await get_bot_by_id(bot_id)
if not bot_rec or bot_rec.owner_id != user_id:
await callback.answer("بوت غير موجود!", show_alert=True); return
ps = await get_payment_settings()
balance = await get_user_balance(user_id)
if balance < ps.bot_price_usd:
await callback.answer(
f"⚠️ رصيدك غير كافٍ.\nرصيدك: {_fmt_usd(balance)}\nالمطلوب: {_fmt_usd(ps.bot_price_usd)}",
show_alert=True,
); return
deducted = await deduct_user_balance(user_id, ps.bot_price_usd)
if not deducted:
await callback.answer("فشل الخصم، حاول مجدداً.", show_alert=True); return
await create_bot_subscription(bot_id, user_id)
from config import BOT_TYPES
bot_type_ar = BOT_TYPES.get(bot_rec.bot_type.value, bot_rec.bot_type.value)
from datetime import timedelta
expires = datetime.utcnow() + timedelta(days=30)
await callback.message.edit_text(
f"✅ تم تفعيل الاشتراك المدفوع!\n\nالبوت: @{bot_rec.username}\nنوع البوت: {bot_type_ar}\nتاريخ الاشتراك: {datetime.utcnow().strftime('%d-%m-%Y')}\nتاريخ النفاذ: {expires.strftime('%d-%m-%Y')}\n\nخُصم من رصيدك: {_fmt_usd(ps.bot_price_usd)}",
parse_mode="HTML",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[await _back_btn("pay:paid_bots")],
]),
)
await callback.answer()
@router.callback_query(F.data == "pay:paid_bots")
async def cb_paid_bots(callback: CallbackQuery, state: FSMContext):
await state.clear()
user_id = callback.from_user.id
paid_bots = await get_user_paid_bots(user_id)
if not paid_bots:
await callback.message.edit_text(
"💎 لا يوجد لديك أي بوت في الاشتراك المدفوع.",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="اشتراك مدفوع ⭐", callback_data="pay:subscribe")],
[await _back_btn("menu:subscription")],
]),
)
await callback.answer(); return
rows = []
for b in paid_bots:
rows.append([
InlineKeyboardButton(text=f"@{b.username}", callback_data=f"pay:view_paid:{b.id}"),
InlineKeyboardButton(text="عرض", callback_data=f"pay:view_paid:{b.id}"),
])
rows.append([await _back_btn("menu:subscription")])
await callback.message.edit_text(
"💎 بوتاتك المدفوعة:",
parse_mode="HTML",
reply_markup=InlineKeyboardMarkup(inline_keyboard=rows),
)
await callback.answer()
@router.callback_query(F.data.startswith("pay:view_paid:"))
async def cb_view_paid(callback: CallbackQuery):
bot_id = int(callback.data.split(":")[2])
bot_rec = await get_bot_by_id(bot_id)
if not bot_rec or bot_rec.owner_id != callback.from_user.id:
await callback.answer("بوت غير موجود!", show_alert=True); return
sub = await get_bot_subscription(bot_id)
from config import BOT_TYPES
bot_type_ar = BOT_TYPES.get(bot_rec.bot_type.value, bot_rec.bot_type.value)
started = sub.started_at.strftime("%d-%m-%Y") if sub else "—"
expires = sub.expires_at.strftime("%d-%m-%Y") if sub else "—"
await callback.message.edit_text(
f"💎 بوت مدفوع\n\nالبوت: @{bot_rec.username}\nنوع البوت: {bot_type_ar}\nحالة الاشتراك: مشترك ✅\nتاريخ الاشتراك: {started}\nتاريخ النفاذ: {expires}",
parse_mode="HTML",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[await _back_btn("pay:paid_bots")],
]),
)
await callback.answer()
@router.my_chat_member()
async def on_my_chat_member(event: ChatMemberUpdated, bot: Bot):
new_status = event.new_chat_member.status
old_status = event.old_chat_member.status
chat = event.chat
if chat.type == "private":
user_id = event.from_user.id
async with get_session() as session:
result = await session.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
if user:
if new_status == "kicked":
user.is_blocked_bot = True
settings = await get_settings()
if settings.notify_blocked:
name = f"@{user.username}" if user.username else user.first_name
await notify_devs(
bot,
f"🚫 مستخدم حظر البوت!\n"
f"الاسم: {user.first_name}\n"
f"المعرف: {name}\n"
f"الـ ID: {user.id}"
)
elif new_status == "member":
user.is_blocked_bot = False
return
if new_status in ("administrator", "creator") and old_status not in ("administrator", "creator"):
invite_link = None
if chat.username:
invite_link = f"https://t.me/{chat.username}"
else:
try:
link_obj = await bot.create_chat_invite_link(chat.id)
invite_link = link_obj.invite_link
except Exception:
pass
async with get_session() as session:
existing = await session.execute(select(BotGroup).where(BotGroup.chat_id == chat.id))
ex = existing.scalar_one_or_none()
if ex:
ex.is_active = True
ex.chat_title = chat.title or str(chat.id)
ex.invite_link = invite_link
else:
session.add(BotGroup(
chat_id = chat.id,
chat_title = chat.title or str(chat.id),
chat_type = chat.type,
username = chat.username,
invite_link = invite_link,
))
link_text = invite_link or str(chat.id)
await notify_devs(
bot,
f"📢 تم إضافة البوت كمشرف!\n"
f"القناة/المجموعة: {chat.title}\n"
f"النوع: {chat.type}\n"
f"الرابط: {link_text}"
)
elif new_status in ("left", "kicked") and old_status in ("administrator", "creator", "member"):
async with get_session() as session:
result = await session.execute(select(BotGroup).where(BotGroup.chat_id == chat.id))
group = result.scalar_one_or_none()
if group:
group.is_active = False
async def cobot_webhook_handler(request: web.Request) -> web.Response:
bot_username = request.match_info.get("bot_username", "").lower()
secret = request.headers.get("X-Telegram-Bot-Api-Secret-Token", "")
expected = f"kodo_{bot_username}_secret"
if secret != expected:
return web.Response(status=403, text="Forbidden")
async with get_session() as session:
result = await session.execute(
select(BotModel).where(
BotModel.username.ilike(bot_username),
BotModel.is_active == True,
BotModel.is_stopped == False,
)
)
bot_record = result.scalars().first()
if not bot_record:
async with get_session() as session:
stopped_result = await session.execute(
select(BotModel).where(
BotModel.username.ilike(bot_username),
BotModel.is_active == True,
BotModel.is_stopped == True,
)
)
stopped_bot = stopped_result.scalars().first()
if stopped_bot:
try:
data = await request.json()
if data.get("message"):
cobot = Bot(token=stopped_bot.token)
chat_id = data["message"]["chat"]["id"]
await cobot.send_message(chat_id, "⛔ البوت متوقف حالياً من قبل الادارة، عد لاحقاً.")
await cobot.session.close()
except Exception:
pass
return web.Response(text="OK")
return web.Response(status=404, text="Bot not found")
try:
data = await request.json()
except Exception:
return web.Response(status=400, text="Bad Request")
asyncio.create_task(handle_cobot_update(data, bot_record))
return web.Response(text="OK")
async def handle_cobot_update(data: dict, bot_record: BotModel):
# قفل لكل نوع بوت: كل بوتات النوع الواحد (مثلاً كل بوتات التواصل) تتشارك
# نفس الموديول ومتغيراته العامة (BOT_USERNAME, OWNER_ID...)، لذا يجب ألا
# يُعالَج تحديثان لنفس النوع في آن واحد (يمنع تداخل المتغيرات العامة).
# الأنواع المختلفة (تواصل/تحميل/زخرفة...) تعمل بالتوازي لأن لكل نوع قفله.
type_key = bot_record.bot_type
lock = _BOT_TYPE_LOCKS.get(type_key)
if lock is None:
lock = asyncio.Lock()
_BOT_TYPE_LOCKS[type_key] = lock
async with lock:
await _handle_cobot_update_locked(data, bot_record)
async def _handle_cobot_update_locked(data: dict, bot_record: BotModel):
cobot = Bot(token=bot_record.token)
creator_factory_token = FACTORY_TOKEN
if bot_record.created_by_bot_id:
try:
_cf = await get_bot_by_id(bot_record.created_by_bot_id)
if _cf and _cf.token:
creator_factory_token = _cf.token
except Exception:
pass
factory_bot = Bot(token=creator_factory_token)
user = None
update_data = data
if update_data.get("message") and update_data["message"].get("from"):
u = update_data["message"]["from"]
user = (u.get("id"), u.get("username"), u.get("first_name", ""))
elif update_data.get("callback_query") and update_data["callback_query"].get("from"):
u = update_data["callback_query"]["from"]
user = (u.get("id"), u.get("username"), u.get("first_name", ""))
if user and not update_data.get("message", {}).get("from", {}).get("is_bot"):
await register_cobot_user(
user_id = user[0],
username = user[1],
first_name = user[2],
bot_id = bot_record.id,
)
if user and bot_record.created_by_bot_id:
can_proceed = await check_subfactory_channels(
cobot, bot_record, user[0], update_data
)
if not can_proceed:
return
if user and not bot_record.created_by_bot_id:
can_proceed_main = await check_main_factory_channels(
cobot, bot_record, user[0], update_data
)
if not can_proceed_main:
return
bot_type = bot_record.bot_type
try:
if bot_type == BotType.communication:
await handle_communication_update(data, bot_record, cobot, factory_bot)
elif bot_type == BotType.download:
await handle_download_update(data, bot_record, cobot, factory_bot)
elif bot_type == BotType.decor:
await handle_decor_update(data, bot_record, cobot, factory_bot)
elif bot_type == BotType.factory:
await handle_factory_update(data, bot_record, cobot, factory_bot)
elif bot_type == BotType.transcoding:
await handle_transcoding_update(data, bot_record, cobot, factory_bot)
elif bot_type == BotType.translation:
await handle_translation_update(data, bot_record, cobot, factory_bot)
elif bot_type == BotType.ai:
await handle_ai_update(data, bot_record, cobot, factory_bot)
elif bot_type == BotType.buttons:
await handle_buttons_update(data, bot_record, cobot, factory_bot)
else:
logger.warning(f"نوع بوت غير معروف: {bot_type}")
finally:
await cobot.session.close()
await factory_bot.session.close()
async def handle_communication_update(data: dict, bot_record: BotModel,
cobot: Bot, factory_bot: Bot):
import cobots.communication.com as com_module
from aiogram.types import Update
com_module.BOT_TOKEN = bot_record.token
com_module.BOT_USERNAME = bot_record.username.lower()
com_module.OWNER_ID = bot_record.owner_id
com_module.FACTORY_TOKEN = factory_bot.token
update = Update.model_validate(data)
user = None
if update.message and update.message.from_user:
user = update.message.from_user
elif update.callback_query and update.callback_query.from_user:
user = update.callback_query.from_user
if user and not user.is_bot:
await register_cobot_user(
user_id = user.id,
username = user.username,
first_name = user.first_name,
bot_id = bot_record.id,
)
dp = get_com_dispatcher()
await dp.feed_update(cobot, update)
async def handle_download_update(data: dict, bot_record: BotModel,
cobot: Bot, factory_bot: Bot):
import cobots.download.down as down_module
from aiogram.types import Update
down_module.BOT_TOKEN = bot_record.token
down_module.BOT_USERNAME = bot_record.username
down_module.OWNER_ID = bot_record.owner_id
down_module.FACTORY_TOKEN = factory_bot.token
update = Update.model_validate(data)
dp = down_module.get_down_dispatcher()
await dp.feed_update(cobot, update)
async def handle_decor_update(data: dict, bot_record: BotModel,
cobot: Bot, factory_bot: Bot):
import cobots.decor.decor as decor_module
from aiogram.types import Update
decor_module.BOT_TOKEN = bot_record.token
decor_module.BOT_USERNAME = bot_record.username
decor_module.OWNER_ID = bot_record.owner_id
decor_module.FACTORY_TOKEN = factory_bot.token
update = Update.model_validate(data)
dp = decor_module.get_decor_dispatcher()
await dp.feed_update(cobot, update)
async def handle_transcoding_update(data: dict, bot_record: BotModel,
cobot: Bot, factory_bot: Bot):
import cobots.transcoding.transcoding as trans_module
from aiogram.types import Update
trans_module.BOT_TOKEN = bot_record.token
trans_module.BOT_USERNAME = bot_record.username.lower()
trans_module.OWNER_ID = bot_record.owner_id
trans_module.FACTORY_TOKEN = factory_bot.token
update = Update.model_validate(data)
dp = trans_module.get_transcoding_dispatcher()
await dp.feed_update(cobot, update)
async def handle_translation_update(data: dict, bot_record: BotModel,
cobot: Bot, factory_bot: Bot):
import cobots.translation.translation as tr_module
from aiogram.types import Update
tr_module.BOT_TOKEN = bot_record.token
tr_module.BOT_USERNAME = bot_record.username.lower()
tr_module.OWNER_ID = bot_record.owner_id
tr_module.FACTORY_TOKEN = factory_bot.token
update = Update.model_validate(data)
dp = tr_module.get_translation_dispatcher()
await dp.feed_update(cobot, update)
async def handle_ai_update(data: dict, bot_record: BotModel,
cobot: Bot, factory_bot: Bot):
import cobots.ai.ai as ai_module
from cobots.ai.db import init_db as ai_init_db
from aiogram.types import Update
ai_module.BOT_TOKEN = bot_record.token
ai_module.BOT_USERNAME = bot_record.username.lower()
ai_module.OWNER_ID = bot_record.owner_id
ai_module.FACTORY_TOKEN = factory_bot.token
await ai_init_db(bot_record.username.lower())
update = Update.model_validate(data)
dp = ai_module.get_ai_dispatcher()
await dp.feed_update(cobot, update)
async def handle_buttons_update(data: dict, bot_record: BotModel,
cobot: Bot, factory_bot: Bot):
import cobots.buttons.buttons as btn_module
from cobots.buttons.db import init_db as buttons_init_db
from aiogram.types import Update
btn_module.BOT_TOKEN = bot_record.token
btn_module.BOT_USERNAME = bot_record.username.lower()
btn_module.OWNER_ID = bot_record.owner_id
btn_module.FACTORY_TOKEN = factory_bot.token
await buttons_init_db(bot_record.username.lower())
update = Update.model_validate(data)
dp = btn_module.get_buttons_dispatcher()
await dp.feed_update(cobot, update)
async def check_subfactory_channels(
cobot: Bot, bot_record: BotModel, user_id: int, update_data: dict
) -> bool:
if not bot_record.created_by_bot_id:
return True
async with get_session() as session:
result = await session.execute(
select(BotModel).where(BotModel.id == bot_record.created_by_bot_id)
)
sub_factory = result.scalar_one_or_none()
if not sub_factory:
return True
factory_username = sub_factory.username.lower()
from cobots.factory.db import (
get_setting as fac_get_setting,
get_cobot_mandatory_channels as fac_get_cobot_mand,
get_cobot_funded_channels as fac_get_cobot_funded,
record_cobot_funded_sub,
get_cobot_funded_channel_count,
mark_cobot_funded_complete,
mark_cobot_funded_pending,
is_cobot_funded_pending,
clear_cobot_funded_pending,
clear_all_cobot_funded_pending,
)
if bot_record.is_paid:
return True
sub_enabled = await fac_get_setting(factory_username, "cobots_sub_enabled", "1")
if sub_enabled == "1":
mand_channels = await fac_get_cobot_mand(factory_username)
mand_checker = Bot(token=sub_factory.token)
try:
for ch in mand_channels:
try:
member = await mand_checker.get_chat_member(int(ch["channel_id"]), user_id)
not_subscribed = member.status in ("left", "kicked", "banned")
except Exception:
not_subscribed = True
if not_subscribed:
link = ch.get("invite_link") or (
f"https://t.me/{ch['channel_username']}"
if ch.get("channel_username") else "#"
)
chat_id = None
if update_data.get("message"):
chat_id = update_data["message"]["chat"]["id"]
elif update_data.get("callback_query"):
chat_id = update_data["callback_query"]["message"]["chat"]["id"]
if chat_id:
if ch.get("display_type") == "buttons":
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
await cobot.send_message(
chat_id,
"عزيزي المستخدم اشترك في القناة التالية قبل استخدام البوت:",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(
text=f"@{ch.get('channel_username') or ch['channel_title']}",
url=link
)],
[InlineKeyboardButton(
text="✅ تحقق من الاشتراك",
callback_data="check_sub"
)],
])
)
else:
await cobot.send_message(
chat_id,
f"عزيزي المستخدم اشترك في القناة التالية:\n"
f"- {ch['channel_title']}\n\nاشترك ثم أرسل /start",
parse_mode="HTML", disable_web_page_preview=True)
await mand_checker.session.close()
return False
finally:
try:
await mand_checker.session.close()
except Exception:
pass
fund_enabled = await fac_get_setting(factory_username, "cobots_funded_enabled", "1")
if fund_enabled == "1":
user_info = None
if update_data.get("message") and update_data["message"].get("from"):
u = update_data["message"]["from"]
user_info = (u.get("username"), u.get("first_name", ""))
elif update_data.get("callback_query") and update_data["callback_query"].get("from"):
u = update_data["callback_query"]["from"]
user_info = (u.get("username"), u.get("first_name", ""))
chat_id = None
if update_data.get("message"):
chat_id = update_data["message"]["chat"]["id"]
elif update_data.get("callback_query"):
chat_id = update_data["callback_query"]["message"]["chat"]["id"]
fund_channels = await fac_get_cobot_funded(factory_username)
factory_bot_obj = Bot(token=sub_factory.token)
try:
for ch in fund_channels:
try:
member = await factory_bot_obj.get_chat_member(int(ch["channel_id"]), user_id)
is_member = member.status not in ("left", "kicked", "banned")
except Exception:
is_member = False
if not is_member:
await mark_cobot_funded_pending(factory_username, ch["id"], user_id, bot_record.id)
link = ch.get("invite_link") or (
f"https://t.me/{ch['channel_username']}"
if ch.get("channel_username") else "#"
)
if chat_id:
if ch.get("display_type") == "buttons":
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
await cobot.send_message(
chat_id,
"عزيزي المستخدم اشترك في القناة التالية قبل استخدام البوت:",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(
text=f"@{ch.get('channel_username') or ch['channel_title']}",
url=link
)],
[InlineKeyboardButton(
text="✅ تحقق من الاشتراك",
callback_data="check_sub"
)],
])
)
else:
await cobot.send_message(
chat_id,
f"عزيزي المستخدم اشترك في القناة التالية:\n"
f"- {ch['channel_title']}\n\nاشترك ثم أرسل /start",
parse_mode="HTML", disable_web_page_preview=True)
await factory_bot_obj.session.close()
return False
was_requested = await is_cobot_funded_pending(factory_username, ch["id"], user_id, bot_record.id)
if was_requested:
added = await record_cobot_funded_sub(
factory_username, ch["id"], user_id, bot_record.id
)
await clear_cobot_funded_pending(factory_username, ch["id"], user_id, bot_record.id)
if added:
current, target = await get_cobot_funded_channel_count(
factory_username, ch["id"]
)
fund_notify = await fac_get_setting(factory_username, "cobots_fund_notify", "1")
if fund_notify == "1" and user_info:
name = f"@{user_info[0]}" if user_info[0] else user_info[1] or str(user_id)
link = ch.get("invite_link") or (
f"https://t.me/{ch['channel_username']}"
if ch.get("channel_username") else str(ch["channel_id"])
)
try:
await factory_bot_obj.send_message(
sub_factory.owner_id,
f"🔔 اشتراك جديد في قناة ممولة (بوتات مصنوعة):\n"
f"الحساب: {name}\n"
f"البوت: @{bot_record.username}\n"
f"القناة: {ch['channel_title']}\n"
f"عدد المشتركين: {current}/{target}",
parse_mode="HTML", disable_web_page_preview=True)
except Exception:
pass
if current >= target:
await mark_cobot_funded_complete(factory_username, ch["id"])
await clear_all_cobot_funded_pending(factory_username, ch["id"])
link = ch.get("invite_link") or str(ch["channel_id"])
try:
await factory_bot_obj.send_message(
sub_factory.owner_id,
f"✅ اكتملت قناة ممولة (بوتات مصنوعة)!\n"
f"القناة: {ch['channel_title']}\n"
f"تم الوصول إلى {target} عضو.",
parse_mode="HTML", disable_web_page_preview=True)
except Exception:
pass
finally:
try:
await factory_bot_obj.session.close()
except Exception:
pass
return True
async def check_main_factory_channels(
cobot: Bot, bot_record: BotModel, user_id: int, update_data: dict
) -> bool:
if bot_record.is_paid:
return True
chat_id = None
if update_data.get("message"):
chat_id = update_data["message"]["chat"]["id"]
elif update_data.get("callback_query"):
chat_id = update_data["callback_query"]["message"]["chat"]["id"]
try:
channels = await get_factory_cobot_mandatory_channels()
except Exception:
channels = []
factory_checker = Bot(token=FACTORY_TOKEN)
try:
for ch in channels:
try:
member = await factory_checker.get_chat_member(int(ch["channel_id"]), user_id)
not_subscribed = member.status in ("left", "kicked", "banned")
except Exception:
not_subscribed = True
if not_subscribed:
link = ch.get("invite_link") or (
f"https://t.me/{ch['channel_username']}"
if ch.get("channel_username") else "#"
)
if chat_id:
if ch.get("display_type") == "buttons":
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
await cobot.send_message(
chat_id,
"عزيزي المستخدم اشترك في القناة التالية قبل استخدام البوت:",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(
text=f"@{ch.get('channel_username') or ch['channel_title']}",
url=link
)],
[InlineKeyboardButton(
text="✅ تحقق من الاشتراك",
callback_data="check_sub"
)],
])
)
else:
await cobot.send_message(
chat_id,
f"عزيزي المستخدم اشترك في القناة التالية:\n"
f"- {ch['channel_title']}\n\nاشترك ثم أرسل /start",
parse_mode="HTML", disable_web_page_preview=True)
return False
finally:
await factory_checker.session.close()
try:
fund_channels = await get_factory_cobot_funded_channels()
except Exception:
fund_channels = []
if fund_channels:
u_username = u_first = None
if update_data.get("message") and update_data["message"].get("from"):
_u = update_data["message"]["from"]
u_username, u_first = _u.get("username"), _u.get("first_name", "")
elif update_data.get("callback_query") and update_data["callback_query"].get("from"):
_u = update_data["callback_query"]["from"]
u_username, u_first = _u.get("username"), _u.get("first_name", "")
fund_checker = Bot(token=FACTORY_TOKEN)
try:
for ch in fund_channels:
try:
member = await fund_checker.get_chat_member(int(ch["channel_id"]), user_id)
is_member = member.status not in ("left", "kicked", "banned")
except Exception:
is_member = False
if not is_member:
await mark_factory_funded_pending(ch["id"], user_id, bot_record.id)
link = ch.get("invite_link") or (
f"https://t.me/{ch['channel_username']}"
if ch.get("channel_username") else "#"
)
if chat_id:
if ch.get("display_type") == "buttons":
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
await cobot.send_message(
chat_id,
"عزيزي المستخدم اشترك في القناة التالية قبل استخدام البوت:",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(
text=f"@{ch.get('channel_username') or ch['channel_title']}",
url=link
)],
[InlineKeyboardButton(
text="✅ تحقق من الاشتراك",
callback_data="check_sub"
)],
])
)
else:
await cobot.send_message(
chat_id,
f"عزيزي المستخدم اشترك في القناة التالية:\n"
f"- {ch['channel_title']}\n\nاشترك ثم أرسل /start",
parse_mode="HTML", disable_web_page_preview=True)
return False
was_requested = await is_factory_funded_pending(ch["id"], user_id, bot_record.id)
if was_requested:
added = await record_factory_cobot_funded_sub(ch["id"], user_id, bot_record.id)
await clear_factory_funded_pending(ch["id"], user_id, bot_record.id)
if added:
current, target = await get_factory_cobot_funded_count(ch["id"])
name = f"@{u_username}" if u_username else (u_first or str(user_id))
link = ch.get("invite_link") or (
f"https://t.me/{ch['channel_username']}"
if ch.get("channel_username") else str(ch["channel_id"])
)
try:
await notify_devs(
fund_checker,
f"🔔 اشتراك جديد في قناة ممولة (بوتات مصنوعة):\n"
f"الحساب: {name}\n"
f"البوت: @{bot_record.username}\n"
f"القناة: {ch['channel_title']}\n"
f"عدد المشتركين: {current}/{target}")
except Exception:
pass
if current >= target:
await mark_factory_cobot_funded_complete(ch["id"])
await clear_all_factory_funded_pending(ch["id"])
try:
await notify_devs(
fund_checker,
f"✅ اكتملت قناة ممولة (بوتات مصنوعة)!\n"
f"القناة: {ch['channel_title']}\n"
f"تم الوصول إلى {target} عضو.")
except Exception:
pass
finally:
await fund_checker.session.close()
return True
async def handle_factory_update(data: dict, bot_record: BotModel,
cobot: Bot, factory_bot: Bot):
import cobots.factory.factory as fac_module
from aiogram.types import Update
fac_module.BOT_TOKEN = bot_record.token
fac_module.BOT_USERNAME = bot_record.username.lower()
fac_module.OWNER_ID = bot_record.owner_id
fac_module.FACTORY_TOKEN = factory_bot.token
fac_module.BOT_ID = bot_record.id
fac_module.PARENT_FACTORY_ID = bot_record.created_by_bot_id or 0
try:
from information.database import _init_payment_settings, _init_payment_packages
await _init_payment_settings(bot_record.id)
await _init_payment_packages(bot_record.id)
except Exception as _e:
logger.error(f"خطأ في تهيئة مدفوعات المصنع {bot_record.username}: {_e}")
update = Update.model_validate(data)
user = None
if update.message and update.message.from_user:
user = update.message.from_user
elif update.callback_query and update.callback_query.from_user:
user = update.callback_query.from_user
if user and not user.is_bot:
await register_cobot_user(
user_id = user.id,
username = user.username,
first_name = user.first_name,
bot_id = bot_record.id,
)
dp = fac_module.get_factory_dispatcher()
await dp.feed_update(cobot, update)
async def register_cobot_user(user_id: int, username: str | None, first_name: str, bot_id: int):
async with get_session() as session:
result = await session.execute(
select(CobotUser).where(
CobotUser.user_id == user_id,
CobotUser.bot_id == bot_id,
)
)
cu = result.scalar_one_or_none()
if not cu:
session.add(CobotUser(
user_id = user_id,
bot_id = bot_id,
username = username,
first_name = first_name,
))
else:
cu.last_active = datetime.utcnow()
cu.username = username
cu.first_name = first_name
async def record_cobot_funded_subscription(
cobot: Bot, user_id: int, channel: CobotFundedChannel,
bot_record: BotModel, first_name: str, username: str | None,
settings, factory_bot: Bot = None
):
async with get_session() as session:
result = await session.execute(
select(CobotFundedSubscription).where(
CobotFundedSubscription.cobot_funded_channel_id == channel.id,
CobotFundedSubscription.user_id == user_id,
CobotFundedSubscription.bot_id == bot_record.id,
)
)
if result.scalar_one_or_none():
return
session.add(CobotFundedSubscription(
cobot_funded_channel_id = channel.id,
user_id = user_id,
bot_id = bot_record.id,
))
ch_result = await session.execute(
select(CobotFundedChannel).where(CobotFundedChannel.id == channel.id)
)
ch = ch_result.scalar_one_or_none()
if not ch:
return
ch.current_count += 1
notify_bot = factory_bot or cobot
if settings.cobots_fund_notify:
name = f"@{username}" if username else first_name or str(user_id)
link = ch.invite_link or (
f"https://t.me/{ch.channel_username}" if ch.channel_username else str(ch.channel_id)
)
await notify_devs(
notify_bot,
f"🔔 اشتراك جديد في قناة ممولة (بوتات مصنوعة):\n"
f"الحساب: {name}\n"
f"البوت: @{bot_record.username}\n"
f"القناة: {ch.channel_title}\n"
f"عدد المشتركين: {ch.current_count}/{ch.target_count}"
)
if ch.current_count >= ch.target_count:
ch.is_active = False
ch.completed_at = datetime.utcnow()
link = ch.invite_link or str(ch.channel_id)
await notify_devs(
notify_bot,
f"✅ اكتملت قناة ممولة (بوتات مصنوعة)!\n"
f"القناة: {ch.channel_title}\n"
f"تم الوصول إلى {ch.target_count} عضو."
)
async def _check_expired_subscriptions(bot: Bot):
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
subs = await get_expiring_subscriptions(factory_bot_id=0)
for sub in subs:
bot_rec = await get_bot_by_id(sub.bot_id)
started = sub.started_at.strftime("%d-%m-%Y") if sub.started_at else "—"
expires = sub.expires_at.strftime("%d-%m-%Y") if sub.expires_at else "—"
await expire_bot_subscription(sub.bot_id)
if not bot_rec:
continue
try:
balance = await get_user_balance(sub.user_id)
await bot.send_message(
sub.user_id,
f"عزيزي المستخدم تم انتهاء اشتراك البوت الخاص بك.\n\n"
f"البوت: @{bot_rec.username}\n"
f"تاريخ الاشتراك: {started}\n"
f"تاريخ الانتهاء: {expires}\n"
f"حالة البوت: غير مدفوع حالياً\n\n"
f"يمكنك ادارة اشتراك البوت من خلال الازرار:",
parse_mode="HTML", disable_web_page_preview=True,
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text=f"💰 رصيد حسابك: {_fmt_usd(balance)}", callback_data="pay:noop")],
[await _dyn_btn("pay_subscribe", "pay:subscribe", "اشتراك مدفوع ⭐"),
await _dyn_btn("pay_topup", "pay:topup", "إضافة رصيد 💳")],
[InlineKeyboardButton(text="تواصل مع الدعم 📞", url="https://t.me/n_u_7")],
])
)
except Exception:
pass
all_subs = await get_expiring_subscriptions(factory_bot_id=None)
for sub in all_subs:
if not sub.factory_bot_id:
continue
bot_rec = await get_bot_by_id(sub.bot_id)
started = sub.started_at.strftime("%d-%m-%Y") if sub.started_at else "—"
expires = sub.expires_at.strftime("%d-%m-%Y") if sub.expires_at else "—"
await expire_bot_subscription(sub.bot_id)
if not bot_rec:
continue
factory_rec = await get_bot_by_id(sub.factory_bot_id)
if not factory_rec:
continue
try:
from aiogram import Bot as _Bot
from aiogram.client.default import DefaultBotProperties
fac_bot = _Bot(token=factory_rec.token, default=DefaultBotProperties(parse_mode="HTML"))
owner_rec = await get_bot_by_id(sub.factory_bot_id)
support_url = f"https://t.me/{factory_rec.username}"
balance = await get_user_balance(sub.user_id, sub.factory_bot_id)
await fac_bot.send_message(
sub.user_id,
f"عزيزي المستخدم تم انتهاء اشتراك البوت الخاص بك.\n\n"
f"البوت: @{bot_rec.username}\n"
f"تاريخ الاشتراك: {started}\n"
f"تاريخ الانتهاء: {expires}\n"
f"حالة البوت: غير مدفوع حالياً\n\n"
f"يمكنك ادارة اشتراك البوت من خلال الازرار:",
parse_mode="HTML", disable_web_page_preview=True,
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text=f"💰 رصيد حسابك: {_fmt_usd(balance)}", callback_data="fac:noop")],
[InlineKeyboardButton(text="اشتراك مدفوع ⭐", callback_data="fac:subscribe"),
InlineKeyboardButton(text="إضافة رصيد 💳", callback_data="fac:topup")],
[InlineKeyboardButton(text="تواصل مع الدعم 📞", url=support_url)],
])
)
await fac_bot.session.close()
except Exception:
pass
async def on_startup(app: web.Application):
bot: Bot = app["bot"]
await init_db()
scheduler = AsyncIOScheduler()
scheduler.add_job(_check_expired_subscriptions, "interval", hours=1, args=[bot])
scheduler.start()
app["scheduler"] = scheduler
await bot.set_webhook(
url = WEBHOOK_URL,
secret_token = WEBHOOK_SECRET,
allowed_updates = ["message", "callback_query", "my_chat_member", "edited_message", "managed_bot"],
)
async with get_session() as session:
result = await session.execute(
select(BotModel).where(BotModel.is_active == True, BotModel.bot_type == BotType.communication)
)
active_bots = result.scalars().all()
for bot_record in active_bots:
username = bot_record.username.lower()
get_com_dispatcher()
logger.info("✅ تم إنشاء Dispatcher مشترك لبوتات التواصل")
logger.info(f"✅ Webhook مُسجَّل: {WEBHOOK_URL}")
async def on_shutdown(app: web.Application):
bot: Bot = app["bot"]
if "scheduler" in app:
app["scheduler"].shutdown()
await bot.delete_webhook()
await bot.session.close()
logger.info("🛑 تم إيقاف البوت.")
def main():
bot = Bot(token=FACTORY_TOKEN)
dp = Dispatcher(storage=MemoryStorage())
dp.include_router(dev_router)
dp.include_router(router)
dp.update.register(on_managed_bot_update)
app = web.Application()
app["bot"] = bot
app.on_startup.append(on_startup)
app.on_shutdown.append(on_shutdown)
SimpleRequestHandler(
dispatcher = dp,
bot = bot,
secret_token = WEBHOOK_SECRET,
).register(app, path=WEBHOOK_PATH)
setup_application(app, dp, bot=bot)
app.router.add_post("/kodo/bots/{bot_username}", cobot_webhook_handler)
logger.info(f"🚀 بدء التشغيل على {WEBAPP_HOST}:{WEBAPP_PORT}")
web.run_app(app, host=WEBAPP_HOST, port=WEBAPP_PORT)
if __name__ == "__main__":
main()