scm-harmonic-cmts-admin/routers/leaf_options.py

# --- 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, asyncio, re
import database
from cmts_scraper import sync_cmts_leaves_async
from shared import CMTS_DEVICE

router = APIRouter(tags=["Options"])

SCAN_STATUS = {}
active_clients = {}

async def broadcast_message(message_dict: dict, host: str):
    channel_key = host
    dead_clients = set()
    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):
    channel_key = host
    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:
                    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:
            if channel_key in active_clients:
                active_clients[channel_key].discard(client_queue)
                # ✅ 安全升級：如果該設備已經沒有任何客戶端監聽，徹底刪除 Key 釋放記憶體
                if not active_clients[channel_key]:
                    del active_clients[channel_key]
            
    return StreamingResponse(event_generator(), media_type="text/event-stream")

@router.get("/scan-status")
async def get_scan_status(host: str):
    return {"is_scanning": SCAN_STATUS.get(host, False)}

class SyncOptionsRequest(BaseModel):
    host: str 
    username: str = "" 
    password: str = "" 
    leaf_paths: List[str]

@router.get("/cmts-leaf-options")
async def get_leaf_options(host: str): 
    try:
        db_options = await database.get_all_leaf_options(host)
        if db_options is not None:
            metadata = await database.get_device_status(host)
            if metadata:
                db_options["__metadata__"] = metadata
            return db_options
        raise HTTPException(status_code=500, detail="資料庫連線異常")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"資料庫讀取失敗: {e}")

async def run_scan_task(host: str, username: str, password: str, paths: List[str]):
    global SCAN_STATUS
    channel_key = host
    try:
        await broadcast_message({"event": "scan_start"}, host)
        async for chunk in sync_cmts_leaves_async(host=host, username=username, password=password, leaf_paths=paths):
            if isinstance(chunk, str):
                await broadcast_message(json.loads(chunk), host)
            else:
                await broadcast_message(chunk, host)
        await broadcast_message({"event": "done"}, host)
    except Exception as e:
        await broadcast_message({"event": "error", "message": str(e)}, host)
    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 = request.host
    if SCAN_STATUS.get(channel_key, False):
        return {"status": "busy", "message": "⚠️ 該設備的掃描任務正在執行中，請勿重複點擊！"}
    
    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))
    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)
    return {"status": "started"}

@router.post("/clear_cache")
async def clear_specific_cache(host: str, paths_to_clear: list = Body(...)): 
    try:
        cleared_count = await database.delete_leaf_options(host, paths_to_clear)
        if cleared_count >= 0:
            return {"status": "success", "cleared_count": cleared_count}
        raise HTTPException(status_code=500, detail="資料庫清除失敗")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"資料庫清除失敗: {str(e)}")