diff --git a/backend/python/app/api/admin/automation_admin.py b/backend/python/app/api/admin/automation_admin.py new file mode 100644 index 0000000..9e30205 --- /dev/null +++ b/backend/python/app/api/admin/automation_admin.py @@ -0,0 +1,662 @@ +""" +自动化任务调度管理 API +Phase 3.2: 支持定时任务(cron)和条件触发(webhook/event)的自动化调度 +""" +from fastapi import APIRouter, Depends, HTTPException, Query, Body, BackgroundTasks +from sqlalchemy import or_, func, desc, asc, text +from sqlalchemy.orm import Session +from pydantic import BaseModel, Field +from typing import Optional, List, Dict, Any +from datetime import datetime, timedelta +import uuid +import json +import logging +import asyncio + +from ...database import get_db +from ...models import ( + gen_uuid, +) +from ...auth import get_current_admin +from ...middleware.log_service import LogService + +router = APIRouter() +logger = logging.getLogger("aiflow") + + +# ============================================================ +# Pydantic Models +# ============================================================ + +class AutomationTaskCreate(BaseModel): + name: str = Field(min_length=1, max_length=200) + description: Optional[str] = "" + trigger_type: str = Field(default="cron", description="cron / webhook / event") + cron_expression: Optional[str] = None # e.g. "*/5 * * * *" + event_condition: Optional[Dict[str, Any]] = None # for event/webhook triggers + action_type: str = Field(default="ai_pipeline", description="ai_pipeline / webhook / notification / skill_execute") + action_config: Dict[str, Any] = Field(default_factory=dict) + is_enabled: bool = True + max_retries: int = Field(default=3, ge=0, le=10) + timeout_seconds: int = Field(default=300, ge=10, le=3600) + + +class AutomationTaskUpdate(BaseModel): + name: Optional[str] = Field(None, min_length=1, max_length=200) + description: Optional[str] = None + trigger_type: Optional[str] = None + cron_expression: Optional[str] = None + event_condition: Optional[Dict[str, Any]] = None + action_type: Optional[str] = None + action_config: Optional[Dict[str, Any]] = None + is_enabled: Optional[bool] = None + max_retries: Optional[int] = Field(None, ge=0, le=10) + timeout_seconds: Optional[int] = Field(None, ge=10, le=3600) + + +# ============================================================ +# Helper Functions +# ============================================================ + +TRIGGER_TYPE_MAP = { + "cron": "定时触发", + "webhook": "Webhook", + "event": "事件触发", +} + +ACTION_TYPE_MAP = { + "ai_pipeline": "AI流水线", + "webhook": "Webhook回调", + "notification": "发送通知", + "skill_execute": "技能执行", +} + + +def task_to_dict(task) -> dict: + """将AutomatedTask模型转换为字典""" + result = { + "id": task.id, + "name": task.name, + "description": getattr(task, "description", "") or "", + "trigger_type": task.trigger_type, + "trigger_type_label": TRIGGER_TYPE_MAP.get(task.trigger_type, task.trigger_type), + "cron_expression": getattr(task, "cron_expression", "") or "", + "action_type": task.action_type, + "action_type_label": ACTION_TYPE_MAP.get(task.action_type, task.action_type), + "is_enabled": task.is_enabled, + "status": task.status or "idle", + "max_retries": getattr(task, "max_retries", 3), + "timeout_seconds": getattr(task, "timeout_seconds", 300), + "last_run_at": task.last_run_at.isoformat() if task.last_run_at else None, + "next_run_at": task.next_run_at.isoformat() if task.next_run_at else None, + "total_runs": getattr(task, "total_runs", 0), + "success_runs": getattr(task, "success_runs", 0), + "created_at": task.created_at.isoformat() if task.created_at else None, + "updated_at": task.updated_at.isoformat() if task.updated_at else None, + } + + # Action config + action_config = getattr(task, "action_config", None) + if action_config: + try: + result["action_config"] = json.loads(action_config) if isinstance(action_config, str) else action_config + except Exception: + result["action_config"] = {} + else: + result["action_config"] = {} + + # Event condition + event_condition = getattr(task, "event_condition", None) + if event_condition: + try: + result["event_condition"] = json.loads(event_condition) if isinstance(event_condition, str) else event_condition + except Exception: + result["event_condition"] = {} + else: + result["event_condition"] = {} + + return result + + +def log_to_dict(log) -> dict: + """将TaskExecutionLog转换为字典""" + return { + "id": log.id, + "task_id": log.task_id, + "task_name": getattr(log, "task_name", ""), + "status": log.status, + "trigger_type": getattr(log, "trigger_type", ""), + "started_at": log.started_at.isoformat() if log.started_at else None, + "finished_at": log.finished_at.isoformat() if log.finished_at else None, + "duration_ms": getattr(log, "duration_ms", 0), + "result_summary": getattr(log, "result_summary", "") or "", + "error_message": getattr(log, "error_message", "") or "", + "created_at": log.created_at.isoformat() if log.created_at else None, + } + + +def calculate_next_run(cron_expression: str) -> Optional[datetime]: + """计算cron表达式的下次执行时间(简化版)""" + if not cron_expression: + return None + + parts = cron_expression.strip().split() + if len(parts) != 5: + return None + + now = datetime.now() + try: + minute, hour, day, month, weekday = parts + + # Handle wildcards + next_run = now.replace(second=0, microsecond=0) + + if minute != "*": + target_min = int(minute) + if target_min > now.minute: + next_run = next_run.replace(minute=target_min) + else: + next_run = next_run.replace(minute=target_min) + timedelta(hours=1) + + if hour != "*": + target_hour = int(hour) + if target_hour < next_run.hour or (target_hour == next_run.hour and next_run.minute <= now.minute): + next_run = next_run.replace(hour=target_hour, minute=int(minute) if minute != "*" else 0) + timedelta(days=1) + else: + next_run = next_run.replace(hour=target_hour) + + # Handle */N patterns for minute + if minute.startswith("*/"): + interval = int(minute[2:]) + next_minute = ((now.minute // interval) + 1) * interval + if next_minute >= 60: + next_run = next_run.replace(minute=0) + timedelta(hours=1) + else: + next_run = next_run.replace(minute=next_minute) + if hour != "*": + next_run = next_run.replace(hour=int(hour)) + + # Handle */N patterns for hour + if hour.startswith("*/"): + interval = int(hour[2:]) + next_hour = ((now.hour // interval) + 1) * interval + if next_hour >= 24: + next_run = next_run.replace(hour=0, minute=int(minute) if minute != "*" and not minute.startswith("*/") else 0) + timedelta(days=1) + else: + next_run = next_run.replace(hour=next_hour, minute=int(minute) if minute != "*" and not minute.startswith("*/") else 0) + + if next_run <= now: + next_run += timedelta(minutes=1) + + return next_run + except Exception as e: + logger.warning(f"Failed to calculate next run for '{cron_expression}': {e}") + return None + + +# ============================================================ +# API Endpoints +# ============================================================ + +@router.get("/stats/overview", summary="自动化任务统计概览") +async def get_automation_stats( + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取自动化任务的统计信息""" + from ...models import AutomatedTask, TaskExecutionLog + + total_tasks = db.query(func.count(AutomatedTask.id)).scalar() or 0 + enabled_tasks = db.query(func.count(AutomatedTask.id)).filter( + AutomatedTask.is_enabled == True + ).scalar() or 0 + running_tasks = db.query(func.count(AutomatedTask.id)).filter( + AutomatedTask.status == "running" + ).scalar() or 0 + + # By trigger type + by_trigger = {} + triggers = db.query(AutomatedTask.trigger_type, func.count(AutomatedTask.id)).group_by( + AutomatedTask.trigger_type + ).all() + for t, c in triggers: + by_trigger[t] = c + + # Recent execution stats (last 24h) + since = datetime.now() - timedelta(hours=24) + total_executions = db.query(func.count(TaskExecutionLog.id)).filter( + TaskExecutionLog.created_at >= since + ).scalar() or 0 + success_executions = db.query(func.count(TaskExecutionLog.id)).filter( + TaskExecutionLog.created_at >= since, + TaskExecutionLog.status == "success" + ).scalar() or 0 + + return { + "code": 0, + "data": { + "total_tasks": total_tasks, + "enabled_tasks": enabled_tasks, + "running_tasks": running_tasks, + "by_trigger": by_trigger, + "recent_executions": total_executions, + "recent_success": success_executions, + }, + "message": "ok" + } + + +@router.get("/", summary="获取自动化任务列表") +async def list_automation_tasks( + trigger_type: str = Query(None), + is_enabled: bool = Query(None), + keyword: str = Query(None), + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取自动化任务列表""" + from ...models import AutomatedTask + + query = db.query(AutomatedTask) + + if trigger_type: + query = query.filter(AutomatedTask.trigger_type == trigger_type) + if is_enabled is not None: + query = query.filter(AutomatedTask.is_enabled == is_enabled) + if keyword: + keyword_filter = or_( + AutomatedTask.name.ilike(f"%{keyword}%"), + AutomatedTask.description.ilike(f"%{keyword}%"), + ) + query = query.filter(keyword_filter) + + total = query.count() + items = query.order_by(desc(AutomatedTask.updated_at)) \ + .offset((page - 1) * page_size).limit(page_size).all() + + return { + "code": 0, + "data": { + "items": [task_to_dict(item) for item in items], + "total": total, + "page": page, + "page_size": page_size, + }, + "message": "ok" + } + + +@router.get("/{task_id}", summary="获取任务详情") +async def get_automation_task( + task_id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取单个自动化任务详情""" + from ...models import AutomatedTask + + task = db.query(AutomatedTask).filter(AutomatedTask.id == task_id).first() + if not task: + raise HTTPException(status_code=404, detail="任务不存在") + + return { + "code": 0, + "data": task_to_dict(task), + "message": "ok" + } + + +@router.post("/", summary="创建自动化任务") +async def create_automation_task( + req: AutomationTaskCreate, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """创建新的自动化任务""" + from ...models import AutomatedTask + + # 检查名称重复 + existing = db.query(AutomatedTask).filter(AutomatedTask.name == req.name).first() + if existing: + raise HTTPException(status_code=400, detail="任务名称已存在") + + # 验证cron表达式格式 + if req.trigger_type == "cron" and req.cron_expression: + parts = req.cron_expression.strip().split() + if len(parts) != 5: + raise HTTPException(status_code=400, detail="Cron表达式格式错误，需要5个字段") + + next_run = None + if req.trigger_type == "cron" and req.cron_expression: + next_run = calculate_next_run(req.cron_expression) + + task = AutomatedTask( + id=gen_uuid(), + name=req.name, + description=req.description or "", + trigger_type=req.trigger_type, + cron_expression=req.cron_expression, + action_type=req.action_type, + action_config=req.action_config, + is_enabled=req.is_enabled, + status="idle", + max_retries=req.max_retries, + timeout_seconds=req.timeout_seconds, + next_run_at=next_run, + event_condition=req.event_condition, + total_runs=0, + success_runs=0, + ) + + db.add(task) + db.commit() + db.refresh(task) + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="create_automation_task", + target_type="automated_task", + target_id=task.id, + detail=f"创建了自动化任务: {task.name}" + ) + + return { + "code": 0, + "data": task_to_dict(task), + "message": "创建成功" + } + + +@router.put("/{task_id}", summary="更新自动化任务") +async def update_automation_task( + task_id: str, + req: AutomationTaskUpdate, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """更新自动化任务""" + from ...models import AutomatedTask + + task = db.query(AutomatedTask).filter(AutomatedTask.id == task_id).first() + if not task: + raise HTTPException(status_code=404, detail="任务不存在") + + # 名称重复检查 + if req.name is not None and req.name != task.name: + existing = db.query(AutomatedTask).filter( + AutomatedTask.name == req.name, + AutomatedTask.id != task_id + ).first() + if existing: + raise HTTPException(status_code=400, detail="任务名称已存在") + + update_fields = ["name", "description", "trigger_type", "cron_expression", + "action_type", "is_enabled", "max_retries", "timeout_seconds"] + for field in update_fields: + value = getattr(req, field, None) + if value is not None: + setattr(task, field, value) + + if req.action_config is not None: + task.action_config = req.action_config + if req.event_condition is not None: + task.event_condition = req.event_condition + + # 重新计算下次执行时间 + if req.cron_expression is not None or req.trigger_type is not None: + cron = req.cron_expression if req.cron_expression is not None else task.cron_expression + trigger = req.trigger_type if req.trigger_type is not None else task.trigger_type + if trigger == "cron" and cron: + task.next_run_at = calculate_next_run(cron) + + task.updated_at = datetime.now() + + db.commit() + db.refresh(task) + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="update_automation_task", + target_type="automated_task", + target_id=task.id, + detail=f"更新了自动化任务: {task.name}" + ) + + return { + "code": 0, + "data": task_to_dict(task), + "message": "更新成功" + } + + +@router.delete("/{task_id}", summary="删除自动化任务") +async def delete_automation_task( + task_id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """删除自动化任务""" + from ...models import AutomatedTask + + task = db.query(AutomatedTask).filter(AutomatedTask.id == task_id).first() + if not task: + raise HTTPException(status_code=404, detail="任务不存在") + + name = task.name + db.delete(task) + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="delete_automation_task", + target_type="automated_task", + target_id=task_id, + detail=f"删除了自动化任务: {name}" + ) + + return { + "code": 0, + "data": None, + "message": f"已删除任务: {name}" + } + + +@router.put("/{task_id}/toggle", summary="启用/禁用自动化任务") +async def toggle_automation_task( + task_id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """切换任务启用/禁用状态""" + from ...models import AutomatedTask + + task = db.query(AutomatedTask).filter(AutomatedTask.id == task_id).first() + if not task: + raise HTTPException(status_code=404, detail="任务不存在") + + task.is_enabled = not task.is_enabled + if not task.is_enabled: + task.next_run_at = None + elif task.trigger_type == "cron" and task.cron_expression: + task.next_run_at = calculate_next_run(task.cron_expression) + + task.updated_at = datetime.now() + db.commit() + db.refresh(task) + + action = "启用" if task.is_enabled else "禁用" + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="toggle_automation_task", + target_type="automated_task", + target_id=task.id, + detail=f"{action}了自动化任务: {task.name}" + ) + + return { + "code": 0, + "data": task_to_dict(task), + "message": f"已{action}" + } + + +@router.post("/{task_id}/execute", summary="手动执行任务") +async def execute_task_now( + task_id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """立即手动执行一个自动化任务""" + from ...models import AutomatedTask, TaskExecutionLog + + task = db.query(AutomatedTask).filter(AutomatedTask.id == task_id).first() + if not task: + raise HTTPException(status_code=404, detail="任务不存在") + + if task.status == "running": + raise HTTPException(status_code=400, detail="任务正在执行中") + + # 创建执行日志 + exec_log = TaskExecutionLog( + id=gen_uuid(), + task_id=task.id, + task_name=task.name, + status="running", + trigger_type="manual", + started_at=datetime.now(), + ) + db.add(exec_log) + + # 更新任务状态 + task.status = "running" + task.total_runs = (task.total_runs or 0) + 1 + db.commit() + db.refresh(exec_log) + + # 异步执行(简化版: 直接模拟执行) + import time + start = time.time() + + try: + action_type = task.action_type + action_config = task.action_config or {} + + # 模拟执行不同类型的操作 + result_summary = "" + if action_type == "ai_pipeline": + result_summary = f"AI流水线执行完成: 处理了 {action_config.get('pipeline_name', 'default')} 管道" + elif action_type == "webhook": + url = action_config.get("url", "") + result_summary = f"Webhook调用完成: {url}" + elif action_type == "notification": + result_summary = f"通知发送完成: {action_config.get('channel', 'system')}" + elif action_type == "skill_execute": + result_summary = f"技能执行完成: {action_config.get('skill_name', 'unknown')}" + else: + result_summary = f"任务执行完成: {action_type}" + + duration = int((time.time() - start) * 1000) + + # 更新执行日志 + exec_log.status = "success" + exec_log.finished_at = datetime.now() + exec_log.duration_ms = duration + exec_log.result_summary = result_summary + + # 更新任务 + task.status = "idle" + task.last_run_at = datetime.now() + task.success_runs = (task.success_runs or 0) + 1 + if task.trigger_type == "cron" and task.cron_expression: + task.next_run_at = calculate_next_run(task.cron_expression) + + db.commit() + + return { + "code": 0, + "data": log_to_dict(exec_log), + "message": "执行成功" + } + + except Exception as e: + duration = int((time.time() - start) * 1000) + exec_log.status = "failed" + exec_log.finished_at = datetime.now() + exec_log.duration_ms = duration + exec_log.error_message = str(e) + task.status = "idle" + db.commit() + + raise HTTPException(status_code=500, detail=f"执行失败: {str(e)}") + + +@router.get("/{task_id}/history", summary="获取任务执行历史") +async def get_task_history( + task_id: str, + status: str = Query(None), + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取任务的执行历史记录""" + from ...models import TaskExecutionLog + + query = db.query(TaskExecutionLog).filter(TaskExecutionLog.task_id == task_id) + if status: + query = query.filter(TaskExecutionLog.status == status) + + total = query.count() + items = query.order_by(desc(TaskExecutionLog.created_at)) \ + .offset((page - 1) * page_size).limit(page_size).all() + + return { + "code": 0, + "data": { + "items": [log_to_dict(item) for item in items], + "total": total, + "page": page, + "page_size": page_size, + }, + "message": "ok" + } + + +@router.get("/history/all", summary="获取所有执行历史") +async def get_all_history( + status: str = Query(None), + task_id: str = Query(None), + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取所有任务的执行历史""" + from ...models import TaskExecutionLog + + query = db.query(TaskExecutionLog) + if status: + query = query.filter(TaskExecutionLog.status == status) + if task_id: + query = query.filter(TaskExecutionLog.task_id == task_id) + + total = query.count() + items = query.order_by(desc(TaskExecutionLog.created_at)) \ + .offset((page - 1) * page_size).limit(page_size).all() + + return { + "code": 0, + "data": { + "items": [log_to_dict(item) for item in items], + "total": total, + "page": page, + "page_size": page_size, + }, + "message": "ok" + } diff --git a/backend/python/app/models.py b/backend/python/app/models.py index e1624fb..c9ad3c0 100644 --- a/backend/python/app/models.py +++ b/backend/python/app/models.py @@ -998,3 +998,41 @@ class OSSStorageConfig(Base): config = Column(JSON, default=None) created_at = Column(DateTime, default=datetime.now) updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now) + +class AutomatedTask(Base): + __tablename__ = "automated_tasks" + + id = Column(String(36), primary_key=True, default=gen_uuid) + name = Column(String(200), nullable=False) + description = Column(Text, default="") + trigger_type = Column(String(20), nullable=False) + cron_expression = Column(String(100), default=None) + event_condition = Column(JSON, default=None) + action_type = Column(String(30), nullable=False) + action_config = Column(JSON, default=None) + is_enabled = Column(Boolean, default=True) + status = Column(String(20), default="idle") + max_retries = Column(Integer, default=3) + timeout_seconds = Column(Integer, default=300) + last_run_at = Column(DateTime, default=None) + next_run_at = Column(DateTime, default=None) + total_runs = Column(Integer, default=0) + success_runs = Column(Integer, default=0) + created_at = Column(DateTime, default=datetime.now) + updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now) + +class TaskExecutionLog(Base): + __tablename__ = "task_execution_logs" + + id = Column(String(36), primary_key=True, default=gen_uuid) + task_id = Column(String(36), nullable=False) + task_name = Column(String(200), default="") + status = Column(String(20), default="running") + trigger_type = Column(String(20), default="manual") + started_at = Column(DateTime, default=None) + finished_at = Column(DateTime, default=None) + duration_ms = Column(Integer, default=0) + result_summary = Column(Text, default="") + error_message = Column(Text, default="") + created_at = Column(DateTime, default=datetime.now) + diff --git a/backend/python/main.py b/backend/python/main.py index 13bbb55..28ee359 100644 --- a/backend/python/main.py +++ b/backend/python/main.py @@ -34,7 +34,7 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app.api import auth, chat, channels, knowledge, market, token_usage, payment, profile, notifications, assets, evolution -from app.api.admin import dashboard, users, models, settings, payment_admin, services, membership_admin, ads, market_admin, logs, skills_admin, agents_admin, knowledge_admin, storage_admin +from app.api.admin import dashboard, users, models, settings, payment_admin, services, membership_admin, ads, market_admin, logs, skills_admin, agents_admin, knowledge_admin, storage_admin, automation_admin from app.config import setup_file_logging, SLOW_REQUEST_THRESHOLD from app.database import SessionLocal @@ -420,6 +420,7 @@ app.include_router(skills_admin.router, prefix="/app/api/admin/skills", tags=[" app.include_router(agents_admin.router, prefix="/app/api/admin/agents", tags=["后台-Agent管理"]) app.include_router(knowledge_admin.router, prefix="/app/api/admin/knowledge", tags=["后台-知识库管理"]) app.include_router(storage_admin.router, prefix="/app/api/admin/storage", tags=["后台-存储配置"]) +app.include_router(automation_admin.router, prefix="/app/api/admin/automation", tags=["后台-自动化任务"]) diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index f6bc76d..6ebcab1 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -73,7 +73,8 @@ import AdminLogs from '@/pages/admin/Logs'; import AdminSkills from '@/pages/admin/Skills' import AdminAgents from '@/pages/admin/Agents' import AdminKnowledge from '@/pages/admin/Knowledge'; -import AdminStorage from '@/pages/admin/Storage';; +import AdminStorage from '@/pages/admin/Storage'; +import AdminAutomation from '@/pages/admin/Automation';; @@ -233,6 +234,7 @@ const App: React.FC = () => { } /> } /> } /> + } /> diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index f5463ea..49b5505 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -512,6 +512,23 @@ export const adminAPI = { setDefaultAdminStorageConfig: (id: string) => put(`/admin/storage/${id}/set-default`), testAdminStorageConfig: (data: any) => post('/admin/storage/test', data), + // 自动化任务调度 (Phase 3.2) + getAdminAutomationStats: () => get('/admin/automation/stats/overview'), + getAdminAutomationTasks: (params?: any) => { + const query = params ? '?' + Object.entries(params).filter(([_, v]) => v !== undefined && v !== '').map(([k, v]) => `${k}=${v}`).join('&') : ''; + return get(`/admin/automation/${query}`); + }, + getAdminAutomationTask: (id: string) => get(`/admin/automation/${id}`), + createAdminAutomationTask: (data: any) => post('/admin/automation/', data), + updateAdminAutomationTask: (id: string, data: any) => put(`/admin/automation/${id}`, data), + deleteAdminAutomationTask: (id: string) => del(`/admin/automation/${id}`), + toggleAdminAutomationTask: (id: string) => put(`/admin/automation/${id}/toggle`), + executeAdminAutomationTask: (id: string) => post(`/admin/automation/${id}/execute`), + getAdminAutomationTaskHistory: (id: string, params?: any) => { + const query = params ? '?' + Object.entries(params).filter(([_, v]) => v !== undefined && v !== '').map(([k, v]) => `${k}=${v}`).join('&') : ''; + return get(`/admin/automation/${id}/history${query}`); + }, + }; // ===== AI自进化API ===== export const evolutionAPI = { diff --git a/frontend/src/components/layout/AdminLayout.tsx b/frontend/src/components/layout/AdminLayout.tsx index 66c7eb6..b292d1f 100644 --- a/frontend/src/components/layout/AdminLayout.tsx +++ b/frontend/src/components/layout/AdminLayout.tsx @@ -35,6 +35,7 @@ const adminNavItems = [ { path: '/admin/agents', icon: 'agents', label: 'Agent管理' }, { path: '/admin/knowledge', icon: 'knowledge', label: '知识库管理' }, { path: '/admin/storage', icon: 'storage', label: '存储配置' }, + { path: '/admin/automation', icon: 'automation', label: '自动化任务' }, { path: '/admin/skills', icon: 'skills', label: '技能管理' },]; @@ -64,6 +65,7 @@ const adminIconPaths: Record = { skills: , + automation: , storage: , }; diff --git a/frontend/src/pages/admin/Automation.tsx b/frontend/src/pages/admin/Automation.tsx new file mode 100644 index 0000000..efee622 --- /dev/null +++ b/frontend/src/pages/admin/Automation.tsx @@ -0,0 +1,599 @@ +import React, { useState, useEffect } from 'react'; +import { adminAPI } from '@/api'; + +interface AutomationTask { + id: string; + name: string; + description: string; + trigger_type: string; + trigger_type_label: string; + cron_expression: string; + action_type: string; + action_type_label: string; + action_config: Record; + event_condition: Record; + is_enabled: boolean; + status: string; + max_retries: number; + timeout_seconds: number; + last_run_at: string; + next_run_at: string; + total_runs: number; + success_runs: number; + created_at: string; + updated_at: string; +} + +interface ExecLog { + id: string; + task_id: string; + task_name: string; + status: string; + trigger_type: string; + started_at: string; + finished_at: string; + duration_ms: number; + result_summary: string; + error_message: string; + created_at: string; +} + +interface AutomationStats { + total_tasks: number; + enabled_tasks: number; + running_tasks: number; + by_trigger: Record; + recent_executions: number; + recent_success: number; +} + +const TRIGGER_CONFIG: Record = { + cron: { label: '定时触发', color: '#6C63FF', bg: '#EDE9FE' }, + webhook: { label: 'Webhook', color: '#059669', bg: '#D1FAE5' }, + event: { label: '事件触发', color: '#D97706', bg: '#FEF3C7' }, +}; + +const STATUS_CONFIG: Record = { + idle: { label: '空闲', color: '#6B7280', bg: '#F3F4F6' }, + running: { label: '运行中', color: '#3B82F6', bg: '#DBEAFE' }, + success: { label: '成功', color: '#10B981', bg: '#D1FAE5' }, + failed: { label: '失败', color: '#EF4444', bg: '#FEE2E2' }, +}; + +const PAGE_SIZE = 15; + +const Automation: React.FC = () => { + // Data state + const [tasks, setTasks] = useState([]); + const [total, setTotal] = useState(0); + const [page, setPage] = useState(1); + const [loading, setLoading] = useState(false); + const [stats, setStats] = useState(null); + + // Filter state + const [filterTrigger, setFilterTrigger] = useState(''); + const [keyword, setKeyword] = useState(''); + + // Modal state + const [modalVisible, setModalVisible] = useState(false); + const [editingTask, setEditingTask] = useState(null); + const [formData, setFormData] = useState({ + name: '', + description: '', + trigger_type: 'cron', + cron_expression: '', + event_condition_str: '', + action_type: 'ai_pipeline', + action_config_str: '{"pipeline_name": "default"}', + is_enabled: true, + max_retries: 3, + timeout_seconds: 300, + }); + const [submitting, setSubmitting] = useState(false); + + // Execution history + const [historyVisible, setHistoryVisible] = useState(false); + const [historyTaskId, setHistoryTaskId] = useState(''); + const [historyTaskName, setHistoryTaskName] = useState(''); + const [historyLogs, setHistoryLogs] = useState([]); + const [historyPage, setHistoryPage] = useState(1); + const [historyTotal, setHistoryTotal] = useState(0); + const [executingId, setExecutingId] = useState(null); + + // Toast + const [toast, setToast] = useState<{ msg: string; type: 'success' | 'error' } | null>(null); + const showToast = (msg: string, type: 'success' | 'error' = 'success') => { + setToast({ msg, type }); + setTimeout(() => setToast(null), 3000); + }; + + // Load stats + const loadStats = async () => { + try { + const result = await adminAPI.getAdminAutomationStats() as any; + setStats(result.data || null); + } catch (e) { /* ignore */ } + }; + + // Load tasks + const loadTasks = async () => { + setLoading(true); + try { + const params: any = { page, page_size: PAGE_SIZE }; + if (filterTrigger) params.trigger_type = filterTrigger; + if (keyword) params.keyword = keyword; + const result = await adminAPI.getAdminAutomationTasks(params) as any; + const data = result.data || result; + setTasks(data.items || []); + setTotal(data.total || 0); + } catch (e: any) { + showToast('加载失败: ' + (e.message || e), 'error'); + } finally { + setLoading(false); + } + }; + + useEffect(() => { loadStats(); }, []); + useEffect(() => { loadTasks(); }, [page, filterTrigger, keyword]); + + // Open create modal + const openCreateModal = () => { + setEditingTask(null); + setFormData({ + name: '', + description: '', + trigger_type: 'cron', + cron_expression: '*/5 * * * *', + event_condition_str: '', + action_type: 'ai_pipeline', + action_config_str: '{"pipeline_name": "default"}', + is_enabled: true, + max_retries: 3, + timeout_seconds: 300, + }); + setModalVisible(true); + }; + + // Open edit modal + const openEditModal = async (task: AutomationTask) => { + try { + const result = await adminAPI.getAdminAutomationTask(task.id) as any; + const detail = result.data; + setEditingTask(task); + setFormData({ + name: detail.name || '', + description: detail.description || '', + trigger_type: detail.trigger_type || 'cron', + cron_expression: detail.cron_expression || '', + event_condition_str: detail.event_condition ? JSON.stringify(detail.event_condition, null, 2) : '', + action_type: detail.action_type || 'ai_pipeline', + action_config_str: detail.action_config ? JSON.stringify(detail.action_config, null, 2) : '{}', + is_enabled: detail.is_enabled !== false, + max_retries: detail.max_retries || 3, + timeout_seconds: detail.timeout_seconds || 300, + }); + setModalVisible(true); + } catch (e: any) { + showToast('获取任务详情失败', 'error'); + } + }; + + // Submit + const handleSubmit = async () => { + if (!formData.name.trim()) { showToast('请输入任务名称', 'error'); return; } + if (formData.trigger_type === 'cron' && !formData.cron_expression.trim()) { + showToast('请输入Cron表达式', 'error'); return; + } + + let actionConfig: any = {}; + if (formData.action_config_str.trim()) { + try { actionConfig = JSON.parse(formData.action_config_str); } + catch { showToast('操作配置JSON格式不正确', 'error'); return; } + } + + let eventCondition: any = undefined; + if (formData.event_condition_str.trim()) { + try { eventCondition = JSON.parse(formData.event_condition_str); } + catch { showToast('事件条件JSON格式不正确', 'error'); return; } + } + + setSubmitting(true); + try { + const data: any = { + name: formData.name.trim(), + description: formData.description.trim(), + trigger_type: formData.trigger_type, + cron_expression: formData.cron_expression.trim(), + action_type: formData.action_type, + action_config: actionConfig, + is_enabled: formData.is_enabled, + max_retries: formData.max_retries, + timeout_seconds: formData.timeout_seconds, + }; + if (eventCondition) data.event_condition = eventCondition; + + if (editingTask) { + await adminAPI.updateAdminAutomationTask(editingTask.id, data); + showToast('更新成功'); + } else { + await adminAPI.createAdminAutomationTask(data); + showToast('创建成功'); + } + + setModalVisible(false); + loadTasks(); + loadStats(); + } catch (e: any) { + showToast('操作失败: ' + (e.message || e), 'error'); + } finally { + setSubmitting(false); + } + }; + + // Delete + const handleDelete = async (task: AutomationTask) => { + if (!window.confirm(`确定要删除任务 "${task.name}" 吗？`)) return; + try { + await adminAPI.deleteAdminAutomationTask(task.id); + showToast('删除成功'); + loadTasks(); + loadStats(); + } catch (e: any) { + showToast('删除失败: ' + (e.message || e), 'error'); + } + }; + + // Toggle + const handleToggle = async (task: AutomationTask) => { + try { + await adminAPI.toggleAdminAutomationTask(task.id); + showToast(task.is_enabled ? '已禁用' : '已启用'); + loadTasks(); + loadStats(); + } catch (e: any) { + showToast('操作失败: ' + (e.message || e), 'error'); + } + }; + + // Execute now + const handleExecuteNow = async (task: AutomationTask) => { + setExecutingId(task.id); + try { + const result = await adminAPI.executeAdminAutomationTask(task.id) as any; + showToast(result.message || '执行完成'); + loadTasks(); + loadStats(); + } catch (e: any) { + showToast('执行失败: ' + (e.message || e), 'error'); + } finally { + setExecutingId(null); + } + }; + + // Show history + const showHistory = async (task: AutomationTask, pageNum: number = 1) => { + setHistoryTaskId(task.id); + setHistoryTaskName(task.name); + setHistoryPage(pageNum); + setHistoryVisible(true); + + try { + const result = await adminAPI.getAdminAutomationTaskHistory(task.id, { page: pageNum, page_size: 20 }) as any; + const data = result.data || result; + setHistoryLogs(data.items || []); + setHistoryTotal(data.total || 0); + } catch (e) { /* ignore */ } + }; + + // Format time + const formatTime = (ts: string) => { + if (!ts) return '-'; + try { return new Date(ts).toLocaleString('zh-CN'); } + catch { return ts; } + }; + + // Format duration + const formatDuration = (ms: number) => { + if (!ms) return '-'; + if (ms < 1000) return `${ms}ms`; + return `${(ms / 1000).toFixed(1)}s`; + }; + + const labelStyle: React.CSSProperties = { display: 'block', marginBottom: '4px', fontSize: '13px', color: '#374151', fontWeight: 500 }; + const inputStyle: React.CSSProperties = { width: '100%', padding: '8px 12px', borderRadius: '6px', border: '1px solid #D1D5DB', fontSize: '14px', boxSizing: 'border-box' }; + const btnStyle = (bg: string, disabled = false): React.CSSProperties => ({ + padding: '8px 20px', borderRadius: '6px', border: 'none', background: bg, + cursor: disabled ? 'not-allowed' : 'pointer', fontSize: '14px', color: '#fff', + opacity: disabled ? 0.7 : 1, + }); + + return ( +
+
+

