feat: add service config templates and extraction script

Former-commit-id: 1de24b7eb79676d1aba9d799a58c5a753290cf52
This commit is contained in:
反编译工作区
2026-05-01 19:38:01 +08:00
parent 3175b7074b
commit 8b15445328
2433 changed files with 8322164 additions and 1604 deletions
+252
View File
@@ -0,0 +1,252 @@
#!/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())