utils/core_engine.py

"""
调度层
配置热加载、CPA 仓管逻辑、主循环、RegEngine 控制类。
邮箱逻辑 → mail_service.py
注册流程 → register.py
配置变量 → config.py
"""

import argparse
import asyncio
import builtins
import io
import json
import os
import random
import re
import threading
import time
import string
import yaml
import urllib.parse
from concurrent.futures import ThreadPoolExecutor
from typing import Any, Optional, Tuple
from curl_cffi import requests, CurlMime
import queue
from datetime import datetime, timezone, timedelta
from utils.email_providers import mail_service
from utils import config as cfg
from utils import db_manager
from utils.config import reload_all_configs, ts, format_docker_url
from utils.email_providers.mail_service import mask_email
from utils.register import run, refresh_oauth_token as _refresh_oauth_token
from utils.proxy_manager import smart_switch_node
from utils.integrations.sub2api_client import Sub2APIClient
from utils.integrations.tg_notifier import send_tg_msg_sync

_stats_lock = threading.Lock()
sub_fail_counts = {}
_heal_lock = threading.Lock()
DEFAULT_CLIPROXY_UA = "codex_cli_rs/0.76.0 (Debian 13.0.0; x86_64) WindowsTerminal"
run_stats = {
    "success": 0,
    "failed": 0,
    "retries": 0,
    "start_time": 0,
    "target": 0,
    "pwd_blocked": 0,
    "phone_verify": 0
}
KNOWN_CLIPROXY_ERROR_LABELS = {
    "usage_limit_reached":  "周限额已耗尽",
    "account_deactivated":  "账号已停用",
    "insufficient_quota":   "额度不足",
    "invalid_api_key":      "凭证无效",
    "unsupported_region":   "地区不支持",
}

log_queue = queue.Queue(maxsize=500)
_orig_print  = builtins.print
_thread_local = threading.local()
_print_lock   = threading.Lock()


def web_print(*args, **kwargs):
    if "file" in kwargs and kwargs["file"] is not None:
        with _print_lock:
            _orig_print(*args, **kwargs)
        return
    if not hasattr(_thread_local, "buffer"):
        _thread_local.buffer = ""
    tmp = io.StringIO()
    _orig_print(*args, file=tmp, **kwargs)
    _thread_local.buffer += tmp.getvalue()
    if _thread_local.buffer.endswith("\n"):
        with _print_lock:
            msg = _thread_local.buffer.lstrip("\n")
            if msg and msg.strip() != ".":
                try:
                    log_queue.put_nowait(msg.strip())
                except queue.Full:
                    pass
        _thread_local.buffer = ""


builtins.print = web_print

def _load_dotenv(path: str = ".env") -> None:
    if not os.path.exists(path):
        return
    try:
        with open(path, "r", encoding="utf-8") as handle:
            for raw in handle:
                line = raw.strip()
                if not line or line.startswith("#") or "=" not in line:
                    continue
                key, value = line.split("=", 1)
                key = key.strip()
                if not key or key in os.environ:
                    continue
                value = value.strip()
                if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}:
                    value = value[1:-1]
                os.environ[key] = value
    except Exception:
        pass


_load_dotenv()

def _normalize_cpa_auth_files_url(api_url: str) -> str:
    normalized = (api_url or "").strip().rstrip("/")
    lower = normalized.lower()
    if not normalized:
        return ""
    if lower.endswith("/auth-files"):
        return normalized
    if lower.endswith("/v0/management") or lower.endswith("/management"):
        return f"{normalized}/auth-files"
    if lower.endswith("/v0"):
        return f"{normalized}/management/auth-files"
    return f"{normalized}/v0/management/auth-files"


def set_cpa_auth_file_status(
    api_url: str, api_token: str, filename: str, disabled: bool = True
) -> bool:
    status_url = f"{_normalize_cpa_auth_files_url(api_url)}/status"
    try:
        res = requests.patch(
            status_url,
            headers={"Authorization": f"Bearer {api_token}", "Content-Type": "application/json"},
            json={"name": filename, "disabled": disabled},
            timeout=15, impersonate="chrome110",
        )
        if res.status_code in (200, 204):
            return True
        print(f"[{ts()}] [ERROR] 切换凭证状态失败 (HTTP {res.status_code}): {res.text}")
        return False
    except Exception as e:
        print(f"[{ts()}] [ERROR] 切换凭证状态异常: {e}")
        return False


def upload_to_cpa_integrated(
    token_data: dict, api_url: str, api_token: str, custom_filename: str = None
) -> Tuple[bool, str]:
    upload_url = _normalize_cpa_auth_files_url(api_url)
    filename   = custom_filename or f"{token_data.get('email', 'unknown')}.json"
    file_content = json.dumps(token_data, ensure_ascii=False, indent=2).encode("utf-8")
    try:
        mime = CurlMime()
        mime.addpart(name="file", data=file_content, filename=filename,
                     content_type="application/json")
        resp = requests.post(
            upload_url, multipart=mime,
            headers={"Authorization": f"Bearer {api_token}"},
            timeout=30, impersonate="chrome110",
        )
        if resp.status_code in (200, 201):
            return True, "上传成功"
        if resp.status_code in (404, 405, 415):
            raw_url = f"{upload_url}?name={urllib.parse.quote(filename)}"
            fb = requests.post(
                raw_url, data=file_content,
                headers={"Authorization": f"Bearer {api_token}",
                         "Content-Type": "application/json"},
                timeout=30, impersonate="chrome110",
            )
            if fb.status_code in (200, 201):
                return True, "上传成功"
            resp = fb
        return False, f"HTTP {resp.status_code}"
    except Exception as e:
        return False, str(e)


def _decode_possible_json_payload(payload: Any) -> Any:
    if isinstance(payload, str):
        text = payload.strip()
        if not text:
            return payload
        try:
            return json.loads(text)
        except Exception:
            return payload
    return payload


def _extract_remaining_percent(window_info: Any) -> Optional[float]:
    if not isinstance(window_info, dict):
        return None
    remaining_percent = window_info.get("remaining_percent")
    if isinstance(remaining_percent, (int, float)):
        return max(0.0, min(100.0, float(remaining_percent)))
    used_percent = window_info.get("used_percent")
    if isinstance(used_percent, (int, float)):
        return max(0.0, min(100.0, 100.0 - float(used_percent)))
    return None


def _format_percent(value: float) -> str:
    n = round(float(value), 2)
    return str(int(n)) if n.is_integer() else f"{n:.2f}".rstrip("0").rstrip(".")


def _format_known_cliproxy_error(error_type: str) -> str:
    label = KNOWN_CLIPROXY_ERROR_LABELS.get(error_type)
    return f"{label} ({error_type})" if label else f"错误类型: {error_type}"


def _extract_rate_limit_reason(
    rate_info: Any, key: str, min_remaining_weekly_percent: int = 0
) -> Optional[str]:
    if not isinstance(rate_info, dict):
        return None
    if rate_info.get("allowed") is False or rate_info.get("limit_reached") is True:
        label = {"rate_limit": "周限额已耗尽", "code_review_rate_limit": "代码审查周限额已耗尽"}.get(
            key, f"{key} 已耗尽"
        )
        return f"{label}（allowed={rate_info.get('allowed')}, limit_reached={rate_info.get('limit_reached')}）"
    if key == "rate_limit" and min_remaining_weekly_percent > 0:
        pct = _extract_remaining_percent(rate_info.get("primary_window"))
        if pct is not None and pct < min_remaining_weekly_percent:
            return f"周限额剩余 {_format_percent(pct)}%，低于阈值 {min_remaining_weekly_percent}%"
    return None


