349 lines
16 KiB
Python
349 lines
16 KiB
Python
# --- 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"}
|