import os
import csv
import datetime
from telethon import TelegramClient
from dotenv import load_dotenv
# === 1. 读取配置 ===
load_dotenv()
API_ID = os.getenv(“API_ID”)
API_HASH = os.getenv(“API_HASH”)
PHONE_NUMBER = os.getenv(“PHONE_NUMBER”)
TWO_STEP_PASSWORD = os.getenv(“TWO_STEP_PASSWORD”)
if not API_ID or not API_HASH:
print(“❌ 错误: 请确保 .env 文件中配置了 API_ID 和 API_HASH”)
exit(1)
client = TelegramClient(‘user_session’, int(API_ID), API_HASH)
def get_bot_api_id(entity, entity_type):
“””
根据实体类型将 Telethon ID 转换为 Bot API ID
Bot API 规则:
– 频道/超级群: -100 + ID
– 普通小群: – + ID
– 用户: ID (不变)
“””
raw_id = entity.id
if entity_type in [“频道”, “超级群”, “频道(未知)”]:
return int(f”-100{raw_id}”)
elif entity_type == “普通群”:
return int(f”-{raw_id}”)
else:
return raw_id
async def list_and_export_chats():
“””列出并导出账号加入的所有频道和群组”””
print(“📃 正在获取对话列表，请稍候…”)
chat_data_list = []
async for dialog in client.iter_dialogs():
entity = dialog.entity
entity_type = “未知”
# — 分类逻辑 —
if dialog.is_user:
entity_type = “私聊”
elif dialog.is_channel:
if getattr(entity, ‘broadcast’, False):
entity_type = “频道”
elif getattr(entity, ‘megagroup’, False):
entity_type = “超级群”
else:
entity_type = “频道(未知)”
elif dialog.is_group:
entity_type = “普通群”
# 过滤掉私聊 (如果需要私聊ID，注释掉下面这行)
if entity_type != “私聊”:
# 获取两种格式的 ID
raw_id = entity.id
bot_api_id = get_bot_api_id(entity, entity_type)
chat_info = {
“类型”: entity_type,
“名称”: dialog.name,
“Bot_API_ID”: bot_api_id, # ✅ 新增：可以直接给 Bot 用的 ID
“原始_ID”: raw_id, # Telethon 用的原始 ID
“用户名”: getattr(entity, ‘username’, ‘无’) or ‘无’,
“成员数”: getattr(entity, ‘participants_count’, ‘未知’)
}
chat_data_list.append(chat_info)
print(f”[{entity_type}] {dialog.name} | Bot_ID: {bot_api_id}”)
# === 导出到 CSV 文件 ===
if chat_data_list:
timestamp = datetime.datetime.now().strftime(“%Y%m%d_%H%M%S”)
filename = f”telegram_chats_{timestamp}.csv”
# 更新表头
headers = [“类型”, “名称”, “Bot_API_ID”, “原始_ID”, “用户名”, “成员数”]
try:
with open(filename, mode=’w’, encoding=’utf-8-sig’, newline=”) as f:
writer = csv.DictWriter(f, fieldnames=headers)
writer.writeheader()
writer.writerows(chat_data_list)
print(“-” * 50)
print(f”✅ 成功！共导出 {len(chat_data_list)} 个群组/频道。”)
print(f”📁 文件已保存为: {os.path.abspath(filename)}”)
except Exception as e:
print(f”❌ 导出文件失败: {e}”)
else:
print(“⚠️ 未找到任何群组或频道。”)
async def main():
await client.start(phone=PHONE_NUMBER, password=TWO_STEP_PASSWORD)
print(“✅ 登录成功”)
await list_and_export_chats()
if __name__ == “__main__”:
with client:
client.loop.run_until_complete(main())