webtest-agent v1.0.0: AI网页测试智能体（LLM决策+agent-browser执行+断言+报告）
This commit is contained in:
@@ -0,0 +1,6 @@
|
||||
__pycache__/
|
||||
*.pyc
|
||||
logs/
|
||||
data/tasks/
|
||||
data/*.db
|
||||
nohup.out
|
||||
@@ -0,0 +1,78 @@
|
||||
# webtest-agent · AI 网页测试智能体
|
||||
|
||||
基于 **LLM 决策 + 无头浏览器执行** 的端到端网页测试系统。用自然语言描述测试目标，AI 自动打开页面、逐步操作、断言验证、生成带截图的测试报告。
|
||||
|
||||
## 架构
|
||||
|
||||
```
|
||||
┌─────────────────────────────────────────────┐
|
||||
│ HTTP API (Flask) │
|
||||
│ POST /api/tasks → 发起测试 │
|
||||
│ GET /api/tasks → 任务列表/状态 │
|
||||
│ GET /api/tasks/<id>/report → HTML 报告 │
|
||||
├─────────────────────────────────────────────┤
|
||||
│ Agent 循环（每任务一线程） │
|
||||
│ LLM(火山引擎 doubao) 决策下一步动作 │
|
||||
│ agent-browser CLI 执行（无头 Chrome） │
|
||||
│ 断言器 + 截图留证 + 报告生成 │
|
||||
└─────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
## 快速开始
|
||||
|
||||
```bash
|
||||
./start.sh # 启动（默认 16051 端口）
|
||||
./start.sh stop # 停止
|
||||
```
|
||||
|
||||
### API 用法
|
||||
|
||||
```bash
|
||||
# 发起测试
|
||||
curl -X POST http://localhost:16051/api/tasks \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"url":"https://example.com","goal":"验证标题包含 Example Domain，点击 Learn more 验证跳转","max_steps":12}'
|
||||
|
||||
# 查看任务
|
||||
curl http://localhost:16051/api/tasks/<task_id>
|
||||
|
||||
# 查看报告
|
||||
curl http://localhost:16051/api/tasks/<task_id>/report
|
||||
|
||||
# 停止任务
|
||||
curl -X POST http://localhost:16051/api/tasks/<task_id>/stop
|
||||
```
|
||||
|
||||
## AI 动作协议
|
||||
|
||||
LLM 每步输出一个 JSON 动作，由 agent-browser 执行：
|
||||
|
||||
| 动作 | 说明 |
|
||||
|------|------|
|
||||
| `click` | 点击元素（`@e1` 或 `text:按钮文字`） |
|
||||
| `fill` | 输入框填值 |
|
||||
| `select` | 下拉框选择 |
|
||||
| `press` | 按键（Enter/Tab 等） |
|
||||
| `wait` | 等待（毫秒 / `text:xx` / `url:xx`） |
|
||||
| `scroll` | 滚动页面 |
|
||||
| `assert` | 断言（text/url/title/element，PASS/FAIL） |
|
||||
| `screenshot` | 截图留证 |
|
||||
| `done` / `fail` | 结束任务并总结 |
|
||||
|
||||
## 关键特性
|
||||
|
||||
- **自然语言驱动**：测试目标直接用中文描述
|
||||
- **自我修复**：元素找不到时 LLM 自动换定位策略重试
|
||||
- **断言验证**：文本/URL/标题/元素存在性，输出 PASS/FAIL
|
||||
- **证据留存**：关键步骤自动截图，HTML 报告内嵌展示
|
||||
- **并发控制**：默认最多 2 个任务并发（`config.py` 可调）
|
||||
|
||||
## 配置
|
||||
|
||||
见 `config.py`：LLM 端点/模型、最大步数、超时、并发数均可通过环境变量覆盖。
|
||||
|
||||
## 依赖
|
||||
|
||||
- Python 3.10+（Flask 3.0）
|
||||
- agent-browser CLI（node 安装）：`npm install -g agent-browser`
|
||||
- 无头 Chrome（agent-browser install 自动安装）
|
||||
@@ -0,0 +1,457 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Agent 循环：LLM 决策 + agent-browser 执行 + 断言 + 报告"""
|
||||
import os
|
||||
import threading
|
||||
import time
|
||||
import traceback
|
||||
|
||||
from browser import AgentBrowser, BrowserError
|
||||
from config import (TASKS_DIR, DEFAULT_MAX_SNAPSHOT_CHARS, MAX_RETRY_SAME_ERROR,
|
||||
HISTORY_LIMIT)
|
||||
from db import update_task, save_step_log, load_step_logs
|
||||
from llm import chat_json, LLMError
|
||||
|
||||
SYSTEM_PROMPT = """你是一个专业的网页自动化测试工程师，正在使用浏览器工具执行端到端测试任务。
|
||||
你的目标：根据用户给的测试目标，实际操作网页完成测试，并通过断言判断测试是否通过。
|
||||
|
||||
## 可用的浏览器能力
|
||||
- click @e1 点击元素（@e1 是快照里的引用）
|
||||
- fill @e1 "值" 在输入框填入文字（会先清空）
|
||||
- select @e1 "值" 选择下拉框选项
|
||||
- press "Enter" 按键（Enter/Tab/Escape/Control+a 等）
|
||||
- wait "2000" 等待毫秒；wait text:"xx" 等待文本出现；wait url:"/path" 等待 URL 变化
|
||||
- scroll "down 500" 滚动页面
|
||||
- screenshot 截图留证（重要步骤请调用）
|
||||
- assert 断言验证（见下）
|
||||
- done 测试完成（全部通过）
|
||||
- fail 测试失败（无法继续或断言不通过）
|
||||
|
||||
## 断言类型 assert
|
||||
{"type":"text","expect":"欢迎","present":true} 页面上应出现"欢迎"文本
|
||||
{"type":"text","expect":"错误提示","present":false} 页面上不应出现"错误提示"
|
||||
{"type":"url","expect":"/dashboard","present":true} URL 应包含 /dashboard
|
||||
{"type":"title","expect":"首页","present":true} 页面标题应包含"首页"
|
||||
{"type":"element","expect":"登录","present":true} 页面上应存在文本为"登录"的元素
|
||||
断言执行后会返回 PASS 或 FAIL，根据结果决定下一步。
|
||||
|
||||
## 输出格式（严格 JSON，不要输出其他内容）
|
||||
{
|
||||
"reason": "简要中文说明当前判断和下一步计划",
|
||||
"action": "click|fill|select|press|wait|scroll|screenshot|assert|done|fail",
|
||||
"target": "@e1 或 text:按钮文字",
|
||||
"value": "填入的值/按键名/等待参数",
|
||||
"assert": {断言对象，action=assert 时必填},
|
||||
"summary": "测试结论（action=done/fail 时必填，说明验证了什么）"
|
||||
}
|
||||
|
||||
## 工作原则
|
||||
1. 先理解页面，再逐步操作；每一步只做一件事。
|
||||
2. 找不到目标元素时，观察快照选择最接近的 ref，或改用 text: 语义定位。
|
||||
3. 操作后必须用 assert 验证结果，不要盲目继续。
|
||||
4. 测试失败(断言 FAIL 且无补救)时用 fail 结束，并说明原因。
|
||||
5. 全部验证通过后用 done 结束，summary 里总结测试覆盖的内容。
|
||||
6. 快照可能被截断，必要时用 wait 等待页面加载完成再操作。
|
||||
"""
|
||||
|
||||
|
||||
def _fmt_snapshot(snap, max_chars=DEFAULT_MAX_SNAPSHOT_CHARS):
|
||||
"""把快照数据转成紧凑文本给 LLM"""
|
||||
lines = []
|
||||
refs = (snap or {}).get('refs') or {}
|
||||
if isinstance(refs, dict):
|
||||
for ref, info in refs.items():
|
||||
if isinstance(info, dict):
|
||||
name = info.get('name') or info.get('text') or info.get('label') or ''
|
||||
role = info.get('role') or ''
|
||||
lines.append(f'{ref} [{role}] {name}')
|
||||
else:
|
||||
lines.append(f'{ref} {info}')
|
||||
text = '\n'.join(lines)
|
||||
if len(text) > max_chars:
|
||||
text = text[:max_chars] + '\n...(快照过长已截断)'
|
||||
return text
|
||||
|
||||
|
||||
def _fmt_history(steps):
|
||||
"""格式化历史步骤给 LLM"""
|
||||
out = []
|
||||
for s in steps[-HISTORY_LIMIT:]:
|
||||
act = s.get('action', '')
|
||||
tgt = s.get('target', '')
|
||||
val = s.get('value', '')
|
||||
res = s.get('result', '')
|
||||
detail = s.get('detail', '')
|
||||
line = f"[步骤{s.get('n')}] {act} {tgt} {val if val else ''} -> {res}"
|
||||
if detail:
|
||||
line += f' ({detail[:120]})'
|
||||
out.append(line)
|
||||
return '\n'.join(out)
|
||||
|
||||
|
||||
class TaskRunner(threading.Thread):
|
||||
def __init__(self, task_id, url, goal, max_steps, timeout):
|
||||
super().__init__(daemon=True, name=f'task-{task_id}')
|
||||
self.task_id = task_id
|
||||
self.url = url
|
||||
self.goal = goal
|
||||
self.max_steps = max_steps
|
||||
self.timeout = timeout
|
||||
self.stop_flag = threading.Event()
|
||||
self.task_dir = os.path.join(TASKS_DIR, task_id)
|
||||
os.makedirs(self.task_dir, exist_ok=True)
|
||||
|
||||
def stop(self):
|
||||
self.stop_flag.set()
|
||||
|
||||
def _log(self, msg):
|
||||
ts = time.strftime('%H:%M:%S')
|
||||
print(f'[{ts}][{self.task_id}] {msg}', flush=True)
|
||||
|
||||
def _screenshot(self, browser, name):
|
||||
try:
|
||||
path = os.path.join(self.task_dir, name)
|
||||
browser.screenshot(path)
|
||||
return os.path.basename(path)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def run(self):
|
||||
update_task(self.task_id, status='running', started_at=time.time(),
|
||||
result='running')
|
||||
steps = []
|
||||
browser = None
|
||||
try:
|
||||
browser = AgentBrowser(namespace=f'task-{self.task_id}')
|
||||
deadline = time.time() + self.timeout
|
||||
self._log(f'打开页面: {self.url}')
|
||||
browser.open(self.url, timeout=60)
|
||||
browser.wait('--load', 'networkidle', timeout=45)
|
||||
self._log('页面已打开')
|
||||
|
||||
step_n = 0
|
||||
while step_n < self.max_steps:
|
||||
if self.stop_flag.is_set():
|
||||
self._finish(browser, 'stopped', '任务被手动停止', steps)
|
||||
return
|
||||
if time.time() > deadline:
|
||||
self._finish(browser, 'error', '任务超时', steps)
|
||||
return
|
||||
|
||||
step_n += 1
|
||||
self._log(f'--- 步骤 {step_n}/{self.max_steps} ---')
|
||||
|
||||
# 1. 快照
|
||||
try:
|
||||
snap = browser.snapshot(interactive=True, compact=True, timeout=45)
|
||||
snap_text = _fmt_snapshot(snap)
|
||||
except BrowserError as e:
|
||||
snap_text = f'(快照失败: {e})'
|
||||
|
||||
cur_url = browser.url()
|
||||
cur_title = browser.title()
|
||||
|
||||
# 2. LLM 决策
|
||||
user_msg = (
|
||||
f'## 测试目标\n{self.goal}\n\n'
|
||||
f'## 当前页面\nURL: {cur_url}\n标题: {cur_title}\n\n'
|
||||
f'## 页面可交互元素\n{snap_text}\n\n'
|
||||
f'## 已执行步骤\n' + (_fmt_history(steps) if steps else '(尚无)') +
|
||||
f'\n\n请输出下一步动作的 JSON。'
|
||||
)
|
||||
try:
|
||||
decision = chat_json(
|
||||
[{'role': 'system', 'content': SYSTEM_PROMPT},
|
||||
{'role': 'user', 'content': user_msg}],
|
||||
temperature=0.2, max_tokens=500)
|
||||
except LLMError as e:
|
||||
self._log(f'LLM 错误: {e}')
|
||||
self._finish(browser, 'error', f'LLM 决策失败: {e}', steps)
|
||||
return
|
||||
|
||||
action = decision.get('action', '')
|
||||
if action not in ('click', 'fill', 'select', 'press', 'wait',
|
||||
'scroll', 'screenshot', 'assert', 'done', 'fail'):
|
||||
self._log(f'非法动作: {action}')
|
||||
steps.append(self._record(step_n, decision, 'error',
|
||||
f'非法动作: {action}'))
|
||||
continue
|
||||
|
||||
# 3. 执行动作（带自愈重试）
|
||||
result, detail, extra = self._execute(browser, decision)
|
||||
self._log(f'动作 {action} -> {result} {detail}')
|
||||
|
||||
if action in ('done', 'fail'):
|
||||
self._finish(browser,
|
||||
'pass' if action == 'done' else 'fail',
|
||||
decision.get('summary', detail), steps)
|
||||
return
|
||||
|
||||
step_rec = self._record(step_n, decision, result, detail, extra)
|
||||
steps.append(step_rec)
|
||||
|
||||
# 关键步骤截图（截图后统一写日志，避免重复）
|
||||
if result == 'ok' and action in ('click', 'fill', 'assert'):
|
||||
shot = self._screenshot(browser, f'step{step_n:02d}.png')
|
||||
if shot:
|
||||
step_rec['screenshot'] = shot
|
||||
save_step_log(self.task_id, step_rec)
|
||||
|
||||
self._finish(browser, 'error', f'超过最大步数({self.max_steps})未完成', steps)
|
||||
except BrowserError as e:
|
||||
self._log(f'浏览器错误: {e}')
|
||||
self._finish(browser, 'error', f'浏览器错误: {e}', steps)
|
||||
except Exception as e:
|
||||
self._log(f'未知异常: {traceback.format_exc()}')
|
||||
self._finish(browser, 'error', f'异常: {e}', steps)
|
||||
|
||||
def _record(self, n, decision, result, detail='', extra=None):
|
||||
rec = {
|
||||
'n': n,
|
||||
'action': decision.get('action'),
|
||||
'reason': decision.get('reason', ''),
|
||||
'target': decision.get('target', ''),
|
||||
'value': decision.get('value', ''),
|
||||
'assert': decision.get('assert'),
|
||||
'result': result,
|
||||
'detail': detail,
|
||||
'ts': time.strftime('%Y-%m-%d %H:%M:%S'),
|
||||
}
|
||||
if extra:
|
||||
rec.update(extra)
|
||||
return rec
|
||||
|
||||
def _execute(self, browser, decision):
|
||||
"""执行单个动作，带自愈重试。返回 (result, detail, extra)"""
|
||||
action = decision.get('action')
|
||||
target = decision.get('target', '')
|
||||
value = decision.get('value', '')
|
||||
assert_obj = decision.get('assert') or {}
|
||||
|
||||
if action == 'done':
|
||||
return 'ok', decision.get('summary', '完成'), {}
|
||||
if action == 'fail':
|
||||
return 'ok', decision.get('summary', '失败'), {}
|
||||
if action == 'screenshot':
|
||||
shot = self._screenshot(browser, f'shot_{int(time.time())}.png')
|
||||
return ('ok', f'已截图 {shot}', {'screenshot': shot}) if shot else \
|
||||
('error', '截图失败', {})
|
||||
if action == 'assert':
|
||||
return self._do_assert(browser, assert_obj)
|
||||
|
||||
# 带重试的动作
|
||||
last_err = None
|
||||
for attempt in range(MAX_RETRY_SAME_ERROR + 1):
|
||||
if attempt > 0:
|
||||
self._log(f'重试 {attempt}: {action} {target}')
|
||||
try:
|
||||
if action == 'click':
|
||||
browser.click(self._resolve(target))
|
||||
return 'ok', '点击成功', {}
|
||||
elif action == 'fill':
|
||||
browser.fill(self._resolve(target), value)
|
||||
return 'ok', '填入成功', {}
|
||||
elif action == 'select':
|
||||
browser.select(self._resolve(target), value)
|
||||
return 'ok', '选择成功', {}
|
||||
elif action == 'press':
|
||||
browser.press(value)
|
||||
return 'ok', f'按键 {value}', {}
|
||||
elif action == 'wait':
|
||||
if value.startswith('text:'):
|
||||
browser.wait('--text', value[5:])
|
||||
elif value.startswith('url:'):
|
||||
browser.wait('--url', value[4:])
|
||||
elif value.isdigit():
|
||||
browser.wait(value)
|
||||
else:
|
||||
browser.wait(value)
|
||||
return 'ok', f'等待完成', {}
|
||||
elif action == 'scroll':
|
||||
parts = value.split()
|
||||
browser.scroll if hasattr(browser, 'scroll') else None
|
||||
args = parts if parts else ['down', '500']
|
||||
self._run_scroll(browser, args)
|
||||
return 'ok', f'滚动 {value}', {}
|
||||
except BrowserError as e:
|
||||
last_err = str(e)
|
||||
# 失败后重新快照让 LLM 换个策略
|
||||
try:
|
||||
snap = browser.snapshot(interactive=True, compact=True, timeout=40)
|
||||
snap_text = _fmt_snapshot(snap, max_chars=5000)
|
||||
except Exception:
|
||||
snap_text = '(快照失败)'
|
||||
retry_dec = chat_json(
|
||||
[{'role': 'system', 'content': SYSTEM_PROMPT},
|
||||
{'role': 'user', 'content': (
|
||||
f'## 测试目标\n{self.goal}\n\n'
|
||||
f'## 刚才执行失败\n动作: {action} 目标: {target} 值: {value}\n'
|
||||
f'错误: {last_err}\n\n'
|
||||
f'## 当前页面元素\n{snap_text}\n\n'
|
||||
f'请换一种方式完成相同意图，输出下一步动作 JSON。'
|
||||
f'如果确认无法完成，输出 {{"action":"fail","reason":"...","summary":"..."}}'
|
||||
)}],
|
||||
temperature=0.2, max_tokens=400)
|
||||
new_action = retry_dec.get('action', '')
|
||||
if new_action == 'fail':
|
||||
return 'error', f'自愈放弃: {retry_dec.get("summary", last_err)}', {}
|
||||
if new_action in ('click', 'fill', 'select', 'press', 'wait', 'scroll'):
|
||||
# 更新决策重试
|
||||
decision = retry_dec
|
||||
target = decision.get('target', target)
|
||||
value = decision.get('value', value)
|
||||
action = new_action
|
||||
continue
|
||||
last_err = f'自愈输出非法动作: {new_action}'
|
||||
return 'error', f'执行失败: {last_err}', {}
|
||||
|
||||
def _run_scroll(self, browser, args):
|
||||
import subprocess
|
||||
from config import AGENT_BROWSER, NODE_BIN_DIR, XDG_RUNTIME_DIR
|
||||
import os
|
||||
env = os.environ.copy()
|
||||
env['PATH'] = f'{NODE_BIN_DIR}:{env.get("PATH", "")}'
|
||||
env['XDG_RUNTIME_DIR'] = XDG_RUNTIME_DIR
|
||||
subprocess.run([AGENT_BROWSER, '--namespace', f'task-{self.task_id}',
|
||||
'scroll'] + args, capture_output=True, timeout=20, env=env,
|
||||
check=True)
|
||||
|
||||
def _resolve(self, target):
|
||||
"""把目标转成 agent-browser 定位：@ref 原样，text:xxx 转 find 语法"""
|
||||
if not target:
|
||||
raise BrowserError('缺少目标元素')
|
||||
if target.startswith('@'):
|
||||
return target
|
||||
if target.startswith('text:'):
|
||||
return f'find text "{target[5:]}"'
|
||||
if target.startswith('role:'):
|
||||
parts = target[5:].split(':', 1)
|
||||
if len(parts) == 2:
|
||||
return f'find role {parts[0]} --name "{parts[1]}"'
|
||||
return f'find role {parts[0]}'
|
||||
return target
|
||||
|
||||
def _do_assert(self, browser, assert_obj):
|
||||
atype = assert_obj.get('type', '')
|
||||
expect = assert_obj.get('expect', '')
|
||||
present = bool(assert_obj.get('present', True))
|
||||
try:
|
||||
if atype == 'text':
|
||||
found = self._page_has_text(browser, expect)
|
||||
elif atype == 'element':
|
||||
found = self._element_exists(browser, expect)
|
||||
elif atype == 'url':
|
||||
cur = browser.url()
|
||||
found = expect in cur
|
||||
if not found:
|
||||
return ('fail', f'URL 应为包含"{expect}"，实际: {cur}', {})
|
||||
return 'ok', f'URL 包含 "{expect}" (PASS)', {}
|
||||
elif atype == 'title':
|
||||
cur = browser.title()
|
||||
found = expect in cur
|
||||
if not found:
|
||||
return ('fail', f'标题应包含"{expect}"，实际: {cur}', {})
|
||||
return 'ok', f'标题包含 "{expect}" (PASS)', {}
|
||||
else:
|
||||
return 'error', f'未知断言类型: {atype}', {}
|
||||
except BrowserError as e:
|
||||
return 'error', f'断言执行出错: {e}', {}
|
||||
|
||||
if found == present:
|
||||
tag = 'PASS' if present else 'PASS(确认不存在)'
|
||||
return 'ok', f'断言通过: {atype}="{expect}" present={present} ({tag})', {}
|
||||
tag = 'FAIL' if present else 'FAIL(不应出现却出现)'
|
||||
return 'fail', f'断言失败: {atype}="{expect}" present={present} ({tag})', {}
|
||||
|
||||
def _page_has_text(self, browser, text):
|
||||
r = browser.eval_js(
|
||||
f'document.body && document.body.innerText.includes({json_dumps(text)})')
|
||||
if isinstance(r, dict):
|
||||
d = r.get('data', {})
|
||||
val = d.get('value') if isinstance(d, dict) else d
|
||||
return bool(val)
|
||||
return bool(r)
|
||||
|
||||
def _element_exists(self, browser, text):
|
||||
r = browser.eval_js(
|
||||
f'!![...document.querySelectorAll("button,a,input,textarea,[role=button]")]'
|
||||
f'.find(el => (el.innerText||el.value||"").trim().includes({json_dumps(text)}))')
|
||||
if isinstance(r, dict):
|
||||
d = r.get('data', {})
|
||||
val = d.get('value') if isinstance(d, dict) else d
|
||||
return bool(val)
|
||||
return bool(r)
|
||||
|
||||
def _finish(self, browser, result, summary, steps):
|
||||
if browser:
|
||||
browser.close()
|
||||
update_task(self.task_id, status='finished', result=result,
|
||||
finished_at=time.time(), steps=len(steps), summary=summary)
|
||||
report = self._build_report(result, summary, steps)
|
||||
rpath = os.path.join(self.task_dir, 'report.html')
|
||||
with open(rpath, 'w', encoding='utf-8') as f:
|
||||
f.write(report)
|
||||
update_task(self.task_id, report_path=f'/api/tasks/{self.task_id}/report')
|
||||
self._log(f'完成: result={result} summary={summary}')
|
||||
|
||||
def _build_report(self, result, summary, steps):
|
||||
"""生成 HTML 报告"""
|
||||
title_map = {'pass': '✅ 测试通过', 'fail': '❌ 测试失败',
|
||||
'error': '⚠️ 测试错误', 'stopped': '⏹️ 已停止'}
|
||||
color_map = {'pass': '#16a34a', 'fail': '#dc2626',
|
||||
'error': '#d97706', 'stopped': '#6b7280'}
|
||||
rows = []
|
||||
for s in steps:
|
||||
cls = {'ok': 'ok', 'fail': 'bad', 'error': 'bad'}.get(s.get('result'), '')
|
||||
shot = ''
|
||||
if s.get('screenshot'):
|
||||
shot = (f'<div class="shot"><img src="/api/tasks/{self.task_id}/'
|
||||
f'screenshot/{s["screenshot"]}" loading="lazy"></div>')
|
||||
rows.append(f'''<tr class="{cls}">
|
||||
<td>{s.get('n')}</td>
|
||||
<td>{s.get('action')}</td>
|
||||
<td>{s.get('target','')} {s.get('value','')}</td>
|
||||
<td class="reason">{s.get('reason','')}</td>
|
||||
<td>{s.get('result')} {s.get('detail','')}</td>
|
||||
<td>{shot}</td></tr>''')
|
||||
steps_html = '\n'.join(rows) if rows else '<tr><td colspan="6">无步骤记录</td></tr>'
|
||||
dur = ''
|
||||
t = get_task_safe(self.task_id)
|
||||
if t and t.get('started_at') and t.get('finished_at'):
|
||||
dur = f'{t["finished_at"] - t["started_at"]:.1f}s'
|
||||
return f'''<!DOCTYPE html>
|
||||
<html lang="zh-CN"><head><meta charset="utf-8">
|
||||
<title>{title_map.get(result, result)} - webtest-agent</title>
|
||||
<style>
|
||||
body{{font-family:-apple-system,'Segoe UI',sans-serif;margin:0;background:#f3f4f6;color:#111}}
|
||||
.wrap{{max-width:1100px;margin:0 auto;padding:24px}}
|
||||
h1{{font-size:22px}}
|
||||
.badge{{display:inline-block;padding:6px 16px;border-radius:999px;color:#fff;background:{color_map.get(result, '#6b7280')};font-size:14px}}
|
||||
.meta{{color:#666;font-size:13px;margin:8px 0 20px}}
|
||||
.summary{{background:#fff;border-radius:10px;padding:16px;margin-bottom:20px;border:1px solid #e5e7eb}}
|
||||
table{{width:100%;border-collapse:collapse;background:#fff;border-radius:10px;overflow:hidden;font-size:13px}}
|
||||
th,td{{padding:10px 12px;border-bottom:1px solid #f0f0f0;text-align:left;vertical-align:top}}
|
||||
th{{background:#f9fafb;font-weight:600}}
|
||||
tr.bad td{{background:#fef2f2}}
|
||||
tr.ok td{{background:#f0fdf4}}
|
||||
.reason{{color:#555;max-width:260px}}
|
||||
.shot img{{max-width:260px;border-radius:6px;border:1px solid #e5e7eb;display:block}}
|
||||
</style></head><body><div class="wrap">
|
||||
<h1>网页测试报告 <span class="badge">{title_map.get(result, result)}</span></h1>
|
||||
<div class="meta">任务: {self.task_id} | 目标: {self.goal} | 步骤数: {len(steps)} | 耗时: {dur}</div>
|
||||
<div class="summary"><b>结论：</b>{summary}</div>
|
||||
<table><thead><tr><th>#</th><th>动作</th><th>目标</th><th>原因</th><th>结果</th><th>截图</th></tr></thead>
|
||||
<tbody>{steps_html}</tbody></table>
|
||||
</div></body></html>'''
|
||||
|
||||
|
||||
def json_dumps(s):
|
||||
import json
|
||||
return json.dumps(str(s))
|
||||
|
||||
|
||||
def get_task_safe(tid):
|
||||
from db import get_task
|
||||
try:
|
||||
return get_task(tid)
|
||||
except Exception:
|
||||
return None
|
||||
@@ -0,0 +1,145 @@
|
||||
#!/usr/bin/env python3
|
||||
"""webtest-agent - AI 网页测试智能体 (Flask 服务)"""
|
||||
import os
|
||||
import threading
|
||||
|
||||
from flask import Flask, request, jsonify, send_from_directory, abort
|
||||
from flask_cors import CORS
|
||||
|
||||
import config
|
||||
from db import init_db, create_task, get_task, list_tasks, load_step_logs, update_task
|
||||
from agent import TaskRunner
|
||||
|
||||
app = Flask(__name__, static_folder='static', static_url_path='')
|
||||
CORS(app)
|
||||
|
||||
init_db()
|
||||
|
||||
_runner_lock = threading.Lock()
|
||||
_runners = {} # task_id -> TaskRunner
|
||||
_semaphore = threading.Semaphore(config.MAX_CONCURRENT_TASKS)
|
||||
|
||||
|
||||
@app.route('/')
|
||||
def index():
|
||||
return send_from_directory('static', 'index.html')
|
||||
|
||||
|
||||
@app.route('/health')
|
||||
def health():
|
||||
return jsonify({'status': 'ok', 'version': '1.0.0',
|
||||
'llm_model': config.LLM_MODEL,
|
||||
'concurrent': len(_runners)})
|
||||
|
||||
|
||||
@app.route('/api/tasks', methods=['POST'])
|
||||
def api_create_task():
|
||||
data = request.get_json(silent=True) or {}
|
||||
url = (data.get('url') or '').strip()
|
||||
goal = (data.get('goal') or '').strip()
|
||||
if not url:
|
||||
return jsonify({'error': '缺少 url 参数'}), 400
|
||||
if not goal:
|
||||
return jsonify({'error': '缺少 goal（测试目标）参数'}), 400
|
||||
if not url.startswith(('http://', 'https://')):
|
||||
url = 'https://' + url
|
||||
max_steps = int(data.get('max_steps') or config.DEFAULT_MAX_STEPS)
|
||||
timeout = int(data.get('timeout') or config.DEFAULT_TIMEOUT)
|
||||
max_steps = max(1, min(max_steps, 100))
|
||||
timeout = max(30, min(timeout, 3600))
|
||||
|
||||
tid = create_task(url, goal, max_steps, timeout)
|
||||
|
||||
def _launch():
|
||||
with _semaphore:
|
||||
if get_task(tid) and get_task(tid).get('status') == 'stopped':
|
||||
return
|
||||
runner = TaskRunner(tid, url, goal, max_steps, timeout)
|
||||
with _runner_lock:
|
||||
_runners[tid] = runner
|
||||
runner.start()
|
||||
runner.join()
|
||||
with _runner_lock:
|
||||
_runners.pop(tid, None)
|
||||
|
||||
threading.Thread(target=_launch, daemon=True).start()
|
||||
return jsonify({'task_id': tid, 'status': 'queued',
|
||||
'url': url, 'goal': goal}), 202
|
||||
|
||||
|
||||
@app.route('/api/tasks', methods=['GET'])
|
||||
def api_list_tasks():
|
||||
tasks = list_tasks(limit=50)
|
||||
for t in tasks:
|
||||
t['created'] = _fmt_time(t.get('created_at'))
|
||||
t['finished'] = _fmt_time(t.get('finished_at'))
|
||||
return jsonify({'tasks': tasks})
|
||||
|
||||
|
||||
@app.route('/api/tasks/<tid>', methods=['GET'])
|
||||
def api_get_task(tid):
|
||||
t = get_task(tid)
|
||||
if not t:
|
||||
abort(404)
|
||||
t['created'] = _fmt_time(t.get('created_at'))
|
||||
t['finished'] = _fmt_time(t.get('finished_at'))
|
||||
t['steps_log'] = load_step_logs(tid)
|
||||
return jsonify(t)
|
||||
|
||||
|
||||
@app.route('/api/tasks/<tid>/stop', methods=['POST'])
|
||||
def api_stop_task(tid):
|
||||
runner = _runners.get(tid)
|
||||
if runner:
|
||||
runner.stop()
|
||||
update_task(tid, status='stopped', result='stopped')
|
||||
return jsonify({'ok': True, 'message': '停止请求已发送'})
|
||||
t = get_task(tid)
|
||||
if not t:
|
||||
abort(404)
|
||||
if t['status'] == 'queued':
|
||||
update_task(tid, status='stopped', result='stopped')
|
||||
return jsonify({'ok': True, 'message': '已取消排队任务'})
|
||||
return jsonify({'ok': False, 'message': '任务不在运行中'})
|
||||
|
||||
|
||||
@app.route('/api/tasks/<tid>/report')
|
||||
def api_report(tid):
|
||||
path = os.path.join(config.TASKS_DIR, tid, 'report.html')
|
||||
if os.path.exists(path):
|
||||
return send_from_directory(config.TASKS_DIR, f'{tid}/report.html')
|
||||
abort(404)
|
||||
|
||||
|
||||
@app.route('/api/tasks/<tid>/report.json')
|
||||
def api_report_json(tid):
|
||||
t = get_task(tid)
|
||||
if not t:
|
||||
abort(404)
|
||||
return jsonify({
|
||||
'task': t,
|
||||
'steps': load_step_logs(tid),
|
||||
'report_url': f'/api/tasks/{tid}/report',
|
||||
})
|
||||
|
||||
|
||||
@app.route('/api/tasks/<tid>/screenshot/<name>')
|
||||
def api_screenshot(tid, name):
|
||||
if not name or '..' in name or '/' in name:
|
||||
abort(400)
|
||||
path = os.path.join(config.TASKS_DIR, tid, name)
|
||||
if os.path.exists(path):
|
||||
return send_from_directory(config.TASKS_DIR, f'{tid}/{name}')
|
||||
abort(404)
|
||||
|
||||
|
||||
def _fmt_time(ts):
|
||||
if not ts:
|
||||
return None
|
||||
import time
|
||||
return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(ts))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
print(f'webtest-agent 启动: http://0.0.0.0:{config.PORT}')
|
||||
app.run(host=config.HOST, port=config.PORT, threaded=True)
|
||||
+125
@@ -0,0 +1,125 @@
|
||||
#!/usr/bin/env python3
|
||||
"""agent-browser CLI 封装"""
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
|
||||
from config import AGENT_BROWSER, NODE_BIN_DIR, XDG_RUNTIME_DIR
|
||||
|
||||
|
||||
class BrowserError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class AgentBrowser:
|
||||
"""每个任务实例一个 namespace，避免 socket 冲突"""
|
||||
|
||||
def __init__(self, namespace='default'):
|
||||
self.namespace = namespace
|
||||
self.env = os.environ.copy()
|
||||
self.env['PATH'] = f'{NODE_BIN_DIR}:{self.env.get("PATH", "")}'
|
||||
self.env['XDG_RUNTIME_DIR'] = XDG_RUNTIME_DIR
|
||||
os.makedirs(XDG_RUNTIME_DIR, exist_ok=True)
|
||||
try:
|
||||
os.chmod(XDG_RUNTIME_DIR, 0o700)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def _run(self, args, timeout=60, check=False):
|
||||
cmd = [AGENT_BROWSER, '--namespace', self.namespace] + args
|
||||
try:
|
||||
p = subprocess.run(cmd, capture_output=True, text=True,
|
||||
timeout=timeout, env=self.env)
|
||||
except subprocess.TimeoutExpired:
|
||||
raise BrowserError(f'命令超时: {" ".join(args)}')
|
||||
out = p.stdout.strip()
|
||||
if check and p.returncode != 0:
|
||||
err = (p.stderr or '').strip() or out
|
||||
raise BrowserError(f'命令失败({p.returncode}): {err[:300]}')
|
||||
if out.startswith('{'):
|
||||
try:
|
||||
return json.loads(out)
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
return out
|
||||
|
||||
def open(self, url, timeout=60):
|
||||
return self._run(['open', url], timeout=timeout, check=True)
|
||||
|
||||
def snapshot(self, interactive=True, compact=False, depth=None, timeout=60):
|
||||
args = ['snapshot']
|
||||
if interactive:
|
||||
args.append('-i')
|
||||
if compact:
|
||||
args.append('-c')
|
||||
if depth:
|
||||
args += ['-d', str(depth)]
|
||||
args.append('--json')
|
||||
data = self._run(args, timeout=timeout, check=True)
|
||||
if isinstance(data, dict) and data.get('success'):
|
||||
return data.get('data') or {}
|
||||
return data
|
||||
|
||||
def click(self, target, timeout=30):
|
||||
return self._run(['click', target], timeout=timeout, check=True)
|
||||
|
||||
def fill(self, target, value, timeout=30):
|
||||
return self._run(['fill', target, value], timeout=timeout, check=True)
|
||||
|
||||
def select(self, target, value, timeout=30):
|
||||
return self._run(['select', target, value], timeout=timeout, check=True)
|
||||
|
||||
def press(self, key, timeout=30):
|
||||
return self._run(['press', key], timeout=timeout, check=True)
|
||||
|
||||
def wait(self, *args, timeout=60):
|
||||
return self._run(['wait'] + list(args), timeout=timeout, check=True)
|
||||
|
||||
def get(self, what, target=None, timeout=30):
|
||||
args = ['get', what]
|
||||
if target:
|
||||
args.append(target)
|
||||
args.append('--json')
|
||||
return self._run(args, timeout=timeout, check=True)
|
||||
|
||||
def eval_js(self, expr, timeout=30):
|
||||
return self._run(['eval', expr], timeout=timeout, check=True)
|
||||
|
||||
def screenshot(self, path, full=False, timeout=30):
|
||||
args = ['screenshot', path]
|
||||
if full:
|
||||
args.append('--full')
|
||||
return self._run(args, timeout=timeout, check=True)
|
||||
|
||||
def is_visible(self, target, timeout=30):
|
||||
try:
|
||||
r = self._run(['is', 'visible', target, '--json'], timeout=timeout)
|
||||
return bool(r.get('data', {}).get('visible')) if isinstance(r, dict) else False
|
||||
except BrowserError:
|
||||
return False
|
||||
|
||||
def url(self, timeout=20):
|
||||
try:
|
||||
r = self.get('url')
|
||||
if isinstance(r, dict):
|
||||
d = r.get('data', {})
|
||||
return d.get('url') or d.get('value') or ''
|
||||
return ''
|
||||
except BrowserError:
|
||||
return ''
|
||||
|
||||
def title(self, timeout=20):
|
||||
try:
|
||||
r = self.get('title')
|
||||
if isinstance(r, dict):
|
||||
d = r.get('data', {})
|
||||
return d.get('title') or d.get('value') or ''
|
||||
return ''
|
||||
except BrowserError:
|
||||
return ''
|
||||
|
||||
def close(self):
|
||||
try:
|
||||
self._run(['close'], timeout=15)
|
||||
except Exception:
|
||||
pass
|
||||
@@ -0,0 +1,39 @@
|
||||
#!/usr/bin/env python3
|
||||
"""webtest-agent 配置文件"""
|
||||
import os
|
||||
|
||||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
DATA_DIR = os.path.join(BASE_DIR, 'data')
|
||||
TASKS_DIR = os.path.join(DATA_DIR, 'tasks')
|
||||
DB_PATH = os.path.join(DATA_DIR, 'webtest.db')
|
||||
LOG_DIR = os.path.join(BASE_DIR, 'logs')
|
||||
|
||||
# agent-browser CLI 路径
|
||||
AGENT_BROWSER = os.environ.get(
|
||||
'AGENT_BROWSER',
|
||||
'/home/openclaw/.nvm/versions/node/v24.15.0/bin/agent-browser'
|
||||
)
|
||||
NODE_BIN_DIR = os.path.dirname(AGENT_BROWSER)
|
||||
# agent-browser 需要可写的 socket 目录，固定用 /tmp 下的（系统 XDG_RUNTIME_DIR 可能属于其他用户）
|
||||
XDG_RUNTIME_DIR = '/tmp/xdg-rt'
|
||||
|
||||
# LLM 配置（OpenAI 兼容接口）
|
||||
LLM_BASE_URL = os.environ.get('LLM_BASE_URL', 'https://ark.cn-beijing.volces.com/api/plan/v3')
|
||||
LLM_API_KEY = os.environ.get('LLM_API_KEY', 'ark-2b06dc9d-8878-4c6e-b201-f422376e79cb-246d9')
|
||||
LLM_MODEL = os.environ.get('LLM_MODEL', 'doubao-seed-evolving')
|
||||
LLM_TEMPERATURE = 0.2
|
||||
LLM_TIMEOUT = 120
|
||||
|
||||
# Agent 默认参数
|
||||
DEFAULT_MAX_STEPS = 30 # 最大动作步数
|
||||
DEFAULT_TIMEOUT = 600 # 任务总超时（秒）
|
||||
DEFAULT_MAX_SNAPSHOT_CHARS = 9000 # 快照截断长度
|
||||
MAX_RETRY_SAME_ERROR = 2 # 相同错误最多重试次数
|
||||
MAX_CONCURRENT_TASKS = 2 # 最大并发任务数
|
||||
HISTORY_LIMIT = 16 # 喂给 LLM 的最近历史步数
|
||||
|
||||
HOST = '0.0.0.0'
|
||||
PORT = int(os.environ.get('PORT', '16051'))
|
||||
|
||||
for d in (DATA_DIR, TASKS_DIR, LOG_DIR):
|
||||
os.makedirs(d, exist_ok=True)
|
||||
@@ -0,0 +1,106 @@
|
||||
#!/usr/bin/env python3
|
||||
"""SQLite 任务持久化"""
|
||||
import json
|
||||
import sqlite3
|
||||
import time
|
||||
import uuid
|
||||
|
||||
from config import DB_PATH
|
||||
|
||||
|
||||
def _conn():
|
||||
c = sqlite3.connect(DB_PATH, timeout=30)
|
||||
c.row_factory = sqlite3.Row
|
||||
return c
|
||||
|
||||
|
||||
def init_db():
|
||||
c = _conn()
|
||||
c.execute('''
|
||||
CREATE TABLE IF NOT EXISTS tasks (
|
||||
id TEXT PRIMARY KEY,
|
||||
url TEXT NOT NULL,
|
||||
goal TEXT NOT NULL,
|
||||
status TEXT DEFAULT 'queued',
|
||||
result TEXT DEFAULT 'pending',
|
||||
max_steps INTEGER DEFAULT 30,
|
||||
timeout INTEGER DEFAULT 600,
|
||||
created_at REAL,
|
||||
started_at REAL,
|
||||
finished_at REAL,
|
||||
steps INTEGER DEFAULT 0,
|
||||
summary TEXT,
|
||||
error TEXT,
|
||||
report_path TEXT
|
||||
)
|
||||
''')
|
||||
c.commit()
|
||||
c.close()
|
||||
|
||||
|
||||
def create_task(url, goal, max_steps, timeout):
|
||||
tid = uuid.uuid4().hex[:12]
|
||||
c = _conn()
|
||||
c.execute(
|
||||
'INSERT INTO tasks (id, url, goal, status, max_steps, timeout, created_at) '
|
||||
'VALUES (?,?,?,?,?,?,?)',
|
||||
(tid, url, goal, 'queued', max_steps, timeout, time.time()))
|
||||
c.commit()
|
||||
c.close()
|
||||
return tid
|
||||
|
||||
|
||||
def update_task(tid, **fields):
|
||||
allowed = {'status', 'result', 'started_at', 'finished_at', 'steps',
|
||||
'summary', 'error', 'report_path'}
|
||||
sets = [f'{k}=?' for k in fields if k in allowed]
|
||||
vals = [fields[k] for k in fields if k in allowed]
|
||||
if not sets:
|
||||
return
|
||||
c = _conn()
|
||||
c.execute(f'UPDATE tasks SET {", ".join(sets)} WHERE id=?', (*vals, tid))
|
||||
c.commit()
|
||||
c.close()
|
||||
|
||||
|
||||
def get_task(tid):
|
||||
c = _conn()
|
||||
row = c.execute('SELECT * FROM tasks WHERE id=?', (tid,)).fetchone()
|
||||
c.close()
|
||||
return dict(row) if row else None
|
||||
|
||||
|
||||
def list_tasks(limit=50):
|
||||
c = _conn()
|
||||
rows = c.execute(
|
||||
'SELECT * FROM tasks ORDER BY created_at DESC LIMIT ?', (limit,)
|
||||
).fetchall()
|
||||
c.close()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def save_step_log(tid, step):
|
||||
"""把单个步骤 JSON 追加到任务目录的 steps.jsonl"""
|
||||
from config import TASKS_DIR
|
||||
import os
|
||||
p = os.path.join(TASKS_DIR, tid, 'steps.jsonl')
|
||||
with open(p, 'a', encoding='utf-8') as f:
|
||||
f.write(json.dumps(step, ensure_ascii=False) + '\n')
|
||||
|
||||
|
||||
def load_step_logs(tid):
|
||||
from config import TASKS_DIR
|
||||
import os
|
||||
p = os.path.join(TASKS_DIR, tid, 'steps.jsonl')
|
||||
if not os.path.exists(p):
|
||||
return []
|
||||
out = []
|
||||
with open(p, encoding='utf-8') as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if line:
|
||||
try:
|
||||
out.append(json.loads(line))
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
return out
|
||||
@@ -0,0 +1,89 @@
|
||||
#!/usr/bin/env python3
|
||||
"""LLM 客户端（OpenAI 兼容接口）"""
|
||||
import json
|
||||
import re
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
|
||||
from config import LLM_BASE_URL, LLM_API_KEY, LLM_MODEL, LLM_TEMPERATURE, LLM_TIMEOUT
|
||||
|
||||
|
||||
class LLMError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def _extract_json(text):
|
||||
"""从 LLM 输出中提取 JSON（容忍 markdown 代码块等包裹）"""
|
||||
if not text:
|
||||
return None
|
||||
text = text.strip()
|
||||
# 去掉 markdown 代码块
|
||||
fence = re.search(r'```(?:json)?\s*(.*?)```', text, re.S)
|
||||
if fence:
|
||||
text = fence.group(1).strip()
|
||||
try:
|
||||
return json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
# 尝试截取第一个 { 到最后一个 }
|
||||
s, e = text.find('{'), text.rfind('}')
|
||||
if s != -1 and e > s:
|
||||
try:
|
||||
return json.loads(text[s:e + 1])
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def chat(messages, temperature=None, max_tokens=None, timeout=None):
|
||||
"""调用 OpenAI 兼容 chat/completions，返回 content 字符串"""
|
||||
url = f'{LLM_BASE_URL}/chat/completions'
|
||||
body = {
|
||||
'model': LLM_MODEL,
|
||||
'messages': messages,
|
||||
'temperature': temperature if temperature is not None else LLM_TEMPERATURE,
|
||||
}
|
||||
if max_tokens:
|
||||
body['max_tokens'] = max_tokens
|
||||
req = urllib.request.Request(
|
||||
url,
|
||||
data=json.dumps(body).encode('utf-8'),
|
||||
headers={
|
||||
'Content-Type': 'application/json',
|
||||
'Authorization': f'Bearer {LLM_API_KEY}',
|
||||
},
|
||||
method='POST',
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=timeout or LLM_TIMEOUT) as resp:
|
||||
data = json.loads(resp.read().decode('utf-8'))
|
||||
except urllib.error.HTTPError as e:
|
||||
detail = e.read().decode('utf-8', 'ignore')[:300]
|
||||
raise LLMError(f'LLM HTTP {e.code}: {detail}')
|
||||
except Exception as e:
|
||||
raise LLMError(f'LLM 调用失败: {e}')
|
||||
|
||||
try:
|
||||
return data['choices'][0]['message']['content']
|
||||
except (KeyError, IndexError, TypeError):
|
||||
raise LLMError(f'LLM 响应异常: {str(data)[:300]}')
|
||||
|
||||
|
||||
def chat_json(messages, temperature=None, max_tokens=None, retries=2):
|
||||
"""调用 LLM 并强制解析 JSON，失败重试"""
|
||||
last_err = None
|
||||
for i in range(retries + 1):
|
||||
try:
|
||||
content = chat(messages, temperature=temperature, max_tokens=max_tokens)
|
||||
obj = _extract_json(content)
|
||||
if obj is not None:
|
||||
return obj
|
||||
last_err = f'无法从输出解析 JSON: {content[:200]}'
|
||||
except LLMError as e:
|
||||
last_err = str(e)
|
||||
if i < retries:
|
||||
messages = messages + [
|
||||
{'role': 'assistant', 'content': content if 'content' in dir() else ''},
|
||||
{'role': 'user', 'content': f'刚才的输出不是合法 JSON，请只输出严格的 JSON 对象。错误: {last_err}'},
|
||||
]
|
||||
raise LLMError(f'LLM JSON 解析失败: {last_err}')
|
||||
@@ -0,0 +1,2 @@
|
||||
flask>=2.3
|
||||
flask-cors>=4.0
|
||||
@@ -0,0 +1,24 @@
|
||||
#!/bin/bash
|
||||
# webtest-agent 启动脚本
|
||||
cd "$(dirname "$0")"
|
||||
export PATH="/home/openclaw/.nvm/versions/node/v24.15.0/bin:$PATH"
|
||||
export XDG_RUNTIME_DIR=/tmp/xdg-rt
|
||||
mkdir -p "$XDG_RUNTIME_DIR" && chmod 700 "$XDG_RUNTIME_DIR"
|
||||
mkdir -p logs
|
||||
|
||||
if [ -n "$1" ] && [ "$1" = "stop" ]; then
|
||||
pkill -f "webtest-agent/app.py" && echo "已停止" || echo "未在运行"
|
||||
exit 0
|
||||
fi
|
||||
|
||||
# 端口占用检查
|
||||
if ss -tlnp 2>/dev/null | grep -q ":16051 "; then
|
||||
echo "⚠️ 端口 16051 已被占用，无法启动！"
|
||||
ss -tlnp | grep ":16051 "
|
||||
exit 1
|
||||
fi
|
||||
|
||||
nohup /home/hz1/miniconda3/envs/openclaw/bin/python3 app.py > logs/app.log 2>&1 &
|
||||
echo "✅ webtest-agent 已启动 (PID $!)"
|
||||
sleep 1
|
||||
curl -s http://localhost:16051/health && echo
|
||||
@@ -0,0 +1,47 @@
|
||||
* { box-sizing: border-box; margin: 0; padding: 0; }
|
||||
body { font-family: -apple-system, 'Segoe UI', 'PingFang SC', 'Microsoft YaHei', sans-serif; background: #f3f4f6; color: #111; }
|
||||
.topbar { background: #111827; color: #fff; padding: 14px 24px; display: flex; justify-content: space-between; align-items: center; }
|
||||
.brand { font-size: 17px; font-weight: 600; }
|
||||
.ver { font-size: 11px; background: #374151; padding: 2px 8px; border-radius: 999px; margin-left: 6px; font-weight: 400; }
|
||||
.health { font-size: 12px; color: #9ca3af; }
|
||||
.health.ok { color: #4ade80; }
|
||||
.wrap { max-width: 1200px; margin: 0 auto; padding: 24px; }
|
||||
.card { background: #fff; border-radius: 12px; padding: 20px; margin-bottom: 20px; border: 1px solid #e5e7eb; }
|
||||
.card h2 { font-size: 16px; margin-bottom: 14px; display: flex; justify-content: space-between; align-items: center; }
|
||||
.form-grid { display: grid; grid-template-columns: 1fr 1fr; gap: 14px; margin-bottom: 16px; }
|
||||
.field { display: flex; flex-direction: column; }
|
||||
.field.span2 { grid-column: span 2; }
|
||||
.field label { font-size: 13px; color: #374151; margin-bottom: 6px; font-weight: 500; }
|
||||
.field input, .field textarea { padding: 10px 12px; border: 1px solid #d1d5db; border-radius: 8px; font-size: 14px; font-family: inherit; }
|
||||
.field input:focus, .field textarea:focus { outline: none; border-color: #2563eb; box-shadow: 0 0 0 3px rgba(37,99,235,.1); }
|
||||
.btn { padding: 10px 20px; border: none; border-radius: 8px; cursor: pointer; font-size: 14px; font-weight: 500; }
|
||||
.btn.primary { background: #2563eb; color: #fff; }
|
||||
.btn.primary:hover { background: #1d4ed8; }
|
||||
.btn.primary:disabled { background: #93c5fd; cursor: not-allowed; }
|
||||
.btn.small { padding: 5px 12px; font-size: 12px; background: #f3f4f6; color: #374151; border: 1px solid #d1d5db; }
|
||||
.btn.small:hover { background: #e5e7eb; }
|
||||
.hint { margin-top: 12px; font-size: 12px; color: #6b7280; }
|
||||
table { width: 100%; border-collapse: collapse; font-size: 13px; }
|
||||
th, td { padding: 10px 12px; border-bottom: 1px solid #f0f0f0; text-align: left; vertical-align: middle; }
|
||||
th { background: #f9fafb; font-weight: 600; color: #374151; white-space: nowrap; }
|
||||
td { word-break: break-all; }
|
||||
td .goal-cell { max-width: 260px; display: -webkit-box; -webkit-line-clamp: 2; -webkit-box-orient: vertical; overflow: hidden; }
|
||||
.empty { text-align: center; color: #9ca3af; padding: 24px; }
|
||||
.status { padding: 3px 10px; border-radius: 999px; font-size: 12px; font-weight: 500; }
|
||||
.status.queued { background: #fef3c7; color: #92400e; }
|
||||
.status.running { background: #dbeafe; color: #1e40af; }
|
||||
.status.finished { background: #dcfce7; color: #166534; }
|
||||
.status.stopped { background: #f3f4f6; color: #4b5563; }
|
||||
.result { padding: 3px 10px; border-radius: 999px; font-size: 12px; font-weight: 500; }
|
||||
.result.pass { background: #dcfce7; color: #166534; }
|
||||
.result.fail { background: #fee2e2; color: #991b1b; }
|
||||
.result.error { background: #ffedd5; color: #9a3412; }
|
||||
.result.stopped { background: #f3f4f6; color: #4b5563; }
|
||||
.result.pending { background: #f3f4f6; color: #9ca3af; }
|
||||
a.report-link { color: #2563eb; text-decoration: none; }
|
||||
a.report-link:hover { text-decoration: underline; }
|
||||
.modal { position: fixed; inset: 0; background: rgba(0,0,0,.5); z-index: 100; display: flex; align-items: center; justify-content: center; }
|
||||
.modal.hidden { display: none; }
|
||||
.modal-body { width: 90vw; height: 90vh; background: #fff; border-radius: 12px; display: flex; flex-direction: column; overflow: hidden; }
|
||||
.modal-head { padding: 12px 16px; border-bottom: 1px solid #e5e7eb; display: flex; justify-content: space-between; align-items: center; font-weight: 600; }
|
||||
#report-frame { flex: 1; border: none; width: 100%; }
|
||||
@@ -0,0 +1,58 @@
|
||||
<!DOCTYPE html>
|
||||
<html lang="zh-CN">
|
||||
<head>
|
||||
<meta charset="UTF-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||||
<title>AI 网页测试智能体</title>
|
||||
<link rel="stylesheet" href="/static/css/style.css">
|
||||
</head>
|
||||
<body>
|
||||
<header class="topbar">
|
||||
<div class="brand">🌐 AI 网页测试智能体 <span class="ver">v1.0.0</span></div>
|
||||
<div class="health" id="health">检测服务中...</div>
|
||||
</header>
|
||||
|
||||
<main class="wrap">
|
||||
<section class="card create-card">
|
||||
<h2>发起测试任务</h2>
|
||||
<div class="form-grid">
|
||||
<div class="field span2">
|
||||
<label>目标网址</label>
|
||||
<input id="url" type="text" placeholder="https://example.com" value="https://example.com">
|
||||
</div>
|
||||
<div class="field span2">
|
||||
<label>测试目标（自然语言描述要测什么）</label>
|
||||
<textarea id="goal" rows="3" placeholder="例如：打开页面后，验证标题为 Example Domain，并点击 Learn more 链接能跳转"></textarea>
|
||||
</div>
|
||||
<div class="field">
|
||||
<label>最大步数</label>
|
||||
<input id="max_steps" type="number" value="30" min="1" max="100">
|
||||
</div>
|
||||
<div class="field">
|
||||
<label>超时（秒）</label>
|
||||
<input id="timeout" type="number" value="600" min="30" max="3600">
|
||||
</div>
|
||||
</div>
|
||||
<button id="submit" class="btn primary">🚀 开始测试</button>
|
||||
<div class="hint">测试由 AI 自动驱动浏览器执行：打开页面 → 逐步操作 → 断言验证 → 生成报告（含截图）</div>
|
||||
</section>
|
||||
|
||||
<section class="card">
|
||||
<h2>任务列表 <button id="refresh" class="btn small">刷新</button></h2>
|
||||
<table id="task-table">
|
||||
<thead><tr><th>任务ID</th><th>目标网址</th><th>测试目标</th><th>状态</th><th>结果</th><th>步骤</th><th>创建时间</th><th>操作</th></tr></thead>
|
||||
<tbody><tr><td colspan="8" class="empty">加载中...</td></tr></tbody>
|
||||
</table>
|
||||
</section>
|
||||
</main>
|
||||
|
||||
<div id="report-modal" class="modal hidden">
|
||||
<div class="modal-body">
|
||||
<div class="modal-head"><span id="modal-title">测试报告</span><button id="modal-close" class="btn small">关闭</button></div>
|
||||
<iframe id="report-frame" src="about:blank"></iframe>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<script src="/static/js/app.js"></script>
|
||||
</body>
|
||||
</html>
|
||||
@@ -0,0 +1,122 @@
|
||||
/* webtest-agent 前端逻辑 */
|
||||
const $ = s => document.querySelector(s);
|
||||
const API = '';
|
||||
|
||||
async function refreshHealth() {
|
||||
try {
|
||||
const r = await fetch(API + '/health');
|
||||
const d = await r.json();
|
||||
$('#health').textContent = `服务正常 | 模型: ${d.llm_model} | 运行中任务: ${d.concurrent}`;
|
||||
$('#health').classList.add('ok');
|
||||
} catch (e) {
|
||||
$('#health').textContent = '服务异常';
|
||||
$('#health').classList.remove('ok');
|
||||
}
|
||||
}
|
||||
|
||||
const STATUS_MAP = {
|
||||
queued: '排队中', running: '测试中', finished: '已完成', stopped: '已停止'
|
||||
};
|
||||
const RESULT_MAP = {
|
||||
pending: '待定', pass: '✅ 通过', fail: '❌ 失败', error: '⚠️ 错误', stopped: '⏹️ 停止'
|
||||
};
|
||||
|
||||
function esc(s) {
|
||||
return String(s ?? '').replace(/[&<>"']/g, c => (
|
||||
{'&':'&','<':'<','>':'>','"':'"',"'":'''}[c]));
|
||||
}
|
||||
|
||||
async function loadTasks() {
|
||||
try {
|
||||
const r = await fetch(API + '/api/tasks');
|
||||
const d = await r.json();
|
||||
const tb = $('#task-table tbody');
|
||||
if (!d.tasks.length) {
|
||||
tb.innerHTML = '<tr><td colspan="8" class="empty">暂无任务，先发起一个吧</td></tr>';
|
||||
return;
|
||||
}
|
||||
tb.innerHTML = d.tasks.map(t => {
|
||||
const ops = t.status === 'running' || t.status === 'queued'
|
||||
? `<button class="btn small" onclick="stopTask('${t.id}')">停止</button> `
|
||||
: '';
|
||||
const report = t.status === 'finished'
|
||||
? `<a class="report-link" href="#" onclick="openReport('${t.id}');return false;">查看报告</a>`
|
||||
: '—';
|
||||
return `<tr>
|
||||
<td>${esc(t.id)}</td>
|
||||
<td>${esc(t.url)}</td>
|
||||
<td><div class="goal-cell" title="${esc(t.goal)}">${esc(t.goal)}</div></td>
|
||||
<td><span class="status ${esc(t.status)}">${STATUS_MAP[t.status] || esc(t.status)}</span></td>
|
||||
<td><span class="result ${esc(t.result)}">${RESULT_MAP[t.result] || esc(t.result)}</span></td>
|
||||
<td>${t.steps || 0}</td>
|
||||
<td>${esc(t.created)}</td>
|
||||
<td>${ops}${report}</td>
|
||||
</tr>`;
|
||||
}).join('');
|
||||
} catch (e) {
|
||||
$('#task-table tbody').innerHTML = '<tr><td colspan="8" class="empty">加载失败</td></tr>';
|
||||
}
|
||||
}
|
||||
|
||||
async function stopTask(tid) {
|
||||
if (!confirm('确定停止该任务？')) return;
|
||||
await fetch(API + `/api/tasks/${tid}/stop`, { method: 'POST' });
|
||||
setTimeout(loadTasks, 500);
|
||||
}
|
||||
|
||||
function openReport(tid) {
|
||||
$('#report-frame').src = API + `/api/tasks/${tid}/report`;
|
||||
$('#modal-title').textContent = `测试报告 - ${tid}`;
|
||||
$('#report-modal').classList.remove('hidden');
|
||||
}
|
||||
|
||||
$('#modal-close').onclick = () => $('#report-modal').classList.add('hidden');
|
||||
$('#report-modal').onclick = e => { if (e.target === $('#report-modal')) $('#report-modal').classList.add('hidden'); };
|
||||
|
||||
$('#submit').onclick = async () => {
|
||||
const url = $('#url').value.trim();
|
||||
const goal = $('#goal').value.trim();
|
||||
if (!url || !goal) { alert('请填写网址和测试目标'); return; }
|
||||
const btn = $('#submit');
|
||||
btn.disabled = true; btn.textContent = '提交中...';
|
||||
try {
|
||||
const r = await fetch(API + '/api/tasks', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({
|
||||
url, goal,
|
||||
max_steps: parseInt($('#max_steps').value) || 30,
|
||||
timeout: parseInt($('#timeout').value) || 600
|
||||
})
|
||||
});
|
||||
const d = await r.json();
|
||||
if (d.error) { alert('提交失败: ' + d.error); }
|
||||
else {
|
||||
$('#goal').value = '';
|
||||
loadTasks();
|
||||
// 轮询直到该任务结束
|
||||
pollTask(d.task_id);
|
||||
}
|
||||
} catch (e) { alert('提交失败: ' + e); }
|
||||
finally { btn.disabled = false; btn.textContent = '🚀 开始测试'; }
|
||||
};
|
||||
|
||||
function pollTask(tid, count = 0) {
|
||||
if (count > 400) return;
|
||||
setTimeout(async () => {
|
||||
try {
|
||||
const r = await fetch(API + `/api/tasks/${tid}`);
|
||||
const t = await r.json();
|
||||
loadTasks();
|
||||
if (t.status === 'finished' || t.status === 'stopped') return;
|
||||
pollTask(tid, count + 1);
|
||||
} catch (e) { /* ignore */ }
|
||||
}, 3000);
|
||||
}
|
||||
|
||||
$('#refresh').onclick = loadTasks;
|
||||
|
||||
refreshHealth();
|
||||
loadTasks();
|
||||
setInterval(refreshHealth, 30000);
|
||||
setInterval(loadTasks, 5000);
|
||||
Reference in New Issue
Block a user