import os  # 导入系统模块用于处理文件路径。 
import time  # 导入时间模块用于记录状态时间。 
import json  # 导入 JSON 模块用于会话 token 载荷。 
import hmac  # 导入 HMAC 模块用于签名比对。 
import base64  # 导入 base64 模块用于编码 token。 
import hashlib  # 导入哈希模块用于 token 签名。 
import secrets  # 导入随机模块用于 token nonce。 
import shutil  # 导入文件复制模块用于准备天翼云运行目录。 
import subprocess  # 导入子进程模块用于隔离每个账号的购买任务。 
import sys  # 导入系统模块用于获取当前 Python 解释器。 
import threading  # 导入线程模块用于后台执行流程。 
from typing import Any, Optional  # 导入请求体类型。 
from fastapi import FastAPI, Request  # 导入 FastAPI 与 Request 用于构建接口与读取会话。 
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, PlainTextResponse  # 导入响应类型用于返回文件、页面、脚本与 JSON。 
from starlette.middleware.sessions import SessionMiddleware  # 导入 Session 中间件用于登录态管理。 
from pydantic import BaseModel  # 导入数据模型用于校验请求体。 
import config as _config  # 导入配置模块用于读取会话与数据库参数。 
from auth_service import init_db, ensure_default_users, authenticate_user, update_password, is_default_password  # 导入认证服务方法。 
import script_share_service  # 导入脚本分享服务。 
import performance_service  # 导入业绩数据服务。 
import ctyun_agent_service  # 导入天翼云代理后台同步服务。 
import volc_agent_service  # 导入火山引擎代理后台同步服务。 
import huawei_agent_service  # 导入华为云代理后台同步服务。 
import huawei_customer_guard  # 导入华为云客户归属校验服务。 
import iam_service  # 导入华为云 IAM 身份查询服务。 
import agent_browser_cookie_service  # 导入天翼云和火山引擎浏览器 Cookie 状态服务。 
import huawei_browser_cookie_service  # 导入华为云浏览器 Cookie 状态服务。 
from proxy_test import service as proxy_test_service  # 导入 SOCKS5 IP 连通性测试服务。 
APP_ROOT = os.path.dirname(os.path.abspath(__file__))  # 计算当前文件目录作为应用根路径。 


def _first_existing_path(*paths: str) -> str:  # 选择第一个存在的路径，部署迁移时用于兼容新旧目录。 
    for path in paths: 
        if path and os.path.exists(path): 
            return path 
    return paths[0] 


PROJECT_ROOT = os.getenv( 
    "WAWA_PROJECT_ROOT", 
    os.path.abspath(os.path.join(APP_ROOT, "..", "..", "..", "..")), 
)  # 项目根目录，本机为 Atuo_Ecs，服务器为 /www/wwwroot。 
WWWROOT = os.getenv( 
    "WAWA_WWWROOT", 
    _first_existing_path(os.path.join(PROJECT_ROOT, "wwwroot"), PROJECT_ROOT), 
)  # 兼容旧 wwwroot 目录和服务器 /www/wwwroot。 
STORAGE_ROOT = os.getenv( 
    "WAWA_STORAGE_DIR", 
    _first_existing_path(os.path.join(WWWROOT, "storage"), os.path.join(PROJECT_ROOT, "storage")), 
)  # 运行数据目录，保存 Cookie、浏览器 profile、业绩库和任务工作区。 
FRONTEND_ROOT = os.getenv( 
    "WAWA_FRONTEND_DIR", 
    _first_existing_path(os.path.join(PROJECT_ROOT, "frontend"), os.path.join(WWWROOT, "wawaok.cn"), APP_ROOT), 
)  # 前端目录，优先使用重构后的 frontend。 
CTYUN_APP_ROOT = os.getenv("WAWA_CTYUN_DIR", os.path.abspath(os.path.join(APP_ROOT, "..", "ctyun")))  # 天翼云购买脚本目录。 
CTYUN_REGIONS_SOURCE = os.path.join(CTYUN_APP_ROOT, "regions_10ip.json")  # 天翼云地区配置源文件。 
CTYUN_WORKSPACE_ROOT = os.path.join(STORAGE_ROOT, "ctyun_runs")  # 天翼云每次任务的独立运行目录。 
INDEX_PATH = os.path.join(FRONTEND_ROOT, "index.html")  # 拼接前端页面路径用于返回静态页面。 
LOGIN_PATH = os.path.join(FRONTEND_ROOT, "login.html")  # 拼接登录页面路径用于返回静态页面。 
MAX_LOG_LINES = 2000  # 日志缓冲区最多保留的行数。 
LOG_BUFFER_BY_USER = {}  # 按账号区分的日志缓冲区字典。 
LOG_LOCK = threading.Lock()  # 日志锁用于多线程安全写入。 
STATE_BY_USER = {}  # 按账号区分的运行状态字典。 
STATE_LOCK = threading.Lock()  # 状态锁用于多线程安全更新。 
PERFORMANCE_SYNC_LOCK = threading.Lock()  # 业绩同步锁，避免手动和自动同步同时执行。 
AUTO_SYNC_LOCK = threading.Lock()  # 自动同步状态锁。 
AUTO_SYNC_STOP = threading.Event()  # 自动同步停止信号。 
AUTO_SYNC_THREAD = None  # 自动同步线程句柄。 
AUTO_SYNC_INTERVAL_SECONDS = 3600  # 每小时同步一次。 
AUTO_SYNC_DAILY_HOUR = 23  # 每天 23:59 做一次当天收尾同步。 
AUTO_SYNC_DAILY_MINUTE = 59  # 每天收尾同步分钟。 
AUTO_SYNC_STATE = {  # 自动同步运行状态。 
    "enabled": True, 
    "running": False, 
    "last_started_at": 0, 
    "last_finished_at": 0, 
    "last_reason": "", 
    "last_results": [], 
    "last_ok": None, 
    "last_message": "", 
    "next_sync_at": 0, 
    "next_reason": "", 
} 
SESSION_TOKEN_TTL_SECONDS = int(os.getenv("WAWACLOUD_SESSION_TOKEN_TTL_SECONDS", "604800") or "604800")  # 标签页 token 默认 7 天有效。 
app = FastAPI(title="wawacloud 华为云自动购买")  # 创建 FastAPI 应用实例。 
app.add_middleware(  # 添加会话中间件用于登录态管理。 
    SessionMiddleware,  # 使用 Starlette 提供的 Session 中间件。 
    secret_key=_config.APP_SESSION_SECRET,  # 设置会话密钥用于 Cookie 加密。 
    session_cookie=_config.APP_SESSION_COOKIE,  # 设置会话 Cookie 名称。 
    same_site="lax",  # 设置 SameSite 保护跨站请求。 
    https_only=_config.APP_SESSION_HTTPS_ONLY,  # 设置是否仅 HTTPS 发送 Cookie。 
)  # 会话中间件配置结束。 
class RunRequest(BaseModel):  # 运行接口请求体模型定义。 
    regions: list[str] = []  # 选择的地区列表，空表示不运行。 
    ak: Optional[str] = None  # 可选 AK 字段，用于单次任务覆盖。 
    sk: Optional[str] = None  # 可选 SK 字段，用于单次任务覆盖。 
    mode: Optional[str] = None  # 可选购买模式，delta 表示三角洲模式。 
    manual_confirm_customer: Optional[bool] = False  # 代理后台掉线时的人工确认放行。 
class LoginRequest(BaseModel):  # 登录接口请求体模型定义。 
    account: str = ""  # 登录账号字段。 
    password: str = ""  # 登录密码字段。 
class ResetPasswordRequest(BaseModel):  # 修改密码接口请求体模型定义。 
    account: str = ""  # 账号字段。 
    old_password: str = ""  # 旧密码字段。 
    new_password: str = ""  # 新密码字段。 
class ScriptShareRequest(BaseModel):  # 脚本分享创建请求模型。 
    script_key: str = "huawei"  # 脚本类型：huawei / volc / ctyun。 
    validity: str = "7day"  # 有效期选项。 
    max_uses: Optional[str] = "7"  # 最大使用次数选项。 
