使用TDLib获取Telegram频道订阅数的问题求助
TDLib调用返回None的核心问题
- TDLib是异步请求-响应模型:你调用
td_send只是把请求发送给TDLib客户端,但不会同步返回结果。你的td_send函数本身没有返回语句,默认就返回None;而且TDLib的响应会通过独立的事件回调/接收逻辑返回,不是调用函数时直接给到你。 - 你完全没处理TDLib的响应接收逻辑,所以打印
td_send的返回值必然是None,这不是TDLib的问题,是你的实现逻辑缺了关键环节。
用TDLib实现订阅数周期统计的补充步骤
- 必须实现响应监听循环:得写一个接收TDLib输出的函数(比如
td_receive),持续读取TDLib返回的数据,解析出对应getChat请求的响应。
示例实现逻辑:import json def td_receive(): # 替换成你实际从TDLib接收数据的逻辑(比如读管道、套接字) response_raw = _td_receive(client_id) return json.loads(response_raw.decode('utf-8')) # 先发送请求,再循环等待响应 td_send({'@type':'getChat', 'chat_id':-11231231231, 'extra':1001011}) while True: response = td_receive() # 通过extra字段匹配我们发送的请求(避免混进其他响应) if response.get('@type') == 'chat' and response.get('extra') == 1001011: # 从chat对象里提取订阅数 subscriber_count = response['member_count'] print(f"当前订阅数:{subscriber_count}") break - 添加定时任务实现周期统计:用定时工具(比如
schedule库,或者简单的time.sleep循环)定期触发getChat请求,每次拿到订阅数后写入日志或数据库,记录时间戳和数值。
示例定时逻辑:import schedule import time from datetime import datetime def log_subscriber_count(): td_send({'@type':'getChat', 'chat_id':-11231231231, 'extra':1001011}) # 监听响应并记录 while True: response = td_receive() if response.get('@type') == 'chat' and response.get('extra') == 1001011: count = response['member_count'] with open('subscriber_log.txt', 'a') as f: f.write(f"{datetime.now().isoformat()},{count}\n") break time.sleep(1) # 防止重复读取相同响应 # 设置每小时执行一次统计 schedule.every().hour.do(log_subscriber_count) # 启动定时任务循环 while True: schedule.run_pending() time.sleep(60) - 验证chat_id有效性:确保你用的
chat_id是真实有效的频道ID(负号开头是频道/超级群的正确格式,但可以先调用TDLib的searchPublicChat通过频道用户名获取正确ID,避免填错)。
备选方案:用Telethon快速实现
如果觉得TDLib的异步逻辑太繁琐,Telethon更适合快速落地,它是封装好的Telegram客户端库,API更直观:
- 初始化客户端并登录:
from telethon import TelegramClient from telethon.tl.functions.channels import GetFullChannelRequest import schedule import time from datetime import datetime # 替换成你从my.telegram.org获取的API信息 api_id = 123456 api_hash = '你的api_hash字符串' client = TelegramClient('订阅统计会话', api_id, api_hash) - 实现订阅数获取与周期记录:
async def fetch_subscriber_count(channel_username): async with client: # 通过频道用户名获取实体 channel = await client.get_entity(channel_username) # 获取完整频道信息,包含订阅数 full_channel = await client(GetFullChannelRequest(channel=channel)) return full_channel.full_chat.participants_count def record_count(): # 同步执行异步函数 count = client.loop.run_until_complete(fetch_subscriber_count('你的频道用户名')) # 写入日志 with open('subscriber_log.txt', 'a') as f: f.write(f"{datetime.now().isoformat()},{count}\n") print(f"已记录:{datetime.now()} - 订阅数{count}") # 设置定时任务,比如每小时执行一次 schedule.every().hour.do(record_count) if __name__ == '__main__': client.start() # 首次运行需要扫码登录 # 保持脚本运行 while True: schedule.run_pending() time.sleep(60) - 优势:不用处理复杂的异步响应循环,内置实体解析功能,直接用频道用户名就能获取信息,减少ID错误的概率,上手更快。
内容的提问来源于stack exchange,提问作者Владимир Пушкарев
相关产品推荐
相关产品推荐

