Spaces:

Arnavpro
/

Tg

Build error

Tg / Shin.py

Create Shin.py

1e4ed2e verified 11 months ago

11.8 kB

import logging
import re
import os
import threading
from datetime import timedelta
from flask import Flask
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup, ChatPermissions, ChatMember
from telegram.constants import ParseMode
from telegram.ext import (
    Application,
    CommandHandler,
    ChatJoinRequestHandler,
    MessageHandler,
    filters,
    ContextTypes,
)

# --- FLASK SERVER SETUP (For 24/7 Hosting) ---
webapp = Flask(__name__)

@webapp.route('/')
def home():
    return "Bot is running 24/7!"

def run_flask():
    # Hugging Face ke liye port 7860 zaroori hai
    webapp.run(host="0.0.0.0", port=7860)

# --- CONFIGURATION ---
BOT_TOKEN = "7974658489:AAFCdZtki5NVOn7ez9UukXcCSYtpvF2fnVg"

# Multiple admin IDs
ADMIN_IDS = [8186973947, 7857898495]

CHANNEL_LINK = "https://t.me/shinchanbannedmovies"
CHANNEL_USERNAME = "@shinchanbannedmovies"

AUTO_APPROVE = True
DEFAULT_MUTE_DURATION_HOURS = 1
# ---------------------

# Logging setup
logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
logger = logging.getLogger(__name__)

# Track warnings
user_warnings = {}  # {user_id: count}


# --- HELPER FUNCTIONS ---

def is_admin(user_id: int) -> bool:
    """Checks if the user ID is in the hardcoded ADMIN_IDS list."""
    return user_id in ADMIN_IDS