class PerformanceImportRequest(BaseModel):  # 业绩 JSON 导入请求模型。 
    provider: str = ""  # 云厂商：ctyun / huawei / volc。 
    payload: Any = None  # 本地登录后台后导出的 JSON 数据。 
class CtyunCleanupRequest(BaseModel):  # 天翼云资源管理请求模型。 
    regions: list[str] = []  # 选择的地区列表，空时资源查询默认全部地区。 
    ak: Optional[str] = None  # 天翼云 AK。 
    sk: Optional[str] = None  # 天翼云 SK。 
    items: list[dict[str, Any]] = []  # 退订或重装时前端勾选的资源。 
class ProxyTestRequest(BaseModel):  # IP 测试请求模型。 
    proxies: str = ""  # 每行一条 IP/端口/账号/密码。 
def _get_log_buffer(user: str):  # 获取指定账号的日志缓冲区。 
    key = user or "_system"  # 为空时归档到系统日志。 
    with LOG_LOCK:  # 加锁保证线程安全。 
        return LOG_BUFFER_BY_USER.setdefault(key, [])  # 返回日志列表。 
def _append_log(line, user: str = ""):  # 追加日志行到缓冲区的函数。 
    text = (line or "").rstrip()  # 取出文本并去掉行尾换行。 
    if not text:  # 若为空则直接返回。 
        return  # 空行不写入日志缓冲。 
    buffer = _get_log_buffer(user)  # 获取日志列表。 
    with LOG_LOCK:  # 加锁保证线程安全。 
        buffer.append(text)  # 追加日志行到缓冲区。 
        overflow = len(buffer) - MAX_LOG_LINES  # 计算超过最大行数的数量。 
        if overflow > 0:  # 若超出最大行数则裁剪。 
            del buffer[:overflow]  # 删除最旧的日志行保持缓冲大小。 
def _blank_state():  # 构造默认状态字典。 
    return {"running": False, "error": "", "started_at": None, "finished_at": None}  # 返回默认状态。 


def _b64_encode(raw: bytes) -> str:  # URL 安全 base64 编码。 
    return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")  # 去掉填充便于放入请求头。 


def _b64_decode(value: str) -> bytes:  # URL 安全 base64 解码。 
    padding = "=" * (-len(value) % 4)  # 补齐 base64 填充。 
    return base64.urlsafe_b64decode((value + padding).encode("ascii"))  # 返回原始字节。 


def _token_signature(payload_b64: str) -> str:  # 计算 token 签名。 
    secret = str(_config.APP_SESSION_SECRET or "CHANGE_ME_TO_RANDOM_SECRET").encode("utf-8")  # 读取统一密钥。 
    digest = hmac.new(secret, payload_b64.encode("ascii"), hashlib.sha256).digest()  # HMAC-SHA256 签名。 
    return _b64_encode(digest)  # 返回 URL 安全签名。 


