Spaces:
Running
Running
File size: 5,479 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 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 |
from __future__ import annotations
import logging
from datetime import datetime
from pyrogram import Client
from typing import Any, Optional
from pyrogram.enums import ParseMode, ChatType
from pyrogram.types import Message
from pyrogram.file_id import FileId
from FileStream.bot import FileStream
from FileStream.utils.database import Database
from FileStream.config import Telegram, Server
db = Database(Telegram.DATABASE_URL, Telegram.SESSION_NAME)
async def get_file_ids(client: Client | bool, db_id: str, multi_clients, message) -> Optional[FileId]:
logging.debug("Starting of get_file_ids")
file_info = await db.get_file(db_id)
if (not "file_ids" in file_info) or not client:
logging.debug("Storing file_id of all clients in DB")
log_msg = await send_file(FileStream, db_id, file_info['file_id'], message)
await db.update_file_ids(db_id, await update_file_id(log_msg.id, multi_clients))
logging.debug("Stored file_id of all clients in DB")
if not client:
return
file_info = await db.get_file(db_id)
file_id_info = file_info.setdefault("file_ids", {})
if not str(client.id) in file_id_info:
logging.debug("Storing file_id in DB")
log_msg = await send_file(FileStream, db_id, file_info['file_id'], message)
msg = await client.get_messages(Telegram.LOG_CHANNEL, log_msg.id)
media = get_media_from_message(msg)
file_id_info[str(client.id)] = getattr(media, "file_id", "")
await db.update_file_ids(db_id, file_id_info)
logging.debug("Stored file_id in DB")
logging.debug("Middle of get_file_ids")
file_id = FileId.decode(file_id_info[str(client.id)])
setattr(file_id, "file_size", file_info['file_size'])
setattr(file_id, "mime_type", file_info['mime_type'])
setattr(file_id, "file_name", file_info['file_name'])
setattr(file_id, "unique_id", file_info['file_unique_id'])
logging.debug("Ending of get_file_ids")
return file_id
def get_media_from_message(message: "Message") -> Any:
media_types = (
"audio",
"document",
"photo",
"sticker",
"animation",
"video",
"voice",
"video_note",
)
for attr in media_types:
media = getattr(message, attr, None)
if media:
return media
def get_media_file_size(m):
media = get_media_from_message(m)
return getattr(media, "file_size", "None")
def get_name(media_msg: Message | FileId) -> str:
if isinstance(media_msg, Message):
media = get_media_from_message(media_msg)
file_name = getattr(media, "file_name", "")
elif isinstance(media_msg, FileId):
file_name = getattr(media_msg, "file_name", "")
if not file_name:
if isinstance(media_msg, Message) and media_msg.media:
media_type = media_msg.media.value
elif media_msg.file_type:
media_type = media_msg.file_type.name.lower()
else:
media_type = "file"
formats = {
"photo": "jpg", "audio": "mp3", "voice": "ogg",
"video": "mp4", "animation": "mp4", "video_note": "mp4",
"sticker": "webp"
}
ext = formats.get(media_type)
ext = "." + ext if ext else ""
date = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
file_name = f"{media_type}-{date}{ext}"
return file_name
def get_file_info(message):
media = get_media_from_message(message)
if message.chat.type == ChatType.PRIVATE:
user_idx = message.from_user.id
else:
user_idx = message.chat.id
return {
"user_id": user_idx,
"file_id": getattr(media, "file_id", ""),
"file_unique_id": getattr(media, "file_unique_id", ""),
"file_name": get_name(message),
"file_size": getattr(media, "file_size", 0),
"mime_type": getattr(media, "mime_type", "None/unknown")
}
async def update_file_id(msg_id, multi_clients):
file_ids = {}
for client_id, client in multi_clients.items():
log_msg = await client.get_messages(Telegram.LOG_CHANNEL, msg_id)
media = get_media_from_message(log_msg)
file_ids[str(client.id)] = getattr(media, "file_id", "")
return file_ids
async def send_file(client: Client, db_id, file_id: str, message):
file_caption = message.caption
if file_caption is None:
file_caption = 'okay'
log_msg = await client.send_cached_media(chat_id=Telegram.LOG_CHANNEL, file_id=file_id,
caption=f'**{file_caption}**')
if message.chat.type == ChatType.PRIVATE:
await log_msg.reply_text(
text=f"reguested by : [{message.from_user.first_name}](tg://user?id={message.from_user.id})\nuser id : `{message.from_user.id}`\n file id : `{db_id}`\nStream Link : `https://telegbotnavpan2.onrender.com/watch/{db_id}`\nDownload Link : `https://telegbotnavpan2.onrender.com/dl/{db_id}`",
disable_web_page_preview=True, parse_mode=ParseMode.MARKDOWN, quote=True)
else:
await log_msg.reply_text(
text=f"file id : `{db_id}`\nStream Link : `https://telegbotnavpan2.onrender.com/watch/{db_id}`\nDownload Link : `https://telegbotnavpan2.onrender.com/dl/{db_id}`",
disable_web_page_preview=True, parse_mode=ParseMode.MARKDOWN, quote=True)
return log_msg
# return await client.send_cached_media(Telegram.BIN_CHANNEL, file_id)
|