File size: 4,724 Bytes
157e137 |
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 |
import logging
import math
from FileStream import __version__
from FileStream.bot import FileStream
from FileStream.server.exceptions import FIleNotFound
from FileStream.utils.bot_utils import gen_linkx, verify_user
from FileStream.config import Telegram
from FileStream.utils.database import Database
from FileStream.utils.translation import LANG, BUTTON
from pyrogram import filters, Client
from pyrogram.types import InlineKeyboardMarkup, InlineKeyboardButton, Message
from pyrogram.enums.parse_mode import ParseMode
db = Database(Telegram.DATABASE_URL, Telegram.SESSION_NAME)
@FileStream.on_message(filters.command('start') & filters.private)
async def start(bot: Client, message: Message):
if not await verify_user(bot, message):
return
usr_cmd = message.text.split("_")[-1]
if usr_cmd == "/start":
await message.reply_text(
text=LANG.START_TEXT.format(message.from_user.mention, FileStream.username),
parse_mode=ParseMode.HTML,
disable_web_page_preview=True,
reply_markup=BUTTON.START_BUTTONS
)
else:
if "stream_" in message.text:
try:
file_check = await db.get_file(usr_cmd)
file_id = str(file_check['_id'])
if file_id == usr_cmd:
reply_markup, stream_text = await gen_linkx(m=message, _id=file_id,
name=[FileStream.username, FileStream.fname])
await message.reply_text(
text=stream_text,
parse_mode=ParseMode.HTML,
disable_web_page_preview=True,
reply_markup=reply_markup,
quote=True
)
except FIleNotFound as e:
await message.reply_text("File Not Found")
except Exception as e:
await message.reply_text("Something Went Wrong")
logging.error(e)
elif "file_" in message.text:
try:
file_check = await db.get_file(usr_cmd)
db_id = str(file_check['_id'])
file_id = file_check['file_id']
file_name = file_check['file_name']
if db_id == usr_cmd:
await message.reply_cached_media(file_id=file_id, caption=f'**{file_name}**')
except FIleNotFound as e:
await message.reply_text("**File Not Found**")
except Exception as e:
await message.reply_text("Something Went Wrong")
logging.error(e)
else:
await message.reply_text(f"**Invalid Command**")
@FileStream.on_message(filters.private & filters.command(["about"]))
async def start(bot, message):
if not await verify_user(bot, message):
return
await message.reply_text(
text=LANG.ABOUT_TEXT.format(FileStream.fname, __version__),
disable_web_page_preview=True,
reply_markup=BUTTON.ABOUT_BUTTONS
)
@FileStream.on_message((filters.command('help')) & filters.private)
async def help_handler(bot, message):
if not await verify_user(bot, message):
return
await message.reply_text(
text=LANG.HELP_TEXT.format(Telegram.OWNER_ID),
parse_mode=ParseMode.HTML,
disable_web_page_preview=True,
reply_markup=BUTTON.HELP_BUTTONS
)
# ---------------------------------------------------------------------------------------------------
@FileStream.on_message(filters.command('files') & filters.private)
async def my_files(bot: Client, message: Message):
if not await verify_user(bot, message):
return
user_files, total_files = await db.find_files(message.from_user.id, [1, 10])
file_list = []
async for x in user_files:
file_list.append([InlineKeyboardButton(x["file_name"], callback_data=f"myfile_{x['_id']}_{1}")])
if total_files > 10:
file_list.append(
[
InlineKeyboardButton("◄", callback_data="N/A"),
InlineKeyboardButton(f"1/{math.ceil(total_files / 10)}", callback_data="N/A"),
InlineKeyboardButton("►", callback_data="userfiles_2")
],
)
if not file_list:
file_list.append(
[InlineKeyboardButton("ᴇᴍᴘᴛʏ", callback_data="N/A")],
)
file_list.append([InlineKeyboardButton("ᴄʟᴏsᴇ", callback_data="close")])
await message.reply_photo(photo=Telegram.IMAGE_FILEID,
caption="Total files: {}".format(total_files),
reply_markup=InlineKeyboardMarkup(file_list))
|