From 12dc9a299fff611adb6fc7cf3f880c2b10998436 Mon Sep 17 00:00:00 2001 From: hz4th_coder Date: Sun, 9 Aug 2026 22:41:00 +0800 Subject: [PATCH] =?UTF-8?q?webtest-agent=20v1.0.0:=20AI=E7=BD=91=E9=A1=B5?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=E6=99=BA=E8=83=BD=E4=BD=93=EF=BC=88LLM?= =?UTF-8?q?=E5=86=B3=E7=AD=96+agent-browser=E6=89=A7=E8=A1=8C+=E6=96=AD?= =?UTF-8?q?=E8=A8=80+=E6=8A=A5=E5=91=8A=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 6 + README.md | 78 ++++++++ agent.py | 457 +++++++++++++++++++++++++++++++++++++++++++ app.py | 145 ++++++++++++++ browser.py | 125 ++++++++++++ config.py | 39 ++++ db.py | 106 ++++++++++ llm.py | 89 +++++++++ requirements.txt | 2 + start.sh | 24 +++ static/css/style.css | 47 +++++ static/index.html | 58 ++++++ static/js/app.js | 122 ++++++++++++ 13 files changed, 1298 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 agent.py create mode 100644 app.py create mode 100644 browser.py create mode 100644 config.py create mode 100644 db.py create mode 100644 llm.py create mode 100644 requirements.txt create mode 100755 start.sh create mode 100644 static/css/style.css create mode 100644 static/index.html create mode 100644 static/js/app.js diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..1b20c2b --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +*.pyc +logs/ +data/tasks/ +data/*.db +nohup.out diff --git a/README.md b/README.md new file mode 100644 index 0000000..91e77d3 --- /dev/null +++ b/README.md @@ -0,0 +1,78 @@ +# webtest-agent · AI 网页测试智能体 + +基于 **LLM 决策 + 无头浏览器执行** 的端到端网页测试系统。用自然语言描述测试目标，AI 自动打开页面、逐步操作、断言验证、生成带截图的测试报告。 + +## 架构 + +``` +┌─────────────────────────────────────────────┐ +│ HTTP API (Flask) │ +│ POST /api/tasks → 发起测试 │ +│ GET /api/tasks → 任务列表/状态 │ +│ GET /api/tasks//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/ + +# 查看报告 +curl http://localhost:16051/api/tasks//report + +# 停止任务 +curl -X POST http://localhost:16051/api/tasks//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 自动安装） diff --git a/agent.py b/agent.py new file mode 100644 index 0000000..4e411e4 --- /dev/null +++ b/agent.py @@ -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'
') + rows.append(f''' +{s.get('n')} +{s.get('action')} +{s.get('target','')} {s.get('value','')} +{s.get('reason','')} +{s.get('result')} {s.get('detail','')} +{shot}''') + steps_html = '\n'.join(rows) if rows else '无步骤记录' + 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''' + +{title_map.get(result, result)} - webtest-agent +
+

网页测试报告 {title_map.get(result, result)}

+
任务: {self.task_id} | 目标: {self.goal} | 步骤数: {len(steps)} | 耗时: {dur}
+
结论：{summary}
+ +{steps_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 diff --git a/app.py b/app.py new file mode 100644 index 0000000..fa25d8d --- /dev/null +++ b/app.py @@ -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/', 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//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//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//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//screenshot/') +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) diff --git a/browser.py b/browser.py new file mode 100644 index 0000000..d6ddeb6 --- /dev/null +++ b/browser.py @@ -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 diff --git a/config.py b/config.py new file mode 100644 index 0000000..c2b765a --- /dev/null +++ b/config.py @@ -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) diff --git a/db.py b/db.py new file mode 100644 index 0000000..3829ab5 --- /dev/null +++ b/db.py @@ -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 diff --git a/llm.py b/llm.py new file mode 100644 index 0000000..4fb064d --- /dev/null +++ b/llm.py @@ -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}') diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..32a092a --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +flask>=2.3 +flask-cors>=4.0 diff --git a/start.sh b/start.sh new file mode 100755 index 0000000..1267598 --- /dev/null +++ b/start.sh @@ -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 diff --git a/static/css/style.css b/static/css/style.css new file mode 100644 index 0000000..f7cf6b7 --- /dev/null +++ b/static/css/style.css @@ -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%; } diff --git a/static/index.html b/static/index.html new file mode 100644 index 0000000..c568bcc --- /dev/null +++ b/static/index.html @@ -0,0 +1,58 @@ + + + + + +AI 网页测试智能体 + + + +
+
🌐 AI 网页测试智能体 v1.0.0
+
检测服务中...
+
+ +
+
+

发起测试任务

+
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ +
测试由 AI 自动驱动浏览器执行：打开页面 → 逐步操作 → 断言验证 → 生成报告（含截图）
+
+ +
+

任务列表

+ + + +
任务ID目标网址测试目标状态结果步骤创建时间操作
加载中...
+
+
+ + + + + + diff --git a/static/js/app.js b/static/js/app.js new file mode 100644 index 0000000..a6eceb7 --- /dev/null +++ b/static/js/app.js @@ -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 = '暂无任务，先发起一个吧'; + return; + } + tb.innerHTML = d.tasks.map(t => { + const ops = t.status === 'running' || t.status === 'queued' + ? ` ` + : ''; + const report = t.status === 'finished' + ? `查看报告` + : '—'; + return ` + ${esc(t.id)} + ${esc(t.url)} +
${esc(t.goal)}
+ ${STATUS_MAP[t.status] || esc(t.status)} + ${RESULT_MAP[t.result] || esc(t.result)} + ${t.steps || 0} + ${esc(t.created)} + ${ops}${report} + `; + }).join(''); + } catch (e) { + $('#task-table tbody').innerHTML = '加载失败'; + } +} + +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);