bupabupala
commented
1 week ago `import re from itertools import cycle
from pyrogram import Client
from config import from config.config import from mhy.utils.logger import logger from better_proxy import Proxy
def parse_proxy(proxy_string):
正则表达式模式
socks5_pattern = re.compile(r'socks5://(?:(?P<user>[^:@]*):(?P<password>[^@]*)@)?(?P<host>[^:]+):(?P<port>\d+)')
http_pattern = re.compile(r'http(?:s)?://(?:(?P<user>[^:@]*):(?P<password>[^@]*)@)?(?P<host>[^:]+):(?P<port>\d+)')
# 尝试匹配Socks5代理
if proxy_string.startswith('socks5://'):
socks5_match = socks5_pattern.match(proxy_string)
if socks5_match:
data = socks5_match.groupdict()
return {
'type': 'socks5',
'host': data['host'],
'port': int(data['port']),
'user': data.get('user'),
'password': data.get('password')
}
# 尝试匹配HTTP或HTTPS代理
if proxy_string.startswith(('http://', 'https://')):
http_match = http_pattern.match(proxy_string)
if http_match:
data = http_match.groupdict()
return {
'type': 'http' if not proxy_string.startswith('https://') else 'https',
'host': data['host'],
'port': int(data['port']),
'user': data.get('user'),
'password': data.get('password')
}
# 如果没有匹配到任何代理类型，返回None
return Noneasync def register_sessions() -> None: if not API_ID or not API_HASH: raise ValueError("API_ID 和 API_HASH 未配置")
session_name = input('\n请输入设置当前session对应别名 (别名仅用于区分账号，输完按 Enter继续下一步): ')
if not session_name:
return None
if USE_PROXY_FROM_FILE:
with open(file='data/proxies.txt', encoding='utf-8-sig') as file:
proxies = [Proxy.from_str(proxy=row.strip()).as_url for row in file]
proxies_cycle = cycle(proxies) if proxies else None
# 在每次操作时，从循环迭代器中获取一个新的代理
proxy = next(proxies_cycle) if proxies_cycle else None
proxy_ = parse_proxy(proxy)
proxy_dict = {
'scheme': proxy_.get('type'),
'hostname': proxy_.get('host'),
'port': proxy_.get('port')
}
if proxy_.get('user') is not None:
proxy_dict['username'] = proxy_.get('user')
if proxy_.get('password') is not None:
proxy_dict['password'] = proxy_.get('password')
session = Client(
name=session_name,
api_id=API_ID,
api_hash=API_HASH,
workdir="sessions/",
proxy=proxy_dict
)
else:
session = Client(
name=session_name,
api_id=API_ID,
api_hash=API_HASH,
workdir="sessions/",
)
async with session:
user_data = await session.get_me()
logger.success(f'账号登录信息本地保存成功{user_data.username} | {user_data.first_name} {user_data.last_name}')`
gitbcat
Информация в шапке, сейчас берется ip с устройства которого вы добавляете эти самые аккаунты. Хотелось бы поддержка прокси в этот момент.