async def get_target_user_id(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int | None:
    """Finds the target user ID either from a reply or an argument."""
    message = update.effective_message
    
    if message.reply_to_message:
        target_user = message.reply_to_message.from_user
    elif context.args and context.args[0].isdigit():
        try:
            target_user = await context.bot.get_chat_member(message.chat_id, int(context.args[0]))
            return target_user.user.id
        except Exception:
            return int(context.args[0]) 
    else:
        await message.reply_text("Please reply to a user's message or provide their User ID.")
        return None
    
    return target_user.id

async def is_group_admin(chat_id: int, user_id: int, context: ContextTypes.DEFAULT_TYPE) -> bool:
    """Checks if a user is an admin or creator of the current chat."""
    try:
        member = await context.bot.get_chat_member(chat_id, user_id)
        return member.status in [ChatMember.ADMINISTRATOR, ChatMember.CREATOR]
    except Exception:
        return False

async def apply_punishment(chat_id: int, user: object, count: int, context: ContextTypes.DEFAULT_TYPE, message_to_delete: Update = None):
    """Handles the 3-strike punishment system."""
    user_mention = user.mention_html()
    
    if message_to_delete:
        try:
            await message_to_delete.delete()
        except Exception as e:
            logger.warning(f"Failed to delete message from {user.id}: {e}")

    if count == 1:
        await context.bot.send_message(
            chat_id=chat_id,
            text=f"⚠️ {user_mention}, this is your *first warning!* Stop sharing unauthorized content/links.",
            parse_mode=ParseMode.HTML,
        )
    elif count == 2:
        try:
            # Mute for 1 hour
            await context.bot.restrict_chat_member(
                chat_id,
                user.id,
                ChatPermissions(can_send_messages=False),
                until_date=timedelta(hours=DEFAULT_MUTE_DURATION_HOURS),
            )
            await context.bot.send_message(
                chat_id=chat_id,
                text=f"🔇 {user_mention} muted for {DEFAULT_MUTE_DURATION_HOURS} hour(s) for the second strike!",
                parse_mode=ParseMode.HTML,
            )
        except Exception as e:
            logger.error(f"Failed to mute {user.id}: {e}")
            await context.bot.send_message(chat_id, f"❌ Failed to mute {user_mention}.", parse_mode=ParseMode.HTML)
    else:
        try:
            # Ban the user
            await context.bot.ban_chat_member(chat_id, user.id)
            await context.bot.send_message(
                chat_id=chat_id,
                text=f"🚫 {user_mention} has been *banned* for repeated offenses (3 strikes).",
                parse_mode=ParseMode.HTML,
            )
            if user.id in user_warnings:
                del user_warnings[user.id]

        except Exception as e:
            logger.error(f"Failed to ban {user.id}: {e}")


# --- BOT COMMANDS ---

async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(
        "🤖 *Auto Approver + Anti-Link Bot Online!*\n\n"
        "Add me to your group and make me admin to start working!",
        parse_mode=ParseMode.MARKDOWN,
    )

async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(
        "🧩 *Bot Commands:*\n\n"
        "• /start - Check bot status\n"
        "• /help - Show this help\n"
        "• /status - Check bot permissions\n"
        "• /broadcast <msg> - Super Admin Only\n\n"
        "🚨 *Moderation:*\n"
        "• /warn [reply/ID]\n"
        "• /mute [reply/ID] [time_h]\n"
        "• /unmute [reply/ID]\n"
        "• /clearwarn [reply/ID]\n",
        parse_mode=ParseMode.MARKDOWN,
    )

async def status_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat = update.effective_chat
    if chat.type == 'private':
        await update.message.reply_text("This command must be used in a group!")
        return
    try:
        member = await chat.get_member(context.bot.id)
        status_msg = f"⚙️ *Bot Status in {chat.title}:*\n\n"
        if member.status in [ChatMember.ADMINISTRATOR, ChatMember.CREATOR]:
            status_msg += "✅ I am an Administrator.\n"
        else:
            status_msg += "❌ I am NOT an Administrator."
        await update.message.reply_text(status_msg, parse_mode=ParseMode.MARKDOWN)
    except Exception as e:
        await update.message.reply_text("❌ Error checking status.")

async def warn_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat = update.effective_chat
    issuer = update.effective_user
    if not (is_admin(issuer.id) or await is_group_admin(chat.id, issuer.id, context)):
        await update.message.reply_text("⛔ Admin rights required.")
        return
    target_id = await get_target_user_id(update, context)
    if target_id:
        try:
            target_user = await context.bot.get_chat_member(chat.id, target_id)
            user = target_user.user
            if await is_group_admin(chat.id, user.id, context):
                await update.message.reply_text("🚫 Cannot warn admins.")
                return
            user_warnings[user.id] = user_warnings.get(user.id, 0) + 1
            count = user_warnings[user.id]
            await apply_punishment(chat.id, user, count, context)
        except Exception as e:
            await update.message.reply_text("❌ Error.")

async def mute_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat = update.effective_chat
    issuer = update.effective_user
    if not (is_admin(issuer.id) or await is_group_admin(chat.id, issuer.id, context)):
        return
    target_id = await get_target_user_id(update, context)
    if target_id:
        duration = int(context.args[-1]) if context.args and context.args[-1].isdigit() else 1
        await context.bot.restrict_chat_member(chat.id, target_id, ChatPermissions(can_send_messages=False), until_date=timedelta(hours=duration))
        await update.message.reply_text(f"✅ Muted for {duration}h.")

async def unmute_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat = update.effective_chat
    issuer = update.effective_user
    if not (is_admin(issuer.id) or await is_group_admin(chat.id, issuer.id, context)):
        return
    target_id = await get_target_user_id(update, context)
    if target_id:
        await context.bot.restrict_chat_member(chat.id, target_id, ChatPermissions(can_send_messages=True, can_send_photos=True, can_send_videos=True, can_send_audios=True, can_send_documents=True, can_send_other_messages=True, can_add_web_page_previews=True))
        await update.message.reply_text("✅ Unmuted.")

async def clearwarn_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    issuer = update.effective_user
    if not (is_admin(issuer.id) or await is_group_admin(update.effective_chat.id, issuer.id, context)):
        return
    target_id = await get_target_user_id(update, context)
    if target_id and target_id in user_warnings:
        del user_warnings[target_id]
        await update.message.reply_text("✅ Warnings reset.")

# --- LISTENERS ---

async def approve_join_request(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if AUTO_APPROVE:
        try:
            await context.bot.approve_chat_join_request(update.chat_join_request.chat.id, update.chat_join_request.from_user.id)
        except Exception as e:
            logger.error(f"Approve error: {e}")

async def welcome_new_member(update: Update, context: ContextTypes.DEFAULT_TYPE):
    for member in update.message.new_chat_members:
        if not member.is_bot:
            buttons = InlineKeyboardMarkup([[InlineKeyboardButton("📢 Join Channel", url=CHANNEL_LINK)]])
            await update.message.reply_text(f"👋 Welcome {member.mention_html()}!", parse_mode=ParseMode.HTML, reply_markup=buttons)

async def detect_links(update: Update, context: ContextTypes.DEFAULT_TYPE):
    message = update.message
    if not message or not message.from_user or message.from_user.is_bot:
        return
    if await is_group_admin(update.effective_chat.id, message.from_user.id, context):
        return
    text = message.text or message.caption or ""
    if re.search(r"(https?://|t\.me/|telegram\.me/|bit\.ly|linktr\.ee|bio\.link)", text, re.IGNORECASE):
        user_warnings[message.from_user.id] = user_warnings.get(message.from_user.id, 0) + 1
        await apply_punishment(update.effective_chat.id, message.from_user, user_warnings[message.from_user.id], context, message_to_delete=message)

async def broadcast(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if not is_admin(update.effective_user.id):
        return
    if not context.args:
        return
    text = " ".join(context.args)
    # Simple broadcast to current chat (Expand this for database use)
    await update.message.reply_text(f"📢 BROADCAST: {text}")

# --- MAIN ---

def main():
    # Start Flask Thread
    threading.Thread(target=run_flask, daemon=True).start()
    
    app = Application.builder().token(BOT_TOKEN).build()

    # Handlers
    app.add_handler(CommandHandler("start", start_command))
    app.add_handler(CommandHandler("help", help_command))
    app.add_handler(CommandHandler("status", status_command, filters=filters.ChatType.GROUPS))
    app.add_handler(CommandHandler("warn", warn_command, filters=filters.ChatType.GROUPS))
    app.add_handler(CommandHandler("mute", mute_command, filters=filters.ChatType.GROUPS))
    app.add_handler(CommandHandler("unmute", unmute_command, filters=filters.ChatType.GROUPS))
    app.add_handler(CommandHandler("clearwarn", clearwarn_command, filters=filters.ChatType.GROUPS))
    app.add_handler(CommandHandler("broadcast", broadcast))

    app.add_handler(ChatJoinRequestHandler(approve_join_request))
    app.add_handler(MessageHandler(filters.ChatType.GROUPS & filters.StatusUpdate.NEW_CHAT_MEMBERS, welcome_new_member))
    app.add_handler(MessageHandler(filters.ChatType.GROUPS & (filters.TEXT | filters.CAPTION), detect_links))

    logger.info("🚀 Bot is now running with Flask on port 7860...")
    app.run_polling(poll_interval=1.0) 

if __name__ == "__main__":
    main()

	import logging
	import re
	import os
	import threading
	from datetime import timedelta
	from flask import Flask
	from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup, ChatPermissions, ChatMember
	from telegram.constants import ParseMode
	from telegram.ext import (
	Application,
	CommandHandler,
	ChatJoinRequestHandler,
	MessageHandler,
	filters,
	ContextTypes,
	)

	# --- FLASK SERVER SETUP (For 24/7 Hosting) ---
	webapp = Flask(__name__)

	@webapp.route('/')
	def home():
	return "Bot is running 24/7!"

	def run_flask():
	# Hugging Face ke liye port 7860 zaroori hai
	webapp.run(host="0.0.0.0", port=7860)

	# --- CONFIGURATION ---
	BOT_TOKEN = "7974658489:AAFCdZtki5NVOn7ez9UukXcCSYtpvF2fnVg"

	# Multiple admin IDs
	ADMIN_IDS = [8186973947, 7857898495]

	CHANNEL_LINK = "https://t.me/shinchanbannedmovies"
	CHANNEL_USERNAME = "@shinchanbannedmovies"

	AUTO_APPROVE = True
	DEFAULT_MUTE_DURATION_HOURS = 1
	# ---------------------

	# Logging setup
	logging.basicConfig(
	format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
	)
	logger = logging.getLogger(__name__)

	# Track warnings
	user_warnings = {} # {user_id: count}


	# --- HELPER FUNCTIONS ---

	def is_admin(user_id: int) -> bool:
	"""Checks if the user ID is in the hardcoded ADMIN_IDS list."""
	return user_id in ADMIN_IDS

	async def get_target_user_id(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int \| None:
	"""Finds the target user ID either from a reply or an argument."""
	message = update.effective_message

	if message.reply_to_message:
	target_user = message.reply_to_message.from_user
	elif context.args and context.args[0].isdigit():
	try:
	target_user = await context.bot.get_chat_member(message.chat_id, int(context.args[0]))
	return target_user.user.id
	except Exception:
	return int(context.args[0])
	else:
	await message.reply_text("Please reply to a user's message or provide their User ID.")
	return None

	return target_user.id

	async def is_group_admin(chat_id: int, user_id: int, context: ContextTypes.DEFAULT_TYPE) -> bool:
	"""Checks if a user is an admin or creator of the current chat."""
	try:
	member = await context.bot.get_chat_member(chat_id, user_id)
	return member.status in [ChatMember.ADMINISTRATOR, ChatMember.CREATOR]
	except Exception:
	return False

	async def apply_punishment(chat_id: int, user: object, count: int, context: ContextTypes.DEFAULT_TYPE, message_to_delete: Update = None):
	"""Handles the 3-strike punishment system."""
	user_mention = user.mention_html()

	if message_to_delete:
	try:
	await message_to_delete.delete()
	except Exception as e:
	logger.warning(f"Failed to delete message from {user.id}: {e}")

	if count == 1:
	await context.bot.send_message(
	chat_id=chat_id,
	text=f"⚠️ {user_mention}, this is your first warning! Stop sharing unauthorized content/links.",
	parse_mode=ParseMode.HTML,
	)
	elif count == 2:
	try:
	# Mute for 1 hour
	await context.bot.restrict_chat_member(
	chat_id,
	user.id,
	ChatPermissions(can_send_messages=False),
	until_date=timedelta(hours=DEFAULT_MUTE_DURATION_HOURS),
	)
	await context.bot.send_message(
	chat_id=chat_id,
	text=f"🔇 {user_mention} muted for {DEFAULT_MUTE_DURATION_HOURS} hour(s) for the second strike!",
	parse_mode=ParseMode.HTML,
	)
	except Exception as e:
	logger.error(f"Failed to mute {user.id}: {e}")
	await context.bot.send_message(chat_id, f"❌ Failed to mute {user_mention}.", parse_mode=ParseMode.HTML)
	else:
	try:
	# Ban the user
	await context.bot.ban_chat_member(chat_id, user.id)
	await context.bot.send_message(
	chat_id=chat_id,
	text=f"🚫 {user_mention} has been banned for repeated offenses (3 strikes).",
	parse_mode=ParseMode.HTML,
	)
	if user.id in user_warnings:
	del user_warnings[user.id]

	except Exception as e:
	logger.error(f"Failed to ban {user.id}: {e}")


	# --- BOT COMMANDS ---

	async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
	await update.message.reply_text(
	"🤖 Auto Approver + Anti-Link Bot Online!\n\n"
	"Add me to your group and make me admin to start working!",
	parse_mode=ParseMode.MARKDOWN,
	)

	async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
	await update.message.reply_text(
	"🧩 Bot Commands:\n\n"
	"• /start - Check bot status\n"
	"• /help - Show this help\n"
	"• /status - Check bot permissions\n"
	"• /broadcast <msg> - Super Admin Only\n\n"
	"🚨 Moderation:\n"
	"• /warn [reply/ID]\n"
	"• /mute [reply/ID] [time_h]\n"
	"• /unmute [reply/ID]\n"
	"• /clearwarn [reply/ID]\n",
	parse_mode=ParseMode.MARKDOWN,
	)

	async def status_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
	chat = update.effective_chat
	if chat.type == 'private':
	await update.message.reply_text("This command must be used in a group!")
	return
	try:
	member = await chat.get_member(context.bot.id)
	status_msg = f"⚙️ Bot Status in {chat.title}:\n\n"
	if member.status in [ChatMember.ADMINISTRATOR, ChatMember.CREATOR]:
	status_msg += "✅ I am an Administrator.\n"
	else:
	status_msg += "❌ I am NOT an Administrator."
	await update.message.reply_text(status_msg, parse_mode=ParseMode.MARKDOWN)
	except Exception as e:
	await update.message.reply_text("❌ Error checking status.")

	async def warn_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
	chat = update.effective_chat
	issuer = update.effective_user
	if not (is_admin(issuer.id) or await is_group_admin(chat.id, issuer.id, context)):
	await update.message.reply_text("⛔ Admin rights required.")
	return
	target_id = await get_target_user_id(update, context)
	if target_id:
	try:
	target_user = await context.bot.get_chat_member(chat.id, target_id)
	user = target_user.user
	if await is_group_admin(chat.id, user.id, context):
	await update.message.reply_text("🚫 Cannot warn admins.")
	return
	user_warnings[user.id] = user_warnings.get(user.id, 0) + 1
	count = user_warnings[user.id]
	await apply_punishment(chat.id, user, count, context)
	except Exception as e:
	await update.message.reply_text("❌ Error.")

	async def mute_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
	chat = update.effective_chat
	issuer = update.effective_user
	if not (is_admin(issuer.id) or await is_group_admin(chat.id, issuer.id, context)):
	return
	target_id = await get_target_user_id(update, context)
	if target_id:
	duration = int(context.args[-1]) if context.args and context.args[-1].isdigit() else 1
	await context.bot.restrict_chat_member(chat.id, target_id, ChatPermissions(can_send_messages=False), until_date=timedelta(hours=duration))
	await update.message.reply_text(f"✅ Muted for {duration}h.")

	async def unmute_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
	chat = update.effective_chat
	issuer = update.effective_user
	if not (is_admin(issuer.id) or await is_group_admin(chat.id, issuer.id, context)):
	return
	target_id = await get_target_user_id(update, context)
	if target_id:
	await context.bot.restrict_chat_member(chat.id, target_id, ChatPermissions(can_send_messages=True, can_send_photos=True, can_send_videos=True, can_send_audios=True, can_send_documents=True, can_send_other_messages=True, can_add_web_page_previews=True))
	await update.message.reply_text("✅ Unmuted.")

	async def clearwarn_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
	issuer = update.effective_user
	if not (is_admin(issuer.id) or await is_group_admin(update.effective_chat.id, issuer.id, context)):
	return
	target_id = await get_target_user_id(update, context)
	if target_id and target_id in user_warnings:
	del user_warnings[target_id]
	await update.message.reply_text("✅ Warnings reset.")

	# --- LISTENERS ---

	async def approve_join_request(update: Update, context: ContextTypes.DEFAULT_TYPE):
	if AUTO_APPROVE:
	try:
	await context.bot.approve_chat_join_request(update.chat_join_request.chat.id, update.chat_join_request.from_user.id)
	except Exception as e:
	logger.error(f"Approve error: {e}")

	async def welcome_new_member(update: Update, context: ContextTypes.DEFAULT_TYPE):
	for member in update.message.new_chat_members:
	if not member.is_bot:
	buttons = InlineKeyboardMarkup([[InlineKeyboardButton("📢 Join Channel", url=CHANNEL_LINK)]])
	await update.message.reply_text(f"👋 Welcome {member.mention_html()}!", parse_mode=ParseMode.HTML, reply_markup=buttons)

	async def detect_links(update: Update, context: ContextTypes.DEFAULT_TYPE):
	message = update.message
	if not message or not message.from_user or message.from_user.is_bot:
	return
	if await is_group_admin(update.effective_chat.id, message.from_user.id, context):
	return
	text = message.text or message.caption or ""
	if re.search(r"(https?://\|t\.me/\|telegram\.me/\|bit\.ly\|linktr\.ee\|bio\.link)", text, re.IGNORECASE):
	user_warnings[message.from_user.id] = user_warnings.get(message.from_user.id, 0) + 1
	await apply_punishment(update.effective_chat.id, message.from_user, user_warnings[message.from_user.id], context, message_to_delete=message)

	async def broadcast(update: Update, context: ContextTypes.DEFAULT_TYPE):
	if not is_admin(update.effective_user.id):
	return
	if not context.args:
	return
	text = " ".join(context.args)
	# Simple broadcast to current chat (Expand this for database use)
	await update.message.reply_text(f"📢 BROADCAST: {text}")

	# --- MAIN ---

	def main():
	# Start Flask Thread
	threading.Thread(target=run_flask, daemon=True).start()

	app = Application.builder().token(BOT_TOKEN).build()

	# Handlers
	app.add_handler(CommandHandler("start", start_command))
	app.add_handler(CommandHandler("help", help_command))
	app.add_handler(CommandHandler("status", status_command, filters=filters.ChatType.GROUPS))
	app.add_handler(CommandHandler("warn", warn_command, filters=filters.ChatType.GROUPS))
	app.add_handler(CommandHandler("mute", mute_command, filters=filters.ChatType.GROUPS))
	app.add_handler(CommandHandler("unmute", unmute_command, filters=filters.ChatType.GROUPS))
	app.add_handler(CommandHandler("clearwarn", clearwarn_command, filters=filters.ChatType.GROUPS))
	app.add_handler(CommandHandler("broadcast", broadcast))

	app.add_handler(ChatJoinRequestHandler(approve_join_request))
	app.add_handler(MessageHandler(filters.ChatType.GROUPS & filters.StatusUpdate.NEW_CHAT_MEMBERS, welcome_new_member))
	app.add_handler(MessageHandler(filters.ChatType.GROUPS & (filters.TEXT \| filters.CAPTION), detect_links))

	logger.info("🚀 Bot is now running with Flask on port 7860...")
	app.run_polling(poll_interval=1.0)

	if __name__ == "__main__":
	main()