Add files via upload

Author: Shiv-x

Dockerfile

@@ -0,0 +1,8 @@
+FROM python:3.9.6-slim-buster
+RUN apt-get update && apt-get upgrade -y
+RUN apt-get install git curl python3-pip ffmpeg -y
+RUN python3.9 -m pip install -U pip
+COPY . /app
+WORKDIR /app
+RUN python3.9 -m pip install -U -r requirements.txt
+CMD python3.9 -m VCPlayBot

Procfile

@@ -0,0 +1 @@
+worker: python3 -m VCPlayBot

VCPlayBot/__main__.py

@@ -0,0 +1,24 @@
+import requests
+from pyrogram import Client as Bot
+
+from VCPlayBot.config import API_HASH
+from VCPlayBot.config import API_ID
+from VCPlayBot.config import BG_IMAGE
+from VCPlayBot.config import BOT_TOKEN
+from VCPlayBot.services.callsmusic import run
+
+response = requests.get(BG_IMAGE)
+file = open("./etc/foreground.png", "wb")
+file.write(response.content)
+file.close()
+
+bot = Bot(
+    ":memory:",
+    API_ID,
+    API_HASH,
+    bot_token=BOT_TOKEN,
+    plugins=dict(root="VCPlayBot.modules"),
+)
+
+bot.start()
+run()

VCPlayBot/config.py

@@ -0,0 +1,29 @@
+import os
+from os import getenv
+
+from dotenv import load_dotenv
+
+if os.path.exists("local.env"):
+    load_dotenv("local.env")
+
+que = {}
+SESSION_NAME = getenv("SESSION_NAME", "session")
+BOT_TOKEN = getenv("BOT_TOKEN")
+BOT_NAME = getenv("BOT_NAME")
+UPDATES_CHANNEL = getenv("UPDATES_CHANNEL", "LaylaList")
+BG_IMAGE = getenv("BG_IMAGE", "https://telegra.ph/file/9b13ea3ce046a1a5c8098.png")
+admins = {}
+API_ID = int(getenv("API_ID"))
+API_HASH = getenv("API_HASH")
+BOT_USERNAME = getenv("BOT_USERNAME")
+ASSISTANT_NAME = getenv("ASSISTANT_NAME", "VCPlayAssistant")
+SUPPORT_GROUP = getenv("SUPPORT_GROUP", "AwesomeSupport")
+PROJECT_NAME = getenv("PROJECT_NAME", "VCPlayBot2.0")
+SOURCE_CODE = getenv("SOURCE_CODE", "github.com/QueenArzoo/VCPlayBot")
+DURATION_LIMIT = int(getenv("DURATION_LIMIT", "7"))
+ARQ_API_KEY = getenv("ARQ_API_KEY", None)
+PMPERMIT = getenv("PMPERMIT", None)
+LOG_GRP = getenv("LOG_GRP", None)
+COMMAND_PREFIXES = list(getenv("COMMAND_PREFIXES", "/ !").split())
+
+SUDO_USERS = list(map(int, getenv("SUDO_USERS", "797768146").split()))

VCPlayBot/function/__init__.py

@@ -0,0 +1,5 @@
+from VCPlayBot.function.admins import admins
+from VCPlayBot.function.admins import get
+from VCPlayBot.function.admins import set
+
+__all__ = ["set", "get", "admins"]

VCPlayBot/function/admins.py

@@ -0,0 +1,15 @@
+from typing import Dict
+from typing import List
+
+
+admins: Dict[int, List[int]] = {}
+
+
+def set(chat_id: int, admins_: List[int]):
+    admins[chat_id] = admins_
+
+
+def get(chat_id: int) -> List[int]:
+    if chat_id in admins:
+        return admins[chat_id]
+    return []

VCPlayBot/helpers/__init__.py

@@ -0,0 +1 @@
+

VCPlayBot/helpers/admins.py

@@ -0,0 +1,25 @@
+
+
+from typing import List
+
+from pyrogram.types import Chat
+
+from VCPlayBot.function.admins import get as gett
+from VCPlayBot.function.admins import set
+
+
+async def get_administrators(chat: Chat) -> List[int]:
+    get = gett(chat.id)
+
+    if get:
+        return get
+    else:
+        administrators = await chat.get_members(filter="administrators")
+        to_set = []
+
+        for administrator in administrators:
+            if administrator.can_manage_voice_chats:
+                to_set.append(administrator.user.id)
+
+        set(chat.id, to_set)
+        return await get_administrators(chat)

VCPlayBot/helpers/channelmusic.py

@@ -0,0 +1,7 @@
+from pyrogram.types import Chat
+
+
+def get_chat_id(chat: Chat):
+    if chat.title.startswith("Channel Music: ") and chat.title[16:].isnumeric():
+        return int(chat.title[15:])
+    return chat.id

VCPlayBot/helpers/decorators.py

@@ -0,0 +1,31 @@
+from typing import Callable
+
+from pyrogram import Client
+from pyrogram.types import Message
+
+from VCPlayBot.config import SUDO_USERS
+from VCPlayBot.helpers.admins import get_administrators
+
+
+def errors(func: Callable) -> Callable:
+    async def decorator(client: Client, message: Message):
+        try:
+            return await func(client, message)
+        except Exception as e:
+            await message.reply(f"{type(e).__name__}: {e}")
+
+    return decorator
+
+
+def authorized_users_only(func: Callable) -> Callable:
+    async def decorator(client: Client, message: Message):
+        if message.from_user.id in SUDO_USERS:
+            return await func(client, message)
+
+        administrators = await get_administrators(message.chat)
+
+        for administrator in administrators:
+            if administrator == message.from_user.id:
+                return await func(client, message)
+
+    return decorator

