tg_code/踢设备.py

import asyncio
import threading
import time

from loguru import logger
from opentele.api import API
from telethon.errors import SessionPasswordNeededError, PhoneCodeInvalidError, RPCError
from telethon import TelegramClient, functions
from telethon.tl.functions.account import GetAuthorizationsRequest, ResetAuthorizationRequest
from telethon.tl.functions.auth import ResetAuthorizationsRequest

from models.device_info import DeviceInfo
from models.tg_models import TelegramAccount
from models.tg_phone_devices import TgPhoneDevices


async def get_device_info():
    """
    获取设备信息
    """
    NewApi = API.TelegramDesktop.Generate(system="macos")
    return {
        "api_id": NewApi.api_id,
        "api_hash": NewApi.api_hash,
        "device_model": NewApi.device_model,
        "system_version": NewApi.system_version,
        "app_version": NewApi.app_version,
        "lang_code": NewApi.lang_code,
        "system_lang_code": NewApi.system_lang_code
    }



async def create_telegram_client(server_type):
    """
    创建并配置 Telegram 客户端
    """

    # gw.dataimpulse.com:824:14fa7e7aa8280a914611__cr.gb:f1c1501b6b535c19
    # 定义代理
    proxy = {
        'proxy_type': server_type.proxy_type,  # 或 'socks4'，具体看你的代理类型
        'addr': server_type.addr,  # 代理服务器地址
        'port': int(server_type.port),  # 代理服务器端口
        'username': server_type.user if server_type.user else "",  # 如果有用户名，填写
        'password': server_type.pwd if server_type.pwd else ""  # 如果有密码，填写
    }

    client = TelegramClient(
        fr"C:\sessions\{server_type.phone}",
        api_id=server_type.api_id,
        api_hash=server_type.api_hash,
        device_model=server_type.device_model,
        system_version=server_type.system_version,
        app_version=server_type.app_version,
        system_lang_code=server_type.system_lang_code,
        lang_code=server_type.lang_code,
        proxy=proxy
    )
    return client


async def login_telegram(client, phone_num):
    """
    处理登录流程
    """
    try:

        sent_code_info = await client.send_code_request(phone_num)
        code = input("请输入验证码：")
        try:
            await client.sign_in(phone=phone_num, code=code)
        except PhoneCodeInvalidError:
            logger.error("验证码无效，请重试")
            return False
        except SessionPasswordNeededError:
            password = input("请输入密码：")
            await client.sign_in(password=password)

        me = await client.get_me()
        logger.info(f'账号 {phone_num} -- {me.first_name} 登录成功！')
        return True

    except (RPCError, Exception) as e:
        logger.error(f'登录失败，错误信息：{e}')
        return False

    finally:
        await client.disconnect()


# async def main(tg_info):
#     # 创建 Telegram 客户端
#
#     client = await create_telegram_client(tg_info)
#
#     try:
#         await client.connect()
#
#         if await client.is_user_authorized():
#             # 发送请求以重置所有授权，即踢出所有其他设备
#             result = await client(ResetAuthorizationsRequest())
#
#             if result:
#                 tg_info.kick_status = 1
#                 tg_info.save()
#
#                 print(f"{tg_info.phone}成功踢出了所有其他设备！")
#             else:
#                 print("尝试踢出其他设备失败。")
#
#                 tg_info.kick_status = 0
#                 tg_info.save()
#     except Exception as e:
#         print(e)
#         logger.error("操作失败，可能是原始2FA不对！！！")
#     finally:
#         await client.disconnect()

async def main(tg_info):
    # 创建 Telegram 客户端

    client = await create_telegram_client(tg_info)

    try:
        await client.connect()

        if await client.is_user_authorized():
            # 发送请求以获取授权信息
            authorizations = await client(GetAuthorizationsRequest())

            dev_info = TelegramAccount.get_or_none(
                TelegramAccount.is_logged_in_telegram == 1,
                TelegramAccount.ld_name == tg_info.phone
            )

            # if not dev_info:
            #     logger.error("没有查询到！！！")
            #     return

            authorizations = await client(GetAuthorizationsRequest())
            tg_info.device_start = len(authorizations.authorizations)

            if dev_info and dev_info.manufacturer:
                # 设定你要踢出的设备名
                target_device_model = dev_info.manufacturer

                # 遍历所有授权会话
                for authorization in authorizations.authorizations:
                    if not authorization.current:
                        if authorization.device_model.lower() != target_device_model.lower():
                            # 踢出指定设备
                            await client(ResetAuthorizationRequest(hash=authorization.hash))
                            logger.success(f"已踢出设备: {authorization.device_model} (授权ID: {authorization.hash})")

                    else:
                        logger.warning(
                            f"未找到或跳过当前设备: {authorization.device_model} (授权ID: {authorization.hash})")
            else:
                if tg_info.device_start > 1:
                    result = await client(ResetAuthorizationsRequest())
                    if result:
                        logger.success(f"{tg_info}，{tg_info.phone}成功踢出了所有其他设备！")

            tg_info.kick_status = 1
            authorizations = await client(GetAuthorizationsRequest())
            tg_info.device_start = len(authorizations.authorizations)
            tg_info.save()

            logger.success(f"{tg_info}，{tg_info.phone}数据更新！！！！")
        else:
            logger.error("尝试踢出其他设备失败。")

            tg_info.kick_status = 0
            tg_info.save()


    except Exception as e:
        print(e)
        logger.error(f"{tg_info.phone},操作失败，可能是原始2FA不对！！！")
    finally:
        await client.disconnect()


def run_main(tg_info1):
    asyncio.run(main(tg_info1))


if __name__ == '__main__':

    for tg_info in TgPhoneDevices().select().where(
            TgPhoneDevices.is_valid_session == 1,
            # TgPhoneDevices.device_start > 2,
            # TgPhoneDevices.kick_status.is_null(),
            # TgPhoneDevices.to_code.is_null(False),
            #
            # TgPhoneDevices.create_time > '2025-02-28 00:10:32',
            # TgPhoneDevices.create_time < '2025-02-28 23:00:00'
            # TgPhoneDevices.create_time > '2025-04-08 00:00:00',

    ):
        # print(tg_info.phone)
        threading.Thread(target=run_main, args=(tg_info,)).start()
        time.sleep(0.1)