149 lines
5.2 KiB
Python
149 lines
5.2 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
|
||
|
||
from cmts_scraper import sync_cmts_leaves_async
|
||
from shared import CMTS_DEVICE
|
||
|
||
router = APIRouter(tags=["Options"])
|
||
CACHE_FILE = "cmts_leaf_options_cache.json"
|
||
|
||
# ==========================================
|
||
# 🌟 全域廣播機制 (SSE)
|
||
# ==========================================
|
||
IS_SCANNING = False
|
||
active_clients = set()
|
||
|
||
async def broadcast_message(message_dict: dict):
|
||
"""將訊息推播給所有連線中的前端"""
|
||
dead_clients = set()
|
||
|
||
# 🌟 關鍵修復 1：SSE 協定嚴格要求必須以 "data: " 開頭，並以 "\n\n" 結尾
|
||
message_str = f"data: {json.dumps(message_dict)}\n\n"
|
||
|
||
for client_queue in active_clients:
|
||
try:
|
||
await client_queue.put(message_str)
|
||
except Exception:
|
||
dead_clients.add(client_queue)
|
||
for dead in dead_clients:
|
||
active_clients.discard(dead)
|
||
|
||
@router.get("/cmts-leaf-options/stream")
|
||
async def sse_stream(request: Request):
|
||
"""前端一載入就會連上這個路由，持續監聽廣播"""
|
||
client_queue = asyncio.Queue()
|
||
active_clients.add(client_queue)
|
||
|
||
async def event_generator():
|
||
try:
|
||
while True:
|
||
try:
|
||
message = await asyncio.wait_for(client_queue.get(), timeout=2.0)
|
||
yield message
|
||
except asyncio.TimeoutError:
|
||
if await request.is_disconnected():
|
||
break
|
||
# 🌟 關鍵修復 2：每 2 秒發送一個空註解，防止瀏覽器因閒置而自動斷線
|
||
yield ": keepalive\n\n"
|
||
finally:
|
||
active_clients.discard(client_queue)
|
||
|
||
return StreamingResponse(event_generator(), media_type="text/event-stream")
|
||
|
||
@router.get("/scan-status")
|
||
async def get_scan_status():
|
||
global IS_SCANNING
|
||
return {"is_scanning": IS_SCANNING}
|
||
|
||
# ==========================================
|
||
# 🌟 快取讀取與背景掃描任務
|
||
# ==========================================
|
||
class SyncOptionsRequest(BaseModel):
|
||
leaf_paths: List[str]
|
||
|
||
@router.get("/cmts-leaf-options")
|
||
async def get_leaf_options():
|
||
if not os.path.exists(CACHE_FILE): return {}
|
||
with open(CACHE_FILE, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
|
||
async def run_scan_task(paths: List[str]):
|
||
"""在背景執行的爬蟲任務，並將進度廣播出去"""
|
||
global IS_SCANNING
|
||
try:
|
||
await broadcast_message({"event": "scan_start"})
|
||
|
||
# 呼叫爬蟲 (帶入設備連線資訊)
|
||
async for chunk in sync_cmts_leaves_async(
|
||
host=CMTS_DEVICE.get("host"),
|
||
username=CMTS_DEVICE.get("username"),
|
||
password=CMTS_DEVICE.get("password"),
|
||
leaf_paths=paths
|
||
):
|
||
if isinstance(chunk, str):
|
||
await broadcast_message(json.loads(chunk))
|
||
else:
|
||
await broadcast_message(chunk)
|
||
|
||
await broadcast_message({"event": "done"})
|
||
except Exception as e:
|
||
await broadcast_message({"event": "error", "message": str(e)})
|
||
finally:
|
||
IS_SCANNING = False
|
||
|
||
@router.post("/cmts-leaf-options/sync")
|
||
async def sync_leaf_options(request: SyncOptionsRequest, background_tasks: BackgroundTasks):
|
||
"""前端點擊掃描時，只負責觸發背景任務並立即回傳 JSON"""
|
||
global IS_SCANNING
|
||
|
||
if not request.leaf_paths:
|
||
raise HTTPException(status_code=400)
|
||
|
||
if IS_SCANNING:
|
||
return {"status": "busy", "message": "⚠️ 掃描任務正在執行中，請勿重複點擊！"}
|
||
|
||
IS_SCANNING = 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))
|
||
|
||
# 將任務丟入背景執行，並立刻回傳 JSON 給前端
|
||
background_tasks.add_task(run_scan_task, cmts_query_paths)
|
||
return {"status": "started"}
|
||
|
||
# ==========================================
|
||
# 🌟 清除快取 API (對應 api.js 的 POST 請求)
|
||
# ==========================================
|
||
@router.post("/clear_cache")
|
||
async def clear_specific_cache(paths_to_clear: list = Body(...)):
|
||
cleared_count = 0
|
||
if not os.path.exists(CACHE_FILE):
|
||
return {"status": "success", "cleared_count": 0, "message": "快取檔案不存在"}
|
||
|
||
try:
|
||
with open(CACHE_FILE, "r", encoding="utf-8") as f:
|
||
cache_data = json.load(f)
|
||
|
||
for path in paths_to_clear:
|
||
if path in cache_data:
|
||
del cache_data[path]
|
||
cleared_count += 1
|
||
|
||
if cleared_count > 0:
|
||
with open(CACHE_FILE, "w", encoding="utf-8") as f:
|
||
json.dump(cache_data, f, ensure_ascii=False, indent=4)
|
||
|
||
return {"status": "success", "cleared_count": cleared_count}
|
||
|
||
except Exception as e:
|
||
return {"status": "error", "message": f"清除快取時發生錯誤: {str(e)}"}
|