ddrwode/mini_code
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ggrg

Browse Source
This commit is contained in:
27942
2025-12-01 17:22:53 +08:00
parent 33d1c2cae5
commit 1d3e26f519
13 changed files with 1243 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

312
tg/text.py Normal file
View File

@@ -0,0 +1,312 @@
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

188
tg/查询所有号存活状态.py Normal file
View File

@@ -0,0 +1,188 @@
import asyncio
import os.path
import random
import threading
import time
import traceback
from urllib.parse import unquote
from loguru import logger
from telethon import TelegramClient
from telethon.errors import PhoneNumberBannedError
from telethon.tl.functions.account import GetAuthorizationsRequest
from telethon.tl.functions.messages import RequestAppWebViewRequest
from telethon.tl.types import InputBotAppShortName
from models.tg_phone_devices import TgPhoneDevices
async def create_telegram_client(server_type):
"""
创建并配置 Telegram 客户端
"""
proxy = {
'proxy_type': server_type.proxy_type, # 或 'socks4'，具体看你的代理类型
# 'addr': server_type.addr, # 代理服务器地址
'addr': "192.168.1.79", # 代理服务器地址
'port': int(server_type.port), # 代理服务器端口
'username': server_type.user if server_type.user else "", # 如果有用户名，填写
'password': server_type.pwd if server_type.pwd else "" # 如果有密码，填写
}
try:
client = TelegramClient(
fr"C:\Users\27942\Desktop\session\sessions\{server_type.phone}",
api_id=server_type.api_id,
api_hash=server_type.api_hash,
system_lang_code=server_type.system_lang_code,
lang_code=server_type.lang_code,
device_model=server_type.device_model,
system_version=server_type.system_version,
app_version=server_type.app_version,
proxy=proxy,
auto_reconnect=False,
)
return client
except Exception as e:
print(e)
traceback.print_stack()
logger.error(f"当前电话号码：{server_type.phone}，有问题！！！")
async def login_and_validate(server):
client = await create_telegram_client(server)
try:
try:
await client.connect()
except Exception as e:
server.is_valid_session = 4
server.save(only=[TgPhoneDevices.is_valid_session])
return
if await client.is_user_authorized():
me = await client.get_me()
logger.info(f'账号 {me.phone} -- {me.first_name} 登录成功！！！')
server.is_valid_session = 1
time.sleep(random.randint(3, 8))
# 获取设备数
authorizations = await client(GetAuthorizationsRequest())
server.device_start = len(authorizations.authorizations)
time.sleep(random.randint(3, 8))
# 判断是否被冻结了
bot_username = "BlumCryptoBot"
start = 0
async for dialog in client.iter_dialogs():
try:
for i in dialog.entity.usernames:
if i.username == bot_username:
logger.info(f'账号 {server.phone}，已与机器人交互: @{bot_username}')
start = 1
break
if start:
break
except:
pass
if dialog.is_user and dialog.entity.username == bot_username:
logger.info(f'账号 {server.phone}，已与机器人交互: @{bot_username}')
start = 1
break
try:
if not start:
logger.info(f'账号 {server.phone}，尚未与机器人交互: @{bot_username}')
bot = await client.get_entity(bot_username) # 获取bot实体
await client.send_message(bot, '/start')
invite_code = "ref_H21tvNwu3n"
await client.get_input_entity(bot_username)
app_info = await client(RequestAppWebViewRequest(
'me',
InputBotAppShortName(await client.get_input_entity(bot_username), "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)
except Exception as e:
print(e)
server.is_valid_session = 3
else:
logger.error(f'账号 {server.phone} 授权失败！！！')
server.is_valid_session = 0
except PhoneNumberBannedError:
logger.error(f"账号 {server.phone}，账号以死！！！")
server.is_valid_session = -1
server.save(only=[TgPhoneDevices.is_valid_session, TgPhoneDevices.device_start])
logger.success('数据已更新')
client.disconnect()
async def start_task(server):
try:
status = check_exist_session_file(server.phone)
if not status:
server.is_valid_session = -1
server.save(only=[TgPhoneDevices.is_valid_session])
return
await login_and_validate(server)
except Exception as e:
a = f"1{e}"
if a == "1Connection to Telegram failed 5 time(s)":
server.is_valid_session = 4
server.save(only=[TgPhoneDevices.is_valid_session])
logger.error(f"当前电话号码：{server.phone}，有问题！！！")
def check_exist_session_file(phone):
path = fr"C:\Users\27942\Desktop\session\sessions\{phone}.session"
is_exist_session_file = os.path.isfile(path)
if not is_exist_session_file:
logger.error(f'未找到session文件：{path}')
return False
return True
def main1(server):
asyncio.run(start_task(server=server))
def main():
devices = TgPhoneDevices.select().where(
TgPhoneDevices.is_valid_session.is_null(False),
# TgPhoneDevices.phone.is_null(False)
# TgPhoneDevices.device_start.is_null()
# TgPhoneDevices.kick_status.is_null()
)
logger.info(f'本轮需要检查：{len(devices)}')
for _, server in enumerate(devices):
logger.info(f'正在检测，第{_}，账号：{server.phone}')
# main1(server=server)
threading.Thread(target=main1, args=(server,)).start()
time.sleep(0.3)
if __name__ == '__main__':
main()

189
tg/查询所有号存活状态1.py Normal file
View File

@@ -0,0 +1,189 @@
import asyncio
import os.path
import random
import threading
import time
import traceback
from urllib.parse import unquote
from loguru import logger
from telethon import TelegramClient
from telethon.errors import PhoneNumberBannedError
from telethon.tl.functions.account import GetAuthorizationsRequest
from telethon.tl.functions.messages import RequestAppWebViewRequest
from telethon.tl.types import InputBotAppShortName
from models.tg_phone_devices import TgPhoneDevices1
async def create_telegram_client(server_type):
"""
创建并配置 Telegram 客户端
"""
proxy = {
'proxy_type': server_type.proxy_type, # 或 'socks4'，具体看你的代理类型
# 'addr': server_type.addr, # 代理服务器地址
'addr': "192.168.1.79", # 代理服务器地址
'port': int(server_type.port), # 代理服务器端口
'username': server_type.user if server_type.user else "", # 如果有用户名，填写
'password': server_type.pwd if server_type.pwd else "" # 如果有密码，填写
}
try:
client = TelegramClient(
fr"C:\sessions\{server_type.phone}",
api_id=server_type.api_id,
api_hash=server_type.api_hash,
system_lang_code=server_type.system_lang_code,
lang_code=server_type.lang_code,
device_model=server_type.device_model,
system_version=server_type.system_version,
app_version=server_type.app_version,
proxy=proxy,
# auto_reconnect=False,
)
return client
except Exception as e:
print(e)
traceback.print_stack()
logger.error(f"当前电话号码：{server_type.phone}，有问题！！！")
async def login_and_validate(server):
client = await create_telegram_client(server)
try:
try:
await client.connect()
except Exception as e:
server.is_valid_session = 4
server.save(only=[TgPhoneDevices1.is_valid_session])
return
if await client.is_user_authorized():
me = await client.get_me()
logger.info(f'账号 {me.phone} -- {me.first_name} 登录成功！！！')
server.is_valid_session = 1
time.sleep(random.randint(3, 8))
# 获取设备数
authorizations = await client(GetAuthorizationsRequest())
server.device_start = len(authorizations.authorizations)
time.sleep(random.randint(3, 8))
# 判断是否被冻结了
bot_username = "BlumCryptoBot"
start = 0
async for dialog in client.iter_dialogs():
try:
for i in dialog.entity.usernames:
if i.username == bot_username:
logger.info(f'账号 {server.phone}，已与机器人交互: @{bot_username}')
start = 1
break
if start:
break
except:
pass
if dialog.is_user and dialog.entity.username == bot_username:
logger.info(f'账号 {server.phone}，已与机器人交互: @{bot_username}')
start = 1
break
try:
if not start:
logger.info(f'账号 {server.phone}，尚未与机器人交互: @{bot_username}')
bot = await client.get_entity(bot_username) # 获取bot实体
await client.send_message(bot, '/start')
invite_code = "ref_H21tvNwu3n"
await client.get_input_entity(bot_username)
app_info = await client(RequestAppWebViewRequest(
'me',
InputBotAppShortName(await client.get_input_entity(bot_username), "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)
except Exception as e:
print(e)
server.is_valid_session = 3
else:
logger.error(f'账号 {server.phone} 授权失败！！！')
server.is_valid_session = 0
except PhoneNumberBannedError:
logger.error(f"账号 {server.phone}，账号以死！！！")
server.is_valid_session = -1
server.save(only=[TgPhoneDevices1.is_valid_session, TgPhoneDevices1.device_start])
logger.success('数据已更新')
client.disconnect()
async def start_task(server):
try:
status = check_exist_session_file(server.phone)
if not status:
server.is_valid_session = -1
server.save(only=[TgPhoneDevices1.is_valid_session])
return
await login_and_validate(server)
except Exception as e:
a = f"1{e}"
if a == "1Connection to Telegram failed 5 time(s)":
server.is_valid_session = 4
server.save(only=[TgPhoneDevices1.is_valid_session])
logger.error(f"当前电话号码：{server.phone}，有问题！！！")
def check_exist_session_file(phone):
path = fr"C:\sessions\{phone}.session"
is_exist_session_file = os.path.isfile(path)
if not is_exist_session_file:
logger.error(f'未找到session文件：{path}')
return False
return True
def main1(server):
asyncio.run(start_task(server=server))
def main():
devices = TgPhoneDevices1.select().where(
TgPhoneDevices1.is_valid_session.is_null(False),
# TgPhoneDevices1.phone.is_null(False)
# TgPhoneDevices1.device_start.is_null()
# TgPhoneDevices1.kick_status.is_null()
)
logger.info(f'本轮需要检查：{len(devices)}')
for _, server in enumerate(devices):
logger.info(f'正在检测，第{_}，账号：{server.phone}')
# main1(server=server)
threading.Thread(target=main1, args=(server,)).start()
time.sleep(0.3)
if __name__ == '__main__':
main()

174
tg/登录账号.py Normal file
View File

@@ -0,0 +1,174 @@
import asyncio
import random
import threading
import time
from loguru import logger
from opentele.api import API
from telethon import TelegramClient
from telethon.errors import SessionPasswordNeededError, PhoneCodeInvalidError, RPCError
from models.tg_phone_devices import TgPhoneDevices, TgPhoneDevices1
from tg.text import main1111
async def get_device_info():
"""
获取设备信息
"""
NewApi = API.TelegramDesktop.Generate(system="windows")
return {
"api_id": NewApi.api_id,
"api_hash": NewApi.api_hash,
"device_model": NewApi.device_model,
'system_version': 'Windows 10',
"app_version": "6.3.4 x64",
"lang_code": NewApi.lang_code,
"system_lang_code": NewApi.system_lang_code
}
async def create_telegram_client(api_id, api_hash, phone_num, server_type1, 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:\sessions\{phone_num}",
api_id=api_id,
api_hash=api_hash,
device_model=server_type1.device_model,
system_version=server_type1.system_version,
app_version=server_type1.app_version,
system_lang_code=server_type1.system_lang_code,
lang_code=server_type1.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("请输入验证码：")
time.sleep(3)
code = await main1111(phone=phone_num)
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="Haxi@123456@")
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(device):
# 读取设备信息和 API 参数
device_info = await get_device_info()
api_id = device_info["api_id"]
api_hash = device_info["api_hash"]
# area_code = input("请输入区号：")
# phone_number = input("请输入电话号码：")
area_code = device.area_code
phone_number = device.phone_number
server, server_type = TgPhoneDevices().get_or_create(
phone_number=phone_number,
)
server1, server_type = TgPhoneDevices1().get_or_create(
phone_number=phone_number,
)
server1.area_code = area_code
server1.api_id = api_id
server1.api_hash = api_hash
# 这里我加了一行代码 by rachel 成功默认为1
server1.is_valid_session = '1'
server1.device_model = device_info["device_model"]
server1.system_version = device_info["system_version"]
server1.app_version = device_info["app_version"]
server1.system_lang_code = device_info["system_lang_code"]
server1.lang_code = device_info["lang_code"]
server1.phone = area_code + phone_number
server1.proxy_type = server.proxy_type
server1.addr = server.addr
server1.port = server.port
server1.save()
# 创建 Telegram 客户端
phone = area_code + phone_number
client = await create_telegram_client(api_id, api_hash, phone, server1, server)
await client.connect()
if await client.is_user_authorized():
me = await client.get_me()
logger.info(f'账号 {phone_number} -- {me.first_name} 登录成功！')
else:
# 登录 Telegram
if await login_telegram(client, phone):
logger.info("Telegram 客户端操作完成")
server1.is_valid_session = 1
server1.save()
else:
logger.error("Telegram 登录失败")
server1.is_valid_session = 0
server1.save()
def run_main(device):
asyncio.run(main(device))
if __name__ == '__main__':
# 筛选出 is_valid_session 为 1 的记录
valid_devices = TgPhoneDevices.select().where(TgPhoneDevices.is_valid_session == 1)
# 将查询结果转换为列表
valid_devices_list = list(valid_devices)
# 随机选择 100 条记录，如果记录总数不足 100 条，则选择全部记录
random_devices = random.sample(valid_devices_list, min(1524, len(valid_devices_list)))
# 输出结果
for device in random_devices:
threading.Thread(target=run_main, args=(device,)).start()
time.sleep(0.5)

125
tg/获取验证码.py Normal file
View File

@@ -0,0 +1,125 @@
import asyncio
import re
from loguru import logger
from opentele.api import API
from telethon.errors import SessionPasswordNeededError, PhoneCodeInvalidError, RPCError
from telethon import TelegramClient
from models.tg_phone_devices import TgPhoneDevices
async def get_device_info():
"""
获取设备信息
"""
PROXY_CONFIG = {
'proxy_type': 'http', # 或 'socks4'，具体看你的代理类型
'addr': '192.168.50.220', # 代理服务器地址
'port': 10809, # 代理服务器端口
'username': '', # 如果有用户名，填写
'password': '' # 如果有密码，填写
}
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():
phone_num = input("请输入电话号码：")
server = TgPhoneDevices().get_or_none(
TgPhoneDevices.phone == phone_num,
)
# 创建 Telegram 客户端
client = await create_telegram_client(server)
try:
await client.connect()
messages = await client.get_messages(777000, 1)
message_text = messages[0].text
print(message_text)
try:
code = re.findall(r'\d{6}', message_text)[0]
logger.info(f"当前验证码：{code}")
except:
logger.error("获取验证码失败！！！")
except:
logger.error("账号已死！！！")
finally:
await client.disconnect()
if __name__ == '__main__':
asyncio.run(main())