"""斗鱼账号检测批次执行器。""" from __future__ import annotations import io import threading import uuid import zipfile from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from datetime import datetime, timezone from types import SimpleNamespace from typing import Optional, cast from core.douyu import DouyuLogin, WgapiLoginAPI from core.douyu.login import AccountLike from core.douyu.proxy_fetcher import ProxyFetcher from ..models import ProxyConfig as ProxyConfigModel def _now() -> datetime: """返回时区感知 UTC 时间。""" return datetime.now(timezone.utc) STATUS_LABELS = { "account_cancelled": "账号已注销", "password_wrong": "账号密码错误", "account_unverified": "账号未认证", "account_verified": "账号已认证", "account_auth_unknown": "认证状态未知", "error": "检测失败", "stopped": "已停止", } EXPORT_STATUS_ORDER = [ "account_cancelled", "password_wrong", "account_unverified", "account_verified", "account_auth_unknown", "error", "stopped", ] @dataclass class AccountCheckInput: """导入的一行斗鱼账号。""" line: int username: str password: str email: str email_password: str def export_line(self) -> str: """导出为统一四段格式。""" return f"{self.username}----{self.password}----{self.email}----{self.email_password}" @dataclass class AccountCheckItemState: """单个账号检测状态。""" line: int username: str email: str export_text: str status: str = "pending" message: str = "等待开始" started_at: datetime | None = None finished_at: datetime | None = None def to_dict(self) -> dict: return { "line": self.line, "username": self.username, "email": self.email, "status": self.status, "message": self.message, "started_at": self.started_at, "finished_at": self.finished_at, } @dataclass class AccountCheckBatch: """账号检测批次内存快照。""" batch_id: str created_by: int concurrency: int max_login_retries: int max_total_time: float use_proxy: bool items: list[AccountCheckItemState] status: str = "pending" message: str = "等待开始" created_at: datetime = field(default_factory=_now) started_at: datetime | None = None finished_at: datetime | None = None def parse_account_check_lines(text: str) -> list[AccountCheckInput]: """解析账号检测导入文本，支持 ---- 和 | 两种分隔符。""" accounts: list[AccountCheckInput] = [] for line_no, raw_line in enumerate(text.splitlines(), 1): line = raw_line.strip() if not line: continue separator = "----" if "----" in line else "|" parts = [part.strip() for part in line.split(separator)] if len(parts) != 4 or any(not part for part in parts): raise ValueError( f"第 {line_no} 行格式错误，请使用：账号----密码----邮箱----邮箱密码 " f"或 账号|密码|邮箱|邮箱密码" ) accounts.append( AccountCheckInput( line=line_no, username=parts[0], password=parts[1], email=parts[2], email_password=parts[3], ) ) return accounts class AccountCheckRunner: """在后台线程中批量检测斗鱼账号。""" def __init__( self, batch: AccountCheckBatch, accounts: list[AccountCheckInput], proxy_config: Optional[ProxyConfigModel] = None, ): self.batch = batch self.accounts = accounts self.proxy_config = proxy_config self._lock = threading.Lock() self._stop = threading.Event() self._shared_proxy_fetcher = self._create_proxy_fetcher() def _create_proxy_fetcher(self) -> ProxyFetcher | None: """按全局代理配置创建 API 代理获取器。""" if ( not self.batch.use_proxy or not self.proxy_config or not self.proxy_config.enabled or not self.proxy_config.api_url ): return None wl_platform = "xiequ" wl_credentials = None if self.proxy_config.whitelist_enabled: wl_platform = ( getattr(self.proxy_config, "whitelist_platform", None) or "xiequ" ) wl_credentials = getattr(self.proxy_config, "whitelist_credentials", None) if ( not wl_credentials and self.proxy_config.whitelist_uid and self.proxy_config.whitelist_ukey ): wl_credentials = { "uid": self.proxy_config.whitelist_uid, "ukey": self.proxy_config.whitelist_ukey, } return ProxyFetcher( api_url=self.proxy_config.api_url, whitelist_platform=wl_platform, whitelist_credentials=wl_credentials, stop_event=self._stop, ) def _resolve_static_proxy(self) -> tuple[dict[str, str] | None, str]: """解析静态代理；API 代理由 DouyuLogin 内部通过 proxy_fetcher 获取。""" if not self.batch.use_proxy: return None, "" 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}, "" if self._shared_proxy_fetcher: return None, "" return None, "代理不可用: 未配置代理" def stop(self): self._stop.set() with self._lock: if self.batch.status == "running": self.batch.message = "正在停止" def _set_item(self, index: int, **updates): with self._lock: item = self.batch.items[index] for key, value in updates.items(): setattr(item, key, value) def _set_batch(self, **updates): with self._lock: for key, value in updates.items(): setattr(self.batch, key, value) def _run_one(self, index: int, account: AccountCheckInput): if self._stop.is_set(): self._set_item( index, status="stopped", message="已停止", finished_at=_now() ) return proxy_dict, proxy_error = self._resolve_static_proxy() if proxy_error: self._set_item( index, status="error", message=proxy_error, finished_at=_now() ) return self._set_item( index, status="running", message="检测中", started_at=_now(), finished_at=None, ) try: result = DouyuLogin( cast( AccountLike, SimpleNamespace( username=account.username, password=account.password, email=account.email, email_password=account.email_password, email_imap_server="", email_imap_port=993, email_imap_ssl=True, ), ), proxy=proxy_dict, max_login_retries=self.batch.max_login_retries, max_total_time=self.batch.max_total_time, proxy_fetcher=self._shared_proxy_fetcher, stop_event=self._stop, api_strategy=WgapiLoginAPI(), ).check_account() except Exception as exc: self._set_item( index, status="error", message=f"检测异常: {exc}", finished_at=_now() ) return if self._stop.is_set() and not result.success: self._set_item( index, status="stopped", message=result.message or "已停止", finished_at=_now(), ) return if result.success: status = ( result.code if result.code in STATUS_LABELS else "account_auth_unknown" ) message = result.message or STATUS_LABELS.get(status, "认证状态未知") else: status = "error" message = result.message or "检测失败" self._set_item(index, status=status, message=message, finished_at=_now()) def snapshot(self) -> dict: with self._lock: items = [item.to_dict() for item in self.batch.items] status_counts = { status: sum(1 for item in self.batch.items if item.status == status) for status in STATUS_LABELS } running_count = sum( 1 for item in self.batch.items if item.status in {"pending", "running"} ) finished_count = len(self.batch.items) - running_count return { "batch_id": self.batch.batch_id, "status": self.batch.status, "message": self.batch.message, "created_by": self.batch.created_by, "concurrency": self.batch.concurrency, "max_login_retries": self.batch.max_login_retries, "max_total_time": self.batch.max_total_time, "use_proxy": self.batch.use_proxy, "total": len(self.batch.items), "finished_count": finished_count, "running_count": running_count, "status_counts": status_counts, "created_at": self.batch.created_at, "started_at": self.batch.started_at, "finished_at": self.batch.finished_at, "items": items, } def build_zip(self) -> tuple[bytes, str]: """按检测状态生成 zip 包。""" with self._lock: items = list(self.batch.items) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") buffer = io.BytesIO() with zipfile.ZipFile(buffer, "w", compression=zipfile.ZIP_DEFLATED) as zf: for status in EXPORT_STATUS_ORDER: label = STATUS_LABELS[status] lines: list[str] = [] for item in items: if item.status != status: continue line = item.export_text if ( status in {"error", "stopped", "account_auth_unknown"} and item.message ): line = f"{line}----{item.message}" lines.append(line) content = "\n".join(lines) if content: content += "\n" zf.writestr(f"{timestamp}_{label}.txt", content.encode("utf-8")) filename = f"account_check_{timestamp}.zip" return buffer.getvalue(), filename def run(self): """线程入口。""" self._set_batch( status="running", message="批次运行中", started_at=_now(), finished_at=None ) try: if self._shared_proxy_fetcher: self._shared_proxy_fetcher.warmup_whitelist() with ThreadPoolExecutor(max_workers=self.batch.concurrency) as executor: futures = [] for index, account in enumerate(self.accounts): if self._stop.is_set(): self._set_item( index, status="stopped", message="已停止", finished_at=_now(), ) continue futures.append(executor.submit(self._run_one, index, account)) for future in as_completed(futures): future.result() except Exception as exc: self._set_batch( status="error", message=f"批次执行异常: {exc}", finished_at=_now() ) return if self._stop.is_set(): self._set_batch(status="stopped", message="批次已停止", finished_at=_now()) else: self._set_batch(status="finished", message="批次已完成", finished_at=_now()) class AccountCheckRegistry: """管理账号检测批次。""" def __init__(self): self._lock = threading.Lock() self._runners: dict[str, AccountCheckRunner] = {} def create( self, accounts: list[AccountCheckInput], created_by: int, concurrency: int, max_login_retries: int, max_total_time: float, use_proxy: bool = False, proxy_config: Optional[ProxyConfigModel] = None, ) -> AccountCheckRunner: batch_id = uuid.uuid4().hex[:12] normalized_retries = max_login_retries if max_login_retries > 0 else 20 batch = AccountCheckBatch( batch_id=batch_id, created_by=created_by, concurrency=max(1, min(int(concurrency or 1), 10)), max_login_retries=normalized_retries, max_total_time=max(0.0, float(max_total_time or 0)), use_proxy=bool(use_proxy), items=[ AccountCheckItemState( line=account.line, username=account.username, email=account.email, export_text=account.export_line(), ) for account in accounts ], ) runner = AccountCheckRunner( batch=batch, accounts=accounts, proxy_config=proxy_config ) with self._lock: self._runners[batch_id] = runner return runner def get(self, batch_id: str) -> AccountCheckRunner | None: with self._lock: return self._runners.get(batch_id) account_check_registry = AccountCheckRegistry()