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 logger import get_logger

logger = get_logger("app.scraper")

def parse_question_mark_output(output: str) -> dict:
    """解析 '?' 回傳內容，無差別掃描支援所有標題排列組合"""
    hint_lines = []
    options = []
    current_value = None
    format_desc = None
    is_format_only = False
    has_bracket_value = False
    
    if "dhcp-relay" in output.lower():
        is_format_only = True
        format_desc = "IP address or options"

    for line in output.splitlines():
        line = line.strip()
        
        # ==========================================
        # 🛡️ 1. 終極雜訊與 Echo 過濾器
        # ==========================================
        if not line or line.startswith('%') or line.startswith('^'):
            continue
            
        # 攔截 Echo 的問號指令 (例如 "logging buffered ?")
        if line.endswith('?'):
            continue
            
        # 攔截終端機 Prompt (例如 "admin@SERCOMM-COS-02(config)# ..." 或 "admin@SERCOMM-COS-02>")
        if re.search(r"[a-zA-Z0-9_.@-]+\(.*\)#", line) or re.search(r"^[a-zA-Z0-9_.@-]+>", line):
            continue
        # ==========================================

        # 2. 無差別收集所有有效行作為 Hint (保留最完整的說明給使用者看)
        hint_lines.append(line)

        # 3. 略過純標題行，不進行選項解析
        if line.startswith('Description:') or line.startswith('Possible completions:'):
            continue

        # 4. 解析格式與選項 (無差別掃描每一行)
        
        # 格式 A: <格式說明>[當前值] (例如: <string, min: 0 chars, max: 128 chars>[Cold Start])
        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
        
        # 格式 B: 型態, 最小值 .. 最大值 (例如: unsignedInt, 1 .. 3000)
        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
        
        # 格式 C: 純文字關鍵字
        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
        
        # 格式 D: 選項列表 [現值] 選項1 選項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()

        # 分離選項與說明 (完美支援水平列表，如 severity   size-mb)
        parts = re.split(r'\s{2,}', clean_line)
        valid_options = []
        for p in parts:
            p = p.strip()
            if not p: continue
            # 如果這個片段包含空白，代表它是說明文字 (Description)，停止解析後續片段
            if ' ' in p:
                break
            valid_options.append(p)
            
        for p in valid_options:
            if p not in ["|", ".."]:
                options.append(p)

    # 子命令防呆：如果抓到一堆單字，但沒有 [現值]，且不是已知格式，很可能是子命令列表
    if 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)

    return {
        "hint": "\n".join(hint_lines).strip(),
        "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:
    # 完美支援 (<string, min: 0 chars, max: 128 chars>) (Cold Start): 格式
    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()}

