tg-cli

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Telegram 名字更新器 - 管理命令行工具
用于添加、删除、查看账号
"""
import os
import sys
import json
import asyncio
import argparse
from datetime import datetime
from pathlib import Path

from telethon import TelegramClient
from telethon.errors import SessionPasswordNeededError, PhoneNumberInvalidError

# 全局变量
SESSIONS_DIR = Path('/app/data/sessions')
SESSIONS_DIR.mkdir(parents=True, exist_ok=True)
CONFIG_FILE = SESSIONS_DIR / 'config.json'


class ConfigManager:
    """配置管理器"""
    
    def __init__(self):
        self.config_file = CONFIG_FILE
        self.config = self._load_config()
    
    def _load_config(self) -> dict:
        """加载配置"""
        if self.config_file.exists():
            try:
                with open(self.config_file, 'r', encoding='utf-8') as f:
                    return json.load(f)
            except Exception as e:
                print(f"❌ 加载配置失败: {e}")
        return {'accounts': {}}
    
    def _save_config(self):
        """保存配置"""
        try:
            with open(self.config_file, 'w', encoding='utf-8') as f:
                json.dump(self.config, f, ensure_ascii=False, indent=2)
        except Exception as e:
            print(f"❌ 保存配置失败: {e}")
    
    def add_account(self, phone: str, api_id: int, api_hash: str, name: str = "", username: str = ""):
        """添加账号"""
        self.config['accounts'][phone] = {
            'phone': phone,
            'api_id': api_id,
            'api_hash': api_hash,
            'name': name,
            'username': username,
            'added_at': datetime.now().isoformat()
        }
        self._save_config()
    
    def remove_account(self, phone: str):
        """删除账号"""
        if phone in self.config['accounts']:
            del self.config['accounts'][phone]
            self._save_config()
            # 删除 session 文件
            session_file = SESSIONS_DIR / f"{phone}.session"
            if session_file.exists():
                session_file.unlink()
    
    def get_account(self, phone: str):
        """获取账号信息"""
        return self.config['accounts'].get(phone)
    
    def get_all_accounts(self) -> dict:
        """获取所有账号"""
        return self.config['accounts']


config_manager = ConfigManager()


async def login_account():
    """登录账号"""
    print("\n" + "="*60)
    print("🔐 添加新账号")
    print("="*60)
    
    # 步骤1: 获取 API 凭证
    print("\n📋 步骤 1/3: API 配置")
    print("💡 访问 https://my.telegram.org/apps 获取 API 凭证")
    
    try:
        api_id = input("请输入 API ID: ").strip()
        api_hash = input("请输入 API Hash: ").strip()
        phone = input("请输入手机号 (格式: +86xxxxxxxxxx): ").strip()
        
        if not api_id or not api_hash or not phone:
            print("❌ 所有字段都是必填的")
            return
        
        api_id = int(api_id)
        
        # 创建客户端
        session_path = str(SESSIONS_DIR / phone)
        client = TelegramClient(session_path, api_id, api_hash)
        
        await client.connect()
        
        # 步骤2: 发送验证码
        print("\n📱 步骤 2/3: 验证码验证")
        print("⏳ 正在发送验证码...")
        
        await client.send_code_request(phone)
        print("✅ 验证码已发送到你的 Telegram 应用")
        
        code = input("请输入验证码: ").strip()
        
        if not code:
            print("❌ 验证码不能为空")
            await client.disconnect()
            return
        
        try:
            # 尝试登录
            await client.sign_in(phone, code)
            
        except SessionPasswordNeededError:
            # 需要两步验证
            print("\n🔒 步骤 3/3: 两步验证")
            password = input("请输入两步验证密码 (如未设置请直接回车): ").strip()
            
            if password:
                await client.sign_in(password=password)
            else:
                print("❌ 需要两步验证密码")
                await client.disconnect()
                return
        
        # 登录成功，获取用户信息
        me = await client.get_me()
        name = me.first_name or "Unknown"
        username = me.username or "unknown"
        
        print(f"\n✅ 登录成功!")
        print(f"👤 用户: {name}")
        print(f"📱 用户名: @{username}")
        print(f"📞 手机号: {phone}")
        
        # 保存配置
        config_manager.add_account(phone, api_id, api_hash, name, username)
        
        # 断开连接（守护进程会自动连接）
        await client.disconnect()
        
        print("\n🚀 账号已添加！")
        print("💡 守护进程会在几秒内自动启动此账号")
        print("💡 使用 'tg-cli list' 查看账号状态")
        
    except PhoneNumberInvalidError:
        print("❌ 手机号格式错误")
    except Exception as e:
        print(f"❌ 登录失败: {e}")


def list_accounts():
    """列出所有账号"""
    accounts = config_manager.get_all_accounts()
    
    if not accounts:
        print("\n📭 暂无账号")
        print("💡 使用 'tg-cli add' 添加账号")
        return
    
    print("\n" + "="*70)
    print(f"📋 账号列表 (共 {len(accounts)} 个)")
    print("="*70)
    
    for i, (phone, data) in enumerate(accounts.items(), 1):
        name = data.get('name', 'Unknown')
        username = data.get('username', 'unknown')
        added_at = data.get('added_at', '')
        
        print(f"\n{i}. {name}")
        print(f"   📱 用户名: @{username}")
        print(f"   📞 手机号: {phone}")
        print(f"   📅 添加时间: {added_at[:10] if added_at else '未知'}")
    
    print("\n" + "="*70)
    print("💡 所有账号由守护进程自动管理")
    print("💡 使用 'docker logs -f tgtime' 查看运行日志")


async def delete_account(phone: str):
    """删除账号"""
    account = config_manager.get_account(phone)
    if not account:
        print(f"❌ 账号不存在: {phone}")
        return
    
    name = account.get('name', 'Unknown')
    print(f"\n⚠️  即将删除账号: {name} ({phone})")
    confirm = input("确定要删除吗? (yes/no): ").strip().lower()
    
    if confirm not in ['yes', 'y']:
        print("❌ 已取消")
        return
    
    # 删除配置
    config_manager.remove_account(phone)
    
    print(f"✅ 已删除账号: {phone}")
    print("💡 守护进程会在几秒内自动停止此账号")


def show_status():
    """显示守护进程状态"""
    pid_file = Path('/app/data/daemon.pid')
    log_file = Path('/app/data/daemon.log')
    
    print("\n" + "="*60)
    print("📊 系统状态")
    print("="*60)
    
    # 检查守护进程
    if pid_file.exists():
        with open(pid_file, 'r') as f:
            pid = f.read().strip()
        print(f"✅ 守护进程运行中 (PID: {pid})")
    else:
        print("❌ 守护进程未运行")
    
    # 账号统计
    accounts = config_manager.get_all_accounts()
    print(f"📱 已配置账号: {len(accounts)} 个")
    
    # 显示最近日志
    if log_file.exists():
        print("\n📝 最近日志 (最后10行):")
        print("-" * 60)
        try:
            with open(log_file, 'r') as f:
                lines = f.readlines()
                for line in lines[-10:]:
                    print(line.rstrip())
        except Exception as e:
            print(f"读取日志失败: {e}")
    
    print("="*60)
    print("💡 使用 'docker logs -f tgtime' 查看实时日志")


def print_help():
    """显示帮助信息"""
    help_text = """
