update-system/server/main.py

from fastapi import FastAPI, Body, Request, HTTPException, UploadFile, File, Form, Header, Depends
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
from pydantic import BaseModel
import os
import sqlite3
from pathlib import Path, PurePosixPath
from dotenv import load_dotenv
from contextlib import asynccontextmanager
import db
import minio_tool
import hashlib
import secrets
import base64
import json
import uuid
from datetime import datetime, timezone, timedelta
from fastapi.middleware.cors import CORSMiddleware
import socket
import subprocess
import time
import shutil
import tempfile
import urllib.request
from cryptography.hazmat.primitives.serialization import load_pem_private_key
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives import hashes

# 加载环境变量
load_dotenv()
BASE_DIR = Path(__file__).resolve().parent


def env_bool(name: str, default: bool = False) -> bool:
    value = os.getenv(name)
    return default if value is None else value.strip().lower() in {"1", "true", "yes", "on"}


def configured_path(name: str, default: str) -> Path:
    path = Path(os.getenv(name, default)).expanduser()
    return path if path.is_absolute() else BASE_DIR / path


SERVER_HOST = os.getenv("SERVER_HOST", "0.0.0.0")
SERVER_PORT = int(os.getenv("SERVER_PORT", "8000"))
VALID_CLIENT_TOKEN = os.getenv("CLIENT_API_TOKEN")
TARGET_PLATFORM = os.getenv("TARGET_PLATFORM", "windows")
TARGET_ARCH = os.getenv("TARGET_ARCH", "x64")
RELEASE_MAIN_EXECUTABLE = os.getenv("RELEASE_MAIN_EXECUTABLE", "MainApp.exe").strip()
SIGNING_KEY_ID = os.getenv("SIGNING_KEY_ID", "manifest-key-v1")

# 大型 multipart 上传使用内存文件系统暂存，避免与 MinIO 对象双重占用根分区。
UPLOAD_SPOOL_DIR = configured_path("UPLOAD_SPOOL_DIR", "upload_spool")
UPLOAD_SPOOL_DIR.mkdir(parents=True, exist_ok=True)
tempfile.tempdir = str(UPLOAD_SPOOL_DIR)
MINIO_DATA_DIR = configured_path("MINIO_DATA_DIR", "minio_data")
UPLOAD_SPACE_RESERVE = int(os.getenv("UPLOAD_SPACE_RESERVE_MB", "256")) * 1024 * 1024

# SimCAE 崩溃报告私有存储与上传限制
CRASH_STORAGE_ROOT = configured_path("CRASH_STORAGE_ROOT", "crash_storage")
CRASH_METADATA_MAX_BYTES = int(os.getenv("CRASH_METADATA_MAX_KB", "256")) * 1024
CRASH_MINIDUMP_MAX_BYTES = int(os.getenv("CRASH_MINIDUMP_MAX_MB", "128")) * 1024 * 1024
CRASH_ATTACHMENTS_MAX_BYTES = int(os.getenv("CRASH_ATTACHMENTS_MAX_MB", "64")) * 1024 * 1024
CRASH_REQUEST_MAX_BYTES = int(os.getenv("CRASH_REQUEST_MAX_MB", "200")) * 1024 * 1024
CRASH_SYMBOLS_MAX_BYTES = int(os.getenv("CRASH_SYMBOLS_MAX_MB", "512")) * 1024 * 1024
CRASH_SERVICE_VERSION = os.getenv("CRASH_SERVICE_VERSION", "1.0.0")

# 后台管理令牌：当前版本直接使用 .env 中的 ADMIN_TOKEN 明文配置。
ADMIN_TOKEN = os.getenv("ADMIN_TOKEN", "").strip()
if not ADMIN_TOKEN:
    raise RuntimeError("请在环境变量或 .env 中配置 ADMIN_TOKEN")

