# --- 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: <格式說明>[當前值] (例如: [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: # 完美支援 () (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"}