自动化任务调度

+ +
+ + {/* Stats Cards */} + {stats && ( +
+
+
{stats.total_tasks}
+
总任务数
+
+
+
{stats.enabled_tasks}
+
已启用
+
+
+
{stats.running_tasks}
+
运行中
+
+
+
{stats.recent_executions}
+
24小时执行
+
+
+
{stats.recent_success}
+
24小时成功
+
+
+ )} + + {/* Filter Bar */} +
+ + { setKeyword(e.target.value); setPage(1); }} + style={{ padding: '8px 12px', borderRadius: '6px', border: '1px solid #D1D5DB', fontSize: '14px', flex: 1, maxWidth: '400px' }} /> +
+ + {/* Table */} +
+ + + + + + + + + + + + + + + {loading ? ( + + ) : tasks.length === 0 ? ( + + ) : ( + tasks.map(task => ( + + + + + + + + + + + )) + )} + +
任务名称触发方式操作类型状态上次执行下次执行成功率操作
加载中...
暂无自动化任务
+
{task.name}
+ {task.description &&
{task.description}
} + 		+ + {(TRIGGER_CONFIG[task.trigger_type] || {}).label || task.trigger_type} + + {task.cron_expression &&
{task.cron_expression}
} + 		{task.action_type_label} + + {(STATUS_CONFIG[task.status] || {}).label || task.status} + + {formatTime(task.last_run_at)}{formatTime(task.next_run_at)} + {task.total_runs > 0 + ? = 0.9 ? '#10B981' : '#D97706' }}> + {Math.round(task.success_runs / task.total_runs * 100)}% + ({task.success_runs}/{task.total_runs}) + + : -} + +
+ + + + + +
+
+
+ + {/* Pagination */} + {total > PAGE_SIZE && ( +
+ + 第 {page} / {Math.ceil(total / PAGE_SIZE)} 页 (共 {total} 条) + +
+ )} + + {/* Create/Edit Modal */} + {modalVisible && ( +
+
+

{editingTask ? '编辑任务' : '创建自动化任务'}

+
+
+ setFormData({ ...formData, name: e.target.value })} style={inputStyle} placeholder="如: 每日报表生成" />
+ +
+ setFormData({ ...formData, description: e.target.value })} style={inputStyle} placeholder="可选描述" />
+ +
+
+
+
+
+
+ + {formData.trigger_type === 'cron' && ( +
+ setFormData({ ...formData, cron_expression: e.target.value })} + style={{ ...inputStyle, fontFamily: 'monospace' }} placeholder="*/5 * * * * (每5分钟)" /> +
+ 格式: 分 时 日 月 周 | 示例: 0 9 * * * (每天9点) +
+ )} + + {formData.trigger_type !== 'cron' && ( +
+