def _extract_cliproxy_failure_reason(
    payload: Any, min_remaining_weekly_percent: int = 0
) -> Optional[str]:
    data = _decode_possible_json_payload(payload)
    if isinstance(data, str):
        for kw in KNOWN_CLIPROXY_ERROR_LABELS:
            if kw in data:
                return _format_known_cliproxy_error(kw)
        return None
    if not isinstance(data, dict):
        return None
    error = data.get("error")
    if isinstance(error, dict):
        et = error.get("type")
        if et:
            return _format_known_cliproxy_error(et)
        msg = error.get("message")
        if msg:
            return str(msg)
    for key in ("rate_limit", "code_review_rate_limit"):
        pct = min_remaining_weekly_percent if key == "rate_limit" else 0
        reason = _extract_rate_limit_reason(data.get(key), key, pct)
        if reason:
            return reason
    arl = data.get("additional_rate_limits")
    if isinstance(arl, list):
        for i, ri in enumerate(arl):
            r = _extract_rate_limit_reason(ri, f"additional_rate_limits[{i}]", 0)
            if r:
                return r
    elif isinstance(arl, dict):
        for k, ri in arl.items():
            r = _extract_rate_limit_reason(ri, f"additional_rate_limits.{k}", 0)
            if r:
                return r
    for k in ("data", "body", "response", "text", "content", "status_message"):
        r = _extract_cliproxy_failure_reason(data.get(k), min_remaining_weekly_percent)
        if r:
            return r
    data_str = json.dumps(data, ensure_ascii=False)
    for kw in KNOWN_CLIPROXY_ERROR_LABELS:
        if kw in data_str:
            return _format_known_cliproxy_error(kw)
    return None


def refresh_oauth_token(refresh_token: str, proxies: Any = None) -> Tuple[bool, dict]:
    """刷新获取新的 access_token 等凭证"""
    return _refresh_oauth_token(refresh_token, proxies=proxies)


def test_cliproxy_auth_file(item: dict, api_url: str, api_token: str) -> Tuple[bool, str]:
    auth_index = item.get("auth_index")
    base_url   = api_url.strip().rstrip("/")
    call_url   = (
        base_url.replace("/auth-files", "/api-call")
        if "/auth-files" in base_url
        else f"{base_url}/v0/management/api-call"
    )
    payload = {
        "authIndex": auth_index,
        "method":    "GET",
        "url":       "https://chatgpt.com/backend-api/wham/usage",
        "header": {
            "Authorization":     "Bearer $TOKEN$",
            "Content-Type":      "application/json",
            "User-Agent":        DEFAULT_CLIPROXY_UA,
            "Chatgpt-Account-Id": str(item.get("account_id") or ""),
        },
    }
    try:
        resp = requests.post(
            call_url,
            headers={"Authorization": f"Bearer {api_token}"},
            json=payload, timeout=60, impersonate="chrome110",
        )
        if resp.status_code != 200:
            return False, f"HTTP {resp.status_code}"
        data        = resp.json()
        item['_raw_usage'] = data
        status_code = data.get("status_code", 0)
        reason      = _extract_cliproxy_failure_reason(data, cfg.MIN_REMAINING_WEEKLY_PERCENT)
        if status_code >= 400 or reason:
            return False, reason or f"HTTP {status_code}"
        return True, "正常"
    except Exception:
        return False, "测活超时"

def test_sub2api_account_direct(item: dict, proxy: str) -> Tuple[bool, str]:
    """直连 OpenAI 接口进行 Sub2API 账号测活，并实时提取真实额度"""
    credentials = item.get("credentials", {})
    access_token = credentials.get("access_token")
    account_id = credentials.get("chatgpt_account_id", "")
    
    if not access_token:
        return False, "缺少 access_token"
        
    headers = {
        "Authorization": f"Bearer {access_token}",
        "User-Agent": DEFAULT_CLIPROXY_UA,
        "Accept": "application/json"
    }
    if account_id:
        headers["Chatgpt-Account-Id"] = account_id

    try:
        proxies = {"http": proxy, "https": proxy} if proxy else None
        
        resp = requests.get(
            "https://chatgpt.com/backend-api/wham/usage",
            headers=headers,
            proxies=proxies,
            timeout=30,
            impersonate="chrome110"
        )
        
        if resp.status_code != 200:
            if resp.status_code == 401: return False, "凭证无效 (HTTP 401)"
            if resp.status_code == 403: return False, "请求被拒绝 (HTTP 403)"
            return False, f"HTTP {resp.status_code}"
            
        data = resp.json()

        reason = _extract_cliproxy_failure_reason(data,0)
        if reason:
            return False, reason
            
        pct_str = "未知"
        rl_data = data.get("rate_limit", {})
        if isinstance(rl_data, dict):
            pct = _extract_remaining_percent(rl_data.get("primary_window"))
            if pct is not None:
                pct_str = f"{pct:.1f}%"
                
        return True, f"实时剩余: {pct_str}"
    except Exception as e:
        return False, f"测活异常: {e}"

