220 lines
8.9 KiB
Python
220 lines
8.9 KiB
Python
# --- routers/leaf_options.py ---
|
||
from fastapi import APIRouter, BackgroundTasks, HTTPException, Request, Body
|
||
from fastapi.responses import StreamingResponse
|
||
from pydantic import BaseModel
|
||
from typing import List
|
||
import json, os, asyncio, re
|
||
import database
|
||
from cmts_scraper import sync_cmts_leaves_async
|
||
from shared import CMTS_DEVICE, USE_DB
|
||
|
||
router = APIRouter(tags=["Options"])
|
||
|
||
# 🌟 1. 動態獲取快取檔名 (加入 host 隔離)
|
||
def get_cache_file(host: str, config_type: str):
|
||
safe_host = host.replace(".", "_")
|
||
return f"{safe_host}_{config_type}_cache.json"
|
||
|
||
# ==========================================
|
||
# 🌟 2. 全域廣播機制 (SSE) - 升級為「IP + 模式」頻道分流
|
||
# ==========================================
|
||
# 將狀態改為空字典，動態根據 host_configType 建立
|
||
SCAN_STATUS = {}
|
||
active_clients = {}
|
||
|
||
async def broadcast_message(message_dict: dict, host: str, config_type: str = "running"):
|
||
"""將訊息推播給特定頻道的連線前端"""
|
||
channel_key = f"{host}_{config_type}"
|
||
dead_clients = set()
|
||
|
||
# SSE 協定嚴格要求必須以 "data: " 開頭，並以 "\n\n" 結尾
|
||
message_str = f"data: {json.dumps(message_dict)}\n\n"
|
||
|
||
for client_queue in active_clients.get(channel_key, set()):
|
||
try:
|
||
await client_queue.put(message_str)
|
||
except Exception:
|
||
dead_clients.add(client_queue)
|
||
for dead in dead_clients:
|
||
active_clients[channel_key].discard(dead)
|
||
|
||
@router.get("/cmts-leaf-options/stream")
|
||
async def sse_stream(request: Request, host: str, config_type: str = "running"):
|
||
"""前端一載入就會連上這個路由，持續監聽特定頻道的廣播"""
|
||
channel_key = f"{host}_{config_type}"
|
||
client_queue = asyncio.Queue()
|
||
|
||
if channel_key not in active_clients:
|
||
active_clients[channel_key] = set()
|
||
active_clients[channel_key].add(client_queue)
|
||
|
||
async def event_generator():
|
||
try:
|
||
while True:
|
||
# 主動檢查斷線，避免死鎖
|
||
if await request.is_disconnected():
|
||
break
|
||
try:
|
||
# [修復 Leak] 延長 timeout 降低輪詢負擔
|
||
message = await asyncio.wait_for(client_queue.get(), timeout=5.0)
|
||
yield message
|
||
except asyncio.TimeoutError:
|
||
yield ": keepalive\n\n"
|
||
|
||
except asyncio.CancelledError:
|
||
pass
|
||
finally:
|
||
# [修復 Leak] 確保斷線時 Queue 絕對會被移出 active_clients 釋放記憶體
|
||
if channel_key in active_clients:
|
||
active_clients[channel_key].discard(client_queue)
|
||
|
||
return StreamingResponse(event_generator(), media_type="text/event-stream")
|
||
|
||
@router.get("/scan-status")
|
||
async def get_scan_status(host: str, config_type: str = "running"):
|
||
channel_key = f"{host}_{config_type}"
|
||
return {"is_scanning": SCAN_STATUS.get(channel_key, False)}
|
||
|
||
# ==========================================
|
||
# 🌟 3. 快取讀取與背景掃描任務
|
||
# ==========================================
|
||
class SyncOptionsRequest(BaseModel):
|
||
host: str # 🌟 新增 host 參數
|
||
username: str = "" # 🌟 新增 username 參數
|
||
password: str = "" # 🌟 新增 password 參數
|
||
leaf_paths: List[str]
|
||
config_type: str = "running"
|
||
|
||
@router.get("/cmts-leaf-options")
|
||
async def get_leaf_options(host: str, config_type: str = "running"):
|
||
if USE_DB:
|
||
try:
|
||
db_options = await database.get_all_leaf_options(host, config_type)
|
||
if db_options is not None:
|
||
# 取得 metadata
|
||
metadata = await database.get_device_status(host, config_type)
|
||
if metadata:
|
||
db_options["__metadata__"] = metadata
|
||
return db_options
|
||
else:
|
||
# 若回傳 None 表示 DB 連線異常，執行 Fallback
|
||
print("⚠️ [Fallback] 資料庫無法取得資料，自動切換至 JSON 快取讀取...")
|
||
except Exception as e:
|
||
print(f"⚠️ [Fallback] 資料庫讀取發生例外: {e}，自動切換至 JSON 快取讀取...")
|
||
|
||
cache_file = get_cache_file(host, config_type)
|
||
if not os.path.exists(cache_file): return {}
|
||
|
||
# 🌟 優化 3：將讀取動作丟到背景執行緒
|
||
def read_json_from_file(filepath):
|
||
with open(filepath, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
|
||
return await asyncio.to_thread(read_json_from_file, cache_file)
|
||
|
||
async def run_scan_task(host: str, username: str, password: str, paths: List[str], config_type: str):
|
||
"""在背景執行的爬蟲任務，並將進度廣播出去"""
|
||
global SCAN_STATUS
|
||
channel_key = f"{host}_{config_type}"
|
||
try:
|
||
await broadcast_message({"event": "scan_start"}, host, config_type)
|
||
|
||
# 呼叫爬蟲 (帶入設備連線資訊與 config_type)
|
||
async for chunk in sync_cmts_leaves_async(
|
||
host=host,
|
||
username=username,
|
||
password=password,
|
||
leaf_paths=paths,
|
||
config_type=config_type
|
||
):
|
||
if isinstance(chunk, str):
|
||
await broadcast_message(json.loads(chunk), host, config_type)
|
||
else:
|
||
await broadcast_message(chunk, host, config_type)
|
||
|
||
await broadcast_message({"event": "done"}, host, config_type)
|
||
except Exception as e:
|
||
await broadcast_message({"event": "error", "message": str(e)}, host, config_type)
|
||
finally:
|
||
# 解除特定頻道的鎖定
|
||
SCAN_STATUS[channel_key] = False
|
||
|
||
@router.post("/cmts-leaf-options/sync")
|
||
async def sync_leaf_options(request: SyncOptionsRequest, background_tasks: BackgroundTasks):
|
||
global SCAN_STATUS
|
||
if not request.leaf_paths:
|
||
raise HTTPException(status_code=400)
|
||
|
||
channel_key = f"{request.host}_{request.config_type}"
|
||
|
||
if SCAN_STATUS.get(channel_key, False):
|
||
return {"status": "busy", "message": f"⚠️ {request.config_type} 模式的掃描任務正在執行中，請勿重複點擊！"}
|
||
|
||
SCAN_STATUS[channel_key] = True
|
||
|
||
# 保留您原本完美的路徑清理邏輯
|
||
cmts_query_paths = []
|
||
for p in request.leaf_paths:
|
||
clean_p = p.replace("::", " ")
|
||
clean_p = re.sub(r"\s*\[\d+\]", "", clean_p)
|
||
cmts_query_paths.append(clean_p)
|
||
|
||
cmts_query_paths = list(dict.fromkeys(cmts_query_paths))
|
||
|
||
# 決定帳密 (優先使用 request 傳來的，否則 fallback 到 shared.CMTS_DEVICE)
|
||
# 🌟 加上 or "" 確保最終結果絕對是字串，消除 Pylance 警告
|
||
user = request.username or CMTS_DEVICE.get("username") or ""
|
||
pwd = request.password or CMTS_DEVICE.get("password") or ""
|
||
|
||
background_tasks.add_task(run_scan_task, request.host, user, pwd, cmts_query_paths, request.config_type)
|
||
return {"status": "started"}
|
||
|
||
# ==========================================
|
||
# 🌟 4. 清除快取 API
|
||
# ==========================================
|
||
@router.post("/clear_cache")
|
||
async def clear_specific_cache(host: str, config_type: str = "running", paths_to_clear: list = Body(...)):
|
||
cleared_count = 0
|
||
|
||
# 1. 嘗試清除 DB
|
||
if USE_DB:
|
||
try:
|
||
db_cleared = await database.delete_leaf_options(host, config_type, paths_to_clear)
|
||
if db_cleared >= 0:
|
||
cleared_count = db_cleared
|
||
else:
|
||
print("⚠️ [Fallback] 資料庫清除失敗，自動切換至 JSON 快取清除...")
|
||
except Exception as e:
|
||
print(f"⚠️ [Fallback] 資料庫清除發生例外: {e}，自動切換至 JSON 快取清除...")
|
||
|
||
# 2. 清除 JSON (如果 DB 沒清掉，或 USE_DB=False，或者是連同舊檔一起清確保乾淨)
|
||
cache_file = get_cache_file(host, config_type)
|
||
|
||
if not os.path.exists(cache_file):
|
||
# 如果是走 DB，且有清掉，回傳 DB 的結果
|
||
if USE_DB and cleared_count > 0:
|
||
return {"status": "success", "cleared_count": cleared_count}
|
||
return {"status": "success", "cleared_count": 0, "message": "快取檔案不存在"}
|
||
|
||
try:
|
||
with open(cache_file, "r", encoding="utf-8") as f:
|
||
cache_data = json.load(f)
|
||
|
||
json_cleared_count = 0
|
||
for path in paths_to_clear:
|
||
if path in cache_data:
|
||
del cache_data[path]
|
||
json_cleared_count += 1
|
||
|
||
if json_cleared_count > 0:
|
||
with open(cache_file, "w", encoding="utf-8") as f:
|
||
json.dump(cache_data, f, ensure_ascii=False, indent=4)
|
||
|
||
# 回傳較大的那個數值
|
||
final_count = max(cleared_count, json_cleared_count)
|
||
return {"status": "success", "cleared_count": final_count}
|
||
|
||
except Exception as e:
|
||
# 確保回傳標準 JSON 格式
|
||
return {"status": "error", "message": f"資料庫連線與快取存取皆失敗: {str(e)}"}
|