|
|
"""
|
|
|
灵枢智能任务平台 —— Flask 后端服务
|
|
|
数据库: MySQL 8.0 (lingshu)
|
|
|
"""
|
|
|
from flask import Flask, request, jsonify, send_from_directory, g
|
|
|
from flask_cors import CORS
|
|
|
import db
|
|
|
import os
|
|
|
import jwt
|
|
|
import bcrypt
|
|
|
import datetime
|
|
|
import threading
|
|
|
from functools import wraps
|
|
|
import openclaw
|
|
|
|
|
|
app = Flask(__name__, static_folder=os.path.join(os.path.dirname(__file__), '..'))
|
|
|
app.config['JSON_AS_ASCII'] = False
|
|
|
CORS(app)
|
|
|
|
|
|
JWT_SECRET = os.environ.get('LINGSHU_JWT_SECRET', 'lingshu-secret-key-change-in-prod')
|
|
|
JWT_EXPIRE_HOURS = 24 * 7 # 7天
|
|
|
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
|
# JWT 认证装饰器
|
|
|
# ─────────────────────────────────────────────
|
|
|
def require_auth(f):
|
|
|
@wraps(f)
|
|
|
def decorated(*args, **kwargs):
|
|
|
auth_header = request.headers.get('Authorization', '')
|
|
|
if not auth_header.startswith('Bearer '):
|
|
|
return jsonify({'error': '未登录，请先登录'}), 401
|
|
|
token = auth_header[7:]
|
|
|
try:
|
|
|
payload = jwt.decode(token, JWT_SECRET, algorithms=['HS256'])
|
|
|
g.current_user = payload
|
|
|
except jwt.ExpiredSignatureError:
|
|
|
return jsonify({'error': 'Token 已过期，请重新登录'}), 401
|
|
|
except jwt.InvalidTokenError:
|
|
|
return jsonify({'error': 'Token 无效'}), 401
|
|
|
return f(*args, **kwargs)
|
|
|
return decorated
|
|
|
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
|
# 静态文件（托管前端 HTML）
|
|
|
# ─────────────────────────────────────────────
|
|
|
@app.route('/')
|
|
|
def index():
|
|
|
return send_from_directory(app.static_folder, '灵枢智能任务平台.html')
|
|
|
|
|
|
@app.route('/logo.png')
|
|
|
def logo():
|
|
|
return send_from_directory(app.static_folder, 'logo.png')
|
|
|
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
|
# 认证接口
|
|
|
# ─────────────────────────────────────────────
|
|
|
@app.route('/api/auth/login', methods=['POST'])
|
|
|
def login():
|
|
|
data = request.get_json() or {}
|
|
|
username = data.get('username', '').strip()
|
|
|
password = data.get('password', '')
|
|
|
if not username or not password:
|
|
|
return jsonify({'error': '用户名和密码不能为空'}), 400
|
|
|
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("SELECT id, username, display_name, password_hash, avatar_color FROM users WHERE username=%s", (username,))
|
|
|
user = cur.fetchone()
|
|
|
cur.close(); conn.close()
|
|
|
|
|
|
if not user:
|
|
|
return jsonify({'error': '用户名或密码错误'}), 401
|
|
|
|
|
|
# 兼容 bcrypt hash 和明文（初始数据用明文，建议后续全部改为 bcrypt）
|
|
|
ph = user['password_hash']
|
|
|
ok = False
|
|
|
if ph and ph.startswith('$2'):
|
|
|
try:
|
|
|
ok = bcrypt.checkpw(password.encode(), ph.encode())
|
|
|
except Exception:
|
|
|
ok = False
|
|
|
else:
|
|
|
ok = (password == ph)
|
|
|
|
|
|
if not ok:
|
|
|
return jsonify({'error': '用户名或密码错误'}), 401
|
|
|
|
|
|
payload = {
|
|
|
'id': user['id'],
|
|
|
'username': user['username'],
|
|
|
'name': user['display_name'],
|
|
|
'color': user['avatar_color'] or '#3b82f6',
|
|
|
'exp': datetime.datetime.utcnow() + datetime.timedelta(hours=JWT_EXPIRE_HOURS)
|
|
|
}
|
|
|
token = jwt.encode(payload, JWT_SECRET, algorithm='HS256')
|
|
|
return jsonify({
|
|
|
'token': token,
|
|
|
'user': {
|
|
|
'id': user['id'],
|
|
|
'name': user['display_name'],
|
|
|
'username': user['username'],
|
|
|
'color': user['avatar_color'] or '#3b82f6',
|
|
|
'avatar': (user['display_name'] or '?')[0]
|
|
|
}
|
|
|
})
|
|
|
|
|
|
|
|
|
@app.route('/api/auth/me')
|
|
|
@require_auth
|
|
|
def auth_me():
|
|
|
u = g.current_user
|
|
|
return jsonify({
|
|
|
'id': u['id'],
|
|
|
'name': u['name'],
|
|
|
'username': u['username'],
|
|
|
'color': u['color'],
|
|
|
'avatar': (u['name'] or '?')[0]
|
|
|
})
|
|
|
|
|
|
|
|
|
@app.route('/api/auth/register', methods=['POST'])
|
|
|
def register():
|
|
|
"""注册新用户（管理员可用，生产环境可关闭此接口）"""
|
|
|
data = request.get_json() or {}
|
|
|
username = data.get('username', '').strip()
|
|
|
password = data.get('password', '')
|
|
|
display_name = data.get('display_name', username).strip()
|
|
|
if not username or not password:
|
|
|
return jsonify({'error': '用户名和密码不能为空'}), 400
|
|
|
if len(password) < 4:
|
|
|
return jsonify({'error': '密码至少4位'}), 400
|
|
|
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("SELECT id FROM users WHERE username=%s", (username,))
|
|
|
if cur.fetchone():
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify({'error': '用户名已存在'}), 409
|
|
|
|
|
|
hashed = bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
|
|
|
color = data.get('color', '#3b82f6')
|
|
|
cur2 = conn.cursor()
|
|
|
cur2.execute(
|
|
|
"INSERT INTO users (username, display_name, password_hash, avatar_color) VALUES (%s, %s, %s, %s)",
|
|
|
(username, display_name, hashed, color)
|
|
|
)
|
|
|
conn.commit()
|
|
|
new_id = cur2.lastrowid
|
|
|
cur.close(); cur2.close(); conn.close()
|
|
|
|
|
|
# 将新用户加入默认团队（id=1）
|
|
|
try:
|
|
|
conn2 = db.get_conn()
|
|
|
c2 = conn2.cursor()
|
|
|
c2.execute("INSERT IGNORE INTO team_members (team_id, user_id, role) VALUES (1, %s, 'viewer')", (new_id,))
|
|
|
conn2.commit()
|
|
|
c2.close(); conn2.close()
|
|
|
except Exception:
|
|
|
pass
|
|
|
|
|
|
return jsonify({'id': new_id, 'message': '注册成功'}), 201
|
|
|
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
|
# 工作台 KPI
|
|
|
# ─────────────────────────────────────────────
|
|
|
@app.route('/api/dashboard/kpi')
|
|
|
@require_auth
|
|
|
def dashboard_kpi():
|
|
|
owner_id = g.current_user['id']
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
|
|
|
cur.execute("SELECT COUNT(*) AS total, SUM(status='running') AS running FROM tasks WHERE owner_id=%s", (owner_id,))
|
|
|
row = cur.fetchone()
|
|
|
|
|
|
cur.execute("""
|
|
|
SELECT COUNT(*) AS monthly
|
|
|
FROM task_executions e
|
|
|
JOIN tasks t ON e.task_id = t.id
|
|
|
WHERE t.owner_id=%s
|
|
|
AND MONTH(e.started_at)=MONTH(CURDATE())
|
|
|
AND YEAR(e.started_at)=YEAR(CURDATE())
|
|
|
""", (owner_id,))
|
|
|
monthly = cur.fetchone()['monthly']
|
|
|
|
|
|
cur.execute("SELECT COUNT(*) AS alerts FROM alert_notifications WHERE target_user_id=%s AND is_read=0", (owner_id,))
|
|
|
alerts = cur.fetchone()['alerts']
|
|
|
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify({
|
|
|
'running': int(row['running'] or 0),
|
|
|
'total': int(row['total'] or 0),
|
|
|
'monthly_triggers': monthly,
|
|
|
'alerts': alerts
|
|
|
})
|
|
|
|
|
|
|
|
|
@app.route('/api/dashboard/exec_trend')
|
|
|
@require_auth
|
|
|
def exec_trend():
|
|
|
days = int(request.args.get('days', 7))
|
|
|
owner_id = g.current_user['id']
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("""
|
|
|
SELECT DATE(e.started_at) AS day,
|
|
|
SUM(e.status='success') AS success,
|
|
|
SUM(e.status='failed' OR e.status='timeout') AS fail
|
|
|
FROM task_executions e
|
|
|
JOIN tasks t ON e.task_id = t.id
|
|
|
WHERE t.owner_id=%s AND e.started_at >= DATE_SUB(CURDATE(), INTERVAL %s DAY)
|
|
|
GROUP BY DATE(e.started_at)
|
|
|
ORDER BY day ASC
|
|
|
""", (owner_id, days))
|
|
|
rows = cur.fetchall()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify([{'day': str(r['day']), 'success': int(r['success']), 'fail': int(r['fail'])} for r in rows])
|
|
|
|
|
|
|
|
|
@app.route('/api/dashboard/agent_usage')
|
|
|
@require_auth
|
|
|
def agent_usage():
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("SELECT name, usage_count FROM agents WHERE is_public=1 ORDER BY usage_count DESC LIMIT 6")
|
|
|
rows = cur.fetchall()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify([{'name': r['name'], 'value': r['usage_count']} for r in rows])
|
|
|
|
|
|
|
|
|
@app.route('/api/dashboard/status_dist')
|
|
|
@require_auth
|
|
|
def status_dist():
|
|
|
owner_id = g.current_user['id']
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("SELECT status, COUNT(*) AS cnt FROM tasks WHERE owner_id=%s GROUP BY status", (owner_id,))
|
|
|
rows = cur.fetchall()
|
|
|
cur.close(); conn.close()
|
|
|
label_map = {'running': '运行中', 'stopped': '已停止', 'error': '错误', 'draft': '草稿'}
|
|
|
return jsonify([{'name': label_map.get(r['status'], r['status']), 'value': r['cnt']} for r in rows])
|
|
|
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
|
# 任务中心
|
|
|
# ─────────────────────────────────────────────
|
|
|
@app.route('/api/tasks')
|
|
|
@require_auth
|
|
|
def list_tasks():
|
|
|
import json as _json
|
|
|
owner_id = g.current_user['id']
|
|
|
keyword = request.args.get('keyword', '').strip()
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
if keyword:
|
|
|
cur.execute("""
|
|
|
SELECT t.id, t.name, t.cron_expression, t.status,
|
|
|
a.name AS agent, t.trigger_type, t.workflow_json
|
|
|
FROM tasks t LEFT JOIN agents a ON t.agent_id=a.id
|
|
|
WHERE t.owner_id=%s AND t.name LIKE %s
|
|
|
ORDER BY t.updated_at DESC
|
|
|
""", (owner_id, f'%{keyword}%'))
|
|
|
else:
|
|
|
cur.execute("""
|
|
|
SELECT t.id, t.name, t.cron_expression, t.status,
|
|
|
a.name AS agent, t.trigger_type, t.workflow_json
|
|
|
FROM tasks t LEFT JOIN agents a ON t.agent_id=a.id
|
|
|
WHERE t.owner_id=%s
|
|
|
ORDER BY t.updated_at DESC
|
|
|
""", (owner_id,))
|
|
|
rows = cur.fetchall()
|
|
|
cur.close(); conn.close()
|
|
|
result = []
|
|
|
for r in rows:
|
|
|
item = dict(r)
|
|
|
wj = item.pop('workflow_json', None)
|
|
|
try:
|
|
|
extra = _json.loads(wj) if wj else {}
|
|
|
except Exception:
|
|
|
extra = {}
|
|
|
item['description'] = extra.get('description', '')
|
|
|
item['category'] = extra.get('category', '')
|
|
|
item['skill_name'] = extra.get('skill_name', '')
|
|
|
item['timeout'] = extra.get('timeout', 300)
|
|
|
item['retry_count'] = extra.get('retry_count', 3)
|
|
|
result.append(item)
|
|
|
return jsonify(result)
|
|
|
|
|
|
|
|
|
@app.route('/api/tasks/<int:task_id>', methods=['GET'])
|
|
|
@require_auth
|
|
|
def get_task(task_id):
|
|
|
import json as _json
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("""
|
|
|
SELECT t.*, a.name AS agent_name
|
|
|
FROM tasks t LEFT JOIN agents a ON t.agent_id=a.id
|
|
|
WHERE t.id=%s
|
|
|
""", (task_id,))
|
|
|
row = cur.fetchone()
|
|
|
cur.close(); conn.close()
|
|
|
if not row:
|
|
|
return jsonify({'error': 'not found'}), 404
|
|
|
wj = row.pop('workflow_json', None)
|
|
|
try:
|
|
|
extra = _json.loads(wj) if wj else {}
|
|
|
except Exception:
|
|
|
extra = {}
|
|
|
row['description'] = extra.get('description', '')
|
|
|
row['category'] = extra.get('category', '')
|
|
|
row['skill_name'] = extra.get('skill_name', '')
|
|
|
row['skill_params'] = extra.get('skill_params', {})
|
|
|
row['timeout'] = extra.get('timeout', 300)
|
|
|
row['retry_count'] = extra.get('retry_count', 3)
|
|
|
return jsonify(db.serialize_row(row))
|
|
|
|
|
|
|
|
|
@app.route('/api/tasks', methods=['POST'])
|
|
|
@require_auth
|
|
|
def create_task():
|
|
|
import json as _json
|
|
|
data = request.get_json() or {}
|
|
|
owner_id = g.current_user['id']
|
|
|
name = data.get('name', '').strip()
|
|
|
if not name:
|
|
|
return jsonify({'error': '任务名称不能为空'}), 400
|
|
|
|
|
|
# 触发方式：有 cron_expression 且不为空则为 cron，否则 manual
|
|
|
cron_expr = data.get('cron_expression', '').strip()
|
|
|
trigger_type = 'cron' if cron_expr else 'manual'
|
|
|
|
|
|
# 扩展字段（先用 task_config 的 JSON 字段存储，待数据库结构扩展后迁移）
|
|
|
extra = {
|
|
|
'description': data.get('description', ''),
|
|
|
'category': data.get('category', ''),
|
|
|
'skill_name': data.get('skill_name', ''),
|
|
|
'skill_params': data.get('skill_params', {}),
|
|
|
'datasource_id': data.get('datasource_id'),
|
|
|
'timeout': data.get('timeout', 300),
|
|
|
'retry_count': data.get('retry_count', 3),
|
|
|
}
|
|
|
extra.update(data.get('task_config', {}))
|
|
|
|
|
|
# 获取用户所在团队
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("SELECT team_id FROM team_members WHERE user_id=%s LIMIT 1", (owner_id,))
|
|
|
tm = cur.fetchone()
|
|
|
team_id = tm['team_id'] if tm else None
|
|
|
|
|
|
cur2 = conn.cursor()
|
|
|
status = data.get('status', 'stopped')
|
|
|
if status not in ('stopped', 'draft'):
|
|
|
status = 'stopped'
|
|
|
cur2.execute("""
|
|
|
INSERT INTO tasks (owner_id, team_id, name, cron_expression, agent_id, trigger_type, status, workflow_json)
|
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
|
|
""", (
|
|
|
owner_id,
|
|
|
team_id,
|
|
|
name,
|
|
|
cron_expr or None,
|
|
|
data.get('agent_id') or None,
|
|
|
trigger_type,
|
|
|
status,
|
|
|
_json.dumps(extra, ensure_ascii=False)
|
|
|
))
|
|
|
conn.commit()
|
|
|
new_id = cur2.lastrowid
|
|
|
cur.close(); cur2.close(); conn.close()
|
|
|
|
|
|
_audit_log(owner_id, team_id, f"创建了任务「{name}」", 'task', new_id)
|
|
|
return jsonify({'id': new_id, 'message': '任务创建成功'}), 201
|
|
|
|
|
|
|
|
|
@app.route('/api/tasks/<int:task_id>', methods=['PUT'])
|
|
|
@require_auth
|
|
|
def update_task(task_id):
|
|
|
import json as _json
|
|
|
data = request.get_json() or {}
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
# 先读取现有 workflow_json，用于合并扩展字段
|
|
|
cur.execute("SELECT workflow_json FROM tasks WHERE id=%s", (task_id,))
|
|
|
row = cur.fetchone()
|
|
|
try:
|
|
|
existing_extra = _json.loads(row['workflow_json']) if row and row['workflow_json'] else {}
|
|
|
except Exception:
|
|
|
existing_extra = {}
|
|
|
|
|
|
# 更新扩展字段到 workflow_json
|
|
|
for key in ['description', 'category', 'skill_name', 'skill_params', 'datasource_id', 'timeout', 'retry_count']:
|
|
|
if key in data:
|
|
|
existing_extra[key] = data[key]
|
|
|
|
|
|
cur2 = conn.cursor()
|
|
|
fields, vals = [], []
|
|
|
for key in ['name', 'cron_expression', 'agent_id', 'trigger_type']:
|
|
|
if key in data:
|
|
|
fields.append(f"{key}=%s")
|
|
|
vals.append(data[key])
|
|
|
fields.append("workflow_json=%s")
|
|
|
vals.append(_json.dumps(existing_extra, ensure_ascii=False))
|
|
|
vals.append(task_id)
|
|
|
cur2.execute(f"UPDATE tasks SET {','.join(fields)}, updated_at=NOW() WHERE id=%s", vals)
|
|
|
conn.commit()
|
|
|
cur.close(); cur2.close(); conn.close()
|
|
|
return jsonify({'message': '更新成功'})
|
|
|
|
|
|
|
|
|
@app.route('/api/tasks/<int:task_id>/copy', methods=['POST'])
|
|
|
@require_auth
|
|
|
def copy_task(task_id):
|
|
|
import json as _json
|
|
|
owner_id = g.current_user['id']
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("SELECT * FROM tasks WHERE id=%s AND owner_id=%s", (task_id, owner_id))
|
|
|
task = cur.fetchone()
|
|
|
if not task:
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify({'error': '任务不存在'}), 404
|
|
|
cur2 = conn.cursor()
|
|
|
cur2.execute("""
|
|
|
INSERT INTO tasks (owner_id, team_id, name, cron_expression, agent_id, trigger_type, status, workflow_json)
|
|
|
VALUES (%s, %s, %s, %s, %s, %s, 'stopped', %s)
|
|
|
""", (
|
|
|
owner_id,
|
|
|
task['team_id'],
|
|
|
task['name'] + '-副本',
|
|
|
task['cron_expression'],
|
|
|
task['agent_id'],
|
|
|
task['trigger_type'],
|
|
|
task['workflow_json']
|
|
|
))
|
|
|
conn.commit()
|
|
|
new_id = cur2.lastrowid
|
|
|
cur.close(); cur2.close(); conn.close()
|
|
|
_audit_log(owner_id, task['team_id'], f"复制了任务「{task['name']}」", 'task', new_id)
|
|
|
return jsonify({'id': new_id, 'message': '复制成功'}), 201
|
|
|
|
|
|
|
|
|
@app.route('/api/tasks/<int:task_id>', methods=['DELETE'])
|
|
|
@require_auth
|
|
|
def delete_task(task_id):
|
|
|
owner_id = g.current_user['id']
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("SELECT id, name, team_id FROM tasks WHERE id=%s AND owner_id=%s", (task_id, owner_id))
|
|
|
task = cur.fetchone()
|
|
|
if not task:
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify({'error': '任务不存在或无权删除'}), 404
|
|
|
# 先停止（如果运行中），再删除执行记录和任务
|
|
|
cur2 = conn.cursor()
|
|
|
cur2.execute("DELETE FROM task_logs WHERE execution_id IN (SELECT id FROM task_executions WHERE task_id=%s)", (task_id,))
|
|
|
cur2.execute("DELETE FROM task_executions WHERE task_id=%s", (task_id,))
|
|
|
cur2.execute("DELETE FROM tasks WHERE id=%s", (task_id,))
|
|
|
conn.commit()
|
|
|
cur.close(); cur2.close(); conn.close()
|
|
|
_audit_log(owner_id, task.get('team_id'), f"删除了任务「{task['name']}」", 'task', task_id)
|
|
|
return jsonify({'message': '删除成功'})
|
|
|
|
|
|
|
|
|
@app.route('/api/tasks/<int:task_id>/status', methods=['PUT'])
|
|
|
@require_auth
|
|
|
def toggle_task_status(task_id):
|
|
|
import json as _json
|
|
|
data = request.get_json()
|
|
|
new_status = data.get('status')
|
|
|
if new_status not in ('running', 'stopped'):
|
|
|
return jsonify({'error': 'invalid status'}), 400
|
|
|
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("SELECT id, name, status, workflow_json FROM tasks WHERE id=%s", (task_id,))
|
|
|
task = cur.fetchone()
|
|
|
cur.close(); conn.close()
|
|
|
if not task:
|
|
|
return jsonify({'error': '任务不存在'}), 404
|
|
|
|
|
|
result_msg = f'状态已切换为 {new_status}'
|
|
|
openclaw_msg = ''
|
|
|
|
|
|
if new_status == 'running':
|
|
|
# 向 OpenClaw 注册调度（异步，不阻塞接口响应）
|
|
|
def _register():
|
|
|
res = openclaw.register_task(task_id)
|
|
|
if not res['ok']:
|
|
|
# 注册失败时将任务改回 stopped 并写告警
|
|
|
try:
|
|
|
c = db.get_conn()
|
|
|
cu = c.cursor()
|
|
|
cu.execute("UPDATE tasks SET status='stopped' WHERE id=%s", (task_id,))
|
|
|
cu.execute("""
|
|
|
INSERT INTO alert_notifications (target_user_id, task_id, alert_type, title, content)
|
|
|
SELECT owner_id, %s, 'task_fail', '任务启动失败', %s FROM tasks WHERE id=%s
|
|
|
""", (task_id, res['message'], task_id))
|
|
|
c.commit(); cu.close(); c.close()
|
|
|
except Exception:
|
|
|
pass
|
|
|
|
|
|
threading.Thread(target=_register, daemon=True).start()
|
|
|
openclaw_msg = '正在向 OpenClaw 注册调度...'
|
|
|
|
|
|
else: # stopped
|
|
|
# 通知 OpenClaw 移除调度
|
|
|
res = openclaw.unregister_task(task_id)
|
|
|
openclaw_msg = res['message']
|
|
|
|
|
|
conn2 = db.get_conn()
|
|
|
cur2 = conn2.cursor()
|
|
|
cur2.execute("UPDATE tasks SET status=%s WHERE id=%s", (new_status, task_id))
|
|
|
conn2.commit()
|
|
|
cur2.close(); conn2.close()
|
|
|
|
|
|
return jsonify({'message': result_msg, 'openclaw': openclaw_msg})
|
|
|
|
|
|
|
|
|
@app.route('/api/tasks/<int:task_id>/execute', methods=['POST'])
|
|
|
@require_auth
|
|
|
def manual_execute(task_id):
|
|
|
trigger_user_id = g.current_user['id']
|
|
|
|
|
|
# 1. 创建执行记录
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor()
|
|
|
cur.execute("""
|
|
|
INSERT INTO task_executions (task_id, trigger_type, trigger_user_id, status, started_at)
|
|
|
VALUES (%s, 'manual', %s, 'running', NOW())
|
|
|
""", (task_id, trigger_user_id))
|
|
|
conn.commit()
|
|
|
exec_id = cur.lastrowid
|
|
|
cur.close(); conn.close()
|
|
|
|
|
|
# 2. 异步调用 OpenClaw（不阻塞前端响应）
|
|
|
def _run():
|
|
|
res = openclaw.trigger_once(task_id, exec_id)
|
|
|
if not res['ok']:
|
|
|
try:
|
|
|
c = db.get_conn()
|
|
|
cu = c.cursor()
|
|
|
cu.execute("""
|
|
|
UPDATE task_executions SET status='failed', ended_at=NOW(), error_message=%s
|
|
|
WHERE id=%s
|
|
|
""", (res['message'], exec_id))
|
|
|
c.commit(); cu.close(); c.close()
|
|
|
except Exception:
|
|
|
pass
|
|
|
|
|
|
threading.Thread(target=_run, daemon=True).start()
|
|
|
|
|
|
return jsonify({'execution_id': exec_id, 'message': '已触发手动执行，正在调用 OpenClaw...'})
|
|
|
|
|
|
|
|
|
@app.route('/api/tasks/<int:task_id>/executions')
|
|
|
@require_auth
|
|
|
def task_executions(task_id):
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("""
|
|
|
SELECT id, trigger_type, status, started_at, ended_at, duration_ms,
|
|
|
retry_attempt, error_message, cpu_usage_pct, memory_usage_pct
|
|
|
FROM task_executions WHERE task_id=%s ORDER BY started_at DESC LIMIT 20
|
|
|
""", (task_id,))
|
|
|
rows = cur.fetchall()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify(db.serialize_rows(rows))
|
|
|
|
|
|
|
|
|
@app.route('/api/tasks/<int:task_id>/logs')
|
|
|
@require_auth
|
|
|
def task_logs(task_id):
|
|
|
level = request.args.get('level', 'all')
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("SELECT id FROM task_executions WHERE task_id=%s ORDER BY started_at DESC LIMIT 1", (task_id,))
|
|
|
exec_row = cur.fetchone()
|
|
|
if not exec_row:
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify([])
|
|
|
exec_id = exec_row['id']
|
|
|
if level == 'all':
|
|
|
cur.execute("SELECT log_level, log_time, message, node_name FROM task_logs WHERE execution_id=%s ORDER BY log_time ASC", (exec_id,))
|
|
|
else:
|
|
|
cur.execute("SELECT log_level, log_time, message, node_name FROM task_logs WHERE execution_id=%s AND log_level=%s ORDER BY log_time ASC", (exec_id, level))
|
|
|
rows = cur.fetchall()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify(db.serialize_rows(rows))
|
|
|
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
|
# 应用空间 —— 智能体广场（公共，不需个人认证也可浏览，但统一加认证）
|
|
|
# ─────────────────────────────────────────────
|
|
|
@app.route('/api/agents')
|
|
|
@require_auth
|
|
|
def list_agents():
|
|
|
keyword = request.args.get('keyword', '').strip()
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
if keyword:
|
|
|
cur.execute("""
|
|
|
SELECT id, agent_code, name, provider, category, description, usage_count, tags
|
|
|
FROM agents WHERE is_public=1 AND status='active'
|
|
|
AND (name LIKE %s OR description LIKE %s)
|
|
|
ORDER BY usage_count DESC
|
|
|
""", (f'%{keyword}%', f'%{keyword}%'))
|
|
|
else:
|
|
|
cur.execute("""
|
|
|
SELECT id, agent_code, name, provider, category, description, usage_count, tags
|
|
|
FROM agents WHERE is_public=1 AND status='active'
|
|
|
ORDER BY usage_count DESC
|
|
|
""")
|
|
|
rows = cur.fetchall()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify(rows)
|
|
|
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
|
# 应用空间 —— 任务模板广场（公共共享）
|
|
|
# ─────────────────────────────────────────────
|
|
|
@app.route('/api/task_templates')
|
|
|
@require_auth
|
|
|
def list_templates():
|
|
|
keyword = request.args.get('keyword', '').strip()
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
if keyword:
|
|
|
cur.execute("""
|
|
|
SELECT tt.id, tt.title, tt.description, tt.category, tt.tags,
|
|
|
tt.usage_count, tt.contributor_alias, u.display_name AS contributor_name
|
|
|
FROM task_templates tt JOIN users u ON tt.contributor_id=u.id
|
|
|
WHERE tt.is_active=1 AND (tt.title LIKE %s OR tt.description LIKE %s)
|
|
|
ORDER BY tt.usage_count DESC
|
|
|
""", (f'%{keyword}%', f'%{keyword}%'))
|
|
|
else:
|
|
|
cur.execute("""
|
|
|
SELECT tt.id, tt.title, tt.description, tt.category, tt.tags,
|
|
|
tt.usage_count, tt.contributor_alias, u.display_name AS contributor_name
|
|
|
FROM task_templates tt JOIN users u ON tt.contributor_id=u.id
|
|
|
WHERE tt.is_active=1 ORDER BY tt.usage_count DESC
|
|
|
""")
|
|
|
rows = cur.fetchall()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify(rows)
|
|
|
|
|
|
|
|
|
@app.route('/api/task_templates', methods=['POST'])
|
|
|
@require_auth
|
|
|
def publish_template():
|
|
|
data = request.get_json()
|
|
|
owner_id = g.current_user['id']
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor()
|
|
|
cur.execute("""
|
|
|
INSERT INTO task_templates (title, description, category, contributor_id, contributor_alias, task_config)
|
|
|
VALUES (%s, %s, %s, %s, %s, %s)
|
|
|
""", (
|
|
|
data['title'],
|
|
|
data.get('description', ''),
|
|
|
data.get('category', '情报处理'),
|
|
|
owner_id,
|
|
|
data.get('contributor_alias', g.current_user['name']),
|
|
|
db.to_json(data.get('task_config', {}))
|
|
|
))
|
|
|
conn.commit()
|
|
|
new_id = cur.lastrowid
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify({'id': new_id, 'message': '模板发布成功'}), 201
|
|
|
|
|
|
|
|
|
@app.route('/api/task_templates/<int:tpl_id>/import', methods=['POST'])
|
|
|
@require_auth
|
|
|
def import_template(tpl_id):
|
|
|
owner_id = g.current_user['id']
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("SELECT * FROM task_templates WHERE id=%s", (tpl_id,))
|
|
|
tpl = cur.fetchone()
|
|
|
if not tpl:
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify({'error': 'template not found'}), 404
|
|
|
|
|
|
config = tpl.get('task_config') or {}
|
|
|
cur2 = conn.cursor()
|
|
|
cur2.execute("""
|
|
|
INSERT INTO tasks (owner_id, name, cron_expression, agent_id, trigger_type, status, source_template_id)
|
|
|
VALUES (%s, %s, %s, NULL, 'cron', 'stopped', %s)
|
|
|
""", (
|
|
|
owner_id,
|
|
|
tpl['title'] + ' (导入)',
|
|
|
config.get('cron_expression', '未配置'),
|
|
|
tpl_id
|
|
|
))
|
|
|
cur2.execute("UPDATE task_templates SET usage_count=usage_count+1 WHERE id=%s", (tpl_id,))
|
|
|
conn.commit()
|
|
|
new_task_id = cur2.lastrowid
|
|
|
cur.close(); cur2.close(); conn.close()
|
|
|
return jsonify({'task_id': new_task_id, 'message': '导入成功'})
|
|
|
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
|
# 团队协作空间
|
|
|
# ─────────────────────────────────────────────
|
|
|
@app.route('/api/teams/<int:team_id>/members')
|
|
|
@require_auth
|
|
|
def team_members(team_id):
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("""
|
|
|
SELECT u.id, u.display_name AS name, u.avatar_color AS color,
|
|
|
tm.role, tm.is_online AS online, tm.joined_at
|
|
|
FROM team_members tm JOIN users u ON tm.user_id=u.id
|
|
|
WHERE tm.team_id=%s ORDER BY tm.joined_at ASC
|
|
|
""", (team_id,))
|
|
|
rows = cur.fetchall()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify(db.serialize_rows(rows))
|
|
|
|
|
|
|
|
|
@app.route('/api/teams/<int:team_id>/members/<int:user_id>/role', methods=['PUT'])
|
|
|
@require_auth
|
|
|
def change_member_role(team_id, user_id):
|
|
|
data = request.get_json()
|
|
|
role = data.get('role')
|
|
|
if role not in ('admin', 'editor', 'viewer'):
|
|
|
return jsonify({'error': 'invalid role'}), 400
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor()
|
|
|
cur.execute("UPDATE team_members SET role=%s WHERE team_id=%s AND user_id=%s", (role, team_id, user_id))
|
|
|
conn.commit()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify({'message': '角色已更新'})
|
|
|
|
|
|
|
|
|
@app.route('/api/teams/<int:team_id>/members/<int:user_id>', methods=['DELETE'])
|
|
|
@require_auth
|
|
|
def remove_member(team_id, user_id):
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor()
|
|
|
cur.execute("DELETE FROM team_members WHERE team_id=%s AND user_id=%s", (team_id, user_id))
|
|
|
conn.commit()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify({'message': '成员已移除'})
|
|
|
|
|
|
|
|
|
@app.route('/api/teams/<int:team_id>/members', methods=['POST'])
|
|
|
@require_auth
|
|
|
def invite_member(team_id):
|
|
|
data = request.get_json()
|
|
|
name = data.get('name', '').strip()
|
|
|
if not name:
|
|
|
return jsonify({'error': 'name required'}), 400
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("SELECT id FROM users WHERE display_name=%s", (name,))
|
|
|
user = cur.fetchone()
|
|
|
if not user:
|
|
|
import random, string
|
|
|
uname = 'user_' + ''.join(random.choices(string.digits, k=6))
|
|
|
cur2 = conn.cursor()
|
|
|
cur2.execute("""
|
|
|
INSERT INTO users (username, display_name, password_hash, avatar_color)
|
|
|
VALUES (%s, %s, 'placeholder', %s)
|
|
|
""", (uname, name, data.get('color', '#3b82f6')))
|
|
|
conn.commit()
|
|
|
user_id = cur2.lastrowid
|
|
|
cur2.close()
|
|
|
else:
|
|
|
user_id = user['id']
|
|
|
cur3 = conn.cursor()
|
|
|
try:
|
|
|
cur3.execute("INSERT INTO team_members (team_id, user_id, role) VALUES (%s, %s, 'viewer')", (team_id, user_id))
|
|
|
conn.commit()
|
|
|
except Exception:
|
|
|
pass
|
|
|
cur3.close(); cur.close(); conn.close()
|
|
|
return jsonify({'user_id': user_id, 'message': f'{name} 已加入团队'})
|
|
|
|
|
|
|
|
|
@app.route('/api/teams/<int:team_id>/shared_resources')
|
|
|
@require_auth
|
|
|
def shared_resources(team_id):
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("""
|
|
|
SELECT sr.id, sr.resource_type, sr.resource_id, sr.resource_name,
|
|
|
sr.usage_count, u.display_name AS creator, sr.shared_at
|
|
|
FROM team_shared_resources sr JOIN users u ON sr.shared_by=u.id
|
|
|
WHERE sr.team_id=%s AND sr.is_active=1 ORDER BY sr.shared_at DESC
|
|
|
""", (team_id,))
|
|
|
rows = cur.fetchall()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify(db.serialize_rows(rows))
|
|
|
|
|
|
|
|
|
@app.route('/api/teams/<int:team_id>/shared_resources/<int:res_id>', methods=['DELETE'])
|
|
|
@require_auth
|
|
|
def unshare_resource(team_id, res_id):
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor()
|
|
|
cur.execute("UPDATE team_shared_resources SET is_active=0 WHERE id=%s AND team_id=%s", (res_id, team_id))
|
|
|
conn.commit()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify({'message': '已取消共享'})
|
|
|
|
|
|
|
|
|
@app.route('/api/teams/<int:team_id>/audit_logs')
|
|
|
@require_auth
|
|
|
def audit_logs(team_id):
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("""
|
|
|
SELECT al.id, u.display_name AS user, al.action, al.resource_type, al.created_at
|
|
|
FROM team_audit_logs al JOIN users u ON al.user_id=u.id
|
|
|
WHERE al.team_id=%s ORDER BY al.created_at DESC LIMIT 30
|
|
|
""", (team_id,))
|
|
|
rows = cur.fetchall()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify(db.serialize_rows(rows))
|
|
|
|
|
|
|
|
|
@app.route('/api/teams/<int:team_id>/comments')
|
|
|
@require_auth
|
|
|
def team_comments(team_id):
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("""
|
|
|
SELECT c.id, u.display_name AS user, u.avatar_color AS color,
|
|
|
c.content AS text, c.created_at AS time, c.parent_id
|
|
|
FROM team_comments c JOIN users u ON c.user_id=u.id
|
|
|
WHERE c.team_id=%s AND c.is_deleted=0 ORDER BY c.created_at ASC
|
|
|
""", (team_id,))
|
|
|
rows = cur.fetchall()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify(db.serialize_rows(rows))
|
|
|
|
|
|
|
|
|
@app.route('/api/teams/<int:team_id>/comments', methods=['POST'])
|
|
|
@require_auth
|
|
|
def post_comment(team_id):
|
|
|
data = request.get_json()
|
|
|
user_id = g.current_user['id']
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor()
|
|
|
cur.execute("""
|
|
|
INSERT INTO team_comments (team_id, user_id, content, parent_id)
|
|
|
VALUES (%s, %s, %s, %s)
|
|
|
""", (team_id, user_id, data['content'], data.get('parent_id')))
|
|
|
conn.commit()
|
|
|
new_id = cur.lastrowid
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify({'id': new_id, 'message': '留言成功'})
|
|
|
|
|
|
|
|
|
@app.route('/api/teams/<int:team_id>/info')
|
|
|
@require_auth
|
|
|
def team_info(team_id):
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("SELECT id, team_code, team_name, description FROM teams WHERE id=%s", (team_id,))
|
|
|
row = cur.fetchone()
|
|
|
cur.close(); conn.close()
|
|
|
if not row:
|
|
|
return jsonify({'error': 'not found'}), 404
|
|
|
return jsonify(row)
|
|
|
|
|
|
|
|
|
# 获取当前用户所属团队
|
|
|
@app.route('/api/my/team')
|
|
|
@require_auth
|
|
|
def my_team():
|
|
|
user_id = g.current_user['id']
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("""
|
|
|
SELECT t.id, t.team_code, t.team_name
|
|
|
FROM team_members tm JOIN teams t ON tm.team_id=t.id
|
|
|
WHERE tm.user_id=%s ORDER BY tm.joined_at ASC LIMIT 1
|
|
|
""", (user_id,))
|
|
|
row = cur.fetchone()
|
|
|
cur.close(); conn.close()
|
|
|
if not row:
|
|
|
return jsonify({'id': 1, 'team_code': 'DEFAULT', 'team_name': '默认团队'})
|
|
|
return jsonify(row)
|
|
|
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
|
# 核心配置中心
|
|
|
# ─────────────────────────────────────────────
|
|
|
@app.route('/api/datasources')
|
|
|
@require_auth
|
|
|
def list_datasources():
|
|
|
owner_id = g.current_user['id']
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("""
|
|
|
SELECT id, name, ds_type, host, database_name, username, connection_status, last_tested_at
|
|
|
FROM datasources WHERE owner_id=%s ORDER BY created_at DESC
|
|
|
""", (owner_id,))
|
|
|
rows = cur.fetchall()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify(db.serialize_rows(rows))
|
|
|
|
|
|
|
|
|
@app.route('/api/datasources/<int:ds_id>', methods=['PUT'])
|
|
|
@require_auth
|
|
|
def update_datasource(ds_id):
|
|
|
data = request.get_json()
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor()
|
|
|
password = data.get('password', '')
|
|
|
enc = db.simple_encrypt(password) if password else None
|
|
|
fields, vals = [], []
|
|
|
if 'host' in data: fields.append('host=%s'); vals.append(data['host'])
|
|
|
if 'username' in data: fields.append('username=%s'); vals.append(data['username'])
|
|
|
if enc: fields.append('credential_enc=%s'); vals.append(enc)
|
|
|
if fields:
|
|
|
vals.append(ds_id)
|
|
|
cur.execute(f"UPDATE datasources SET {','.join(fields)} WHERE id=%s", vals)
|
|
|
conn.commit()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify({'message': '凭证已保存'})
|
|
|
|
|
|
|
|
|
@app.route('/api/datasources', methods=['POST'])
|
|
|
@require_auth
|
|
|
def create_datasource():
|
|
|
data = request.get_json()
|
|
|
owner_id = g.current_user['id']
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor()
|
|
|
enc = db.simple_encrypt(data.get('password', ''))
|
|
|
cur.execute("""
|
|
|
INSERT INTO datasources (owner_id, name, ds_type, host, database_name, username, credential_enc, connection_status)
|
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, 'unknown')
|
|
|
""", (
|
|
|
owner_id,
|
|
|
data['name'], data.get('ds_type', 'mysql'),
|
|
|
data.get('host', ''), data.get('database_name', ''),
|
|
|
data.get('username', ''), enc
|
|
|
))
|
|
|
conn.commit()
|
|
|
new_id = cur.lastrowid
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify({'id': new_id, 'message': '数据源已创建'}), 201
|
|
|
|
|
|
|
|
|
@app.route('/api/ai_engines')
|
|
|
@require_auth
|
|
|
def list_ai_engines():
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("SELECT id, engine_type, name, endpoint, description, status, last_checked_at FROM ai_engines")
|
|
|
rows = cur.fetchall()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify(db.serialize_rows(rows))
|
|
|
|
|
|
|
|
|
@app.route('/api/ai_engines/<int:eng_id>/restart', methods=['POST'])
|
|
|
@require_auth
|
|
|
def restart_engine(eng_id):
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor()
|
|
|
cur.execute("UPDATE ai_engines SET status='ready', last_checked_at=NOW() WHERE id=%s", (eng_id,))
|
|
|
conn.commit()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify({'message': '重启指令已发送'})
|
|
|
|
|
|
|
|
|
@app.route('/api/alerts')
|
|
|
@require_auth
|
|
|
def list_alerts():
|
|
|
user_id = g.current_user['id']
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
cur.execute("""
|
|
|
SELECT id, alert_type, title, content, is_read, created_at
|
|
|
FROM alert_notifications WHERE target_user_id=%s
|
|
|
ORDER BY created_at DESC LIMIT 20
|
|
|
""", (user_id,))
|
|
|
rows = cur.fetchall()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify(db.serialize_rows(rows))
|
|
|
|
|
|
|
|
|
@app.route('/api/alerts/read_all', methods=['POST'])
|
|
|
@require_auth
|
|
|
def mark_alerts_read():
|
|
|
user_id = g.current_user['id']
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor()
|
|
|
cur.execute("UPDATE alert_notifications SET is_read=1 WHERE target_user_id=%s", (user_id,))
|
|
|
conn.commit()
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify({'message': '全部已读'})
|
|
|
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
|
# OpenClaw 回调接口（无需登录验证，由 OpenClaw 直接调用）
|
|
|
# ─────────────────────────────────────────────
|
|
|
@app.route('/api/openclaw/callback/<int:task_id>', methods=['POST'])
|
|
|
def openclaw_callback(task_id):
|
|
|
"""
|
|
|
OpenClaw 执行完成后回调此接口，写入执行结果和日志。
|
|
|
请求体示例：
|
|
|
{
|
|
|
"execution_id": 123, # 灵枢侧执行记录ID（可选，优先用）
|
|
|
"job_id": "lingshu_task_1_exec_123",
|
|
|
"status": "success", # success / failed / timeout
|
|
|
"duration_ms": 1500,
|
|
|
"cpu_usage_pct": 35.2,
|
|
|
"memory_usage_pct": 28.1,
|
|
|
"error_message": "",
|
|
|
"logs": [
|
|
|
{"level": "info", "time": "2026-01-01 12:00:00.000", "message": "xxx", "node": "step1"},
|
|
|
...
|
|
|
]
|
|
|
}
|
|
|
"""
|
|
|
import json as _json
|
|
|
data = request.get_json(silent=True) or {}
|
|
|
|
|
|
status = data.get('status', 'success')
|
|
|
duration_ms = data.get('duration_ms')
|
|
|
cpu = data.get('cpu_usage_pct')
|
|
|
mem = data.get('memory_usage_pct')
|
|
|
error_msg = data.get('error_message', '')
|
|
|
logs = data.get('logs', [])
|
|
|
exec_id_from_cb = data.get('execution_id')
|
|
|
|
|
|
# 找到对应的执行记录
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor(dictionary=True)
|
|
|
|
|
|
if exec_id_from_cb:
|
|
|
exec_id = exec_id_from_cb
|
|
|
else:
|
|
|
# 取该任务最近一条 running 的执行记录
|
|
|
cur.execute("""
|
|
|
SELECT id FROM task_executions
|
|
|
WHERE task_id=%s AND status='running'
|
|
|
ORDER BY started_at DESC LIMIT 1
|
|
|
""", (task_id,))
|
|
|
row = cur.fetchone()
|
|
|
exec_id = row['id'] if row else None
|
|
|
|
|
|
if not exec_id:
|
|
|
cur.close(); conn.close()
|
|
|
return jsonify({'ok': False, 'message': '未找到对应执行记录'}), 404
|
|
|
|
|
|
# 更新执行记录
|
|
|
cur2 = conn.cursor()
|
|
|
cur2.execute("""
|
|
|
UPDATE task_executions
|
|
|
SET status=%s, ended_at=NOW(), duration_ms=%s,
|
|
|
cpu_usage_pct=%s, memory_usage_pct=%s, error_message=%s
|
|
|
WHERE id=%s
|
|
|
""", (status, duration_ms, cpu, mem, error_msg or None, exec_id))
|
|
|
|
|
|
# 更新任务最后运行时间
|
|
|
cur2.execute("UPDATE tasks SET last_run_at=NOW() WHERE id=%s", (task_id,))
|
|
|
|
|
|
# 写入日志
|
|
|
now_str = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
|
|
|
for log in logs:
|
|
|
cur2.execute("""
|
|
|
INSERT INTO task_logs (execution_id, task_id, log_level, log_time, message, node_name)
|
|
|
VALUES (%s, %s, %s, %s, %s, %s)
|
|
|
""", (
|
|
|
exec_id, task_id,
|
|
|
log.get('level', 'info'),
|
|
|
log.get('time', now_str),
|
|
|
log.get('message', ''),
|
|
|
log.get('node')
|
|
|
))
|
|
|
|
|
|
# 失败时写告警通知
|
|
|
if status in ('failed', 'timeout'):
|
|
|
cur2.execute("""
|
|
|
INSERT INTO alert_notifications (target_user_id, task_id, alert_type, title, content)
|
|
|
SELECT owner_id, %s, 'task_fail', '任务执行失败',
|
|
|
CONCAT('任务执行', IF(%s='timeout','超时','失败'), IFNULL(CONCAT('：', %s), ''))
|
|
|
FROM tasks WHERE id=%s
|
|
|
""", (task_id, status, error_msg or None, task_id))
|
|
|
|
|
|
conn.commit()
|
|
|
cur.close(); cur2.close(); conn.close()
|
|
|
|
|
|
return jsonify({'ok': True, 'message': '回调已处理'})
|
|
|
|
|
|
|
|
|
@app.route('/api/openclaw/status', methods=['GET'])
|
|
|
def openclaw_status():
|
|
|
"""检查 OpenClaw 连接状态：通过 WebSocket 握手探测"""
|
|
|
try:
|
|
|
result = openclaw._acp_call('cron.status', {})
|
|
|
return jsonify({'connected': True, 'message': 'OpenClaw 在线',
|
|
|
'base_url': openclaw.OPENCLAW_BASE})
|
|
|
except Exception as e:
|
|
|
err = str(e)
|
|
|
connected = 'not reachable' not in err.lower() and 'ConnectionRefusedError' not in err
|
|
|
return jsonify({'connected': False, 'message': err,
|
|
|
'base_url': openclaw.OPENCLAW_BASE})
|
|
|
|
|
|
|
|
|
# ─────────────────────────────────────────────
|
|
|
# 工具函数
|
|
|
# ─────────────────────────────────────────────
|
|
|
def _audit_log(user_id, team_id, action, res_type=None, res_id=None):
|
|
|
if not team_id:
|
|
|
return
|
|
|
try:
|
|
|
conn = db.get_conn()
|
|
|
cur = conn.cursor()
|
|
|
cur.execute("""
|
|
|
INSERT INTO team_audit_logs (team_id, user_id, action, resource_type, resource_id)
|
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
|
""", (team_id, user_id, action, res_type, res_id))
|
|
|
conn.commit()
|
|
|
cur.close(); conn.close()
|
|
|
except Exception:
|
|
|
pass
|
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
|
app.run(host='0.0.0.0', port=5000, debug=True, use_reloader=False)
|
