"""斗鱼账号检测批次执行器。""" 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 from core.douyu import DouyuLogin, WgapiLoginAPI 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( 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()