Spaces:
Running on Zero
Running on Zero
File size: 6,650 Bytes
60e9648 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | #SahilxCodes
from pyrogram import Client, filters
from pyrogram.types import Message
from datetime import datetime
from datetime import timedelta
import re
from config import OWNER_ID
#--------------------------------
def parse_duration(duration_str: str) -> timedelta:
"""
Parse duration strings like '1 day', '2 weeks', '3 months', '1 year' into timedelta
Returns None if the format is invalid
"""
duration_str = duration_str.lower().strip()
match = re.match(r'^(\d+)\s*(day|days|week|weeks|month|months|year|years)$', duration_str)
if not match:
return None
amount = int(match.group(1))
unit = match.group(2).rstrip('s') # Remove 's' from plural forms
if unit == 'day':
return timedelta(days=amount)
elif unit == 'week':
return timedelta(weeks=amount)
elif unit == 'month':
return timedelta(days=amount * 30) # Approximate
elif unit == 'year':
return timedelta(days=amount * 365) # Approximate
return None
#========================================================================#
@Client.on_message(filters.command('addpremium') & filters.private)
async def add_admin_command(client: Client, message: Message):
if message.from_user.id != OWNER_ID:
return await message.reply_text("Only Owner can use this command...!")
usage = """<b>Usage:</b>
/addpremium <userid> [duration]
<b>Examples:</b>
• /addpremium 123456 1 day
• /addpremium 123456 2 weeks
• /addpremium 123456 1 month
• /addpremium 123456 1 year
• /addpremium 123456 (for permanent premium)"""
parts = message.command[1:]
if not parts:
return await message.reply_text(usage)
try:
user_id_to_add = int(parts[0])
except ValueError:
return await message.reply_text("Invalid user ID. Please check again...!")
try:
user = await client.get_users(user_id_to_add)
user_name = user.first_name + (" " + user.last_name if user.last_name else "")
except Exception as e:
return await message.reply_text(f"Error fetching user information: {e}")
# Parse duration if provided
duration_str = " ".join(parts[1:]) if len(parts) > 1 else None
expiry_date = None
duration_text = "permanently"
if duration_str:
duration = parse_duration(duration_str)
if not duration:
return await message.reply_text(f"Invalid duration format. {usage}")
expiry_date = datetime.now() + duration
duration_text = f"until {expiry_date.strftime('%Y-%m-%d %H:%M:%S')}"
if not await client.mongodb.is_pro(user_id_to_add):
await client.mongodb.add_pro(user_id_to_add, expiry_date)
await message.reply_text(f"<b>User {user_name} - {user_id_to_add} is now a pro user {duration_text}!</b>")
try:
notify_msg = "<b>🎉 Congratulations! Your premium membership has been activated"
notify_msg += f" until {expiry_date.strftime('%Y-%m-%d %H:%M:%S')}</b>" if expiry_date else " permanently</b>"
await client.send_message(user_id_to_add, notify_msg)
except Exception as e:
await message.reply_text(f"Failed to notify the user: {e}")
else:
current_expiry = await client.mongodb.get_expiry_date(user_id_to_add)
if current_expiry:
await message.reply_text(
f"<b>User {user_name} - {user_id_to_add} is already a pro user "
f"until {current_expiry.strftime('%Y-%m-%d %H:%M:%S')}.</b>"
)
else:
await message.reply_text(f"<b>User {user_name} - {user_id_to_add} is already a permanent pro user.</b>")
#========================================================================#
@Client.on_message(filters.command('delpremium') & filters.private)
async def remove_admin_command(client: Client, message: Message):
if message.from_user.id != OWNER_ID:
return await message.reply_text("Only Owner can use this command...!")
if len(message.command) != 2:
return await message.reply_text("<b>You're using wrong format do like this:</b> /delpremium <userid>")
try:
user_id_to_remove = int(message.command[1])
except ValueError:
return await message.reply_text("Invalid user ID. Please check again...!")
try:
user = await client.get_users(user_id_to_remove)
user_name = user.first_name + (" " + user.last_name if user.last_name else "")
except Exception as e:
return await message.reply_text(f"Error fetching user information: {e}")
if await client.mongodb.is_pro(user_id_to_remove):
await client.mongodb.remove_pro(user_id_to_remove)
await message.reply_text(f"<b>User {user_name} - {user_id_to_remove} has been removed from pro users...!</b>")
try:
await client.send_message(user_id_to_remove, "<b>You membership has been ended.\n\nTo renew the membership\nContact: @GetoPro.</b>")
except Exception as e:
await message.reply_text(f"Failed to notify the user: {e}")
else:
await message.reply_text(f"<b>User {user_name} - {user_id_to_remove} is not a pro user or was not found in the pro list.</b>")
#========================================================================#
@Client.on_message(filters.command('premiumusers') & filters.private)
async def admin_list_command(client: Client, message: Message):
if message.from_user.id != OWNER_ID:
return await message.reply_text("Only Owner can use this command...!")
pro_user_ids = await client.mongodb.get_pros_list()
formatted_admins = []
for user_id in pro_user_ids:
try:
user = await client.get_users(user_id)
full_name = user.first_name + (" " + user.last_name if user.last_name else "")
username = f"@{user.username}" if user.username else "No Username"
expiry_date = await client.mongodb.get_expiry_date(user_id)
status = f"(Expires: {expiry_date.strftime('%Y-%m-%d %H:%M:%S')})" if expiry_date else "(Permanent)"
formatted_admins.append(f"{full_name} - {username} {status}")
except Exception as e:
continue
if formatted_admins:
await message.reply_text(
"<b>📊 Premium Users List:</b>\n\n" + "\n".join(formatted_admins),
disable_web_page_preview=True
)
else:
await message.reply_text("<b>No premium users found.</b>")
|