scm-harmonic-cmts-admin/cmts_scraper.py

# --- cmts_scraper.py ---
import asyncio
import asyncssh
import re
import json
import os
import time
import database
from shared import USE_DB

def parse_question_mark_output(output: str) -> dict:
    """解析 '?' 回傳內容，支援無 Description、子命令判定與 dhcp-relay 混合欄位"""
    hint_lines = []
    options = []
    current_value = None
    format_desc = None
    is_format_only = False
    
    state = "INIT"
    has_description = False
    has_bracket_value = False
    
    # 🌟 Case-5 特殊處理：dhcp-relay 混合型欄位 (選項 + IP輸入)
    # 直接在開頭掃描完整輸出，若包含 dhcp-relay，強制轉為純文字輸入框
    if "dhcp-relay" in output.lower():
        is_format_only = True
        format_desc = "IP address or options"

    for line in output.splitlines():
        line = line.strip()
        # 過濾雜訊與終端機提示字元
        if not line or line.startswith('admin@') or line.startswith('cable') or line.startswith('%') or line.startswith('^'):
            continue
            
        if line.startswith('Description:'):
            state = "DESC"
            has_description = True
            hint_lines.append(line)
            continue
            
        if line.startswith('Possible completions:'):
            state = "COMPLETIONS"
            # 🌟 解除限制：無論有沒有 Description，都把這行加進 hint
            hint_lines.append(line)
            continue
            
        if state == "DESC":
            hint_lines.append(line)
            
        elif state == "COMPLETIONS":
            # Case 1~4: 無 Description 時，將後續選項說明也納入提示
            # 🌟 解除限制：無論有沒有 Description，都把這行加進 hint
            hint_lines.append(line)
                
            # 規則 1: <格式說明>[當前值]
            match_format = re.search(r"<([^>]+)>\s*(?:\[([^\]]+)\])?", line)
            if match_format:
                is_format_only = True
                format_desc = match_format.group(1).strip()
                if match_format.group(2):
                    current_value = match_format.group(2).strip()
                continue
            
            # 規則 1.5: 型態, 最小值 .. 最大值
            match_range = re.search(r"^\s*([a-zA-Z0-9_]+),\s*(\d+)\s*\.\.\s*(\d+)\s*$", line)
            if match_range:
                is_format_only = True
                format_desc = f"{match_range.group(1)} ({match_range.group(2)} .. {match_range.group(3)})"
                continue
            
            # 規則 1.6: IP address 等純文字關鍵字
            match_keyword = re.search(r"^(IP address|IPv4 address|IPv6 address|MAC address)$", line, re.IGNORECASE)
            if match_keyword:
                is_format_only = True
                format_desc = match_keyword.group(1)
                continue
            
            # 規則 2: 選項列表處理
            match_current = re.search(r"^\[([^\]]+)\]", line)
            if match_current:
                has_bracket_value = True # 標記：這是一個帶有現值的標準選項清單
                if not current_value:
                    current_value = match_current.group(1).strip()
                clean_line = re.sub(r"^\[[^\]]+\]", "", line).strip()
            else:
                clean_line = line.strip()

            # 🌟 Case-6 防呆：過濾垂直列表的說明文字
            # 利用「3 個以上的連續空白」作為選項與說明文字的分界線
            if re.search(r"\s{3,}", clean_line):
                clean_line = re.split(r"\s{3,}", clean_line)[0]
                
            parts = clean_line.split()
            for p in parts:
                if p and p not in ["|", ".."]:
                    options.append(p)

    # 🌟 Case 2, 3, 4: 子命令防呆機制
    # 若在選項區塊從未發現 [現值]，且非已知格式，判定為子命令，強制轉純文字
    if state == "COMPLETIONS" and not has_bracket_value and not is_format_only and options:
        is_format_only = True
        options = []

    # 確保現值一定包含在選項中（如果它是一般的下拉選單）
    if current_value and options and current_value not in options:
        options.append(current_value)

    hint_text = "\n".join(hint_lines).strip()
    
    return {
        "hint": hint_text,
        # 若為純文字模式，強制回傳空陣列，確保前端正確渲染為 input
        "options": list(dict.fromkeys(options)) if not is_format_only else [],
        "current_value": current_value,
        "format_desc": format_desc,
        "is_format_only": is_format_only
    }