def process_account_worker(i: int, total: int, item: dict, args: Any) -> bool:
    if hasattr(args, 'check_stop') and args.check_stop(): return False
    name        = item.get("name")
    is_disabled = item.get("disabled", False)
    is_ok, msg  = test_cliproxy_auth_file(item, cfg.CPA_API_URL, cfg.CPA_API_TOKEN)

    if is_ok:
        if is_disabled:
            print(f"[{ts()}] [INFO] 测活: {mask_email(name)} 额度已恢复且有效，准备启用...")
            ok = set_cpa_auth_file_status(cfg.CPA_API_URL, cfg.CPA_API_TOKEN, name, disabled=False)
            print(
                f"[{ts()}] [{'SUCCESS' if ok else 'ERROR'}] 凭证 {mask_email(name)} "
                f"{'已成功启用！' if ok else '启用失败。'}"
            )
            return ok
        print(f"[{ts()}] [INFO] 测活: {mask_email(name)} 状态健康")
        return True

    print(f"[{ts()}] [WARNING] 测活: 凭证 {mask_email(name)} 失效，原因: {msg}")

    if "周限额" in msg or "usage_limit_reached" in msg:
        if cfg.REMOVE_ON_LIMIT_REACHED:
            print(f"[{ts()}] [INFO] 触发限额剔除规则，执行物理剔除...")
            requests.delete(
                _normalize_cpa_auth_files_url(cfg.CPA_API_URL),
                headers={"Authorization": f"Bearer {cfg.CPA_API_TOKEN}"},
                params={"name": name},
            )
        elif not is_disabled:
            print(f"[{ts()}] [INFO] 测活: 凭证额度耗尽，正在禁用...")
            ok = set_cpa_auth_file_status(cfg.CPA_API_URL, cfg.CPA_API_TOKEN, name, disabled=True)
            print(
                f"[{ts()}] [{'SUCCESS' if ok else 'ERROR'}] "
                f"测活: 凭证 {mask_email(name)} {'已成功禁用，等待额度重置。' if ok else '禁用失败！'}"
            )
        else:
            print(f"[{ts()}] [INFO] 测活: 账号额度尚未恢复，继续保持禁用状态。")
        return False

    if not cfg.ENABLE_TOKEN_REVIVE:
        print(f"[{ts()}] [INFO] 检测到 Token 已失效，但【复活】已关闭，仅记录状态。")
        _handle_dead_account(name, is_disabled)
        return False

    print(f"[{ts()}] [INFO] 测活: 凭证 {mask_email(name)} 准备尝试刷新 Token 复活...")
    refresh_success = False

    if item.get("runtime_only") or item.get("source") == "memory":
        print(f"[{ts()}] [WARNING] {mask_email(name)} 属于纯内存凭据，跳过抢救。")
        full_item_data: dict = {}
    else:
        try:
            dl_url = f"{_normalize_cpa_auth_files_url(cfg.CPA_API_URL)}/download"
            content_resp = requests.get(
                dl_url, params={"name": name},
                headers={"Authorization": f"Bearer {cfg.CPA_API_TOKEN}"},
                timeout=20,
            )
            full_item_data = content_resp.json() if content_resp.status_code == 200 else {}
            if content_resp.status_code != 200:
                print(f"[{ts()}] [ERROR] 获取 {mask_email(name)} 完整内容失败 "
                      f"(HTTP {content_resp.status_code})")
        except Exception as e:
            print(f"[{ts()}] [ERROR] 获取 {mask_email(name)} 完整内容异常: {e}")
            full_item_data = {}

    refresh_token_val = full_item_data.get("refresh_token")
    if refresh_token_val:
        proxies = {"http": args.proxy, "https": args.proxy} if args.proxy else None
        ok, new_tokens = refresh_oauth_token(refresh_token_val, proxies=proxies)
        if ok:
            print(f"[{ts()}] [INFO] {mask_email(name)} Token 刷新成功，正在同步至CPA...")
            full_item_data.update(new_tokens)
            if "email" not in full_item_data:
                full_item_data["email"] = name.replace(".json", "")
            up_ok, up_msg = upload_to_cpa_integrated(
                full_item_data, cfg.CPA_API_URL, cfg.CPA_API_TOKEN, custom_filename=name
            )
            if up_ok:
                time.sleep(3)
                is_ok2, msg2 = test_cliproxy_auth_file(item, cfg.CPA_API_URL, cfg.CPA_API_TOKEN)
                if is_ok2:
                    refresh_success = True
                    print(f"[{ts()}] [SUCCESS] 测活: {mask_email(name)} 刷新后复活成功！")
                else:
                    print(f"[{ts()}] [WARNING] {mask_email(name)} 刷新后二次测活依然失败({msg2})")
            else:
                print(f"[{ts()}] [ERROR] 刷新后覆盖CPA失败: {up_msg}")
        else:
            print(f"[{ts()}] [WARNING] {mask_email(name)} Token 复活请求被拒绝: "
                  f"{new_tokens.get('error','未知错误')}")
    else:
        print(f"[{ts()}] [WARNING] {mask_email(name)} 未找到有效数据，无法抢救")

    if not refresh_success:
        _handle_dead_account(name, is_disabled)
    return refresh_success


def _handle_dead_account(name: str, is_disabled: bool) -> None:
    """统一处理彻底死亡账号（删除或禁用）。"""
    if cfg.REMOVE_DEAD_ACCOUNTS:
        print(f"[{ts()}] [WARNING] 凭证 {mask_email(name)} 彻底死亡，执行物理剔除...")
        requests.delete(
            _normalize_cpa_auth_files_url(cfg.CPA_API_URL),
            headers={"Authorization": f"Bearer {cfg.CPA_API_TOKEN}"},
            params={"name": name},
        )
    elif not is_disabled:
        print(f"[{ts()}] [INFO] 凭证 {mask_email(name)} 死亡，根据配置保留，正在禁用...")
        if set_cpa_auth_file_status(cfg.CPA_API_URL, cfg.CPA_API_TOKEN, name, disabled=True):
            print(f"[{ts()}] [SUCCESS] 死亡凭证 {mask_email(name)} 已成功禁用。")
    else:
        print(f"[{ts()}] [WARNING] 凭证 {mask_email(name)} 已死亡，当前已是禁用状态，根据配置保留不删除。")

def handle_registration_result(result: Any, cpa_upload: bool = False, run_ctx: dict = None) -> str:
    if getattr(cfg, 'GLOBAL_STOP', False):
        return "stopped"
    global run_stats

    last_email = mail_service.get_last_email()
    if not last_email or "@" not in last_email:
        return "failed"

    if "+" in last_email:
        u_part, d_part = last_email.split("@")
        master_email = f"{u_part.split('+')[0]}@{d_part}"
        is_raw = False
    else:
        master_email = last_email
        is_raw = True

    is_dead = False
    if run_ctx:
        if run_ctx.get('pwd_blocked'):
            with _stats_lock: run_stats["pwd_blocked"] += 1
            is_dead = True
        if run_ctx.get('phone_verify'):
            with _stats_lock: run_stats["phone_verify"] += 1
            is_dead = True
    signup_blocked = run_ctx.get('signup_blocked', False) if run_ctx else False
    if (signup_blocked or is_dead) and getattr(cfg, "EMAIL_API_MODE", "") == "local_microsoft":
        if getattr(cfg, "LOCAL_MS_POOL_FISSION", False):
            db_manager.update_pool_fission_result(master_email, is_blocked=True, is_raw=is_raw)
        elif not getattr(cfg, "LOCAL_MS_ENABLE_FISSION", False):
            db_manager.update_local_mailbox_status(master_email, 3)
            print(f"[{ts()}] [WARNING] 触发风控，已将主号标记为死号: {mask_email(master_email)}")

    cur_dom = last_email.split("@")[-1] if last_email and "@" in last_email else None

    token_json_str = None
    password = None
    if result and isinstance(result, (tuple, list)) and len(result) >= 2:
        token_json_str, password = result
        
    ret_status = "success"
     
    if not token_json_str or token_json_str == "retry_403":
        if token_json_str == "retry_403":
            with _stats_lock: run_stats["retries"] += 1
            print(f"[{ts()}] [WARNING] 检测到 403 频率限制，挂起重试...")
            ret_status = "retry_403"
        else:
            with _stats_lock: run_stats["failed"] += 1
            ret_status = "failed"
        if cfg.ENABLE_SUB_DOMAINS:
            mail_service.clear_sticky_domain() 
            print(f"[{ts()}] [系统] 域名 {mask_email(cur_dom or '')} 注册失败，下一轮重新生成。")
            
    else:
        with _stats_lock: run_stats["success"] += 1
        token_data    = json.loads(token_json_str)
        account_email = token_data.get("email", "unknown")

        # 存入本地数据库
        if (cpa_upload and cfg.SAVE_TO_LOCAL_IN_CPA_MODE) or not cpa_upload:
            if db_manager.save_account_to_db(account_email, password, token_json_str):
                print(f"[{ts()}] [SUCCESS] 账号密码与 Token 已安全存入: {mask_email(account_email)}")

        # CPA 云端上传
        if cpa_upload:
            success, up_msg = upload_to_cpa_integrated(token_data, cfg.CPA_API_URL, cfg.CPA_API_TOKEN)
            if success:
                print(f"[{ts()}] [SUCCESS] 补货凭证 {mask_email(account_email)} 云端上传成功！")
            else:
                print(f"[{ts()}] [ERROR] 云端上传失败: {up_msg}")

        if getattr(cfg, "LOCAL_MS_POOL_FISSION", False) and cfg.EMAIL_API_MODE == "local_microsoft":
            db_manager.update_pool_fission_result(master_email, is_blocked=False, is_raw=is_raw)
        elif not getattr(cfg, "LOCAL_MS_ENABLE_FISSION", False) and cfg.EMAIL_API_MODE == "local_microsoft":
            db_manager.update_local_mailbox_status(master_email, 2)

        safe_pwd = str(password) if password else ""
        orig_masked_email = mail_service.mask_email(account_email, force_mask=True)
        orig_masked_password = f"{safe_pwd[:2]}****{safe_pwd[-2:]}" if len(safe_pwd) > 4 else "****"

        final_email = orig_masked_email if getattr(cfg, 'TG_BOT', {}).get("mask_email", False) else account_email
        final_password = orig_masked_password if getattr(cfg, 'TG_BOT', {}).get("mask_password", False) else safe_pwd

        template_str = getattr(cfg, 'TG_BOT', {}).get("template_success", "成功: {email} / {password} 时间: {time}")
        beijing_tz = timezone(timedelta(hours=8))
        current_time = datetime.now(beijing_tz).strftime("%Y-%m-%d %H:%M:%S")
        try:
            success_text = template_str.format(email=final_email, password=final_password, time=current_time)
        except Exception:
            success_text = f"🎉 注册成功\n账号: {final_email}\n密码: {final_password}\n时间: {current_time}\n(温馨提示: 您的TG单号自定义模板配置有误)"

        send_tg_msg_sync(success_text)
    return ret_status