async def sync_cmts_leaves_async(host, username, password, leaf_paths: list):
    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):
            logger.info(f"🔄 正在處理第 {batch_idx + 1}/{len(batches)} 批次 (共 {len(batch)} 個路徑)...")
            
            try:
                async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as conn:
                    async with conn.create_process(term_type='xterm-256color', term_size=(200, 24), encoding='utf-8') as process:

                        async def read_until_quiet(timeout=1.0, prompt_pattern: str = None):
                            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()
                                    if prompt_pattern and re.search(prompt_pattern, output):
                                        break
                                except asyncio.TimeoutError:
                                    break
                            return output

                        # 🌟 關鍵修復 1：等待登入歡迎詞 (MOTD) 結束，確保設備準備好接收指令
                        await read_until_quiet(timeout=1.5, prompt_pattern=r"(?:#|>)")

                        if cmts_version == "unknown":
                            process.stdin.write("show version | nomore\n")
                            await process.stdin.drain()
                            version_output = await read_until_quiet(timeout=2.0, prompt_pattern=r"(?:#|>)")
                            match = re.search(r"(?:infra|vcmts-cd-0|CableOS)\s+([\w\.\-]+)", version_output, re.IGNORECASE)
                            if match:
                                cmts_version = match.group(1)
                            else:
                                cmts_version = "parse_failed" # 🌟 避免正則失敗導致每批次都重查

                        # 🌟 關鍵修復 2：確保成功進入 config 模式
                        process.stdin.write("config\n")
                        await process.stdin.drain()
                        await read_until_quiet(timeout=1.5, prompt_pattern=r"\(config.*\)#")

                        for path in batch:
                            await asyncio.sleep(0.01) 

                            command_to_send = f"{path} ?"
                            process.stdin.write(command_to_send)
                            await process.stdin.drain()
                            
                            output_question = await read_until_quiet(timeout=0.8)
                            q_data = parse_question_mark_output(output_question)
                            
                            # 🌟 頂級優雅解法：使用 Ctrl+U (\x15) 瞬間清空整行輸入緩衝區
                            process.stdin.write("\x15")
                            await process.stdin.drain()
                            await read_until_quiet(timeout=0.2)

                            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")
                            }

                            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=0.8)
                                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":
                                    # 如果進入了互動式輸入 (例如 prompt 變成 (val): )，按 Enter 接受預設值並退出
                                    process.stdin.write("\n") 
                                    await process.stdin.drain()
                                else:
                                    # 如果只是印出錯誤，我們不需要做什麼
                                    pass
                                    
                                await read_until_quiet(timeout=0.5)

                            result_data[path] = parsed_data
                            
                            processed_count += 1
                            yield json.dumps({
                                "event": "progress", 
                                "current": processed_count, 
                                "total": total_paths,
                                "current_path": path
                            }) + "\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 
            
            try:
                current_ts = int(time.time())
                formatted_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
                
                # 寫入資料庫
                if cmts_version not in ["unknown", "parse_failed"]:
                    await database.upsert_device_status(host, {"cmts_version": cmts_version, "last_scanned": formatted_time})
                else:
                    await database.upsert_device_status(host, {"last_scanned": formatted_time})
                    
                for p in batch:
                    if p in result_data:
                        result_data[p]["updated_at"] = current_ts
                        await database.upsert_leaf_option(host, p, result_data[p])
                        
                logger.info(f"💾 [DB] 第 {batch_idx + 1}/{len(batches)} 批次已寫入資料庫！")
            except Exception as e:
                logger.error(f"⚠️ 資料庫寫入失敗: {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"

async def fetch_raw_config(host: str, username: str, password: str, config_type: str = "running") -> str:
    try:
        async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as conn:
            async with conn.create_process(term_type='xterm-256color', term_size=(200, 24), encoding='utf-8') as process:

                async def read_until_quiet(timeout=2.0, prompt_pattern: str = None):
                    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()
                            if prompt_pattern and re.search(prompt_pattern, output):
                                break
                        except asyncio.TimeoutError:
                            break
                    return output

                await read_until_quiet(timeout=1.0, prompt_pattern=r"(?:#|>)")

                if config_type == "full":
                    process.stdin.write("config\n")
                    await process.stdin.drain()
                    await read_until_quiet(timeout=1.5, prompt_pattern=r"\(config\)#") 
                    
                    cmd = "show full-configuration | nomore"
                    process.stdin.write(f"{cmd}\n")
                    await process.stdin.drain()
                    raw_output = await read_until_quiet(timeout=3.0, prompt_pattern=r"\(config\)#")
                    
                    process.stdin.write("exit\n")
                    await process.stdin.drain()
                    
                else:
                    cmd = "show running-config | nomore"
                    process.stdin.write(f"{cmd}\n")
                    await process.stdin.drain()
                    raw_output = await read_until_quiet(timeout=3.0)
                
                process.stdin.write("exit\n")
                await process.stdin.drain()
                
                cleaned_output = re.sub(r'\x1b\[[0-9;?]*[a-zA-Z]|\x08', '', raw_output)
                cleaned_output = re.sub(r'\[7m\s*--More--\s*\[27m\[\d+D\[K', '', cleaned_output)
                cleaned_output = re.sub(r'\s*--More--\s*', '', cleaned_output)
                cleaned_output = re.sub(r'\s*\(END\)\s*', '', cleaned_output)
                
                lines = cleaned_output.splitlines()

                clean_lines = [
                    line for line in lines 
                    if not line.strip().startswith(cmd) 
                    and not line.strip().startswith("admin@")
                ]
                return "\n".join(clean_lines).strip()

    except Exception as e:
        raise Exception(f"SSH 連線或抓取設定失敗: {str(e)}")

def parse_config_to_tree(raw_cli: str) -> dict:
    return {"_raw_length": len(raw_cli), "status": "pending_parser"}