[solved] issue with resending file through pyrofork

main.py

import pyrogram.methods.advanced.save_file
import pyrogram.session.session
import pyrogram.client
import pyrogram.types
import pyrogram.raw
import pyrogram
import asyncio
import typing
import types


async def save_file_custom_wrapper(
    self: pyrogram.client.Client,
    path: str | typing.BinaryIO | typing.AsyncGenerator[bytes, None],
    file_id: int | None = None,
    file_part: int = 0,
    progress: typing.Callable | None = None,
    progress_args: tuple = ()
):
    if isinstance(path, typing.AsyncGenerator):
        # runs custom function if got async bytes generator
        return await save_file_from_bytes_gen(
            self=self,
            bytes_gen=path,
        )
    else:
        # runs default pyrofork's save_file function if got something else
        return await pyrogram.methods.advanced.save_file.SaveFile.save_file( # type: ignore
            self=self,
            path=path,
            file_id=file_id,
            file_part=file_part,
            progress=progress,
            progress_args=progress_args,
        )


async def save_file_from_bytes_gen(
    self: pyrogram.client.Client,
    bytes_gen: typing.AsyncGenerator[bytes, None],
    file_name: str = 'filename',
):
    '''
    custom save_file variant that accepts async bytes generator

    can be used only with big files

    file size should be more than 10 * 1024 * 1024 bytes (10 megabytes)

    can resend big file without using disk and without writing while file to disk

    useful when file size is more then ram size
    '''
    assert self.me
    file_size_limit_mib = 4000 if self.me.is_premium else 2000
    session = pyrogram.session.session.Session(
        self,
        await self.storage.dc_id(), # type: ignore
        await self.storage.auth_key(), # type: ignore
        await self.storage.test_mode(), # type: ignore
        is_media=True,
    )
    await session.start()
    file_id = self.rnd_id()
    file_total_parts: int = 0
    file_size: int = 0
    async for chunk in bytes_gen:
        file_total_parts += 1
        file_size += len(chunk)
        if file_size > file_size_limit_mib * 1024 * 1024:
            raise ValueError(f"can't upload files bigger than {file_size_limit_mib} MiB")
        rpc = pyrogram.raw.functions.upload.SaveBigFilePart( # type: ignore
            file_id=file_id,
            file_part=file_total_parts - 1,
            file_total_parts=file_total_parts,
            bytes=chunk
        )
        await session.invoke(rpc)
    return pyrogram.raw.types.input_file_big.InputFileBig(
        id=file_id,
        parts=file_total_parts,
        name=file_name,
    )


async def main():
    client = pyrogram.client.Client(
        name='tg_bot',
    )
    # replacing pyrofork's native save_file method with custom wrapper
    client.save_file = types.MethodType(save_file_custom_wrapper, client)
    await client.start()
    source_msg = await client.get_messages(
        chat_id='@TAndroidAPK',
        message_ids=385,
    )
    assert isinstance(source_msg, pyrogram.types.Message)
    assert source_msg.document.file_size > 10 * 1024 * 1024
    stream_media = client.stream_media(
        message=source_msg,
    )
    assert isinstance(stream_media, typing.AsyncGenerator)
    await client.send_document(
        chat_id='me',
        document=stream_media, # type: ignore
    )


asyncio.run(main())

fixed code:

import pyrogram.methods.advanced.save_file
import pyrogram.session.session
import pyrogram.client
import pyrogram.types
import pyrogram.raw
import pyrogram
import rich.progress
import asyncio
import typing
import types
import math


media_file = typing.Union[
    pyrogram.types.Audio,
    pyrogram.types.Photo,
    pyrogram.types.Video,
    pyrogram.types.Voice,
    pyrogram.types.Document,
    pyrogram.types.Animation,
    pyrogram.types.VideoNote,
]