def run_and_refresh(proxy, args, cpa_upload=False, skip_switch=False):
    proxy = format_docker_url(proxy)
    """切节点 → 注册 → 处理结果。"""
    if not skip_switch:
        if not smart_switch_node(proxy):
            print(f"[{ts()}] [WARNING] {proxy} 节点切换失败，将使用当前 IP 继续尝试...")
    
    result = None
    run_ctx = {}
    try:
        result = run(proxy, run_ctx=run_ctx)
    except Exception as e:
        print(f"[{ts()}] [ERROR] 注册线程发生未捕获异常{e}")

    return handle_registration_result(result, cpa_upload=cpa_upload, run_ctx=run_ctx)

# def auto_heal_subdomain(failed_domain: str):
    # print(f"[{ts()}] [自愈] 域名 {failed_domain} 达到失败阈值，触发更替程序...")
    # import wfxl_openai_regst 
    # cf_cfg = getattr(cfg, '_c', {})
    # api_email = cf_cfg.get("cf_api_email")
    # api_key = cf_cfg.get("cf_api_key")
    # root_str = cf_cfg.get("mail_domains", "")
    # root_domains = [d.strip() for d in root_str.split(",") if d.strip()]
    
    # main_dom = None
    # for root in root_domains:
        # if failed_domain.endswith(root):
            # main_dom = root
            # break
    # if not main_dom:
        # print(f"[{ts()}] [ERROR] 无法识别 {failed_domain} 所属的主域，请检查配置！")
        # return
        
        
    # level = cf_cfg.get("sub_domain_level", 1)
    
    # try:
        # from cloudflare import Cloudflare
        # cf = Cloudflare(api_email=api_email, api_key=api_key)
        # zones = cf.zones.list(name=main_dom)
        # if zones.result:
            # zone_id = zones.result[0].id
            # url = f"https://api.cloudflare.com/client/v4/zones/{zone_id}/email/routing/dns"
            # headers = {"X-Auth-Email": api_email, "X-Auth-Key": api_key, "Content-Type": "application/json"}
            # payload = json.dumps({"name": failed_domain}).encode('utf-8')
            # requests.delete(url, data=payload, headers=headers, impersonate="chrome110")
            # wfxl_openai_regst.dispatch_email_backend_delete(failed_domain, cf_cfg)
            # print(f"[{ts()}] [自愈] 已成功注销失效域名: {mask_email(failed_domain)}")
    # except Exception as e:
        # print(f"[{ts()}] [ERROR] 销毁失效域名异常: {e}")
        # return

    # refill_num = int(getattr(cfg, 'SUB_DOMAIN_REFILL_COUNT', 1))
    # new_domains = []
    # for _ in range(refill_num):
        # random_parts = []
        # for _ in range(level):
            # random_parts.append(''.join(random.choices(string.ascii_lowercase + string.digits, k=8)))
        # new_domains.append(".".join(random_parts) + f".{main_dom}")

    # with _heal_lock:
        # current_list = [d.strip() for d in cfg.SUB_DOMAINS_LIST.split(",") if d.strip()]
        # if failed_domain in current_list:
            # current_list.remove(failed_domain)
        # current_list.extend(new_domains)
        
        # config_path = "config.yaml"
        # try:
            # with open(config_path, "r", encoding="utf-8") as f:
                # y = yaml.safe_load(f) or {}
            # y["sub_domains_list"] = ",".join(current_list)
            # y["sub_domain_fail_threshold"] = cfg.SUB_DOMAIN_FAIL_THRESHOLD
            # y["sub_domain_refill_count"] = cfg.SUB_DOMAIN_REFILL_COUNT
            
            # with open(config_path, "w", encoding="utf-8") as f:
                # yaml.dump(y, f, allow_unicode=True, sort_keys=False)
            # reload_all_configs()
        # except Exception as e:
            # print(f"[{ts()}] [ERROR] 自愈配置保存失败: {e}")

    # for ns in new_domains:
        # try:
            # cf.email_routing.dns.create(zone_id=zone_id, name=ns)
            # wfxl_openai_regst.dispatch_email_backend_add(ns, cf_cfg)
            # print(f"[{ts()}] [自愈] 已补货新域名 {ns}，等待生效...")
        # except: pass

    # print(f"[{ts()}] [自愈] 正在进入状态监控，等待 Cloudflare 激活路由...")
    # retry_count = 0
    # while True:
        # try:
            # info = cf.email_routing.get(zone_id=zone_id)
            # res_data = getattr(info, 'result', info)
            # status = getattr(res_data, 'status', 'unknown')
            # synced = getattr(res_data, 'synced', False)

            # retry_count += 1
            
            # print(f"[{ts()}] [监控] (等待中...)")
            
            # if status == 'ready':
                # if synced is True or retry_count > 20: 
                    # print(f"[{ts()}] [SUCCESS] 域名池状态确认完成，准备恢复业务线程。")
                    # break
                    
        # except Exception as e:
            # print(f"[{ts()}] [WARNING] 状态监控请求异常 (重试中): {e}")
            # if retry_count > 6: break
            
        # time.sleep(10)
        
