main.py

#!/usr/bin/env python3
"""
req_bot_telethon_mongo.py

Telethon & Mongo-backed Telegram request bot with persistent throttling/queueing.
- All background tasks (DMs, approvals, broadcasts) are saved to MongoDB.
- Resumes tasks automatically if the bot restarts.
- Compatible with all previously generated invite links.

Environment variables (recommended):
- BOT_TOKEN
- API_ID (required for Telethon)
- API_HASH (required for Telethon)
- MONGO_URI
- OWNER_ID (numeric)
"""

import os
import logging
import asyncio
from datetime import datetime
from typing import Optional, Any, List

from pymongo import MongoClient
from telethon import TelegramClient, events, functions, types, errors

# ---------------- CONFIG ----------------
API_ID = int(os.environ.get("API_ID", "22138159"))
API_HASH = os.environ.get("API_HASH", "3fe4592e4cad72f366b6c564505f2d57")
BOT_TOKEN = os.environ.get("BOT_TOKEN")

MONGO_URI = os.environ.get("MONGO_URI", "mongodb+srv://userbot:userbot@cluster0.iweqz.mongodb.net/test?retryWrites=true&w=majority")
DB_NAME = os.environ.get("DB_NAME", "reqbot_dbbb")

PENDING_COLLECTION = "pending_requests"
SETTINGS_COLLECTION = "settings"
USERS_COLLECTION = "users"
TASKS_COLLECTION = "tasks" # Added for persistent queues

OWNER_ID = int(os.environ.get("OWNER_ID", "7161228754"))

# Rate limits
DM_RATE_PER_MIN = int(os.environ.get("DM_RATE_PER_MIN", "300"))
APPROVE_RATE_PER_MIN = int(os.environ.get("APPROVE_RATE_PER_MIN", "200"))
BROADCAST_RATE_PER_MIN = int(os.environ.get("BROADCAST_RATE_PER_MIN", "200"))

logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)

# ---------------- MongoDB setup ----------------
db_client = MongoClient(MONGO_URI)
db = db_client[DB_NAME]
pending_col = db[PENDING_COLLECTION]
settings_col = db[SETTINGS_COLLECTION]
users_col = db[USERS_COLLECTION]
tasks_col = db[TASKS_COLLECTION]

# Best-effort indexes
try:
    pending_col.create_index([("chat_id", 1), ("user_id", 1)], unique=True)
    settings_col.create_index("key", unique=True)
    users_col.create_index("user_id", unique=True)
    tasks_col.create_index([("type", 1), ("status", 1), ("added_at", 1)])
except Exception:
    pass

# Reset any interrupted tasks from previous runs
tasks_col.update_many({"status": "processing"}, {"$set": {"status": "pending"}})

# ---------------- Helpers & persistence ----------------
def register_user(user_id: int, username: Optional[str], full_name: Optional[str]) -> None:
    try:
        users_col.update_one(
            {"user_id": user_id},
            {"$set": {"username": username, "full_name": full_name, "seen_at": datetime.utcnow()}},
            upsert=True,
        )
    except Exception:
        log.exception("Failed to register user %s", user_id)

def remove_blocked_user(user_id: int) -> None:
    """Removes a user from all DB collections if they have blocked the bot."""
    try:
        users_col.delete_one({"user_id": user_id})
        pending_col.delete_many({"user_id": user_id})
        log.info(f"Successfully deleted blocked user {user_id} from database.")
    except Exception:
        log.exception(f"Failed to delete blocked user {user_id} from database.")

def get_setting(key: str, default: Any = None) -> Any:
    doc = settings_col.find_one({"key": key})
    return doc["value"] if doc else default

def set_setting(key: str, value: Any) -> None:
    settings_col.find_one_and_update(
        {"key": key},
        {"$set": {"value": value, "updated_at": datetime.utcnow()}},
        upsert=True,
    )

