Spaces:
Paused
Paused
| from telethon.events import CallbackQuery | |
| from bot import TelegramBot | |
| from bot.modules.decorators import verify_user,byte_to_gb_mb | |
| from bot.modules.static import * | |
| from bot.modules.telegram import get_message, get_file_properties | |
| from bot.config import Telegram, Server | |
| '''async def upload_file_in_chunks(TelegramBot, chat_id, file_path, chunk_size_kb): | |
| file_id = None | |
| total_parts = math.ceil(file_size / (chunk_size_kb * 1024)) | |
| current_part = 1 | |
| with open(file_path, "rb") as f: | |
| while True: | |
| data = f.read(chunk_size_kb * 1024) | |
| if not data: | |
| break | |
| # Upload the current chunk | |
| file_part = await client.upload_file_part(file_id, current_part, data) | |
| if not file_id: | |
| file_id = file_part.file_id | |
| current_part += 1 | |
| # Show progress if desired | |
| progress = current_part / total_parts * 100 | |
| print(f"Uploaded {progress:.2f}%") | |
| # Complete the upload | |
| complete_file = await client.upload_file_completed(file_id, total_parts) | |
| # Send the uploaded file to the chat | |
| await client.send_file(chat_id, complete_file) | |
| ''' | |
| #@verify_user(private=True) | |
| async def delete_file(event: CallbackQuery.Event): | |
| query_data = event.query.data.decode().split('_') | |
| if len(query_data) != 3: | |
| return await event.answer(InvalidQueryText, alert=True) | |
| message = await get_message(int(query_data[1])) | |
| if not message: | |
| return await event.answer(MessageNotExist, alert=True) | |
| await message.delete() | |
| return await event.answer(LinkRevokedText, alert=True) | |
| #@verify_user(private=True) | |
| async def up_file(event:CallbackQuery.Event): | |
| query_data = event.query.data.decode().split('_') | |
| if len(query_data) != 5: | |
| return await event.answer(InvalidQueryText, alert=True) | |
| message_id=int(query_data[1]) | |
| event_id=int(query_data[3]) | |
| event_peer=int(query_data[4]) | |
| message = await get_message(int(query_data[1])) | |
| main_event = await TelegramBot.get_messages(event_peer, ids=event_id) | |
| #print(event_id,event_peer) | |
| #print(main_event) | |
| code = str(query_data[2]) | |
| def callback(current, total,event_peer,event_id): | |
| print('Uploaded', current, 'out of', total,'bytes: {:.2%}'.format(current / total)) | |
| TelegramBot.edit_message(event_peer, event_id, f'Uploaded, {current}, out of, {total}') | |
| if not message or code != message.raw_text: | |
| return await event.answer(MessageNotExist, alert=True) | |
| else: | |
| dl_link = f'{Server.BASE_URL}/dl/{message_id}?code={code}' | |
| file_name, file_size, mime_type = get_file_properties(main_event) | |
| await main_event.reply( | |
| message= f'**File name:{file_name} \nFile size:{byte_to_gb_mb(file_size)} \nID:{message_id} \nAccess Code:{code} \n*If you want to upload with different name please send name with reply of the message \n reply with n for original file name ***', | |
| parse_mode="Markdown" | |
| ) | |