Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Corrupted file video and half image being loaded #16

Open
0tanvi00 opened this issue Aug 18, 2024 · 1 comment
Open

Corrupted file video and half image being loaded #16

0tanvi00 opened this issue Aug 18, 2024 · 1 comment

Comments

@0tanvi00
Copy link

for small size image this thing works but for larger size it fails

import os
import asyncio
from telethon import TelegramClient
from telethon_secret_chat import SecretChatManager
from telethon_secret_chat.secret_sechma.secretTL import DecryptedMessageMediaPhoto, DecryptedMessageMediaVideo
phone_number = os.getenv("tg_number")
client = TelegramClient(os.getenv("tg_session_name"), os.getenv("tg_api_id"), os.getenv("tg_api_hash"))


async def downloader(event):
    
    if event.decrypted_event.media is None:
        return

    if isinstance(event.decrypted_event.media, DecryptedMessageMediaPhoto):
        file_name = 'see.jpg'
    elif isinstance(event.decrypted_event.media, DecryptedMessageMediaVideo):
        file_name = 'see.mp4'
    else:
        file_name = 'see.unknown' 

    print(f"Processing {file_name}")

    cache = await manager.download_secret_media(event.decrypted_event)

    with open(f"./{file_name}", "wb") as f:
        f.write(cache)
        f.close()


async def new_chat(chat, created_by_me):
    if created_by_me:
        print(f"User {chat} has accepted our secret chat request")
    else:
        print(f"We have accepted the secret chat request of {chat}")

manager = SecretChatManager(client, auto_accept=True, new_chat_created=new_chat)
manager.add_secret_event_handler(func=downloader)



async def main():
    await client.start(phone_number)
    print("Connected to Telegram \/")

    await client.run_until_disconnected()

asyncio.get_event_loop().run_until_complete(main())
@0tanvi00
Copy link
Author

I have figured out a solution for images, we need to supply rewrite this function as

 async def download_secret_media(self, message):
        if not message.file or not isinstance(message.file, EncryptedFile):
            return b""
        key_fingerprint = message.file.key_fingerprint
        key = message.media.key
        iv = message.media.iv
        digest = md5(key + iv).digest()

        fingerprint = int.from_bytes(digest[:4], byteorder="little", signed=True) ^ int.from_bytes(digest[4:8],
                                                                                                   byteorder="little",
                                                                                                   signed=True)
        if fingerprint != key_fingerprint:
            raise SecurityError("Wrong fingerprint")
        media = await self.client.download_file(InputEncryptedFileLocation(message.file.id, message.file.access_hash),
                                                part_size_kb=512, # here we need to give part_size 
                                                key=message.media.key,
                                                iv=message.media.iv)
        return media

I still need help for videos

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

1 participant