"""登录服务：复用 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 .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"

    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 在哪，健康状态就在哪）。
                        account.ck_check_status = (
                            "valid" if check_result["valid"] else "invalid"
                        )
                        account.ck_check_result = {
                            "fish_ball": check_result["fish_ball"],
                            "nickname": check_result["nickname"],
                            "level": check_result["level"],
                            "message": check_result["message"],
                        }
                        account.ck_checked_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