feifei.xu
608e53ed2f
- 创建 bots.py 路由文件 - 实现 9 个 REST API 端点： - GET /api/bots - 获取列表 - POST /api/bots - 创建 - GET /api/bots/:id - 详情 - PUT /api/bots/:id - 更新 - DELETE /api/bots/:id - 删除 - POST /api/bots/:id/bind - 绑定 Agent - POST /api/bots/:id/unbind - 解绑 Agent - GET /api/bots/:id/status - 状态 - POST /api/bots/:id/heartbeat - 心跳 - POST /api/bots/:id/token - 重新生成 Token - GET /api/bots/:id/stats - 统计 - 实现 JWT 权限检查 - 实现 X-Bot-Token 认证 - 注册蓝图到应用 - 更新版本号到 0.9.2
363 lines
9.7 KiB
Python
363 lines
9.7 KiB
Python
"""
|
||
Bot 路由 - 机器人管理 API
|
||
"""
|
||
from flask import Blueprint, request, jsonify
|
||
from flask_jwt_extended import jwt_required, get_jwt_identity
|
||
from datetime import datetime
|
||
|
||
from app.models import User, Agent
|
||
from app.services.bot_service import BotService
|
||
|
||
bots_bp = Blueprint('bots', __name__)
|
||
|
||
|
||
@bots_bp.route('/', methods=['GET'])
|
||
@jwt_required()
|
||
def get_bots():
|
||
"""
|
||
获取机器人列表
|
||
|
||
Query params:
|
||
- owner_only: bool - 只返回自己的 Bot
|
||
- include_system: bool - 包含系统 Bot
|
||
"""
|
||
user_id = get_jwt_identity()
|
||
user = User.query.get(user_id)
|
||
|
||
if not user:
|
||
return jsonify({'error': 'User not found'}), 404
|
||
|
||
owner_only = request.args.get('owner_only', 'false').lower() == 'true'
|
||
|
||
if owner_only:
|
||
bots = BotService.get_bots_by_owner(user_id)
|
||
else:
|
||
bots = BotService.get_available_bots(user)
|
||
|
||
return jsonify({
|
||
'bots': [bot.to_dict() for bot in bots],
|
||
'count': len(bots)
|
||
}), 200
|
||
|
||
|
||
@bots_bp.route('/', methods=['POST'])
|
||
@jwt_required()
|
||
def create_bot():
|
||
"""
|
||
创建机器人
|
||
|
||
Request body:
|
||
- name: str - 机器人名称（必填，唯一）
|
||
- display_name: str - 显示名称
|
||
- avatar: str - 头像 URL
|
||
- description: str - 描述
|
||
- is_system: bool - 是否为系统级 Bot（仅 admin）
|
||
- agent_id: str - 绑定的 Agent ID
|
||
- capabilities: list - 能力标签
|
||
- config: dict - 配置
|
||
"""
|
||
user_id = get_jwt_identity()
|
||
user = User.query.get(user_id)
|
||
|
||
if not user:
|
||
return jsonify({'error': 'User not found'}), 404
|
||
|
||
data = request.get_json()
|
||
|
||
# 验证必填字段
|
||
if not data or 'name' not in data:
|
||
return jsonify({'error': 'name is required'}), 400
|
||
|
||
name = data['name'].strip()
|
||
|
||
# 检查名称是否已存在
|
||
if BotService.get_bot_by_name(name):
|
||
return jsonify({'error': 'Bot name already exists'}), 400
|
||
|
||
# 检查 is_system 权限
|
||
is_system = data.get('is_system', False)
|
||
if is_system and user.role != 'admin':
|
||
return jsonify({'error': 'Only admin can create system bots'}), 403
|
||
|
||
# 检查 agent_id 是否有效
|
||
agent_id = data.get('agent_id')
|
||
if agent_id:
|
||
agent = Agent.query.get(agent_id)
|
||
if not agent:
|
||
return jsonify({'error': 'Agent not found'}), 404
|
||
|
||
# 创建 Bot
|
||
bot, token = BotService.create_bot(
|
||
name=name,
|
||
owner_id=user_id,
|
||
display_name=data.get('display_name'),
|
||
avatar=data.get('avatar'),
|
||
description=data.get('description'),
|
||
is_system=is_system,
|
||
agent_id=agent_id,
|
||
capabilities=data.get('capabilities'),
|
||
config=data.get('config')
|
||
)
|
||
|
||
return jsonify({
|
||
'bot': bot.to_dict(),
|
||
'token': token # 只在创建时返回一次
|
||
}), 201
|
||
|
||
|
||
@bots_bp.route('/<bot_id>', methods=['GET'])
|
||
@jwt_required()
|
||
def get_bot(bot_id):
|
||
"""获取机器人详情"""
|
||
user_id = get_jwt_identity()
|
||
user = User.query.get(user_id)
|
||
|
||
if not user:
|
||
return jsonify({'error': 'User not found'}), 404
|
||
|
||
bot = BotService.get_bot_by_id(bot_id)
|
||
if not bot:
|
||
return jsonify({'error': 'Bot not found'}), 404
|
||
|
||
# 权限检查
|
||
if not BotService.check_permission(user, bot, 'view'):
|
||
return jsonify({'error': 'Permission denied'}), 403
|
||
|
||
return jsonify({'bot': bot.to_dict()}), 200
|
||
|
||
|
||
@bots_bp.route('/<bot_id>', methods=['PUT'])
|
||
@jwt_required()
|
||
def update_bot(bot_id):
|
||
"""
|
||
更新机器人配置
|
||
|
||
Request body:
|
||
- display_name: str
|
||
- avatar: str
|
||
- description: str
|
||
- capabilities: list
|
||
- config: dict
|
||
"""
|
||
user_id = get_jwt_identity()
|
||
user = User.query.get(user_id)
|
||
|
||
if not user:
|
||
return jsonify({'error': 'User not found'}), 404
|
||
|
||
bot = BotService.get_bot_by_id(bot_id)
|
||
if not bot:
|
||
return jsonify({'error': 'Bot not found'}), 404
|
||
|
||
# 权限检查
|
||
if not BotService.check_permission(user, bot, 'edit'):
|
||
return jsonify({'error': 'Permission denied'}), 403
|
||
|
||
data = request.get_json() or {}
|
||
|
||
bot = BotService.update_bot(
|
||
bot=bot,
|
||
display_name=data.get('display_name'),
|
||
avatar=data.get('avatar'),
|
||
description=data.get('description'),
|
||
capabilities=data.get('capabilities'),
|
||
config=data.get('config')
|
||
)
|
||
|
||
return jsonify({'bot': bot.to_dict()}), 200
|
||
|
||
|
||
@bots_bp.route('/<bot_id>', methods=['DELETE'])
|
||
@jwt_required()
|
||
def delete_bot(bot_id):
|
||
"""删除机器人"""
|
||
user_id = get_jwt_identity()
|
||
user = User.query.get(user_id)
|
||
|
||
if not user:
|
||
return jsonify({'error': 'User not found'}), 404
|
||
|
||
bot = BotService.get_bot_by_id(bot_id)
|
||
if not bot:
|
||
return jsonify({'error': 'Bot not found'}), 404
|
||
|
||
# 权限检查
|
||
if not BotService.check_permission(user, bot, 'delete'):
|
||
return jsonify({'error': 'Permission denied'}), 403
|
||
|
||
# 不允许删除系统 Bot（除非是 admin）
|
||
if bot.is_system and user.role != 'admin':
|
||
return jsonify({'error': 'Cannot delete system bot'}), 403
|
||
|
||
success = BotService.delete_bot(bot)
|
||
|
||
if success:
|
||
return jsonify({'message': 'Bot deleted'}), 200
|
||
else:
|
||
return jsonify({'error': 'Failed to delete bot'}), 500
|
||
|
||
|
||
@bots_bp.route('/<bot_id>/bind', methods=['POST'])
|
||
@jwt_required()
|
||
def bind_agent(bot_id):
|
||
"""
|
||
绑定 Agent
|
||
|
||
Request body:
|
||
- agent_id: str - Agent ID
|
||
"""
|
||
user_id = get_jwt_identity()
|
||
user = User.query.get(user_id)
|
||
|
||
if not user:
|
||
return jsonify({'error': 'User not found'}), 404
|
||
|
||
bot = BotService.get_bot_by_id(bot_id)
|
||
if not bot:
|
||
return jsonify({'error': 'Bot not found'}), 404
|
||
|
||
# 权限检查
|
||
if not BotService.check_permission(user, bot, 'bind'):
|
||
return jsonify({'error': 'Permission denied'}), 403
|
||
|
||
data = request.get_json()
|
||
if not data or 'agent_id' not in data:
|
||
return jsonify({'error': 'agent_id is required'}), 400
|
||
|
||
agent_id = data['agent_id']
|
||
agent = Agent.query.get(agent_id)
|
||
if not agent:
|
||
return jsonify({'error': 'Agent not found'}), 404
|
||
|
||
bot = BotService.bind_agent(bot, agent_id)
|
||
|
||
return jsonify({
|
||
'bot': bot.to_dict(),
|
||
'message': f'Bot bound to agent {agent.name}'
|
||
}), 200
|
||
|
||
|
||
@bots_bp.route('/<bot_id>/unbind', methods=['POST'])
|
||
@jwt_required()
|
||
def unbind_agent(bot_id):
|
||
"""解绑 Agent"""
|
||
user_id = get_jwt_identity()
|
||
user = User.query.get(user_id)
|
||
|
||
if not user:
|
||
return jsonify({'error': 'User not found'}), 404
|
||
|
||
bot = BotService.get_bot_by_id(bot_id)
|
||
if not bot:
|
||
return jsonify({'error': 'Bot not found'}), 404
|
||
|
||
# 权限检查
|
||
if not BotService.check_permission(user, bot, 'bind'):
|
||
return jsonify({'error': 'Permission denied'}), 403
|
||
|
||
bot = BotService.bind_agent(bot, None)
|
||
|
||
return jsonify({
|
||
'bot': bot.to_dict(),
|
||
'message': 'Bot unbound from agent'
|
||
}), 200
|
||
|
||
|
||
@bots_bp.route('/<bot_id>/status', methods=['GET'])
|
||
@jwt_required()
|
||
def get_bot_status(bot_id):
|
||
"""获取机器人状态"""
|
||
user_id = get_jwt_identity()
|
||
user = User.query.get(user_id)
|
||
|
||
if not user:
|
||
return jsonify({'error': 'User not found'}), 404
|
||
|
||
bot = BotService.get_bot_by_id(bot_id)
|
||
if not bot:
|
||
return jsonify({'error': 'Bot not found'}), 404
|
||
|
||
# 权限检查
|
||
if not BotService.check_permission(user, bot, 'view'):
|
||
return jsonify({'error': 'Permission denied'}), 403
|
||
|
||
return jsonify({
|
||
'bot_id': bot.id,
|
||
'name': bot.name,
|
||
'status': bot.status,
|
||
'agent_id': bot.agent_id,
|
||
'last_active_at': bot.last_active_at.isoformat() if bot.last_active_at else None
|
||
}), 200
|
||
|
||
|
||
@bots_bp.route('/<bot_id>/heartbeat', methods=['POST'])
|
||
def bot_heartbeat(bot_id):
|
||
"""
|
||
机器人心跳上报
|
||
|
||
用于 Bot Token 认证（非 JWT）
|
||
Headers: X-Bot-Token: <token>
|
||
"""
|
||
bot = BotService.get_bot_by_id(bot_id)
|
||
if not bot:
|
||
return jsonify({'error': 'Bot not found'}), 404
|
||
|
||
# Token 认证
|
||
token = request.headers.get('X-Bot-Token')
|
||
if not token or not BotService.verify_token(bot, token):
|
||
return jsonify({'error': 'Invalid bot token'}), 401
|
||
|
||
# 更新状态
|
||
BotService.update_status(bot, 'online')
|
||
|
||
return jsonify({'status': 'ok'}), 200
|
||
|
||
|
||
@bots_bp.route('/<bot_id>/token', methods=['POST'])
|
||
@jwt_required()
|
||
def regenerate_token(bot_id):
|
||
"""重新生成 Bot Token"""
|
||
user_id = get_jwt_identity()
|
||
user = User.query.get(user_id)
|
||
|
||
if not user:
|
||
return jsonify({'error': 'User not found'}), 404
|
||
|
||
bot = BotService.get_bot_by_id(bot_id)
|
||
if not bot:
|
||
return jsonify({'error': 'Bot not found'}), 404
|
||
|
||
# 权限检查
|
||
if not BotService.check_permission(user, bot, 'edit'):
|
||
return jsonify({'error': 'Permission denied'}), 403
|
||
|
||
token = BotService.regenerate_token(bot)
|
||
|
||
return jsonify({
|
||
'message': 'Token regenerated',
|
||
'token': token
|
||
}), 200
|
||
|
||
|
||
@bots_bp.route('/<bot_id>/stats', methods=['GET'])
|
||
@jwt_required()
|
||
def get_bot_stats(bot_id):
|
||
"""获取机器人统计信息"""
|
||
user_id = get_jwt_identity()
|
||
user = User.query.get(user_id)
|
||
|
||
if not user:
|
||
return jsonify({'error': 'User not found'}), 404
|
||
|
||
bot = BotService.get_bot_by_id(bot_id)
|
||
if not bot:
|
||
return jsonify({'error': 'Bot not found'}), 404
|
||
|
||
# 权限检查
|
||
if not BotService.check_permission(user, bot, 'view'):
|
||
return jsonify({'error': 'Permission denied'}), 403
|
||
|
||
stats = BotService.get_bot_stats(bot)
|
||
|
||
return jsonify(stats), 200
|