🤖 Telegram 名字自动更新 - 管理工具

📖 使用方法:
    tg-cli add           添加新账号（登录）
    tg-cli list          查看所有账号
    tg-cli delete PHONE  删除指定账号
    tg-cli status        查看系统状态
    tg-cli help          显示此帮助信息

📝 示例:
    tg-cli add                    # 交互式添加账号
    tg-cli list                   # 查看账号列表
    tg-cli delete +8613800138000  # 删除指定账号
    tg-cli status                 # 查看运行状态

💡 提示:
    - 添加账号后会自动开始更新名字
    - 守护进程在后台持续运行
    - 容器重启后自动恢复所有账号
    - 使用 'docker logs -f tgtime' 查看实时日志
"""
    print(help_text)


async def main():
    """主函数"""
    parser = argparse.ArgumentParser(
        description='Telegram 名字自动更新管理工具',
        add_help=False
    )
    parser.add_argument('command', nargs='?', help='命令: add/list/delete/status/help')
    parser.add_argument('args', nargs='*', help='命令参数')
    
    args = parser.parse_args()
    
    if not args.command or args.command == 'help':
        print_help()
    
    elif args.command == 'add':
        await login_account()
    
    elif args.command == 'list':
        list_accounts()
    
    elif args.command == 'delete':
        if not args.args:
            print("❌ 请指定要删除的手机号")
            print("💡 使用方法: tg-cli delete +8613800138000")
        else:
            await delete_account(args.args[0])
    
    elif args.command == 'status':
        show_status()
    
    else:
        print(f"❌ 未知命令: {args.command}")
        print("💡 使用 'tg-cli help' 查看帮助")


if __name__ == '__main__':
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n\n⚠️  已取消")
    except Exception as e:
        print(f"❌ 错误: {e}")
        sys.exit(1)