import os
import csv
import datetime
from telethon import TelegramClient
from dotenv import load_dotenv
# === 1. \u8bfb\u53d6\u914d\u7f6e \u00a0 \u83b7\u53d6\u9891\u9053\u7fa4\u7ec4\u4fe1\u606f ===
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(“\u274c \u9519\u8bef: \u8bf7\u786e\u4fdd .env \u6587\u4ef6\u4e2d\u914d\u7f6e\u4e86 API_ID \u548c API_HASH”)
exit(1)
client = TelegramClient(‘session\/user_session’, int(API_ID), API_HASH)
def get_bot_api_id(entity, entity_type):
“””
\u6839\u636e\u5b9e\u4f53\u7c7b\u578b\u5c06 Telethon ID \u8f6c\u6362\u4e3a Bot API ID
Bot API \u89c4\u5219:
– \u9891\u9053\/\u8d85\u7ea7\u7fa4: -100 + ID
– \u666e\u901a\u5c0f\u7fa4: – + ID
– \u7528\u6237: ID (\u4e0d\u53d8)
“””
raw_id = entity.id
if entity_type in [“\u9891\u9053”, “\u8d85\u7ea7\u7fa4”, “\u9891\u9053(\u672a\u77e5)”]:
return int(f”-100{raw_id}”)
elif entity_type == “\u666e\u901a\u7fa4”:
return int(f”-{raw_id}”)
else:
return raw_id
async def list_and_export_chats():
“””\u5217\u51fa\u5e76\u5bfc\u51fa\u8d26\u53f7\u52a0\u5165\u7684\u6240\u6709\u9891\u9053\u548c\u7fa4\u7ec4”””
print(“\ud83d\udcc3 \u6b63\u5728\u83b7\u53d6\u5bf9\u8bdd\u5217\u8868\uff0c\u8bf7\u7a0d\u5019…”)
chat_data_list = []
async for dialog in client.iter_dialogs():
entity = dialog.entity
entity_type = “\u672a\u77e5”
# — \u5206\u7c7b\u903b\u8f91 —
if dialog.is_user:
entity_type = “\u79c1\u804a”
elif dialog.is_channel:
if getattr(entity, ‘broadcast’, False):
entity_type = “\u9891\u9053”
elif getattr(entity, ‘megagroup’, False):
entity_type = “\u8d85\u7ea7\u7fa4”
else:
entity_type = “\u9891\u9053(\u672a\u77e5)”
elif dialog.is_group:
entity_type = “\u666e\u901a\u7fa4”
# \u8fc7\u6ee4\u6389\u79c1\u804a (\u5982\u679c\u9700\u8981\u79c1\u804aID\uff0c\u6ce8\u91ca\u6389\u4e0b\u9762\u8fd9\u884c)
if entity_type != “\u79c1\u804a”:
# \u83b7\u53d6\u4e24\u79cd\u683c\u5f0f\u7684 ID
raw_id = entity.id
bot_api_id = get_bot_api_id(entity, entity_type)
chat_info = {
“\u7c7b\u578b”: entity_type,
“\u540d\u79f0”: dialog.name,
“Bot_API_ID”: bot_api_id, # \u2705 \u65b0\u589e\uff1a\u53ef\u4ee5\u76f4\u63a5\u7ed9 Bot \u7528\u7684 ID
“\u539f\u59cb_ID”: raw_id, # Telethon \u7528\u7684\u539f\u59cb ID
“\u7528\u6237\u540d”: getattr(entity, ‘username’, ‘\u65e0’) or ‘\u65e0’,
“\u6210\u5458\u6570”: getattr(entity, ‘participants_count’, ‘\u672a\u77e5’)
}
chat_data_list.append(chat_info)
print(f”[{entity_type}] {dialog.name} | Bot_ID: {bot_api_id}”)
# === \u5bfc\u51fa\u5230 CSV \u6587\u4ef6 ===
if chat_data_list:
# — \u6587\u4ef6\u5939\u5904\u7406\u903b\u8f91 —
folder_name = “namelist”
if not os.path.exists(folder_name):
os.makedirs(folder_name)
print(f”\ud83d\udcc1 \u5df2\u81ea\u52a8\u521b\u5efa\u6587\u4ef6\u5939: {folder_name}”)
timestamp = datetime.datetime.now().strftime(“%Y%m%d_%H%M%S”)
filename = f”telegram_chats_{timestamp}.csv”
# \u4f7f\u7528 os.path.join \u786e\u4fdd\u8de8\u5e73\u53f0\u517c\u5bb9\u6027
file_path = os.path.join(folder_name, filename)
# \u66f4\u65b0\u8868\u5934
headers = [“\u7c7b\u578b”, “\u540d\u79f0”, “Bot_API_ID”, “\u539f\u59cb_ID”, “\u7528\u6237\u540d”, “\u6210\u5458\u6570”]
try:
with open(file_path, 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”\u2705 \u6210\u529f\uff01\u5171\u5bfc\u51fa {len(chat_data_list)} \u4e2a\u7fa4\u7ec4\/\u9891\u9053\u3002”)
print(f”\ud83d\udcc1 \u6587\u4ef6\u5df2\u4fdd\u5b58\u5230: {os.path.abspath(file_path)}”)
except Exception as e:
print(f”\u274c \u5bfc\u51fa\u6587\u4ef6\u5931\u8d25: {e}”)
else:
print(“\u26a0\ufe0f \u672a\u627e\u5230\u4efb\u4f55\u7fa4\u7ec4\u6216\u9891\u9053\u3002”)
async def main():
await client.start(phone=PHONE_NUMBER, password=TWO_STEP_PASSWORD)
print(“\u2705 \u767b\u5f55\u6210\u529f”)
await list_and_export_chats()
if __name__ == “__main__”:
with client:
client.loop.run_until_complete(main())<\/p>\n","protected":false},"excerpt":{"rendered":"