def _issue_session_token(user: str) -> str:  # 给当前标签页签发独立 token。 
    now = int(time.time())  # 当前时间戳。 
    payload = { 
        "user": user, 
        "iat": now, 
        "exp": now + max(3600, SESSION_TOKEN_TTL_SECONDS), 
        "nonce": secrets.token_urlsafe(12), 
    } 
    payload_b64 = _b64_encode(json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")) 
    return f"{payload_b64}.{_token_signature(payload_b64)}"  # 返回 payload.signature。 


def _token_user(request: Request) -> str:  # 从请求头读取标签页 token 对应账号。 
    token = request.headers.get("x-wawa-session") or request.headers.get("x-wawacloud-session") or "" 
    token = token.strip() 
    if not token or "." not in token: 
        return "" 
    payload_b64, signature = token.rsplit(".", 1) 
    if not payload_b64 or not signature: 
        return "" 
    if not hmac.compare_digest(signature, _token_signature(payload_b64)): 
        return "" 
    try: 
        payload = json.loads(_b64_decode(payload_b64).decode("utf-8")) 
    except Exception: 
        return "" 
    try: 
        if int(payload.get("exp") or 0) < int(time.time()): 
            return "" 
    except (TypeError, ValueError): 
        return "" 
    return str(payload.get("user") or "").strip() 


def _get_state_key(user: str):  # 获取状态字典键。 
    return user or "_system"  # 用户为空时归档到系统键。 

def _service_state_key(service: str, user: str) -> str:  # 生成带服务名的状态键。 
    return f"{service}:{_get_state_key(user)}"  # 避免天翼云任务与华为云任务共用同一状态。 


def _service_user_from_key(key: str) -> str:  # 从服务状态键提取真实账号。 
    return key.split(":", 1)[1] if ":" in key else key  # 无服务前缀时直接返回原值。 


def _ensure_state_unlocked(user: str):  # 在已持锁前提下获取账号状态。 
    return STATE_BY_USER.setdefault(_get_state_key(user), _blank_state())  # 初始化并返回状态。 


def _set_state(user: str, running=None, error=None, started_at=None, finished_at=None):  # 更新指定账号状态。 
    with STATE_LOCK:  # 加锁保证状态更新线程安全。 
        state = _ensure_state_unlocked(user)  # 获取指定账号状态。 
        if running is not None:  # 若传入运行状态则更新。 
            state["running"] = running  # 更新运行状态字段。 
        if error is not None:  # 若传入错误信息则更新。 
            state["error"] = error  # 更新错误信息字段。 
        if started_at is not None:  # 若传入开始时间则更新。 
            state["started_at"] = started_at  # 更新开始时间字段。 
        if finished_at is not None:  # 若传入结束时间则更新。 
            state["finished_at"] = finished_at  # 更新结束时间字段。 


def _set_state_fields(user: str, **updates):  # 批量写入状态附加字段。 
    with STATE_LOCK:  # 加锁保证状态一致。 
        state = _ensure_state_unlocked(user)  # 获取状态字典。 
        state.update(updates)  # 写入附加状态。 


def _reserve_huawei_task(user: str) -> bool:  # 原子占用当前账号的华为云任务槽位。 
    with STATE_LOCK: 
        state = _ensure_state_unlocked(user) 
        if state.get("running"): 
            return False 
        state.update( 
            { 
                "running": True, 
                "error": "", 
                "started_at": time.time(), 
                "finished_at": None, 
            } 
        ) 
    return True 


def _summarize_errors(errors):  # 将完整 SDK 异常压缩成适合状态栏展示的短错误。 
    if not errors:  # 没有错误时返回空。 
        return ""  # 返回空字符串。 
    joined = "; ".join(str(item) for item in errors)  # 拼接完整错误用于判断类型。 
    lower = joined.lower()  # 统一小写便于匹配。 
    regions = []  # 收集异常地区。 
    for item in errors:  # 遍历地区错误。 
        region = str(item).split(":", 1)[0].strip()  # 取冒号前地区名。 
        if region and region not in regions:  # 去重加入。 
            regions.append(region) 
    region_text = "、".join(regions[:4])  # 状态栏最多展示前 4 个地区。 
    if len(regions) > 4:  # 地区过多时加省略。 
        region_text += f" 等{len(regions)}个地区" 
    if "apigw.0301" in lower or "verify ak sk signature failed" in lower or "status_code:401" in lower: 
        return f"{region_text}: AK/SK 认证失败，请检查访问密钥" if region_text else "AK/SK 认证失败，请检查访问密钥" 
    if "failed to get domain id" in lower: 
        return f"{region_text}: 获取华为云账号 Domain ID 失败" if region_text else "获取华为云账号 Domain ID 失败" 
    return f"{region_text}: 任务异常，详情见实时日志" if region_text else "任务异常，详情见实时日志" 


def _summarize_guard_exception(provider: str, exc: Exception) -> str:  # 压缩客户归属校验异常。 
    text = str(exc)  # 转成字符串用于匹配。 
    lower = text.lower()  # 小写后便于判断。 
    if "apigw.0301" in lower or "verify ak sk signature failed" in lower or "status_code:401" in lower: 
        return f"{provider} AK/SK 认证失败，请检查访问密钥" 
    short = text if len(text) <= 180 else text[:180] + "..." 
    return f"{provider}客户归属校验失败: {short}" 


def _manual_confirm_response(message: str, user: str) -> JSONResponse:  # 返回人工确认弹窗提示。 
    _append_log(f"步骤: 客户归属需要人工确认 {message}", user) 
    _set_state(user, running=False, error=message, finished_at=time.time()) 
    return JSONResponse( 
        { 
            "ok": False, 
            "message": message, 
            "manual_confirm_available": True, 
            "manual_confirm_text": "已手动确认用户继续执行？", 
        }, 
        status_code=403, 
    ) 


def _get_state_snapshot(user: str):  # 获取指定账号的状态快照。 
    with STATE_LOCK:  # 加锁保证读取一致。 
        return dict(_ensure_state_unlocked(user))  # 返回状态副本避免外部修改。 
def _get_session_user(request: Request) -> str:  # 获取当前登录账号的函数。 
    token_user = _token_user(request)  # 优先使用标签页独立 token，避免同浏览器多账号串号。 
    if token_user: 
        return token_user 
    if not hasattr(request, "session"):  # 若请求对象不含会话则返回空。 
        return ""  # 返回空字符串表示未登录。 
    return str(request.session.get("user") or "")  # 读取会话中的用户名并返回。 
def _require_login(request: Request):  # 登录校验辅助函数。 
    user = _get_session_user(request)  # 获取当前会话用户。 
    if not user:  # 若用户为空则返回未登录响应。 
        return JSONResponse({"ok": False, "message": "未登录或登录已过期"}, status_code=401)  # 返回未登录错误。 
    return None  # 登录通过时返回空。 

def _require_password_reset(request: Request):  # 检查是否需要修改默认密码。 
    token_user = _token_user(request)  # 标签页 token 场景下实时查询默认密码状态。 
    if token_user: 
        try: 
            if is_default_password(token_user): 
                return JSONResponse({"ok": False, "message": "请先修改默认密码"}, status_code=403) 
        except Exception as exc: 
            return JSONResponse({"ok": False, "message": f"数据库异常: {exc}"}, status_code=500) 
        return None 
    if not hasattr(request, "session"):  # 若请求对象不含会话则跳过。 
        return None  # 无会话信息时不拦截。 
    if request.session.get("force_reset"):  # 若仍为默认密码则阻止操作。 
        return JSONResponse({"ok": False, "message": "请先修改默认密码"}, status_code=403)  # 返回禁止响应。 
    return None  # 无需拦截。 
def _public_base_url(request: Request) -> str:  # 生成给客户使用的外部基础地址。 
    host = request.headers.get("host") or request.url.netloc  # 优先使用代理后的 Host。 
    proto = request.headers.get("x-forwarded-proto") or request.url.scheme or "https"  # 读取协议。 
    if "wawaok.cn" in host:  # 线上域名固定使用 HTTPS。 
        proto = "https" 
    return f"{proto}://{host}".rstrip("/")  # 返回外部基础 URL。 
def _run_task(user, regions, ak, sk, mode=None):  # 后台执行流程任务的函数。 
    _set_state(user, running=True, error="", started_at=time.time(), finished_at=None)  # 设置任务开始状态。 
    try:  # 捕获执行过程中的异常。 
        mode_key = (mode or "").strip().lower()  # 标准化购买模式。 
        mode_label = f" 模式={mode_key}" if mode_key else ""  # 构造日志中的模式说明。 
        _append_log(f"步骤: 接口启动 地区={regions}{mode_label}", user)  # 输出 API 启动日志。 
        env = os.environ.copy()  # 复制当前环境，避免修改主服务进程。 
        env["PYTHONUNBUFFERED"] = "1"  # 强制子进程实时输出日志。 
        env["HUAWEI_REGIONS"] = ",".join(regions)  # 传递本次任务选择的地区。 
        if mode_key:  # 若前端指定购买模式则传给子进程。 
            env["HUAWEI_PURCHASE_MODE"] = mode_key  # 设置购买模式环境变量。 
        else:  # 未指定时清理可能残留的模式环境变量。 
            env.pop("HUAWEI_PURCHASE_MODE", None)  # 清理购买模式环境变量。 
        if ak and sk:  # 若本次请求传入 AK/SK，则只注入到子进程。 
            env["HUAWEI_AK"] = ak  # 设置子进程 AK。 
            env["HUAWEI_SK"] = sk  # 设置子进程 SK。 
        else:  # 未传入时清理可能残留的环境变量。 
            env.pop("HUAWEI_AK", None)  # 清理 AK 环境变量。 
            env.pop("HUAWEI_SK", None)  # 清理 SK 环境变量。 
        command = [sys.executable, "-u", "main.py"]  # 使用当前服务解释器启动购买脚本。 
        process = subprocess.Popen(  # 启动独立子进程，避免多个账号共享 stdout。 
            command, 
            cwd=APP_ROOT, 
            env=env, 
            stdout=subprocess.PIPE, 
            stderr=subprocess.STDOUT, 
            text=True, 
            encoding="utf-8", 
            errors="replace", 
            bufsize=1, 
        ) 
        error_items = []  # 记录子进程输出的错误汇总，供状态栏压缩展示。 
        if process.stdout is not None:  # 理论上 stdout 一定存在。 
            for line in process.stdout:  # 逐行读取子进程输出，实时写入当前账号日志。 
                text = line.rstrip("\r\n")  # 去掉换行符。 
                _append_log(text, user)  # 写入当前账号独立日志缓冲。 
                marker = "步骤: 错误汇总 "  # main.py 的错误汇总标记。 
                if marker in text:  # 捕获地区错误汇总。 
                    summary_text = text.split(marker, 1)[1].strip()  # 取出汇总内容。 
                    error_items.extend(item.strip() for item in summary_text.split(";") if item.strip())  # 拆分地区错误。 
        exit_code = process.wait()  # 等待子进程结束并获取退出码。 
        if exit_code:  # 非零退出表示至少一个地区失败或脚本异常。 
            message = _summarize_errors(error_items) if error_items else f"exit_code={exit_code}"  # 状态栏展示短错误。 
            _set_state(user, error=message)  # 保存错误信息到当前账号状态。 
            _append_log(f"步骤: 接口异常 exit_code={exit_code}", user)  # 记录退出码到日志。 
            return  # 异常场景直接结束。 
        _append_log("步骤: 接口完成", user)  # 输出 API 完成日志。 
    except Exception as exc:  # 捕获执行异常并记录。 
        _append_log(f"步骤: 接口异常 {exc}", user)  # 记录异常信息到日志。 
        _set_state(user, error=str(exc))  # 保存错误信息到状态。 
    finally:  # 无论成功失败均更新结束状态。 
        _set_state(user, running=False, finished_at=time.time())  # 设置任务结束状态。 


CTYUN_PURCHASE_PHASES = [  # 天翼云完整购买阶段。 
    "buy-eip", 
    "has-active-regions", 
    "discover", 
    "ensure-network", 
    "buy-ecs", 
    "havip-bind", 
    "ecs-eip-security", 
    "export", 
] 
CTYUN_PHASE_TITLES = {  # 天翼云阶段展示名。 
    "ensure-network": "准备 VPC/子网/安全组", 
    "discover": "读取地区参数", 
    "buy-eip": "购买弹性 IP", 
    "has-active-regions": "检查可继续地区", 
    "buy-ecs": "购买云服务器", 
    "havip-bind": "创建并绑定子网虚拟 IP", 
    "ecs-eip-security": "绑定云服务器 IP 并完善安全组", 
    "cleanup-query": "查询待退订/重装资源", 
    "cleanup-execute": "退订勾选资源", 
    "reinstall-execute": "重装勾选云服务器", 
    "export": "导出运行结果", 
} 


def _safe_slug(value: str) -> str:  # 生成适合文件夹使用的短名称。 
    cleaned = "".join(ch if ch.isalnum() or ch in ("-", "_") else "_" for ch in str(value or "")) 
    return (cleaned or "system")[:80] 


def _ctyun_all_region_names() -> list[str]:  # 读取天翼云全部可选地区。 
    with open(CTYUN_REGIONS_SOURCE, "r", encoding="utf-8-sig") as f: 
        data = json.load(f) 
    return [str(item.get("displayName") or "").strip() for item in data.get("regions", []) if item.get("displayName")] 


def _ctyun_regions(payload_regions: list[str], allow_all: bool = False) -> list[str]:  # 解析天翼云地区列表。 
    regions = [str(item or "").strip() for item in (payload_regions or []) if str(item or "").strip()] 
    if regions: 
        return regions 
    if allow_all: 
        return _ctyun_all_region_names() 
    return [] 


def _require_ctyun_credentials(ak: str, sk: str):  # 校验天翼云 AK/SK。 
    if not ak or not sk: 
        return JSONResponse({"ok": False, "message": "天翼云自动购买必须填写 AK 和 SK"}, status_code=400) 
    if (ak and not sk) or (sk and not ak): 
        return JSONResponse({"ok": False, "message": "AK 和 SK 需要同时填写"}, status_code=400) 
    return None 


def _prepare_ctyun_workspace(user: str, task_label: str) -> str:  # 为每次天翼云任务准备独立目录。 
    if not os.path.isdir(CTYUN_APP_ROOT): 
        raise RuntimeError(f"天翼云脚本目录不存在: {CTYUN_APP_ROOT}") 
    if not os.path.isfile(CTYUN_REGIONS_SOURCE): 
        raise RuntimeError(f"天翼云地区配置不存在: {CTYUN_REGIONS_SOURCE}") 
    user_dir = os.path.join(CTYUN_WORKSPACE_ROOT, _safe_slug(user)) 
    os.makedirs(user_dir, exist_ok=True) 
    stamp = time.strftime("%Y%m%d-%H%M%S") 
    workspace = os.path.join(user_dir, f"{stamp}-{_safe_slug(task_label)}-{secrets.token_hex(4)}") 
    os.makedirs(workspace, exist_ok=True) 
    shutil.copyfile(CTYUN_REGIONS_SOURCE, os.path.join(workspace, "regions_10ip.json")) 
    return workspace 


def _write_ctyun_input(workspace: str, ak: str, sk: str, regions: list[str]) -> str:  # 写入当前任务专用输入文件。 
    input_path = os.path.join(workspace, "input.txt") 
    with open(input_path, "w", encoding="utf-8") as f: 
        f.write((ak or "").strip() + "\n") 
        f.write((sk or "").strip() + "\n") 
        for region in regions[:2]: 
            f.write(region + "\n") 
    return input_path 


def _reserve_ctyun_task(user: str, task_label: str) -> tuple[str, str]:  # 原子占用当前账号的天翼云任务槽位。 
    service_user = _service_state_key("ctyun", user) 
    with STATE_LOCK: 
        state = _ensure_state_unlocked(service_user) 
        if state.get("running"): 
            return "", user 
        state.update( 
            { 
                "running": True, 
                "error": "", 
                "started_at": time.time(), 
                "finished_at": None, 
                "task": task_label, 
                "workspace": "", 
            } 
        ) 
    return service_user, "" 


def _run_ctyun_phase(service_user: str, workspace: str, input_path: str, regions: list[str], phase: str, selection_file: str = "") -> int:  # 执行天翼云单阶段。 
    title = CTYUN_PHASE_TITLES.get(phase, phase) 
    _append_log(f"步骤: 天翼云{title}开始", service_user) 
    command = [ 
        sys.executable, 
        "-u", 
        "workflow_2region.py", 
        "--input-file", 
        input_path, 
        "--regions", 
        ",".join(regions), 
        "--phase", 
        phase, 
        "--execute", 
    ] 
    if selection_file: 
        command.extend(["--cleanup-selection-file", selection_file]) 
    env = os.environ.copy() 
    env["PYTHONUNBUFFERED"] = "1" 
    env["PYTHONUTF8"] = "1" 
    env["PYTHONIOENCODING"] = "utf-8" 
    env["CTYUN_WORKFLOW_ROOT"] = workspace 
    env["CTYUN_WORKFLOW_REGIONS_FILE"] = os.path.join(workspace, "regions_10ip.json") 
    process = subprocess.Popen( 
        command, 
        cwd=CTYUN_APP_ROOT, 
        env=env, 
        stdout=subprocess.PIPE, 
        stderr=subprocess.STDOUT, 
        text=True, 
        encoding="utf-8", 
        errors="replace", 
        bufsize=1, 
    ) 
    if process.stdout is not None: 
        for line in process.stdout: 
            _append_log(line.rstrip("\r\n"), service_user) 
    exit_code = process.wait() 
    if exit_code: 
        _append_log(f"步骤: 天翼云{title}异常 exit_code={exit_code}", service_user) 
    else: 
        _append_log(f"步骤: 天翼云{title}完成", service_user) 
    return exit_code 


def _run_ctyun_workflow(user: str, service_user: str, regions: list[str], ak: str, sk: str, phases: list[str], task_label: str, items: Optional[list[dict[str, Any]]] = None):  # 后台执行天翼云任务。 
    input_path = "" 
    try: 
        workspace = _prepare_ctyun_workspace(user, task_label) 
        _set_state_fields(service_user, workspace=workspace, task=task_label) 
        input_path = _write_ctyun_input(workspace, ak, sk, regions) 
        selection_file = "" 
        if items is not None: 
            selection_file = os.path.join(workspace, "cleanup_selection.json") 
            with open(selection_file, "w", encoding="utf-8") as f: 
                json.dump(items, f, ensure_ascii=False, indent=2) 
                f.write("\n") 
        _append_log(f"步骤: 天翼云任务启动 类型={task_label} 地区={regions}", service_user) 
        for phase in phases: 
            code = _run_ctyun_phase(service_user, workspace, input_path, regions, phase, selection_file=selection_file) 
            if phase == "has-active-regions" and code == 2: 
                _append_log("步骤: 天翼云所有地区已跳过，开始导出结果", service_user) 
                _run_ctyun_phase(service_user, workspace, input_path, regions, "export") 
                _set_state(service_user, error="所有地区均已跳过") 
                return 
            if code: 
                _set_state(service_user, error=f"{CTYUN_PHASE_TITLES.get(phase, phase)}失败 exit_code={code}") 
                return 
        _append_log("步骤: 天翼云任务完成", service_user) 
    except Exception as exc: 
        _append_log(f"步骤: 天翼云任务异常 {exc}", service_user) 
        _set_state(service_user, error=str(exc)) 
    finally: 
        if input_path: 
            try: 
                os.remove(input_path) 
            except OSError: 
                pass 
        _set_state(service_user, running=False, finished_at=time.time()) 


def _latest_ctyun_workspace(user: str) -> str:  # 获取当前账号最近一次天翼云工作目录。 
    service_user = _service_state_key("ctyun", user) 
    with STATE_LOCK: 
        workspace = str(_ensure_state_unlocked(service_user).get("workspace") or "") 
    if workspace and os.path.isdir(workspace): 
        return workspace 
    user_dir = os.path.join(CTYUN_WORKSPACE_ROOT, _safe_slug(user)) 
    if not os.path.isdir(user_dir): 
        return "" 
    children = [os.path.join(user_dir, name) for name in os.listdir(user_dir)] 
    dirs = [path for path in children if os.path.isdir(path)] 
    return max(dirs, key=os.path.getmtime) if dirs else "" 


def _start_ctyun_background_task(request: Request, payload, phases: list[str], task_label: str, allow_all_regions: bool = False, items: Optional[list[dict[str, Any]]] = None):  # 启动天翼云后台任务。 
    auth_error = _require_login(request) 
    if auth_error: 
        return auth_error 
    reset_error = _require_password_reset(request) 
    if reset_error: 
        return reset_error 
    user = _get_session_user(request) 
    regions = _ctyun_regions(payload.regions or [], allow_all=allow_all_regions) 
    if not regions: 
        return JSONResponse({"ok": False, "message": "请至少选择一个天翼云地区"}, status_code=400) 
    ak = (payload.ak or "").strip() 
    sk = (payload.sk or "").strip() 
    credential_error = _require_ctyun_credentials(ak, sk) 
    if credential_error: 
        return credential_error 
    if items is not None and not items: 
        return JSONResponse({"ok": False, "message": "请先勾选需要处理的资源"}, status_code=400) 
    service_user, running_user = _reserve_ctyun_task(user, task_label) 
    if running_user: 
        return JSONResponse( 
            {"ok": False, "message": "本账号天翼云任务正在运行，请等待执行完毕", "running_user": running_user}, 
            status_code=409, 
        ) 
    thread = threading.Thread( 
        target=_run_ctyun_workflow, 
        args=(user, service_user, regions, ak, sk, phases, task_label, items), 
        daemon=True, 
    ) 
    try: 
        thread.start() 
    except Exception as exc: 
        _set_state(service_user, running=False, error=str(exc), finished_at=time.time()) 
        return JSONResponse({"ok": False, "message": f"启动天翼云任务失败: {exc}"}, status_code=500) 
    return {"ok": True, "message": "天翼云任务已启动"} 


def _sync_provider_metrics(wait: bool = False) -> dict[str, Any]:  # 同步三家云代理后台业绩。 
    acquired = PERFORMANCE_SYNC_LOCK.acquire(blocking=wait)  # 尝试获取同步锁。 
    if not acquired:  # 已有同步任务时直接返回。 
        return {"ok": False, "running": True, "message": "已有业绩同步任务正在执行", "results": [], "summary": performance_service.summary()} 
    providers = [  # 三家云同步入口。 
        ("ctyun", "天翼云", ctyun_agent_service.sync_metrics), 
        ("huawei", "华为云", huawei_agent_service.sync_metrics), 
        ("volc", "火山引擎", volc_agent_service.sync_metrics), 
    ] 
    results = []  # 记录每家云同步结果。 
    try: 
        for key, label, sync_func in providers: 
            try: 
                result = sync_func() 
                metrics = result["metrics"] 
                performance_service.upsert_provider_metrics(key, label, metrics) 
                results.append({"provider": key, "provider_label": label, "ok": True, "metrics": metrics}) 
            except Exception as exc: 
                results.append({"provider": key, "provider_label": label, "ok": False, "message": str(exc)}) 
        return {"ok": all(item["ok"] for item in results), "results": results, "summary": performance_service.summary()} 
    finally: 
        PERFORMANCE_SYNC_LOCK.release() 


def _next_daily_sync_ts(now: Optional[float] = None) -> float:  # 计算下一次每日收尾同步时间。 
    value = time.time() if now is None else now 
    local = time.localtime(value) 
    target = time.mktime(( 
        local.tm_year, local.tm_mon, local.tm_mday, 
        AUTO_SYNC_DAILY_HOUR, AUTO_SYNC_DAILY_MINUTE, 0, 
        local.tm_wday, local.tm_yday, local.tm_isdst, 
    )) 
    if target <= value: 
        target += 24 * 60 * 60 
    return target 


def _next_hourly_sync_ts(now: Optional[float] = None) -> float:  # 计算下一次整点同步时间。 
    value = time.time() if now is None else now 
    local = time.localtime(value) 
    current_hour = time.mktime(( 
        local.tm_year, local.tm_mon, local.tm_mday, local.tm_hour, 0, 0, 
        local.tm_wday, local.tm_yday, local.tm_isdst, 
    )) 
    return current_hour + AUTO_SYNC_INTERVAL_SECONDS 


def _calculate_next_auto_sync(now: Optional[float] = None) -> tuple[float, str]:  # 计算下一次自动同步。 
    value = time.time() if now is None else now 
    hourly = _next_hourly_sync_ts(value) 
    daily = _next_daily_sync_ts(value) 
    if daily <= hourly: 
        return daily, "daily_2359" 
    return hourly, "hourly" 


def _set_auto_sync_state(**updates):  # 更新自动同步状态。 
    with AUTO_SYNC_LOCK: 
        AUTO_SYNC_STATE.update(updates) 


def _get_auto_sync_state() -> dict[str, Any]:  # 获取自动同步状态。 
    with AUTO_SYNC_LOCK: 
        return dict(AUTO_SYNC_STATE) 


def _cookie_file_mtime(path: str) -> int:  # 读取 Cookie 文件更新时间。 
    try: 
        return int(os.path.getmtime(path)) if path and os.path.isfile(path) else 0 
    except OSError: 
        return 0 


def _agent_provider_status(provider: str, label: str) -> dict[str, Any]:  # 查询代理后台登录态是否有效。 
    checked_at = int(time.time())  # 记录检查时间。 
    try: 
        if provider == "huawei":  # 华为云使用独立浏览器 Cookie 服务。 
            cookies = huawei_browser_cookie_service.refresh_cookies_from_browser_profile() 
            cookie_path = getattr(huawei_browser_cookie_service, "COOKIE_PATH", "") 
        else:  # 天翼云和火山引擎使用统一代理浏览器 Cookie 服务。 
            cookies = agent_browser_cookie_service.refresh_cookies_from_browser(provider) 
            provider_config = getattr(agent_browser_cookie_service, "PROVIDERS", {}).get(provider, {}) 
            cookie_path = str(provider_config.get("cookie_path") or "") 
        if provider == "ctyun":  # 天翼云实际读取代理后台指标，确认 Cookie 不是空壳。 
            data_result = ctyun_agent_service.sync_metrics() 
        elif provider == "huawei":  # 华为云实际读取伙伴中心指标，确认登录态有效。 
            data_result = huawei_agent_service.sync_metrics() 
        elif provider == "volc":  # 火山引擎实际读取代理后台指标，确认登录态有效。 
            data_result = volc_agent_service.sync_metrics() 
        else: 
            data_result = {} 
        metrics = data_result.get("metrics", {}) if isinstance(data_result, dict) else {} 
        return { 
            "provider": provider, 
            "provider_label": label, 
            "ok": True, 
            "status": "valid", 
            "message": "登录态和数据接口有效", 
            "cookie_count": len(cookies), 
            "checked_at": checked_at, 
            "cookie_updated_at": _cookie_file_mtime(cookie_path), 
            "month_performance": metrics.get("month_performance"), 
            "today_performance": metrics.get("today_performance"), 
            "customers_total": metrics.get("customers_total"), 
        } 
    except Exception as exc: 
        return { 
            "provider": provider, 
            "provider_label": label, 
            "ok": False, 
            "status": "invalid", 
            "message": str(exc), 
            "cookie_count": 0, 
            "checked_at": checked_at, 
            "cookie_updated_at": 0, 
        } 


def _run_auto_sync(reason: str) -> None:  # 执行一次自动同步。 
    started_at = int(time.time()) 
    _set_auto_sync_state(running=True, last_started_at=started_at, last_reason=reason, last_message="自动同步运行中") 
    _append_log(f"步骤: 自动业绩同步开始 reason={reason}", "_system") 
    result = _sync_provider_metrics(wait=False) 
    finished_at = int(time.time()) 
    message = result.get("message") or ("自动同步完成" if result.get("ok") else "自动同步部分失败") 
    _set_auto_sync_state( 
        running=False, 
        last_finished_at=finished_at, 
        last_ok=result.get("ok"), 
        last_results=result.get("results", []), 
        last_message=message, 
    ) 
    _append_log(f"步骤: 自动业绩同步结束 ok={result.get('ok')} message={message}", "_system") 


def _auto_sync_loop() -> None:  # 自动同步后台循环。 
    while not AUTO_SYNC_STOP.is_set(): 
        next_sync_at, reason = _calculate_next_auto_sync() 
        _set_auto_sync_state(next_sync_at=int(next_sync_at), next_reason=reason) 
        wait_seconds = max(1, int(next_sync_at - time.time())) 
        if AUTO_SYNC_STOP.wait(wait_seconds): 
            break 
        _run_auto_sync(reason) 


def _start_auto_sync_worker() -> None:  # 启动自动同步线程。 
    global AUTO_SYNC_THREAD 
    if AUTO_SYNC_THREAD and AUTO_SYNC_THREAD.is_alive(): 
        return 
    AUTO_SYNC_STOP.clear() 
    AUTO_SYNC_THREAD = threading.Thread(target=_auto_sync_loop, name="performance-auto-sync", daemon=True) 
    AUTO_SYNC_THREAD.start() 


@app.on_event("startup")  # 注册启动事件用于初始化数据库。 
def _startup():  # 启动初始化函数。 
    try:  # 捕获数据库初始化异常。 
        init_db()  # 初始化数据库表结构。 
        ensure_default_users()  # 确保默认账号存在。 
        script_share_service.init_db()  # 初始化脚本分享数据库。 
        performance_service.init_db()  # 初始化业绩数据库。 
        _start_auto_sync_worker()  # 启动业绩自动同步。 
    except Exception as exc:  # 记录初始化异常避免服务中断。 
        _append_log(f"步骤: 登录初始化异常 {exc}", "_system")  # 输出初始化异常日志。 


@app.on_event("shutdown")  # 注册关闭事件用于停止自动同步线程。 
def _shutdown():  # 服务关闭时停止后台线程。 
    AUTO_SYNC_STOP.set() 


@app.get("/")  # 定义主页路由用于返回前端页面。 
def index(request: Request):  # 主页接口函数。 
    if not _get_session_user(request):  # 若未登录则返回登录页。 
        return FileResponse(LOGIN_PATH)  # 返回 login.html 页面内容。 
    return FileResponse(INDEX_PATH)  # 返回 index.html 页面内容。 
@app.get("/login")  # 定义登录页面路由。 
def login_page():  # 登录页面接口函数。 
    return FileResponse(LOGIN_PATH)  # 返回 login.html 页面内容。 
@app.get("/api/me")  # 定义登录状态查询接口。 
def get_me(request: Request):  # 登录状态查询函数。 
    user = _get_session_user(request)  # 获取当前会话用户。 
    if not user:  # 未登录返回状态。 
        return {"logged_in": False}  # 返回未登录状态。 
    try: 
        force_reset = is_default_password(user)  # 实时读取是否需要修改默认密码，避免 token 与 Cookie 状态不一致。 
    except Exception: 
        force_reset = bool(getattr(request, "session", {}).get("force_reset"))  # 数据库异常时回退旧会话标记。 
    return {"logged_in": True, "account": user, "force_reset": force_reset}  # 返回已登录状态与账号。 
@app.post("/api/login")  # 定义登录接口路由。 
def login(payload: LoginRequest, request: Request):  # 登录接口函数。 
    account = (payload.account or "").strip()  # 获取账号并去除空白。 
    password = payload.password or ""  # 获取密码字段。 
    if not account or not password:  # 校验账号密码不能为空。 
        return JSONResponse({"ok": False, "message": "账号和密码不能为空"}, status_code=400)  # 返回参数错误。 
    try:  # 捕获数据库异常。 
        ok = authenticate_user(account, password)  # 校验账号密码。 
    except Exception as exc:  # 返回数据库错误信息。 
        return JSONResponse({"ok": False, "message": f"数据库异常: {exc}"}, status_code=500)  # 返回服务错误。 
    if not ok:  # 登录失败返回提示。 
        return JSONResponse({"ok": False, "message": "账号或密码错误"}, status_code=401)  # 返回未授权错误。 
    request.session["user"] = account  # 登录成功写入会话。 
    request.session["force_reset"] = is_default_password(account)  # 标记是否需要修改默认密码。 
    return {"ok": True, "account": account, "session_token": _issue_session_token(account)}  # 返回登录成功响应。 
@app.post("/api/logout")  # 定义退出登录接口路由。 
def logout(request: Request):  # 退出登录接口函数。 
    request.session.clear()  # 清空会话信息。 
    return {"ok": True}  # 返回退出成功响应。 
@app.post("/api/reset_password")  # 定义修改密码接口路由。 
def reset_password(payload: ResetPasswordRequest, request: Request):  # 修改密码接口函数。 
    account = (payload.account or "").strip()  # 获取账号并去除空白。 
    if not account:  # 若未传账号则使用当前会话账号。 
        account = _get_session_user(request)  # 从会话读取账号。 
    old_password = payload.old_password or ""  # 获取旧密码字段。 
    new_password = payload.new_password or ""  # 获取新密码字段。 
    if not account:  # 校验账号不能为空。 
        return JSONResponse({"ok": False, "message": "账号不能为空"}, status_code=400)  # 返回参数错误。 
    if not old_password or not new_password:  # 校验密码字段不能为空。 
        return JSONResponse({"ok": False, "message": "旧密码和新密码不能为空"}, status_code=400)  # 返回参数错误。 
    try:  # 捕获数据库异常。 
        ok, message = update_password(account, old_password, new_password)  # 调用更新密码逻辑。 
    except Exception as exc:  # 返回数据库错误信息。 
        return JSONResponse({"ok": False, "message": f"数据库异常: {exc}"}, status_code=500)  # 返回服务错误。 
    if not ok:  # 修改失败返回提示。 
        return JSONResponse({"ok": False, "message": message}, status_code=400)  # 返回修改失败响应。 
    request.session["user"] = account  # 修改成功后更新会话。 
    request.session["force_reset"] = False  # 修改成功后取消强制提示。 
    return {"ok": True, "message": message}  # 返回修改成功响应。 
@app.post("/api/run")  # 定义启动任务的接口路由。 
def run_task(payload: RunRequest, request: Request):  # 运行任务接口函数。 
    auth_error = _require_login(request)  # 校验登录状态。 
    if auth_error:  # 未登录则返回错误响应。 
        return auth_error  # 返回未登录响应。 
    reset_error = _require_password_reset(request)  # 校验是否需修改默认密码。 
    if reset_error:  # 若需修改则返回错误。 
        return reset_error  # 返回禁止响应。 
    user = _get_session_user(request)  # 获取当前账号。 
    regions = payload.regions or []  # 获取地区列表并处理空值。 
    if not regions:  # 若地区列表为空则返回错误。 
        return JSONResponse({"ok": False, "message": "请至少选择一个地区"}, status_code=400)  # 返回错误响应。 
    ak = (payload.ak or "").strip()  # 获取并清理 AK 字符串。 
    sk = (payload.sk or "").strip()  # 获取并清理 SK 字符串。 
    if (ak and not sk) or (sk and not ak):  # 若 AK/SK 只填一个则返回错误。 
        return JSONResponse({"ok": False, "message": "AK 和 SK 需要同时填写"}, status_code=400)  # 返回错误响应。 
    if ak and sk and payload.manual_confirm_customer:  # 人工确认放行时仍先验证华为云 AK/SK 有效。 
        try: 
            identity = iam_service.get_account_identity_for_credentials(ak, sk) 
        except Exception as exc: 
            guard_message = _summarize_guard_exception("华为云", exc) 
            _append_log(f"步骤: 人工确认被拒绝 {guard_message}", user) 
            _set_state(user, running=False, error=guard_message, finished_at=time.time()) 
            return JSONResponse({"ok": False, "message": guard_message}, status_code=403) 
        account_id = identity.get("account_id") or identity.get("account") or "unknown" 
        _append_log(f"步骤: 人工确认客户归属通过 account_id={account_id}", user) 
    elif ak and sk:  # 客户传入 AK/SK 时先确认属于代理后台客户。 
        try: 
            guard_ok, guard_message, _guard_detail = huawei_customer_guard.validate_customer_ownership(ak, sk) 
        except huawei_customer_guard.PartnerCustomerLookupError as exc: 
            guard_message = f"华为云代理后台客户查询失败，可人工确认后继续: {exc}" 
            return _manual_confirm_response(guard_message, user) 
        except Exception as exc: 
            guard_message = _summarize_guard_exception("华为云", exc) 
            _append_log(f"步骤: 客户归属校验失败 {guard_message}", user) 
            _set_state(user, running=False, error=guard_message, finished_at=time.time()) 
            return JSONResponse({"ok": False, "message": guard_message}, status_code=403) 
        if not guard_ok: 
            return _manual_confirm_response(guard_message, user) 
        _append_log(f"步骤: 客户归属校验通过 {guard_message}", user) 
    if not _reserve_huawei_task(user):  # 启动前原子占用当前账号，避免同账号重复点击。 
        return JSONResponse({"ok": False, "message": "本账号华为云任务正在运行，请等待执行完毕", "running_user": user}, status_code=409)  # 返回冲突响应。 
    mode = (payload.mode or "").strip().lower()  # 获取并清理购买模式。 
    allowed_modes = {"delta", "x1.2u.2g", "x1.2u.29", "fixed", "x1", "x1.1u.1g", "20", "wawacloud_20", "twenty", "x1.20"}  # 兼容前端不同版本的 x1.2u.2g/20 模式名。 
    if mode and mode not in allowed_modes:  # 当前只允许标准模式、x1.2u.2g 模式或 20 条 IP 购买模式。 
        _set_state(user, running=False, finished_at=time.time())  # 回滚已占用的账号任务状态。 
        return JSONResponse({"ok": False, "message": "未知购买模式"}, status_code=400)  # 返回错误响应。 
    thread = threading.Thread(target=_run_task, args=(user, regions, ak, sk, mode), daemon=True)  # 创建后台线程执行任务。 
    try: 
        thread.start()  # 启动后台线程。 
    except Exception as exc: 
        _set_state(user, running=False, error=str(exc), finished_at=time.time())  # 回滚启动失败状态。 
        return JSONResponse({"ok": False, "message": f"任务启动失败: {exc}"}, status_code=500)  # 返回启动失败。 
    return {"ok": True, "message": "任务已启动"}  # 返回启动成功响应。 
@app.get("/api/status")  # 定义状态查询接口路由。 
def get_status(request: Request):  # 状态查询接口函数。 
    auth_error = _require_login(request)  # 校验登录状态。 
    if auth_error:  # 未登录则返回错误响应。 
        return auth_error  # 返回未登录响应。 
    user = _get_session_user(request)  # 获取当前账号。 
    return _get_state_snapshot(user)  # 返回当前账号独立状态。 
@app.get("/api/logs")  # 定义日志拉取接口路由。 
def get_logs(request: Request, cursor: int = 0):  # 日志拉取接口函数。 
    auth_error = _require_login(request)  # 校验登录状态。 
    if auth_error:  # 未登录则返回错误响应。 
        return auth_error  # 返回未登录响应。 
    user = _get_session_user(request)  # 获取当前账号。 
    buffer = _get_log_buffer(user)  # 获取当前账号的日志缓冲。 
    with LOG_LOCK:  # 加锁读取日志缓冲区。 
        safe_cursor = max(cursor, 0)  # 对游标进行非负修正。 
        if safe_cursor > len(buffer):  # 若前端携带的是过期游标则回退到开头重拉。 
            safe_cursor = 0  # 回退到首行避免错过本次任务日志。 
        lines = buffer[safe_cursor:]  # 取出指定游标后的日志行。 
        next_cursor = len(buffer)  # 计算下一个游标位置。 
    return {"lines": lines, "next_cursor": next_cursor}  # 返回日志数据与下一个游标。 

@app.post("/api/clear_logs")  # 定义清空日志接口路由。 
def clear_logs(request: Request):  # 清空当前账号的日志缓冲。 
    auth_error = _require_login(request)  # 校验登录状态。 
    if auth_error:  # 未登录则返回错误响应。 
        return auth_error  # 返回未登录响应。 
    user = _get_session_user(request)  # 获取当前账号。 
    buffer = _get_log_buffer(user)  # 获取当前账号的日志缓冲。 
    with LOG_LOCK:  # 加锁清空日志缓冲。 
        buffer.clear()  # 清空当前账号的历史日志。 
    return {"ok": True}  # 返回清空成功响应。 

@app.post("/api/proxy_test")  # 批量测试 SOCKS5 IP 的 TCP/UDP 连通性。 
def proxy_test(payload: ProxyTestRequest, request: Request): 
    auth_error = _require_login(request)  # 校验登录状态。 
    if auth_error: 
        return auth_error 
    reset_error = _require_password_reset(request)  # 默认密码必须先修改。 
    if reset_error: 
        return reset_error 
    try: 
        return proxy_test_service.test_proxy_lines(payload.proxies or "") 
    except ValueError as exc: 
        return JSONResponse({"ok": False, "message": str(exc)}, status_code=400) 
    except Exception as exc: 
        return JSONResponse({"ok": False, "message": f"IP 测试失败: {exc}"}, status_code=500) 

@app.post("/api/ctyun/run")  # 启动天翼云自动购买。 
def run_ctyun_task(payload: RunRequest, request: Request): 
    return _start_ctyun_background_task(request, payload, CTYUN_PURCHASE_PHASES, "purchase") 


@app.post("/api/ctyun/cleanup_query")  # 查询天翼云可退订/重装资源。 
def query_ctyun_cleanup(payload: CtyunCleanupRequest, request: Request): 
    return _start_ctyun_background_task(request, payload, ["cleanup-query"], "cleanup-query", allow_all_regions=True) 


@app.post("/api/ctyun/cleanup_execute")  # 退订天翼云勾选资源。 
def execute_ctyun_cleanup(payload: CtyunCleanupRequest, request: Request): 
    return _start_ctyun_background_task( 
        request, 
        payload, 
        ["cleanup-execute"], 
        "cleanup-execute", 
        allow_all_regions=True, 
        items=payload.items or [], 
    ) 


@app.post("/api/ctyun/reinstall_execute")  # 重装天翼云勾选云服务器。 
def execute_ctyun_reinstall(payload: CtyunCleanupRequest, request: Request): 
    ecs_items = [item for item in (payload.items or []) if str(item.get("type") or "").upper() == "ECS"] 
    return _start_ctyun_background_task( 
        request, 
        payload, 
        ["reinstall-execute"], 
        "reinstall-execute", 
        allow_all_regions=True, 
        items=ecs_items, 
    ) 


@app.get("/api/ctyun/status")  # 查询天翼云任务状态。 
def get_ctyun_status(request: Request): 
    auth_error = _require_login(request) 
    if auth_error: 
        return auth_error 
    user = _get_session_user(request) 
    service_user = _service_state_key("ctyun", user) 
    snapshot = _get_state_snapshot(service_user) 
    snapshot.pop("workspace", None) 
    return snapshot 


@app.get("/api/ctyun/logs")  # 拉取天翼云任务日志。 
def get_ctyun_logs(request: Request, cursor: int = 0): 
    auth_error = _require_login(request) 
    if auth_error: 
        return auth_error 
    user = _get_session_user(request) 
    buffer = _get_log_buffer(_service_state_key("ctyun", user)) 
    with LOG_LOCK: 
        safe_cursor = max(cursor, 0) 
        if safe_cursor > len(buffer): 
            safe_cursor = 0 
        lines = buffer[safe_cursor:] 
        next_cursor = len(buffer) 
    return {"lines": lines, "next_cursor": next_cursor} 


@app.post("/api/ctyun/clear_logs")  # 清空天翼云任务日志。 
def clear_ctyun_logs(request: Request): 
    auth_error = _require_login(request) 
    if auth_error: 
        return auth_error 
    user = _get_session_user(request) 
    buffer = _get_log_buffer(_service_state_key("ctyun", user)) 
    with LOG_LOCK: 
        buffer.clear() 
    return {"ok": True} 


@app.get("/api/ctyun/cleanup_resources")  # 读取最近一次资源查询结果。 
def get_ctyun_cleanup_resources(request: Request): 
    auth_error = _require_login(request) 
    if auth_error: 
        return auth_error 
    user = _get_session_user(request) 
    workspace = _latest_ctyun_workspace(user) 
    if not workspace: 
        return {"items": [], "generatedAt": "", "message": "还没有查询过资源"} 
    path = os.path.join(workspace, "cleanup_resources.json") 
    if not os.path.isfile(path): 
        return {"items": [], "generatedAt": "", "message": "最近一次任务没有资源查询结果"} 
    try: 
        with open(path, "r", encoding="utf-8-sig") as f: 
            data = json.load(f) 
    except Exception as exc: 
        return JSONResponse({"ok": False, "message": f"读取资源查询结果失败: {exc}"}, status_code=500) 
    return data 

@app.get("/api/performance/summary")  # 获取业绩汇总。 
def get_performance_summary(request: Request):  # 业绩汇总接口。 
    auth_error = _require_login(request)  # 校验登录状态。 
    if auth_error: 
        return auth_error 
    return performance_service.summary()  # 返回当日、当月、客户数等汇总。 

@app.get("/api/performance/growth")  # 获取每日业绩增长报表。 
def get_performance_growth(request: Request, days: int = 14):  # 业绩增长接口。 
    auth_error = _require_login(request)  # 校验登录状态。 
    if auth_error: 
        return auth_error 
    return performance_service.growth_report(days)  # 返回三家云每日增长和图表数据。 

@app.get("/api/performance/auto_sync")  # 获取业绩自动同步状态。 
def get_performance_auto_sync(request: Request):  # 自动同步状态接口。 
    auth_error = _require_login(request)  # 校验登录状态。 
    if auth_error: 
        return auth_error 
    state = _get_auto_sync_state() 
    return { 
        "ok": True, 
        **state, 
        "interval_seconds": AUTO_SYNC_INTERVAL_SECONDS, 
        "daily_hour": AUTO_SYNC_DAILY_HOUR, 
        "daily_minute": AUTO_SYNC_DAILY_MINUTE, 
    } 

@app.get("/api/performance/agent_status")  # 查询三家代理后台账号数据是否有效。 
def get_agent_status(request: Request, provider: str = ""):  # 代理后台状态查询接口。 
    auth_error = _require_login(request)  # 校验登录状态。 
    if auth_error: 
        return auth_error 
    reset_error = _require_password_reset(request)  # 默认密码未改时不允许读取后台登录态。 
    if reset_error: 
        return reset_error 
    providers = [ 
        ("ctyun", "天翼云"), 
        ("huawei", "华为云"), 
        ("volc", "火山引擎"), 
    ] 
    provider_key = (provider or "").strip().lower() 
    if provider_key: 
        providers = [item for item in providers if item[0] == provider_key or item[1] == provider] 
        if not providers: 
            return JSONResponse({"ok": False, "message": "不支持的云厂商"}, status_code=400) 
    statuses = [_agent_provider_status(key, label) for key, label in providers] 
    return {"ok": all(item["ok"] for item in statuses), "statuses": statuses, "checked_at": int(time.time())} 

@app.post("/api/performance/sync_all")  # 同步三家云代理后台业绩。 
def sync_all_performance(request: Request):  # 三家云同步接口。 
    auth_error = _require_login(request)  # 校验登录状态。 
    if auth_error: 
        return auth_error 
    reset_error = _require_password_reset(request)  # 默认密码未改时不允许同步敏感后台数据。 
    if reset_error: 
        return reset_error 
    result = _sync_provider_metrics(wait=False) 
    status_code = 409 if result.get("running") else 200 
    return JSONResponse(result, status_code=status_code) 

@app.post("/api/performance/import")  # 上传本地导出的云厂商后台 JSON。 
def import_performance(payload: PerformanceImportRequest, request: Request):  # 业绩导入接口。 
    auth_error = _require_login(request)  # 校验登录状态。 
    if auth_error: 
        return auth_error 
    reset_error = _require_password_reset(request)  # 默认密码未改时不允许导入敏感数据。 
    if reset_error: 
        return reset_error 
    if payload.payload is None:  # JSON 内容不能为空。 
        return JSONResponse({"ok": False, "message": "请上传或粘贴 JSON 数据"}, status_code=400) 
    try: 
        return performance_service.import_snapshot( 
            provider=payload.provider, 
            payload=payload.payload, 
            created_by=_get_session_user(request), 
        ) 
    except Exception as exc: 
        return JSONResponse({"ok": False, "message": str(exc)}, status_code=400) 

@app.get("/api/performance/customers")  # 按手机号、账号 ID 或账号查询客户。 
def search_performance_customers(request: Request, keyword: str = "", provider: str = ""):  # 客户查询接口。 
    auth_error = _require_login(request)  # 校验登录状态。 
    if auth_error: 
        return auth_error 
    try: 
        provider_key = (provider or "").strip().lower()
        if provider_key in {"ctyun", "天翼云"}: 
            return ctyun_agent_service.search_customers(keyword=keyword) 
        if provider_key in {"huawei", "华为云"}: 
            return huawei_agent_service.search_customers(keyword=keyword) 
        if provider_key in {"volc", "volcengine", "火山引擎"}: 
            return volc_agent_service.search_customers(keyword=keyword) 
        return performance_service.search_customers(keyword=keyword, provider=provider) 
    except Exception as exc: 
        return JSONResponse({"ok": False, "message": str(exc)}, status_code=400) 

@app.post("/api/performance/ctyun/sync")  # 使用已保存 Cookie 同步天翼云代理后台业绩。 
def sync_ctyun_performance(request: Request):  # 天翼云代理后台同步接口。 
    auth_error = _require_login(request)  # 校验登录状态。 
    if auth_error: 
        return auth_error 
    reset_error = _require_password_reset(request)  # 默认密码未改时不允许同步敏感后台数据。 
    if reset_error: 
        return reset_error 
    try: 
        result = ctyun_agent_service.sync_metrics() 
        metrics = result["metrics"] 
        performance_service.upsert_provider_metrics("ctyun", "天翼云", metrics) 
        return {"ok": True, "metrics": metrics} 
    except Exception as exc: 
        return JSONResponse({"ok": False, "message": str(exc)}, status_code=400) 

@app.post("/api/performance/huawei/sync")  # 使用已保存 Cookie 同步华为云代理后台业绩。 
def sync_huawei_performance(request: Request):  # 华为云代理后台同步接口。 
    auth_error = _require_login(request)  # 校验登录状态。 
    if auth_error: 
        return auth_error 
    reset_error = _require_password_reset(request)  # 默认密码未改时不允许同步敏感后台数据。 
    if reset_error: 
        return reset_error 
    try: 
        result = huawei_agent_service.sync_metrics() 
        metrics = result["metrics"] 
        performance_service.upsert_provider_metrics("huawei", "华为云", metrics) 
        return {"ok": True, "metrics": metrics} 
    except Exception as exc: 
        return JSONResponse({"ok": False, "message": str(exc)}, status_code=400) 

@app.post("/api/performance/volc/sync")  # 使用已保存 Cookie 同步火山引擎代理后台业绩。 
def sync_volc_performance(request: Request):  # 火山引擎代理后台同步接口。 
    auth_error = _require_login(request)  # 校验登录状态。 
    if auth_error: 
        return auth_error 
    reset_error = _require_password_reset(request)  # 默认密码未改时不允许同步敏感后台数据。 
    if reset_error: 
        return reset_error 
    try: 
        result = volc_agent_service.sync_metrics() 
        metrics = result["metrics"] 
        performance_service.upsert_provider_metrics("volc", "火山引擎", metrics) 
        return {"ok": True, "metrics": metrics} 
    except Exception as exc: 
        return JSONResponse({"ok": False, "message": str(exc)}, status_code=400) 

@app.post("/api/script_shares")  # 创建脚本分享链接。 
def create_script_share(payload: ScriptShareRequest, request: Request):  # 脚本分享创建接口。 
    auth_error = _require_login(request)  # 校验登录状态。 
    if auth_error: 
        return auth_error 
    reset_error = _require_password_reset(request)  # 默认密码未改时不允许创建。 
    if reset_error: 
        return reset_error 
    try: 
        record = script_share_service.create_share( 
            script_key=payload.script_key, 
            validity=payload.validity, 
            max_uses=payload.max_uses, 
            created_by=_get_session_user(request), 
            base_url=_public_base_url(request), 
        ) 
    except Exception as exc: 
        return JSONResponse({"ok": False, "message": str(exc)}, status_code=400) 
    return {"ok": True, **record} 

@app.get("/api/script_share/{token}")  # 客户查看分享页，不展示源码。 
def get_script_share_page(token: str, request: Request): 
    return HTMLResponse(script_share_service.render_customer_page(token, _public_base_url(request))) 

@app.get("/api/script_share/run/{token}")  # 客户执行命令拉取的入口脚本。 
def get_script_share_runner(token: str, request: Request): 
    body, status_code = script_share_service.build_runner(token, _public_base_url(request)) 
    return PlainTextResponse( 
        body, 
        status_code=status_code, 
        media_type="text/x-shellscript; charset=utf-8", 
        headers={"Cache-Control": "no-store"}, 
    ) 

@app.get("/api/script_share/payload/{token}")  # 入口脚本校验签名后消耗次数并返回真实脚本。 
def get_script_share_payload(token: str, sig: str = ""): 
    body, status_code = script_share_service.consume_payload(token, sig) 
    return PlainTextResponse( 
        body, 
        status_code=status_code, 
        media_type="text/x-shellscript; charset=utf-8", 
        headers={"Cache-Control": "no-store"}, 
    ) 
