146 lines
4.3 KiB
Python
146 lines
4.3 KiB
Python
#!/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)
|