[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())

error i am getting:

[N] ➜  test git:(main) ✗ .venv/bin/python test.py
[10] Retrying "upload.SaveBigFilePart" due to: [Errno 32] Broken pipe
Traceback (most recent call last):
  File "/var/home/hmk/proj/test/test.py", line 109, in <module>
    asyncio.run(main())
  File "/usr/lib64/python3.12/asyncio/runners.py", line 194, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/usr/lib64/python3.12/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib64/python3.12/asyncio/base_events.py", line 687, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/var/home/hmk/proj/test/test.py", line 103, in main
    await client.send_document(
  File "/var/home/hmk/proj/test/.venv/lib64/python3.12/site-packages/pyrogram/methods/messages/send_document.py", line 229, in send_document
    file = await self.save_file(document, progress=progress, progress_args=progress_args)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/var/home/hmk/proj/test/test.py", line 22, in save_file_custom_wrapper
    return await save_file_from_bytes_gen(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/var/home/hmk/proj/test/test.py", line 78, in save_file_from_bytes_gen
    await session.invoke(rpc)
  File "/var/home/hmk/proj/test/.venv/lib64/python3.12/site-packages/pyrogram/session/session.py", line 430, in invoke
    return await self.invoke(query, retries - 1, timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/var/home/hmk/proj/test/.venv/lib64/python3.12/site-packages/pyrogram/session/session.py", line 430, in invoke
    return await self.invoke(query, retries - 1, timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/var/home/hmk/proj/test/.venv/lib64/python3.12/site-packages/pyrogram/session/session.py", line 430, in invoke
    return await self.invoke(query, retries - 1, timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  [Previous line repeated 7 more times]
  File "/var/home/hmk/proj/test/.venv/lib64/python3.12/site-packages/pyrogram/session/session.py", line 420, in invoke
    raise e from None
  File "/var/home/hmk/proj/test/.venv/lib64/python3.12/site-packages/pyrogram/session/session.py", line 407, in invoke
    return await self.send(query, timeout=timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/var/home/hmk/proj/test/.venv/lib64/python3.12/site-packages/pyrogram/session/session.py", line 347, in send
    raise e
  File "/var/home/hmk/proj/test/.venv/lib64/python3.12/site-packages/pyrogram/session/session.py", line 344, in send
    await self.connection.send(payload)
  File "/var/home/hmk/proj/test/.venv/lib64/python3.12/site-packages/pyrogram/connection/connection.py", line 71, in send
    await self.protocol.send(data)
  File "/var/home/hmk/proj/test/.venv/lib64/python3.12/site-packages/pyrogram/connection/transport/tcp/tcp_abridged.py", line 39, in send
    await super().send(
  File "/var/home/hmk/proj/test/.venv/lib64/python3.12/site-packages/pyrogram/connection/transport/tcp/tcp.py", line 105, in send
    raise OSError(e)
OSError: [Errno 32] Broken pipe

how to run this code and reproduce problem:

curl -sSL gist.githubusercontent.com/gmankab/753a5225aa291c7f04d0bc17110b432e/raw/01c428e7e2f688152b3e2e4dffd7e772d89894c7/main.py -o main.py
python -m venv
.venb/bin/pip install -U pip
.venv/bin/pip install -U tgcrypto-pyrofork pyrofork
.venb/bin/python main.py

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())