# def auto_heal_subdomain(failed_domain: str):
#     """
#     功能：仅销毁本地失效域名记录。
#     """
#     print(f"[{ts()}] [自愈] 域名 {failed_domain} 达到失败阈值，启动快速更替程序...")
#
#     cf_cfg = getattr(cfg, '_c', {})
#     root_str = cf_cfg.get("mail_domains", "")
#     root_domains = [d.strip() for d in root_str.split(",") if d.strip()]
#
#     main_dom = None
#     for root in root_domains:
#         if failed_domain.endswith(root):
#             main_dom = root
#             break
#
#     if not main_dom:
#         print(f"[{ts()}] [ERROR] 无法识别 {failed_domain} 所属的主域，跳过自愈。")
#         return
#
#     level = cf_cfg.get("sub_domain_level", 1)
#     refill_num = int(getattr(cfg, 'SUB_DOMAIN_REFILL_COUNT', 1))
#     new_domains = []
#     for _ in range(refill_num):
#         random_parts = []
#         for _ in range(level):
#             random_parts.append(''.join(random.choices(string.ascii_lowercase + string.digits, k=8)))
#         new_domains.append(".".join(random_parts) + f".{main_dom}")
#
#     with _heal_lock:
#         current_list = [d.strip() for d in cfg.SUB_DOMAINS_LIST.split(",") if d.strip()]
#         if failed_domain in current_list:
#             current_list.remove(failed_domain)
#         current_list.extend(new_domains)
#
#         config_path = "config.yaml"
#         try:
#             with open(config_path, "r", encoding="utf-8") as f:
#                 y = yaml.safe_load(f) or {}
#
#             y["sub_domains_list"] = ",".join(current_list)
#             y["sub_domain_fail_threshold"] = cfg.SUB_DOMAIN_FAIL_THRESHOLD
#             y["sub_domain_refill_count"] = cfg.SUB_DOMAIN_REFILL_COUNT
#
#             with open(config_path, "w", encoding="utf-8") as f:
#                 yaml.dump(y, f, allow_unicode=True, sort_keys=False)
#
#             reload_all_configs()
#             for ns in new_domains:
#                 print(f"[{ts()}] [自愈] 已成功补货新域名: {ns}")
#
#             print(f"[{ts()}] [SUCCESS] 配置文件已更新，业务线程将无缝切换新域名。")
#         except Exception as e:
#             print(f"[{ts()}] [ERROR] 自愈配置保存失败: {e}")

def _handle_sub2api_dead_account(item: dict, client: Any, is_disabled: bool) -> None:
    """统一处理 Sub2API 彻底死亡账号（删除或禁用）"""
    name = item.get("name", "unknown")
    account_id = item.get("id") 

    if cfg.SUB2API_REMOVE_DEAD_ACCOUNTS:
        print(f"[{ts()}] [ERROR] 凭证 {mask_email(name)} 彻底死亡，执行物理剔除...")
        if hasattr(client, "delete_account") and account_id:
            client.delete_account(account_id) 
    elif not is_disabled:
        print(f"[{ts()}] [ERROR] 凭证 {mask_email(name)} 死亡，根据配置保留，正在禁用...")
        if hasattr(client, "set_account_status") and account_id:
            client.set_account_status(account_id, disabled=True)
    else:
        print(f"[{ts()}] [ERROR] 凭证 {mask_email(name)} 已死亡，当前已是禁用状态，根据配置保留不删除。")

def process_sub2api_worker(i: int, total: int, item: dict, client: Any, args: Any) -> bool:
    """Sub2API 测活 Worker（使用 Sub2API /test SSE 接口）"""
    if hasattr(args, 'check_stop') and args.check_stop(): return False
    name = item.get("name", "unknown")
    account_id = item.get("id")
    result, reason = client.test_account(account_id)

    if result == "ok":
        print(f"[{ts()}] [SUCCESS] Sub2API测活: {mask_email(name)} 状态健康")
        return True

    if result == "quota":
        if cfg.SUB2API_REMOVE_ON_LIMIT_REACHED:
            print(f"[{ts()}] [WARNING] Sub2API测活: {mask_email(name)} 额度耗尽，执行物理删除...")
            if account_id:
                client.delete_account(account_id)
            return False
        print(f"[{ts()}] [WARNING] Sub2API测活: {mask_email(name)} 额度限流，暂不计入有效库存，Sub2API 自动管理")
        return False

    print(f"[{ts()}] [ERROR] Sub2API测活: {mask_email(name)} 测活失败 ({reason})")

    if not cfg.SUB2API_ENABLE_TOKEN_REVIVE:
        print(f"[{ts()}] [ERROR] Token 复活已关闭，直接执行死亡处理")
        _handle_sub2api_dead_account(item, client, is_disabled=False)
        return False

    refresh_token_val = item.get("credentials", {}).get("refresh_token")
    if not refresh_token_val:
        print(f"[{ts()}] [ERROR] {mask_email(name)} 无 refresh_token，执行死亡处理")
        _handle_sub2api_dead_account(item, client, is_disabled=False)
        return False

    print(f"[{ts()}] [INFO] {mask_email(name)} 尝试刷新 Token...")
    proxies = {"http": args.proxy, "https": args.proxy} if args.proxy else None
    ok, new_tokens = refresh_oauth_token(refresh_token_val, proxies=proxies)

    if not ok:
        err_info = new_tokens.get('error', '未知') if isinstance(new_tokens, dict) else str(new_tokens)
        print(f"[{ts()}] [ERROR] {mask_email(name)} Token 刷新失败: {err_info}")
        _handle_sub2api_dead_account(item, client, is_disabled=False)
        return False

    print(f"[{ts()}] [INFO] {mask_email(name)} Token 刷新成功，同步至 Sub2API...")
    item.setdefault("credentials", {}).update(new_tokens)
    up_ok, up_msg = client.update_account(account_id, item)
    if not up_ok:
        print(f"[{ts()}] [ERROR] {mask_email(name)} 更新回 Sub2API 失败: {up_msg}")
        _handle_sub2api_dead_account(item, client, is_disabled=False)
        return False

    print(f"[{ts()}] [INFO] {mask_email(name)} Token 已更新，二次验证中...")
    result2, reason2 = client.test_account(account_id)

    if result2 == "ok":
        print(f"[{ts()}] [SUCCESS] {mask_email(name)} 刷新复活成功，二次验证通过！")
        return True

    print(f"[{ts()}] [ERROR] {mask_email(name)} 二次验证失败 ({reason2})，账号确认已死")
    _handle_sub2api_dead_account(item, client, is_disabled=False)
    return False

