660 lines
29 KiB
Python
660 lines
29 KiB
Python
import os
|
||
import json
|
||
import argparse
|
||
import subprocess
|
||
import time
|
||
import socket
|
||
import re
|
||
from pathlib import Path
|
||
from urllib.parse import urljoin, urlparse
|
||
from typing import List, Dict, Any, Optional
|
||
|
||
from selenium import webdriver
|
||
from selenium.webdriver.edge.options import Options
|
||
from selenium.webdriver.edge.service import Service
|
||
from selenium.webdriver.common.by import By
|
||
from selenium.webdriver.support.ui import WebDriverWait
|
||
from selenium.webdriver.support import expected_conditions as EC
|
||
from webdriver_manager.microsoft import EdgeChromiumDriverManager
|
||
|
||
|
||
# ---------------------------- 工具函数 ----------------------------
|
||
def get_absolute_url(base_url: str, href: str) -> Optional[str]:
|
||
if not href:
|
||
return None
|
||
href = href.strip()
|
||
if href.startswith(('javascript:', 'mailto:', 'tel:', '#')):
|
||
return None
|
||
absolute = urljoin(base_url, href)
|
||
parsed = urlparse(absolute)
|
||
if parsed.scheme in ('http', 'https', 'file'):
|
||
return absolute
|
||
return None
|
||
|
||
|
||
def sanitize_filename(url: str) -> str:
|
||
"""根据 URL 生成合法的文件名（不含扩展名）"""
|
||
parsed = urlparse(url)
|
||
path = parsed.path.strip('/')
|
||
if not path:
|
||
path = "index"
|
||
safe = re.sub(r'[\\/*?:"<>|]', '_', path)
|
||
if len(safe) > 100:
|
||
safe = safe[:100]
|
||
return safe
|
||
|
||
|
||
# ---------------------------- 语义上下文推断（加强版） ----------------------------
|
||
def infer_context_from_attrs(tag_name: str, attrs: Dict[str, str], text: str = "") -> str:
|
||
class_list = attrs.get("class", "").split()
|
||
id_ = attrs.get("id", "")
|
||
href = attrs.get("href", "")
|
||
features = " ".join(class_list) + " " + id_
|
||
|
||
# 对 <a> 标签的判断
|
||
if tag_name == "a":
|
||
if any(w in features for w in ["btn", "button", "tab", "nav-btn", "menu-item", "action-btn"]):
|
||
return "按钮"
|
||
|
||
# 通用特征判断
|
||
if "search" in features or "nav-search" in features:
|
||
return "搜索框"
|
||
if any(w in features for w in ["header", "top-bar", "masthead", "nav"]):
|
||
return "顶部导航"
|
||
if any(w in features for w in ["sidebar", "aside", "right-entry"]):
|
||
return "侧边栏"
|
||
if "footer" in features:
|
||
return "页脚"
|
||
if any(w in features for w in ["channel", "partition", "channel-icons"]):
|
||
return "频道/分区"
|
||
if any(w in features for w in ["carousel", "banner", "slider"]):
|
||
return "轮播/横幅"
|
||
if any(w in features for w in ["video-card", "feed-card", "card-item", "card"]):
|
||
return "视频卡片"
|
||
if "main" in features or "content" in features:
|
||
return "主要内容区"
|
||
if tag_name == "a":
|
||
if "cm.bilibili.com" in href or "ad" in features:
|
||
return "广告链接"
|
||
return "普通链接"
|
||
if tag_name in ("input", "textarea"):
|
||
return "输入控件"
|
||
if tag_name == "button":
|
||
return "按钮"
|
||
if tag_name == "select":
|
||
return "下拉选择"
|
||
# 对新加入的交互元素（div/li/span/img等）做一些推断
|
||
if tag_name in ("div", "li", "span", "img", "td", "label"):
|
||
if any(w in features for w in ["btn", "button", "action", "clickable", "tab", "menu-item"]):
|
||
return "按钮"
|
||
if any(w in features for w in ["card", "item", "cell", "list-item"]):
|
||
return "卡片/列表项"
|
||
if "dropdown" in features or "menu" in features:
|
||
return "菜单"
|
||
if "dialog" in features or "modal" in features:
|
||
return "弹窗"
|
||
if "icon" in features:
|
||
return "图标"
|
||
if tag_name == "img":
|
||
return "图片"
|
||
return "交互区域"
|
||
return "其他"
|
||
|
||
|
||
# ---------------------------- 事件绑定与递归（增强版：模拟点击所有可交互控件） ----------------------------
|
||
def add_events_and_recurse(controls: List[Dict[str, Any]], base_url: str,
|
||
depth: int, max_depth: int, visited: set, driver):
|
||
"""
|
||
增强版：对于点击后 URL 不变的控件，会检测页面中动态出现的新控件，
|
||
并对新控件（如下拉菜单项）进行递归导航抓取。
|
||
"""
|
||
def get_control_signature(ctrl):
|
||
"""生成控件的唯一签名"""
|
||
tp = ctrl["type"]
|
||
attrs = ctrl["attributes"]
|
||
text = ctrl.get("text", "")
|
||
cls = attrs.get("class", "")
|
||
id_ = attrs.get("id", "")
|
||
return f"{tp}|{text}|{cls}|{id_}"
|
||
|
||
original_signatures = {get_control_signature(c) for c in controls}
|
||
|
||
for ctrl in controls:
|
||
tp = ctrl["type"]
|
||
attrs = ctrl["attributes"]
|
||
idx = ctrl.get("index")
|
||
events = []
|
||
navigate_url = None
|
||
reason = ""
|
||
clicked_for_navigation = False
|
||
|
||
# ---------- URL 预判 ----------
|
||
if tp == "a":
|
||
href = attrs.get("href")
|
||
if href:
|
||
abs_url = get_absolute_url(base_url, href)
|
||
if abs_url:
|
||
navigate_url = abs_url
|
||
reason = "点击链接跳转"
|
||
|
||
if tp == "button" or (tp == "input" and attrs.get("type", "").lower() in ("submit", "button", "image")):
|
||
form_action = ctrl.get("form_action")
|
||
if form_action:
|
||
abs_action = get_absolute_url(base_url, form_action)
|
||
if abs_action:
|
||
navigate_url = abs_action
|
||
reason = "提交表单跳转"
|
||
else:
|
||
events.append({"type": "submit", "description": "提交表单（无明确跳转）"})
|
||
else:
|
||
events.append({"type": "submit", "description": "提交表单（无 action）"})
|
||
|
||
if "alert" in attrs.get("onclick", "") or "confirm" in attrs.get("onclick", ""):
|
||
events.append({"type": "popup", "description": "触发弹窗"})
|
||
|
||
if tp == "input":
|
||
input_type = attrs.get("type", "text").lower()
|
||
if input_type in ("text", "password", "email", "search", "number", "tel", "url", ""):
|
||
events.append({"type": "input", "description": "输入文字"})
|
||
events.append({"type": "change", "description": "内容改变"})
|
||
elif input_type == "range":
|
||
events.append({"type": "input", "description": "滑动改变值"})
|
||
elif input_type in ("checkbox", "radio"):
|
||
events.append({"type": "change", "description": "切换状态"})
|
||
|
||
if tp == "select":
|
||
events.append({"type": "change", "description": "选择项改变"})
|
||
|
||
if tp == "textarea":
|
||
events.append({"type": "input", "description": "输入多行文字"})
|
||
events.append({"type": "change", "description": "内容改变"})
|
||
|
||
if tp in ("div", "li", "span", "img", "td", "label") and ctrl.get("is_interactive"):
|
||
onclick = attrs.get("onclick", "")
|
||
data_href = attrs.get("data-href", "")
|
||
if data_href:
|
||
abs_url = get_absolute_url(base_url, data_href)
|
||
if abs_url:
|
||
navigate_url = abs_url
|
||
reason = "点击交互跳转"
|
||
elif "location.href" in onclick or "window.open" in onclick:
|
||
match = re.search(r"""['"]([^'"]+)['"]""", onclick)
|
||
if match:
|
||
abs_url = get_absolute_url(base_url, match.group(1))
|
||
if abs_url:
|
||
navigate_url = abs_url
|
||
reason = "点击交互跳转"
|
||
|
||
# 弹窗兜底
|
||
if tp in ("input", "button") or (tp in ("div", "li", "span", "img") and ctrl.get("is_interactive")):
|
||
if ("alert" in attrs.get("onclick", "") or "confirm" in attrs.get("onclick", "")) and \
|
||
not any(e["type"] == "popup" for e in events):
|
||
events.append({"type": "popup", "description": "触发弹窗"})
|
||
|
||
# ========== 模拟点击与动态内容探索 ==========
|
||
if depth < max_depth and idx is not None:
|
||
# 情况1：有预知 URL
|
||
if navigate_url and navigate_url not in visited:
|
||
try:
|
||
target_element = driver.find_element(By.CSS_SELECTOR, f'[data-ctrl-idx="{idx}"]')
|
||
pre_url = driver.current_url
|
||
target_element.click()
|
||
time.sleep(1)
|
||
WebDriverWait(driver, 10).until(
|
||
lambda d: d.execute_script("return document.readyState") == "complete"
|
||
)
|
||
|
||
handles = driver.window_handles
|
||
if len(handles) > 1:
|
||
original_handle = driver.current_window_handle
|
||
for handle in handles:
|
||
if handle != original_handle:
|
||
driver.switch_to.window(handle)
|
||
break
|
||
new_url = driver.current_url
|
||
if new_url not in visited:
|
||
visited.add(new_url)
|
||
target_page = process_source(new_url, depth + 1, max_depth, visited, driver)
|
||
event = {"type": "navigate", "url": new_url, "description": reason,
|
||
"target_page": target_page}
|
||
events.append(event)
|
||
driver.close()
|
||
driver.switch_to.window(original_handle)
|
||
else:
|
||
new_url = driver.current_url
|
||
if new_url != pre_url and new_url not in visited:
|
||
visited.add(new_url)
|
||
target_page = process_source(new_url, depth + 1, max_depth, visited, driver)
|
||
event = {"type": "navigate", "url": new_url, "description": reason,
|
||
"target_page": target_page}
|
||
events.append(event)
|
||
driver.back()
|
||
WebDriverWait(driver, 10).until(
|
||
lambda d: d.execute_script("return document.readyState") == "complete"
|
||
)
|
||
clicked_for_navigation = True
|
||
except Exception as e:
|
||
events.append({"type": "navigate", "url": navigate_url, "description": reason,
|
||
"target_page": {"url": navigate_url, "error": str(e)}})
|
||
clicked_for_navigation = True
|
||
|
||
# 情况2：无预知 URL，但可交互 → 尝试点击并检查是否出现下拉菜单
|
||
elif ctrl.get("is_interactive") and not clicked_for_navigation:
|
||
try:
|
||
target_element = driver.find_element(By.CSS_SELECTOR, f'[data-ctrl-idx="{idx}"]')
|
||
pre_url = driver.current_url
|
||
target_element.click()
|
||
time.sleep(0.8)
|
||
WebDriverWait(driver, 10).until(
|
||
lambda d: d.execute_script("return document.readyState") == "complete"
|
||
)
|
||
|
||
new_url = driver.current_url
|
||
if new_url != pre_url and new_url not in visited:
|
||
# 直接跳转了
|
||
visited.add(new_url)
|
||
target_page = process_source(new_url, depth + 1, max_depth, visited, driver)
|
||
event = {"type": "navigate", "url": new_url, "description": "点击交互跳转（运行时发现）",
|
||
"target_page": target_page}
|
||
events.append(event)
|
||
driver.back()
|
||
WebDriverWait(driver, 10).until(
|
||
lambda d: d.execute_script("return document.readyState") == "complete"
|
||
)
|
||
clicked_for_navigation = True
|
||
else:
|
||
# URL 没变，检测动态出现的下拉菜单项
|
||
# 使用更可靠的方法：直接查找可能是下拉项的 <li> 或 <a>，并排除原始控件
|
||
sub_items = driver.execute_script("""
|
||
let candidates = [];
|
||
// 查找常见菜单结构：ul.submenu, div.dropdown-menu 下的 li/a
|
||
let menus = document.querySelectorAll('ul, div[class*="menu"], div[class*="dropdown"]');
|
||
for (let menu of menus) {
|
||
let items = menu.querySelectorAll('li, a');
|
||
for (let item of items) {
|
||
// 跳过隐藏元素
|
||
if (item.offsetParent === null) continue;
|
||
let text = item.innerText.trim().substring(0, 50);
|
||
if (!text) continue;
|
||
// 排除与原始控件文本相同的元素（父菜单本身可能被包含）
|
||
candidates.push({
|
||
tag: item.tagName.toLowerCase(),
|
||
text: text,
|
||
outerHTML: item.outerHTML.substring(0, 200)
|
||
});
|
||
}
|
||
}
|
||
return candidates;
|
||
""")
|
||
|
||
# 去重，只保留之前没有的项
|
||
clicked_sub_items = set()
|
||
for item in sub_items:
|
||
sig = f"{item['tag']}|{item['text']}"
|
||
if sig not in original_signatures and sig not in clicked_sub_items:
|
||
# 尝试定位并点击
|
||
try:
|
||
# 用文本查找具体元素（更稳定）
|
||
elem = driver.find_element(By.XPATH,
|
||
f"//{item['tag']}[normalize-space()='{item['text']}']")
|
||
elem.click()
|
||
time.sleep(0.8)
|
||
WebDriverWait(driver, 10).until(
|
||
lambda d: d.execute_script("return document.readyState") == "complete"
|
||
)
|
||
new_url2 = driver.current_url
|
||
if new_url2 != pre_url and new_url2 not in visited:
|
||
visited.add(new_url2)
|
||
sub_target = process_source(new_url2, depth + 1, max_depth, visited, driver)
|
||
event = {"type": "navigate", "url": new_url2,
|
||
"description": f"下拉菜单项: {item['text']}",
|
||
"target_page": sub_target}
|
||
events.append(event)
|
||
driver.back()
|
||
WebDriverWait(driver, 10).until(
|
||
lambda d: d.execute_script("return document.readyState") == "complete"
|
||
)
|
||
# 回退后重新点击父菜单，确保下拉菜单仍然展开
|
||
target_element = driver.find_element(By.CSS_SELECTOR, f'[data-ctrl-idx="{idx}"]')
|
||
target_element.click()
|
||
time.sleep(0.5)
|
||
clicked_sub_items.add(sig)
|
||
except Exception:
|
||
continue
|
||
except Exception as e:
|
||
# 可以在这里打印调试信息，但不影响流程
|
||
pass
|
||
|
||
if not events:
|
||
events.append({"type": "unknown", "description": "未识别的事件"})
|
||
|
||
ctrl["events"] = events
|
||
|
||
|
||
# ---------------------------- 核心抓取逻辑（增强版：去重+注入索引） ----------------------------
|
||
def build_page_tree_from_url(driver, url: str, depth: int, max_depth: int,
|
||
visited: set) -> Dict[str, Any]:
|
||
print(f"正在抓取 (深度 {depth}/{max_depth}): {url}")
|
||
visited.add(url)
|
||
|
||
try:
|
||
driver.get(url)
|
||
WebDriverWait(driver, 15).until(
|
||
lambda d: d.execute_script("return document.readyState") == "complete"
|
||
)
|
||
driver.implicitly_wait(2)
|
||
except Exception as e:
|
||
return {"url": url, "error": str(e)}
|
||
|
||
# ---------- 提取所有交互元素（去重：只保留最外层交互元素，并注入索引） ----------
|
||
try:
|
||
raw_controls = driver.execute_script("""
|
||
const tags = ['a', 'input', 'button', 'select', 'textarea',
|
||
'div', 'li', 'span', 'img', 'td', 'label'];
|
||
let allElements = [];
|
||
for (let tag of tags) {
|
||
let elems = document.querySelectorAll(tag);
|
||
for (let elem of elems) {
|
||
if (tag === 'input' && elem.type === 'hidden') continue;
|
||
if (tag === 'a' && !elem.href && !elem.onclick) continue;
|
||
|
||
let isInteractive = false;
|
||
if (['div', 'li', 'span', 'img', 'td', 'label'].includes(tag)) {
|
||
if (elem.onclick) isInteractive = true;
|
||
let role = elem.getAttribute('role');
|
||
if (role && ['button', 'link', 'tab', 'menuitem'].includes(role)) isInteractive = true;
|
||
if (elem.hasAttribute('tabindex') && elem.getAttribute('tabindex') !== '-1') isInteractive = true;
|
||
if (elem.hasAttribute('data-action') || elem.hasAttribute('data-href')) isInteractive = true;
|
||
let cls = elem.className || '';
|
||
if (/(btn|button|clickable|action|link|tab|menu-item|card|cell)/i.test(cls)) {
|
||
isInteractive = true;
|
||
}
|
||
if (!isInteractive) continue;
|
||
}
|
||
|
||
allElements.push({element: elem, tag: tag, isInteractive: isInteractive});
|
||
}
|
||
}
|
||
|
||
// 去重：如果某元素被另一个更外层的交互元素包含，则丢弃它
|
||
let filtered = [];
|
||
let idx = 0;
|
||
for (let item of allElements) {
|
||
let elem = item.element;
|
||
let isChild = false;
|
||
// 检查是否被之前的某个元素包含
|
||
for (let prev of filtered) {
|
||
if (prev.element.contains(elem) && prev.element !== elem) {
|
||
isChild = true;
|
||
break;
|
||
}
|
||
}
|
||
if (!isChild) {
|
||
// 注入唯一索引
|
||
elem.setAttribute('data-ctrl-idx', idx);
|
||
let attrs = {};
|
||
for (let attr of elem.attributes) {
|
||
attrs[attr.name] = attr.value;
|
||
}
|
||
let formAction = null;
|
||
if (item.tag === 'button' || (item.tag === 'input' && ['submit','button','image'].includes(elem.type))) {
|
||
let form = elem.closest('form');
|
||
if (form) formAction = form.getAttribute('action') || '';
|
||
}
|
||
let text = elem.innerText ? elem.innerText.trim().substring(0, 50) : '';
|
||
filtered.push({
|
||
element: elem, // 用于后续 contains 检查
|
||
tag: item.tag,
|
||
attributes: attrs,
|
||
text: text,
|
||
form_action: formAction,
|
||
is_interactive: item.isInteractive,
|
||
index: idx
|
||
});
|
||
idx++;
|
||
}
|
||
}
|
||
// 只返回需要的数据（去掉 DOM 引用）
|
||
return filtered.map(({tag, attributes, text, form_action, is_interactive, index}) => ({
|
||
tag, attributes, text, form_action, is_interactive, index
|
||
}));
|
||
""")
|
||
except Exception as e:
|
||
return {"url": url, "error": f"JavaScript提取失败: {str(e)}"}
|
||
|
||
# 转换为统一格式
|
||
controls = []
|
||
for rc in raw_controls:
|
||
ctrl = {
|
||
"type": rc["tag"],
|
||
"attributes": rc["attributes"],
|
||
"text": rc.get("text", ""),
|
||
"context": infer_context_from_attrs(rc["tag"], rc["attributes"], rc.get("text", "")),
|
||
"events": [],
|
||
"form_action": rc.get("form_action"),
|
||
"is_interactive": rc.get("is_interactive", False),
|
||
"index": rc.get("index") # 保存索引，用于模拟点击
|
||
}
|
||
controls.append(ctrl)
|
||
|
||
# 填充事件和递归（将 driver 传入，实现模拟点击）
|
||
add_events_and_recurse(controls, url, depth, max_depth, visited, driver)
|
||
|
||
# 删除内部使用的字段
|
||
for ctrl in controls:
|
||
ctrl.pop("form_action", None)
|
||
if not ctrl["is_interactive"]:
|
||
del ctrl["is_interactive"]
|
||
|
||
return {"url": url, "controls": controls}
|
||
|
||
|
||
def process_source(source: str, depth: int, max_depth: int,
|
||
visited: set, driver) -> Dict[str, Any]:
|
||
if source.startswith(('http://', 'https://', 'file://')):
|
||
url = source
|
||
else:
|
||
abs_path = os.path.abspath(source)
|
||
url = Path(abs_path).as_uri()
|
||
return build_page_tree_from_url(driver, url, depth, max_depth, visited)
|
||
|
||
|
||
# ---------------------------- 调试浏览器管理 ----------------------------
|
||
def is_port_open(host, port):
|
||
try:
|
||
with socket.create_connection((host, port), timeout=1):
|
||
return True
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
def ensure_debug_browser():
|
||
debug_host = "127.0.0.1"
|
||
debug_port = 9222
|
||
|
||
if not is_port_open(debug_host, debug_port):
|
||
print("未检测到调试浏览器，正在启动 Edge 调试窗口...")
|
||
edge_path = r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe"
|
||
user_data_dir = r"C:\temp\edge_debug_profile"
|
||
try:
|
||
subprocess.Popen([
|
||
edge_path,
|
||
f"--remote-debugging-port={debug_port}",
|
||
f"--user-data-dir={user_data_dir}"
|
||
], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
except FileNotFoundError:
|
||
edge_path = "msedge"
|
||
subprocess.Popen([
|
||
edge_path,
|
||
f"--remote-debugging-port={debug_port}",
|
||
f"--user-data-dir={user_data_dir}"
|
||
], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
print("请在新打开的 Edge 窗口中访问目标网站并完成登录，然后回到这里按 Enter 继续...")
|
||
input()
|
||
else:
|
||
print("检测到已有的调试浏览器，将直接使用。")
|
||
|
||
if not is_port_open(debug_host, debug_port):
|
||
raise RuntimeError("无法连接到调试端口，请检查 Edge 是否正常启动。")
|
||
|
||
|
||
# ---------------------------- 人性化报告生成 ----------------------------
|
||
def short_text_for_control(ctrl: Dict[str, Any]) -> str:
|
||
tp = ctrl["type"]
|
||
attrs = ctrl["attributes"]
|
||
is_interactive = ctrl.get("is_interactive", False)
|
||
base = ""
|
||
|
||
if tp == "a":
|
||
href = attrs.get("href", "")
|
||
text = ctrl.get("text") or attrs.get("title") or attrs.get("aria-label") or href[:60]
|
||
base = f"链接 →{text} : {href}"
|
||
elif tp == "button":
|
||
text = ctrl.get("text") or attrs.get("value") or attrs.get("aria-label") or "按钮"
|
||
base = f"按钮: {text}"
|
||
elif tp == "input":
|
||
input_type = attrs.get("type", "text")
|
||
text = ctrl.get("text") or attrs.get("value") or attrs.get("placeholder") or ""
|
||
base = f"输入框 ({input_type})"
|
||
if text:
|
||
base += f" [{text}]"
|
||
elif tp == "textarea":
|
||
text = ctrl.get("text") or attrs.get("placeholder") or ""
|
||
base = "多行文本框"
|
||
if text:
|
||
base += f" [{text}]"
|
||
elif tp == "select":
|
||
text = ctrl.get("text") or ""
|
||
base = "下拉选择框"
|
||
if text:
|
||
base += f" [{text}]"
|
||
elif is_interactive:
|
||
text = ctrl.get("text") or ""
|
||
role = attrs.get("role", "")
|
||
tag_desc = f"{tp}控件"
|
||
if role == "button":
|
||
tag_desc = "按钮(ARIA)"
|
||
elif role == "link":
|
||
tag_desc = "链接(ARIA)"
|
||
elif "card" in (attrs.get("class", "") + attrs.get("id", "")):
|
||
tag_desc = "卡片"
|
||
base = f"可交互{tag_desc}"
|
||
if text:
|
||
base += f" [{text}]"
|
||
if attrs.get("data-href"):
|
||
base += f" (目标: {attrs['data-href']})"
|
||
else:
|
||
text = ctrl.get("text") or ""
|
||
base = f"{tp} 控件"
|
||
if text:
|
||
base += f" [{text}]"
|
||
return base
|
||
|
||
|
||
def describe_page_tree(page: Dict[str, Any], prefix: str = "",
|
||
max_depth: int = 1, current_depth: int = 1,
|
||
is_last: bool = True) -> str:
|
||
url = page.get("url", "未知页面")
|
||
if "error" in page:
|
||
return f"{prefix}页面 {url} 抓取失败: {page['error']}\n"
|
||
|
||
controls = page.get("controls", [])
|
||
lines = []
|
||
connector = "└── " if is_last else "├── "
|
||
lines.append(f"{prefix}{connector}页面: {url}")
|
||
|
||
groups: Dict[str, List[Dict]] = {}
|
||
for ctrl in controls:
|
||
ctx = ctrl["context"]
|
||
groups.setdefault(ctx, []).append(ctrl)
|
||
|
||
sorted_groups = sorted(groups.items(), key=lambda x: x[0])
|
||
group_count = len(sorted_groups)
|
||
|
||
for idx, (context, items) in enumerate(sorted_groups):
|
||
is_last_group = (idx == group_count - 1)
|
||
group_connector = "└── " if is_last_group else "├── "
|
||
group_line = f"{prefix}{' ' if is_last else '│ '}{group_connector}【{context}】({len(items)}个)"
|
||
lines.append(group_line)
|
||
|
||
item_prefix = prefix + (" " if is_last else "│ ") + (" " if is_last_group else "│ ")
|
||
|
||
for j, ctrl in enumerate(items):
|
||
is_last_item = (j == len(items) - 1)
|
||
item_connector = "└── " if is_last_item else "├── "
|
||
desc = short_text_for_control(ctrl)
|
||
lines.append(f"{item_prefix}{item_connector}{desc}")
|
||
|
||
for event in ctrl.get("events", []):
|
||
if "target_page" in event and current_depth < max_depth:
|
||
sub_page = event["target_page"]
|
||
new_prefix = item_prefix + (" " if is_last_item else "│ ")
|
||
sub_desc = describe_page_tree(
|
||
sub_page,
|
||
prefix=new_prefix,
|
||
max_depth=max_depth,
|
||
current_depth=current_depth + 1,
|
||
is_last=True
|
||
)
|
||
lines.append(sub_desc)
|
||
|
||
return "\n".join(lines) + "\n"
|
||
|
||
|
||
def generate_human_report(root_page: Dict[str, Any], max_depth: int) -> str:
|
||
header = f"🌐 页面交互控件分析报告（增强版）\n深度限制: {max_depth}\n" + "=" * 40 + "\n"
|
||
tree = describe_page_tree(root_page, prefix="", max_depth=max_depth,
|
||
current_depth=1, is_last=True)
|
||
return header + tree
|
||
|
||
|
||
# ---------------------------- 命令行入口 ----------------------------
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="通用网页交互控件分析工具（增强版：模拟点击、去重）")
|
||
parser.add_argument("source", help="起始 URL 或本地 HTML 文件路径")
|
||
parser.add_argument("-N", "--depth", type=int, default=1, help="最大递归深度（默认1）")
|
||
parser.add_argument("-o", "--output", help="输出文本文件路径，默认打印到控制台")
|
||
parser.add_argument("--json", action="store_true", help="同时保存机器可读的 JSON 文件（与 -o 同名 .json）")
|
||
args = parser.parse_args()
|
||
|
||
max_depth = max(1, args.depth)
|
||
|
||
if args.source.startswith(('http://', 'https://')):
|
||
start_url = args.source
|
||
else:
|
||
abs_path = os.path.abspath(args.source)
|
||
start_url = Path(abs_path).as_uri()
|
||
visited = {start_url}
|
||
|
||
ensure_debug_browser()
|
||
|
||
service = Service(EdgeChromiumDriverManager().install())
|
||
options = Options()
|
||
options.add_experimental_option("debuggerAddress", "127.0.0.1:9222")
|
||
|
||
driver = webdriver.Edge(service=service, options=options)
|
||
driver.implicitly_wait(5)
|
||
|
||
try:
|
||
tree = process_source(start_url, depth=1, max_depth=max_depth,
|
||
visited=visited, driver=driver)
|
||
finally:
|
||
driver.quit()
|
||
|
||
report = generate_human_report(tree, max_depth)
|
||
|
||
if args.output:
|
||
with open(args.output, 'w', encoding='utf-8') as f:
|
||
f.write(report)
|
||
print(f"人性化报告已保存到 {args.output}")
|
||
if args.json:
|
||
json_path = args.output.rsplit('.', 1)[0] + '.json'
|
||
with open(json_path, 'w', encoding='utf-8') as jf:
|
||
json.dump(tree, jf, ensure_ascii=False, indent=2)
|
||
print(f"机器可读 JSON 已保存到 {json_path}")
|
||
else:
|
||
print(report)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |