KyrosDev's picture
新增用戶管理編輯功能和版本發布表單資料保存
588c4fc
"""
KSTools License Manager - 授權 API 路由
"""
from fastapi import APIRouter, HTTPException, Request, Depends
from fastapi.responses import JSONResponse
from typing import List, Dict, Any
from app.models.license import (
LicenseActivation, LicenseValidation, LicenseCreate, LicenseExtend, LicenseUpdate,
ActivationResponse, ValidationResponse, APIResponse
)
from app.services.license_service import license_service
router = APIRouter()
def get_client_ip(request: Request) -> str:
"""取得客戶端 IP"""
forwarded = request.headers.get("X-Forwarded-For")
if forwarded:
return forwarded.split(",")[0].strip()
return request.client.host
@router.post("/license/activate", response_model=ActivationResponse)
async def activate_license(
activation: LicenseActivation,
request: Request,
client_ip: str = Depends(get_client_ip)
):
"""
Revit Plugin 授權啟用 API
**使用方法:**
```json
{
"license_code": "KS12345678",
"hardware_id": "HW_ABC123456789",
"machine_name": "DESKTOP-DEV01"
}
```
**回應:**
- success: 啟用是否成功
- message: 狀態訊息
- expires_at: 授權到期時間
- user_name: 授權用戶名稱
- hardware_id: 硬體ID
"""
try:
result = await license_service.activate_license(activation, client_ip)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=f"系統錯誤: {str(e)}")
@router.post("/license/validate", response_model=ValidationResponse)
async def validate_license(
validation: LicenseValidation,
request: Request,
client_ip: str = Depends(get_client_ip)
):
"""
Revit Plugin 授權驗證 API
**使用方法:**
```json
{
"license_code": "KS12345678",
"hardware_id": "HW_ABC123456789",
"machine_name": "DESKTOP-DEV01"
}
```
**回應:**
- valid: 授權是否有效
- message: 狀態訊息
- expires_at: 授權到期時間
- user_name: 授權用戶名稱
- hardware_id: 硬體ID
"""
try:
result = await license_service.validate_license(validation, client_ip)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=f"系統錯誤: {str(e)}")
# === 管理 API (供 Web 界面使用) ===
@router.post("/license/create", response_model=APIResponse)
async def create_license(license_data: LicenseCreate):
"""建立新授權 (管理界面使用)"""
try:
result = await license_service.create_license(license_data)
return APIResponse(
success=result["success"],
message=result["message"],
data=result if result["success"] else None
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"系統錯誤: {str(e)}")
@router.get("/licenses")
async def get_all_licenses():
"""取得所有授權列表 (管理界面使用)"""
try:
licenses = await license_service.get_all_licenses()
return {"success": True, "data": licenses}
except Exception as e:
raise HTTPException(status_code=500, detail=f"系統錯誤: {str(e)}")
@router.get("/licenses/stats")
async def get_license_stats():
"""取得授權統計 (管理界面使用)"""
try:
stats = await license_service.get_license_stats()
return {"success": True, "data": stats.dict()}
except Exception as e:
raise HTTPException(status_code=500, detail=f"系統錯誤: {str(e)}")
@router.get("/licenses/logs")
async def get_usage_logs(limit: int = 100):
"""取得使用記錄 (管理界面使用)"""
try:
logs = await license_service.get_usage_logs(limit)
return {"success": True, "data": logs}
except Exception as e:
raise HTTPException(status_code=500, detail=f"系統錯誤: {str(e)}")
@router.patch("/license/{license_id}/toggle")
async def toggle_license_status(license_id: str):
"""切換授權啟用狀態 (管理界面使用)"""
try:
result = await license_service.toggle_license_status(license_id)
return APIResponse(
success=result["success"],
message=result["message"],
data={"new_status": result.get("new_status")} if result["success"] else None
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"系統錯誤: {str(e)}")
@router.delete("/license/{license_id}")
async def delete_license(license_id: str):
"""刪除授權 (管理界面使用)"""
try:
result = await license_service.delete_license(license_id)
return APIResponse(
success=result["success"],
message=result["message"]
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"系統錯誤: {str(e)}")
@router.get("/license/check/{license_code}")
async def check_license_info(license_code: str):
"""檢查授權資訊 (不記錄使用日誌)"""
try:
# 使用服務層方法而不是直接存取資料庫
result = await license_service.get_license_info(license_code)
return APIResponse(
success=result["success"],
message=result["message"],
data=result.get("data")
)
except Exception as e:
raise HTTPException(status_code=500, detail="系統錯誤,請稍後重試")
@router.patch("/license/{license_id}/extend")
async def extend_license(license_id: str, extend_data: LicenseExtend):
"""延長授權天數 (管理界面使用)"""
try:
result = await license_service.extend_license(
license_id,
extend_data.extend_days,
extend_data.reason
)
return APIResponse(
success=result["success"],
message=result["message"],
data=result.get("data")
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"系統錯誤: {str(e)}")
@router.patch("/license/{license_id}")
async def update_license(license_id: str, update_data: LicenseUpdate):
"""更新授權資訊 (管理界面使用)"""
try:
result = await license_service.update_license(license_id, update_data)
return APIResponse(
success=result["success"],
message=result["message"],
data=result.get("data")
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"系統錯誤: {str(e)}")