def parse_device_response(output: str) -> dict:
    """支援多種編輯狀態格式解析"""
    match = re.search(r"(?:\[(.*?)\]|\(<(.*?)>\))\s*\((.*?)\):", output)
    
    if match:
        enum_content = match.group(1)
        desc_content = match.group(2)
        current_value = match.group(3).strip()
        
        if enum_content:
            if enum_content.strip().lower() == "list":
                return {"type": "list", "options": [], "current_value": current_value}
            else:
                return {
                    "type": "enum",
                    "options": [opt.strip() for opt in enum_content.split(",") if opt.strip()],
                    "current_value": current_value
                }
        elif desc_content:
            return {
                "type": "string_or_number",
                "options": [], 
                "current_value": current_value,
                "format_desc": desc_content.strip()
            }

    return {"type": "unknown", "options": [], "current_value": None, "raw_output": output.strip()}

# 🌟 1. 函數簽名加上 config_type 參數，預設為 "running"
async def sync_cmts_leaves_async(host, username, password, leaf_paths: list, config_type: str = "running"):
    try:
        total_paths = len(leaf_paths)
        processed_count = 0
        result_data = {}
        BATCH_SIZE = 30  
        batches = [leaf_paths[i:i + BATCH_SIZE] for i in range(0, len(leaf_paths), BATCH_SIZE)]
        
        cmts_version = "unknown" # 🌟 新增：用來記錄當前爬蟲抓到的版本
        
        for batch_idx, batch in enumerate(batches):
            print(f"🔄 正在處理第 {batch_idx + 1}/{len(batches)} 批次 (共 {len(batch)} 個路徑)...")
            conn = None
            
            try:
                conn = await asyncssh.connect(host, username=username, password=password, known_hosts=None)
                process = await conn.create_process(term_type='xterm-256color', term_size=(200, 24), encoding='utf-8')

                async def read_until_quiet(timeout=1.0):
                    output = ""
                    while True:
                        try:
                            chunk = await asyncio.wait_for(process.stdout.read(4096), timeout=timeout)
                            if not chunk: break
                            output += chunk
                            if "--More--" in chunk or "More" in chunk:
                                process.stdin.write(" ")
                                await process.stdin.drain()
                        except asyncio.TimeoutError:
                            break
                    return output

                # 🌟 新增：在第一批次連線時，先抓取設備版本
                if cmts_version == "unknown":
                    process.stdin.write("show version\n")
                    await process.stdin.drain()
                    version_output = await read_until_quiet(timeout=1.5)
                    match = re.search(r"(?:infra|vcmts-cd-0)\s+([\w\.\-]+)", version_output)
                    if match:
                        cmts_version = match.group(1)

                process.stdin.write("config\n")
                await process.stdin.drain()
                await read_until_quiet(timeout=1.5)

                for path in batch:
                    # 🌟 優化 1：強迫讓出事件迴圈控制權，讓 FastAPI 去處理其他使用者的請求
                    await asyncio.sleep(0.01) 

                    # --- 步驟 1：發送 '?' 查詢 ---
                    command_to_send = f"{path} ?"
                    process.stdin.write(command_to_send)
                    await process.stdin.drain()
                    
                    output_question = await read_until_quiet(timeout=1.0)
                    q_data = parse_question_mark_output(output_question)
                    
                    backspaces = "\x08" * (len(command_to_send) + 5)
                    process.stdin.write(backspaces)
                    await process.stdin.drain()
                    await read_until_quiet(timeout=0.5)

                    parsed_data = {
                        "hint": q_data["hint"],
                        "options": q_data["options"],
                        "type": "list" if q_data["options"] else "unknown",
                        "current_value": q_data.get("current_value")
                    }

                    # --- 步驟 2：判斷是否需要發送 Enter ---
                    if q_data.get("is_format_only"):
                        parsed_data["type"] = "string_or_number"
                            
                    elif not q_data["options"]:
                        process.stdin.write(f"{path}\n")
                        await process.stdin.drain()
                        
                        output_enter = await read_until_quiet(timeout=1.0)
                        enter_data = parse_device_response(output_enter)
                        
                        parsed_data["type"] = enter_data["type"]
                        parsed_data["options"] = enter_data["options"]
                        parsed_data["current_value"] = enter_data["current_value"]
                        
                        if enter_data["type"] != "unknown":
                            process.stdin.write("\x03") 
                            await process.stdin.drain()
                        else:
                            print(f"⚠️ [Debug] 未知格式 ({path}):\n{enter_data['raw_output']}")
                            process.stdin.write("\n")
                            await process.stdin.drain()
                            
                        await read_until_quiet(timeout=0.5)

                    result_data[path] = parsed_data
                    
                    processed_count += 1
                    event_data = {
                        "event": "progress", 
                        "current": processed_count, 
                        "total": total_paths,
                        "current_path": path
                    }
                    yield json.dumps(event_data) + "\n"

                process.stdin.write("exit\n")
                await process.stdin.drain()
                await read_until_quiet(timeout=1.0)

            except Exception as e:
                yield json.dumps({"event": "error", "message": f"批次錯誤: {str(e)}"}) + "\n"
                continue 
            finally:
                if conn: conn.close()
            
                        # --- 步驟 3：寫入快取 (DB or JSON) ---
            try:
                db_success = False
                current_ts = int(time.time())
                formatted_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
                
                # 1. 嘗試寫入資料庫
                if USE_DB:
                    try:
                        # 寫入 metadata
                        if cmts_version != "unknown":
                            await database.upsert_device_status(host, config_type, {"cmts_version": cmts_version, "last_scanned": formatted_time})
                        else:
                            await database.upsert_device_status(host, config_type, {"last_scanned": formatted_time})
                            
                        # 寫入 options
                        db_write_count = 0
                        for p in batch:
                            if p in result_data:
                                result_data[p]["updated_at"] = current_ts
                                success = await database.upsert_leaf_option(host, config_type, p, result_data[p])
                                if success:
                                    db_write_count += 1
                                    
                        if db_write_count > 0:
                            db_success = True
                            print(f"💾 [DB] 第 {batch_idx + 1}/{len(batches)} 批次已非同步寫入資料庫！")
                        else:
                            print("⚠️ [Fallback] 資料庫寫入 0 筆，退回寫入 JSON 快取...")
                    except Exception as e:
                        print(f"⚠️ [Fallback] 資料庫寫入發生例外: {e}，自動切換至 JSON 快取寫入...")

                # 2. 如果 DB 寫入失敗或 USE_DB=False，則 Fallback 寫入 JSON
                if not db_success:
                    # 🌟 動態決定快取檔名 (加入 IP 隔離)
                    safe_host = host.replace(".", "_") 
                    cache_file = f"{safe_host}_{config_type}_cache.json"
                    
                    def read_json_from_file(filepath):
                        if os.path.exists(filepath):
                            with open(filepath, "r", encoding="utf-8") as f:
                                return json.load(f)
                        return {}
                        
                    cache_data = await asyncio.to_thread(read_json_from_file, cache_file)
                    
                    # 🌟 新增：寫入 Metadata (版本與掃描時間)
                    if "__metadata__" not in cache_data:
                        cache_data["__metadata__"] = {}
                    
                    if cmts_version != "unknown":
                        cache_data["__metadata__"]["cmts_version"] = cmts_version
                        
                    cache_data["__metadata__"]["last_scanned"] = formatted_time
                    
                    for p in batch:
                        if p in result_data:
                            result_data[p]["updated_at"] = current_ts
                            cache_data[p] = result_data[p]
                            
                    def write_json_to_file(filepath, data):
                        with open(filepath, "w", encoding="utf-8") as f:
                            json.dump(data, f, indent=4, ensure_ascii=False)
                    
                    await asyncio.to_thread(write_json_to_file, cache_file, cache_data)
                    print(f"💾 [JSON] 第 {batch_idx + 1}/{len(batches)} 批次已非同步寫入快取檔！(版本: {cmts_version}, 檔案: {cache_file})")

            except Exception as e:
                print(f"⚠️ 寫入快取失敗 (DB 與 JSON 皆失敗): {e}")
                    
            if batch_idx < len(batches) - 1:
                await asyncio.sleep(2)

        yield json.dumps({"event": "done", "message": "快取更新完成"}) + "\n"

    except Exception as e:
        yield json.dumps({"event": "error", "message": f"爬蟲嚴重錯誤: {str(e)}"}) + "\n"