205 lines
6.8 KiB
Python
205 lines
6.8 KiB
Python
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)
|