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)