def normal_main_loop(args, stop_event: threading.Event):
    """常规量产模式（纯数据库保存）"""
    sleep_min    = max(1, cfg.NORMAL_SLEEP_MIN)
    sleep_max    = max(sleep_min, cfg.NORMAL_SLEEP_MAX)
    target_count = cfg.NORMAL_TARGET_COUNT

    print(f"\n[{ts()}] [系统] >>> 启动常规量产模式 <<<")
    if target_count > 0:
        print(f"[{ts()}] [系统] 任务目标: 注册 {target_count} 个账号后自动停止")
    else:
        print(f"[{ts()}] [系统] 任务目标: 无限挂机注册 (按 Ctrl+C 停止)")

    success_count  = 0
    total_attempts = 0

    while not stop_event.is_set() and not cfg.POOL_EXHAUSTED:
        if target_count > 0 and success_count >= target_count:
            print(f"\n[{ts()}] [SUCCESS] 已达到目标注册数量 ({target_count})，任务圆满结束！")
            break

        total_attempts += 1
        print(f"\n[{ts()}] [系统] 开始第 {total_attempts} 次注册 (已成功: {success_count}) ---")
        if stop_event.wait(1.0):
            break

        try:
            if cfg._clash_enable and not cfg._clash_pool_mode:
                print(f"[{ts()}] [INFO] 触发单端口共享模式，正在进行全局节点切换...")
                if not smart_switch_node(args.proxy):
                    print(f"[{ts()}] [WARNING] 全局节点切换失败，将使用当前 IP 继续尝试...")

            if cfg.ENABLE_MULTI_THREAD_REG:
                current_batch = (
                    min(cfg.REG_THREADS, target_count - success_count)
                    if target_count > 0 else cfg.REG_THREADS
                )
                print(f"[{ts()}] [INFO] 启用多线程并发 ({current_batch} 条通道)")

                def _worker():
                    if stop_event.is_set(): return "stopped"
                    if cfg._clash_enable and cfg._clash_pool_mode:
                        p = cfg.PROXY_QUEUE.get()
                        try:
                            return run_and_refresh(p, args, False, skip_switch=False)
                        finally:
                            cfg.PROXY_QUEUE.put(p)
                            cfg.PROXY_QUEUE.task_done()
                    return run_and_refresh(args.proxy, args, False, skip_switch=True)

                with ThreadPoolExecutor(max_workers=current_batch) as ex:
                    futures = [ex.submit(_worker) for _ in range(current_batch)]
                    for f in futures:
                        if f.result() == "success":
                            success_count += 1
            else:
                if cfg._clash_enable and cfg._clash_pool_mode:
                    p = cfg.PROXY_QUEUE.get()
                    try:
                        status = run_and_refresh(p, args, False, skip_switch=False)
                    finally:
                        cfg.PROXY_QUEUE.put(p)
                        cfg.PROXY_QUEUE.task_done()
                else:
                    status = run_and_refresh(args.proxy, args, False, skip_switch=True)

                if status == "success":
                    success_count += 1

        except Exception as e:
            print(f"[{ts()}] [ERROR] 发生未捕获全局异常: {e}")

        if target_count > 0 and success_count >= target_count:
            print(f"\n[{ts()}] [SUCCESS] 已达到目标注册数量 ({target_count})，任务圆满结束！")
            break

        if args.once:
            break

        wait_time = random.randint(sleep_min, sleep_max)
        print(f"[{ts()}] [INFO] 缓冲防风控，等待 {wait_time} 秒后继续...")
        if stop_event.wait(wait_time):
            break


async def perform_cpa_check(args, async_stop_event, loop):
    print(f"[{ts()}] [INFO] 开始执行 CPA 仓库全量测活巡检...")
    res = requests.get(
        _normalize_cpa_auth_files_url(cfg.CPA_API_URL),
        headers={"Authorization": f"Bearer {cfg.CPA_API_TOKEN}"},
        timeout=20,
    )
    all_files = res.json().get("files", [])
    codex_files = [
        f for f in all_files
        if "codex" in str(f.get("type", "")).lower()
           or "codex" in str(f.get("provider", "")).lower()
    ]
    total_files = len(codex_files)

    with ThreadPoolExecutor(max_workers=cfg.CPA_THREADS) as executor:
        futures = [
            loop.run_in_executor(executor, process_account_worker, i, total_files, item, args)
            for i, item in enumerate(codex_files, 1)
        ]
        results = await asyncio.gather(*futures)

    valid_count = sum(1 for r in results if r)
    print(f"[{ts()}] [INFO] CPA 测活结束，当前有效数: {valid_count} / {total_files}")
    return valid_count, total_files


async def perform_sub2api_check(args, async_stop_event, loop, client):
    print(f"[{ts()}] [INFO] 开始执行 Sub2API 仓库全量测活巡检...")
    success, account_list = client.get_all_accounts()
    if not success:
        print(f"[{ts()}] [ERROR] 获取 Sub2API 全量库存失败: {account_list}")
        return 0, 0

    total_files = len(account_list)

    with ThreadPoolExecutor(max_workers=cfg.SUB2API_THREADS) as executor:
        futures = [
            loop.run_in_executor(executor, process_sub2api_worker, i, total_files, item, client, args)
            for i, item in enumerate(account_list, 1)
        ]
        results = await asyncio.gather(*futures)

    valid_count = sum(1 for r in results if r)
    print(f"[{ts()}] [INFO] Sub2API 测活结束，当前有效数: {valid_count} / {total_files}")
    return valid_count, total_files

async def manual_check_main_loop(args, async_stop_event: asyncio.Event):
    print("=" * 60)
    print(f"\n[{ts()}] [系统] >>> 启动独立测活清理任务 <<<")
    print("=" * 60)
    loop = asyncio.get_running_loop()

    if cfg.ENABLE_CPA_MODE:
        await perform_cpa_check(args, async_stop_event, loop)
    elif cfg.ENABLE_SUB2API_MODE:
        client = Sub2APIClient(api_url=cfg.SUB2API_URL, api_key=cfg.SUB2API_KEY)
        await perform_sub2api_check(args, async_stop_event, loop, client)
    else:
        print(f"[{ts()}] [WARNING] 当前未开启 CPA 或 Sub2API 模式，无法执行仓管测活。")

    print(f"\n[{ts()}] [SUCCESS] 独立测活任务执行完毕！")
    cfg.GLOBAL_STOP = True
    async_stop_event.set()


