update-server/app/api/routes/admin_config.py

import json
import os
from datetime import datetime, timezone

from fastapi import APIRouter, Depends, HTTPException, Request

from app.core.config import ENV_FILE_PATH
from app.core.security import persist_env_value, require_permission, set_admin_token, token_digest
from app.repositories import app_channel_repository, license_repository
from app.schemas.admin_config import ChangeAdminTokenRequest, ClientConfigGenerateRequest
from app.services.admin_config_service import (
    PUBLISH_MAX_FIELDS,
    PUBLISH_MAX_FILES,
    PUBLISH_MAX_REQUEST_BYTES,
    RELEASE_MAIN_EXECUTABLE,
    TARGET_ARCH,
    TARGET_PLATFORM,
    default_client_config_values,
    crash_report_test_values,
    crash_report_test_text,
    normalize_executable_name,
    platform_executable_path,
    public_api_base_url,
)
from app.services.common_service import validate_channel_code


router = APIRouter(tags=["admin-config"])


def client_config_license_warning(license_key: str, app_id: str, channel: str) -> str:
    if not license_key:
        return ""
    license_row = license_repository.get_license_by_hash_with_usage(token_digest(license_key))
    if not license_row:
        return "当前 License 不存在或已删除，生成的客户端配置未写入 License，请创建或申请新的 License。"
    if license_row["app_id"] != app_id or license_row["channel_code"] != channel:
        return "当前 License 不属于所选应用或渠道，生成的客户端配置未写入 License，请选择匹配的 License。"
    if license_row["status"] != "active":
        return "当前 License 已禁用，生成的客户端配置未写入 License，请启用该 License 或创建新的 License。"
    try:
        expiry = datetime.fromisoformat(str(license_row["valid_until"]).replace("Z", "+00:00"))
    except ValueError:
        expiry = datetime.now(timezone.utc)
    if expiry.tzinfo is None:
        expiry = expiry.replace(tzinfo=timezone.utc)
    if expiry <= datetime.now(timezone.utc):
        return "当前 License 已过期，生成的客户端配置未写入 License，请创建或申请新的 License。"
    used_devices = int(license_row["used_devices"] or 0)
    max_devices = int(license_row["max_devices"] or 0)
    if used_devices >= max_devices:
        return f"当前 License 已绑定 {used_devices}/{max_devices} 台设备，已达到上限。生成的客户端配置未写入 License，请创建或申请新的 License。"
    return ""


def normalize_install_root(value: str, fallback: str = "..") -> str:
    normalized = str(value or fallback).strip().replace("\\", "/")
    if normalized in ("", "."):
        return "."
    if normalized == "..":
        return ".."
    raise HTTPException(status_code=400, detail="install_root 目前仅支持 . 或 ..")


@router.post("/admin/token/change")
def admin_change_token(body: ChangeAdminTokenRequest, auth=Depends(require_permission("admin:manage"))):
    new_token = body.new_token.strip()
    if len(new_token) < 8:
        raise HTTPException(status_code=400, detail="新服务端兜底令牌至少需要 8 个字符")
    if len(new_token) > 128:
        raise HTTPException(status_code=400, detail="新服务端兜底令牌不能超过 128 个字符")
    try:
        persist_env_value(ENV_FILE_PATH, "ADMIN_TOKEN", new_token)
    except Exception as exc:
        raise HTTPException(status_code=500, detail=f"令牌已通过校验，但写入 .env 失败: {exc}")
    set_admin_token(new_token)
    os.environ["ADMIN_TOKEN"] = new_token
    return {"code": 0, "msg": f"服务端兜底令牌已更新，并已写入 {ENV_FILE_PATH}"}


@router.get("/admin/runtime-config")
def admin_runtime_config(request: Request, auth=Depends(require_permission("config:view"))):
    return {
        "release_main_executable": RELEASE_MAIN_EXECUTABLE,
        "target_platform": TARGET_PLATFORM,
        "target_arch": TARGET_ARCH,
        "publish_max_request_bytes": PUBLISH_MAX_REQUEST_BYTES,
        "publish_max_files": PUBLISH_MAX_FILES,
        "publish_max_fields": PUBLISH_MAX_FIELDS,
        "api_base_url": public_api_base_url(request),
        "default_client_config": default_client_config_values(request),
    }


@router.post("/admin/client-config/generate")
def admin_generate_client_config(body: ClientConfigGenerateRequest, request: Request, auth=Depends(require_permission("config:view"))):
    app_id = body.app_id.strip()
    channel = body.channel.strip() or "stable"
    current_version = body.current_version.strip() or "1.0.0"
    client_protocol = max(1, int(body.client_protocol or 3))
    api_base_url = (body.api_base_url.strip() or public_api_base_url(request)).rstrip("/")
    if not app_id:
        raise HTTPException(status_code=400, detail="请先选择或填写 App ID")
    if not validate_channel_code(channel):
        raise HTTPException(status_code=400, detail="渠道代码无效")
    if not api_base_url.startswith(("http://", "https://")):
        raise HTTPException(status_code=400, detail="api_base_url 必须以 http:// 或 https:// 开头")

    app_row = app_channel_repository.get_app(app_id)
    if not app_row:
        raise HTTPException(status_code=404, detail="应用不存在，请先在应用管理中创建")
    channel_row = app_channel_repository.get_channel(app_id, channel)
    if not channel_row:
        raise HTTPException(status_code=400, detail={"error": "channel_not_found", "msg": f"渠道 {channel} 不存在"})

    defaults = default_client_config_values(request)
    install_root = normalize_install_root(body.install_root, defaults["install_root"])
    main_executable = platform_executable_path(
        normalize_executable_name(body.main_executable),
        defaults["main_executable"],
    )
    license_key = body.license_key.strip()
    license_warning = client_config_license_warning(license_key, app_id, channel)
    if license_warning:
        license_key = ""
    config = {
        "app_id": app_id,
        "app_name": app_row["app_name"],
        "channel": channel,
        "current_version": current_version,
        "client_protocol": str(client_protocol),
        "launch_token": defaults["launch_token"],
        "license_key": license_key,
        "client_token": defaults["client_token"],
        "request_timeout_ms": defaults["request_timeout_ms"],
        "temp_folder": defaults["temp_folder"],
        "device_id": "",
        "install_root": install_root,
        "main_executable": main_executable,
        "launcher_executable": defaults["launcher_executable"],
        "updater_executable": defaults["updater_executable"],
        "bootstrap_executable": defaults["bootstrap_executable"],
        "health_check_timeout_ms": defaults["health_check_timeout_ms"],
        "platform": defaults["platform"],
        "arch": defaults["arch"],
    }
    crash_test_config = crash_report_test_values(api_base_url)
    server_config = {
        "api_base_url": api_base_url,
    }
    return {
        "config": config,
        "json_text": json.dumps(config, ensure_ascii=False, indent=2),
        "license_warning": license_warning,
        "server_config": server_config,
        "server_config_text": json.dumps(server_config, ensure_ascii=False, indent=2),
        "crash_test_config": crash_test_config,
        "crash_test_text": crash_report_test_text(api_base_url),
    }