108 lines
3.9 KiB
Python
108 lines
3.9 KiB
Python
# --- routers/lock.py ---
|
||
import time
|
||
from fastapi import APIRouter, HTTPException
|
||
from pydantic import BaseModel
|
||
from typing import Dict, Optional
|
||
|
||
# 🌟 只定義自己的子路徑
|
||
router = APIRouter(prefix="/locks", tags=["Lock Management"])
|
||
|
||
# ==========================================
|
||
# 💡 全域鎖定表 (In-Memory Lock Table)
|
||
# 結構: { "path": {"user_id": "...", "username": "...", "expires_at": 1234567890.123} }
|
||
# ==========================================
|
||
ACTIVE_LOCKS: Dict[str, dict] = {}
|
||
|
||
# 鎖定超時時間 (秒)：前端需在此時間內發送 Heartbeat，否則自動釋放
|
||
LOCK_TIMEOUT = 30
|
||
|
||
class LockAcquireReq(BaseModel):
|
||
path: str
|
||
user_id: str # 前端產生的 UUID (Session ID)
|
||
username: str # 登入的帳號 (用於 UI 提示，例如 "admin")
|
||
|
||
class LockActionReq(BaseModel):
|
||
path: str
|
||
user_id: str
|
||
|
||
def clean_expired_locks():
|
||
"""清除已經超時的鎖 (被動式清理)"""
|
||
current_time = time.time()
|
||
expired_paths = [path for path, lock in ACTIVE_LOCKS.items() if lock["expires_at"] < current_time]
|
||
for path in expired_paths:
|
||
del ACTIVE_LOCKS[path]
|
||
|
||
def is_path_locked_by_others(target_path: str, user_id: str) -> tuple[Optional[dict], str]:
|
||
"""
|
||
檢查路徑是否被其他人鎖定，並回傳 (衝突的鎖定資訊, 具體錯誤訊息)
|
||
"""
|
||
clean_expired_locks()
|
||
|
||
for locked_path, lock_info in ACTIVE_LOCKS.items():
|
||
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, ""
|
||
|
||
# ==========================================
|
||
# 💡 API Endpoints
|
||
# ==========================================
|
||
|
||
@router.post("/acquire")
|
||
async def acquire_lock(req: LockAcquireReq):
|
||
"""請求鎖定特定路徑"""
|
||
# 接收回傳的鎖定資訊與具體訊息
|
||
conflict_lock, error_msg = is_path_locked_by_others(req.path, req.user_id)
|
||
|
||
if conflict_lock:
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail=error_msg # 💡 將精準的錯誤訊息傳給前端
|
||
)
|
||
|
||
# 寫入或更新鎖定狀態
|
||
ACTIVE_LOCKS[req.path] = {
|
||
"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()
|
||
|
||
if req.path not in ACTIVE_LOCKS:
|
||
raise HTTPException(status_code=404, detail="鎖定已失效，請重新獲取")
|
||
|
||
if ACTIVE_LOCKS[req.path]["user_id"] != req.user_id:
|
||
raise HTTPException(status_code=403, detail="無權更新他人的鎖定")
|
||
|
||
# 延長鎖定時間
|
||
ACTIVE_LOCKS[req.path]["expires_at"] = time.time() + LOCK_TIMEOUT
|
||
return {"status": "success", "message": "心跳更新成功"}
|
||
|
||
@router.post("/release")
|
||
async def release_lock(req: LockActionReq):
|
||
"""主動釋放鎖定"""
|
||
if req.path in ACTIVE_LOCKS and ACTIVE_LOCKS[req.path]["user_id"] == req.user_id:
|
||
del ACTIVE_LOCKS[req.path]
|
||
return {"status": "success", "message": "鎖定已釋放"}
|
||
|
||
@router.get("/status")
|
||
async def get_lock_status():
|
||
"""獲取目前所有被鎖定的路徑 (供前端 UI 標示 '編輯中' 狀態)"""
|
||
clean_expired_locks()
|
||
return {"status": "success", "data": ACTIVE_LOCKS}
|