async def cpa_main_loop(args, async_stop_event: asyncio.Event):
    """CPA 智能仓管模式（接入发牌器，防止撞车）。"""
    print("=" * 60)
    print(f"\n[{ts()}] [系统] 目标库存阈值: {cfg.MIN_ACCOUNTS_THRESHOLD} | 单次补发量: {cfg.BATCH_REG_COUNT}")
    print(
        f"\n[{ts()}] [系统] 周限额剔除规则: 剩余低于 {cfg.MIN_REMAINING_WEEKLY_PERCENT}%"
        if cfg.MIN_REMAINING_WEEKLY_PERCENT > 0
        else f"\n[{ts()}] [系统] 周限额剔除规则: 完全耗尽才剔除"
    )
    print("=" * 60)

    loop = asyncio.get_running_loop()

    while not async_stop_event.is_set() and not cfg.POOL_EXHAUSTED:
        try:
            if cfg.CPA_AUTO_CHECK:
                valid_count, total_files = await perform_cpa_check(args, async_stop_event, loop)
            else:
                print(f"\n[{ts()}] [INFO] 自动测活已关闭，直接读取云端列表进行补发判断...")
                res = requests.get(
                    _normalize_cpa_auth_files_url(cfg.CPA_API_URL),
                    headers={"Authorization": f"Bearer {cfg.CPA_API_TOKEN}"},
                    timeout=20,
                )
                all_files = res.json().get("files", [])
                codex_files = [
                    f for f in all_files
                    if "codex" in str(f.get("type", "")).lower()
                       or "codex" in str(f.get("provider", "")).lower()
                ]
                total_files = len(codex_files)
                valid_count = total_files
                print(f"[{ts()}] [INFO] 当前云端总数: {total_files} (未开启自动巡检，默认全部视为有效)")

            if valid_count < cfg.MIN_ACCOUNTS_THRESHOLD:
                need_to_reg          = cfg.BATCH_REG_COUNT
                global run_stats
                run_stats["target"] += need_to_reg
                success_in_this_cycle = 0
                print(f"[{ts()}] [INFO] 库存不足 ({valid_count} < {cfg.MIN_ACCOUNTS_THRESHOLD})，启动补货...")
                await asyncio.sleep(1)

                def _cpa_worker():
                    if async_stop_event.is_set(): return "stopped"
                    if cfg._clash_enable and cfg._clash_pool_mode:
                        p = cfg.PROXY_QUEUE.get()
                        try:
                            return run_and_refresh(p, args, cpa_upload=True, skip_switch=False)
                        finally:
                            cfg.PROXY_QUEUE.put(p)
                            cfg.PROXY_QUEUE.task_done()
                    return run_and_refresh(args.proxy, args, cpa_upload=True, skip_switch=True)

                while success_in_this_cycle < need_to_reg and not async_stop_event.is_set() and not cfg.POOL_EXHAUSTED:
                    remaining  = need_to_reg - success_in_this_cycle
                    batch_size = min(cfg.REG_THREADS, remaining)

                    if cfg._clash_enable and not cfg._clash_pool_mode:
                        print(f"[{ts()}] [INFO] [CPA补货] 切换全局节点...")
                        if not smart_switch_node(args.proxy):
                            print(f"[{ts()}] [WARNING] [CPA补货] 全局节点切换失败，使用当前 IP 继续...")

                    if cfg.ENABLE_MULTI_THREAD_REG:
                        print(f"[{ts()}] [INFO] 多线程补货: {success_in_this_cycle}/{need_to_reg} "
                              f"({batch_size} 线程)")
                        with ThreadPoolExecutor(max_workers=batch_size) as ex:
                            reg_futures = [
                                loop.run_in_executor(ex, _cpa_worker)
                                for _ in range(batch_size)
                            ]
                            reg_results = await asyncio.gather(*reg_futures)
                        for status in reg_results:
                            if status == "success":
                                success_in_this_cycle += 1
                            elif status == "retry_403":
                                print(f"[{ts()}] [WARNING] 遇到 403 频率限制，给服务器 15 秒冷却时间...")
                                await asyncio.sleep(15)
                    else:
                        print(f"[{ts()}] [INFO] 单线程补货: {success_in_this_cycle}/{need_to_reg}")
                        if cfg._clash_enable and cfg._clash_pool_mode:
                            p = cfg.PROXY_QUEUE.get()
                            try:
                                status = await loop.run_in_executor(None, run_and_refresh, p, args, True, False)
                            finally:
                                cfg.PROXY_QUEUE.put(p)
                                cfg.PROXY_QUEUE.task_done()
                        else:
                            status = await loop.run_in_executor(
                                None, run_and_refresh, args.proxy, args, True, True
                            )
                        if status == "success":
                            success_in_this_cycle += 1
                        elif status == "retry_403":
                            await asyncio.sleep(10)
                        await asyncio.sleep(5)

                print(f"[{ts()}] [SUCCESS] 本轮补货完成！累计入库: {success_in_this_cycle} 个。")
            else:
                print(f"[{ts()}] [INFO] 仓库存量充足，无需补发。")

            if async_stop_event.is_set() or getattr(cfg, 'GLOBAL_STOP', False):
                print(f"[{ts()}] [系统] 主调度循环已彻底退出。")
                break

            print(f"[{ts()}] [INFO] 维护周期结束，{cfg.CHECK_INTERVAL_MINUTES} 分钟后进行下一次巡检...")
            try:
                await asyncio.wait_for(
                    async_stop_event.wait(),
                    timeout=cfg.CHECK_INTERVAL_MINUTES * 60,
                )
            except asyncio.TimeoutError:
                pass

        except Exception as e:
            print(f"[{ts()}] [ERROR] 主循环异常: {e}")
            try:
                await asyncio.wait_for(async_stop_event.wait(), timeout=60)
            except asyncio.TimeoutError:
                pass

async def sub2api_main_loop(args, async_stop_event: asyncio.Event):
    """Sub2API 智能仓管模式"""
    print("=" * 60)
    print(f"\n[{ts()}] [系统] Sub2API 目标库存阈值: {cfg.SUB2API_MIN_THRESHOLD} | 单次补发量: {cfg.SUB2API_BATCH_COUNT}")
    print(f"\n[{ts()}] [系统] Sub2API 限额处理: 仅在真实耗尽后禁用或剔除")
    print("=" * 60)

    loop = asyncio.get_running_loop()
    client = Sub2APIClient(api_url=cfg.SUB2API_URL, api_key=cfg.SUB2API_KEY)

    while not async_stop_event.is_set() and not cfg.POOL_EXHAUSTED:

        try:
            if cfg.SUB2API_AUTO_CHECK:
                print(f"\n[{ts()}] [INFO] 开始执行 Sub2API 仓库例行巡检与测活...")
                success, account_list = client.get_all_accounts()
                if not success:
                    print(f"[{ts()}] [ERROR] 获取 Sub2API 全量库存失败: {account_list}")
                    try: await asyncio.wait_for(async_stop_event.wait(), timeout=60)
                    except asyncio.TimeoutError: pass
                    continue

                total_files = len(account_list)

                with ThreadPoolExecutor(max_workers=cfg.SUB2API_THREADS) as executor:
                    futures = [
                        loop.run_in_executor(executor, process_sub2api_worker, i, total_files, item, client, args)
                        for i, item in enumerate(account_list, 1)
                    ]
                    results = await asyncio.gather(*futures)

                valid_count = sum(1 for r in results if r)
                print(f"[{ts()}] [INFO] 巡检结束，当前 Sub2API 仓库有效数: {valid_count}")
            else:
                print(f"[{ts()}] [INFO] Sub2API 自动测活已关闭，直接读取云端列表进行补发判断...")
                success, account_list = client.get_all_accounts()
                if not success:
                    print(f"[{ts()}] [ERROR] 获取 Sub2API 全量库存失败: {account_list}")
                    try:
                        await asyncio.wait_for(async_stop_event.wait(), timeout=60)
                    except asyncio.TimeoutError:
                        pass
                    continue
                total_files = len(account_list)
                valid_count = total_files
                print(f"[{ts()}] [INFO] 当前云端总数: {total_files} (未开启自动巡检，默认全部视为有效)")

            if valid_count < cfg.SUB2API_MIN_THRESHOLD:
                need_to_reg          = cfg.SUB2API_BATCH_COUNT
                global run_stats
                run_stats["target"] += need_to_reg
                success_in_this_cycle = 0
                print(f"[{ts()}] [INFO] 库存不足 ({valid_count} < {cfg.SUB2API_MIN_THRESHOLD})，启动补货...")
                await asyncio.sleep(1)

                def _sub2api_run_wrapper(p, skip_switch):
                    p = format_docker_url(p)
                    if not skip_switch:
                        if not smart_switch_node(p):
                            print(f"[{ts()}] [WARNING] [Sub2API补货] 全局节点切换失败...")
                    run_ctx = {}
                    result = run(p, run_ctx=run_ctx)
                    status = handle_registration_result(result, cpa_upload=False, run_ctx=run_ctx)

                    if status == "success":
                        token_dict = json.loads(result[0])
                        if hasattr(client, "add_account"):
                            ok, msg = client.add_account(token_dict)
                            if ok: print(f"[{ts()}] [SUCCESS] Sub2API 补货入库成功")
                            else: print(f"[{ts()}] [ERROR] Sub2API 补货入库失败: {msg}")
                    return status

                def _sub2api_worker():
                    if async_stop_event.is_set(): return "stopped"
                    if cfg._clash_enable and cfg._clash_pool_mode:
                        p = cfg.PROXY_QUEUE.get()
                        try:
                            return _sub2api_run_wrapper(p, False)
                        finally:
                            cfg.PROXY_QUEUE.put(p)
                            cfg.PROXY_QUEUE.task_done()
                    return _sub2api_run_wrapper(args.proxy, True)

                while success_in_this_cycle < need_to_reg and not async_stop_event.is_set() and not cfg.POOL_EXHAUSTED:
                    remaining  = need_to_reg - success_in_this_cycle
                    batch_size = min(cfg.REG_THREADS, remaining)

                    if cfg._clash_enable and not cfg._clash_pool_mode:
                        print(f"[{ts()}] [INFO] [Sub2API补货] 切换全局节点...")
                        if not smart_switch_node(args.proxy):
                            print(f"[{ts()}] [WARNING] [Sub2API补货] 全局节点切换失败，使用当前 IP 继续...")

                    if cfg.ENABLE_MULTI_THREAD_REG:
                        print(f"[{ts()}] [INFO] 多线程补货: {success_in_this_cycle}/{need_to_reg} "
                              f"({batch_size} 线程)")
                        with ThreadPoolExecutor(max_workers=batch_size) as ex:
                            reg_futures = [
                                loop.run_in_executor(ex, _sub2api_worker)
                                for _ in range(batch_size)
                            ]
                            reg_results = await asyncio.gather(*reg_futures)

                        for status in reg_results:
                            if status == "success":
                                success_in_this_cycle += 1
                            elif status == "retry_403":
                                print(f"[{ts()}] [WARNING] 遇到 403 频率限制，给服务器 15 秒冷却时间...")
                                try: await asyncio.wait_for(async_stop_event.wait(), timeout=15)
                                except asyncio.TimeoutError: pass

                    else:
                        print(f"[{ts()}] [INFO] 单线程补货: {success_in_this_cycle}/{need_to_reg}")
                        if cfg._clash_enable and cfg._clash_pool_mode:
                            p = cfg.PROXY_QUEUE.get()
                            try:
                                status = await loop.run_in_executor(None, _sub2api_run_wrapper, p, False)
                            finally:
                                cfg.PROXY_QUEUE.put(p)
                                cfg.PROXY_QUEUE.task_done()
                        else:
                            status = await loop.run_in_executor(
                                None, _sub2api_run_wrapper, args.proxy, True
                            )

                        if status == "success":
                            success_in_this_cycle += 1
                        elif status == "retry_403":
                            try: await asyncio.wait_for(async_stop_event.wait(), timeout=10)
                            except asyncio.TimeoutError: pass

                        try: await asyncio.wait_for(async_stop_event.wait(), timeout=5)
                        except asyncio.TimeoutError: pass

                print(f"[{ts()}] [SUCCESS] 本轮补货完成！累计入库 Sub2API: {success_in_this_cycle} 个。")
            else:
                print(f"[{ts()}] [INFO] 仓库存量充足，无需补发。")

            if async_stop_event.is_set() or getattr(cfg, 'GLOBAL_STOP', False):
                print(f"[{ts()}] [系统] 主调度循环已彻底退出。")
                break
            print(f"[{ts()}] [INFO] 维护周期结束，{cfg.SUB2API_CHECK_INTERVAL} 分钟后进行下一次巡检...")
            try:
                await asyncio.wait_for(
                    async_stop_event.wait(),
                    timeout=cfg.SUB2API_CHECK_INTERVAL * 60,
                )
            except asyncio.TimeoutError:
                pass

        except Exception as e:
            print(f"[{ts()}] [ERROR] Sub2API 循环发生致命异常: {e}")
            print(f"[{ts()}] [INFO] 触发安全保护，系统已自动停止运行。")
            async_stop_event.set()
            break


