yml2213
0a922ce56d
S3 让 `status` 可信，S4 让**时间线完整**：账号上发生的事，时间线里都能查到。 ## 补齐「定义了却从没发过」的事件 `assigned` / `tag_changed` 这两个类型早就定义了、前端 `EVENT_STYLE` 也渲染了， 但**全仓没有任何代码发过**（`grep` 只有定义处）。本次接上： - 单个分配 / 批量分配 → `assigned`（detail 带 assigned_to / assigned_username） - 单个改标签 / 批量改标签 / 按筛选批量改标签 → `tag_changed`（detail 带 from/to） ## 新增 `ck_cleared`，并修掉 `ck_invalid` 的刷屏 - **`ck_cleared`**：清 CK 是**不可逆**操作（重登当前成功率极低），时间线里却不留痕。 现在单个清除与批量清除共用 `account_lifecycle.on_ck_cleared()`。 - **`ck_invalid` 改为只在「由好变坏」时发**。旧实现每轮巡检只要判定 invalid 就发一条， 源码注释写着「避免每轮刷屏」，**但代码并没有做到** —— 6 小时一轮、失效一周就是 28 条一模一样的事件。判定失效是**一次状态转变**，不是每轮读数。`error`（查不到） 同样不改变结论、不发事件。 ## 审计与时间线一次写完（防漏一边） 新增 `audit_service.record_account_action()`：**审计 + 账号事件一起写**，并把 `account_id` 落到审计上。这两件事必须成对出现 —— 只写审计，时间线就缺「谁改了什么」； 只发事件，合规审计就缺一条。分成两次调用写，迟早有人漏一边（`assigned` 就是这么漏的）。 顺带把批量删除的审计补上被删账号 id（`detail.account_ids`，截前 200 条）： 删账号会**连带删掉它的事件时间线**（外键），审计是唯一痕迹 —— 这正是 10-03 那次事故要留的东西。 ## 批量操作不再逐条开 SAVEPOINT 批量分配/改标签可能覆盖几千个账号，而每个账号的时间线都该各留一条。 逐条走 `emit()` 会给每条开一个 SAVEPOINT，几千条时请求明显变慢。 新增 `emit_bulk()`：一次 executemany，整批包在**一个** SAVEPOINT 里， 与 `emit()` 一样**绝不抛异常**（整批失败就整批丢弃 + warning，不影响业务写入）。 ## 检测入口收口：4 处 → 1 个 `on_ck_verdict()` `ck_health` / 人工检测 / 工作台执行前检测 / 重登后复检 原本各自写 `ck_check_status` + `ck_check_result` + `ck_checked_at` + 发事件 + 调 `sync_status`， 口径容易分叉（`ck_invalid` 的刷屏就是这么来的）。现在统一走 `account_lifecycle.on_ck_verdict()`，顺序固定：**写结论 → 由好变坏记事件 → 重算状态**。 至此 `account_lifecycle` 成为**唯一**写生命周期字段的模块，并加了一条**源码扫描用例** 把它钉死（任何地方直接 `acc.status = ...` 都会让它红，并指出 file:line）—— 比写在文档里的约定可靠。虎牙的 `huya_accounts.status` 是同一个病，但属另一套模型， 已在用例里显式排除并注明。 ## 接口与前端 - 时间线支持 `?category=` 筛选（lifecycle / auth / business / ops）； 响应补 `category` / `subject_*` / `actor_*` - 前端 `EVENT_STYLE` 补上 `status_changed`（**S3 我漏的**：加了事件类型却没加样式， 页面上显示成原始英文）与 `ck_cleared`；时间线加「分类」筛选 ## 验证 - `TZ=Asia/Shanghai`：**420 passed / 6 skipped**（S3 后 404 → 新增 16 条） - 迁移链 3 passed；ruff check/format 全绿；前端 `tsc -b && vite build` 与 `npm run lint`（0 errors）均通过 - **两条关键用例都做过「反向验证」**（先注入旧行为确认会红）： - 把 `ck_invalid` 改回「每轮都发」→ 用例报 `3 == 1`； - 在路由里直接写 `acc.status = ...` → 不变量用例给出精确 file:line。 - 新用例还覆盖：批量分配/改标签逐账号发事件、清 CK 事件与审计关联、 `emit_bulk` 的跳过与容错、时间线分类筛选、批量删除的 id 留痕。 未做（后续）：S5 业务事实直发事件 + `douyu_tasks` 保留策略、S6 账号页重构 + `bind_status` 三列拆解。`remark_changed` 目前没有改备注的接口，故无发射点（已在文档标注）。
828 lines
34 KiB
Python
828 lines
34 KiB
Python
"""登录服务：复用 core/ 核心模块，在线程池中并发执行登录并推送日志。"""
|
||
|
||
import asyncio
|
||
import os
|
||
import threading
|
||
import time
|
||
import uuid
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from datetime import UTC, datetime
|
||
from types import SimpleNamespace
|
||
from typing import cast
|
||
|
||
from loguru import logger
|
||
from sqlalchemy import and_, func, or_
|
||
from sqlalchemy.orm import Session
|
||
|
||
from core.douyu import DouyuLogin, IframeLoginAPI, WgapiLoginAPI
|
||
from core.douyu.login import AccountLike
|
||
from core.douyu.proxy_fetcher import ProxyFetcher
|
||
|
||
from ..models import Account as AccountModel
|
||
from ..models import LoginTask
|
||
from ..models import ProxyConfig as ProxyConfigModel
|
||
from .account_events_service import EVENT_LOGIN_OK, EVENT_RELOGIN_OK, emit
|
||
from .account_lifecycle import on_ck_verdict, on_login_success
|
||
from .cookie_check_service import check_douyu_cookie
|
||
from .douyu_service import LOGIN_ATTEMPT_IN_FLIGHT_STATUSES
|
||
|
||
|
||
def _create_api_strategy(strategy_name: str):
|
||
"""根据名称创建登录接口策略实例。"""
|
||
if strategy_name == "iframe":
|
||
return IframeLoginAPI()
|
||
return WgapiLoginAPI()
|
||
|
||
|
||
CHECK_STATUS_MESSAGES = {
|
||
"account_cancelled": "账号已注销",
|
||
"password_wrong": "账号密码错误",
|
||
"account_unverified": "账号未认证",
|
||
"account_verified": "账号已认证",
|
||
"account_auth_unknown": "账号认证状态未知",
|
||
}
|
||
|
||
CHECK_STATUS_LOG_LEVELS = {
|
||
"account_cancelled": "warning",
|
||
"password_wrong": "error",
|
||
"account_unverified": "warning",
|
||
"account_verified": "success",
|
||
"account_auth_unknown": "warning",
|
||
}
|
||
|
||
|
||
def _positive_env_int(name: str, default: int) -> int:
|
||
"""读取正整数环境变量，非法值回退到默认值。"""
|
||
try:
|
||
value = int(os.getenv(name, str(default)))
|
||
except ValueError:
|
||
return default
|
||
return value if value > 0 else default
|
||
|
||
|
||
def get_relogin_limits() -> tuple[int, int]:
|
||
"""读取客服 CK 重登的重试与总时长限制。"""
|
||
return (
|
||
_positive_env_int("COOKIE_RELOGIN_MAX_RETRIES", 5),
|
||
_positive_env_int("COOKIE_RELOGIN_MAX_TOTAL_TIME", 600),
|
||
)
|
||
|
||
|
||
def _snapshot_proxy_config(
|
||
proxy_config: ProxyConfigModel | None,
|
||
) -> ProxyConfigModel | None:
|
||
"""复制代理配置，避免后台线程访问已关闭会话中的 ORM 对象。"""
|
||
if proxy_config is None:
|
||
return None
|
||
credentials = getattr(proxy_config, "whitelist_credentials", None)
|
||
return cast(
|
||
ProxyConfigModel,
|
||
SimpleNamespace(
|
||
enabled=bool(getattr(proxy_config, "enabled", False)),
|
||
http=getattr(proxy_config, "http", "") or "",
|
||
https=getattr(proxy_config, "https", "") or "",
|
||
api_url=getattr(proxy_config, "api_url", "") or "",
|
||
whitelist_enabled=bool(getattr(proxy_config, "whitelist_enabled", False)),
|
||
whitelist_platform=getattr(proxy_config, "whitelist_platform", None),
|
||
whitelist_credentials=dict(credentials)
|
||
if isinstance(credentials, dict)
|
||
else credentials,
|
||
whitelist_uid=getattr(proxy_config, "whitelist_uid", "") or "",
|
||
whitelist_ukey=getattr(proxy_config, "whitelist_ukey", "") or "",
|
||
),
|
||
)
|
||
|
||
|
||
ATTEMPT_IN_FLIGHT_STATUSES = ("pending", "running")
|
||
# 升级前把尝试状态写进 status 的历史取值；判断与清理时一并兼容。
|
||
_LEGACY_IN_FLIGHT_STATUSES = (
|
||
"pending",
|
||
"running",
|
||
"relogin_pending",
|
||
"relogin_running",
|
||
)
|
||
|
||
|
||
def _attempt_in_flight(task: LoginTask) -> bool:
|
||
"""判断这条记录是否正被某个批次处理（只看尝试字段，不看 CK 字段）。"""
|
||
if task.attempt_status in ATTEMPT_IN_FLIGHT_STATUSES:
|
||
return True
|
||
return not task.attempt_status and task.status in _LEGACY_IN_FLIGHT_STATUSES
|
||
|
||
|
||
def _begin_attempt(task: LoginTask, batch_id: str, message: str) -> None:
|
||
"""登记一次登录尝试。
|
||
|
||
只写 attempt_*：账号当前持有的 CK（status/cookie/message）在登录成功前
|
||
一律不动，所以批次被停止、失败或进程重启都不会让账号失去 CK。
|
||
"""
|
||
task.attempt_status = "pending"
|
||
task.attempt_message = message
|
||
task.attempt_batch_id = batch_id
|
||
task.attempt_at = None
|
||
|
||
|
||
def _mark_attempt_failed(
|
||
task: LoginTask, message: str, *, status: str = "failed"
|
||
) -> None:
|
||
"""记录一次失败/被中断的尝试；**账号已有的 CK 原样不动**。
|
||
|
||
这里读的是 `task.account.cookie`（CK 自 2026-10-04 起挂在账号上），
|
||
而不是这条登录记录 —— 记录不再持有凭证。
|
||
"""
|
||
completed_at = datetime.now(UTC)
|
||
task.attempt_status = status if status in {"failed", "error"} else "failed"
|
||
task.attempt_message = message
|
||
task.attempt_at = completed_at
|
||
account = task.account
|
||
if not (account and account.cookie):
|
||
# 账号还没有任何 CK 需要保护：保持旧的失败语义，便于复用记录与页面统计。
|
||
task.status = task.attempt_status
|
||
task.message = message
|
||
task.finished_at = completed_at
|
||
|
||
|
||
def _write_login_success(
|
||
db: Session,
|
||
task: LoginTask,
|
||
*,
|
||
cookie: str,
|
||
message: str,
|
||
batch_id: str,
|
||
event_type: str = EVENT_LOGIN_OK,
|
||
) -> None:
|
||
"""登录成功后把新 CK 写到**账号**上：这是唯一替换旧 CK 的入口。
|
||
|
||
顺带记一条账号事件 —— CK 是**覆盖式**字段，换掉之后就看不出「这账号什么时候
|
||
登录成功过、换过几次 CK」。事件是旁路（emit 内部吞异常），记不上也不影响登录。
|
||
"""
|
||
completed_at = datetime.now(UTC)
|
||
account = task.account
|
||
task.status = "success"
|
||
task.message = message
|
||
task.batch_id = batch_id
|
||
task.finished_at = completed_at
|
||
task.attempt_status = "success"
|
||
task.attempt_message = message
|
||
task.attempt_batch_id = batch_id
|
||
task.attempt_at = completed_at
|
||
|
||
if account is not None:
|
||
# CK 是**账号级凭证**，不是「某次登录的产物」—— 写账号，不写记录。
|
||
account.cookie = cookie
|
||
account.cookie_updated_at = completed_at
|
||
account.cookie_source = "web"
|
||
# 登录成功同时作废上一轮重登的退避痕迹，并把状态重算回正常。
|
||
# 人工手动登录也走这条 →「待人工」的账号被人工救回来时状态会正确恢复。
|
||
on_login_success(db, account, occurred_at=completed_at)
|
||
|
||
emit(
|
||
db,
|
||
account_id=task.account_id,
|
||
event_type=event_type,
|
||
title="重新登录成功，Cookie 已更新"
|
||
if event_type == EVENT_RELOGIN_OK
|
||
else "登录成功",
|
||
occurred_at=completed_at,
|
||
detail={"message": message, "batch_id": batch_id},
|
||
source="login",
|
||
# 幂等键用登录记录 + 完成时间：同一账号多次登录是多个事件，
|
||
# 但同一次登录重复写回不会重复入库。
|
||
source_ref=f"login_task:{task.id}:{int(completed_at.timestamp())}",
|
||
)
|
||
|
||
|
||
class LoginBatchRunner:
|
||
"""批量登录执行器，在线程中运行，通过 ThreadPoolExecutor 并发登录多个账号。"""
|
||
|
||
def __init__(
|
||
self,
|
||
db: Session,
|
||
account_ids: list[int],
|
||
created_by: int,
|
||
creator_permissions: list[str],
|
||
max_login_retries: int = 0,
|
||
max_total_time: float = 0,
|
||
proxy_config: ProxyConfigModel | None = None,
|
||
log_queue: asyncio.Queue | None = None,
|
||
loop: asyncio.AbstractEventLoop | None = None,
|
||
concurrency: int = 3,
|
||
api_strategy: str = "wgapi",
|
||
mode: str = "login",
|
||
):
|
||
self.db = db
|
||
self.account_ids = account_ids
|
||
self.created_by = created_by
|
||
self.creator_permissions = creator_permissions
|
||
# 限制最大重试次数，避免无限重试导致资源耗尽
|
||
# 0 表示使用默认值 20，其他值保持原样
|
||
self.max_login_retries = max_login_retries if max_login_retries > 0 else 20
|
||
self.max_total_time = max_total_time
|
||
self.proxy_config = _snapshot_proxy_config(proxy_config)
|
||
self.log_queue = log_queue
|
||
self.loop = loop
|
||
self.batch_id = uuid.uuid4().hex[:12]
|
||
self.concurrency = max(1, min(concurrency, 10)) # 限制 1-10
|
||
self.api_strategy = _create_api_strategy(api_strategy)
|
||
self.mode = mode if mode in ("login", "check", "relogin") else "login"
|
||
self._stop = threading.Event()
|
||
self._counter_lock = threading.Lock()
|
||
self._completed = 0
|
||
|
||
# 共享代理获取器（无池，每次取新代理）
|
||
self._shared_proxy_fetcher = None
|
||
proxy_config = self.proxy_config
|
||
if proxy_config and proxy_config.enabled and proxy_config.api_url:
|
||
wl_platform = "xiequ"
|
||
wl_credentials = None
|
||
if proxy_config.whitelist_enabled:
|
||
wl_platform = (
|
||
getattr(proxy_config, "whitelist_platform", None) or "xiequ"
|
||
)
|
||
wl_credentials = getattr(proxy_config, "whitelist_credentials", None)
|
||
# 向后兼容
|
||
if (
|
||
not wl_credentials
|
||
and proxy_config.whitelist_uid
|
||
and proxy_config.whitelist_ukey
|
||
):
|
||
wl_platform = "xiequ"
|
||
wl_credentials = {
|
||
"uid": proxy_config.whitelist_uid,
|
||
"ukey": proxy_config.whitelist_ukey,
|
||
}
|
||
|
||
self._shared_proxy_fetcher = ProxyFetcher(
|
||
api_url=proxy_config.api_url,
|
||
whitelist_platform=wl_platform,
|
||
whitelist_credentials=wl_credentials,
|
||
stop_event=self._stop,
|
||
)
|
||
|
||
def stop(self):
|
||
self._stop.set()
|
||
|
||
def _sleep_or_stop(self, seconds: float) -> bool:
|
||
"""可中断等待；返回 True 表示收到停止信号。"""
|
||
deadline = time.monotonic() + seconds
|
||
while time.monotonic() < deadline:
|
||
if self._stop.is_set():
|
||
return True
|
||
time.sleep(min(0.2, deadline - time.monotonic()))
|
||
return self._stop.is_set()
|
||
|
||
def _mark_relogin_stopped(
|
||
self,
|
||
task_id: int,
|
||
message: str = "重新登录已停止，旧 Cookie 已保留",
|
||
) -> None:
|
||
"""把未开始或已中断的重登尝试收敛为终态，旧 CK 原封不动。"""
|
||
worker_db = SessionLocal()
|
||
try:
|
||
task = worker_db.query(LoginTask).filter(LoginTask.id == task_id).first()
|
||
if task and _attempt_in_flight(task):
|
||
_mark_attempt_failed(task, message)
|
||
worker_db.commit()
|
||
finally:
|
||
worker_db.close()
|
||
|
||
def _mark_account_relogin_stopped(
|
||
self,
|
||
account_id: int,
|
||
message: str = "重新登录已停止，旧 Cookie 已保留",
|
||
) -> None:
|
||
"""按**账号**收敛在途的重登尝试（新模型下调度单位是账号）。
|
||
|
||
与 `_mark_relogin_stopped` 的区别只是入口：批次挂了但记录还没建出来时，
|
||
这里是唯一能把「该账号的在途尝试」收干净的路径。
|
||
"""
|
||
worker_db = SessionLocal()
|
||
try:
|
||
tasks = (
|
||
worker_db.query(LoginTask)
|
||
.filter(LoginTask.account_id == account_id)
|
||
.order_by(LoginTask.id.desc())
|
||
.all()
|
||
)
|
||
changed = False
|
||
for task in tasks:
|
||
if _attempt_in_flight(task):
|
||
_mark_attempt_failed(task, message)
|
||
changed = True
|
||
if changed:
|
||
worker_db.commit()
|
||
finally:
|
||
worker_db.close()
|
||
|
||
def _push_log(self, level: str, message: str):
|
||
# 即使没有页面实时日志，也要保留批次进度到 app.log，便于排查卡点。
|
||
if message:
|
||
log_level = (
|
||
level
|
||
if level in {"debug", "info", "warning", "error", "success"}
|
||
else "debug"
|
||
)
|
||
getattr(logger, log_level)(f"[登录批次 {self.batch_id}] {message}")
|
||
if self.log_queue and self.loop:
|
||
asyncio.run_coroutine_threadsafe(
|
||
self.log_queue.put({"level": level, "message": message}),
|
||
self.loop,
|
||
)
|
||
|
||
def _resolve_static_proxy(self) -> tuple[dict | None, str]:
|
||
"""解析静态代理配置。"""
|
||
if not self.proxy_config or not self.proxy_config.enabled:
|
||
return None, ""
|
||
|
||
# 静态代理
|
||
if self.proxy_config.http or self.proxy_config.https:
|
||
proxy_url = self.proxy_config.http or self.proxy_config.https
|
||
return {"http": proxy_url, "https": proxy_url}, f"使用静态代理: {proxy_url}"
|
||
|
||
# API代理：由 DouyuLogin 通过 proxy_fetcher 内部管理
|
||
return None, ""
|
||
|
||
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']}")
|
||
self._mark_relogin_stopped(task_id, "任务已停止，旧 Cookie 已保留")
|
||
return
|
||
|
||
worker_db = SessionLocal()
|
||
try:
|
||
task = worker_db.query(LoginTask).filter(LoginTask.id == task_id).first()
|
||
if not task:
|
||
return
|
||
# CK、健康与重登状态都挂在账号上；登录记录只是一次尝试的历史。
|
||
account = task.account
|
||
|
||
# 只推进尝试状态；账号持有的旧 CK 在成功前保持不变。
|
||
task.attempt_status = "running"
|
||
task.attempt_message = (
|
||
"正在重新登录，旧 Cookie 保留中"
|
||
if self.mode == "relogin"
|
||
else "正在检测"
|
||
if self.mode == "check"
|
||
else "正在登录"
|
||
)
|
||
worker_db.commit()
|
||
|
||
with self._counter_lock:
|
||
self._completed += 1
|
||
current = self._completed
|
||
|
||
action_name = (
|
||
"检测"
|
||
if self.mode == "check"
|
||
else "重新登录"
|
||
if self.mode == "relogin"
|
||
else "登录"
|
||
)
|
||
self._push_log(
|
||
"info", f"[{current}/{total}] 开始{action_name}: {acc_info['username']}"
|
||
)
|
||
|
||
try:
|
||
# 代理配置也可能异常，必须由当前任务的失败处理收敛状态。
|
||
proxy_dict, proxy_msg = self._resolve_static_proxy()
|
||
if proxy_msg:
|
||
self._push_log("info", f"[{current}] {proxy_msg}")
|
||
|
||
# 静态代理启用但配置为空 → 不可用
|
||
if (
|
||
self.proxy_config
|
||
and self.proxy_config.enabled
|
||
and not (self.proxy_config.http or self.proxy_config.https)
|
||
and not self._shared_proxy_fetcher
|
||
and not proxy_dict
|
||
):
|
||
if self.mode == "relogin":
|
||
_mark_attempt_failed(
|
||
task,
|
||
"重新登录失败: 代理不可用: 未配置代理（旧 Cookie 已保留）",
|
||
)
|
||
else:
|
||
_mark_attempt_failed(
|
||
task, "代理不可用: 未配置代理", status="error"
|
||
)
|
||
worker_db.commit()
|
||
self._push_log(
|
||
"error", f"[{current}] {acc_info['username']} 代理不可用"
|
||
)
|
||
return
|
||
|
||
account = cast(
|
||
AccountLike,
|
||
SimpleNamespace(
|
||
username=acc_info["username"],
|
||
password=acc_info["password"],
|
||
email=acc_info["email"],
|
||
email_password=acc_info["email_password"],
|
||
email_imap_server=acc_info["email_imap_server"] or "",
|
||
email_imap_port=acc_info["email_imap_port"] or 993,
|
||
email_imap_ssl=acc_info["email_imap_ssl"],
|
||
),
|
||
)
|
||
|
||
loginer = DouyuLogin(
|
||
account,
|
||
proxy=proxy_dict,
|
||
max_login_retries=self.max_login_retries,
|
||
max_total_time=self.max_total_time,
|
||
proxy_fetcher=self._shared_proxy_fetcher,
|
||
stop_event=self._stop,
|
||
api_strategy=self.api_strategy,
|
||
)
|
||
result = (
|
||
loginer.check_account() if self.mode == "check" else loginer.login()
|
||
)
|
||
|
||
if self.mode == "check" and result.success:
|
||
status = (
|
||
result.code
|
||
if result.code in CHECK_STATUS_MESSAGES
|
||
else "account_auth_unknown"
|
||
)
|
||
task.status = status
|
||
# ⚠️ 行为变更（2026-10-04 CK 解耦）：**检测不再清空账号的 CK**。
|
||
# 旧代码在这里写 `task.cookie = ""`，而那时记录的 cookie 就是账号的
|
||
# 唯一凭证 —— 于是一次「批量检测」就会把账号的登录态删掉。CK 换到账号
|
||
# 上之后，「检测账号凭据」与「CK 是否有效」本就是两件事（后者归
|
||
# ck_health 判），继续清空等于凭空制造不可逆的数据丢失
|
||
# （重登当前成功率极低，丢了补不回来）。
|
||
task.message = result.message or CHECK_STATUS_MESSAGES[status]
|
||
# 检测结果本身就是这次尝试的结论，页面按 attempt_* 展示。
|
||
task.attempt_status = status
|
||
task.attempt_message = task.message
|
||
task.attempt_at = datetime.now(UTC)
|
||
task.finished_at = datetime.now(UTC)
|
||
level = CHECK_STATUS_LOG_LEVELS.get(status, "info")
|
||
self._push_log(
|
||
level,
|
||
f"[{current}] {acc_info['username']} 检测结果: {task.message}",
|
||
)
|
||
elif result.success:
|
||
message = result.message or "登录成功"
|
||
if self.mode == "relogin" and account is not None:
|
||
check_result = check_douyu_cookie(result.cookie)
|
||
# 复检结论写账号（CK 在哪，健康状态就在哪），走统一入口：
|
||
# 写结论 + 由好变坏记事件 + 重算状态。
|
||
on_ck_verdict(
|
||
worker_db,
|
||
account,
|
||
"valid" if check_result["valid"] else "invalid",
|
||
detail=check_result,
|
||
occurred_at=check_result["checked_at"],
|
||
)
|
||
if check_result["valid"]:
|
||
message = "重新登录成功，Cookie 有效"
|
||
else:
|
||
message = f"重新登录成功，但 Cookie 有效性检测失败: {check_result['message']}"
|
||
# 唯一替换旧 CK 的入口：登录成功后写回账号。
|
||
#
|
||
# ⚠️ 必须传 **worker_db**（本线程的会话），不能传 self.db：
|
||
# ① `self.db` 由主线程持有，SQLAlchemy Session 不是线程安全的；
|
||
# ② 更致命的是 `emit()` 会往 account_events 插一行，其
|
||
# `account_id` 外键会对 accounts 那一行加**共享锁**。如果这
|
||
# 把锁加在**另一个**会话上，紧接着本会话对同一个账号的
|
||
# `UPDATE accounts SET cookie=...` 就会被自己人挡住 ——
|
||
# 实测直接 50 秒后 `Lock wait timeout exceeded`，登录明明成功
|
||
# 却写不回 CK。事件本就该和状态变更在同一个事务里提交。
|
||
_write_login_success(
|
||
worker_db,
|
||
task,
|
||
cookie=result.cookie,
|
||
message=message,
|
||
batch_id=self.batch_id,
|
||
event_type=(
|
||
EVENT_RELOGIN_OK
|
||
if self.mode == "relogin"
|
||
else EVENT_LOGIN_OK
|
||
),
|
||
)
|
||
if self.mode == "relogin":
|
||
if account is not None and account.ck_check_status == "valid":
|
||
self._push_log(
|
||
"success",
|
||
f"[{current}] {acc_info['username']} 重新登录成功，Cookie 已替换并验证有效",
|
||
)
|
||
else:
|
||
detail = (
|
||
account.ck_check_result.get("message")
|
||
if account is not None and account.ck_check_result
|
||
else ""
|
||
)
|
||
self._push_log(
|
||
"warning",
|
||
f"[{current}] {acc_info['username']} 重新登录成功，但 Cookie 有效性检测失败: {detail}",
|
||
)
|
||
else:
|
||
self._push_log(
|
||
"success",
|
||
f"[{current}] {acc_info['username']} {task.message}",
|
||
)
|
||
else:
|
||
if self.mode == "relogin":
|
||
# 重登失败不动旧 CK 与成功状态，只记录这次尝试失败
|
||
_mark_attempt_failed(
|
||
task, f"重新登录失败: {result.message}（旧 Cookie 已保留）"
|
||
)
|
||
self._push_log(
|
||
"error",
|
||
f"[{current}] {acc_info['username']} 重新登录失败: {result.message}",
|
||
)
|
||
else:
|
||
_mark_attempt_failed(task, result.message or "登录失败")
|
||
self._push_log(
|
||
"error",
|
||
f"[{current}] {acc_info['username']} {action_name}失败: {result.message}",
|
||
)
|
||
|
||
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
|
||
if self.mode == "relogin":
|
||
_mark_attempt_failed(task, f"重新登录异常: {e}（旧 Cookie 已保留）")
|
||
self._push_log(
|
||
"error", f"[{current}] {acc_info['username']} 重新登录异常: {e}"
|
||
)
|
||
else:
|
||
_mark_attempt_failed(task, str(e), status="error")
|
||
self._push_log(
|
||
"error",
|
||
f"[{current}] {acc_info['username']} {action_name}异常: {e}",
|
||
)
|
||
|
||
worker_db.commit()
|
||
|
||
finally:
|
||
worker_db.close()
|
||
|
||
def run(self):
|
||
"""在线程中执行批量登录。"""
|
||
batch_id = self.batch_id
|
||
concurrency = self.concurrency
|
||
action_name = (
|
||
"账号检测"
|
||
if self.mode == "check"
|
||
else "重新登录"
|
||
if self.mode == "relogin"
|
||
else "登录"
|
||
)
|
||
self._push_log(
|
||
"info",
|
||
f"批量{action_name}任务 {batch_id} 开始，共 {len(self.account_ids)} 个账号，并发数: {concurrency}",
|
||
)
|
||
|
||
# 批次开始前同步一次出口 IP 到白名单，后续 fetch_new_proxy 不再主动同步
|
||
if self._shared_proxy_fetcher:
|
||
ok, msg = self._shared_proxy_fetcher.warmup_whitelist()
|
||
if msg != "无白名单凭据，跳过":
|
||
self._push_log("info" if ok else "warning", f"白名单预热: {msg}")
|
||
|
||
def _append_task_info(task: LoginTask, acc: AccountModel):
|
||
# 只登记这次尝试：账号当前持有的 CK（status/cookie/message）在
|
||
# 登录成功前一律不动，批次被停止或失败都不会让账号失去 CK。
|
||
_begin_attempt(
|
||
task,
|
||
batch_id,
|
||
"等待重新登录，旧 Cookie 已保留"
|
||
if self.mode == "relogin"
|
||
else "等待检测"
|
||
if self.mode == "check"
|
||
else "等待登录",
|
||
)
|
||
self.db.flush()
|
||
task_infos.append(
|
||
{
|
||
"task_id": task.id,
|
||
"acc_info": {
|
||
"username": acc.username,
|
||
"password": acc.password,
|
||
"email": acc.email,
|
||
"email_password": acc.email_password,
|
||
"email_imap_server": acc.email_imap_server or "",
|
||
"email_imap_port": acc.email_imap_port or 993,
|
||
"email_imap_ssl": acc.email_imap_ssl
|
||
if acc.email_imap_ssl is not None
|
||
else True,
|
||
},
|
||
}
|
||
)
|
||
|
||
try:
|
||
# 创建或复用任务记录（顺序执行，线程安全）
|
||
#
|
||
# **重登模式也从 account_ids 走**（2026-10-04 CK 解耦）：以前重登传的是
|
||
# 「持有 CK 的记录 id」，因为 CK 挂在记录上；现在 CK 在账号上，调度单位就是
|
||
# 账号，登录记录只是「这次尝试」的历史 —— 顺带让**没有登录历史的账号**
|
||
# 也能被重登（以前它根本进不了候选）。
|
||
task_infos: list[dict] = [] # {task_id, acc_info}
|
||
seen_account_ids = set()
|
||
for aid in self.account_ids:
|
||
if aid in seen_account_ids:
|
||
continue
|
||
seen_account_ids.add(aid)
|
||
acc = (
|
||
self.db.query(AccountModel)
|
||
.filter(AccountModel.id == aid)
|
||
.with_for_update()
|
||
.first()
|
||
)
|
||
if not acc:
|
||
continue
|
||
# 权限检查：客服只能跑分配给自己的
|
||
if (
|
||
"login:view_all" not in self.creator_permissions
|
||
and acc.assigned_to != self.created_by
|
||
):
|
||
self._push_log("warning", f"跳过无权账号: {acc.username}")
|
||
continue
|
||
|
||
# 解耦后不再需要「哪一行持有 CK」这个不变量（CK 在账号上）。
|
||
# 这里只需要保证**同一账号不并发两行**：
|
||
# ① 复用该账号在途的尝试（避免并发两个批次各建一行）；
|
||
# ② 否则复用最近一条失败/中断的历史，避免刷屏；
|
||
# ③ 都没有就新建一行（历史而已，允许一个账号多行）。
|
||
task = (
|
||
self.db.query(LoginTask)
|
||
.filter(
|
||
LoginTask.account_id == aid,
|
||
or_(
|
||
func.coalesce(LoginTask.attempt_status, "").in_(
|
||
LOGIN_ATTEMPT_IN_FLIGHT_STATUSES
|
||
),
|
||
and_(
|
||
LoginTask.attempt_status == "",
|
||
LoginTask.status.in_(LOGIN_ATTEMPT_IN_FLIGHT_STATUSES),
|
||
),
|
||
),
|
||
)
|
||
.order_by(LoginTask.id.desc())
|
||
.first()
|
||
)
|
||
if task is None:
|
||
# 复用最近一条**失败/中断**的历史，避免每次失败都刷一行。
|
||
# 注意要同时看 attempt_status：账号有 CK 时 `_mark_attempt_failed`
|
||
# 刻意不动 status（保护旧语义），所以失败记录的 status 可能还停在
|
||
# pending —— 只看 status 就永远命不中，行数会无限长。
|
||
task = (
|
||
self.db.query(LoginTask)
|
||
.filter(
|
||
LoginTask.account_id == aid,
|
||
or_(
|
||
LoginTask.status.in_(["failed", "error"]),
|
||
LoginTask.attempt_status.in_(["failed", "error"]),
|
||
),
|
||
)
|
||
.order_by(LoginTask.id.desc())
|
||
.first()
|
||
)
|
||
if task is None:
|
||
task = LoginTask(
|
||
batch_id=batch_id,
|
||
account_id=aid,
|
||
status="pending",
|
||
created_by=self.created_by,
|
||
)
|
||
self.db.add(task)
|
||
|
||
_append_task_info(task, acc)
|
||
|
||
self.db.commit()
|
||
total = len(task_infos)
|
||
if total == 0:
|
||
self._push_log("warning", "没有可执行的账号")
|
||
self._push_log("result", "")
|
||
return
|
||
|
||
# 并发执行登录，每个账号独立获取代理
|
||
with ThreadPoolExecutor(max_workers=concurrency) as executor:
|
||
futures = []
|
||
for item_index, item in enumerate(task_infos):
|
||
if self._stop.is_set():
|
||
self._push_log("warning", "任务已停止，跳过剩余账号")
|
||
# 所有模式都要把没轮到的尝试收敛掉，否则会一直停在
|
||
# 「等待中」；已有 CK 的记录本身不受影响。
|
||
for pending_item in task_infos[item_index:]:
|
||
self._mark_relogin_stopped(
|
||
pending_item["task_id"], "任务已停止，旧 Cookie 已保留"
|
||
)
|
||
break
|
||
future = executor.submit(
|
||
self._execute_one,
|
||
item["task_id"],
|
||
item["acc_info"],
|
||
total,
|
||
)
|
||
futures.append(future)
|
||
|
||
# 等待所有任务完成
|
||
for future in as_completed(futures):
|
||
try:
|
||
future.result()
|
||
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
|
||
self._push_log("error", f"Worker 异常: {e}")
|
||
|
||
self._push_log("info", f"批量{action_name}任务 {batch_id} 完成")
|
||
self._push_log("result", "")
|
||
finally:
|
||
# 标记批次结束：WS 端据此决定何时清理注册表（断线重连可继续订阅日志）。
|
||
batch_registry.mark_finished(batch_id)
|
||
# 确保 DB Session 被关闭，避免连接泄漏
|
||
self.db.close()
|
||
|
||
|
||
class BatchRegistry:
|
||
"""管理运行中的登录批次状态。"""
|
||
|
||
def __init__(self):
|
||
self._batches: dict[str, dict] = {}
|
||
|
||
def register(
|
||
self,
|
||
batch_id: str,
|
||
log_queue: asyncio.Queue | None,
|
||
loop: asyncio.AbstractEventLoop | None,
|
||
runner: LoginBatchRunner,
|
||
owner_id: int | None = None,
|
||
):
|
||
self._batches[batch_id] = {
|
||
"log_queue": log_queue,
|
||
"loop": loop,
|
||
"runner": runner,
|
||
"owner_id": owner_id,
|
||
"finished": False,
|
||
"finished_at": None,
|
||
}
|
||
|
||
def get(self, batch_id: str):
|
||
return self._batches.get(batch_id)
|
||
|
||
def mark_finished(self, batch_id: str):
|
||
"""标记批次已结束（幂等；不在本注册表的批次为无操作）。
|
||
|
||
与虎牙批次一致：WS 端只在 finished 后 pop，客户端断线重连仍可订阅
|
||
到运行中批次的实时日志。
|
||
"""
|
||
batch = self._batches.get(batch_id)
|
||
if batch:
|
||
batch["finished"] = True
|
||
batch["finished_at"] = time.time()
|
||
|
||
def pop(self, batch_id: str):
|
||
return self._batches.pop(batch_id, None)
|
||
|
||
|
||
# 模块级单例
|
||
batch_registry = BatchRegistry()
|
||
|
||
|
||
def _converge_interrupted_attempts(db: Session, message: str) -> int:
|
||
"""把遗留的在途尝试收敛为失败。
|
||
|
||
只写 attempt_*；对已经没有 CK 的记录才顺带落一个失败终态。返回收敛条数。
|
||
"""
|
||
tasks = (
|
||
db.query(LoginTask)
|
||
.filter(
|
||
or_(
|
||
LoginTask.attempt_status.in_(ATTEMPT_IN_FLIGHT_STATUSES),
|
||
LoginTask.status.in_(_LEGACY_IN_FLIGHT_STATUSES),
|
||
)
|
||
)
|
||
.all()
|
||
)
|
||
cleaned = 0
|
||
for task in tasks:
|
||
if not _attempt_in_flight(task):
|
||
continue
|
||
_mark_attempt_failed(task, message)
|
||
cleaned += 1
|
||
if cleaned:
|
||
db.commit()
|
||
return cleaned
|
||
|
||
|
||
def cleanup_orphan_relogin_tasks(
|
||
db: Session,
|
||
message: str = "重新登录已中断（服务重启），旧 Cookie 已保留",
|
||
) -> int:
|
||
"""服务重启后收敛遗留的重登尝试，旧 CK 原封不动。"""
|
||
return _converge_interrupted_attempts(db, message)
|
||
|
||
|
||
def cleanup_orphan_login_tasks(
|
||
db: Session,
|
||
message: str = "登录已中断（服务重启），旧 Cookie 已保留",
|
||
) -> int:
|
||
"""服务重启后收敛遗留的登录尝试。
|
||
|
||
在途尝试只体现在 attempt_* 上，收敛它不会碰到账号已有的 CK；只有本来
|
||
就没有 CK 的记录才会顺带标记失败，避免页面永久显示「排队中」。
|
||
"""
|
||
return _converge_interrupted_attempts(db, message)
|
||
|
||
|
||
# 在模块末尾导入 SessionLocal（避免循环导入）
|
||
from ..database import SessionLocal
|