def add_pending_request(chat_id: int, user: dict) -> None:
    try:
        pending_col.update_one(
            {"chat_id": chat_id, "user_id": user["user_id"]},
            {"$setOnInsert": {**user, "chat_id": chat_id, "added_at": datetime.utcnow()}},
            upsert=True,
        )
    except Exception:
        log.exception("Failed to add pending request to DB")

def list_pending_for_chat(chat_id: int) -> List[dict]:
    return list(pending_col.find({"chat_id": chat_id}).sort("added_at", 1))

def remove_pending(chat_id: int, user_id: int) -> None:
    pending_col.delete_one({"chat_id": chat_id, "user_id": user_id})

def enqueue_task(task_type: str, payload: dict) -> None:
    """Saves a task to MongoDB so it persists across restarts."""
    tasks_col.insert_one({
        "type": task_type,
        "status": "pending",
        "payload": payload,
        "added_at": datetime.utcnow()
    })

if get_setting("dm_default") is None:
    set_setting("dm_default", "Your request has been received. We'll review it soon.")

def parse_channel_id(arg: str) -> Optional[int]:
    try:
        return int(arg)
    except Exception:
        return None

# ---------------- Admin checks ----------------
async def verify_admin(client: TelegramClient, channel_id: int, user_id: int) -> tuple[bool, str]:
    try:
        perms = await client.get_permissions(channel_id, user_id)
        if perms.is_admin or perms.is_creator:
            return True, ""
        return False, "You/Bot are not an administrator in that channel."
    except Exception as e:
        return False, f"Could not verify permissions: {e}"

# ---------------- Persistent Queue Workers ----------------
def per_min_to_interval(rate_per_min: int) -> float:
    return 60.0 / rate_per_min if rate_per_min > 0 else 60.0

async def task_worker(client: TelegramClient, task_type: str, interval: float, handler_func):
    log.info(f"{task_type.capitalize()} worker started, interval={interval:.3f}s")
    while True:
        task = tasks_col.find_one_and_update(
            {"type": task_type, "status": "pending"},
            {"$set": {"status": "processing"}},
            sort=[("added_at", 1)]
        )
        
        if task:
            try:
                await handler_func(client, task["payload"])
                tasks_col.delete_one({"_id": task["_id"]})
            except errors.FloodWaitError as e:
                log.warning(f"Flood wait for {e.seconds}s. Pausing {task_type} queue.")
                await asyncio.sleep(e.seconds)
                # Revert task to pending
                tasks_col.update_one({"_id": task["_id"]}, {"$set": {"status": "pending"}})
            except Exception as e:
                log.exception(f"{task_type} failed")
                tasks_col.update_one({"_id": task["_id"]}, {"$set": {"status": "failed", "error": str(e)}})
                
            await asyncio.sleep(interval)
        else:
            await asyncio.sleep(1.0) # wait before polling db again

async def process_dm(client, payload):
    try:
        await client.send_message(payload["user_id"], payload["text"])
    except errors.UserIsBlockedError:
        log.warning(f"User {payload['user_id']} blocked the bot. Removing from DB.")
        remove_blocked_user(payload["user_id"])

async def process_approve(client, payload):
    try:
        await client(functions.messages.HideChatJoinRequestRequest(
            peer=payload["channel_id"],
            user_id=payload["user_id"],
            approved=True
        ))
        remove_pending(payload["channel_id"], payload["user_id"])
    except Exception as e:
        if "USER_ALREADY_PARTICIPANT" in str(e):
            remove_pending(payload["channel_id"], payload["user_id"])
        else:
            raise e

async def process_broadcast(client, payload):
    try:
        await client.send_message(payload["user_id"], payload["text"])
    except errors.UserIsBlockedError:
        log.warning(f"Broadcast failed: User {payload['user_id']} blocked the bot. Removing from DB.")
        remove_blocked_user(payload["user_id"])