def main() -> None:
    reload_all_configs()
    parser = argparse.ArgumentParser(description="OpenAI 自动注册 & CPA 检测一体")
    parser.add_argument("--proxy", default=None, help="代理地址")
    # parser.add_argument("--once", action="store_true", help="只运行一次")
    args       = parser.parse_args()
    args.proxy = cfg.DEFAULT_PROXY if cfg.DEFAULT_PROXY.strip() else None

    if cfg.ENABLE_CPA_MODE:
        print("   当前状态: [ CPA 智能仓管模式 ] 已开启")
    else:
        print("   当前状态: [ 常规量产模式 ] 已开启")
    print("=" * 65)

    if cfg.ENABLE_CPA_MODE:
        try:
            asyncio.run(cpa_main_loop(args, asyncio.Event()))
        except KeyboardInterrupt:
            print(f"\n[{ts()}] [INFO] 用户终止了系统运行。")
    else:
        stop_event = threading.Event()
        try:
            normal_main_loop(args, stop_event)
        except KeyboardInterrupt:
            print(f"\n[{ts()}] [INFO] 用户终止了系统运行。")


class RegEngine:
    """GUI 用控制类，封装线程/协程生命周期。"""

    def __init__(self):
        self.thread_stop_event = threading.Event()
        self.async_stop_event  = None
        self.current_thread    = None
        self.loop              = None
        self._force_stopped    = False

    def start_normal(self, args):
        if self.is_running():
            return
        self._force_stopped = False
        cfg.GLOBAL_STOP = False
        cfg.POOL_EXHAUSTED = False
        self.thread_stop_event.clear()
        args.check_stop = lambda: self.thread_stop_event.is_set()
        self.current_thread = threading.Thread(
            target=normal_main_loop,
            args=(args, self.thread_stop_event),
            daemon=True,
        )
        self.current_thread.start()

    def start_cpa(self, args):
        if self.is_running():
            return
        self._force_stopped = False
        cfg.GLOBAL_STOP = False
        cfg.POOL_EXHAUSTED = False
        self.thread_stop_event.clear()
        self.current_thread = threading.Thread(
            target=self._run_cpa_in_thread, args=(args,), daemon=True
        )
        self.current_thread.start()
        
    def start_sub2api(self, args):
        if self.is_running():
            return
        self._force_stopped = False
        cfg.GLOBAL_STOP = False
        cfg.POOL_EXHAUSTED = False
        self.thread_stop_event.clear()
        self.current_thread = threading.Thread(
            target=self._run_sub2api_in_thread, args=(args,), daemon=True
        )
        self.current_thread.start()

    def _run_cpa_in_thread(self, args):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        try:
            self.loop.run_until_complete(self._cpa_wrapper(args))
        finally:
            self.loop.close()

    def _run_sub2api_in_thread(self, args):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        try:
            self.async_stop_event = asyncio.Event()
            self.loop.run_until_complete(sub2api_main_loop(args, self.async_stop_event))
        finally:
            self.loop.close()
            
    async def _cpa_wrapper(self, args):
        self.async_stop_event = asyncio.Event()
        await cpa_main_loop(args, self.async_stop_event)

    def stop(self):
        self._force_stopped = True
        cfg.GLOBAL_STOP = True
        cfg.POOL_EXHAUSTED = True
        self.thread_stop_event.set()
        if self.loop and self.async_stop_event:
            self.loop.call_soon_threadsafe(self.async_stop_event.set)

    def is_running(self) -> bool:
        if self._force_stopped:
            return False
        return self.current_thread is not None and self.current_thread.is_alive()

    def start_check(self, args):
        if self.is_running(): return
        self._force_stopped = False
        cfg.GLOBAL_STOP = False
        cfg.POOL_EXHAUSTED = False
        self.thread_stop_event.clear()
        self.current_thread = threading.Thread(
            target=self._run_check_in_thread, args=(args,), daemon=True
        )
        self.current_thread.start()

    def _run_check_in_thread(self, args):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        try:
            self.async_stop_event = asyncio.Event()
            self.loop.run_until_complete(manual_check_main_loop(args, self.async_stop_event))
        finally:
            self.loop.close()
            self._force_stopped = True

if __name__ == "__main__":
    main()