mirror of
https://github.com/hpd840321/starRiverProperty.git
synced 2026-06-09 16:30:29 +08:00
hpd840321
7b2bd307f1
- backend/: 13 Maven modules (cw-elevator-application, cloudwalk-cloud, intelligent-cwoscomponent, ninca-crk, etc.) - frontend/: 4 Vue projects (elevator-front, cwos-portal, alarm-front, front_acs) + decompiled + scripts - scripts/: build, test-env, tools (Docker Compose, service templates, API parity) - docs/: AGENTS.md, superpowers specs, architecture docs - .gitignore: standard Java/Maven exclusions Moved from legacy maven-*/ root layout to backend/ organized structure.
253 lines
8.9 KiB
Python
253 lines
8.9 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
前端运行时自动验收（Playwright）：
|
||
1) 打开页面并等待首屏稳定
|
||
2) 采集 console/pageerror/requestfailed
|
||
3) 统计重复资源请求（按 URL 去 query 后聚合）
|
||
4) 输出 markdown 报告 + 截图
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import datetime as dt
|
||
import json
|
||
import re
|
||
import sys
|
||
from collections import Counter
|
||
from pathlib import Path
|
||
from typing import Dict, List, Tuple
|
||
from urllib.parse import urlsplit, urlunsplit
|
||
|
||
from playwright.async_api import Error as PlaywrightError
|
||
from playwright.async_api import async_playwright
|
||
|
||
|
||
def _strip_query(url: str) -> str:
|
||
p = urlsplit(url)
|
||
return urlunsplit((p.scheme, p.netloc, p.path, "", ""))
|
||
|
||
|
||
def _now_tag() -> str:
|
||
return dt.datetime.now().strftime("%Y%m%d-%H%M%S")
|
||
|
||
|
||
def _classify_resource(url: str) -> str:
|
||
path = urlsplit(url).path.lower()
|
||
if path.endswith(".js"):
|
||
return "js"
|
||
if path.endswith(".css"):
|
||
return "css"
|
||
if re.search(r"\.(png|jpg|jpeg|gif|svg|ico|webp)$", path):
|
||
return "image"
|
||
if "/api/" in path or "/elevator/" in path:
|
||
return "api"
|
||
return "other"
|
||
|
||
|
||
async def run_check(url: str, wait_ms: int, screenshot_path: Path) -> Dict[str, object]:
|
||
console_errors: List[str] = []
|
||
console_warnings: List[str] = []
|
||
page_errors: List[str] = []
|
||
failed_requests: List[str] = []
|
||
requests: List[Tuple[str, str]] = []
|
||
|
||
async with async_playwright() as p:
|
||
browser = await p.chromium.launch(headless=True)
|
||
context = await browser.new_context(ignore_https_errors=True)
|
||
page = await context.new_page()
|
||
page.set_default_timeout(5_000)
|
||
|
||
page.on(
|
||
"console",
|
||
lambda msg: (
|
||
console_errors.append(msg.text)
|
||
if msg.type == "error"
|
||
else console_warnings.append(msg.text)
|
||
if msg.type in {"warning", "warn"}
|
||
else None
|
||
),
|
||
)
|
||
page.on("pageerror", lambda err: page_errors.append(str(err)))
|
||
def _on_request_failed(req) -> None:
|
||
failure = req.failure
|
||
if isinstance(failure, dict):
|
||
err_text = failure.get("errorText", "unknown")
|
||
elif isinstance(failure, str):
|
||
err_text = failure
|
||
elif failure is None:
|
||
err_text = "unknown"
|
||
else:
|
||
err_text = getattr(failure, "error_text", str(failure))
|
||
failed_requests.append(f"{req.url} | {err_text}")
|
||
|
||
page.on("requestfailed", _on_request_failed)
|
||
page.on("requestfinished", lambda req: requests.append((req.method, req.url)))
|
||
|
||
navigation_mode = "domcontentloaded"
|
||
try:
|
||
try:
|
||
await page.goto(url, wait_until="domcontentloaded", timeout=12_000)
|
||
except PlaywrightError:
|
||
# 某些历史页面会因脚本阻塞导致 domcontentloaded 不触发，降级到 commit 保证可诊断
|
||
navigation_mode = "commit(fallback)"
|
||
await page.goto(url, wait_until="commit", timeout=20_000)
|
||
await page.wait_for_timeout(wait_ms)
|
||
|
||
try:
|
||
# 避免 full_page 在超长文档上触发超时，先抓取可视区域。
|
||
await page.screenshot(path=str(screenshot_path), full_page=False, timeout=15_000)
|
||
except Exception:
|
||
screenshot_path.write_bytes(b"")
|
||
|
||
try:
|
||
title = await page.title()
|
||
except Exception:
|
||
title = ""
|
||
|
||
try:
|
||
body_non_empty = bool(await page.evaluate("Boolean(document.body && document.body.innerText && document.body.innerText.trim())"))
|
||
except Exception:
|
||
body_non_empty = False
|
||
|
||
try:
|
||
app_count = await page.locator("#app *").count()
|
||
except Exception:
|
||
app_count = -1
|
||
except PlaywrightError as e:
|
||
await browser.close()
|
||
return {
|
||
"fatal_error": str(e),
|
||
"console_errors": console_errors,
|
||
"console_warnings": console_warnings,
|
||
"page_errors": page_errors,
|
||
"failed_requests": failed_requests,
|
||
"requests": requests,
|
||
}
|
||
|
||
await browser.close()
|
||
|
||
normalized_urls = [_strip_query(u) for _, u in requests]
|
||
url_counter = Counter(normalized_urls)
|
||
duplicate_urls = [(u, c) for u, c in url_counter.items() if c > 1]
|
||
duplicate_urls.sort(key=lambda x: x[1], reverse=True)
|
||
|
||
resource_counter = Counter(_classify_resource(u) for u in normalized_urls)
|
||
return {
|
||
"fatal_error": "",
|
||
"navigation_mode": navigation_mode,
|
||
"title": title,
|
||
"body_non_empty": body_non_empty,
|
||
"app_count": app_count,
|
||
"console_errors": console_errors,
|
||
"console_warnings": console_warnings,
|
||
"page_errors": page_errors,
|
||
"failed_requests": failed_requests,
|
||
"requests_total": len(requests),
|
||
"requests_unique": len(url_counter),
|
||
"resource_counter": dict(resource_counter),
|
||
"duplicate_urls": duplicate_urls[:30],
|
||
}
|
||
|
||
|
||
def _render_report(url: str, result: Dict[str, object], screenshot_path: Path) -> str:
|
||
lines: List[str] = []
|
||
lines.append("# 前端运行时自动验收报告")
|
||
lines.append("")
|
||
lines.append(f"- URL: `{url}`")
|
||
lines.append(f"- 时间: `{dt.datetime.now().isoformat(timespec='seconds')}`")
|
||
lines.append(f"- 截图: `{screenshot_path}`")
|
||
lines.append("")
|
||
|
||
fatal = str(result.get("fatal_error", "")).strip()
|
||
if fatal:
|
||
lines.append("## 结论")
|
||
lines.append("- **失败**：页面打开阶段异常")
|
||
lines.append(f"- 异常: `{fatal}`")
|
||
return "\n".join(lines) + "\n"
|
||
|
||
lines.append("## 页面状态")
|
||
lines.append(f"- 导航模式: `{result.get('navigation_mode', '')}`")
|
||
lines.append(f"- title: `{result.get('title', '')}`")
|
||
lines.append(f"- body 是否非空: `{result.get('body_non_empty', False)}`")
|
||
lines.append(f"- `#app` 子节点数: `{result.get('app_count', 0)}`")
|
||
lines.append("")
|
||
|
||
lines.append("## 请求统计")
|
||
lines.append(f"- 总请求数: `{result.get('requests_total', 0)}`")
|
||
lines.append(f"- 去重后 URL 数: `{result.get('requests_unique', 0)}`")
|
||
lines.append(f"- 类型分布: `{json.dumps(result.get('resource_counter', {}), ensure_ascii=False)}`")
|
||
lines.append("")
|
||
|
||
dup = result.get("duplicate_urls", [])
|
||
lines.append("## 重复请求（Top）")
|
||
if dup:
|
||
for u, c in dup:
|
||
lines.append(f"- `{c}x` {u}")
|
||
else:
|
||
lines.append("- 无重复 URL 请求")
|
||
lines.append("")
|
||
|
||
def _section(title: str, rows: List[str]) -> None:
|
||
lines.append(f"## {title}")
|
||
if rows:
|
||
for row in rows[:50]:
|
||
lines.append(f"- {row}")
|
||
else:
|
||
lines.append("- 无")
|
||
lines.append("")
|
||
|
||
_section("Console Error", result.get("console_errors", []))
|
||
_section("Page Error", result.get("page_errors", []))
|
||
_section("Request Failed", result.get("failed_requests", []))
|
||
_section("Console Warning", result.get("console_warnings", []))
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
async def _amain() -> int:
|
||
parser = argparse.ArgumentParser(description="自动检测前端运行时错误/重复加载")
|
||
parser.add_argument("--url", default="http://127.0.0.1:8090/login", help="待检测 URL")
|
||
parser.add_argument(
|
||
"--wait-ms", type=int, default=5000, help="页面打开后额外等待毫秒数（默认 5000）"
|
||
)
|
||
parser.add_argument(
|
||
"--out-dir",
|
||
default="artifacts/frontend-check",
|
||
help="报告与截图输出目录（默认 artifacts/frontend-check）",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
out_dir = Path(args.out_dir).resolve()
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
tag = _now_tag()
|
||
screenshot = out_dir / f"frontend-{tag}.png"
|
||
report = out_dir / f"frontend-{tag}.md"
|
||
|
||
result = await run_check(args.url, args.wait_ms, screenshot)
|
||
report.write_text(_render_report(args.url, result, screenshot), encoding="utf-8")
|
||
|
||
print(f"[frontend-check] report={report}")
|
||
print(f"[frontend-check] screenshot={screenshot}")
|
||
if result.get("fatal_error"):
|
||
return 2
|
||
if result.get("console_errors") or result.get("page_errors") or result.get("failed_requests"):
|
||
return 1
|
||
return 0
|
||
|
||
|
||
def main() -> int:
|
||
try:
|
||
return asyncio.run(_amain())
|
||
except KeyboardInterrupt:
|
||
return 130
|
||
except Exception as e:
|
||
print(f"[frontend-check] unexpected error: {e}", file=sys.stderr)
|
||
return 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|