优化代理获取和登录逻辑
- 每个账号独立获取代理，不再共用一个代理IP - 代理预检失败只标记该账号失败，不终止整个批次 - 代理验证改为并发：verify_proxies_concurrent 多IP同时验证 - resolve_working_proxy 多IP并发验证，找到可用即返回 - parse_proxy_response 改为返回列表，支持多行/逗号分隔 - 邮箱验证码匹配收件人地址，防止多账号并发取错验证码 - 修正 mail.bdhg.xyz 旧数据端口和SSL配置 - bdhg.xyz 默认配置改为 143端口非SSL Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
yml2213
co-authored by
Claude Fable 5
Claude Fable 5
parent
9f6b405571
commit
8fdf8d4b69
@@ -158,9 +158,10 @@ class EmailVerifier:
|
||||
|
||||
subject = self._decode_subject(msg.get('Subject', ''))
|
||||
from_addr = msg.get('From', '')
|
||||
to_addr = msg.get('To', '') or msg.get('Delivered-To', '') or ''
|
||||
body = self._get_email_body(msg)
|
||||
|
||||
if not self._is_douyu_email(subject, from_addr, body):
|
||||
if not self._is_douyu_email(subject, from_addr, body, to_addr=to_addr):
|
||||
continue
|
||||
|
||||
code = self._extract_verification_code(body)
|
||||
@@ -257,8 +258,20 @@ class EmailVerifier:
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
def _is_douyu_email(self, subject: str, from_addr: str, body: str = "") -> bool:
|
||||
"""判断是否是斗鱼的邮件"""
|
||||
def _is_douyu_email(self, subject: str, from_addr: str, body: str = "", to_addr: str = "") -> bool:
|
||||
"""判断是否是发给当前账号的斗鱼验证码邮件"""
|
||||
# 先检查收件人：多个账号共用同一个IMAP时，必须匹配收件人地址
|
||||
if to_addr and self.username:
|
||||
# to_addr 可能是 "name <email>" 格式，提取邮箱
|
||||
to_email = re.search(r'[\w.+-]+@[\w.-]+', to_addr)
|
||||
if to_email:
|
||||
to_lower = to_email.group(0).lower()
|
||||
my_lower = self.username.lower()
|
||||
# 精确匹配，或者 catch-all 前缀匹配
|
||||
if to_lower != my_lower and not my_lower.endswith('@' + to_lower.split('@')[-1]):
|
||||
logger.debug(f"跳过非当前账号邮件: 收件人={to_lower}, 当前={my_lower}")
|
||||
return False
|
||||
|
||||
douyu_keywords = ['斗鱼', 'douyu', 'douyutv']
|
||||
verify_keywords = ['验证码', '安全验证', '登录验证', '动态码', '校验码']
|
||||
combined = f"{subject}\n{from_addr}\n{body[:500]}".lower()
|
||||
@@ -354,4 +367,15 @@ def get_email_config_for_account(email_address: str) -> dict:
|
||||
}
|
||||
|
||||
domain = email_address.split('@')[-1].lower()
|
||||
return configs.get(domain, {'server': '111.229.206.54', 'port': 143, 'ssl': False})
|
||||
configs = {
|
||||
'qq.com': {'server': 'imap.qq.com', 'port': 993, 'ssl': True},
|
||||
'163.com': {'server': 'imap.163.com', 'port': 993, 'ssl': True},
|
||||
'126.com': {'server': 'imap.126.com', 'port': 993, 'ssl': True},
|
||||
'gmail.com': {'server': 'imap.gmail.com', 'port': 993, 'ssl': True},
|
||||
'outlook.com': {'server': 'outlook.office365.com', 'port': 993, 'ssl': True},
|
||||
'hotmail.com': {'server': 'outlook.office365.com', 'port': 993, 'ssl': True},
|
||||
'bdhg.xyz': {'server': 'mail.bdhg.xyz', 'port': 143, 'ssl': False},
|
||||
}
|
||||
|
||||
domain = email_address.split('@')[-1].lower()
|
||||
return configs.get(domain, {'server': domain, 'port': 143, 'ssl': False})
|
||||
|
||||
+51
-8
@@ -164,6 +164,51 @@ def verify_proxy_url(proxy_url: str, timeout: tuple = (4, 6)) -> tuple[bool, str
|
||||
return False, '代理验证失败（所有目标不可达）'
|
||||
|
||||
|
||||
def verify_proxies_concurrent(proxy_urls: list[str], timeout: tuple = (4, 6), max_workers: int = 5) -> tuple[Optional[str], str]:
|
||||
"""
|
||||
并发验证多个代理URL，返回第一个可用的。
|
||||
|
||||
Args:
|
||||
proxy_urls: 代理URL列表
|
||||
timeout: 验证超时
|
||||
max_workers: 最大并发数
|
||||
|
||||
Returns:
|
||||
(可用的代理URL, 消息)
|
||||
"""
|
||||
if not proxy_urls:
|
||||
return None, '无代理可验证'
|
||||
|
||||
if len(proxy_urls) == 1:
|
||||
ok, msg = verify_proxy_url(proxy_urls[0], timeout)
|
||||
if ok:
|
||||
return proxy_urls[0], msg
|
||||
return None, msg
|
||||
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
with ThreadPoolExecutor(max_workers=min(max_workers, len(proxy_urls))) as executor:
|
||||
future_map = {
|
||||
executor.submit(verify_proxy_url, p, timeout): p
|
||||
for p in proxy_urls
|
||||
}
|
||||
for future in as_completed(future_map):
|
||||
proxy_url = future_map[future]
|
||||
try:
|
||||
ok, msg = future.result()
|
||||
if ok:
|
||||
logger.success(f"并发验证找到可用代理: {proxy_url}")
|
||||
# 取消剩余任务
|
||||
for f in future_map:
|
||||
if f != future:
|
||||
f.cancel()
|
||||
return proxy_url, msg
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
return None, f'共 {len(proxy_urls)} 个代理均不可用'
|
||||
|
||||
|
||||
def resolve_working_proxy(
|
||||
api_url: str,
|
||||
whitelist_uid: str = "",
|
||||
@@ -210,14 +255,12 @@ def resolve_working_proxy(
|
||||
proxy_urls, whitelist_ip = parse_proxy_response(text)
|
||||
|
||||
if proxy_urls:
|
||||
log('info', f'获取到 {len(proxy_urls)} 个代理，逐一验证')
|
||||
for proxy_url in proxy_urls:
|
||||
ok, msg = verify_proxy_url(proxy_url)
|
||||
if ok:
|
||||
log('success', f'代理预检成功: {proxy_url}')
|
||||
return proxy_url, msg
|
||||
log('warning', f'代理 {proxy_url} 不可用: {msg}')
|
||||
last_error = f'共 {len(proxy_urls)} 个代理均不可用'
|
||||
log('info', f'获取到 {len(proxy_urls)} 个代理，并发验证')
|
||||
proxy_url, msg = verify_proxies_concurrent(proxy_urls)
|
||||
if proxy_url:
|
||||
log('success', f'代理预检成功: {proxy_url}')
|
||||
return proxy_url, msg
|
||||
last_error = msg
|
||||
log('warning', f'代理预检 {attempt}/{max_attempts}: {last_error}')
|
||||
continue
|
||||
|
||||
|
||||
@@ -64,6 +64,12 @@ def _migrate():
|
||||
"WHERE email_imap_server = '111.229.206.54'"
|
||||
))
|
||||
conn.commit()
|
||||
# 修正旧数据中使用 mail.bdhg.xyz 的账号：993端口不通，改用143非SSL
|
||||
conn.execute(text(
|
||||
"UPDATE accounts SET email_imap_port = 143, email_imap_ssl = 0 "
|
||||
"WHERE email_imap_server = 'mail.bdhg.xyz'"
|
||||
))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def _seed():
|
||||
|
||||
@@ -93,8 +93,8 @@ class LoginBatchRunner:
|
||||
|
||||
return None, ''
|
||||
|
||||
def _execute_one(self, task_id: int, acc_info: dict, proxy_dict: Optional[dict], total: int):
|
||||
"""在独立线程中执行单个账号登录，使用独立的 DB 会话。"""
|
||||
def _execute_one(self, task_id: int, acc_info: dict, total: int):
|
||||
"""在独立线程中执行单个账号登录，使用独立的 DB 会话和代理。"""
|
||||
if self._stop.is_set():
|
||||
self._push_log("warning", f"任务已停止，跳过: {acc_info['username']}")
|
||||
return
|
||||
@@ -114,6 +114,20 @@ class LoginBatchRunner:
|
||||
|
||||
self._push_log("info", f"[{current}/{total}] 开始登录: {acc_info['username']}")
|
||||
|
||||
# 每个账号独立获取代理
|
||||
proxy_dict, proxy_msg = self._resolve_proxy()
|
||||
if proxy_msg:
|
||||
self._push_log("info", f"[{current}] {proxy_msg}")
|
||||
|
||||
# 代理预检失败时，该账号标记失败但不终止整个批次
|
||||
if self.proxy_config and self.proxy_config.enabled and not proxy_dict:
|
||||
task.status = "error"
|
||||
task.message = f"代理不可用: {proxy_msg}"
|
||||
task.finished_at = datetime.now(timezone.utc)
|
||||
worker_db.commit()
|
||||
self._push_log("error", f"[{current}] {acc_info['username']} 代理不可用: {proxy_msg}")
|
||||
return
|
||||
|
||||
try:
|
||||
account = Account(
|
||||
username=acc_info["username"],
|
||||
@@ -217,29 +231,7 @@ class LoginBatchRunner:
|
||||
self._push_log("result", "")
|
||||
return
|
||||
|
||||
# 代理预检
|
||||
proxy_dict, proxy_msg = self._resolve_proxy()
|
||||
if proxy_msg:
|
||||
self._push_log("info", proxy_msg)
|
||||
|
||||
# 如果启用了代理但预检失败，终止任务
|
||||
if self.proxy_config and self.proxy_config.enabled and not proxy_dict:
|
||||
self._push_log("error", f"代理不可用，任务终止: {proxy_msg}")
|
||||
for item in task_infos:
|
||||
worker_db = SessionLocal()
|
||||
try:
|
||||
task = worker_db.query(LoginTask).filter(LoginTask.id == item["task_id"]).first()
|
||||
if task:
|
||||
task.status = "error"
|
||||
task.message = f"代理不可用: {proxy_msg}"
|
||||
task.finished_at = datetime.now(timezone.utc)
|
||||
worker_db.commit()
|
||||
finally:
|
||||
worker_db.close()
|
||||
self._push_log("result", "")
|
||||
return
|
||||
|
||||
# 并发执行登录
|
||||
# 并发执行登录，每个账号独立获取代理
|
||||
with ThreadPoolExecutor(max_workers=concurrency) as executor:
|
||||
futures = []
|
||||
for item in task_infos:
|
||||
@@ -250,7 +242,6 @@ class LoginBatchRunner:
|
||||
self._execute_one,
|
||||
item["task_id"],
|
||||
item["acc_info"],
|
||||
proxy_dict,
|
||||
total,
|
||||
)
|
||||
futures.append(future)
|
||||
|
||||
Reference in New Issue
Block a user