268 lines
11 KiB
Python
268 lines
11 KiB
Python
# --- cmts_scraper.py ---
|
||
import asyncio
|
||
import asyncssh
|
||
import re
|
||
import json
|
||
import os
|
||
import time
|
||
|
||
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()}
|
||
|
||
async def sync_cmts_leaves_async(host, username, password, leaf_paths: list) -> dict:
|
||
try:
|
||
result_data = {}
|
||
BATCH_SIZE = 30
|
||
batches = [leaf_paths[i:i + BATCH_SIZE] for i in range(0, len(leaf_paths), BATCH_SIZE)]
|
||
|
||
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
|
||
|
||
process.stdin.write("config\n")
|
||
await process.stdin.drain()
|
||
await read_until_quiet(timeout=1.5)
|
||
|
||
for path in batch:
|
||
# --- 步驟 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 ---
|
||
# 🌟 配合新規則：如果 '?' 已經明確告訴我們這是 <格式> 欄位，直接跳過 Enter！
|
||
if q_data.get("is_format_only"):
|
||
parsed_data["type"] = "string_or_number"
|
||
# if q_data.get("format_desc"):
|
||
# parsed_data["hint"] += f"\nFormat: {q_data['format_desc']}"
|
||
|
||
elif not q_data["options"]:
|
||
# 原本的 Enter 邏輯
|
||
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 "format_desc" in enter_data:
|
||
# parsed_data["hint"] += f"\nFormat: {enter_data['format_desc']}"
|
||
|
||
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
|
||
|
||
process.stdin.write("exit\n")
|
||
await process.stdin.drain()
|
||
await read_until_quiet(timeout=1.0)
|
||
|
||
except Exception as e:
|
||
print(f"❌ 第 {batch_idx + 1} 批次發生錯誤: {str(e)}")
|
||
continue
|
||
finally:
|
||
if conn: conn.close()
|
||
|
||
try:
|
||
cache_file = "cmts_leaf_options_cache.json"
|
||
cache_data = {}
|
||
if os.path.exists(cache_file):
|
||
with open(cache_file, "r", encoding="utf-8") as f:
|
||
cache_data = json.load(f)
|
||
|
||
current_ts = int(time.time())
|
||
for p in batch:
|
||
if p in result_data:
|
||
result_data[p]["updated_at"] = current_ts
|
||
cache_data[p] = result_data[p]
|
||
|
||
with open(cache_file, "w", encoding="utf-8") as f:
|
||
json.dump(cache_data, f, indent=4, ensure_ascii=False)
|
||
print(f"💾 [Debug] 第 {batch_idx + 1}/{len(batches)} 批次已提早寫入快取檔！")
|
||
except Exception as e:
|
||
print(f"⚠️ 提早寫入快取失敗: {e}")
|
||
|
||
if batch_idx < len(batches) - 1:
|
||
await asyncio.sleep(2)
|
||
|
||
return {"status": "success", "data": result_data}
|
||
|
||
except Exception as e:
|
||
return {"status": "error", "message": f"AsyncSSH 爬蟲錯誤: {str(e)}"}
|