import asyncio import csv import random import re from urllib.parse import unquote from loguru import logger from telethon import TelegramClient, functions from telethon.errors import NewSaltInvalidError, NewSettingsInvalidError, PasswordHashInvalidError, FloodWaitError, \ PeerFloodError, UserPrivacyRestrictedError, ChatAdminRequiredError, UserAlreadyParticipantError, UserKickedError, \ ChannelsTooMuchError, UsersTooMuchError, ChatWriteForbiddenError, BadRequestError, RPCError from telethon.tl import types from telethon.tl.functions.account import GetAuthorizationsRequest, ResetAuthorizationRequest from telethon.tl.functions.channels import JoinChannelRequest, InviteToChannelRequest, GetFullChannelRequest from telethon.tl.functions.messages import RequestAppWebViewRequest, StartBotRequest, ExportChatInviteRequest, \ AddChatUserRequest, ImportChatInviteRequest from telethon.tl.types import InputBotAppShortName, ChannelParticipantsSearch, Chat, Channel, User from models.tg_phone_devices import TgPhoneDevices async def create_telegram_client(server): """ 创建并配置 Telegram 客户端 """ proxy = { 'proxy_type': server.proxy_type, # 或 'socks4'，具体看你的代理类型 'addr': "192.168.1.79", # 代理服务器地址 'port': int(server.port), # 代理服务器端口 'username': server.user if server.user else "", # 如果有用户名，填写 'password': server.pwd if server.pwd else "" # 如果有密码，填写 } client = TelegramClient( fr"C:\Users\27942\Desktop\session\sessions\{server.phone}", api_id=server.api_id, api_hash=server.api_hash, device_model=server.device_model, system_version=server.system_version, app_version=server.app_version, system_lang_code=server.system_lang_code, lang_code=server.lang_code, proxy=proxy ) return client async def main1111(phone=None): server, server_type = TgPhoneDevices().get_or_create( phone=phone, ) client = await create_telegram_client(server) await client.connect() # await client.log_out() # if await client.is_user_authorized(): # me = await client.get_me() # logger.info(f'账号 {8617882004254} -- {me.username} 登录成功！') # # print(me.first_name) # print(me.last_name) # else: # logger.error(f'账号 {8617882004254} 登录失败！') # =========================================================================================================== # await client.send_message('dockerse', 'hrhrerhrhrhrhml2') # -------------------------------------------------------------------------------- # # 查询所有聊天 # bot_username = "pengu_clash_bot" # async for dialog in client.iter_dialogs(): # # print(dialog) # try: # for i in dialog.entity.usernames: # if i.username == bot_username: # logger.info(f'已与机器人交互: @{bot_username}') # return True # except: # pass # if dialog.is_user and dialog.entity.username == bot_username: # logger.info(f'已与机器人交互: @{bot_username}') # return True # logger.info(f'尚未与机器人交互: @{bot_username}') # bot_username = "BlumCryptoBot" # # # 获取目标 Bot 的实体信息 # bot_entity = await client.get_input_entity(bot_username) # print(bot_entity) # # 获取用户的所有对话列表 # dialogs = await client.get_dialogs() # # # 遍历对话列表，检查是否存在该 Bot 的私聊 # for dialog in dialogs: # print(dialog) # if dialog.entity.id == bot_entity.id: # print(1) # ------------------------------------------------------------------------------ # ###点击 confirm # # 获取最近5条消息（避免漏掉） # messages = await client.get_messages(777000, limit=5) # if not messages: # print("未找到验证消息") # return # # # 遍历消息，寻找包含按钮的验证消息 # for message in messages: # if message.buttons: # for row in message.buttons: # for button in row: # if 'confirm' in button.text.lower(): # try: # await button.click() # print(f"已点击消息 {message.id} 的 Confirm 按钮") # except: # pass # # --------------------------------------------------------- # # 获取所有登录设别信息 # # 发送请求以获取授权信息 # authorizations = await client(GetAuthorizationsRequest()) # # # 打印每个授权会话的信息 # for authorization in authorizations.authorizations: # print(f"授权ID: {authorization.hash}") # print(f"设备模型: {authorization.device_model}") # print(f"平台: {authorization.platform}") # print(f"系统版本: {authorization.system_version}") # print(f"应用名称: {authorization.app_name}") # print(f"应用版本: {authorization.app_version}") # print(f"IP 地址: {authorization.ip}") # print(f"国家: {authorization.country}") # print(f"创建时间: {authorization.date_created}") # print(f"最后活跃时间: {authorization.date_active}") # print(f"是否官方应用: {'是' if authorization.official_app else '否'}") # print(f"是否当前设备: {'是' if authorization.current else '否'}") # print('-' * 40) # ---------------------------------------------------------------------------------------- # # 踢出指定设备名的设备 # # # 发送请求以获取授权信息 # authorizations = await client(GetAuthorizationsRequest()) # # # 设定你要踢出的设备名 # target_device_model = 'Microsoft Edge 121' # # # 遍历所有授权会话 # for authorization in authorizations.authorizations: # if authorization.device_model == target_device_model and not authorization.current: # # 踢出指定设备 # await client(ResetAuthorizationRequest(hash=authorization.hash)) # print(f"已踢出设备: {authorization.device_model} (授权ID: {authorization.hash})") # else: # print(f"未找到或跳过当前设备: {authorization.device_model} (授权ID: {authorization.hash})") # # --------------------------------------------------------------------------------------------- # try: # await client.edit_2fa(current_password="Haxi@123456@", new_password="123456qwe") # except NewSettingsInvalidError: # print(1) # except PasswordHashInvalidError: # print(2) # --------------------------------------------------------------------------------------------------- # 使用邀请码启动bot # 使用邀请码启动bot # bot_name = "BlumCryptoBot" # invite_code = "ref_H21tvNwu3n" # # result = await client(StartBotRequest( # bot=bot_name, # peer=bot_name, # start_param=invite_code # )) # # # logger.info(result) # ------------------------------------------------------------------------------------------------- # 获取bot实体 # 使用start_params请求WebApp # bot_name = "BlumCryptoBot" # invite_code = "ref_H21tvNwu3n" # # await client.get_input_entity(bot_name) # # app_info = await client(RequestAppWebViewRequest( # 'me', # InputBotAppShortName(await client.get_input_entity(bot_name), "app"), # platform="web", # start_param=invite_code if invite_code else None # )) # # tgData = unquote(app_info.url.split('tgWebAppData=')[1].split('&tgWebAppVersion')[0]) # print(tgData) # ------------------------------------------------------------------------------------------------ # a = await client(JoinChannelRequest('webseaml2')) # print(a) # messages = await client.get_messages('webseaml2', 100) # for line in messages: # print(line) # # ✅ 用链接获取群实体 # group = await client.get_entity('https://t.me/webseaml2') # # # ✅ 获取目标用户实体（可以是用户名或 user_id） # user = await client.get_entity('dockerse') # # # ✅ 邀请（必须是群管理员） # await client(InviteToChannelRequest( # channel=group, # users=[user] # )) # user = await client.get_entity('dockerse') # print(user.id) # 打印用户 ID # messages = await client.get_messages(bot, limit=1) # # print(messages[0].Message.data.message) # # # if not messages: # # print("未找到验证消息") # # return # # # # 遍历消息，寻找包含按钮的验证消息 # for message in messages: # print(message.message) # # if message.buttons: # # for row in message.buttons: # # for button in row: # # if 'now' in button.text.lower(): # # try: # # await button.click() # # print(f"已点击消息 {message.id} 的 Confirm 按钮") # # except: # # pass # ============================================================================= # 拉人进群 # target_group = await client.get_entity("webseaml2") # users_to_add = [ # await client.get_entity("pantakill"), # # 可放更多用户 # ] # await client(InviteToChannelRequest(target_group, users_to_add)) # # 2️⃣ 发送消息 # update = await client(ImportChatInviteRequest("+E6E4i3svka81MGY1")) # chat = update.chats[0] # try: # await client.send_message(chat.id, "你好，这是一条来自 Telethon 的测试消息！") # print("消息已发送！") # except Exception as e: # print(f"发送失败: {e}") # #+++++++++++++++++++++++++++++++++++++++++++++++++++++++++ # print("正在获取对话列表...") # # 获取所有对话 # dialogs = await client.get_dialogs() # # # 退出所有群聊 # for dialog in dialogs: # if dialog.is_group or dialog.is_channel: # try: # await client.delete_dialog(dialog.entity) # print(f"已退出群聊/频道: {dialog.title}") # except Exception as e: # print(f"退出群聊/频道 {dialog.title} 时出错: {e}") # # # 取消关注所有机器人 # for dialog in dialogs: # if hasattr(dialog.entity, 'bot') and dialog.entity.bot: # try: # await client.delete_dialog(dialog.entity) # print(f"已取消关注机器人: {dialog.title}") # except Exception as e: # print(f"取消关注机器人 {dialog.title} 时出错: {e}") # ============================================================================================ # 获取验证码 messages = await client.get_messages(777000, 1) message_text = messages[0].text try: code = re.findall(r'\d{6}', message_text)[0] logger.info(f"当前验证码：{code}") await client.disconnect() return code except: logger.error("获取验证码失败！！！") # if __name__ == '__main__': # asyncio.run(main()) # .\xray.exe -c .\1_3078_client_liu_http_ip_1v1.json