VCPlayBot/helpers/errors.py

@@ -0,0 +1,6 @@
+class DurationLimitError(Exception):
+    pass
+
+
+class FFmpegReturnCodeError(Exception):
+    pass

VCPlayBot/helpers/filters.py

@@ -0,0 +1,15 @@
+from typing import List
+from typing import Union
+
+from pyrogram import filters
+
+from VCPlayBot.config import COMMAND_PREFIXES
+
+other_filters = filters.group & ~filters.edited & ~filters.via_bot & ~filters.forwarded
+other_filters2 = (
+    filters.private & ~filters.edited & ~filters.via_bot & ~filters.forwarded
+)
+
+
+def command(commands: Union[str, List[str]]):
+    return filters.command(commands, COMMAND_PREFIXES)

VCPlayBot/helpers/gets.py

@@ -0,0 +1,36 @@
+from typing import Union
+
+from pyrogram.types import Audio
+from pyrogram.types import Message
+from pyrogram.types import Voice
+
+
+def get_url(message_1: Message) -> Union[str, None]:
+    messages = [message_1]
+
+    if message_1.reply_to_message:
+        messages.append(message_1.reply_to_message)
+
+    text = ""
+    offset = None
+    length = None
+
+    for message in messages:
+        if offset:
+            break
+
+        if message.entities:
+            for entity in message.entities:
+                if entity.type == "url":
+                    text = message.text or message.caption
+                    offset, length = entity.offset, entity.length
+                    break
+
+    if offset in (None,):
+        return None
+
+    return text[offset : offset + length]
+
+
+def get_file_name(audio: Union[Audio, Voice]):
+    return f'{audio.file_unique_id}.{audio.file_name.split(".")[-1] if not isinstance(audio, Voice) else "ogg"}'

VCPlayBot/modules/__init__.py

@@ -0,0 +1 @@
+

VCPlayBot/modules/admins.py

@@ -0,0 +1,113 @@
+
+
+from asyncio import QueueEmpty
+from pyrogram import Client 
+from pyrogram import filters
+from pyrogram.types import Message
+
+from VCPlayBot.config import que
+from VCPlayBot.function.admins import set
+from VCPlayBot.helpers.channelmusic import get_chat_id
+from VCPlayBot.helpers.decorators import authorized_users_only
+from VCPlayBot.helpers.decorators import errors
+from VCPlayBot.helpers.filters import command
+from VCPlayBot.helpers.filters import other_filters
+from VCPlayBot.services.callsmusic import callsmusic
+from VCPlayBot.services.queues import queues
+
+
+@Client.on_message(filters.command("adminreset"))
+async def update_admin(client, message: Message):
+    chat_id = get_chat_id(message.chat)
+    set(
+        chat_id,
+        [
+            member.user
+            for member in await message.chat.get_members(filter="administrators")
+        ],
+    )
+    await message.reply_text("❇️ Admin cache refreshed!")
+
+
+@Client.on_message(command("pause") & other_filters)
+@errors
+@authorized_users_only
+async def pause(_, message: Message):
+    chat_id = get_chat_id(message.chat)
+    if (chat_id not in callsmusic.active_chats) or (
+        callsmusic.active_chats[chat_id] == "paused"
+    ):
+        await message.reply_text("❗ Nothing is playing!")
+    else:
+        callsmusic.pause(chat_id)
+        await message.reply_text("▶️ Paused!")
+
+
+@Client.on_message(command("resume") & other_filters)
+@errors
+@authorized_users_only
+async def resume(_, message: Message):
+    chat_id = get_chat_id(message.chat)
+    if (chat_id not in callsmusic.active_chats) or (
+        callsmusic.active_chats[chat_id] == "playing"
+    ):
+        await message.reply_text("❗ Nothing is paused!")
+    else:
+        callsmusic.resume(chat_id)
+        await message.reply_text("⏸ Resumed!")
+
+
+@Client.on_message(command("end") & other_filters)
+@errors
+@authorized_users_only
+async def stop(_, message: Message):
+    chat_id = get_chat_id(message.chat)
+    if chat_id not in callsmusic.active_chats:
+        await message.reply_text("❗ Nothing is streaming!")
+    else:
+        try:
+            queues.clear(chat_id)
+        except QueueEmpty:
+            pass
+
+        await callsmusic.stop(chat_id)
+        await message.reply_text("❌ Stopped streaming!")
+
+
+@Client.on_message(command("skip") & other_filters)
+@errors
+@authorized_users_only
+async def skip(_, message: Message):
+    global que
+    chat_id = get_chat_id(message.chat)
+    if chat_id not in callsmusic.active_chats:
+        await message.reply_text("❗ Nothing is playing to skip!")
+    else:
+        queues.task_done(chat_id)
+        if queues.is_empty(chat_id):
+            await callsmusic.stop(chat_id)
+        else:
+            await callsmusic.set_stream(
+                chat_id, 
+                queues.get(chat_id)["file_path"]
+            )
+
+    qeue = que.get(chat_id)
+    if qeue:
+        skip = qeue.pop(0)
+    if not qeue:
+        return
+    await message.reply_text(f"- Skipped **{skip[0]}**\n- Now Playing **{qeue[0][0]}**")
+
+
+@Client.on_message(filters.command("admincache"))
+@errors
+async def admincache(client, message: Message):
+    set(
+        message.chat.id,
+        [
+            member.user
+            for member in await message.chat.get_members(filter="administrators")
+        ],
+    )
+    await message.reply_text("❇️ Admin cache refreshed!")