Transfer files between Telegram and Google Drive with Colab via Telethon & FastTelethonHelper

KhushwantKhushwant
5 min read


Flow of the article ⏬
1. Get your telegram API ID and hash
2. Install necessary packages for Telethon and FastTelethonhelper
3. How to list all messages of a channel
4. Downloading files from Telegram to mounted Google Drive
5. Uploading multiple files from mounted Google Drive to Telegram using respective file paths
6. Some progress bars :)

Get your telegram API ID and hash

  1. Login to your Telegram account with the phone number on https://my.telegram.org/

  2. Click under API Development tools.

  3. A Create new application window will appear. Fill in your application details. There is no need to enter any URL, and only the first two fields (App title and Short name) can currently be changed later.

  4. Click on Create application at the end. Remember that your API hash is secret and Telegram won’t let you revoke it. Don’t post it anywhere!

  5. You need to use these in the code.

Now head to https://colab.research.google.com and mount your Google Drive.

from google.colab import drive
drive.mount('/content/drive')

Install necessary packages for Telethon and FastTelethonhelper

Telethon's Documentation - Telethon 1.30.3 documentation
Telegram is a popular messaging application. This library is meant to make it easy for you to write Python programs… docs.telethon.dev

FastTelethonhelper
Make Telethon files upload/download fasterpypi.org

!pip install telethon cryptg FastTelethonhelper PySocks
OR
!pip install telethon pyaesni FastTelethonhelper PySocks

While browsing github, I came across devs suggesting using pyaesni instead of cryptg for a little faster processing speed*.*

List all messages of a channel

from telethon import TelegramClient
from FastTelethonhelper import fast_upload
from FastTelethonhelper import fast_download
import time
import asyncio
from telethon.tl.types import PeerUser, PeerChat, PeerChannel
import nest_asyncio
nest_asyncio.apply()

api_id='xxxxxx' #Use the API ID and Hash here
api_hash='xxxxxx'
session_name= 'session1'

# Either use the channel invite link or use the Channel Code with PeerChannel()
# Get Channel code either through Telegram PC or Nekogram etc.
# channel_invite_link = 't.me/+xxxxx'
# List all messages of channel

channel_id=xxxxxxx
async def func(client, channel_id):
    entity = await client.get_entity(PeerChannel(channel_id))
    # entity = await client.get_entity(channel_invite_link)
    # entity = await client.get_entity('Channel_Name')

    async for message in reversed(client.iter_messages(entity)):
      try:
          print("--------------------------------------------------------------------------")
          print(f"{message.id} - {message.message[:100]}...")  # Print only first 100 characters for large messages
          print("--------------------------------------------------------------------------")
      except Exception as e:
          print(f"Error printing message {message.id}: {e}")
          continue

async def main():
    async with TelegramClient(session_name, api_id, api_hash) as client:
        await func(client, channel_id)

# Run the main function
asyncio.run(main())

Using the message.id we can choose what files to download from Telegram to Drive

First cd to directory where you want to download cd /content/drive/ABC

We can do it through Telethon only but the speed will be veryy slow. So we will be using FastTelethonhelper.

Through previous code you will find out the message id’s of the files to download.

# FastTelethon download

