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)