如何用Python结合Telegram API下载频道文件并检测新文件

准备工作 #

• 安装依赖库：pip install telethon
• 获取Telegram API凭证：前往my.telegram.org注册应用，获取api_id和api_hash
• 确保你有权限访问目标频道（公开频道无需额外权限，私有频道需先加入）

核心实现步骤 #

1. 初始化客户端 #

创建Telegram客户端实例，用于和Telegram API交互：

from telethon import TelegramClient, events
from telethon.tl.types import MessageMediaDocument
import os
import asyncio

# 替换为你的实际参数
API_ID = 123456
API_HASH = "your_api_hash_here"
SESSION_NAME = "telegram_file_bot"
CHANNEL_USERNAME = "@target_channel"
DOWNLOAD_DIR = "./telegram_downloads"

# 创建下载目录
os.makedirs(DOWNLOAD_DIR, exist_ok=True)

# 初始化客户端
client = TelegramClient(SESSION_NAME, API_ID, API_HASH)

2. 下载频道全部文件 #

遍历频道所有消息，筛选出带文件的消息并下载，同时记录已下载的消息ID避免重复：

async def download_all_files():
    channel = await client.get_entity(CHANNEL_USERNAME)
    # 读取已下载的消息ID
    downloaded_ids = set()
    if os.path.exists("downloaded_ids.txt"):
        with open("downloaded_ids.txt", "r") as f:
            downloaded_ids = set(map(int, f.read().splitlines()))
    
    # 遍历频道消息（reverse=True从最早到最新遍历）
    async for message in client.iter_messages(channel, reverse=True):
        if isinstance(message.media, MessageMediaDocument) and message.id not in downloaded_ids:
            print(f"下载文件: {message.file.name}")
            await message.download_media(file=os.path.join(DOWNLOAD_DIR, message.file.name))
            # 记录已下载ID
            with open("downloaded_ids.txt", "a") as f:
                f.write(f"{message.id}\n")

3. 检测并下载新增文件 #

提供两种实现方式，按需选择：

方式一：定时轮询检测

适合不需要实时响应的场景，每隔固定时间检查频道新增内容：

async def check_new_files():
    channel = await client.get_entity(CHANNEL_USERNAME)
    # 读取上次处理的最后消息ID
    last_id = 0
    if os.path.exists("last_processed_id.txt"):
        with open("last_processed_id.txt", "r") as f:
            last_id = int(f.read().strip())
    
    # 获取last_id之后的新消息
    async for message in client.iter_messages(channel, min_id=last_id):
        if isinstance(message.media, MessageMediaDocument):
            print(f"检测到新文件: {message.file.name}，开始下载")
            await message.download_media(file=os.path.join(DOWNLOAD_DIR, message.file.name))
            last_id = message.id
    
    # 更新最后处理的消息ID
    with open("last_processed_id.txt", "w") as f:
        f.write(str(last_id))

方式二：实时监听新消息

基于事件驱动，一旦频道有新消息立即处理：

@client.on(events.NewMessage(chats=CHANNEL_USERNAME))
async def on_new_message(event):
    message = event.message
    if isinstance(message.media, MessageMediaDocument):
        print(f"实时收到新文件: {message.file.name}，开始下载")
        await message.download_media(file=os.path.join(DOWNLOAD_DIR, message.file.name))
        # 更新最后处理ID
        with open("last_processed_id.txt", "w") as f:
            f.write(str(message.id))

4. 主执行逻辑 #

根据选择的检测方式，编写主函数：

async def main():
    await client.start()
    # 可选：先执行一次全量下载
    # await download_all_files()
    
    # 方式一：启动定时轮询（每5分钟检查一次）
    while True:
        await check_new_files()
        await asyncio.sleep(300)
    
    # 方式二：启动实时监听（注释上面的循环，启用此行）
    # await client.run_until_disconnected()

if __name__ == "__main__":
    asyncio.run(main())

注意事项 #

• 首次运行客户端会要求输入手机号和验证码，登录后生成会话文件，后续运行无需重复登录
• 对于消息量极大的频道，建议在iter_messages中添加limit参数分批处理，避免内存占用过高
• 若下载大文件出现超时，可在客户端初始化时添加timeout参数，例如TelegramClient(..., timeout=300)
• 私有频道需确保你的账号已加入，否则无法获取消息

内容的提问来源于stack exchange，提问作者Mr. Magic