class ChunkGen:
    def __init__(
        self,
        msg: pyrogram.types.Message,
        media: media_file,
        client: pyrogram.client.Client,
        remove_progress: bool = True,
    ):
        self.remove_progress: bool = remove_progress
        self.progress = rich.progress.Progress()
        self.total: int = media.file_size
        self.name: str = getattr(media, 'file_name', 'noname')
        self.msg: pyrogram.types.Message = msg
        self.return_next: bytes = b''
        self.max_chunk_size = 512 * 1024
        self.client: pyrogram.client.Client = client
        if msg.chat.username:
            self.description = f'big file {msg.chat.username}/{msg.id}'
        else:
            self.description = f'big file {msg.chat.full_name}/{msg.id}'
        assert self.total > 10 * 1024 * 1024

    def __aiter__(self):
        return self.stream_512()

    async def stream_512(
        self,
    ) -> typing.AsyncGenerator[bytes, None]:
        task_id = self.progress.add_task(
            description=self.description,
            total=self.total,
            transfer=True,
        )
        stream_1024 = self.client.stream_media(self.msg)
        assert isinstance(stream_1024, typing.AsyncGenerator)
        completed: int = 0
        async for chunk_1024 in stream_1024:
            while len(chunk_1024) > self.max_chunk_size:
                completed += self.max_chunk_size
                self.progress.update(
                    task_id=task_id,
                    completed=completed,
                )
                yield chunk_1024[:self.max_chunk_size]
                chunk_1024 = chunk_1024[self.max_chunk_size:]
            completed += len(chunk_1024)
            self.progress.update(
                task_id=task_id,
                completed=completed,
            )
            yield chunk_1024
        self.progress.stop_task(
            task_id=task_id,
        )
        if self.remove_progress:
            self.progress.update(
                task_id=task_id,
                visible=False,
            )


async def save_file_custom_wrapper(
    self: pyrogram.client.Client,
    path: str | typing.BinaryIO | ChunkGen,
    file_id: int | None = None,
    file_part: int = 0,
    progress: typing.Callable | None = None,
    progress_args: tuple = ()
):
    if isinstance(path, ChunkGen):
        # runs custom function if got async bytes generator
        return await save_file_from_bytes_gen(
            self=self,
            chunk_gen=path,
        )
    else:
        # runs default pyrofork's save_file function if got something else
        return await pyrogram.methods.advanced.save_file.SaveFile.save_file( # type: ignore
            self=self,
            path=path,
            file_id=file_id,
            file_part=file_part,
            progress=progress,
            progress_args=progress_args,
        )


async def save_file_from_bytes_gen(
    self: pyrogram.client.Client,
    chunk_gen: ChunkGen,
):
    '''
    - custom save_file variant that accepts async bytes generator
    - can be used only with big files
    - file size should be more than 10 * 1024 * 1024 bytes (10 megabytes)
    - can resend big file without using disk and without writing while file to disk
    - useful when file size is more then ram size
    '''
    assert self.me
    file_size_limit_mib = 4000 if self.me.is_premium else 2000
    session = pyrogram.session.session.Session(
        self,
        await self.storage.dc_id(), # type: ignore
        await self.storage.auth_key(), # type: ignore
        await self.storage.test_mode(), # type: ignore
        is_media=True,
    )
    await session.start()
    file_id = self.rnd_id()
    max_chunk_size = 512 * 1024
    file_total_parts: int = math.ceil(
        chunk_gen.total / max_chunk_size
    )
    chunk_index: int = 0
    file_size: int = 0
    async for chunk in chunk_gen:
        assert len(chunk) <= max_chunk_size
        assert chunk_index < file_total_parts
        if chunk_index + 1 != file_total_parts:
            assert len(chunk) % 1024 == 0
            assert max_chunk_size % len(chunk) == 0
        file_size += len(chunk)
        if file_size > file_size_limit_mib * 1024 * 1024:
            raise ValueError(f"can't upload files bigger than {file_size_limit_mib} MiB")
        rpc = pyrogram.raw.functions.upload.SaveBigFilePart( # type: ignore
            file_id=file_id,
            file_part=chunk_index,
            file_total_parts=file_total_parts,
            bytes=chunk
        )
        await session.invoke(rpc)
        chunk_index += 1
    await session.stop()
    return pyrogram.raw.types.input_file_big.InputFileBig(
        id=file_id,
        parts=file_total_parts,
        name=chunk_gen.name,
    )


async def main():
    client = pyrogram.client.Client(
        name='tg_bot',
    )
    # replacing pyrofork's native save_file method with custom wrapper
    client.save_file = types.MethodType(save_file_custom_wrapper, client)
    await client.start()
    source_msg = await client.get_messages(
        chat_id='@gmanka',
        message_ids=11354,
    )
    assert isinstance(source_msg, pyrogram.types.Message)
    file_chunk_gen = ChunkGen(
        msg=source_msg,
        media=source_msg.document,
        client=client,
    )
    await client.send_document(
        chat_id='me',
        document=file_chunk_gen, # type: ignore
    )


asyncio.run(main())