def token_digest(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()

CRASH_REPORT_TOKEN = os.getenv("CRASH_REPORT_TOKEN") or os.getenv("SIMCAE_CRASH_CLIENT_TOKEN") or VALID_CLIENT_TOKEN
CRASH_SYMBOL_TOKEN = os.getenv("CRASH_SYMBOL_TOKEN") or os.getenv("SIMCAE_SYMBOL_TOKEN") or ADMIN_TOKEN
CRASH_ADMIN_TOKEN = os.getenv("CRASH_ADMIN_TOKEN")

# ========== 生命周期初始化 ==========
@asynccontextmanager
async def lifespan(app: FastAPI):
    print("===== 进程启动，开始初始化数据库 =====")
    db_file = db.DB_FILE
    sql_file_path = db.SQL_FILE
    conn = sqlite3.connect(db_file, check_same_thread=False)
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()

    if sql_file_path.exists():
        with open(sql_file_path, "r", encoding="utf-8") as f:
            full_sql = f.read()
        cur.executescript(full_sql)
        version_columns = {row[1] for row in cur.execute("PRAGMA table_info(versions)").fetchall()}
        if "client_protocol" not in version_columns:
            cur.execute("ALTER TABLE versions ADD COLUMN client_protocol INTEGER NOT NULL DEFAULT 1")
        device_columns = {row[1] for row in cur.execute("PRAGMA table_info(devices)").fetchall()}
        if "license_id" not in device_columns: cur.execute("ALTER TABLE devices ADD COLUMN license_id TEXT NOT NULL DEFAULT ''")
        if "channel" not in device_columns: cur.execute("ALTER TABLE devices ADD COLUMN channel TEXT NOT NULL DEFAULT 'stable'")
        for app_row in cur.execute("SELECT app_id FROM apps").fetchall():
            if not cur.execute("SELECT 1 FROM channels WHERE app_id=? LIMIT 1", (app_row[0],)).fetchone():
                cur.executemany("INSERT INTO channels(app_id,channel_code,display_name,enabled,sort_order) VALUES(?,?,?,?,?)",
                    [(app_row[0], "stable", "正式版", 1, 10), (app_row[0], "preview", "预览版", 1, 20), (app_row[0], "dev", "开发版", 1, 30)])
        conn.commit()
        print(f"✅ 成功执行tables.sql，创建数据表与测试数据")
    else:
        print(f"❌ 错误：当前目录找不到 tables.sql 文件！")
    conn.close()
    print("===== 数据库初始化完成 =====")

    CRASH_STORAGE_ROOT.mkdir(parents=True, exist_ok=True)
    (CRASH_STORAGE_ROOT / "reports").mkdir(parents=True, exist_ok=True)
    (CRASH_STORAGE_ROOT / "symbols").mkdir(parents=True, exist_ok=True)

    start_minio_if_needed()

    yield
    print("服务进程正常退出")

def is_minio_running(endpoint: str) -> bool:
    try:
        if endpoint.startswith("http://") or endpoint.startswith("https://"):
            base_url = endpoint.rstrip("/")
        else:
            scheme = "https" if env_bool("MINIO_SECURE") else "http"
            base_url = f"{scheme}://{endpoint.rstrip('/')}"
        health_url = base_url + "/minio/health/live"
        with urllib.request.urlopen(health_url, timeout=float(os.getenv("MINIO_HEALTH_TIMEOUT_SEC", "2"))) as resp:
            return resp.status == 200
    except Exception:
        return False


def download_minio_binary(target_path: Path) -> str | None:
    url = os.getenv("MINIO_DOWNLOAD_URL", "https://dl.min.io/server/minio/release/linux-amd64/minio")
    try:
        print(f"MinIO 二进制未找到，尝试下载到 {target_path}")
        target_path.parent.mkdir(parents=True, exist_ok=True)
        with urllib.request.urlopen(url, timeout=30) as response, open(target_path, 'wb') as out_file:
            out_file.write(response.read())
        target_path.chmod(0o755)
        abs_path = str(target_path.resolve())
        print(f"已下载 MinIO 二进制到 {abs_path}")
        return abs_path
    except Exception as err:
        print(f"自动下载 MinIO 失败：{err}")
        return None


def start_minio_if_needed():
    if not env_bool("MINIO_AUTO_START", True):
        return
    endpoint = os.getenv('MINIO_ENDPOINT')
    if not endpoint:
        return

    if is_minio_running(endpoint):
        print(f"MinIO 已在 {endpoint} 运行，跳过启动")
        return

    minio_cmd = shutil.which('minio')
    if not minio_cmd:
        minio_bin = Path(os.getenv('MINIO_BIN_PATH', 'minio'))
        minio_cmd = str(minio_bin.resolve()) if minio_bin.exists() else None
        if not minio_cmd and env_bool("MINIO_AUTO_DOWNLOAD", False):
            minio_cmd = download_minio_binary(minio_bin)
        if not minio_cmd:
            print("MinIO 二进制未找到，且未启用或未完成自动下载")
            return

    data_dir = MINIO_DATA_DIR
    data_dir.mkdir(parents=True, exist_ok=True)
    minio_address = os.getenv("MINIO_LISTEN_ADDRESS", ":9000")
    print(f"尝试自动启动 MinIO：{minio_cmd} server {data_dir} --address {minio_address}")
    try:
        subprocess.Popen([minio_cmd, 'server', str(data_dir), '--address', minio_address], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        time.sleep(float(os.getenv("MINIO_START_WAIT_SEC", "5")))
        if is_minio_running(endpoint):
            print("MinIO 已启动成功")
        else:
            print("MinIO 启动失败，仍然无法连接")
    except Exception as err:
        print(f"启动 MinIO 失败：{err}")


# Manifest signing key path
MANIFEST_PRIVATE_KEY_PATH = str(configured_path("MANIFEST_PRIVATE_KEY_PATH", "keys/manifest_private_key.pem"))

# 创建APP绑定生命周期
app = FastAPI(title=os.getenv("SERVICE_TITLE", "软件自动升级服务"), lifespan=lifespan)

ADMIN_HTML_PATH = configured_path("ADMIN_HTML_PATH", "../client/admin.html")


@app.get("/", include_in_schema=False)
@app.get("/admin.html", include_in_schema=False)
def admin_page():
    if not ADMIN_HTML_PATH.is_file():
        raise HTTPException(status_code=404, detail="管理页面文件不存在")
    return FileResponse(ADMIN_HTML_PATH, headers={"Cache-Control": "no-store"})


@app.get("/health", include_in_schema=False)
def health_check():
    return {"status": "ok"}

# 本地上传备份文件目录（MinIO 不通时会保存到这里）
app.mount("/static", StaticFiles(directory=str(minio_tool.LOCAL_UPLOAD_ROOT)), name="static")

# 跨域配置（给admin前端页面用）
cors_origins = [item.strip() for item in os.getenv("CORS_ALLOW_ORIGINS", "*").split(",") if item.strip()]
app.add_middleware(
    CORSMiddleware,
    allow_origins=cors_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 管理后台写操作统一审计；不保存令牌明文和请求正文。
@app.middleware("http")
async def admin_audit_middleware(request: Request, call_next):
    should_audit = request.url.path.startswith("/admin/") and request.method not in ("GET", "HEAD", "OPTIONS")
    status_code = 500
    try:
        response = await call_next(request); status_code = response.status_code; return response
    finally:
        if should_audit:
            try:
                token = request.headers.get("X-Admin-Token", "")
                conn = db.get_conn(); conn.execute("""INSERT INTO admin_audit_logs
                    (actor_hash,action,method,path,target,result,status_code,ip,user_agent) VALUES(?,?,?,?,?,?,?,?,?)""",
                    (token_digest(token)[:16] if token else "anonymous", request.url.path.removeprefix("/admin/"),
                     request.method, request.url.path, request.url.query[:500],
                     "success" if 200 <= status_code < 400 else "fail", status_code,
                     request.client.host if request.client else "", request.headers.get("user-agent", "")[:300])); conn.commit(); conn.close()
            except Exception as audit_error:
                print("管理员审计日志写入失败:", audit_error)

# ===================== 全局客户端鉴权中间件 =====================
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
    path = request.url.path
    # 后台接口、文档接口、静态回退链接全部跳过客户端token校验
    skip_paths = ("/admin", "/docs", "/openapi.json", "/redoc", "/static", "/health", "/api/v1/health", "/api/v1/crash-reports", "/api/v1/symbols")
    if path == "/" or path.startswith(skip_paths):
        response = await call_next(request)
        return response

    client_token = request.headers.get("X-Client-Token")
    if client_token != VALID_CLIENT_TOKEN:
        return JSONResponse(status_code=401, content={"detail": "非法客户端，令牌校验失败"})
    if path != "/api/v1/device/issue":
        credential = request.headers.get("X-Device-Credential", "")
        if not credential:
            return JSONResponse(status_code=401, content={"detail": {
                "error": "device_credential_required", "msg": "缺少设备身份凭证"
            }})
        try:
            request.state.device_identity = verify_device_credential(credential)
        except HTTPException as exc:
            return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})
    response = await call_next(request)
    return response

# ========== 请求体模型 ==========
class CheckUpdateReq(BaseModel):
    app_id: str
    current_version: str
    channel: str
    client_protocol: int = 1

class DownloadUrlReq(BaseModel):
    app_id: str
    channel: str
    version: str
    version_id: int
    files: list[str]

class ManifestReq(BaseModel):
    app_id: str
    channel: str
    version: str
    version_id: int

class ReportReq(BaseModel):
    app_id: str
    device_id: str
    from_version: str
    to_version: str
    result: str

class ChangeAdminTokenReq(BaseModel):
    new_token: str

class DownloadReportReq(BaseModel):
    app_id: str
    channel: str
    version: str
    result: str
    files: list[dict]

class DeviceIssueReq(BaseModel):
    app_id: str
    channel: str
    license_key: str
    installation_id: str
    machine_hash: str = ""

# 统一后台鉴权：全部接口从Header读取token，不再区分表单/头
def admin_auth(X_Admin_Token: str = Header("")):
    if not secrets.compare_digest(X_Admin_Token, ADMIN_TOKEN):
        raise HTTPException(status_code=403, detail="后台密钥错误，禁止访问")
    return True

@app.get("/admin/auth/check")
def admin_auth_check(auth=Depends(admin_auth)):
    return {"code": 0, "msg": "管理员令牌有效"}

# Manifest signing helpers

def load_manifest_private_key():
    key_path = Path(MANIFEST_PRIVATE_KEY_PATH)
    if not key_path.exists():
        raise FileNotFoundError(f"Manifest private key not found: {key_path}")
    key_data = key_path.read_bytes()
    return load_pem_private_key(key_data, password=None)


def canonical_manifest_bytes(manifest_obj: dict) -> bytes:
    copy = {k: manifest_obj[k] for k in manifest_obj if k != "signature"}
    return json.dumps(copy, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")


def sign_manifest(manifest_obj: dict) -> str:
    json_text = canonical_manifest_bytes(manifest_obj)
    private_key = load_manifest_private_key()
    signature = private_key.sign(
        json_text,
        padding.PKCS1v15(),
        hashes.SHA256()
    )
    return base64.b64encode(signature).decode("ascii")


def canonical_signed_bytes(obj: dict) -> bytes:
    copy = {k: obj[k] for k in obj if k != "signature"}
    return json.dumps(copy, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")


def sign_policy(policy_obj: dict) -> tuple[str, str]:
    payload = canonical_signed_bytes(policy_obj)
    signature = load_manifest_private_key().sign(payload, padding.PKCS1v15(), hashes.SHA256())
    return payload.decode("utf-8"), base64.b64encode(signature).decode("ascii")


def sign_device_identity(identity: dict) -> tuple[str, str]:
    payload = canonical_signed_bytes(identity)
    signature = load_manifest_private_key().sign(payload, padding.PKCS1v15(), hashes.SHA256())
    return payload.decode("utf-8"), base64.b64encode(signature).decode("ascii")

def verify_device_credential(encoded: str) -> dict:
    try:
        wrapper = json.loads(base64.b64decode(encoded, validate=True).decode("utf-8"))
        identity_text = wrapper["identity_text"]
        signature = base64.b64decode(wrapper["signature"], validate=True)
        load_manifest_private_key().public_key().verify(signature, identity_text.encode(), padding.PKCS1v15(), hashes.SHA256())
        identity = json.loads(identity_text)
        if any(not identity.get(k) for k in ("device_id", "license_id", "app_id", "channel", "installation_id", "credential_seq", "valid_until")):
            raise ValueError("missing field")
    except Exception:
        raise HTTPException(status_code=401, detail={"error":"invalid_device_credential","msg":"设备凭证格式或签名无效"})
    conn = db.get_conn(); row = conn.execute("SELECT * FROM devices WHERE device_id=?", (identity["device_id"],)).fetchone()
    if not row:
        conn.close(); raise HTTPException(status_code=401, detail={"error":"device_not_registered","msg":"设备未登记"})
    if row["disabled"]:
        reason=row["disabled_reason"] or "设备已被管理员禁用"; conn.close()
        raise HTTPException(status_code=403, detail={"error":"device_disabled","msg":reason})
    if row["app_id"] != identity["app_id"] or row["installation_id"] != identity["installation_id"] or int(row["credential_seq"]) != int(identity["credential_seq"]):
        conn.close(); raise HTTPException(status_code=401, detail={"error":"device_credential_revoked","msg":"设备凭证已失效"})
    license_row = conn.execute("SELECT * FROM licenses WHERE license_id=?", (identity.get("license_id", ""),)).fetchone()
    now = datetime.now(timezone.utc)
    try: license_expiry = datetime.fromisoformat((license_row["valid_until"] if license_row else "").replace("Z", "+00:00"))
    except ValueError: license_expiry = now - timedelta(seconds=1)
    if not license_row or license_row["status"] != "active" or license_expiry <= now:
        conn.close(); raise HTTPException(status_code=403, detail={"error":"license_invalid","msg":"授权不存在、已禁用或已过期"})
    if license_row["app_id"] != identity["app_id"] or license_row["channel_code"] != identity.get("channel"):
        conn.close(); raise HTTPException(status_code=403, detail={"error":"license_scope_mismatch","msg":"授权应用或渠道不匹配"})
    conn.execute("UPDATE devices SET last_seen_at=datetime('now') WHERE device_id=?", (identity["device_id"],)); conn.commit(); conn.close()
    return identity

def policy_row_to_dict(row) -> dict:
    if not row:
        return {
            "policy_seq": 1, "force_update": False, "allow_rollback": False,
            "offline_allowed": True, "valid_until": "2099-12-31T23:59:59Z",
            "min_supported_version": "", "disabled_versions": [], "message": ""
        }
    try:
        disabled = json.loads(row["disabled_versions"] or "[]")
    except (TypeError, json.JSONDecodeError):
        disabled = []
    return {
        "policy_seq": int(row["policy_seq"]), "force_update": bool(row["force_update"]),
        "allow_rollback": bool(row["allow_rollback"]), "offline_allowed": bool(row["offline_allowed"]),
        "valid_until": row["valid_until"], "min_supported_version": row["min_supported_version"] or "",
        "disabled_versions": disabled if isinstance(disabled, list) else [], "message": row["message"] or ""
    }


def validate_channel_code(code: str) -> bool:
    return bool(code) and len(code) <= 32 and code[0].isalnum() and all(c.isalnum() or c in "_-" for c in code)

def require_channel(conn, app_id: str, channel: str, require_enabled: bool = True):
    row = conn.execute("SELECT * FROM channels WHERE app_id=? AND channel_code=?", (app_id, channel)).fetchone()
    if not row: raise HTTPException(status_code=400, detail={"error":"channel_not_found","msg":f"渠道 {channel} 不存在"})
    if require_enabled and not row["enabled"]: raise HTTPException(status_code=403, detail={"error":"channel_disabled","msg":f"渠道 {channel} 已停用"})
    return row

def is_executable_path(path: str) -> bool:
    return path.lower().endswith((".exe", ".dll"))

def version_key(version: str):
    parts = []
    for part in version.split("."):
        digits = "".join(ch for ch in part if ch.isdigit())
        parts.append(int(digits) if digits else 0)
    return tuple(parts)

def normalize_relative_path(raw_path: str) -> str:
    if not isinstance(raw_path, str):
        raise HTTPException(status_code=400, detail="文件相对路径无效")
    normalized = raw_path.replace(chr(92), "/")
    if normalized.startswith("/"):
        raise HTTPException(status_code=400, detail=f"文件路径必须是相对路径: {raw_path}")
    normalized = normalized.strip("/")
    if not normalized or chr(0) in normalized:
        raise HTTPException(status_code=400, detail="文件相对路径为空或包含非法字符")
    path = PurePosixPath(normalized)
    if path.is_absolute() or any(part in ("", ".", "..") for part in path.parts):
        raise HTTPException(status_code=400, detail=f"不安全的文件路径: {raw_path}")
    if path.parts and ":" in path.parts[0]:
        raise HTTPException(status_code=400, detail=f"文件路径不能包含盘符: {raw_path}")
    return path.as_posix()


def validate_release_path(relative_path: str):
    folded = relative_path.casefold()
    parts = PurePosixPath(folded).parts
    blocked_directories = {".git", ".vs", "cmakefiles", "debug", "update", "update_temp"}
    blocked_suffixes = ("_autogen",)
    for part in parts[:-1]:
        if (part in blocked_directories or part.startswith("build")
                or part.endswith(blocked_suffixes)):
            raise HTTPException(
                status_code=400,
                detail=f"发布目录包含构建或运行时目录，必须选择干净的 Release 输出目录: {relative_path}"
            )

# 文件sha256工具
def calc_sha256(data: bytes):
    sha = hashlib.sha256()
    sha.update(data)
    return sha.hexdigest()


class CrashApiException(Exception):
    def __init__(self, status_code: int, code: str, message: str, retryable: bool = False):
        self.status_code = status_code
        self.code = code
        self.message = message
        self.retryable = retryable


@app.exception_handler(CrashApiException)
async def crash_api_exception_handler(request: Request, exc: CrashApiException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": {"code": exc.code, "message": exc.message, "retryable": exc.retryable}},
    )


def utc_now_text() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def crash_api_error(status_code: int, code: str, message: str, retryable: bool = False):
    raise CrashApiException(status_code, code, message, retryable)


def bearer_token_from_request(request: Request) -> str:
    value = request.headers.get("authorization", "")
    if not value.lower().startswith("bearer "):
        crash_api_error(401, "unauthorized", "missing bearer token", False)
    token = value[7:].strip()
    if not token:
        crash_api_error(401, "unauthorized", "missing bearer token", False)
    return token


def require_crash_bearer(request: Request, allowed_roles: set[str]) -> str:
    token = bearer_token_from_request(request)
    roles: set[str] = set()
    if CRASH_REPORT_TOKEN and secrets.compare_digest(token, CRASH_REPORT_TOKEN):
        roles.add("crash_report")
    if CRASH_SYMBOL_TOKEN and secrets.compare_digest(token, CRASH_SYMBOL_TOKEN):
        roles.add("symbols")
    if CRASH_ADMIN_TOKEN:
        if secrets.compare_digest(token, CRASH_ADMIN_TOKEN):
            roles.add("admin")
    elif secrets.compare_digest(token, ADMIN_TOKEN):
        roles.add("admin")
    if not roles:
        crash_api_error(401, "unauthorized", "invalid bearer token", False)
    if roles.isdisjoint(allowed_roles):
        crash_api_error(403, "forbidden", "token is not allowed to access this endpoint", False)
    return token


def ensure_multipart_request(request: Request, max_bytes: int = CRASH_REQUEST_MAX_BYTES):
    content_type = request.headers.get("content-type", "")
    if "multipart/form-data" not in content_type.lower():
        crash_api_error(415, "unsupported_media_type", "Content-Type must be multipart/form-data", False)
    raw_length = request.headers.get("content-length")
    if raw_length:
        try:
            content_length = int(raw_length)
        except ValueError:
            crash_api_error(400, "metadata_invalid", "Content-Length is invalid", False)
        if content_length > max_bytes:
            crash_api_error(413, "payload_too_large", "request body exceeds limit", False)


def is_upload_file(value) -> bool:
    return hasattr(value, "filename") and hasattr(value, "read")


async def read_metadata_field(value, max_bytes: int) -> bytes:
    if value is None:
        crash_api_error(400, "metadata_invalid", "metadata is required", False)
    if is_upload_file(value):
        data = bytearray()
        while True:
            chunk = await value.read(64 * 1024)
            if not chunk:
                break
            data.extend(chunk)
            if len(data) > max_bytes:
                crash_api_error(413, "payload_too_large", "metadata exceeds size limit", False)
        return bytes(data)
    data = str(value).encode("utf-8")
    if len(data) > max_bytes:
        crash_api_error(413, "payload_too_large", "metadata exceeds size limit", False)
    return data


async def save_upload_limited(upload, target: Path, max_bytes: int, field_name: str) -> dict:
    if not is_upload_file(upload):
        crash_api_error(400, "metadata_invalid", f"{field_name} file is required", False)
    target.parent.mkdir(parents=True, exist_ok=True)
    total = 0
    digest = hashlib.sha256()
    try:
        with open(target, "wb") as out_file:
            while True:
                chunk = await upload.read(1024 * 1024)
                if not chunk:
                    break
                total += len(chunk)
                if total > max_bytes:
                    crash_api_error(413, "payload_too_large", f"{field_name} exceeds size limit", False)
                digest.update(chunk)
                out_file.write(chunk)
    except Exception:
        if target.exists():
            target.unlink()
        raise
    if total <= 0:
        crash_api_error(400, "metadata_invalid", f"{field_name} file is empty", False)
    return {"sha256": digest.hexdigest(), "size": total, "filename": upload.filename or field_name}


def load_json_metadata(raw: bytes, error_code: str = "metadata_invalid") -> dict:
    try:
        obj = json.loads(raw.decode("utf-8"))
    except Exception:
        crash_api_error(400, error_code, "metadata must be valid UTF-8 JSON", False)
    if not isinstance(obj, dict):
        crash_api_error(400, error_code, "metadata must be a JSON object", False)
    return obj


def require_metadata_fields(metadata: dict, fields: list[str], error_code: str = "metadata_invalid"):
    for field in fields:
        if metadata.get(field) in (None, ""):
            crash_api_error(400, error_code, f"metadata.{field} is required", False)
    if metadata.get("schemaVersion") != 1:
        crash_api_error(400, error_code, "metadata.schemaVersion must be 1", False)


def normalize_declared_sha256(value, field_path: str) -> str:
    if value in (None, ""):
        return ""
    text = str(value).strip().lower()
    if len(text) != 64 or any(ch not in "0123456789abcdef" for ch in text):
        crash_api_error(400, "metadata_invalid", f"{field_path} must be a SHA-256 hex string", False)
    return text


def declared_file_sha256(metadata: dict, kind: str, filename: str) -> str:
    files = metadata.get("files")
    if not isinstance(files, list):
        crash_api_error(400, "metadata_invalid", "metadata.files must be an array", False)
    for index, item in enumerate(files):
        if not isinstance(item, dict):
            crash_api_error(400, "metadata_invalid", f"metadata.files[{index}] must be an object", False)
        if item.get("kind") == kind or item.get("name") == filename:
            return normalize_declared_sha256(item.get("sha256"), f"metadata.files[{index}].sha256")
    return ""


def validate_client_report_id(value: str):
    try:
        uuid.UUID(value)
    except Exception:
        crash_api_error(400, "metadata_invalid", "metadata.clientReportId must be a UUID", False)


def safe_storage_component(value, fallback: str = "unknown") -> str:
    text = str(value or fallback).strip()
    cleaned = "".join(ch if ch.isalnum() or ch in "._-" else "_" for ch in text)
    return (cleaned[:96] or fallback)


def crash_content_hash(metadata_sha256: str, minidump_sha256: str, attachments_sha256: str) -> str:
    payload = json.dumps(
        {"metadata": metadata_sha256, "minidump": minidump_sha256, "attachments": attachments_sha256 or ""},
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


def generate_prefixed_id(conn, table: str, column: str, prefix: str) -> str:
    day = datetime.now(timezone.utc).strftime("%Y%m%d")
    for _ in range(20):
        value = f"{prefix}_{day}_{secrets.token_hex(4)}"
        if not conn.execute(f"SELECT 1 FROM {table} WHERE {column}=?", (value,)).fetchone():
            return value
    crash_api_error(503, "service_unavailable", "could not allocate report id", True)


def cleanup_path(path: Path | None):
    if not path:
        return
    try:
        if path.is_dir():
            shutil.rmtree(path)
        elif path.exists():
            path.unlink()
    except Exception as err:
        print(f"清理临时文件失败: {path}: {err}")


def crash_report_response(row, duplicate: bool, status_code: int):
    return JSONResponse(
        status_code=status_code,
        content={
            "reportId": row["report_id"],
            "duplicate": duplicate,
            "status": row["status"],
            "acceptedAtUtc": row["received_at_utc"],
        },
    )


def symbol_upload_response(row, duplicate: bool, status_code: int):
    return JSONResponse(
        status_code=status_code,
        content={
            "symbolUploadId": row["symbol_upload_id"],
            "status": row["status"],
            "duplicate": duplicate,
            "acceptedAtUtc": row["received_at_utc"],
        },
    )


@app.get("/api/v1/health")
def crash_health():
    return {
        "status": "ok",
        "service": "simcae-crash-server",
        "version": CRASH_SERVICE_VERSION,
        "timeUtc": utc_now_text(),
    }


@app.post("/api/v1/crash-reports")
async def create_crash_report(request: Request):
    tmp_dir: Path | None = None
    final_dir: Path | None = None
    stored = False
    conn = None
    try:
        require_crash_bearer(request, {"crash_report"})
        ensure_multipart_request(request)
        try:
            form = await request.form()
        except Exception:
            crash_api_error(400, "metadata_invalid", "multipart form is invalid", False)

        metadata_raw = await read_metadata_field(form.get("metadata"), CRASH_METADATA_MAX_BYTES)
        metadata_sha = hashlib.sha256(metadata_raw).hexdigest()
        metadata = load_json_metadata(metadata_raw)
        require_metadata_fields(
            metadata,
            ["schemaVersion", "clientReportId", "crashTimeUtc", "product", "appVersion", "gitCommit", "buildType", "channel", "platform", "crash", "userConsent", "files"],
        )
        if metadata.get("product") != "SIMCAE":
            crash_api_error(400, "metadata_invalid", "metadata.product must be SIMCAE", False)
        if not isinstance(metadata.get("platform"), dict):
            crash_api_error(400, "metadata_invalid", "metadata.platform must be an object", False)
        if not isinstance(metadata.get("crash"), dict):
            crash_api_error(400, "metadata_invalid", "metadata.crash must be an object", False)
        if not isinstance(metadata.get("userConsent"), dict):
            crash_api_error(400, "metadata_invalid", "metadata.userConsent must be an object", False)

        client_report_id = str(metadata.get("clientReportId") or "").strip()
        validate_client_report_id(client_report_id)
        idempotency_key = request.headers.get("idempotency-key", "").strip()
        if not idempotency_key:
            crash_api_error(400, "metadata_invalid", "Idempotency-Key header is required", False)
        if idempotency_key != client_report_id:
            crash_api_error(400, "metadata_invalid", "Idempotency-Key must equal metadata.clientReportId", False)

        tmp_dir = CRASH_STORAGE_ROOT / "_incoming" / f"crash_{secrets.token_hex(12)}"
        tmp_dir.mkdir(parents=True, exist_ok=False)
        (tmp_dir / "metadata.json").write_bytes(metadata_raw)
        minidump_info = await save_upload_limited(form.get("minidump"), tmp_dir / "crash.dmp", CRASH_MINIDUMP_MAX_BYTES, "minidump")
        declared_dump_sha = declared_file_sha256(metadata, "minidump", "crash.dmp")
        if declared_dump_sha and declared_dump_sha != minidump_info["sha256"]:
            crash_api_error(400, "file_hash_mismatch", "minidump SHA-256 does not match metadata.files", False)

        attachments_info = None
        if form.get("attachments") is not None:
            attachments_info = await save_upload_limited(form.get("attachments"), tmp_dir / "attachments.zip", CRASH_ATTACHMENTS_MAX_BYTES, "attachments")

        package_hash = crash_content_hash(metadata_sha, minidump_info["sha256"], attachments_info["sha256"] if attachments_info else "")
        received_at = utc_now_text()
        content_length = int(request.headers.get("content-length") or 0)
        remote_addr = request.client.host if request.client else ""
        server_info = {
            "storageVersion": 1,
            "clientReportId": client_report_id,
            "receivedAtUtc": received_at,
            "remoteAddress": remote_addr,
            "contentLength": content_length,
        }

        conn = db.get_conn()
        existing = conn.execute("SELECT * FROM crash_reports WHERE client_report_id=?", (client_report_id,)).fetchone()
        if existing:
            if existing["content_hash"] != package_hash:
                crash_api_error(409, "idempotency_conflict", "same clientReportId was uploaded with different content", False)
            return crash_report_response(existing, True, 200)

        report_id = generate_prefixed_id(conn, "crash_reports", "report_id", "srv")
        server_info["reportId"] = report_id
        (tmp_dir / "server.json").write_text(json.dumps(server_info, ensure_ascii=False, indent=2), encoding="utf-8")
        final_dir = CRASH_STORAGE_ROOT / "reports" / report_id
        final_dir.parent.mkdir(parents=True, exist_ok=True)
        tmp_dir.rename(final_dir)
        tmp_dir = None

        crash_obj = metadata.get("crash") or {}
        conn.execute(
            """
            INSERT INTO crash_reports(
              report_id,client_report_id,content_hash,status,symbolication_status,product,app_version,
              git_commit,build_type,channel,exception_code,crash_time_utc,received_at_utc,remote_address,
              content_length,metadata_sha256,metadata_size,minidump_sha256,minidump_size,attachments_sha256,
              attachments_size,storage_path,metadata_json,server_json
            ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
            """,
            (
                report_id,
                client_report_id,
                package_hash,
                "stored",
                "not_started",
                metadata.get("product"),
                metadata.get("appVersion"),
                metadata.get("gitCommit"),
                metadata.get("buildType"),
                metadata.get("channel"),
                str(crash_obj.get("exceptionCode") or ""),
                metadata.get("crashTimeUtc"),
                received_at,
                remote_addr,
                content_length,
                metadata_sha,
                len(metadata_raw),
                minidump_info["sha256"],
                minidump_info["size"],
                attachments_info["sha256"] if attachments_info else "",
                attachments_info["size"] if attachments_info else 0,
                str(final_dir),
                metadata_raw.decode("utf-8", errors="replace"),
                json.dumps(server_info, ensure_ascii=False, separators=(",", ":")),
            ),
        )
        conn.commit()
        stored = True
        row = conn.execute("SELECT * FROM crash_reports WHERE report_id=?", (report_id,)).fetchone()
        return crash_report_response(row, False, 201)
    except sqlite3.IntegrityError:
        if conn:
            conn.rollback()
            client_report_id = ""
            try:
                metadata = load_json_metadata(metadata_raw)
                client_report_id = str(metadata.get("clientReportId") or "").strip()
            except Exception:
                pass
            if client_report_id:
                existing = conn.execute("SELECT * FROM crash_reports WHERE client_report_id=?", (client_report_id,)).fetchone()
                if existing and existing["content_hash"] == package_hash:
                    return crash_report_response(existing, True, 200)
        crash_api_error(409, "idempotency_conflict", "same clientReportId was uploaded with different content", False)
    except CrashApiException:
        if conn:
            conn.rollback()
        raise
    except Exception as err:
        if conn:
            conn.rollback()
        print("崩溃报告上传失败:", err)
        crash_api_error(500, "server_error", "server failed to store crash report", True)
    finally:
        if conn:
            conn.close()
        cleanup_path(tmp_dir)
        if final_dir and not stored:
            cleanup_path(final_dir)


@app.get("/api/v1/crash-reports/{report_id}")
def get_crash_report(report_id: str, request: Request):
    require_crash_bearer(request, {"admin"})
    conn = db.get_conn()
    row = conn.execute("SELECT * FROM crash_reports WHERE report_id=?", (report_id,)).fetchone()
    conn.close()
    if not row:
        crash_api_error(404, "metadata_invalid", "crash report not found", False)
    return {
        "reportId": row["report_id"],
        "clientReportId": row["client_report_id"],
        "status": row["status"],
        "receivedAtUtc": row["received_at_utc"],
        "symbolicationStatus": row["symbolication_status"],
        "appVersion": row["app_version"],
        "gitCommit": row["git_commit"],
        "product": row["product"],
        "buildType": row["build_type"],
        "channel": row["channel"],
        "exceptionCode": row["exception_code"],
        "files": {
            "metadata": {"name": "metadata.json", "sha256": row["metadata_sha256"], "size": row["metadata_size"]},
            "minidump": {"name": "crash.dmp", "sha256": row["minidump_sha256"], "size": row["minidump_size"]},
            "attachments": {"name": "attachments.zip", "sha256": row["attachments_sha256"], "size": row["attachments_size"]} if row["attachments_size"] else None,
            "server": {"name": "server.json"},
        },
    }


@app.get("/api/v1/crash-reports/{report_id}/files/{file_name}")
def download_crash_report_file(report_id: str, file_name: str, request: Request):
    token = require_crash_bearer(request, {"admin"})
    allowed = {
        "metadata": "metadata.json",
        "metadata.json": "metadata.json",
        "minidump": "crash.dmp",
        "crash.dmp": "crash.dmp",
        "attachments": "attachments.zip",
        "attachments.zip": "attachments.zip",
        "server": "server.json",
        "server.json": "server.json",
    }
    stored_name = allowed.get(file_name)
    if not stored_name:
        crash_api_error(404, "metadata_invalid", "file not found", False)
    conn = db.get_conn()
    row = conn.execute("SELECT * FROM crash_reports WHERE report_id=?", (report_id,)).fetchone()
    if not row:
        conn.close()
        crash_api_error(404, "metadata_invalid", "crash report not found", False)
    target = Path(row["storage_path"]) / stored_name
    try:
        conn.execute(
            "INSERT INTO crash_file_access_logs(report_id,file_name,actor_hash,ip,user_agent) VALUES(?,?,?,?,?)",
            (report_id, stored_name, token_digest(token)[:16], request.client.host if request.client else "", request.headers.get("user-agent", "")[:300]),
        )
        conn.commit()
    finally:
        conn.close()
    if not target.is_file():
        crash_api_error(404, "metadata_invalid", "file not found", False)
    return FileResponse(target, media_type="application/octet-stream", filename=stored_name)


@app.post("/api/v1/symbols")
async def upload_symbols(request: Request):
    tmp_dir: Path | None = None
    final_dir: Path | None = None
    stored = False
    conn = None
    try:
        require_crash_bearer(request, {"symbols"})
        ensure_multipart_request(request, CRASH_SYMBOLS_MAX_BYTES + CRASH_METADATA_MAX_BYTES)
        try:
            form = await request.form()
        except Exception:
            crash_api_error(400, "metadata_invalid", "multipart form is invalid", False)
        metadata_raw = await read_metadata_field(form.get("metadata"), CRASH_METADATA_MAX_BYTES)
        metadata_sha = hashlib.sha256(metadata_raw).hexdigest()
        metadata = load_json_metadata(metadata_raw)
        require_metadata_fields(
            metadata,
            ["schemaVersion", "product", "appVersion", "gitCommit", "buildType", "platform", "toolchain", "createdAtUtc", "files"],
        )
        if metadata.get("product") != "SIMCAE":
            crash_api_error(400, "metadata_invalid", "metadata.product must be SIMCAE", False)
        if not isinstance(metadata.get("files"), list):
            crash_api_error(400, "metadata_invalid", "metadata.files must be an array", False)

        tmp_dir = CRASH_STORAGE_ROOT / "_incoming" / f"symbols_{secrets.token_hex(12)}"
        tmp_dir.mkdir(parents=True, exist_ok=False)
        (tmp_dir / "metadata.json").write_bytes(metadata_raw)
        symbols_info = await save_upload_limited(form.get("symbols"), tmp_dir / "symbols.zip", CRASH_SYMBOLS_MAX_BYTES, "symbols")
        identity_payload = json.dumps(
            {
                "product": metadata.get("product"),
                "appVersion": metadata.get("appVersion"),
                "gitCommit": metadata.get("gitCommit"),
                "buildType": metadata.get("buildType"),
                "platform": metadata.get("platform"),
            },
            sort_keys=True,
            separators=(",", ":"),
        )
        identity_hash = hashlib.sha256(identity_payload.encode("utf-8")).hexdigest()
        received_at = utc_now_text()
        conn = db.get_conn()
        existing = conn.execute("SELECT * FROM crash_symbol_uploads WHERE identity_hash=?", (identity_hash,)).fetchone()
        if existing:
            if existing["symbols_sha256"] != symbols_info["sha256"]:
                crash_api_error(409, "idempotency_conflict", "same build identity already has different symbols", False)
            return symbol_upload_response(existing, True, 200)

        upload_id = generate_prefixed_id(conn, "crash_symbol_uploads", "symbol_upload_id", "sym")
        final_dir = CRASH_STORAGE_ROOT / "symbols" / safe_storage_component(metadata.get("product")) / safe_storage_component(metadata.get("appVersion")) / safe_storage_component(metadata.get("gitCommit")) / safe_storage_component(metadata.get("buildType")) / safe_storage_component(metadata.get("platform"))
        final_dir.parent.mkdir(parents=True, exist_ok=True)
        if final_dir.exists():
            shutil.rmtree(final_dir)
        server_info = {
            "storageVersion": 1,
            "symbolUploadId": upload_id,
            "receivedAtUtc": received_at,
            "remoteAddress": request.client.host if request.client else "",
            "contentLength": int(request.headers.get("content-length") or 0),
        }
        (tmp_dir / "server.json").write_text(json.dumps(server_info, ensure_ascii=False, indent=2), encoding="utf-8")
        tmp_dir.rename(final_dir)
        tmp_dir = None
        conn.execute(
            """
            INSERT INTO crash_symbol_uploads(
              symbol_upload_id,identity_hash,status,product,app_version,git_commit,build_type,platform,toolchain,
              created_at_utc,received_at_utc,metadata_sha256,metadata_size,symbols_sha256,symbols_size,storage_path,metadata_json
            ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
            """,
            (
                upload_id,
                identity_hash,
                "stored",
                metadata.get("product"),
                metadata.get("appVersion"),
                metadata.get("gitCommit"),
                metadata.get("buildType"),
                metadata.get("platform"),
                metadata.get("toolchain"),
                metadata.get("createdAtUtc"),
                received_at,
                metadata_sha,
                len(metadata_raw),
                symbols_info["sha256"],
                symbols_info["size"],
                str(final_dir),
                metadata_raw.decode("utf-8", errors="replace"),
            ),
        )
        conn.commit()
        stored = True
        row = conn.execute("SELECT * FROM crash_symbol_uploads WHERE symbol_upload_id=?", (upload_id,)).fetchone()
        return symbol_upload_response(row, False, 201)
    except sqlite3.IntegrityError:
        if conn:
            conn.rollback()
        crash_api_error(409, "idempotency_conflict", "same build identity already exists", False)
    except CrashApiException:
        if conn:
            conn.rollback()
        raise
    except Exception as err:
        if conn:
            conn.rollback()
        print("符号包上传失败:", err)
        crash_api_error(500, "server_error", "server failed to store symbols", True)
    finally:
        if conn:
            conn.close()
        cleanup_path(tmp_dir)
        if final_dir and not stored:
            cleanup_path(final_dir)

# ==================== 原有客户端接口 完整保留 ====================
# 0. 首次设备登记；仅此客户端接口允许没有设备凭证
@app.post("/api/v1/device/issue")
async def issue_device(request: Request, body: DeviceIssueReq = Body(...)):
    app_id, channel, installation_id = body.app_id.strip(), body.channel.strip(), body.installation_id.strip()
    if not app_id or not validate_channel_code(channel) or not (16 <= len(installation_id) <= 128): raise HTTPException(status_code=400, detail="设备登记参数无效")
    key_hash=token_digest(body.license_key.strip()); now=datetime.now(timezone.utc)
    conn=db.get_conn(); conn.execute("BEGIN IMMEDIATE")
    require_channel(conn, app_id, channel)
    lic=conn.execute("SELECT * FROM licenses WHERE license_key_hash=?",(key_hash,)).fetchone()
    try: expiry=datetime.fromisoformat((lic["valid_until"] if lic else "").replace("Z","+00:00"))
    except ValueError: expiry=now-timedelta(seconds=1)
    if not lic or lic["status"]!="active" or expiry<=now or lic["app_id"]!=app_id or lic["channel_code"]!=channel:
        conn.rollback();conn.close();raise HTTPException(status_code=403,detail={"error":"license_invalid","msg":"授权密钥无效、已过期或不适用于当前应用/渠道"})
    row=conn.execute("SELECT * FROM devices WHERE app_id=? AND installation_id=?",(app_id,installation_id)).fetchone()
    if row and row["disabled"]: reason=row["disabled_reason"] or "设备已被禁用";conn.rollback();conn.close();raise HTTPException(status_code=403,detail={"error":"device_disabled","msg":reason})
    if row and row["license_id"] not in ("",lic["license_id"]): conn.rollback();conn.close();raise HTTPException(status_code=409,detail="该安装实例已绑定其他授权")
    device_id=row["device_id"] if row else "dev_"+secrets.token_hex(16); seq=int(row["credential_seq"]) if row else 1
    bound=conn.execute("SELECT 1 FROM license_devices WHERE license_id=? AND device_id=?",(lic["license_id"],device_id)).fetchone()
    used=conn.execute("SELECT COUNT(*) FROM license_devices WHERE license_id=?",(lic["license_id"],)).fetchone()[0]
    if not bound and used>=int(lic["max_devices"]): conn.rollback();conn.close();raise HTTPException(status_code=403,detail={"error":"license_device_limit","msg":"授权设备数量已达到上限"})
    ip=request.client.host if request.client else ""
    if row: conn.execute("UPDATE devices SET machine_hash=?,license_id=?,channel=?,last_seen_at=datetime('now'),last_ip=? WHERE device_id=?",(body.machine_hash.strip().lower(),lic["license_id"],channel,ip,device_id))
    else: conn.execute("INSERT INTO devices(device_id,app_id,installation_id,machine_hash,credential_seq,last_ip,license_id,channel) VALUES(?,?,?,?,?,?,?,?)",(device_id,app_id,installation_id,body.machine_hash.strip().lower(),seq,ip,lic["license_id"],channel))
    conn.execute("INSERT OR IGNORE INTO license_devices(license_id,device_id) VALUES(?,?)",(lic["license_id"],device_id));conn.commit();conn.close()
    identity={"device_id":device_id,"license_id":lic["license_id"],"app_id":app_id,"channel":channel,"installation_id":installation_id,"credential_seq":seq,"issued_at":now.replace(microsecond=0).isoformat().replace("+00:00","Z"),"valid_until":lic["valid_until"],"signature_alg":"RSA-2048-SHA256","key_id":SIGNING_KEY_ID}
    identity_text,signature=sign_device_identity(identity);return {"identity":identity,"identity_text":identity_text,"signature":signature}

# 1. 查询更新接口
@app.post("/api/v1/update/check")
async def check_update(request: Request, body: CheckUpdateReq = Body(...)):
    if request.state.device_identity["app_id"] != body.app_id or request.state.device_identity["channel"] != body.channel: raise HTTPException(status_code=403, detail="设备凭证与应用/渠道不匹配")
    print("收到客户端版本检测请求，参数：", body.model_dump())
    app_id, cur_ver, channel = body.app_id, body.current_version, body.channel
    conn = db.get_conn()
    require_channel(conn, app_id, channel)
    ver = conn.execute(
        "SELECT * FROM versions WHERE app_id=? AND channel=? AND latest=1", (app_id, channel)
    ).fetchone()
    policy_row = conn.execute(
        "SELECT * FROM version_policies WHERE app_id=? AND channel=?", (app_id, channel)
    ).fetchone()
    conn.close()

    settings = policy_row_to_dict(policy_row)
    latest_ver = ver["version"] if ver else ""
    version_id = int(ver["id"]) if ver else 0
    target_protocol = int(ver["client_protocol"] or 1) if ver else 0
    protocol_compatible = not ver or target_protocol >= body.client_protocol
    upgrade_available = bool(latest_ver) and version_key(latest_ver) > version_key(cur_ver)
    rollback_candidate = bool(latest_ver) and version_key(latest_ver) < version_key(cur_ver)
    rollback_allowed = rollback_candidate and bool(settings["allow_rollback"]) and protocol_compatible
    need_update = upgrade_available or rollback_allowed
    disabled = cur_ver in settings["disabled_versions"]
    below_min = bool(settings["min_supported_version"]) and version_key(cur_ver) < version_key(settings["min_supported_version"])
    allow_run = not disabled and not below_min
    force_update = bool(settings["force_update"] and upgrade_available) or disabled or below_min
    if disabled:
        action, message = "blocked", settings["message"] or f"当前版本 {cur_ver} 已被管理员禁用"
    elif below_min:
        action, message = "force_update", settings["message"] or f"当前版本低于最低支持版本 {settings['min_supported_version']}"
    elif force_update:
        action, message = "force_update", settings["message"] or "必须升级到最新版本后才能继续使用"
    elif rollback_allowed:
        action, message = "rollback_allowed", settings["message"] or f"管理员要求回退到版本 {latest_ver}"
    elif rollback_candidate and not protocol_compatible:
        action, message = "rollback_denied", f"目标版本 {latest_ver} 的客户端协议为 {target_protocol}，低于当前协议 {body.client_protocol}，禁止降级"
    elif rollback_candidate:
        action, message = "rollback_denied", settings["message"] or f"渠道目标版本 {latest_ver} 低于当前版本，策略禁止降级"
    elif upgrade_available:
        action, message = "optional_update", settings["message"] or "发现可用新版本"
    else:
        action, message = "allow", settings["message"] or "当前版本允许使用"

    issued_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
    policy = {
        "app_id": app_id, "channel": channel, "current_version": cur_ver,
        "policy_seq": settings["policy_seq"], "allow_run": allow_run,
        "force_update": force_update, "allow_rollback": settings["allow_rollback"],
        "rollback_targets": [], "offline_allowed": settings["offline_allowed"],
        "issued_at": issued_at, "valid_until": settings["valid_until"],
        "latest_version": latest_ver, "min_supported_version": settings["min_supported_version"],
        "disabled_versions": settings["disabled_versions"], "message": message,
        "signature_alg": "RSA-2048-SHA256", "key_id": SIGNING_KEY_ID
    }
    policy_text, policy_signature = sign_policy(policy)
    policy["signature"] = policy_signature
    return {
        "allow_run": allow_run, "action": action, "message": message,
        "need_update": need_update, "release_available": bool(ver),
        "current_version": cur_ver, "latest_version": latest_ver, "version_id": version_id,
        "force_update": force_update, "allow_rollback": settings["allow_rollback"],
        "rollback": rollback_allowed, "client_protocol": body.client_protocol,
        "target_client_protocol": target_protocol, "protocol_compatible": protocol_compatible,
        "policy_seq": settings["policy_seq"], "policy_valid_until": settings["valid_until"],
        "policy": policy, "policy_text": policy_text, "policy_signature": policy_signature
    }

# 2. 获取文件下载链接
@app.post("/api/v1/update/download-url")
async def get_download_url(request: Request, body: DownloadUrlReq = Body(...)):
    if request.state.device_identity["app_id"] != body.app_id or request.state.device_identity["channel"] != body.channel: raise HTTPException(status_code=403, detail="设备凭证与应用/渠道不匹配")
    ver_id = body.version_id
    conn = db.get_conn()
    version_row = conn.execute("SELECT app_id,channel,version FROM versions WHERE id=?", (ver_id,)).fetchone()
    if not version_row or version_row["app_id"] != body.app_id or version_row["channel"] != body.channel or version_row["version"] != body.version:
        conn.close(); raise HTTPException(status_code=404, detail="版本与应用/渠道不匹配")
    rows = conn.execute(
        "SELECT path, sha256, size FROM version_files WHERE version_id=?",
        (ver_id,)
    ).fetchall()
    identity=request.state.device_identity; ip=request.client.host if request.client else ""; ua=request.headers.get("user-agent", "")[:300]
    conn.executemany("""INSERT INTO download_logs(device_id,license_id,app_id,version,channel_code,file_path,file_size,result,ip,user_agent)
        VALUES(?,?,?,?,?,?,?,?,?,?)""", [(identity["device_id"],identity["license_id"],body.app_id,body.version,body.channel,r["path"],r["size"],"authorized",ip,ua) for r in rows])
    conn.commit(); conn.close()
    res = []
    base_path = f"{body.app_id}/{body.channel}/{body.version}/files/"
    for r in rows:
        full_path = base_path + r["path"]
        url = minio_tool.get_url(full_path)
        res.append({
            "path": r["path"],
            "url": url,
            "sha256": r["sha256"],
            "size": r["size"]
        })
    return {"files": res}

@app.post("/api/v1/update/download-report")
async def report_download(request: Request, body: DownloadReportReq = Body(...)):
    identity=request.state.device_identity
    if identity["app_id"]!=body.app_id or identity["channel"]!=body.channel: raise HTTPException(status_code=403,detail="设备凭证与应用/渠道不匹配")
    if body.result not in ("success","fail"): raise HTTPException(status_code=400,detail="下载结果无效")
    conn=db.get_conn(); ip=request.client.host if request.client else "";ua=request.headers.get("user-agent","")[:300]; values=[]
    for item in body.files[:5000]:
        path=normalize_relative_path(str(item.get("path") or ""));size=max(0,int(item.get("size") or 0));values.append((identity["device_id"],identity["license_id"],body.app_id,body.version,body.channel,path,size,body.result,ip,ua))
    if values: conn.executemany("INSERT INTO download_logs(device_id,license_id,app_id,version,channel_code,file_path,file_size,result,ip,user_agent) VALUES(?,?,?,?,?,?,?,?,?,?)",values)
    conn.commit();conn.close();return {"code":0,"logged":len(values)}

# 2b. 获取版本 manifest
@app.post("/api/v1/update/manifest")
async def get_manifest(request: Request, body: ManifestReq = Body(...)):
    if request.state.device_identity["app_id"] != body.app_id or request.state.device_identity["channel"] != body.channel: raise HTTPException(status_code=403, detail="设备凭证与应用/渠道不匹配")
    ver_id = body.version_id
    conn = db.get_conn()
    ver = conn.execute(
        "SELECT app_id, channel, version, create_time FROM versions WHERE id=?",
        (ver_id,)
    ).fetchone()
    if not ver:
        conn.close()
        raise HTTPException(status_code=404, detail="Version not found")

    rows = conn.execute(
        "SELECT path, sha256, size FROM version_files WHERE version_id=?",
        (ver_id,)
    ).fetchall()
    conn.close()

    files = []
    for r in rows:
        files.append({
            "path": r["path"],
            "sha256": r["sha256"],
            "size": r["size"],
            "executable": is_executable_path(r["path"])
        })

    manifest = {
        "app_id": ver["app_id"],
        "version": ver["version"],
        "channel": ver["channel"],
        "platform": TARGET_PLATFORM,
        "arch": TARGET_ARCH,
        "manifest_seq": int(ver_id),
        "created_at": ver["create_time"],
        "files": files
    }
    manifest_text = canonical_manifest_bytes(manifest).decode("utf-8")
    signature = sign_manifest(manifest)
    manifest["signature"] = signature
    return {
        "manifest": manifest,
        "manifest_text": manifest_text,
        "signature": signature
    }

# 3. 升级结果上报
@app.post("/api/v1/update/report")
async def report(request: Request, body: ReportReq = Body(...)):
    identity=request.state.device_identity
    if identity["app_id"] != body.app_id or identity["device_id"] != body.device_id: raise HTTPException(status_code=403, detail="上报身份与设备凭证不匹配")
    conn = db.get_conn()
    insert_sql = """
    INSERT INTO upgrade_logs(device_id,from_ver,to_ver,result,create_time)
    VALUES (?,?,?,?,datetime('now'))
    """
    conn.execute(insert_sql, (body.device_id, body.from_version, body.to_version, body.result))
    conn.commit()
    conn.close()
    return {"code": 0, "msg": "上报成功"}

# ==================== 管理后台 Admin 全套接口（统一Header鉴权） ====================
@app.post("/admin/token/change")
def admin_change_token(body: ChangeAdminTokenReq, auth=Depends(admin_auth)):
    global ADMIN_TOKEN
    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 个字符")
    ADMIN_TOKEN = new_token
    return {"code": 0, "msg": "管理员令牌已在当前服务进程生效；如需重启后继续使用，请同步修改 .env 的 ADMIN_TOKEN"}

# 应用列表 GET
@app.get("/admin/app/list")
def admin_get_app_list(auth=Depends(admin_auth)):
    conn = db.get_conn()
    rows = conn.execute("SELECT app_id, app_name FROM apps").fetchall()
    conn.close()
    return {"list": [{"app_id": r["app_id"], "app_name": r["app_name"]} for r in rows]}

# 新增应用 POST json
@app.post("/admin/app/add")
def admin_add_app(body: dict, auth=Depends(admin_auth)):
    aid = body["app_id"]
    aname = body["app_name"]
    conn = db.get_conn()
    try:
        conn.execute("INSERT INTO apps(app_id, app_name) VALUES (?, ?)", (aid, aname))
        conn.executemany("INSERT INTO channels(app_id,channel_code,display_name,enabled,sort_order) VALUES(?,?,?,?,?)",
            [(aid,"stable","正式版",1,10),(aid,"preview","预览版",1,20),(aid,"dev","开发版",1,30)])
        conn.commit()
    except Exception as e:
        return {"msg": f"创建失败：{str(e)}"}
    conn.close()
    return {"msg": "应用创建成功"}

@app.get("/admin/channel/list")
def admin_channel_list(app_id: str, include_disabled: bool = True, auth=Depends(admin_auth)):
    conn=db.get_conn();sql="SELECT channel_code,display_name,enabled,sort_order FROM channels WHERE app_id=?"+("" if include_disabled else " AND enabled=1")+" ORDER BY sort_order,channel_code";rows=conn.execute(sql,(app_id,)).fetchall();conn.close();return {"list":[{"channel_code":r["channel_code"],"display_name":r["display_name"],"enabled":bool(r["enabled"]),"sort_order":r["sort_order"]} for r in rows]}

@app.post("/admin/channel/save")
def admin_channel_save(body: dict, auth=Depends(admin_auth)):
    app_id=str(body.get("app_id") or "").strip();code=str(body.get("channel_code") or "").strip();name=str(body.get("display_name") or "").strip();enabled=bool(body.get("enabled",True));order=int(body.get("sort_order") or 100)
    if not app_id or not validate_channel_code(code) or not name or len(name)>64: raise HTTPException(status_code=400,detail="渠道参数无效；代码仅允许字母、数字、下划线和连字符")
    conn=db.get_conn();
    if not conn.execute("SELECT 1 FROM apps WHERE app_id=?",(app_id,)).fetchone():conn.close();raise HTTPException(status_code=404,detail="应用不存在")
    conn.execute("INSERT INTO channels(app_id,channel_code,display_name,enabled,sort_order) VALUES(?,?,?,?,?) ON CONFLICT(app_id,channel_code) DO UPDATE SET display_name=excluded.display_name,enabled=excluded.enabled,sort_order=excluded.sort_order,updated_at=datetime('now')",(app_id,code,name,int(enabled),order));conn.commit();conn.close();return {"msg":"渠道已保存"}

# 查询和保存应用渠道策略
@app.get("/admin/policy")
def admin_get_policy(app_id: str, channel: str, auth=Depends(admin_auth)):
    conn = db.get_conn()
    row = conn.execute("SELECT * FROM version_policies WHERE app_id=? AND channel=?", (app_id, channel)).fetchone()
    conn.close()
    result = policy_row_to_dict(row)
    result.update({"app_id": app_id, "channel": channel})
    return result


@app.post("/admin/policy/save")
def admin_save_policy(body: dict, auth=Depends(admin_auth)):
    app_id = str(body.get("app_id") or "").strip()
    channel = str(body.get("channel") or "").strip()
    if not app_id or not validate_channel_code(channel):
        raise HTTPException(status_code=400, detail="App ID 或渠道无效")
    disabled = body.get("disabled_versions") or []
    if not isinstance(disabled, list) or any(not isinstance(v, str) for v in disabled):
        raise HTTPException(status_code=400, detail="disabled_versions 必须是版本字符串数组")
    valid_until = str(body.get("valid_until") or "").strip()
    try:
        datetime.fromisoformat(valid_until.replace("Z", "+00:00"))
    except ValueError:
        raise HTTPException(status_code=400, detail="valid_until 必须是 ISO-8601 时间")
    conn = db.get_conn()
    require_channel(conn, app_id, channel)
    row = conn.execute("SELECT policy_seq FROM version_policies WHERE app_id=? AND channel=?", (app_id, channel)).fetchone()
    next_seq = (int(row["policy_seq"]) + 1) if row else 1
    values = (next_seq, int(bool(body.get("force_update"))), int(bool(body.get("allow_rollback"))),
              int(bool(body.get("offline_allowed"))), valid_until,
              str(body.get("min_supported_version") or "").strip(),
              json.dumps(disabled, ensure_ascii=False), str(body.get("message") or "").strip(), app_id, channel)
    conn.execute("""
        INSERT INTO version_policies(policy_seq,force_update,allow_rollback,offline_allowed,valid_until,
          min_supported_version,disabled_versions,message,app_id,channel)
        VALUES(?,?,?,?,?,?,?,?,?,?)
        ON CONFLICT(app_id,channel) DO UPDATE SET policy_seq=excluded.policy_seq,
          force_update=excluded.force_update,allow_rollback=excluded.allow_rollback,
          offline_allowed=excluded.offline_allowed,valid_until=excluded.valid_until,
          min_supported_version=excluded.min_supported_version,disabled_versions=excluded.disabled_versions,
          message=excluded.message,updated_at=datetime('now')
    """, values)
    conn.commit(); conn.close()
    return {"msg": "策略已保存并递增 policy_seq", "policy_seq": next_seq}


# 发布新版本（文件上传）
@app.post("/admin/publish")
async def admin_publish_version(request: Request, auth=Depends(admin_auth)):
    # 在 Starlette 解析 multipart 前检查空间；否则大文件 rollover 会直接抛出 Errno 28。
    content_length_header = request.headers.get("content-length")
    try:
        content_length = int(content_length_header or 0)
    except ValueError:
        raise HTTPException(status_code=400, detail="无效的 Content-Length")
    if content_length <= 0:
        raise HTTPException(status_code=411, detail="发布请求必须提供 Content-Length")

    spool_free = shutil.disk_usage(UPLOAD_SPOOL_DIR).free
    storage_free = shutil.disk_usage(MINIO_DATA_DIR).free
    if spool_free < content_length + 64 * 1024 * 1024:
        raise HTTPException(status_code=507, detail={
            "error": "upload_temp_space_insufficient",
            "msg": "上传临时空间不足，请减小发布包或扩容 /dev/shm",
            "required_bytes": content_length + 64 * 1024 * 1024,
            "available_bytes": spool_free,
        })
    if storage_free < content_length + UPLOAD_SPACE_RESERVE:
        raise HTTPException(status_code=507, detail={
            "error": "storage_space_insufficient",
            "msg": "版本存储空间不足，请扩容服务器根卷或删除不再需要的历史版本",
            "required_bytes": content_length + UPLOAD_SPACE_RESERVE,
            "available_bytes": storage_free,
        })

    # 手动解析表单，增加兼容性并便于调试前端上传问题
    try:
        form = await request.form()
    except OSError as err:
        if getattr(err, "errno", None) == 28:
            raise HTTPException(status_code=507, detail="上传临时空间已满，请扩容后重试")
        raise
    # 尝试从 form 中获取字段
    app_id = form.get("app_id")
    channel = form.get("channel") or ""
    version = form.get("version")
    try:
        client_protocol = int(form.get("client_protocol") or 2)
    except (TypeError, ValueError):
        raise HTTPException(status_code=400, detail="client_protocol 必须是正整数")
    if client_protocol < 1:
        raise HTTPException(status_code=400, detail="client_protocol 必须是正整数")

    # 收集 files（可能是多个同名字段）
    files = []
    if hasattr(form, "getlist"):
        files = form.getlist("files")
    else:
        # 兼容性降级：遍历所有表单项
        for k, v in form.multi_items():
            if k == "files":
                files.append(v)

    # 如果关键信息缺失，返回详细调试信息，方便前端定位问题
    if not app_id or not version:
        headers = {k: v for k, v in request.headers.items()}
        detail = {
            "error": "missing form fields",
            "have_app_id": bool(app_id),
            "have_version": bool(version),
            "content_type": request.headers.get("content-type"),
            "headers": headers,
            "form_keys": list(form.keys()),
            "app_id_value": app_id,
            "version_value": version,
            "app_id_len": len(app_id or ""),
            "version_len": len(version or "")
        }
        raise HTTPException(status_code=422, detail=detail)

    if not files:
        raise HTTPException(status_code=400, detail="未选择任何文件，请选择程序文件夹")

    relative_paths = form.getlist("relative_paths") if hasattr(form, "getlist") else []
    if relative_paths and len(relative_paths) != len(files):
        raise HTTPException(status_code=400, detail="文件数量与相对路径数量不一致")

    upload_items = []
    seen_paths = set()
    for index, file in enumerate(files):
        if not hasattr(file, "filename"):
            raise HTTPException(status_code=400, detail="上传内容中包含无效文件")
        raw_path = relative_paths[index] if relative_paths else file.filename
        rel_path = normalize_relative_path(str(raw_path))
        validate_release_path(rel_path)
        path_key = rel_path.lower()
        if path_key in seen_paths:
            raise HTTPException(status_code=400, detail=f"存在重复文件路径: {rel_path}")
        seen_paths.add(path_key)
        upload_items.append((file, rel_path))

    required_main_key = RELEASE_MAIN_EXECUTABLE.casefold()
    if required_main_key not in seen_paths:
        nested_main = next((path for path in seen_paths if path.endswith("/" + required_main_key)), None)
        if nested_main:
            raise HTTPException(status_code=400, detail=f"{RELEASE_MAIN_EXECUTABLE} 不在发布根级，请改为选择其所在目录: {nested_main}")
        raise HTTPException(status_code=400, detail=f"发布根目录中缺少 {RELEASE_MAIN_EXECUTABLE}")
    nested_main = next((path for path in seen_paths if path.endswith("/" + required_main_key)), None)
    if nested_main:
        raise HTTPException(
            status_code=400,
            detail=f"发布目录中存在嵌套的重复主程序 {nested_main}，请使用干净的 Release 输出目录"
        )


    conn = db.get_conn()
    require_channel(conn, str(app_id), str(channel))
    publish_started = False
    publish_prefix = f"{app_id}/{channel}/{version}/"
    try:
        cur = conn.cursor()
        exists = cur.execute(
            "SELECT 1 FROM versions WHERE app_id=? AND channel=? AND version=?",
            (app_id, channel, version)
        ).fetchone()
        if exists:
            raise HTTPException(status_code=400, detail="该版本已存在，请更换版本号或先删除旧版本")

        cur.execute("UPDATE versions SET latest=0 WHERE app_id=? AND channel=?", (app_id, channel))
        cur.execute(
            "INSERT INTO versions(app_id, channel, version, latest, client_protocol) VALUES (?,?,?,1,?)",
            (app_id, channel, version, client_protocol)
        )
        new_vid = cur.lastrowid
        publish_started = True

        for file, rel_path in upload_items:
            content = await file.read()

            f_size = len(content)
            f_sha = calc_sha256(content)
            obj_path = f"{app_id}/{channel}/{version}/files/{rel_path}"
            put_result = minio_tool.put_file(obj_path, content, f_size)
            if put_result.get('storage') == 'local':
                print(f"MinIO 不可用，已保存到本地回退：{put_result.get('path')}")
            cur.execute(
                "INSERT INTO version_files(version_id, path, sha256, size) VALUES (?,?,?,?)",
                (new_vid, rel_path, f_sha, f_size)
            )

        conn.commit()
    except sqlite3.IntegrityError as err:
        conn.rollback()
        if publish_started:
            minio_tool.remove_prefix(publish_prefix)
        if 'UNIQUE constraint failed: versions.app_id, versions.channel, versions.version' in str(err):
            raise HTTPException(status_code=400, detail="该版本已存在，请更换版本号或先删除旧版本")
        raise HTTPException(status_code=500, detail=str(err))
    except sqlite3.OperationalError as err:
        conn.rollback()
        if publish_started:
            minio_tool.remove_prefix(publish_prefix)
        raise HTTPException(status_code=500, detail={"error": "database locked", "msg": str(err)})
    except HTTPException:
        conn.rollback()
        if publish_started:
            minio_tool.remove_prefix(publish_prefix)
        raise
    except Exception as err:
        conn.rollback()
        if publish_started:
            minio_tool.remove_prefix(publish_prefix)
        if isinstance(err, OSError) and getattr(err, "errno", None) == 28:
            raise HTTPException(status_code=507, detail="发布过程中存储空间耗尽，已清理未完成版本")
        raise HTTPException(status_code=500, detail={"error": "publish_failed", "msg": str(err)})
    finally:
        conn.close()

    return {"msg": f"版本 {version} 发布完成，共上传 {len(upload_items)} 个文件"}

@app.post("/admin/version/offline-package")
def admin_offline_package(body: dict, auth=Depends(admin_auth)):
    version_id = int(body.get("version_id") or 0)
    conn=db.get_conn();ver=conn.execute("SELECT app_id,channel,version,create_time FROM versions WHERE id=?",(version_id,)).fetchone();rows=conn.execute("SELECT path,sha256,size FROM version_files WHERE version_id=? ORDER BY id",(version_id,)).fetchall();conn.close()
    if not ver or not rows: raise HTTPException(status_code=404,detail="版本或版本文件不存在")
    files=[{"path":r["path"],"sha256":r["sha256"],"size":r["size"],"executable":is_executable_path(r["path"])} for r in rows]
    manifest={"app_id":ver["app_id"],"version":ver["version"],"channel":ver["channel"],"platform":TARGET_PLATFORM,"arch":TARGET_ARCH,"manifest_seq":version_id,"created_at":ver["create_time"],"files":files}
    manifest_text=canonical_manifest_bytes(manifest).decode();manifest_signature=sign_manifest(manifest)
    offset=0;entries=[]
    for r in rows: entries.append({"path":r["path"],"offset":offset,"size":r["size"],"sha256":r["sha256"]});offset+=int(r["size"] or 0)
    package={"format":"MUPD0001","app_id":ver["app_id"],"channel":ver["channel"],"version":ver["version"],"version_id":version_id,"created_at":datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00","Z"),"manifest_sha256":hashlib.sha256(manifest_text.encode()).hexdigest(),"payload_size":offset,"files":entries,"signature_alg":"RSA-2048-SHA256","key_id":SIGNING_KEY_ID}
    package_text,package_signature=sign_policy(package)
    wrapper=json.dumps({"package_text":package_text,"package_signature":package_signature,"manifest_text":manifest_text,"manifest_signature":manifest_signature},ensure_ascii=False,separators=(",",":")).encode()
    prefix=f"{ver['app_id']}/{ver['channel']}/{ver['version']}/files/"
    def stream():
        yield b"MUPD0001";yield len(wrapper).to_bytes(8,"little");yield wrapper
        for r in rows:
            yield from minio_tool.iter_file(prefix+r["path"])
    filename=f"{ver['app_id']}_{ver['channel']}_{ver['version']}.upd"
    return StreamingResponse(stream(),media_type="application/octet-stream",headers={"Content-Disposition":f'attachment; filename="{filename}"',"Content-Length":str(16+len(wrapper)+offset)})

# 获取版本列表 GET
@app.get("/admin/version/list")
def admin_get_version_list(app_id: str, auth=Depends(admin_auth)):
    conn = db.get_conn()
    rows = conn.execute("""
        SELECT id, version, channel, latest, client_protocol, create_time
        FROM versions WHERE app_id=? ORDER BY create_time DESC
    """, (app_id,)).fetchall()
    conn.close()
    out = []
    for r in rows:
        out.append({
            "id": r["id"],
            "version": r["version"],
            "channel": r["channel"],
            "latest": bool(r["latest"]),
            "client_protocol": int(r["client_protocol"] or 1),
            "create_time": r["create_time"]
        })
    return {"list": out}

# 修正历史版本的客户端协议标记
@app.post("/admin/version/set-protocol")
def admin_set_version_protocol(body: dict, auth=Depends(admin_auth)):
    version_id = int(body.get("version_id") or 0)
    client_protocol = int(body.get("client_protocol") or 0)
    if version_id < 1 or client_protocol < 1:
        raise HTTPException(status_code=400, detail="版本 ID 和客户端协议必须是正整数")
    conn = db.get_conn()
    cur = conn.execute("UPDATE versions SET client_protocol=? WHERE id=?", (client_protocol, version_id))
    if cur.rowcount != 1:
        conn.close(); raise HTTPException(status_code=404, detail="版本不存在")
    conn.commit(); conn.close()
    return {"msg": "客户端协议已更新", "version_id": version_id, "client_protocol": client_protocol}


# 设置指定版本为渠道最新 POST json
@app.post("/admin/version/set-latest")
def admin_set_latest(body: dict, auth=Depends(admin_auth)):
    vid = body["version_id"]
    conn = db.get_conn()
    cur = conn.cursor()
    v_info = cur.execute("SELECT app_id, channel FROM versions WHERE id=?", (vid,)).fetchone()
    aid, ch = v_info["app_id"], v_info["channel"]
    cur.execute("UPDATE versions SET latest=0 WHERE app_id=? AND channel=?", (aid, ch))
    cur.execute("UPDATE versions SET latest=1 WHERE id=?", (vid,))
    conn.commit()
    conn.close()
    return {"msg": "已切换为渠道最新版本"}

# 删除版本 POST json
@app.post("/admin/version/delete")
def admin_delete_version(body: dict, auth=Depends(admin_auth)):
    vid = body["version_id"]
    conn = db.get_conn()
    cur = conn.cursor()
    v_info = cur.execute("SELECT app_id, channel, version, latest FROM versions WHERE id=?", (vid,)).fetchone()
    if not v_info:
        conn.close()
        raise HTTPException(status_code=404, detail="版本不存在")
    aid, ch, ver = v_info["app_id"], v_info["channel"], v_info["version"]
    was_latest = bool(v_info["latest"])
    # 删除minio下该版本全部文件
    prefix = f"{aid}/{ch}/{ver}/files/"
    minio_tool.remove_prefix(prefix)
    # 删除数据库记录
    cur.execute("DELETE FROM version_files WHERE version_id=?", (vid,))
    cur.execute("DELETE FROM versions WHERE id=?", (vid,))
    promoted_version = None
    if was_latest:
        replacement = cur.execute(
            "SELECT id, version FROM versions WHERE app_id=? AND channel=? ORDER BY create_time DESC, id DESC LIMIT 1",
            (aid, ch)
        ).fetchone()
        if replacement:
            cur.execute("UPDATE versions SET latest=1 WHERE id=?", (replacement["id"],))
            promoted_version = replacement["version"]
    conn.commit()
    conn.close()
    return {
        "msg": "版本、云端文件、数据库记录全部删除完成",
        "promoted_latest": promoted_version
    }

# 获取升级日志 GET
@app.get("/admin/report/list")
def admin_get_report_log(auth=Depends(admin_auth)):
    conn = db.get_conn()
    rows = conn.execute("""
        SELECT device_id,from_ver,to_ver,result,create_time
        FROM upgrade_logs ORDER BY create_time DESC LIMIT 200
    """).fetchall()
    conn.close()
    res = []
    for r in rows:
        res.append({
            "device_id": r["device_id"],
            "app_id": "",
            "from_version": r["from_ver"],
            "to_version": r["to_ver"],
            "result": r["result"],
            "create_time": r["create_time"]
        })
    return {"list": res}


@app.get("/admin/runtime-config")
def admin_runtime_config(auth=Depends(admin_auth)):
    return {
        "release_main_executable": RELEASE_MAIN_EXECUTABLE,
        "target_platform": TARGET_PLATFORM,
        "target_arch": TARGET_ARCH,
    }


@app.get("/admin/download-log/list")
def admin_download_log_list(app_id: str="", limit: int=300, auth=Depends(admin_auth)):
    limit=max(1,min(limit,1000));conn=db.get_conn();rows=conn.execute("SELECT * FROM download_logs WHERE app_id=? ORDER BY id DESC LIMIT ?",(app_id,limit)).fetchall() if app_id else conn.execute("SELECT * FROM download_logs ORDER BY id DESC LIMIT ?",(limit,)).fetchall();conn.close();return {"list":[dict(r) for r in rows]}

@app.get("/admin/audit-log/list")
def admin_audit_log_list(limit: int=300, auth=Depends(admin_auth)):
    conn=db.get_conn();rows=conn.execute("SELECT * FROM admin_audit_logs ORDER BY id DESC LIMIT ?",(max(1,min(limit,1000)),)).fetchall();conn.close();return {"list":[dict(r) for r in rows]}

@app.post("/admin/license/create")
def admin_license_create(body: dict, auth=Depends(admin_auth)):
    app_id=str(body.get("app_id") or "").strip();channel=str(body.get("channel") or "").strip();customer=str(body.get("customer_name") or "").strip();max_devices=int(body.get("max_devices") or 1);valid_until=str(body.get("valid_until") or "").strip()
    try: expiry=datetime.fromisoformat(valid_until.replace("Z","+00:00"))
    except ValueError: raise HTTPException(status_code=400,detail="valid_until 必须是 ISO-8601 时间")
    if not app_id or not customer or not validate_channel_code(channel) or max_devices<1 or expiry<=datetime.now(timezone.utc): raise HTTPException(status_code=400,detail="授权参数无效")
    license_id="lic_"+secrets.token_hex(12);license_key="MARSCO-"+secrets.token_urlsafe(24);conn=db.get_conn();require_channel(conn,app_id,channel);conn.execute("INSERT INTO licenses(license_id,license_key_hash,customer_name,app_id,channel_code,max_devices,valid_until,status) VALUES(?,?,?,?,?,?,?,'active')",(license_id,token_digest(license_key),customer,app_id,channel,max_devices,valid_until));conn.commit();conn.close();return {"license_id":license_id,"license_key":license_key,"msg":"授权已创建；密钥仅本次返回，请立即保存"}

@app.get("/admin/license/list")
def admin_license_list(app_id: str="", auth=Depends(admin_auth)):
    conn=db.get_conn();rows=conn.execute("SELECT l.*,(SELECT COUNT(*) FROM license_devices d WHERE d.license_id=l.license_id) used_devices FROM licenses l WHERE app_id=? ORDER BY created_at DESC",(app_id,)).fetchall() if app_id else conn.execute("SELECT l.*,(SELECT COUNT(*) FROM license_devices d WHERE d.license_id=l.license_id) used_devices FROM licenses l ORDER BY created_at DESC").fetchall();conn.close();return {"list":[{k:r[k] for k in ("license_id","customer_name","app_id","channel_code","max_devices","valid_until","status","created_at")}|{"used_devices":r["used_devices"]} for r in rows]}

@app.post("/admin/license/set-status")
def admin_license_set_status(body: dict, auth=Depends(admin_auth)):
    status=str(body.get("status") or "");license_id=str(body.get("license_id") or "");
    if status not in ("active","disabled"): raise HTTPException(status_code=400,detail="授权状态无效")
    conn=db.get_conn();cur=conn.execute("UPDATE licenses SET status=?,updated_at=datetime('now') WHERE license_id=?",(status,license_id));conn.commit();conn.close()
    if cur.rowcount!=1: raise HTTPException(status_code=404,detail="授权不存在")
    return {"msg":"授权状态已更新"}

@app.get("/admin/device/list")
def admin_device_list(app_id: str = "", auth=Depends(admin_auth)):
    conn=db.get_conn(); rows=conn.execute("SELECT * FROM devices WHERE app_id=? ORDER BY last_seen_at DESC",(app_id,)).fetchall() if app_id else conn.execute("SELECT * FROM devices ORDER BY last_seen_at DESC LIMIT 500").fetchall(); conn.close()
    return {"list":[{"device_id":r["device_id"],"app_id":r["app_id"],"installation_id":r["installation_id"],"disabled":bool(r["disabled"]),"disabled_reason":r["disabled_reason"],"credential_seq":r["credential_seq"],"first_seen_at":r["first_seen_at"],"last_seen_at":r["last_seen_at"],"last_ip":r["last_ip"]} for r in rows]}

@app.post("/admin/device/set-disabled")
def admin_device_set_disabled(body: dict, auth=Depends(admin_auth)):
    device_id=str(body.get("device_id") or "").strip(); disabled=bool(body.get("disabled")); reason=str(body.get("reason") or "").strip(); conn=db.get_conn(); cur=conn.execute("UPDATE devices SET disabled=?,disabled_reason=? WHERE device_id=?",(int(disabled),reason if disabled else "",device_id))
    if cur.rowcount != 1: conn.close(); raise HTTPException(status_code=404,detail="设备不存在")
    conn.commit(); conn.close(); return {"msg":"设备已禁用" if disabled else "设备已恢复","device_id":device_id}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host=SERVER_HOST, port=SERVER_PORT, workers=1)