107 lines
4.0 KiB
Python
107 lines
4.0 KiB
Python
# --- routers/lock.py ---
|
||
import time
|
||
from fastapi import APIRouter, HTTPException, Query
|
||
from pydantic import BaseModel
|
||
from typing import Dict, Optional
|
||
|
||
router = APIRouter(prefix="/locks", tags=["Lock Management"])
|
||
|
||
# ==========================================
|
||
# 💡 全域鎖定表 (In-Memory Lock Table)
|
||
# 🌟 結構升級: { "host@@path": {"user_id": "...", "username": "...", "expires_at": ...} }
|
||
# ⚠️ 嚴格規則：Lock Key 絕對不包含 config_type，確保 running 與 full 視圖共用同一把鎖！
|
||
# ==========================================
|
||
ACTIVE_LOCKS: Dict[str, dict] = {}
|
||
LOCK_TIMEOUT = 120
|
||
|
||
# 🌟 請求模型嚴格排除 config_type
|
||
class LockAcquireReq(BaseModel):
|
||
host: str
|
||
path: str
|
||
user_id: str
|
||
username: str
|
||
|
||
class LockActionReq(BaseModel):
|
||
host: str
|
||
path: str
|
||
user_id: str
|
||
|
||
def clean_expired_locks():
|
||
current_time = time.time()
|
||
expired_keys = [k for k, lock in ACTIVE_LOCKS.items() if lock["expires_at"] < current_time]
|
||
for k in expired_keys:
|
||
del ACTIVE_LOCKS[k]
|
||
|
||
def is_path_locked_by_others(target_host: str, target_path: str, user_id: str) -> tuple[Optional[dict], str]:
|
||
clean_expired_locks()
|
||
prefix = f"{target_host}@@"
|
||
|
||
for locked_key, lock_info in ACTIVE_LOCKS.items():
|
||
# 🌟 只檢查同一台設備 (IP) 的鎖定
|
||
if not locked_key.startswith(prefix):
|
||
continue
|
||
|
||
locked_path = locked_key.split("@@", 1)[1]
|
||
if lock_info["user_id"] == user_id:
|
||
continue
|
||
|
||
if target_path == locked_path:
|
||
return lock_info, f"此區塊正由 [{lock_info['username']}] 編輯中"
|
||
|
||
if target_path.startswith(locked_path + "::"):
|
||
return lock_info, f"父層級 [{locked_path}] 已被 [{lock_info['username']}] 鎖定，無法編輯子區塊"
|
||
|
||
if locked_path.startswith(target_path + "::"):
|
||
return lock_info, f"子層級 [{locked_path}] 已被 [{lock_info['username']}] 鎖定，無法鎖定整個父區塊"
|
||
|
||
return None, ""
|
||
|
||
@router.post("/acquire")
|
||
async def acquire_lock(req: LockAcquireReq):
|
||
conflict_lock, error_msg = is_path_locked_by_others(req.host, req.path, req.user_id)
|
||
|
||
if conflict_lock:
|
||
raise HTTPException(status_code=409, detail=error_msg)
|
||
|
||
# ⚠️ 嚴格綁定 host 與 path，跨視圖共用
|
||
lock_key = f"{req.host}@@{req.path}"
|
||
ACTIVE_LOCKS[lock_key] = {
|
||
"user_id": req.user_id,
|
||
"username": req.username,
|
||
"expires_at": time.time() + LOCK_TIMEOUT
|
||
}
|
||
return {"status": "success", "message": f"成功鎖定路徑: {req.path}"}
|
||
|
||
@router.post("/heartbeat")
|
||
async def heartbeat_lock(req: LockActionReq):
|
||
clean_expired_locks()
|
||
lock_key = f"{req.host}@@{req.path}"
|
||
|
||
if lock_key not in ACTIVE_LOCKS:
|
||
raise HTTPException(status_code=404, detail="鎖定已失效，請重新獲取")
|
||
|
||
if ACTIVE_LOCKS[lock_key]["user_id"] != req.user_id:
|
||
raise HTTPException(status_code=403, detail="無權更新他人的鎖定")
|
||
|
||
ACTIVE_LOCKS[lock_key]["expires_at"] = time.time() + LOCK_TIMEOUT
|
||
return {"status": "success", "message": "心跳更新成功"}
|
||
|
||
@router.post("/release")
|
||
async def release_lock(req: LockActionReq):
|
||
lock_key = f"{req.host}@@{req.path}"
|
||
if lock_key in ACTIVE_LOCKS and ACTIVE_LOCKS[lock_key]["user_id"] == req.user_id:
|
||
del ACTIVE_LOCKS[lock_key]
|
||
return {"status": "success", "message": "鎖定已釋放"}
|
||
|
||
@router.get("/status")
|
||
async def get_lock_status(host: str = Query(None)):
|
||
"""獲取特定設備的鎖定狀態"""
|
||
clean_expired_locks()
|
||
if host:
|
||
prefix = f"{host}@@"
|
||
# 🌟 貼心設計：拔除 host@@ 前綴，讓前端收到的依然是乾淨的 path，UI 不用改！
|
||
filtered_locks = {k.split("@@", 1)[1]: v for k, v in ACTIVE_LOCKS.items() if k.startswith(prefix)}
|
||
return {"status": "success", "data": filtered_locks}
|
||
|
||
return {"status": "success", "data": {}}
|