# --- 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)} 個路徑)...") try: # 🧹 [穩定性修復] 全面改用 async with 管理生命週期，確保連線與 process 絕對釋放 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): 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 # --- 步驟 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" async def fetch_raw_config(host: str, username: str, password: str, config_type: str = "running") -> str: """連線至設備並抓取完整的純文字設定檔""" try: # 🧹 [穩定性修復] 全面改用 async with 管理生命週期 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): 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 # 🌟 新增這行：剛連線成功後，先清空終端機的登入歡迎詞 (MOTD) 與雜訊 await read_until_quiet(timeout=1.0) # 關鍵修改：根據 config_type 切換模式與指令，並加上 | nomore 關閉分頁 if config_type == "full": # 1. 先進入 config 模式 process.stdin.write("config\n") await process.stdin.drain() await read_until_quiet(timeout=1.5) # 等待提示字元變成 (config)# # 2. 下達 full-configuration 指令 (🌟 加上 | nomore) cmd = "show full-configuration | nomore" process.stdin.write(f"{cmd}\n") await process.stdin.drain() raw_output = await read_until_quiet(timeout=3.0) # 3. 抓完後退回上一層 (保持良好習慣) process.stdin.write("exit\n") await process.stdin.drain() else: # running-config 在一般模式即可下達 (🌟 加上 | nomore) 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() # 簡單清理頭尾的雜訊 (例如指令本身的 echo) # 🌟 關鍵修正 1：強化 ANSI 正規表達式，加入對 '?' 的支援 (精準捕捉 \x1b[?7h) cleaned_output = re.sub(r'\x1b\[[0-9;?]*[a-zA-Z]|\x08', '', raw_output) # 2. 清除失去 \x1b 殘留的字面分頁符號 (精準捕捉 [7m--More--[27m[8D[K) cleaned_output = re.sub(r'\[7m\s*--More--\s*\[27m\[\d+D\[K', '', cleaned_output) # 3. 清除純文字的 --More-- (防呆) cleaned_output = re.sub(r'\s*--More--\s*', '', cleaned_output) # 🌟 新增 4：清除 CableOS 終端機特有的 (END) 結尾標記 cleaned_output = re.sub(r'\s*\(END\)\s*', '', cleaned_output) # 關鍵修正：這裡改用 cleaned_output 來切行！ lines = cleaned_output.splitlines() # 🌟 關鍵修正 2：加入 strip() 避免空白干擾，確保精準踢掉提示字元 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)}") finally: if conn: conn.close() def parse_config_to_tree(raw_cli: str) -> dict: """將純文字設定檔轉換為簡單的階層式 JSON 樹狀圖 (Phase 3 會用到)""" # 這裡先實作一個基礎的縮排解析器，未來可依據您的設備格式優化 tree = {} # 暫時回傳空字典，確保 Phase 1 & 2 能順利走通 # 真正的樹狀解析邏輯我們可以在 Phase 3 完善 return {"_raw_length": len(raw_cli), "status": "pending_parser"}