# ---------------- Client Initialization ----------------
bot = TelegramClient('bot_session', API_ID, API_HASH)

# ---------------- Commands ----------------
@bot.on(events.NewMessage(pattern=r'^/start$'))
async def start_cmd(event):
    sender = await event.get_sender()
    register_user(sender.id, sender.username, f"{sender.first_name or ''} {sender.last_name or ''}".strip())
    await event.reply("Request-bot ready. Use /help to see commands.")

@bot.on(events.NewMessage(pattern=r'^/help$'))
async def help_cmd(event):
    txt = (
        "Available commands:\n\n"
        "/createreq <channel_id> - Create a request-type invite link (must be admin)\n"
        "/setmsg <channel_id> <message...>  - Save DM template. (Or reply to msg)\n"
        "/listpending <channel_id> - List pending join requests\n"
        "/acceptall <channel_id> - Approve all pending requests for that channel\n"
        "/broadcast <message...> - (OWNER only). Broadcast to all users.\n"
        "/help - Show this help message\n\n"
        "Notes:\n"
        "- All tasks are saved securely to DB and will survive bot restarts.\n"
    )
    await event.reply(txt)

@bot.on(events.NewMessage(pattern=r'^/createreq(?:\s+(.+))?$'))
async def createreq_cmd(event):
    if not event.pattern_match.group(1):
        return await event.reply("Usage: /createreq <channel_id>")
    
    channel_id = parse_channel_id(event.pattern_match.group(1))
    
    ok, reason = await verify_admin(bot, channel_id, event.sender_id)
    if not ok: return await event.reply(f"User check failed: {reason}")
    
    ok, reason = await verify_admin(bot, channel_id, (await bot.get_me()).id)
    if not ok: return await event.reply(f"Bot check failed: {reason}")

    try:
        result = await bot(functions.messages.ExportChatInviteRequest(
            peer=channel_id,
            title="RequestLink",
            request_needed=True
        ))
        await event.reply(f"Request-type invite link created:\n{result.link}")
    except Exception as e:
        await event.reply(f"Failed to create request invite link: {e}")

@bot.on(events.NewMessage(pattern=r'^/setmsg(?:\s+(.+))?$'))
async def setmsg_cmd(event):
    args = event.pattern_match.group(1)
    if not args and not event.is_reply:
        return await event.reply("Usage: /setmsg <channel_id> <message...> OR reply to a message with /setmsg <channel_id>")
    
    split_args = args.split(' ', 1) if args else []
    channel_id = parse_channel_id(split_args[0] if split_args else "")
    if channel_id is None:
        return await event.reply("Invalid channel id.")

    ok, reason = await verify_admin(bot, channel_id, event.sender_id)
    if not ok: return await event.reply(f"Permission check failed: {reason}")

    stored_text = ""
    if event.is_reply:
        reply_msg = await event.get_reply_message()
        stored_text = reply_msg.text or ""
    elif len(split_args) > 1:
        stored_text = split_args[1]

    set_setting(f"dm_{channel_id}", stored_text)
    await event.reply("DM message saved for channel.")

@bot.on(events.NewMessage(pattern=r'^/listpending(?:\s+(.+))?$'))
async def listpending_cmd(event):
    if not event.pattern_match.group(1):
        return await event.reply("Usage: /listpending <channel_id>")
    
    channel_id = parse_channel_id(event.pattern_match.group(1))
    ok, reason = await verify_admin(bot, channel_id, event.sender_id)
    if not ok: return await event.reply(f"Permission check failed: {reason}")

    pending = list_pending_for_chat(channel_id)
    if not pending:
        return await event.reply("No pending requests for this channel.")

    lines = [f"- {p.get('full_name') or p.get('username') or p.get('user_id')} ({p.get('user_id')})" for p in pending]
    await event.reply("Pending requests:\n" + "\n".join(lines))

