telegram-dl/telegram-dl.py

#!/usr/bin/env python3
import asyncio
import base64
import datetime
import json
import os
import re
import sys
import telethon


API_ID = os.environ['TELEGRAM_API_ID']
API_HASH = os.environ['TELEGRAM_API_HASH']
BOT_TOKEN = os.environ['TELEGRAM_BOT_TOKEN']
URL_PATTERN = re.compile(r'^https?://t\.me/(?:s/)?(?P<channel>[^/]+)/(?P<message>\d+)$')


def stuff_to_json(o):
	if isinstance(o, datetime.datetime):
		return o.isoformat()
	if isinstance(o, bytes):
		return f'binary data: {base64.b64encode(o).decode("ascii")}'
	raise TypeError(f'Object of type {type(o)} is not JSON serializable')


async def main():
	# Parse URLs
	targets = []
	for url in sys.argv[1:]:
		m = URL_PATTERN.match(url)
		if not m:
			print(f'Error: {url} is not a recognised Telegram URL', file = sys.stderr)
			sys.exit(1)
		targets.append((m['channel'], int(m['message'])))
	if not targets:
		print(f'Usage: telegram-dl.py URL [URL...]', file = sys.stderr)
		sys.exit(1)
	channelName = targets[0][0]
	if not all(x[0] == channelName for x in targets[1:]):
		print(f'Error: all URLs must be of the same channel', file = sys.stderr)
		sys.exit(1)
	ids = [x[1] for x in targets]

	# Let's go...
	client = telethon.TelegramClient('telegram-dl', API_ID, API_HASH)
	await client.start(bot_token = BOT_TOKEN)
	messages = await client.get_messages(channelName, ids = ids)
	for message in messages:
		if not message:
			continue
		with open(f'{channelName}_{message.id}.json', 'x') as fp:
			json.dump(message.to_dict(), fp, default = stuff_to_json)

		await client.download_media(message)


asyncio.run(main())