""" 账号池维护模块 支持 CPA 平台和 Sub2Api 平台的探测、清理、计数和补号 """ from __future__ import annotations import asyncio import json import logging import re import threading import time from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime from typing import Any, Dict, List, Optional from urllib.parse import quote import requests as _requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry try: import aiohttp except ImportError: aiohttp = None logger = logging.getLogger(__name__) DEFAULT_MGMT_UA = "codex_cli_rs/0.76.0 (Debian 13.0.0; x86_64) WindowsTerminal" def _mgmt_headers(token: str) -> Dict[str, str]: return {"Authorization": f"Bearer {token}", "Accept": "application/json"} def _build_session(proxy: str = "") -> _requests.Session: s = _requests.Session() retry = Retry(total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504]) adapter = HTTPAdapter(max_retries=retry) s.mount("https://", adapter) s.mount("http://", adapter) if proxy: s.proxies = {"http": proxy, "https": proxy} return s def _get_item_type(item: Dict[str, Any]) -> str: return str(item.get("type") or item.get("typo") or "") def _safe_json(text: str) -> Dict[str, Any]: try: return json.loads(text) except Exception: return {} def _extract_account_id(item: Dict[str, Any]) -> Optional[str]: for key in ("chatgpt_account_id", "chatgptAccountId", "account_id", "accountId"): val = item.get(key) if val: return str(val) return None def _parse_time_to_epoch(raw: Any) -> float: text = str(raw or "").strip() if not text: return 0.0 iso_text = text[:-1] + "+00:00" if text.endswith("Z") else text try: return datetime.fromisoformat(iso_text).timestamp() except Exception: pass for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"): try: return datetime.strptime(text, fmt).timestamp() except Exception: continue return 0.0 class PoolMaintainer: def __init__( self, cpa_base_url: str, cpa_token: str, target_type: str = "codex", min_candidates: int = 800, used_percent_threshold: int = 95, user_agent: str = DEFAULT_MGMT_UA, ): self.base_url = cpa_base_url.rstrip("/") self.token = cpa_token self.target_type = target_type self.min_candidates = min_candidates self.used_percent_threshold = used_percent_threshold self.user_agent = user_agent def fetch_auth_files(self, timeout: int = 15) -> List[Dict[str, Any]]: resp = _requests.get( f"{self.base_url}/v0/management/auth-files", headers=_mgmt_headers(self.token), timeout=timeout, ) resp.raise_for_status() raw = resp.json() data = raw if isinstance(raw, dict) else {} files = data.get("files", []) return files if isinstance(files, list) else [] def get_pool_status(self, timeout: int = 15) -> Dict[str, Any]: try: files = self.fetch_auth_files(timeout) candidates = [f for f in files if _get_item_type(f).lower() == self.target_type.lower()] total = len(files) cand_count = len(candidates) return { "total": total, "candidates": cand_count, "error_count": max(0, total - cand_count), "threshold": self.min_candidates, "healthy": cand_count >= self.min_candidates, "percent": round(cand_count / self.min_candidates * 100, 1) if self.min_candidates > 0 else 100, "last_checked": time.strftime("%Y-%m-%d %H:%M:%S"), "error": None, } except Exception as e: return { "total": 0, "candidates": 0, "error_count": 0, "threshold": self.min_candidates, "healthy": False, "percent": 0, "last_checked": time.strftime("%Y-%m-%d %H:%M:%S"), "error": str(e), } def test_connection(self, timeout: int = 10) -> Dict[str, Any]: try: files = self.fetch_auth_files(timeout) candidates = [f for f in files if _get_item_type(f).lower() == self.target_type.lower()] return { "ok": True, "total": len(files), "candidates": len(candidates), "message": f"连接成功，共 {len(files)} 个账号，{len(candidates)} 个 {self.target_type} 账号", } except Exception as e: return {"ok": False, "total": 0, "candidates": 0, "message": f"连接失败: {e}"} async def probe_accounts_async( self, workers: int = 20, timeout: int = 10, retries: int = 1, ) -> Dict[str, Any]: if aiohttp is None: raise RuntimeError("需要安装 aiohttp: pip install aiohttp") files = self.fetch_auth_files(timeout) candidates = [f for f in files if _get_item_type(f).lower() == self.target_type.lower()] if not candidates: return {"total": len(files), "candidates": 0, "invalid": [], "files": files} semaphore = asyncio.Semaphore(max(1, workers)) connector = aiohttp.TCPConnector(limit=max(1, workers)) client_timeout = aiohttp.ClientTimeout(total=max(1, timeout)) async def probe_one(session: aiohttp.ClientSession, item: Dict[str, Any]) -> Dict[str, Any]: auth_index = item.get("auth_index") name = item.get("name") or item.get("id") result = { "name": name, "auth_index": auth_index, "invalid_401": False, "invalid_used_percent": False, "used_percent": None, "error": None, } if not auth_index: result["error"] = "missing auth_index" return result account_id = _extract_account_id(item) call_header = { "Authorization": "Bearer $TOKEN$", "Content-Type": "application/json", "User-Agent": self.user_agent, } if account_id: call_header["Chatgpt-Account-Id"] = account_id payload = { "authIndex": auth_index, "method": "GET", "url": "https://chatgpt.com/backend-api/wham/usage", "header": call_header, } for attempt in range(retries + 1): try: async with semaphore: async with session.post( f"{self.base_url}/v0/management/api-call", headers={**_mgmt_headers(self.token), "Content-Type": "application/json"}, json=payload, timeout=timeout, ) as resp: text = await resp.text() if resp.status >= 400: raise RuntimeError(f"HTTP {resp.status}: {text[:200]}") data = _safe_json(text) sc = data.get("status_code") result["invalid_401"] = sc == 401 if sc == 200: try: body_data = _safe_json(data.get("body", "")) used_pct = (body_data.get("rate_limit", {}).get("primary_window", {}).get("used_percent")) if used_pct is not None: result["used_percent"] = used_pct result["invalid_used_percent"] = used_pct >= self.used_percent_threshold except Exception: pass return result except Exception as e: result["error"] = str(e) if attempt >= retries: return result return result async def delete_one(session: aiohttp.ClientSession, name: str) -> Dict[str, Any]: encoded = quote(name, safe="") try: async with semaphore: async with session.delete( f"{self.base_url}/v0/management/auth-files?name={encoded}", headers=_mgmt_headers(self.token), timeout=timeout, ) as resp: text = await resp.text() data = _safe_json(text) ok = resp.status == 200 and data.get("status") == "ok" return {"name": name, "deleted": ok} except Exception: return {"name": name, "deleted": False} invalid_list = [] async with aiohttp.ClientSession(connector=connector, timeout=client_timeout, trust_env=True) as session: tasks = [asyncio.create_task(probe_one(session, item)) for item in candidates] for task in asyncio.as_completed(tasks): result = await task if result.get("invalid_401") or result.get("invalid_used_percent"): invalid_list.append(result) return { "total": len(files), "candidates": len(candidates), "invalid": invalid_list, "files": files, } async def clean_invalid_async(self, workers: int = 20, timeout: int = 10, retries: int = 1) -> Dict[str, Any]: if aiohttp is None: raise RuntimeError("需要安装 aiohttp: pip install aiohttp") probe_result = await self.probe_accounts_async(workers, timeout, retries) invalid = probe_result["invalid"] names = [str(r["name"]) for r in invalid if r.get("name")] deleted_ok = 0 deleted_fail = 0 if names: semaphore = asyncio.Semaphore(max(1, workers)) connector = aiohttp.TCPConnector(limit=max(1, workers)) client_timeout = aiohttp.ClientTimeout(total=max(1, timeout)) async with aiohttp.ClientSession(connector=connector, timeout=client_timeout, trust_env=True) as session: async def do_delete(name: str) -> bool: encoded = quote(name, safe="") try: async with semaphore: async with session.delete( f"{self.base_url}/v0/management/auth-files?name={encoded}", headers=_mgmt_headers(self.token), timeout=timeout, ) as resp: text = await resp.text() data = _safe_json(text) return resp.status == 200 and data.get("status") == "ok" except Exception: return False tasks = [asyncio.create_task(do_delete(n)) for n in names] for task in asyncio.as_completed(tasks): if await task: deleted_ok += 1 else: deleted_fail += 1 return { "total": probe_result["total"], "candidates": probe_result["candidates"], "invalid_count": len(invalid), "deleted_ok": deleted_ok, "deleted_fail": deleted_fail, } def probe_and_clean_sync(self, workers: int = 20, timeout: int = 10, retries: int = 1) -> Dict[str, Any]: return asyncio.run(self.clean_invalid_async(workers, timeout, retries)) def calculate_gap(self, current_candidates: Optional[int] = None) -> int: if current_candidates is None: status = self.get_pool_status() if status.get("error"): raise RuntimeError(f"CPA 池状态查询失败: {status['error']}") current_candidates = status["candidates"] gap = self.min_candidates - current_candidates return max(0, gap) def upload_token(self, filename: str, token_data: Dict[str, Any], proxy: str = "") -> bool: if not self.base_url or not self.token: return False content = json.dumps(token_data, ensure_ascii=False).encode("utf-8") files = {"file": (filename, content, "application/json")} headers = {"Authorization": f"Bearer {self.token}"} with _build_session(proxy) as session: for attempt in range(3): try: resp = session.post( f"{self.base_url}/v0/management/auth-files", files=files, headers=headers, verify=False, timeout=30, ) if resp.status_code in (200, 201, 204): return True except Exception: pass if attempt < 2: time.sleep(2 ** attempt) return False class Sub2ApiMaintainer: """Sub2Api 平台池维护 — 通过 Admin API 管理账号池""" def __init__( self, base_url: str, bearer_token: str, min_candidates: int = 200, email: str = "", password: str = "", ): self.base_url = base_url.rstrip("/") self.bearer_token = bearer_token self.min_candidates = min_candidates self.email = email self.password = password self._auth_lock = threading.Lock() def _headers(self) -> Dict[str, str]: return { "Authorization": f"Bearer {self.bearer_token}", "Accept": "application/json", "Content-Type": "application/json", } def _login(self) -> str: with _build_session() as session: resp = session.post( f"{self.base_url}/api/v1/auth/login", json={"email": self.email, "password": self.password}, timeout=15, ) resp.raise_for_status() data = resp.json() token = ( data.get("token") or data.get("access_token") or (data.get("data") or {}).get("token") or (data.get("data") or {}).get("access_token") or "" ) if token: self.bearer_token = token return token def _request(self, method: str, path: str, **kwargs) -> _requests.Response: kwargs.setdefault("timeout", 15) url = f"{self.base_url}{path}" with _build_session() as session: resp = session.request(method, url, headers=self._headers(), **kwargs) if resp.status_code == 401 and self.email and self.password: current_token = self.bearer_token with self._auth_lock: if self.bearer_token == current_token: self._login() refreshed_token = self.bearer_token if refreshed_token or self.bearer_token != current_token: resp = session.request(method, url, headers=self._headers(), **kwargs) return resp resp = session.request(method, url, headers=self._headers(), **kwargs) return resp def get_dashboard_stats(self, timeout: int = 15) -> Dict[str, Any]: resp = self._request( "GET", "/api/v1/admin/dashboard/stats", params={"timezone": "Asia/Shanghai"}, timeout=timeout, ) resp.raise_for_status() data = resp.json() return data.get("data") if isinstance(data.get("data"), dict) else data def list_accounts( self, page: int = 1, page_size: int = 100, timeout: int = 15, ) -> Dict[str, Any]: params = { "page": page, "page_size": page_size, "platform": "openai", "type": "oauth", } resp = self._request( "GET", "/api/v1/admin/accounts", params=params, timeout=timeout, ) resp.raise_for_status() data = resp.json() return data.get("data") if isinstance(data.get("data"), dict) else data def _list_all_accounts(self, timeout: int = 15, page_size: int = 100) -> List[Dict[str, Any]]: all_accounts: List[Dict[str, Any]] = [] page = 1 while True: data = self.list_accounts(page=page, page_size=page_size, timeout=timeout) items = data.get("items") or [] if not isinstance(items, list): items = [] all_accounts.extend([i for i in items if isinstance(i, dict)]) if not items or len(items) < page_size: break total = data.get("total") if isinstance(total, int) and total > 0 and len(all_accounts) >= total: break page += 1 return all_accounts def _account_identity(self, item: Dict[str, Any]) -> Dict[str, str]: email = "" rt = "" extra = item.get("extra") if isinstance(extra, dict): email = str(extra.get("email") or "").strip().lower() if not email: name = str(item.get("name") or "").strip().lower() if "@" in name: email = name creds = item.get("credentials") if isinstance(creds, dict): rt = str(creds.get("refresh_token") or "").strip() return {"email": email, "refresh_token": rt} @staticmethod def _account_sort_key(item: Dict[str, Any]) -> tuple[float, int]: updated = _parse_time_to_epoch(item.get("updated_at") or item.get("updatedAt")) try: item_id = int(item.get("id") or 0) except (TypeError, ValueError): item_id = 0 return (updated, item_id) @staticmethod def _normalize_account_id(raw: Any) -> Optional[int]: try: account_id = int(raw) except (TypeError, ValueError): return None if account_id <= 0: return None return account_id @staticmethod def _is_abnormal_status(status: Any) -> bool: return str(status or "").strip().lower() in ("error", "disabled") def _build_dedupe_plan(self, all_accounts: List[Dict[str, Any]], details_limit: int = 120) -> Dict[str, Any]: id_to_account: Dict[int, Dict[str, Any]] = {} parent: Dict[int, int] = {} key_to_ids: Dict[str, List[int]] = {} for item in all_accounts: acc_id = self._normalize_account_id(item.get("id")) if acc_id is None: continue id_to_account[acc_id] = item parent[acc_id] = acc_id identity = self._account_identity(item) email = identity["email"] refresh_token = identity["refresh_token"] if email: key_to_ids.setdefault(f"email:{email}", []).append(acc_id) if refresh_token: key_to_ids.setdefault(f"rt:{refresh_token}", []).append(acc_id) def find(x: int) -> int: root = x while parent[root] != root: root = parent[root] while parent[x] != x: nxt = parent[x] parent[x] = root x = nxt return root def union(a: int, b: int) -> None: ra = find(a) rb = find(b) if ra != rb: parent[rb] = ra for ids in key_to_ids.values(): if len(ids) > 1: head = ids[0] for acc_id in ids[1:]: union(head, acc_id) components: Dict[int, List[int]] = {} for acc_id in id_to_account.keys(): root = find(acc_id) components.setdefault(root, []).append(acc_id) duplicate_groups = [ids for ids in components.values() if len(ids) > 1] delete_ids: List[int] = [] group_details: List[Dict[str, Any]] = [] for group_ids in duplicate_groups: group_items = [id_to_account[i] for i in group_ids] keep_item = max(group_items, key=self._account_sort_key) keep_id = self._normalize_account_id(keep_item.get("id")) or 0 group_delete_ids = sorted([i for i in group_ids if i != keep_id], reverse=True) delete_ids.extend(group_delete_ids) if len(group_details) < details_limit: emails_set = set() for it in group_items: identity = self._account_identity(it) if identity["email"]: emails_set.add(identity["email"]) emails = sorted(emails_set) group_details.append({ "keep_id": keep_id, "delete_ids": group_delete_ids, "size": len(group_ids), "emails": emails, }) return { "duplicate_groups": len(duplicate_groups), "duplicate_accounts": sum(len(g) for g in duplicate_groups), "delete_ids": delete_ids, "groups_preview": group_details, "truncated_groups": max(0, len(duplicate_groups) - len(group_details)), } def list_account_inventory(self, timeout: int = 15) -> Dict[str, Any]: all_accounts = self._list_all_accounts(timeout=timeout, page_size=100) dedupe_plan = self._build_dedupe_plan( all_accounts, details_limit=max(1, len(all_accounts)), ) duplicate_delete_ids = { int(account_id) for account_id in (dedupe_plan.get("delete_ids") or []) if isinstance(account_id, int) } duplicate_map: Dict[int, Dict[str, Any]] = {} for group in dedupe_plan.get("groups_preview") or []: keep_id = self._normalize_account_id(group.get("keep_id")) delete_ids = [ account_id for account_id in ( self._normalize_account_id(item) for item in (group.get("delete_ids") or []) ) if account_id is not None ] group_ids = ([keep_id] if keep_id is not None else []) + delete_ids group_size = max(1, int(group.get("size") or len(group_ids) or 1)) emails = [str(email).strip().lower() for email in (group.get("emails") or []) if str(email).strip()] for account_id in group_ids: duplicate_map[account_id] = { "group_size": group_size, "keep_id": keep_id, "delete_candidate": account_id in duplicate_delete_ids, "emails": emails, } items: List[Dict[str, Any]] = [] abnormal_count = 0 for raw_item in sorted(all_accounts, key=self._account_sort_key, reverse=True): account_id = self._normalize_account_id(raw_item.get("id")) if account_id is None: continue identity = self._account_identity(raw_item) status = str(raw_item.get("status") or "").strip().lower() or "unknown" if self._is_abnormal_status(status): abnormal_count += 1 duplicate_info = duplicate_map.get(account_id) or {} items.append({ "id": account_id, "name": str(raw_item.get("name") or "").strip(), "email": identity.get("email") or str(raw_item.get("name") or "").strip(), "status": status, "updated_at": raw_item.get("updated_at") or raw_item.get("updatedAt") or "", "created_at": raw_item.get("created_at") or raw_item.get("createdAt") or "", "is_duplicate": bool(duplicate_info), "duplicate_group_size": int(duplicate_info.get("group_size") or 0), "duplicate_keep": duplicate_info.get("keep_id") == account_id, "duplicate_delete_candidate": bool(duplicate_info.get("delete_candidate")), "duplicate_emails": duplicate_info.get("emails") or [], }) return { "total": len(items), "error_count": abnormal_count, "duplicate_groups": int(dedupe_plan.get("duplicate_groups", 0)), "duplicate_accounts": int(dedupe_plan.get("duplicate_accounts", 0)), "items": items, } def _refresh_accounts_parallel(self, account_ids: List[int], timeout: int = 30, workers: int = 8) -> Dict[str, List[int]]: success_ids: List[int] = [] failed_ids: List[int] = [] ids = [i for i in account_ids if isinstance(i, int) and i > 0] if not ids: return {"success_ids": success_ids, "failed_ids": failed_ids} pool_workers = max(1, min(workers, 16, len(ids))) with ThreadPoolExecutor(max_workers=pool_workers) as executor: future_to_id = { executor.submit(self.refresh_account, account_id, timeout=timeout): account_id for account_id in ids } for future in as_completed(future_to_id): account_id = future_to_id[future] try: ok = bool(future.result()) except Exception: ok = False if ok: success_ids.append(account_id) else: failed_ids.append(account_id) return {"success_ids": success_ids, "failed_ids": failed_ids} def _delete_accounts_parallel(self, account_ids: List[int], timeout: int = 15, workers: int = 12) -> Dict[str, Any]: deleted_ok_ids: List[int] = [] failed_ids: List[int] = [] unique_ids = sorted({i for i in account_ids if isinstance(i, int) and i > 0}, reverse=True) if not unique_ids: return {"deleted_ok": 0, "deleted_fail": 0, "deleted_ok_ids": deleted_ok_ids, "failed_ids": failed_ids} pool_workers = max(1, min(workers, 24, len(unique_ids))) with ThreadPoolExecutor(max_workers=pool_workers) as executor: future_to_id = { executor.submit(self.delete_account, account_id, timeout=timeout): account_id for account_id in unique_ids } for future in as_completed(future_to_id): account_id = future_to_id[future] try: ok = bool(future.result()) except Exception: ok = False if ok: deleted_ok_ids.append(account_id) else: failed_ids.append(account_id) return { "deleted_ok": len(deleted_ok_ids), "deleted_fail": len(failed_ids), "deleted_ok_ids": deleted_ok_ids, "failed_ids": failed_ids, } def dedupe_duplicate_accounts(self, timeout: int = 15, dry_run: bool = True, details_limit: int = 120) -> Dict[str, Any]: """ 清理 Sub2Api 中 OpenAI OAuth 重复账号（按 email 或 refresh_token 判重）。 - 同一连通重复组保留“最新”账号（updated_at 优先，其次 id 最大）。 - dry_run=True 时仅预览，不执行删除。 """ all_accounts = self._list_all_accounts(timeout=timeout, page_size=100) dedupe_plan = self._build_dedupe_plan(all_accounts, details_limit=details_limit) delete_ids = dedupe_plan["delete_ids"] deleted_ok = 0 deleted_fail = 0 failed_ids: List[int] = [] if not dry_run and delete_ids: delete_result = self._delete_accounts_parallel(delete_ids, timeout=timeout, workers=12) deleted_ok = int(delete_result.get("deleted_ok", 0)) deleted_fail = int(delete_result.get("deleted_fail", 0)) failed_ids = list(delete_result.get("failed_ids") or []) return { "dry_run": dry_run, "total": len(all_accounts), "duplicate_groups": int(dedupe_plan["duplicate_groups"]), "duplicate_accounts": int(dedupe_plan["duplicate_accounts"]), "to_delete": len(delete_ids), "deleted_ok": deleted_ok, "deleted_fail": deleted_fail, "failed_delete_ids": failed_ids[:200], "groups_preview": dedupe_plan["groups_preview"], "truncated_groups": int(dedupe_plan["truncated_groups"]), } def probe_accounts(self, account_ids: List[int], timeout: int = 30) -> Dict[str, Any]: ids = sorted({ account_id for account_id in ( self._normalize_account_id(item) for item in (account_ids or []) ) if account_id is not None }) if not ids: return { "requested": 0, "refreshed_ok": 0, "refreshed_fail": 0, "recovered": 0, "still_abnormal": 0, "details": [], } before_status = self._list_accounts_by_ids(ids, timeout=timeout) refresh_result = self._refresh_accounts_parallel(ids, timeout=max(30, timeout), workers=8) success_ids = set(refresh_result.get("success_ids") or []) failed_ids = set(refresh_result.get("failed_ids") or []) if success_ids: time.sleep(2) after_status = self._list_accounts_by_ids(ids, timeout=timeout) recovered_ids: List[int] = [] abnormal_after_ids: List[int] = [] details: List[Dict[str, Any]] = [] for account_id in ids: before = str(before_status.get(account_id) or "unknown").strip().lower() after = str(after_status.get(account_id) or before or "unknown").strip().lower() if self._is_abnormal_status(before) and not self._is_abnormal_status(after): recovered_ids.append(account_id) if self._is_abnormal_status(after): abnormal_after_ids.append(account_id) if len(details) < 200: details.append({ "id": account_id, "before_status": before, "after_status": after, "refresh_ok": account_id in success_ids, }) return { "requested": len(ids), "refreshed_ok": len(success_ids), "refreshed_fail": len(failed_ids), "recovered": len(recovered_ids), "still_abnormal": len(abnormal_after_ids), "details": details, } def delete_accounts_batch(self, account_ids: List[int], timeout: int = 15) -> Dict[str, Any]: ids = [ account_id for account_id in ( self._normalize_account_id(item) for item in (account_ids or []) ) if account_id is not None ] delete_result = self._delete_accounts_parallel(ids, timeout=timeout, workers=12) return { "requested": len({*ids}), "deleted_ok": int(delete_result.get("deleted_ok", 0)), "deleted_fail": int(delete_result.get("deleted_fail", 0)), "deleted_ok_ids": list(delete_result.get("deleted_ok_ids") or []), "failed_ids": list(delete_result.get("failed_ids") or []), } def handle_exception_accounts( self, account_ids: Optional[List[int]] = None, timeout: int = 30, delete_unresolved: bool = True, ) -> Dict[str, Any]: requested_ids = [ account_id for account_id in ( self._normalize_account_id(item) for item in (account_ids or []) ) if account_id is not None ] if requested_ids: current_status = self._list_accounts_by_ids(requested_ids, timeout=timeout) target_ids = [ account_id for account_id in requested_ids if self._is_abnormal_status(current_status.get(account_id)) ] skipped_non_abnormal = max(0, len(set(requested_ids)) - len(target_ids)) else: all_accounts = self._list_all_accounts(timeout=timeout, page_size=100) target_ids = [ account_id for account_id in ( self._normalize_account_id(item.get("id")) for item in all_accounts if self._is_abnormal_status(item.get("status")) ) if account_id is not None ] skipped_non_abnormal = 0 unique_target_ids = sorted(set(target_ids)) if not unique_target_ids: return { "requested": len(set(requested_ids)) if requested_ids else 0, "targeted": 0, "refreshed_ok": 0, "refreshed_fail": 0, "recovered": 0, "remaining_abnormal": 0, "deleted_ok": 0, "deleted_fail": 0, "skipped_non_abnormal": skipped_non_abnormal, } refresh_result = self._refresh_accounts_parallel(unique_target_ids, timeout=max(30, timeout), workers=8) if refresh_result.get("success_ids"): time.sleep(2) after_status = self._list_accounts_by_ids(unique_target_ids, timeout=timeout) remaining_abnormal_ids = [ account_id for account_id in unique_target_ids if self._is_abnormal_status(after_status.get(account_id)) ] remaining_abnormal_set = set(remaining_abnormal_ids) recovered_ids = [ account_id for account_id in unique_target_ids if account_id not in remaining_abnormal_set ] delete_result = { "deleted_ok": 0, "deleted_fail": 0, "deleted_ok_ids": [], "failed_ids": [], } if delete_unresolved and remaining_abnormal_ids: delete_result = self._delete_accounts_parallel(remaining_abnormal_ids, timeout=timeout, workers=12) return { "requested": len(set(requested_ids)) if requested_ids else len(unique_target_ids), "targeted": len(unique_target_ids), "refreshed_ok": len(refresh_result.get("success_ids") or []), "refreshed_fail": len(refresh_result.get("failed_ids") or []), "recovered": len(recovered_ids), "remaining_abnormal": len(remaining_abnormal_ids), "deleted_ok": int(delete_result.get("deleted_ok", 0)), "deleted_fail": int(delete_result.get("deleted_fail", 0)), "deleted_ok_ids": list(delete_result.get("deleted_ok_ids") or []), "failed_ids": list(delete_result.get("failed_ids") or []), "skipped_non_abnormal": skipped_non_abnormal, } def refresh_account(self, account_id: int, timeout: int = 30) -> bool: try: resp = self._request( "POST", f"/api/v1/admin/accounts/{account_id}/refresh", timeout=timeout, ) return resp.status_code in (200, 201) except Exception: return False def delete_account(self, account_id: int, timeout: int = 15) -> bool: try: resp = self._request( "DELETE", f"/api/v1/admin/accounts/{account_id}", timeout=timeout, ) return resp.status_code in (200, 204) except Exception: return False def get_pool_status(self, timeout: int = 15) -> Dict[str, Any]: try: all_accounts = self._list_all_accounts(timeout=timeout, page_size=100) error = sum( 1 for account in all_accounts if self._is_abnormal_status(account.get("status")) ) total = len(all_accounts) normal = max(0, total - error) return { "total": total, "candidates": normal, "error_count": error, "threshold": self.min_candidates, "healthy": normal >= self.min_candidates, "percent": round(normal / self.min_candidates * 100, 1) if self.min_candidates > 0 else 100, "last_checked": time.strftime("%Y-%m-%d %H:%M:%S"), "error": None, } except Exception as e: return { "total": 0, "candidates": 0, "error_count": 0, "threshold": self.min_candidates, "healthy": False, "percent": 0, "last_checked": time.strftime("%Y-%m-%d %H:%M:%S"), "error": str(e), } def test_connection(self, timeout: int = 10) -> Dict[str, Any]: try: status = self.get_pool_status(timeout) total = int(status.get("total", 0)) normal = int(status.get("candidates", 0)) error = int(status.get("error_count", 0)) return { "ok": True, "total": total, "normal": normal, "error": error, "message": f"连接成功，共 {total} 个账号，{normal} 正常，{error} 异常", } except Exception as e: return {"ok": False, "total": 0, "normal": 0, "error": 0, "message": f"连接失败: {e}"} def _list_accounts_by_ids( self, ids: List[int], timeout: int = 15, ) -> Dict[int, str]: """查询指定 ID 的账号当前状态，返回 {id: status}""" result: Dict[int, str] = {} id_set = set(ids) page = 1 while id_set: data = self.list_accounts(page=page, page_size=100, timeout=timeout) items = data.get("items") or [] if not items: break for item in items: aid = item.get("id") if aid in id_set: result[aid] = str(item.get("status", "")) id_set.discard(aid) total = data.get("total", 0) if page * 100 >= total or len(items) < 100: break page += 1 return result def probe_and_clean_sync(self, timeout: int = 15, actions: Optional[Dict[str, bool]] = None) -> Dict[str, Any]: action_flags = { "refresh_abnormal_accounts": bool((actions or {}).get("refresh_abnormal_accounts", True)), "delete_abnormal_accounts": bool((actions or {}).get("delete_abnormal_accounts", True)), "dedupe_duplicate_accounts": bool((actions or {}).get("dedupe_duplicate_accounts", True)), } started = time.time() all_accounts = self._list_all_accounts(timeout=timeout, page_size=100) error_accounts = [ account for account in all_accounts if self._is_abnormal_status(account.get("status")) ] error_ids = [ self._normalize_account_id(acc.get("id")) for acc in error_accounts ] error_ids = [i for i in error_ids if i is not None] initial_error_ids = set(error_ids) refresh_result = {"success_ids": [], "failed_ids": []} if action_flags["refresh_abnormal_accounts"] and error_ids: refresh_result = self._refresh_accounts_parallel(error_ids, timeout=30, workers=8) refreshed_ids = list(refresh_result.get("success_ids") or []) refresh_failed_ids = list(refresh_result.get("failed_ids") or []) current_accounts = all_accounts current_error_ids = set(initial_error_ids) if refreshed_ids: time.sleep(2) if action_flags["refresh_abnormal_accounts"] and (error_ids or refreshed_ids): current_accounts = self._list_all_accounts(timeout=timeout, page_size=100) current_error_ids = { int(acc_id) for acc_id in ( self._normalize_account_id(account.get("id")) for account in current_accounts if self._is_abnormal_status(account.get("status")) ) if isinstance(acc_id, int) } recovered = len(initial_error_ids - current_error_ids) dedupe_plan = { "duplicate_groups": 0, "duplicate_accounts": 0, "delete_ids": [], "groups_preview": [], "truncated_groups": 0, } duplicate_delete_ids: List[int] = [] if action_flags["dedupe_duplicate_accounts"]: dedupe_plan = self._build_dedupe_plan(current_accounts, details_limit=120) duplicate_delete_ids = [int(i) for i in dedupe_plan["delete_ids"] if isinstance(i, int)] normal_count = len(current_accounts) - len(current_error_ids) delete_targets: set[int] = set() if action_flags["delete_abnormal_accounts"]: delete_targets.update(current_error_ids) if action_flags["dedupe_duplicate_accounts"]: delete_targets.update(duplicate_delete_ids) delete_result = self._delete_accounts_parallel(sorted(delete_targets, reverse=True), timeout=timeout, workers=12) deleted_ok = int(delete_result.get("deleted_ok", 0)) deleted_fail = int(delete_result.get("deleted_fail", 0)) deleted_ok_ids = set(int(i) for i in (delete_result.get("deleted_ok_ids") or []) if isinstance(i, int)) deleted_from_error = len(deleted_ok_ids & set(current_error_ids)) deleted_from_duplicate = len(deleted_ok_ids & set(duplicate_delete_ids)) elapsed_ms = int((time.time() - started) * 1000) return { "actions": action_flags, "total": len(current_accounts), "normal": normal_count, "initial_error_count": len(initial_error_ids), "error_count": len(current_error_ids), "refreshed": recovered, "refresh_attempted": len(error_ids) if action_flags["refresh_abnormal_accounts"] else 0, "refresh_failed": len(refresh_failed_ids), "deleted_ok": deleted_ok, "deleted_fail": deleted_fail, "duplicate_groups": int(dedupe_plan["duplicate_groups"]), "duplicate_accounts": int(dedupe_plan["duplicate_accounts"]), "duplicate_to_delete": len(duplicate_delete_ids), "deleted_from_error": deleted_from_error, "deleted_from_duplicate": deleted_from_duplicate, "duration_ms": elapsed_ms, } def calculate_gap(self, current_candidates: Optional[int] = None) -> int: if current_candidates is None: status = self.get_pool_status() if status.get("error"): raise RuntimeError(f"Sub2Api 池状态查询失败: {status['error']}") current_candidates = status["candidates"] return max(0, self.min_candidates - current_candidates)