qiuruiandClaude Opus 4.6
bbea70dc9c
文档刷新： - CLAUDE.md/README.md/DEVELOP.md/RELEASE.md 同步 org v2 架构 （Boss/GroupLeader/SeniorDev/Ops 组织层、EventBus/DailyReport/OpsHub/ AgentScorer/VersionPipeline 工具层、nrf-dev/esp32-dev/market-* 新角色） - 清除所有 /data/rockchip 路径引用 → /data/company（源码+测试+前端） - 保留 "Rockchip" 芯片厂商名作为技术描述（hw_engineer/algo_vision 等） - docs/plans/ 历史文档保持原样（反映当时路径状态） Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
401 lines
14 KiB
Python
401 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
24小时项目自动化运行测试 - 基于 Playwright 浏览器自动化
|
||
|
||
用法：
|
||
python scripts/test_24h.py --hours 24 --project /data/company/algo-base/logodetect-x86
|
||
python scripts/test_24h.py --hours 1 --project /data/company/algo-base/logodetect-x86 --dry-run
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import time
|
||
import urllib.request
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
|
||
|
||
DASHBOARD_URL = "http://localhost:9080"
|
||
REPORT_DIR = Path("tmp/test_24h_report")
|
||
CHECK_INTERVAL = 900 # 每 15 分钟一次
|
||
|
||
|
||
# ── 辅助函数 ─────────────────────────────────────────────────────────
|
||
|
||
def _api(method: str, path: str, body: dict | None = None) -> dict:
|
||
url = DASHBOARD_URL + path
|
||
data = json.dumps(body).encode() if body else None
|
||
headers = {"Content-Type": "application/json"} if data else {}
|
||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||
with urllib.request.urlopen(req, timeout=10) as r:
|
||
return json.loads(r.read())
|
||
|
||
|
||
def _get_project_tasks(project_name: str) -> list[dict]:
|
||
return _api("GET", f"/api/tasks?project={project_name}")
|
||
|
||
|
||
def _task_stats(tasks: list[dict]) -> dict:
|
||
statuses = [t["status"] for t in tasks]
|
||
return {
|
||
"total": len(tasks),
|
||
"pending": statuses.count("pending"),
|
||
"running": statuses.count("running"),
|
||
"done": statuses.count("done"),
|
||
"failed": statuses.count("failed"),
|
||
"vetoed": statuses.count("vetoed"),
|
||
"waiting": statuses.count("waiting_approval"),
|
||
}
|
||
|
||
|
||
# ── Phase 1: 浏览器 UI 操作（Playwright）─────────────────────────────
|
||
|
||
def setup_via_browser(
|
||
page, # playwright Page
|
||
project_path: str,
|
||
linked_project: str,
|
||
project_name: str,
|
||
) -> dict:
|
||
"""通过浏览器 UI 完整设置项目（导入 + 生成任务 + 启动执行器）"""
|
||
from playwright.sync_api import expect
|
||
|
||
print(f"[setup] 打开 Dashboard: {DASHBOARD_URL}")
|
||
page.goto(DASHBOARD_URL)
|
||
page.wait_for_timeout(2000)
|
||
|
||
# 截图：初始状态
|
||
_screenshot(page, "00_initial")
|
||
|
||
# 点击「导入项目」
|
||
print("[setup] 点击「导入项目」")
|
||
page.click("button:has-text('导入项目')")
|
||
page.wait_for_timeout(500)
|
||
|
||
# 填写路径（触发项目名自动推断）
|
||
print(f"[setup] 填写路径: {project_path}")
|
||
# 第 1 个 input = 目录路径
|
||
page.locator("input").nth(0).fill(project_path)
|
||
page.wait_for_timeout(500)
|
||
|
||
# 填写关联项目（第 3 个 input，placeholder 含 logo-collect）
|
||
if linked_project:
|
||
print(f"[setup] 填写关联项目: {linked_project}")
|
||
page.locator("input").nth(2).fill(linked_project)
|
||
|
||
# 确认 mode = auto
|
||
mode_select = page.locator("select")
|
||
mode_select.select_option("auto")
|
||
|
||
# 点击「预览任务」展开模板
|
||
try:
|
||
page.click("button:has-text('预览任务')")
|
||
page.wait_for_timeout(300)
|
||
except Exception:
|
||
pass
|
||
|
||
_screenshot(page, "01_import_modal_filled")
|
||
|
||
# 点击「导入并生成任务」
|
||
print("[setup] 提交导入表单")
|
||
page.click("button:has-text('导入并生成任务')")
|
||
page.wait_for_timeout(3000)
|
||
|
||
_screenshot(page, "02_import_success")
|
||
|
||
# 点击「完成」关闭成功弹窗
|
||
try:
|
||
page.click("button:has-text('完成')")
|
||
page.wait_for_timeout(1000)
|
||
except Exception:
|
||
pass
|
||
|
||
# 切换到目标项目 Tab
|
||
print(f"[setup] 切换到项目: {project_name}")
|
||
try:
|
||
page.click(f"button:has-text('{project_name}')")
|
||
page.wait_for_timeout(500)
|
||
except Exception:
|
||
pass
|
||
|
||
_screenshot(page, "03_project_kanban")
|
||
|
||
# 启动执行器（点击「开始执行」）
|
||
print("[setup] 启动执行器")
|
||
try:
|
||
page.click("button:has-text('开始执行')")
|
||
page.wait_for_timeout(2000)
|
||
except Exception as e:
|
||
print(f"[WARN] 启动执行器按钮未找到: {e}")
|
||
|
||
_screenshot(page, "04_executor_started")
|
||
|
||
# 收集初始任务统计
|
||
tasks = _get_project_tasks(project_name)
|
||
stats = _task_stats(tasks)
|
||
print(f"[setup] 初始任务统计: {stats}")
|
||
return stats
|
||
|
||
|
||
# ── Phase 2: 24 小时监控循环 ─────────────────────────────────────────
|
||
|
||
def monitoring_loop(
|
||
page,
|
||
project_name: str,
|
||
total_seconds: int,
|
||
dry_run: bool = False,
|
||
) -> list[dict]:
|
||
"""每 CHECK_INTERVAL 秒检查一次，记录任务进展"""
|
||
metrics: list[dict] = []
|
||
start_time = time.time()
|
||
check_num = 0
|
||
interval = 60 if dry_run else CHECK_INTERVAL # dry-run 用 1 分钟
|
||
|
||
print(f"\n[monitor] 开始监控，总时长 {total_seconds/3600:.1f}h，间隔 {interval}s")
|
||
|
||
while time.time() - start_time < total_seconds:
|
||
check_num += 1
|
||
elapsed_s = time.time() - start_time
|
||
elapsed_h = elapsed_s / 3600
|
||
|
||
print(f"\n[monitor] 第 {check_num} 次检查 ({elapsed_h:.2f}h / {total_seconds/3600:.1f}h)")
|
||
|
||
# 刷新页面，截图
|
||
try:
|
||
page.goto(DASHBOARD_URL)
|
||
try:
|
||
page.click(f"button:has-text('{project_name}')")
|
||
except Exception:
|
||
pass
|
||
page.wait_for_timeout(2000)
|
||
_screenshot(page, f"monitor_{check_num:04d}_{elapsed_h:.1f}h")
|
||
except Exception as e:
|
||
print(f"[WARN] 截图失败: {e}")
|
||
|
||
# 收集任务统计
|
||
try:
|
||
tasks = _get_project_tasks(project_name)
|
||
stats = _task_stats(tasks)
|
||
exec_status = _api("GET", "/api/executor/status")
|
||
|
||
record = {
|
||
"check": check_num,
|
||
"elapsed_h": round(elapsed_h, 2),
|
||
"timestamp": datetime.utcnow().isoformat(),
|
||
"tasks": stats,
|
||
"executor": exec_status,
|
||
"done_titles": [t["title"] for t in tasks if t["status"] == "done"],
|
||
"failed_titles": [t["title"] for t in tasks if t["status"] == "failed"],
|
||
}
|
||
metrics.append(record)
|
||
print(f" 任务: {stats}")
|
||
print(f" 执行器: running={exec_status.get('running')}, "
|
||
f"active={exec_status.get('active')}")
|
||
|
||
except Exception as e:
|
||
print(f"[WARN] 指标收集失败: {e}")
|
||
|
||
# 检查执行器是否停止，若停止则重启
|
||
try:
|
||
exec_st = _api("GET", "/api/executor/status")
|
||
if not exec_st.get("running") and exec_st.get("pending", 0) > 0:
|
||
print("[monitor] 执行器已停止，重新启动...")
|
||
_api("POST", "/api/executor/start")
|
||
except Exception:
|
||
pass
|
||
|
||
# 等待下次检查
|
||
next_check = start_time + check_num * interval
|
||
wait_s = max(0, next_check - time.time())
|
||
if wait_s > 0 and time.time() - start_time < total_seconds:
|
||
print(f" 下次检查：{wait_s:.0f}s 后")
|
||
time.sleep(wait_s)
|
||
|
||
return metrics
|
||
|
||
|
||
# ── Phase 3: 评估报告生成 ─────────────────────────────────────────────
|
||
|
||
def generate_report(
|
||
project_name: str,
|
||
project_path: str,
|
||
linked_project: str,
|
||
metrics: list[dict],
|
||
total_hours: float,
|
||
page,
|
||
) -> Path:
|
||
"""生成 Markdown 评估报告"""
|
||
if not metrics:
|
||
print("[report] 无指标数据")
|
||
return REPORT_DIR / "report.md"
|
||
|
||
first = metrics[0]["tasks"]
|
||
last = metrics[-1]["tasks"]
|
||
|
||
tasks_done = last["done"]
|
||
tasks_total = last["total"]
|
||
tasks_failed = last["failed"]
|
||
completion_pct = round(tasks_done / max(tasks_total, 1) * 100, 1)
|
||
|
||
# 收集所有 done 任务标题
|
||
all_done = []
|
||
for m in metrics:
|
||
for title in m.get("done_titles", []):
|
||
if title not in all_done:
|
||
all_done.append(title)
|
||
|
||
all_failed = []
|
||
for m in metrics:
|
||
for title in m.get("failed_titles", []):
|
||
if title not in all_failed:
|
||
all_failed.append(title)
|
||
|
||
# 执行器在线率
|
||
online_checks = sum(1 for m in metrics if m.get("executor", {}).get("running"))
|
||
online_pct = round(online_checks / max(len(metrics), 1) * 100, 1)
|
||
|
||
report_lines = [
|
||
f"# {total_hours:.0f}h 自动化运行评估报告",
|
||
f"",
|
||
f"**项目**: `{project_name}` ({project_path})",
|
||
f"**关联项目**: `{linked_project or '无'}`",
|
||
f"**报告时间**: {datetime.now().strftime('%Y-%m-%d %H:%M')}",
|
||
f"**运行时长**: {total_hours:.1f} 小时",
|
||
f"",
|
||
f"---",
|
||
f"",
|
||
f"## 执行摘要",
|
||
f"",
|
||
f"| 指标 | 数值 |",
|
||
f"|------|------|",
|
||
f"| 总任务数 | {tasks_total} |",
|
||
f"| 完成任务 | {tasks_done} ({completion_pct}%) |",
|
||
f"| 失败任务 | {tasks_failed} |",
|
||
f"| 待执行 | {last['pending']} |",
|
||
f"| 执行器在线率 | {online_pct}% |",
|
||
f"",
|
||
f"## 已完成任务",
|
||
f"",
|
||
]
|
||
for t in all_done:
|
||
report_lines.append(f"- ✅ {t}")
|
||
|
||
if all_failed:
|
||
report_lines += ["", "## 失败任务", ""]
|
||
for t in all_failed:
|
||
report_lines.append(f"- ❌ {t}")
|
||
|
||
report_lines += [
|
||
"",
|
||
"## 进度时间线",
|
||
"",
|
||
"| 时间点 | 完成 | 失败 | 待执行 |",
|
||
"|--------|------|------|--------|",
|
||
]
|
||
for m in metrics[::4] + ([metrics[-1]] if metrics else []): # 每4条取1条
|
||
s = m["tasks"]
|
||
report_lines.append(
|
||
f"| {m['elapsed_h']:.1f}h | {s['done']} | {s['failed']} | {s['pending']} |"
|
||
)
|
||
|
||
report_lines += [
|
||
"",
|
||
"## 截图目录",
|
||
f"",
|
||
f"见 `{REPORT_DIR}/` 目录，共 {len(list(REPORT_DIR.glob('*.png')))} 张截图。",
|
||
"",
|
||
"---",
|
||
f"*由 `scripts/test_24h.py` 自动生成*",
|
||
]
|
||
|
||
report_path = REPORT_DIR / "report.md"
|
||
report_path.write_text("\n".join(report_lines), encoding="utf-8")
|
||
print(f"\n[report] 报告已生成: {report_path}")
|
||
return report_path
|
||
|
||
|
||
# ── 主入口 ────────────────────────────────────────────────────────────
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="24h 项目自动化运行测试")
|
||
parser.add_argument("--project", default="/data/company/algo-base/logodetect-x86",
|
||
help="项目目录路径")
|
||
parser.add_argument("--linked", default="/data/test-platform/logo-collect",
|
||
help="关联项目路径（用于集成任务）")
|
||
parser.add_argument("--hours", type=float, default=24.0,
|
||
help="运行时长（小时），默认 24")
|
||
parser.add_argument("--dry-run", action="store_true",
|
||
help="快速测试模式（1 分钟间隔，5 分钟总时长）")
|
||
parser.add_argument("--headless", action="store_true", default=True,
|
||
help="无头模式运行浏览器（默认开启）")
|
||
parser.add_argument("--no-headless", dest="headless", action="store_false",
|
||
help="显示浏览器窗口（需要显示器）")
|
||
args = parser.parse_args()
|
||
|
||
project_path = args.project
|
||
project_name = Path(project_path).name
|
||
linked_project = args.linked
|
||
total_hours = 5 / 60 if args.dry_run else args.hours # dry-run = 5 分钟
|
||
total_seconds = int(total_hours * 3600)
|
||
|
||
REPORT_DIR.mkdir(parents=True, exist_ok=True)
|
||
metrics_path = REPORT_DIR / "metrics.jsonl"
|
||
|
||
print("=" * 60)
|
||
print(f"NMFS Agents 自动化测试")
|
||
print(f" 项目: {project_name} ({project_path})")
|
||
print(f" 关联: {linked_project}")
|
||
print(f" 时长: {total_hours:.2f}h ({'dry-run' if args.dry_run else '正式'})")
|
||
print(f" 报告: {REPORT_DIR}/")
|
||
print("=" * 60)
|
||
|
||
try:
|
||
from playwright.sync_api import sync_playwright
|
||
except ImportError:
|
||
print("[ERROR] playwright 未安装，运行: pip install playwright && playwright install chromium")
|
||
raise SystemExit(1)
|
||
|
||
with sync_playwright() as pw:
|
||
browser = pw.chromium.launch(headless=args.headless)
|
||
ctx = browser.new_context(viewport={"width": 1440, "height": 900})
|
||
page = ctx.new_page()
|
||
|
||
try:
|
||
# Phase 1: 设置
|
||
print("\n=== Phase 1: 浏览器 UI 设置 ===")
|
||
setup_via_browser(page, project_path, linked_project, project_name)
|
||
|
||
# Phase 2: 监控
|
||
print("\n=== Phase 2: 监控循环 ===")
|
||
metrics = monitoring_loop(page, project_name, total_seconds, args.dry_run)
|
||
|
||
# 保存原始指标
|
||
with open(metrics_path, "w") as f:
|
||
for m in metrics:
|
||
f.write(json.dumps(m, ensure_ascii=False) + "\n")
|
||
|
||
# Phase 3: 生成报告
|
||
print("\n=== Phase 3: 生成评估报告 ===")
|
||
_screenshot(page, "final_state")
|
||
report_path = generate_report(
|
||
project_name, project_path, linked_project,
|
||
metrics, total_hours, page,
|
||
)
|
||
print(f"\n完成！评估报告: {report_path}")
|
||
|
||
finally:
|
||
browser.close()
|
||
|
||
|
||
def _screenshot(page, name: str) -> None:
|
||
try:
|
||
path = REPORT_DIR / f"{name}.png"
|
||
page.screenshot(path=str(path), full_page=False)
|
||
print(f" [screenshot] {path.name}")
|
||
except Exception as e:
|
||
print(f" [WARN] 截图失败 {name}: {e}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|