import sys
import subprocess
import os
import csv
import datetime

==================== 0. 面板专属：环境依赖自动安装器 ====================

lib_path = os.path.join(os.path.dirname(os.path.abspath(file)), ‘libs’)
sys.path.insert(0, lib_path)

try:
import telethon
import dotenv
except ImportError:
print(“>>> 首次运行：检测到缺少依赖，正在局部目录(libs)自动安装，请稍候…”)
subprocess.check_call([
sys.executable, “-m”, “pip”, “install”, “-t”, lib_path,
“telethon”, “python-dotenv”
])
print(“>>> 依赖安装完成！准备启动程序…”)

确保安装后可以正常导入

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”)
sys.exit(1)

提前创建 session 文件夹，防止面板系统因无权限报错

if not os.path.exists(“session”):
os.makedirs(“session”)

client = TelegramClient(‘session/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,
            "原始_ID": raw_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:
    # --- 文件夹处理逻辑 ---
    folder_name = "namelist"
    if not os.path.exists(folder_name):
        os.makedirs(folder_name)
        print(f"📁 已自动创建文件夹: {folder_name}")

    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    filename = f"telegram_chats_{timestamp}.csv"
    
    # 使用 os.path.join 确保跨平台兼容性
    file_path = os.path.join(folder_name, filename)
    
    # 更新表头
    headers = ["类型", "名称", "Bot_API_ID", "原始_ID", "用户名", "成员数"]
    
    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"✅ 成功！共导出 {len(chat_data_list)} 个群组/频道。")
        print(f"📁 文件已保存到: {os.path.abspath(file_path)}")
        
    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())