From d921ab32acee81697f1e680958a8a066e92eca2a Mon Sep 17 00:00:00 2001 From: SasukeXDev Date: Fri, 17 Apr 2026 21:24:43 +0530 Subject: [PATCH 1/3] Add production-ready Telegram bot workflow with force-sub, search, and admin tooling --- README.md | 62 ++++++ bot/__main__.py | 2 + bot/config.py | 45 +++- bot/helper/bot_database.py | 191 +++++++++++++++++ bot/helper/retry.py | 26 +++ bot/telegram/keyboards.py | 29 +++ bot/telegram/plugins/start.py | 392 ++++++++++++++++++++++++++-------- bot/telegram/services.py | 105 +++++++++ requirements.txt | 1 + 9 files changed, 751 insertions(+), 102 deletions(-) create mode 100644 bot/helper/bot_database.py create mode 100644 bot/helper/retry.py create mode 100644 bot/telegram/keyboards.py create mode 100644 bot/telegram/services.py diff --git a/README.md b/README.md index ab96cb6..ef98371 100644 --- a/README.md +++ b/README.md @@ -275,3 +275,65 @@ will. Specifically you can redistribute and/or modify it under the terms of the [GNU Affero General Public License](https://www.gnu.org/licenses/agpl-3.0.en.html) as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. Also keep in mind that all the forks of this repository MUST BE OPEN-SOURCE and MUST BE UNDER THE SAME LICENSE. + +--- + +## Telegram Bot (Production Workflow) + +This repository now includes a production-ready **Pyrogram + MongoDB** bot workflow with: + +- Force subscription flow (`/start` + join check + retry button + pending content delivery) +- Movie/series keyword search with pagination and inline result buttons +- Admin commands: `/stats`, `/users`, `/broadcast` +- Mongo persistence for users, content index, global metrics, and broadcast logs +- Channel auto-indexing for new files and `/index` command for history indexing + +### Folder Structure (Telegram Features) + +```text +bot/ +├── config.py +├── helper/ +│ ├── bot_database.py +│ └── retry.py +└── telegram/ + ├── keyboards.py + ├── services.py + └── plugins/ + └── start.py +``` + +### Required Environment Variables + +```env +BOT_TOKEN= +MONGO_URI= # fallback: DATABASE_URL +FORCE_SUB_CHANNELS=@channelA,@channelB +START_IMAGE_URL=https://... +ADMIN_IDS=123456789,987654321 +``` + +Existing values like `AUTH_CHANNEL` remain supported for indexing and fallback force-sub channels. + +### Commands + +- `/start [payload]` → force-sub check + welcome photo + optional content delivery +- `/search ` or plain text in private chat → content search +- `/index` (in indexed channel) → import channel history +- `/stats` (admin) → users + request metrics +- `/users` (admin) → user count + preview +- `/broadcast` (admin, reply required) → mass broadcast with progress + +### Deployment (VPS / Render / Koyeb) + +1. Set env vars above and existing Telegram API variables (`API_ID`, `API_HASH`, `BOT_TOKEN`). +2. Install dependencies: + ```bash + pip install -r requirements.txt + ``` +3. Run: + ```bash + python -m bot + ``` +4. Ensure MongoDB is reachable and bot is admin in required channels. + diff --git a/bot/__main__.py b/bot/__main__.py index 5c1e368..b95cd30 100644 --- a/bot/__main__.py +++ b/bot/__main__.py @@ -6,6 +6,7 @@ from bot import __version__, LOGGER from bot.config import Telegram +from bot.helper.bot_database import bot_db from bot.server import web_server from bot.telegram import StreamBot, UserBot from bot.telegram.clients import initialize_clients @@ -14,6 +15,7 @@ async def start_services(): LOGGER.info(f'Initializing Surf-TG v-{__version__}') + await bot_db.ensure_indexes() await asleep(1.2) await StreamBot.start() diff --git a/bot/config.py b/bot/config.py index 96bf2f0..b112a31 100644 --- a/bot/config.py +++ b/bot/config.py @@ -1,25 +1,54 @@ from os import getenv -from dotenv import load_dotenv from pathlib import Path +from dotenv import load_dotenv + if Path("config.env").exists(): load_dotenv("config.env") + +def _csv_list(key: str, default: str = "") -> list[str]: + raw = getenv(key, default) + return [item.strip() for item in raw.split(",") if item.strip()] + + +def _int_list(key: str) -> list[int]: + values: list[int] = [] + for item in _csv_list(key): + try: + values.append(int(item)) + except ValueError: + continue + return values + + class Telegram: API_ID = int(getenv("API_ID", "0")) API_HASH = getenv("API_HASH", "") BOT_TOKEN = getenv("BOT_TOKEN", "") PORT = int(getenv("PORT", 8080)) SESSION_STRING = getenv("SESSION_STRING", "") - BASE_URL = getenv("BASE_URL", "").rstrip('/') - DATABASE_URL = getenv("DATABASE_URL", "") - AUTH_CHANNEL = [channel.strip() for channel in getenv("AUTH_CHANNEL", "").split(",") if channel.strip()] + + BASE_URL = getenv("BASE_URL", "").rstrip("/") + MONGO_URI = getenv("MONGO_URI") or getenv("DATABASE_URL", "") + + AUTH_CHANNEL = _csv_list("AUTH_CHANNEL") + FORCE_SUB_CHANNELS = _csv_list("FORCE_SUB_CHANNELS") or AUTH_CHANNEL + START_IMAGE_URL = getenv( + "START_IMAGE_URL", + "https://placehold.co/1280x720/png?text=Welcome+to+Movie+Bot", + ) + ADMIN_IDS = _int_list("ADMIN_IDS") + THEME = getenv("THEME", "vapor").lower() USERNAME = getenv("USERNAME", "admin") PASSWORD = getenv("PASSWORD", "admin") ADMIN_USERNAME = getenv("ADMIN_USERNAME", "surfTG") ADMIN_PASSWORD = getenv("ADMIN_PASSWORD", "surfTG") - SLEEP_THRESHOLD = int(getenv('SLEEP_THRESHOLD', '60')) - WORKERS = int(getenv('WORKERS', '10')) - MULTI_CLIENT = getenv('MULTI_CLIENT', 'False') - HIDE_CHANNEL = getenv('HIDE_CHANNEL', 'False') + + SLEEP_THRESHOLD = int(getenv("SLEEP_THRESHOLD", "60")) + WORKERS = int(getenv("WORKERS", "10")) + MULTI_CLIENT = getenv("MULTI_CLIENT", "False") + HIDE_CHANNEL = getenv("HIDE_CHANNEL", "False") + + SEARCH_PAGE_SIZE = int(getenv("SEARCH_PAGE_SIZE", "8")) diff --git a/bot/helper/bot_database.py b/bot/helper/bot_database.py new file mode 100644 index 0000000..eb51b46 --- /dev/null +++ b/bot/helper/bot_database.py @@ -0,0 +1,191 @@ +from __future__ import annotations + +import re +from datetime import datetime, timezone +from typing import Any + +from motor.motor_asyncio import AsyncIOMotorClient +from pymongo import DESCENDING, IndexModel + +from bot.config import Telegram + + +class BotDatabase: + """Async MongoDB helper for bot runtime data.""" + + def __init__(self) -> None: + self.client = AsyncIOMotorClient(Telegram.MONGO_URI) + self.db = self.client["surftg_bot"] + self.users = self.db["users"] + self.content = self.db["content"] + self.broadcast_logs = self.db["broadcast_logs"] + self.metrics = self.db["metrics"] + + async def ensure_indexes(self) -> None: + await self.users.create_indexes( + [ + IndexModel([("user_id", DESCENDING)], unique=True), + IndexModel([("last_seen", DESCENDING)]), + ] + ) + await self.content.create_indexes( + [ + IndexModel([("chat_id", DESCENDING), ("msg_id", DESCENDING)], unique=True), + IndexModel([("title", "text"), ("tags", "text")]), + IndexModel([("created_at", DESCENDING)]), + ] + ) + await self.broadcast_logs.create_index([("created_at", DESCENDING)]) + + async def upsert_user(self, user_id: int, username: str | None = None) -> None: + now = datetime.now(timezone.utc) + await self.users.update_one( + {"user_id": user_id}, + { + "$set": { + "username": username, + "last_seen": now, + }, + "$setOnInsert": { + "created_at": now, + "join_verified": False, + "requests_total": 0, + "success_total": 0, + "failure_total": 0, + }, + }, + upsert=True, + ) + + async def set_join_status(self, user_id: int, join_verified: bool) -> None: + await self.users.update_one({"user_id": user_id}, {"$set": {"join_verified": join_verified}}) + + async def set_pending_request(self, user_id: int, payload: str | None) -> None: + await self.users.update_one({"user_id": user_id}, {"$set": {"pending_request": payload}}) + + async def get_user(self, user_id: int) -> dict[str, Any] | None: + return await self.users.find_one({"user_id": user_id}) + + async def add_content( + self, + chat_id: int, + msg_id: int, + file_id: str, + file_unique_id: str, + title: str, + tags: list[str], + mime_type: str | None, + file_size: int | None, + ) -> None: + now = datetime.now(timezone.utc) + await self.content.update_one( + {"chat_id": chat_id, "msg_id": msg_id}, + { + "$set": { + "file_id": file_id, + "file_unique_id": file_unique_id, + "title": title, + "tags": tags, + "mime_type": mime_type, + "file_size": file_size, + "created_at": now, + } + }, + upsert=True, + ) + + async def search_content(self, query: str, page: int, page_size: int) -> tuple[list[dict[str, Any]], int]: + words = re.findall(r"\w+", query.lower()) + filters: dict[str, Any] = {} + if words: + regex_parts = [f"(?=.*{re.escape(word)})" for word in words] + filters["title"] = {"$regex": "".join(regex_parts), "$options": "i"} + + total = await self.content.count_documents(filters) + cursor = ( + self.content.find(filters, {"_id": 0}) + .sort("created_at", DESCENDING) + .skip((page - 1) * page_size) + .limit(page_size) + ) + docs = await cursor.to_list(length=page_size) + return docs, total + + async def log_request(self, user_id: int, success: bool) -> None: + await self.users.update_one( + {"user_id": user_id}, + { + "$inc": { + "requests_total": 1, + "success_total": 1 if success else 0, + "failure_total": 0 if success else 1, + } + }, + ) + await self.metrics.update_one( + {"_id": "global"}, + { + "$inc": { + "total_requests": 1, + "success_total": 1 if success else 0, + "failure_total": 0 if success else 1, + } + }, + upsert=True, + ) + + async def create_broadcast_log(self, admin_id: int, source_message_id: int, total: int) -> str: + now = datetime.now(timezone.utc) + doc = { + "admin_id": admin_id, + "source_message_id": source_message_id, + "total": total, + "sent": 0, + "failed": 0, + "pending": total, + "status": "running", + "created_at": now, + "updated_at": now, + "failures": [], + } + result = await self.broadcast_logs.insert_one(doc) + return str(result.inserted_id) + + async def update_broadcast_log(self, log_id: str, sent: int, failed: int, pending: int, failures: list[dict[str, Any]]) -> None: + await self.broadcast_logs.update_one( + {"_id": self._oid(log_id)}, + { + "$set": { + "sent": sent, + "failed": failed, + "pending": pending, + "failures": failures, + "status": "completed" if pending == 0 else "running", + "updated_at": datetime.now(timezone.utc), + } + }, + ) + + async def get_stats(self) -> dict[str, int]: + total_users = await self.users.count_documents({}) + active_users = await self.users.count_documents({"join_verified": True}) + global_metrics = await self.metrics.find_one({"_id": "global"}) or {} + return { + "total_users": total_users, + "active_users": active_users, + "total_requests": global_metrics.get("total_requests", 0), + "success_total": global_metrics.get("success_total", 0), + "failure_total": global_metrics.get("failure_total", 0), + } + + async def list_user_ids(self) -> list[int]: + return [doc["user_id"] async for doc in self.users.find({}, {"user_id": 1, "_id": 0})] + + @staticmethod + def _oid(value: str): + from bson import ObjectId + + return ObjectId(value) + + +bot_db = BotDatabase() diff --git a/bot/helper/retry.py b/bot/helper/retry.py new file mode 100644 index 0000000..ee99161 --- /dev/null +++ b/bot/helper/retry.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +from asyncio import sleep +from typing import Any, Awaitable, Callable + +from pyrogram.errors import FloodWait, RPCError + + +async def tg_retry( + func: Callable[..., Awaitable[Any]], + *args: Any, + retries: int = 3, + base_delay: float = 1.2, + **kwargs: Any, +) -> Any: + attempt = 0 + while True: + try: + return await func(*args, **kwargs) + except FloodWait as err: + await sleep(err.value + 1) + except RPCError: + attempt += 1 + if attempt > retries: + raise + await sleep(base_delay * attempt) diff --git a/bot/telegram/keyboards.py b/bot/telegram/keyboards.py new file mode 100644 index 0000000..621ff11 --- /dev/null +++ b/bot/telegram/keyboards.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +from pyrogram.types import InlineKeyboardButton, InlineKeyboardMarkup + + +def force_sub_keyboard(channels: list[str]) -> InlineKeyboardMarkup: + join_buttons = [] + for channel in channels: + label = channel.replace("@", "") + if channel.startswith("-100"): + continue + join_buttons.append([InlineKeyboardButton(f"📢 Join {label}", url=f"https://t.me/{label}")]) + join_buttons.append([InlineKeyboardButton("✅ Try Again", callback_data="fs:try_again")]) + return InlineKeyboardMarkup(join_buttons) + + +def search_results_keyboard(items: list[dict], page: int, total_pages: int, query_id: str) -> InlineKeyboardMarkup: + rows = [ + [InlineKeyboardButton(f"🎬 {item['title'][:45]}", callback_data=f"content:{item['chat_id']}:{item['msg_id']}")] + for item in items + ] + nav_row = [] + if page > 1: + nav_row.append(InlineKeyboardButton("⬅️ Prev", callback_data=f"sp:{query_id}:{page-1}")) + if page < total_pages: + nav_row.append(InlineKeyboardButton("Next ➡️", callback_data=f"sp:{query_id}:{page+1}")) + if nav_row: + rows.append(nav_row) + return InlineKeyboardMarkup(rows) diff --git a/bot/telegram/plugins/start.py b/bot/telegram/plugins/start.py index 99e8cb9..455546a 100644 --- a/bot/telegram/plugins/start.py +++ b/bot/telegram/plugins/start.py @@ -1,101 +1,305 @@ +from __future__ import annotations + import re +from datetime import datetime, timedelta, timezone + +from pyrogram import Client, filters +from pyrogram.errors import RPCError +from pyrogram.types import CallbackQuery, Message + from bot import LOGGER from bot.config import Telegram -from bot.helper.database import Database -from bot.helper.file_size import get_readable_file_size -from bot.helper.index import get_messages -from bot.helper.media import is_media +from bot.helper.bot_database import bot_db +from bot.helper.retry import tg_retry from bot.telegram import StreamBot -from pyrogram import filters, Client -from pyrogram.types import Message -from os.path import splitext -from pyrogram.errors import FloodWait -from pyrogram.enums.parse_mode import ParseMode -from asyncio import sleep - -db = Database() - - -@StreamBot.on_message(filters.command('start') & filters.private) -async def start(bot: Client, message: Message): - if "file_" in message.text: - try: - usr_cmd = message.text.split("_")[-1] - data = usr_cmd.split("-") - message_id, chat_id = data[0], f"-{data[1]}" - file = await bot.get_messages(int(chat_id), int(message_id)) - media = is_media(file) - await message.reply_cached_media(file_id=media.file_id, caption=f'**{media.file_name}**') - except Exception as e: - print(f"An error occurred: {e}") - - -@StreamBot.on_message(filters.command('index')) -async def start(bot: Client, message: Message): - channel_id = message.chat.id - AUTH_CHANNEL = await db.get_variable('auth_channel') - if AUTH_CHANNEL is None or AUTH_CHANNEL.strip() == '': - AUTH_CHANNEL = Telegram.AUTH_CHANNEL - else: - AUTH_CHANNEL = [channel.strip() for channel in AUTH_CHANNEL.split(",")] - if str(channel_id) in AUTH_CHANNEL: - try: - last_id = message.id - start_message = ( - "🔄 Please perform this action only once at the beginning of Surf-Tg usage.\n\n" - "📋 File listing is currently in progress.\n\n" - "🚫 Please refrain from sending any additional files or indexing other channels until this process completes.\n\n" - "⏳ Please be patient and wait a few moments." - ) +from bot.telegram.services import ( + build_search_page, + build_search_response, + deliver_payload, + is_user_subscribed, + send_force_sub_prompt, +) + + +async def safe_error_reply(target: Message | CallbackQuery, err: Exception) -> None: + text = f"Error: {str(err)}" + try: + if isinstance(target, CallbackQuery): + await target.answer(text, show_alert=True) + else: + await target.reply_text(text) + except Exception: + LOGGER.exception("Failed to deliver error reply") + + +def is_admin(_, __, message: Message) -> bool: + return bool(message.from_user and message.from_user.id in Telegram.ADMIN_IDS) + + +admin_filter = filters.create(is_admin) + + +@StreamBot.on_message(filters.private & filters.command("start")) +async def start_handler(client: Client, message: Message) -> None: + try: + user = message.from_user + if not user: + return + + await bot_db.upsert_user(user.id, user.username) + payload = message.text.split(maxsplit=1)[1].strip() if len(message.command) > 1 else None + + is_subscribed = await is_user_subscribed(client, user.id) + await bot_db.set_join_status(user.id, is_subscribed) + + if not is_subscribed: + await bot_db.set_pending_request(user.id, payload) + await send_force_sub_prompt(message) + return + + caption = ( + f"👋 Welcome {user.mention}!\n\n" + "🔎 Send movie/series name to search content instantly.\n" + "⚡ Fast indexed delivery with pagination support.\n\n" + "Use /search <keyword> for direct lookup." + ) + await tg_retry( + message.reply_photo, + photo=Telegram.START_IMAGE_URL, + caption=caption, + ) + + if payload: + sent = await deliver_payload(client, message, payload) + await bot_db.log_request(user.id, sent) + if sent: + await bot_db.set_pending_request(user.id, None) + except Exception as err: + await safe_error_reply(message, err) + + +@StreamBot.on_callback_query(filters.regex(r"^fs:try_again$")) +async def force_sub_try_again(client: Client, query: CallbackQuery) -> None: + try: + user = query.from_user + if not user: + return + subscribed = await is_user_subscribed(client, user.id) + await bot_db.set_join_status(user.id, subscribed) + + if not subscribed: + await query.answer("You still need to join all required channels.", show_alert=True) + return + + user_doc = await bot_db.get_user(user.id) or {} + pending = user_doc.get("pending_request") + await query.answer("✅ Subscription verified") + await query.message.edit_text("✅ Subscription verified! Sending your content...") + if pending: + sent = await deliver_payload(client, query.message, pending) + await bot_db.log_request(user.id, sent) + if sent: + await bot_db.set_pending_request(user.id, None) + except Exception as err: + await safe_error_reply(query, err) + + +@StreamBot.on_message(filters.private & (filters.command("search") | (filters.text & ~filters.command(["start", "stats", "users", "broadcast"])))) +async def search_handler(client: Client, message: Message) -> None: + try: + if not message.from_user: + return + if not await is_user_subscribed(client, message.from_user.id): + await send_force_sub_prompt(message) + return + + query = " ".join(message.command[1:]).strip() if message.command else (message.text or "").strip() + if not query: + await message.reply_text("Usage: /search ") + return + + text, keyboard = await build_search_response(query, page=1) + await message.reply_text(text, reply_markup=keyboard) + except Exception as err: + await safe_error_reply(message, err) + - wait_msg = await message.reply(text=start_message) - files = await get_messages(message.chat.id, 1, last_id) - await db.add_btgfiles(files) - await wait_msg.delete() - done_message = ( - "✅ All your files have been successfully stored in the database. You're all set!\n\n" - "📁 You don't need to index again unless you make changes to the database." +@StreamBot.on_callback_query(filters.regex(r"^sp:")) +async def search_page_handler(_, query: CallbackQuery) -> None: + try: + _, query_id, page = query.data.split(":") + page_data = await build_search_page(query_id, int(page)) + if not page_data: + await query.answer("Search session expired. Run search again.", show_alert=True) + return + text, keyboard = page_data + await query.message.edit_text(text, reply_markup=keyboard) + await query.answer() + except Exception as err: + await safe_error_reply(query, err) + + +@StreamBot.on_callback_query(filters.regex(r"^content:")) +async def content_delivery_handler(client: Client, query: CallbackQuery) -> None: + try: + user = query.from_user + if not user: + return + + if not await is_user_subscribed(client, user.id): + await bot_db.set_pending_request(user.id, query.data) + await query.answer("Join required channels first.", show_alert=True) + await send_force_sub_prompt(query.message) + return + + sent = await deliver_payload(client, query.message, query.data) + await bot_db.log_request(user.id, sent) + await query.answer("Sent ✅" if sent else "Unable to send", show_alert=not sent) + except Exception as err: + await safe_error_reply(query, err) + + +@StreamBot.on_message(filters.channel & (filters.document | filters.video)) +async def auto_index_channel_posts(_, message: Message) -> None: + try: + auth_channels = {str(ch) for ch in (Telegram.AUTH_CHANNEL or [])} + if auth_channels and str(message.chat.id) not in auth_channels: + return + + file = message.document or message.video + if not file: + return + + raw_title = file.file_name or message.caption or file.file_unique_id + title = re.sub(r"[.,|_'-]", " ", raw_title).strip() + tags = [token.lower() for token in re.findall(r"\w+", title)[:20]] + await bot_db.add_content( + chat_id=message.chat.id, + msg_id=message.id, + file_id=file.file_id, + file_unique_id=file.file_unique_id, + title=title, + tags=tags, + mime_type=file.mime_type, + file_size=file.file_size, + ) + except Exception: + LOGGER.exception("Failed to index channel post") + + +@StreamBot.on_message(filters.command("index") & filters.channel) +async def index_history(client: Client, message: Message) -> None: + try: + auth_channels = {str(ch) for ch in (Telegram.AUTH_CHANNEL or [])} + if auth_channels and str(message.chat.id) not in auth_channels: + await message.reply_text("Channel is not in AUTH_CHANNEL") + return + + wait = await message.reply_text("🔄 Indexing previous messages. Please wait...") + imported = 0 + async for msg in client.get_chat_history(message.chat.id, limit=5000): + file = msg.document or msg.video + if not file: + continue + raw_title = file.file_name or msg.caption or file.file_unique_id + title = re.sub(r"[.,|_'-]", " ", raw_title).strip() + tags = [token.lower() for token in re.findall(r"\w+", title)[:20]] + await bot_db.add_content( + chat_id=msg.chat.id, + msg_id=msg.id, + file_id=file.file_id, + file_unique_id=file.file_unique_id, + title=title, + tags=tags, + mime_type=file.mime_type, + file_size=file.file_size, ) + imported += 1 - await bot.send_message(chat_id=message.chat.id, text=done_message) - except FloodWait as e: - LOGGER.info(f"Sleeping for {str(e.value)}s") - await sleep(e.value) - await message.reply(text=f"Got Floodwait of {str(e.value)}s", - disable_web_page_preview=True, parse_mode=ParseMode.MARKDOWN) - else: - await message.reply(text="Channel is not in AUTH_CHANNEL") - - -@StreamBot.on_message( - filters.channel - & ( - filters.document - | filters.video - ) -) -async def file_receive_handler(bot: Client, message: Message): - channel_id = message.chat.id - AUTH_CHANNEL = await db.get_variable('auth_channel') - if AUTH_CHANNEL is None or AUTH_CHANNEL.strip() == '': - AUTH_CHANNEL = Telegram.AUTH_CHANNEL - else: - AUTH_CHANNEL = [channel.strip() for channel in AUTH_CHANNEL.split(",")] - if str(channel_id) in AUTH_CHANNEL: - try: - file = message.video or message.document - title = file.file_name or message.caption or file.file_id - title, _ = splitext(title) - title = re.sub(r'[.,|_\',]', ' ', title) - msg_id = message.id - hash = file.file_unique_id[:6] - size = get_readable_file_size(file.file_size) - type = file.mime_type - await db.add_tgfiles(str(channel_id), str(msg_id), str(hash), str(title), str(size), str(type)) - except FloodWait as e: - LOGGER.info(f"Sleeping for {str(e.value)}s") - await sleep(e.value) - await message.reply(text=f"Got Floodwait of {str(e.value)}s", - disable_web_page_preview=True, parse_mode=ParseMode.MARKDOWN) - else: - await message.reply(text="Channel is not in AUTH_CHANNEL") + await wait.edit_text(f"✅ Indexing completed. Imported {imported} files.") + except Exception as err: + await safe_error_reply(message, err) + + +@StreamBot.on_message(filters.private & filters.command("stats") & admin_filter) +async def stats_handler(_, message: Message) -> None: + try: + stats = await bot_db.get_stats() + text = ( + "📊 Bot Statistics\n\n" + f"👥 Total users: {stats['total_users']}\n" + f"✅ Active users: {stats['active_users']}\n" + f"📨 Total requests: {stats['total_requests']}\n" + f"🎯 Success: {stats['success_total']}\n" + f"❌ Failed: {stats['failure_total']}" + ) + await message.reply_text(text) + except Exception as err: + await safe_error_reply(message, err) + + +@StreamBot.on_message(filters.private & filters.command("users") & admin_filter) +async def users_handler(_, message: Message) -> None: + try: + users = await bot_db.list_user_ids() + preview = ", ".join(map(str, users[:20])) if users else "No users" + await message.reply_text(f"👥 Total users: {len(users)}\n{preview}") + except Exception as err: + await safe_error_reply(message, err) + + +@StreamBot.on_message(filters.private & filters.command("broadcast") & admin_filter) +async def broadcast_handler(client: Client, message: Message) -> None: + try: + if not message.reply_to_message: + await message.reply_text("Reply to a message with /broadcast") + return + + user_ids = await bot_db.list_user_ids() + total = len(user_ids) + if total == 0: + await message.reply_text("No users available for broadcast.") + return + + progress = await message.reply_text(f"📢 Broadcast started for {total} users...") + log_id = await bot_db.create_broadcast_log(message.from_user.id, message.reply_to_message.id, total) + + sent = failed = 0 + failures = [] + for index, user_id in enumerate(user_ids, start=1): + try: + await tg_retry(message.reply_to_message.copy, chat_id=user_id) + sent += 1 + except RPCError as err: + failed += 1 + failures.append({"user_id": user_id, "error": str(err)}) + if index % 50 == 0 or index == total: + pending = total - index + await bot_db.update_broadcast_log(log_id, sent, failed, pending, failures[-100:]) + await progress.edit_text( + f"📢 Broadcast Progress\nTotal: {total}\nSent: {sent}\nFailed: {failed}\nPending: {pending}" + ) + + await progress.edit_text(f"✅ Broadcast completed.\nSent: {sent}\nFailed: {failed}") + except Exception as err: + await safe_error_reply(message, err) + + +@StreamBot.on_message(filters.private & filters.command(["stats", "users", "broadcast"])) +async def admin_blocked_reply(_, message: Message) -> None: + if message.from_user and message.from_user.id not in Telegram.ADMIN_IDS: + await message.reply_text("❌ Admin only command.") + + +@StreamBot.on_message(filters.private) +async def user_touchpoint(_, message: Message) -> None: + """Low priority tracker to keep user activity fresh without spam.""" + + if not message.from_user: + return + await bot_db.upsert_user(message.from_user.id, message.from_user.username) + if datetime.now(timezone.utc).minute % 15 == 0: + await bot_db.users.update_one( + {"user_id": message.from_user.id}, + {"$set": {"last_ping": datetime.now(timezone.utc) + timedelta(minutes=15)}}, + ) diff --git a/bot/telegram/services.py b/bot/telegram/services.py new file mode 100644 index 0000000..0c10830 --- /dev/null +++ b/bot/telegram/services.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from math import ceil +from uuid import uuid4 + +from pyrogram import Client +from pyrogram.enums import ChatMemberStatus +from pyrogram.types import Message + +from bot.config import Telegram +from bot.helper.bot_database import bot_db +from bot.helper.retry import tg_retry +from bot.telegram.keyboards import force_sub_keyboard, search_results_keyboard + + +@dataclass +class SearchState: + query: str + expires_at: datetime + + +SEARCH_CACHE: dict[str, SearchState] = {} + + +def _prune_cache() -> None: + now = datetime.now(timezone.utc) + expired = [k for k, v in SEARCH_CACHE.items() if v.expires_at < now] + for key in expired: + SEARCH_CACHE.pop(key, None) + + +async def is_user_subscribed(client: Client, user_id: int) -> bool: + channels = Telegram.FORCE_SUB_CHANNELS + if not channels: + return True + + for channel in channels: + try: + member = await tg_retry(client.get_chat_member, channel, user_id) + except Exception: + return False + if member.status in {ChatMemberStatus.BANNED, ChatMemberStatus.LEFT}: + return False + return True + + +async def send_force_sub_prompt(message: Message) -> None: + text = ( + "🔒 Subscription Required\n\n" + "Please join all required channels, then tap Try Again to continue." + ) + await message.reply_text(text, reply_markup=force_sub_keyboard(Telegram.FORCE_SUB_CHANNELS)) + + +async def deliver_payload(client: Client, message: Message, payload: str) -> bool: + if payload.startswith("file_"): + parts = payload.replace("file_", "").split("-") + if len(parts) != 2: + return False + msg_id = int(parts[0]) + chat_id = int(f"-{parts[1]}") + await tg_retry(client.copy_message, message.chat.id, chat_id, msg_id) + return True + + if payload.startswith("content:"): + _, chat_id, msg_id = payload.split(":", 2) + await tg_retry(client.copy_message, message.chat.id, int(chat_id), int(msg_id)) + return True + + return False + + +async def build_search_response(query: str, page: int) -> tuple[str, object]: + items, total = await bot_db.search_content(query, page, Telegram.SEARCH_PAGE_SIZE) + if total == 0: + return "❌ No results found. Try another keyword.", None + + total_pages = ceil(total / Telegram.SEARCH_PAGE_SIZE) + query_id = uuid4().hex[:10] + _prune_cache() + SEARCH_CACHE[query_id] = SearchState( + query=query, + expires_at=datetime.now(timezone.utc) + timedelta(minutes=10), + ) + text = f"🔎 Results for: {query}\nPage {page}/{total_pages}" + keyboard = search_results_keyboard(items, page, total_pages, query_id) + return text, keyboard + + +async def build_search_page(query_id: str, page: int) -> tuple[str, object] | None: + _prune_cache() + state = SEARCH_CACHE.get(query_id) + if not state: + return None + + items, total = await bot_db.search_content(state.query, page, Telegram.SEARCH_PAGE_SIZE) + if total == 0: + return None + + total_pages = ceil(total / Telegram.SEARCH_PAGE_SIZE) + text = f"🔎 Results for: {state.query}\nPage {page}/{total_pages}" + keyboard = search_results_keyboard(items, page, total_pages, query_id) + return text, keyboard diff --git a/requirements.txt b/requirements.txt index fa571a1..c873217 100644 --- a/requirements.txt +++ b/requirements.txt @@ -11,3 +11,4 @@ pyrogram==2.0.106 tmdbv3api requests +motor From 73a21db5c399b508c503b05eecab5f2f5c53e92e Mon Sep 17 00:00:00 2001 From: SasukeXDev Date: Fri, 17 Apr 2026 22:02:17 +0530 Subject: [PATCH 2/3] Fix broken bot handlers and enforce @Stream4u_bot restriction --- README.md | 2 + bot/__main__.py | 58 +++++++------ bot/config.py | 3 + bot/helper/bot_database.py | 8 +- bot/telegram/clients.py | 35 ++++---- bot/telegram/keyboards.py | 31 ++++--- bot/telegram/plugins/start.py | 158 +++++++++++++++++++--------------- bot/telegram/services.py | 46 +++++++--- 8 files changed, 212 insertions(+), 129 deletions(-) diff --git a/README.md b/README.md index ef98371..2fefacc 100644 --- a/README.md +++ b/README.md @@ -324,6 +324,8 @@ Existing values like `AUTH_CHANNEL` remain supported for indexing and fallback f - `/users` (admin) → user count + preview - `/broadcast` (admin, reply required) → mass broadcast with progress +> Security restriction: bot startup is locked to **@Stream4u_bot**. If token/username does not match, the process exits. + ### Deployment (VPS / Render / Koyeb) 1. Set env vars above and existing Telegram API variables (`API_ID`, `API_HASH`, `BOT_TOKEN`). diff --git a/bot/__main__.py b/bot/__main__.py index b95cd30..ff9e1f1 100644 --- a/bot/__main__.py +++ b/bot/__main__.py @@ -1,4 +1,4 @@ -from asyncio import get_event_loop, sleep as asleep, gather +from asyncio import get_event_loop, sleep as asleep from traceback import format_exc from aiohttp import web @@ -13,49 +13,59 @@ loop = get_event_loop() -async def start_services(): - LOGGER.info(f'Initializing Surf-TG v-{__version__}') + +def _validate_restricted_bot() -> None: + username = (StreamBot.me.username or "").lower() + if username != Telegram.REQUIRED_BOT_USERNAME: + raise RuntimeError( + f"Restricted bot check failed. Expected @{Telegram.REQUIRED_BOT_USERNAME}, got @{username or 'unknown'}." + ) + + token_prefix = Telegram.BOT_TOKEN.split(":", 1)[0].strip() + if token_prefix and token_prefix != str(StreamBot.me.id): + raise RuntimeError("BOT_TOKEN does not belong to the connected bot account.") + + +async def start_services() -> None: + LOGGER.info(f"Initializing Surf-TG v-{__version__}") await bot_db.ensure_indexes() - await asleep(1.2) - + await asleep(0.8) + await StreamBot.start() + _validate_restricted_bot() StreamBot.username = StreamBot.me.username LOGGER.info(f"Bot Client : [@{StreamBot.username}]") - if len(Telegram.SESSION_STRING) != 0: + + if Telegram.SESSION_STRING: await UserBot.start() UserBot.username = UserBot.me.username or UserBot.me.first_name or UserBot.me.id LOGGER.info(f"User Client : {UserBot.username}") - - await asleep(1.2) + + await asleep(0.8) LOGGER.info("Initializing Multi Clients") await initialize_clients() - - await asleep(2) - LOGGER.info('Initalizing Surf Web Server..') + + LOGGER.info("Initializing Surf Web Server..") server = web.AppRunner(await web_server()) - LOGGER.info("Server CleanUp!") - await server.cleanup() - - await asleep(2) - LOGGER.info("Server Setup Started !") - await server.setup() - await web.TCPSite(server, '0.0.0.0', Telegram.PORT).start() + await web.TCPSite(server, "0.0.0.0", Telegram.PORT).start() - LOGGER.info("Surf-TG Started Revolving !") + LOGGER.info("Surf-TG Started") await idle() -async def stop_clients(): - await StreamBot.stop() - if len(Telegram.SESSION_STRING) != 0: + +async def stop_clients() -> None: + if StreamBot.is_connected: + await StreamBot.stop() + if Telegram.SESSION_STRING and UserBot.is_connected: await UserBot.stop() -if __name__ == '__main__': +if __name__ == "__main__": try: loop.run_until_complete(start_services()) except KeyboardInterrupt: - LOGGER.info('Service Stopping...') + LOGGER.info("Service Stopping...") except Exception: LOGGER.error(format_exc()) finally: diff --git a/bot/config.py b/bot/config.py index b112a31..745090c 100644 --- a/bot/config.py +++ b/bot/config.py @@ -31,6 +31,7 @@ class Telegram: BASE_URL = getenv("BASE_URL", "").rstrip("/") MONGO_URI = getenv("MONGO_URI") or getenv("DATABASE_URL", "") + DATABASE_URL = MONGO_URI # backward compatibility for existing modules AUTH_CHANNEL = _csv_list("AUTH_CHANNEL") FORCE_SUB_CHANNELS = _csv_list("FORCE_SUB_CHANNELS") or AUTH_CHANNEL @@ -40,6 +41,8 @@ class Telegram: ) ADMIN_IDS = _int_list("ADMIN_IDS") + REQUIRED_BOT_USERNAME = "stream4u_bot" + THEME = getenv("THEME", "vapor").lower() USERNAME = getenv("USERNAME", "admin") PASSWORD = getenv("PASSWORD", "admin") diff --git a/bot/helper/bot_database.py b/bot/helper/bot_database.py index eb51b46..3ebbdfa 100644 --- a/bot/helper/bot_database.py +++ b/bot/helper/bot_database.py @@ -14,6 +14,8 @@ class BotDatabase: """Async MongoDB helper for bot runtime data.""" def __init__(self) -> None: + if not Telegram.MONGO_URI: + raise ValueError("MONGO_URI/DATABASE_URL is required") self.client = AsyncIOMotorClient(Telegram.MONGO_URI) self.db = self.client["surftg_bot"] self.users = self.db["users"] @@ -99,7 +101,11 @@ async def search_content(self, query: str, page: int, page_size: int) -> tuple[l filters: dict[str, Any] = {} if words: regex_parts = [f"(?=.*{re.escape(word)})" for word in words] - filters["title"] = {"$regex": "".join(regex_parts), "$options": "i"} + pattern = "".join(regex_parts) + filters["$or"] = [ + {"title": {"$regex": pattern, "$options": "i"}}, + {"tags": {"$regex": pattern, "$options": "i"}}, + ] total = await self.content.count_documents(filters) cursor = ( diff --git a/bot/telegram/clients.py b/bot/telegram/clients.py index de11c5a..eacf9cb 100644 --- a/bot/telegram/clients.py +++ b/bot/telegram/clients.py @@ -1,24 +1,25 @@ -from asyncio import sleep as asleep, gather +from asyncio import gather, sleep as asleep + from pyrogram import Client from bot import LOGGER from bot.config import Telegram from bot.helper.parser import TokenParser -from bot.telegram import multi_clients, work_loads, StreamBot +from bot.telegram import StreamBot, multi_clients, work_loads -async def initialize_clients(): +async def initialize_clients() -> None: multi_clients[0], work_loads[0] = StreamBot, 0 all_tokens = TokenParser().parse_from_env() if not all_tokens: - LOGGER.info("No additional Bot Clients found, Using default client") + LOGGER.info("No additional Bot Clients found, using default client") return - async def start_client(client_id, token): + async def start_client(client_id: int, token: str): try: - LOGGER.info(f"Starting - Bot Client {client_id}") + LOGGER.info(f"Starting Bot Client {client_id}") if client_id == len(all_tokens): - await asleep(2) + await asleep(1.2) client = await Client( name=str(client_id), api_id=Telegram.API_ID, @@ -26,19 +27,23 @@ async def start_client(client_id, token): bot_token=token, sleep_threshold=Telegram.SLEEP_THRESHOLD, no_updates=True, - in_memory=True + in_memory=True, ).start() work_loads[client_id] = 0 return client_id, client except Exception: - LOGGER.error( - f"Failed starting Client - {client_id} Error:", exc_info=True) + LOGGER.error(f"Failed starting Client {client_id}", exc_info=True) + return None + + started = await gather(*[start_client(i, token) for i, token in all_tokens.items()]) + for item in started: + if item is None: + continue + idx, cli = item + multi_clients[idx] = cli - clients = await gather(*[start_client(i, token) for i, token in all_tokens.items()]) - multi_clients.update(dict(clients)) - if len(multi_clients) != 1: + if len(multi_clients) > 1: Telegram.MULTI_CLIENT = True LOGGER.info("Multi-Client Mode Enabled") else: - LOGGER.info( - "No additional clients were initialized, using default client") + LOGGER.info("No additional clients initialized, using default client") diff --git a/bot/telegram/keyboards.py b/bot/telegram/keyboards.py index 621ff11..d73ba08 100644 --- a/bot/telegram/keyboards.py +++ b/bot/telegram/keyboards.py @@ -3,15 +3,24 @@ from pyrogram.types import InlineKeyboardButton, InlineKeyboardMarkup +def _channel_button(channel: str) -> InlineKeyboardButton | None: + if channel.startswith("https://t.me/"): + label = channel.rsplit("/", 1)[-1] + return InlineKeyboardButton(f"📢 Join {label}", url=channel) + if channel.startswith("@"): + label = channel[1:] + return InlineKeyboardButton(f"📢 Join {label}", url=f"https://t.me/{label}") + return None + + def force_sub_keyboard(channels: list[str]) -> InlineKeyboardMarkup: - join_buttons = [] + rows: list[list[InlineKeyboardButton]] = [] for channel in channels: - label = channel.replace("@", "") - if channel.startswith("-100"): - continue - join_buttons.append([InlineKeyboardButton(f"📢 Join {label}", url=f"https://t.me/{label}")]) - join_buttons.append([InlineKeyboardButton("✅ Try Again", callback_data="fs:try_again")]) - return InlineKeyboardMarkup(join_buttons) + btn = _channel_button(channel) + if btn: + rows.append([btn]) + rows.append([InlineKeyboardButton("✅ Try Again", callback_data="fs:try_again")]) + return InlineKeyboardMarkup(rows) def search_results_keyboard(items: list[dict], page: int, total_pages: int, query_id: str) -> InlineKeyboardMarkup: @@ -19,11 +28,13 @@ def search_results_keyboard(items: list[dict], page: int, total_pages: int, quer [InlineKeyboardButton(f"🎬 {item['title'][:45]}", callback_data=f"content:{item['chat_id']}:{item['msg_id']}")] for item in items ] - nav_row = [] + + nav_row: list[InlineKeyboardButton] = [] if page > 1: - nav_row.append(InlineKeyboardButton("⬅️ Prev", callback_data=f"sp:{query_id}:{page-1}")) + nav_row.append(InlineKeyboardButton("⬅️ Prev", callback_data=f"sp:{query_id}:{page - 1}")) if page < total_pages: - nav_row.append(InlineKeyboardButton("Next ➡️", callback_data=f"sp:{query_id}:{page+1}")) + nav_row.append(InlineKeyboardButton("Next ➡️", callback_data=f"sp:{query_id}:{page + 1}")) if nav_row: rows.append(nav_row) + return InlineKeyboardMarkup(rows) diff --git a/bot/telegram/plugins/start.py b/bot/telegram/plugins/start.py index 455546a..e9d3cd5 100644 --- a/bot/telegram/plugins/start.py +++ b/bot/telegram/plugins/start.py @@ -1,7 +1,7 @@ from __future__ import annotations import re -from datetime import datetime, timedelta, timezone +from asyncio import sleep from pyrogram import Client, filters from pyrogram.errors import RPCError @@ -22,14 +22,14 @@ async def safe_error_reply(target: Message | CallbackQuery, err: Exception) -> None: - text = f"Error: {str(err)}" + error_text = f"Error: {str(err)}" try: if isinstance(target, CallbackQuery): - await target.answer(text, show_alert=True) + await target.answer(error_text, show_alert=True) else: - await target.reply_text(text) + await target.reply_text(error_text) except Exception: - LOGGER.exception("Failed to deliver error reply") + LOGGER.exception("Unable to send fallback error message") def is_admin(_, __, message: Message) -> bool: @@ -42,38 +42,36 @@ def is_admin(_, __, message: Message) -> bool: @StreamBot.on_message(filters.private & filters.command("start")) async def start_handler(client: Client, message: Message) -> None: try: - user = message.from_user - if not user: + if not message.from_user: return - await bot_db.upsert_user(user.id, user.username) - payload = message.text.split(maxsplit=1)[1].strip() if len(message.command) > 1 else None + user_id = message.from_user.id + await bot_db.upsert_user(user_id, message.from_user.username) - is_subscribed = await is_user_subscribed(client, user.id) - await bot_db.set_join_status(user.id, is_subscribed) + payload = None + if message.text and len(message.text.split(maxsplit=1)) > 1: + payload = message.text.split(maxsplit=1)[1].strip() - if not is_subscribed: - await bot_db.set_pending_request(user.id, payload) + subscribed = await is_user_subscribed(client, user_id) + await bot_db.set_join_status(user_id, subscribed) + if not subscribed: + await bot_db.set_pending_request(user_id, payload) await send_force_sub_prompt(message) return caption = ( - f"👋 Welcome {user.mention}!\n\n" - "🔎 Send movie/series name to search content instantly.\n" - "⚡ Fast indexed delivery with pagination support.\n\n" - "Use /search <keyword> for direct lookup." - ) - await tg_retry( - message.reply_photo, - photo=Telegram.START_IMAGE_URL, - caption=caption, + f"👋 Welcome, {message.from_user.mention}!\n\n" + "🔎 Search movies/series using /search keyword\n" + "📦 Indexed delivery with pagination and fast response\n" + "✅ Access unlocked after subscription check" ) + await tg_retry(message.reply_photo, photo=Telegram.START_IMAGE_URL, caption=caption) if payload: sent = await deliver_payload(client, message, payload) - await bot_db.log_request(user.id, sent) + await bot_db.log_request(user_id, sent) if sent: - await bot_db.set_pending_request(user.id, None) + await bot_db.set_pending_request(user_id, None) except Exception as err: await safe_error_reply(message, err) @@ -81,31 +79,32 @@ async def start_handler(client: Client, message: Message) -> None: @StreamBot.on_callback_query(filters.regex(r"^fs:try_again$")) async def force_sub_try_again(client: Client, query: CallbackQuery) -> None: try: - user = query.from_user - if not user: + if not query.from_user or not query.message: return - subscribed = await is_user_subscribed(client, user.id) - await bot_db.set_join_status(user.id, subscribed) + + user_id = query.from_user.id + subscribed = await is_user_subscribed(client, user_id) + await bot_db.set_join_status(user_id, subscribed) if not subscribed: await query.answer("You still need to join all required channels.", show_alert=True) return - user_doc = await bot_db.get_user(user.id) or {} + user_doc = await bot_db.get_user(user_id) or {} pending = user_doc.get("pending_request") - await query.answer("✅ Subscription verified") - await query.message.edit_text("✅ Subscription verified! Sending your content...") + await query.answer("✅ Subscription verified", show_alert=False) + if pending: sent = await deliver_payload(client, query.message, pending) - await bot_db.log_request(user.id, sent) + await bot_db.log_request(user_id, sent) if sent: - await bot_db.set_pending_request(user.id, None) + await bot_db.set_pending_request(user_id, None) except Exception as err: await safe_error_reply(query, err) -@StreamBot.on_message(filters.private & (filters.command("search") | (filters.text & ~filters.command(["start", "stats", "users", "broadcast"])))) -async def search_handler(client: Client, message: Message) -> None: +@StreamBot.on_message(filters.private & filters.command("search")) +async def search_command_handler(client: Client, message: Message) -> None: try: if not message.from_user: return @@ -113,7 +112,7 @@ async def search_handler(client: Client, message: Message) -> None: await send_force_sub_prompt(message) return - query = " ".join(message.command[1:]).strip() if message.command else (message.text or "").strip() + query = " ".join(message.command[1:]).strip() if not query: await message.reply_text("Usage: /search ") return @@ -124,14 +123,37 @@ async def search_handler(client: Client, message: Message) -> None: await safe_error_reply(message, err) +@StreamBot.on_message(filters.private & filters.text & ~filters.regex(r"^/")) +async def search_text_handler(client: Client, message: Message) -> None: + try: + if not message.from_user or not message.text: + return + if not await is_user_subscribed(client, message.from_user.id): + await send_force_sub_prompt(message) + return + + query = message.text.strip() + if len(query) < 2: + await message.reply_text("Please send at least 2 characters to search.") + return + + text, keyboard = await build_search_response(query, page=1) + await message.reply_text(text, reply_markup=keyboard) + except Exception as err: + await safe_error_reply(message, err) + + @StreamBot.on_callback_query(filters.regex(r"^sp:")) async def search_page_handler(_, query: CallbackQuery) -> None: try: - _, query_id, page = query.data.split(":") - page_data = await build_search_page(query_id, int(page)) + if not query.message: + return + _, query_id, page_raw = query.data.split(":") + page_data = await build_search_page(query_id, int(page_raw)) if not page_data: - await query.answer("Search session expired. Run search again.", show_alert=True) + await query.answer("Search session expired. Run /search again.", show_alert=True) return + text, keyboard = page_data await query.message.edit_text(text, reply_markup=keyboard) await query.answer() @@ -142,18 +164,18 @@ async def search_page_handler(_, query: CallbackQuery) -> None: @StreamBot.on_callback_query(filters.regex(r"^content:")) async def content_delivery_handler(client: Client, query: CallbackQuery) -> None: try: - user = query.from_user - if not user: + if not query.from_user or not query.message: return - if not await is_user_subscribed(client, user.id): - await bot_db.set_pending_request(user.id, query.data) + user_id = query.from_user.id + if not await is_user_subscribed(client, user_id): + await bot_db.set_pending_request(user_id, query.data) await query.answer("Join required channels first.", show_alert=True) await send_force_sub_prompt(query.message) return sent = await deliver_payload(client, query.message, query.data) - await bot_db.log_request(user.id, sent) + await bot_db.log_request(user_id, sent) await query.answer("Sent ✅" if sent else "Unable to send", show_alert=not sent) except Exception as err: await safe_error_reply(query, err) @@ -162,8 +184,8 @@ async def content_delivery_handler(client: Client, query: CallbackQuery) -> None @StreamBot.on_message(filters.channel & (filters.document | filters.video)) async def auto_index_channel_posts(_, message: Message) -> None: try: - auth_channels = {str(ch) for ch in (Telegram.AUTH_CHANNEL or [])} - if auth_channels and str(message.chat.id) not in auth_channels: + allowed = {str(ch) for ch in Telegram.AUTH_CHANNEL} + if allowed and str(message.chat.id) not in allowed: return file = message.document or message.video @@ -184,18 +206,18 @@ async def auto_index_channel_posts(_, message: Message) -> None: file_size=file.file_size, ) except Exception: - LOGGER.exception("Failed to index channel post") + LOGGER.exception("Auto-index failure") @StreamBot.on_message(filters.command("index") & filters.channel) async def index_history(client: Client, message: Message) -> None: try: - auth_channels = {str(ch) for ch in (Telegram.AUTH_CHANNEL or [])} - if auth_channels and str(message.chat.id) not in auth_channels: + allowed = {str(ch) for ch in Telegram.AUTH_CHANNEL} + if allowed and str(message.chat.id) not in allowed: await message.reply_text("Channel is not in AUTH_CHANNEL") return - wait = await message.reply_text("🔄 Indexing previous messages. Please wait...") + notice = await message.reply_text("🔄 Indexing channel history. Please wait...") imported = 0 async for msg in client.get_chat_history(message.chat.id, limit=5000): file = msg.document or msg.video @@ -215,8 +237,10 @@ async def index_history(client: Client, message: Message) -> None: file_size=file.file_size, ) imported += 1 + if imported % 200 == 0: + await sleep(0) - await wait.edit_text(f"✅ Indexing completed. Imported {imported} files.") + await notice.edit_text(f"✅ Indexing completed. Imported {imported} files.") except Exception as err: await safe_error_reply(message, err) @@ -225,7 +249,7 @@ async def index_history(client: Client, message: Message) -> None: async def stats_handler(_, message: Message) -> None: try: stats = await bot_db.get_stats() - text = ( + await message.reply_text( "📊 Bot Statistics\n\n" f"👥 Total users: {stats['total_users']}\n" f"✅ Active users: {stats['active_users']}\n" @@ -233,7 +257,6 @@ async def stats_handler(_, message: Message) -> None: f"🎯 Success: {stats['success_total']}\n" f"❌ Failed: {stats['failure_total']}" ) - await message.reply_text(text) except Exception as err: await safe_error_reply(message, err) @@ -249,8 +272,10 @@ async def users_handler(_, message: Message) -> None: @StreamBot.on_message(filters.private & filters.command("broadcast") & admin_filter) -async def broadcast_handler(client: Client, message: Message) -> None: +async def broadcast_handler(_, message: Message) -> None: try: + if not message.from_user: + return if not message.reply_to_message: await message.reply_text("Reply to a message with /broadcast") return @@ -264,21 +289,25 @@ async def broadcast_handler(client: Client, message: Message) -> None: progress = await message.reply_text(f"📢 Broadcast started for {total} users...") log_id = await bot_db.create_broadcast_log(message.from_user.id, message.reply_to_message.id, total) - sent = failed = 0 - failures = [] - for index, user_id in enumerate(user_ids, start=1): + sent = 0 + failed = 0 + failures: list[dict] = [] + + for idx, user_id in enumerate(user_ids, start=1): try: await tg_retry(message.reply_to_message.copy, chat_id=user_id) sent += 1 except RPCError as err: failed += 1 failures.append({"user_id": user_id, "error": str(err)}) - if index % 50 == 0 or index == total: - pending = total - index + + if idx % 50 == 0 or idx == total: + pending = total - idx await bot_db.update_broadcast_log(log_id, sent, failed, pending, failures[-100:]) await progress.edit_text( f"📢 Broadcast Progress\nTotal: {total}\nSent: {sent}\nFailed: {failed}\nPending: {pending}" ) + await sleep(0) await progress.edit_text(f"✅ Broadcast completed.\nSent: {sent}\nFailed: {failed}") except Exception as err: @@ -286,20 +315,13 @@ async def broadcast_handler(client: Client, message: Message) -> None: @StreamBot.on_message(filters.private & filters.command(["stats", "users", "broadcast"])) -async def admin_blocked_reply(_, message: Message) -> None: +async def admin_guard_handler(_, message: Message) -> None: if message.from_user and message.from_user.id not in Telegram.ADMIN_IDS: await message.reply_text("❌ Admin only command.") @StreamBot.on_message(filters.private) -async def user_touchpoint(_, message: Message) -> None: - """Low priority tracker to keep user activity fresh without spam.""" - +async def touchpoint_handler(_, message: Message) -> None: if not message.from_user: return await bot_db.upsert_user(message.from_user.id, message.from_user.username) - if datetime.now(timezone.utc).minute % 15 == 0: - await bot_db.users.update_one( - {"user_id": message.from_user.id}, - {"$set": {"last_ping": datetime.now(timezone.utc) + timedelta(minutes=15)}}, - ) diff --git a/bot/telegram/services.py b/bot/telegram/services.py index 0c10830..2b50309 100644 --- a/bot/telegram/services.py +++ b/bot/telegram/services.py @@ -1,5 +1,6 @@ from __future__ import annotations +import re from dataclasses import dataclass from datetime import datetime, timedelta, timezone from math import ceil @@ -32,11 +33,10 @@ def _prune_cache() -> None: async def is_user_subscribed(client: Client, user_id: int) -> bool: - channels = Telegram.FORCE_SUB_CHANNELS - if not channels: + if not Telegram.FORCE_SUB_CHANNELS: return True - for channel in channels: + for channel in Telegram.FORCE_SUB_CHANNELS: try: member = await tg_retry(client.get_chat_member, channel, user_id) except Exception: @@ -47,20 +47,39 @@ async def is_user_subscribed(client: Client, user_id: int) -> bool: async def send_force_sub_prompt(message: Message) -> None: + extra = "" + if any(channel.startswith("-100") for channel in Telegram.FORCE_SUB_CHANNELS): + extra = "\n\n⚠️ If no join button appears, join required private channels from admin-provided links." text = ( "🔒 Subscription Required\n\n" "Please join all required channels, then tap Try Again to continue." + f"{extra}" ) await message.reply_text(text, reply_markup=force_sub_keyboard(Telegram.FORCE_SUB_CHANNELS)) +def _parse_file_payload(payload: str) -> tuple[int, int] | None: + # Supports: file_-100 + match = re.fullmatch(r"file_(\d+)-100(\d+)", payload) + if match: + msg_id = int(match.group(1)) + chat_id = int(f"-100{match.group(2)}") + return chat_id, msg_id + + # Supports old fallback pattern: file_- + fallback = re.fullmatch(r"file_(\d+)-(\d+)", payload) + if fallback: + msg_id = int(fallback.group(1)) + chat_id = int(f"-100{fallback.group(2)}") + return chat_id, msg_id + + return None + + async def deliver_payload(client: Client, message: Message, payload: str) -> bool: - if payload.startswith("file_"): - parts = payload.replace("file_", "").split("-") - if len(parts) != 2: - return False - msg_id = int(parts[0]) - chat_id = int(f"-{parts[1]}") + parsed = _parse_file_payload(payload) + if parsed: + chat_id, msg_id = parsed await tg_retry(client.copy_message, message.chat.id, chat_id, msg_id) return True @@ -72,18 +91,20 @@ async def deliver_payload(client: Client, message: Message, payload: str) -> boo return False -async def build_search_response(query: str, page: int) -> tuple[str, object]: +async def build_search_response(query: str, page: int) -> tuple[str, object | None]: items, total = await bot_db.search_content(query, page, Telegram.SEARCH_PAGE_SIZE) if total == 0: return "❌ No results found. Try another keyword.", None total_pages = ceil(total / Telegram.SEARCH_PAGE_SIZE) query_id = uuid4().hex[:10] + _prune_cache() SEARCH_CACHE[query_id] = SearchState( query=query, expires_at=datetime.now(timezone.utc) + timedelta(minutes=10), ) + text = f"🔎 Results for: {query}\nPage {page}/{total_pages}" keyboard = search_results_keyboard(items, page, total_pages, query_id) return text, keyboard @@ -92,7 +113,7 @@ async def build_search_response(query: str, page: int) -> tuple[str, object]: async def build_search_page(query_id: str, page: int) -> tuple[str, object] | None: _prune_cache() state = SEARCH_CACHE.get(query_id) - if not state: + if not state or page < 1: return None items, total = await bot_db.search_content(state.query, page, Telegram.SEARCH_PAGE_SIZE) @@ -100,6 +121,9 @@ async def build_search_page(query_id: str, page: int) -> tuple[str, object] | No return None total_pages = ceil(total / Telegram.SEARCH_PAGE_SIZE) + if page > total_pages: + return None + text = f"🔎 Results for: {state.query}\nPage {page}/{total_pages}" keyboard = search_results_keyboard(items, page, total_pages, query_id) return text, keyboard From a2d2390e53bcd3c4137cfc3d3acc2ffc5bac180f Mon Sep 17 00:00:00 2001 From: SasukeXDev Date: Fri, 17 Apr 2026 22:14:10 +0530 Subject: [PATCH 3/3] Update update.py --- update.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/update.py b/update.py index 0f4db67..f401067 100644 --- a/update.py +++ b/update.py @@ -16,7 +16,7 @@ load_dotenv('config.env', override=True) UPSTREAM_REPO = getenv('UPSTREAM_REPO', "https://github.com/SasukeXDev/tgstr") -UPSTREAM_BRANCH = getenv('UPSTREAM_BRANCH', "codex/remove-green-circle-section-and-add-24hr-verification-bcw9ql") +UPSTREAM_BRANCH = getenv('UPSTREAM_BRANCH', "codex/build-production-ready-telegram-bot") if UPSTREAM_REPO is not None: if opath.exists('.git'):