@bot.on(events.NewMessage(pattern=r'^/acceptall(?:\s+(.+))?$'))
async def acceptall_cmd(event):
    if not event.pattern_match.group(1):
        return await event.reply("Usage: /acceptall <channel_id>")
    
    channel_id = parse_channel_id(event.pattern_match.group(1))
    ok, reason = await verify_admin(bot, channel_id, event.sender_id)
    if not ok: return await event.reply(f"Permission check failed: {reason}")

    pending = list_pending_for_chat(channel_id)
    if not pending:
        return await event.reply("No pending requests to accept.")

    for doc in pending:
        enqueue_task("approve", {"channel_id": channel_id, "user_id": doc["user_id"]})

    await event.reply(f"Saved {len(pending)} approvals to database. They will process continuously in the background.")

@bot.on(events.NewMessage(pattern=r'^/broadcast(?:\s+(.+))?$'))
async def broadcast_cmd(event):
    if event.sender_id != OWNER_ID:
        return await event.reply("Only OWNER can use this command.")

    args = event.pattern_match.group(1)
    btext = ""
    
    if event.is_reply:
        reply_msg = await event.get_reply_message()
        btext = reply_msg.text or ""
    elif args:
        btext = args
    else:
        return await event.reply("Usage: /broadcast <message...> OR reply to message with /broadcast")

    users = list(users_col.find({}, {"user_id": 1}))
    if not users:
        return await event.reply("No users to broadcast to.")

    for u in users:
        enqueue_task("broadcast", {"user_id": u["user_id"], "text": btext})

    await event.reply(f"Saved broadcast tasks for {len(users)} users to database. Processing will continue even if bot restarts.")

# ---------------- Join request handler ----------------
@bot.on(events.Raw(types.UpdateBotChatInviteRequester))
async def handle_join_request(event):
    chat_id = getattr(event.peer, 'channel_id', None) or getattr(event.peer, 'chat_id', None)
    if chat_id:
        chat_id = int(f"-100{chat_id}") # format for telethon channel ids
    
    try:
        user_entity = await bot.get_entity(event.user_id)
        full_name = f"{user_entity.first_name or ''} {user_entity.last_name or ''}".strip()
        username = f"@{user_entity.username}" if user_entity.username else None
    except Exception:
        full_name, username = None, None

    register_user(event.user_id, username, full_name)

    user_doc = {
        "user_id": event.user_id,
        "username": username,
        "full_name": full_name,
        "date": datetime.utcnow(),
    }
    
    add_pending_request(chat_id, user_doc)

    dm_text = get_setting(f"dm_{chat_id}", None) or get_setting("dm_default")

    # Save to MongoDB queue (survives crashes)
    enqueue_task("dm", {"user_id": event.user_id, "text": dm_text})


# ---------------- Catch-all ----------------
@bot.on(events.NewMessage())
async def catch_all_messages(event):
    if not event.is_private:
        return
    if event.text and not event.text.startswith('/'):
        sender = await event.get_sender()
        if sender:
            register_user(sender.id, sender.username, f"{sender.first_name or ''} {sender.last_name or ''}".strip())

# ---------------- Main ----------------
async def main():
    await bot.start(bot_token=BOT_TOKEN)
    log.info("Bot started successfully.")

    # Start workers as background tasks
    asyncio.create_task(task_worker(bot, "dm", per_min_to_interval(DM_RATE_PER_MIN), process_dm))
    asyncio.create_task(task_worker(bot, "approve", per_min_to_interval(APPROVE_RATE_PER_MIN), process_approve))
    asyncio.create_task(task_worker(bot, "broadcast", per_min_to_interval(BROADCAST_RATE_PER_MIN), process_broadcast))

    # Keep bot running
    await bot.run_until_disconnected()

if __name__ == "__main__":
    if API_ID == 1234567:
        log.error("Please set a valid API_ID and API_HASH environment variable.")
    else:
        asyncio.run(main())