update-server/main.py

import os
import shutil
import sqlite3
import subprocess
import tempfile
import time
import urllib.request
from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles

import db
import minio_tool
from app.api.router import include_api_routes
from app.core.audit import admin_audit_middleware
from app.core.config import configured_path, env_bool, settings
from app.core.security import ensure_default_admin_user
from app.services.crash_api_service import ensure_storage_dirs
from app.services.device_credential_service import verify_device_credential
from app.services.publish_service import UPLOAD_SPOOL_DIR


SERVER_HOST = settings.server_host
SERVER_PORT = settings.server_port
VALID_CLIENT_TOKEN = os.getenv("CLIENT_API_TOKEN")
MINIO_DATA_DIR = configured_path("MINIO_DATA_DIR", "minio_data")
tempfile.tempdir = str(UPLOAD_SPOOL_DIR)


@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'")
        license_columns = {row[1] for row in cur.execute("PRAGMA table_info(licenses)").fetchall()}
        if "license_key_cipher" not in license_columns:
            cur.execute("ALTER TABLE licenses ADD COLUMN license_key_cipher TEXT NOT NULL DEFAULT ''")
        policy_columns = {row[1] for row in cur.execute("PRAGMA table_info(version_policies)").fetchall()}
        if "git_tags_enabled" not in policy_columns:
            cur.execute("ALTER TABLE version_policies ADD COLUMN git_tags_enabled INTEGER NOT NULL DEFAULT 0")
        audit_columns = {row[1] for row in cur.execute("PRAGMA table_info(admin_audit_logs)").fetchall()}
        if "actor_username" not in audit_columns:
            cur.execute("ALTER TABLE admin_audit_logs ADD COLUMN actor_username TEXT NOT NULL DEFAULT ''")
        if "actor_roles" not in audit_columns:
            cur.execute("ALTER TABLE admin_audit_logs ADD COLUMN actor_roles TEXT NOT NULL DEFAULT ''")
        if "auth_type" not in audit_columns:
            cur.execute("ALTER TABLE admin_audit_logs ADD COLUMN auth_type TEXT NOT NULL DEFAULT ''")
        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("✅ 成功执行tables.sql，创建数据表与测试数据")
    else:
        print("❌ 错误：当前目录找不到 tables.sql 文件！")
    conn.close()
    print("===== 数据库初始化完成 =====")

    ensure_storage_dirs()
    ensure_default_admin_user()
    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}")


app = FastAPI(title=settings.service_title, lifespan=lifespan)
include_api_routes(app)

app.mount("/static", StaticFiles(directory=str(minio_tool.LOCAL_UPLOAD_ROOT)), name="static")

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")(admin_audit_middleware)


@app.middleware("http")
async def auth_middleware(request: Request, call_next):
    path = request.url.path
    skip_paths = (
        "/admin",
        "/assets",
        "/docs",
        "/openapi.json",
        "/redoc",
        "/static",
        "/favicon.ico",
        "/platform-config.json",
        "/logo.svg",
        "/health",
        "/login",
        "/refresh-token",
        "/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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host=SERVER_HOST, port=SERVER_PORT, workers=1)