def progress_bar_str(done, total):
    percent = round(done/total*100, 2)
    strin = "░░░░░░░░░░"
    strin = list(strin)
    for i in range(round(percent)//10):
        strin[i] = "█"
    strin = "".join(strin)
    print(f"\r Downloaded {done}/{total} Percent: {percent}% {strin}",end='')
    if percent == 100.0:
        print('\t✅')

async def func(client, channel_id):
    entity = await client.get_entity(PeerChannel(channel_id))
    # entity = await client.get_entity('ChannelName')

    async for message in reversed(client.iter_messages(entity)):
      # Use start end here if needed
        # if 343 <= message.id <= 370:
            await handle_message_download(client, message)


async def handle_message_download(client, message):
    try:
        print("\033[93m --------------------------------------------------------------------------")
        print(f"\nDownloading File... {message.id} ---- {message.message[:100]}...")  # Limit message preview to first 100 chars
        downloaded_location = await fast_download(client, message, reply=message, progress_bar_function=progress_bar_str)
        print(f"\nDownloaded to: {downloaded_location}")
        print("\033[93m --------------------------------------------------------------------------")
        return downloaded_location
    except Exception as e:
        print("\033[93m --------------------------------------------------------------------------")
        print(f"Error downloading message {message.id}")
        # print(e)
        print("\033[93m --------------------------------------------------------------------------")
        return None


# Main entry point
channel_id=xxxxxxx
async def main():
    async with TelegramClient(session_name, api_id, api_hash) as client:
        await func(client, channel_id)


asyncio.run(main())

Sometimes you may encounter FloodWait errors. Some errors will tell you to wait for some time others nothing. To handle these we have to revert to the original telethon download methods instead of FastTelethonhelper.

await client.download_media(message, progress_callback=progress_bar_str)

Uploading multiple files from Drive to Telegram using the file paths

To get paths first go the desired folder by cd /content/drive/ABC and then use the command below to get the file paths. Copy them to use later.

!find $(pwd) -maxdepth 1 -type f -exec echo "\"{}\"," \; | sort
from telethon import TelegramClient
from FastTelethonhelper import fast_upload
import time
import asyncio
import nest_asyncio
nest_asyncio.apply()

api_id='xxxxxx'
api_hash='xxxxxx'
session_name= 'session1'

# Create a public channel and use it's invite link here.
channel_invite_link = 't.me/xxxx'

file_paths = (
"/content/drive/MyDrive/VID/9",
"/content/drive/MyDrive/VID/10",
"/content/drive/MyDrive/VID/11",
"/content/drive/MyDrive/VID/12",
 )

# Progressbar
def callback(current, total):
    print("\r {}%".format(current * 100 / total),end='')

async def func():
  for f in file_paths:
    entity = await client.get_entity(channel_invite_link)
    print(f"Uploading {f}")
    new_file = await fast_upload(client, f,reply=f,progress_bar_function=callback)
    await client.send_file(entity, new_file,force_document=True)

async with TelegramClient(session_name, api_id, api_hash) as client:
    client.loop.run_until_complete(func())

Sometimes there may be issue with where the downloaded files are. This may happen when you forget to cd before downloading. On the left panel of colab, use the file explorer to lacate your files and move them.

Sometimes you may also encounter FloodWait errors again and again. Some errors will tell you to wait for some time others nothing. To handle these we have to revert to the original telethon upload files methods instead of FastTelethonhelper.

async def func():
  for f in file_paths:
    entity = await client.get_entity(channel_invite_link)
    print(f"Uploading {f}")
    await client.send_file(entity,f,force_document=True)

Link to telethon upload methods documentation : https://docs.telethon.dev/en/stable/modules/client.html#telethon.client.uploads.UploadMethods.send_file

Some progress bars

We are using \r in the starting of each print and end= ‘’ so that the progress output gets updated inplace instead of showing new messages ie overwriting the previous message.
Credit => https://github.com/googlecolab/colabtools/issues/166

def callback(current, total):
    print('\r Downloaded', current, 'out of', total,'bytes: {:.2%}'.format(current / total),end='')

def progress_bar_str(done, total):
    percent = round(done/total*100, 2)
    strin = "░░░░░░░░░░"
    strin = list(strin)
    for i in range(round(percent)//10):
        strin[i] = "█"
    strin = "".join(strin)
    print(f"\r Percent: {percent}% {strin}",end='')
    if round(percent) == 100:
        print('\t✅')

def progress_bar(current, total):
  print("\r {}%".format(current * 100 / total),end='')

I will keep updating this article as I myself use this a lot😉.

Last update 22 August 2025

0
Subscribe to my newsletter

Read articles from Khushwant directly inside your inbox. Subscribe to the newsletter, and don't miss out.

Written by

Khushwant
Khushwant

Current...Masters@IIT Kanpur