kstools-license-manager / unified_server.py
KyrosDev's picture
新增資料庫保持活躍API端點
f42dbe4
#!/usr/bin/env python3
"""
統一的 FastAPI 服務器,同時處理前端和 API 請求
專為 Hugging Face Spaces 設計
"""
import uvicorn
from fastapi import FastAPI, Request
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
from fastapi.middleware.cors import CORSMiddleware
import os
from pathlib import Path
# 導入 API 路由模組,而不是整個 app
from app.api import license, hardware
from app import config_endpoint
from app.api import version_routes
# 創建主應用
app = FastAPI(
title="KSTools License Manager",
description="統一的前後端服務",
version="2.0.0",
docs_url="/api/docs",
redoc_url="/api/redoc"
)
# 添加 CORS 中間件
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 直接包含 API 路由 (避免掛載衝突)
app.include_router(license.router, prefix="/api", tags=["license"])
app.include_router(hardware.router, prefix="/api", tags=["hardware"])
app.include_router(config_endpoint.router, prefix="/api", tags=["config"])
app.include_router(version_routes.router, prefix="/api", tags=["versions"])
# 靜態檔案路徑
frontend_path = Path(__file__).parent / "frontend"
# 處理靜態檔案
@app.get("/css/{filename:path}")
async def serve_css(filename: str):
file_path = frontend_path / "css" / filename
if file_path.exists():
return FileResponse(file_path, media_type="text/css")
return {"error": "File not found"}, 404
@app.get("/js/{filename:path}")
async def serve_js(filename: str):
file_path = frontend_path / "js" / filename
if file_path.exists():
return FileResponse(file_path, media_type="application/javascript")
return {"error": "File not found"}, 404
@app.get("/config.js")
async def serve_config():
file_path = frontend_path / "config.js"
if file_path.exists():
return FileResponse(file_path, media_type="application/javascript")
return {"error": "File not found"}, 404
# 前端頁面路由
@app.get("/")
async def serve_index():
return FileResponse(frontend_path / "index.html")
@app.get("/login.html")
async def serve_login():
return FileResponse(frontend_path / "login.html")
@app.get("/index.html")
async def serve_index_html():
return FileResponse(frontend_path / "index.html")
# API 健康檢查端點
@app.get("/api/health")
async def api_health_check():
return {
"status": "healthy",
"service": "KSTools License Manager API",
"frontend": "HTML/CSS/JS",
"backend": "FastAPI",
"port": 7860
}
# 根路徑健康檢查
@app.get("/health")
async def health_check():
return {
"status": "healthy",
"service": "KSTools License Manager (Unified)",
"frontend": "HTML/CSS/JS",
"backend": "FastAPI",
"port": 7860
}
# 根 API 資訊
@app.get("/api")
async def api_root():
return {
"message": "KSTools License Manager API",
"status": "running",
"version": "2.0.0",
"docs": "/api/docs"
}
# 資料庫保持活躍端點
@app.get("/api/keep-alive")
async def keep_alive():
"""
資料庫保持活躍 endpoint
執行簡單的資料庫查詢以保持 Supabase 連接活躍
"""
from datetime import datetime, timezone
from app.models.supabase_clients import supabase_clients
try:
# 查詢資料庫以保持連接活躍
client = supabase_clients.get_license_client()
if client:
# 執行簡單查詢:統計授權數量
response = client.table("licenses").select("id", count="exact").limit(1).execute()
db_status = "active"
license_count = response.count if hasattr(response, 'count') else 0
else:
db_status = "disconnected"
license_count = 0
return {
"status": "success",
"database": db_status,
"license_count": license_count,
"timestamp": datetime.now(timezone.utc).isoformat()
}
except Exception as e:
return {
"status": "error",
"database": "error",
"error": str(e),
"timestamp": datetime.now(timezone.utc).isoformat()
}
# 通配符路由最後處理 (避免攔截 API)
@app.get("/{full_path:path}")
async def serve_spa_routes(full_path: str):
# 檢查是否是靜態檔案
file_path = frontend_path / full_path
if file_path.exists() and file_path.is_file():
return FileResponse(file_path)
# 對於所有其他路徑,返回 index.html (SPA 路由)
return FileResponse(frontend_path / "index.html")
if __name__ == "__main__":
port = int(os.environ.get("PORT", 7860))
print(f"🚀 Starting unified KSTools License Manager on port {port}")
print(f"📁 Frontend path: {frontend_path}")
print(f"🔗 Access: http://0.0.0.0:{port}")
uvicorn.run(
"unified_server:app",
host="0.0.0.0",
port=port,
reload=False,
log_level="info"
)