Spaces:
Paused
Paused
File size: 3,273 Bytes
2875d7e 67b5fbf f0aea90 2875d7e 98e44f0 2875d7e 98e44f0 2875d7e 98e44f0 2875d7e 67b5fbf 2875d7e f0aea90 2875d7e f0aea90 2875d7e 98e44f0 2875d7e |
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 |
from pyrogram import filters
from pyrogram.types import Message, ChatPermissions
from pyrogram.enums import ChatMembersFilter, ChatMemberStatus
from pyrogram.errors import PeerIdInvalid
from DragMusic import app
from DragMusic.utils.decorators.admins import AdminActual
# Helper function to extract user info from a message
async def extract_user(message: Message, client):
if message.reply_to_message:
return message.reply_to_message.from_user
if len(message.command) > 1:
identifier = message.command[1]
try:
return await client.get_users(int(identifier))
except (ValueError, PeerIdInvalid):
try:
return await client.get_users(identifier)
except PeerIdInvalid:
await message.reply_text("Invalid user ID or username.")
return None
await message.reply_text("Reply to a user or provide a username/user ID.")
return None
@app.on_message(filters.command("promote") & filters.group)
@AdminActual
async def promote_user(client, message: Message, _):
target_user = await extract_user(message, client)
if not target_user:
return
try:
await client.promote_chat_member(message.chat.id, target_user.id)
await message.reply_text(f"Successfully promoted {target_user.mention}.")
except Exception as e:
await message.reply_text(f"Failed to promote user. Error: {e}")
@app.on_message(filters.command("demote") & filters.group)
@AdminActual
async def demote_user(client, message: Message, _):
target_user = await extract_user(message, client)
if not target_user:
return
try:
await client.promote_chat_member(
message.chat.id,
target_user.id,
privileges=ChatPermissions(can_manage_chat=False)
)
await message.reply_text(f"Successfully demoted {target_user.mention}.")
except Exception as e:
await message.reply_text(f"Failed to demote user. Error: {e}")
@app.on_message(filters.command("adminlist") & filters.group)
async def admin_list(client, message: Message):
admin_list_text = "Admins in this chat:\n\n"
admins = []
async for member in client.get_chat_members(message.chat.id, filter=ChatMembersFilter.ADMINISTRATORS):
admins.append(member)
# Sort to put creator at the top
admins.sort(key=lambda x: x.status == ChatMemberStatus.OWNER, reverse=True)
for admin in admins:
user = admin.user
status = "Owner" if admin.status == ChatMemberStatus.OWNER else "Admin"
if user.username:
admin_list_text += f"• @{user.username} ({status})\n"
else:
admin_list_text += f"• {user.mention} ({status})\n"
if len(admins) > 0:
await message.reply_text(admin_list_text)
else:
await message.reply_text("No admins found in this chat.")
@app.on_message(filters.command("admincache") & filters.group)
@AdminActual
async def admin_cache(client, message: Message, _):
from DragMusic.utils.decorators.admins import adminlist
if message.chat.id in adminlist:
del adminlist[message.chat.id]
await message.reply_text("Admin cache has been refreshed.") |