diff --git a/backend/app/main.py b/backend/app/main.py index 724ceda..e4aee76 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -15,6 +15,7 @@ from core.db import init_tortoise, close_tortoise from core.logger import logger from contextlib import asynccontextmanager from fastapi.security import OAuth2PasswordBearer +from fastapi.responses import StreamingResponse from apscheduler.schedulers.asyncio import AsyncIOScheduler # 简单的登陆状态存储 @@ -197,21 +198,11 @@ def create_app(): async def root(): return {"message": "hello"} - @app.get( - "/api/check_init", - tags=["System"], - description="系统是否初始化", - response_model=InitStatusResponse, - ) + @app.get("/api/check_init", tags=["System"], description="系统是否初始化", response_model=InitStatusResponse) async def check_init(): return InitStatusResponse(data=await is_init()) - @app.post( - "/api/init_system", - tags=["System"], - description="系统初始化", - response_model=InitStatusResponse, - ) + @app.post("/api/init_system", tags=["System"], description="系统初始化", response_model=InitStatusResponse) async def init_system(request: InitConfigRequest): logger.info("init_system") try: @@ -238,12 +229,7 @@ def create_app(): except Exception as e: return InitStatusResponse(success=False, message=str(e)) - @app.post( - "/api/login", - tags=["User"], - description="用户登录", - response_model=LoginResponse, - ) + @app.post("/api/login", tags=["User"], description="用户登录", response_model=LoginResponse) async def login(login_info: LoginRequest): user = await DB.User.filter(username=login_info.username, password=login_info.get_md5_password(), is_deleted=0).first() if user: @@ -253,32 +239,17 @@ def create_app(): return LoginResponse(data={"token": token, "user": user.json(del_columns=["password"])}) return LoginResponse(success=False, message="帐号或密码错误") - @app.post( - "/api/logout", - tags=["User"], - description="退出登录", - response_model=BaseResponse, - ) + @app.post("/api/logout", tags=["User"], description="退出登录", response_model=BaseResponse) async def logout(token: str): sessions.pop(token, None) logger.info(sessions) return BaseResponse() - @app.get( - "/api/current_user", - tags=["User"], - description="当前用户信息", - response_model=LoginResponse, - ) + @app.get("/api/current_user", tags=["User"], description="当前用户信息", response_model=LoginResponse) async def current_user(user: DB.User = Depends(get_current_user)): return LoginResponse(data=user.json(del_columns=["password"])) - @app.post( - "/api/run", - tags=["Flow"], - description="执行命令", - response_model=BaseResponse, - ) + @app.post("/api/run", tags=["Flow"], description="执行任务", response_model=BaseResponse) async def flow_run(work_uuid: str): work = await DB.Work.filter(uuid=work_uuid).first() if not work: @@ -337,24 +308,19 @@ def create_app(): app.state.tasks[task_id] = asyncio.create_task(_task(task_id, work)) return BaseResponse(data=task_id) - @app.post("/api/stop", tags=["Flow"], description="停止执行指令", response_model=BaseResponse) - async def stop_run(task_id: str): - task = app.state.tasks.pop(task_id, None) - app.state.task_nodes.pop(task_id, None) + @app.post("/api/stop", tags=["Flow"], description="停止任务", response_model=BaseResponse) + async def stop_run(work_uuid: str): + task = app.state.tasks.pop(work_uuid, None) + app.state.task_nodes.pop(work_uuid, None) if task: task.cancel() return BaseResponse() - @app.get( - "/api/flows", - tags=["Flow"], - description="任务列表", - response_model=TableResponse, - ) - async def task_list(): + @app.get("/api/flows", tags=["Flow"], description="任务列表", response_model=TableResponse) + async def flows_list(): global_tasks = app.state.tasks global_status = app.state.task_nodes - work_list = await DB.Work.filter(is_deleted=0).order_by('-id') + work_list = await DB.Work.filter(is_deleted=0).order_by("-id") work_content_list = await DB.WorkContent.filter(work_id__in=[x.id for x in work_list]).all() res = [] for work in work_list: @@ -367,6 +333,25 @@ def create_app(): res.append(obj) return TableResponse(data=res, total=len(res)) + @app.get("/api/flows/status", tags=["Flow"], description="获取任务状态") + async def flows_status(token: str = ""): + # 如果出现长时间链接被中断,可以设置 nginx 的 proxy_read_timeout + async def _event(): + while True: + try: + if not token or token not in sessions: + data = json.dumps({"code": 403}) + yield f"data: {data}\n\n" + break + + data = json.dumps(app.state.task_nodes) + yield f"data: {data}\n\n" + await asyncio.sleep(3) + except: + break + + return StreamingResponse(_event(), media_type="text/event-stream") + return app diff --git a/frontend/src/api/flow.js b/frontend/src/api/flow.js index bd1aba0..409282c 100644 --- a/frontend/src/api/flow.js +++ b/frontend/src/api/flow.js @@ -4,9 +4,14 @@ export const tableList = async (data = {}) => { return request("/api/flows?" + new URLSearchParams(data).toString(), "GET") } -export const run = async (data = {}) => { - return request("/api/current_user", "POST", data) +export const flowRun = async (id, data = {}) => { + return request("/api/run?work_uuid=" + id, "POST", data) +} + +export const flowStop = async (id, data = {}) => { + return request("/api/stop?work_uuid=" + id, "POST", data) } + diff --git a/frontend/src/pages/Flow/index.jsx b/frontend/src/pages/Flow/index.jsx index 86ab949..9426cf9 100644 --- a/frontend/src/pages/Flow/index.jsx +++ b/frontend/src/pages/Flow/index.jsx @@ -2,16 +2,40 @@ import { ProTable } from "@ant-design/pro-components" import { Button, Card, Flex, Modal, Popconfirm, Space, Steps, Typography } from "antd" import { useAuth } from "../../context/AuthContext"; import { LoadingOutlined } from "@ant-design/icons"; -import { useState } from "react"; -import { tableList } from "../../api/flow"; +import { useEffect, useRef, useState } from "react"; +import { tableList, flowRun, flowStop } from "../../api/flow"; export default () => { - const { user } = useAuth(); + const { user, logout } = useAuth(); + const eventSourceRef = useRef(null); + const pageInit = useRef(false); + const [status, setStatus] = useState({}); + const isRunning = (record) => { + return status?.[record.uuid] + } const run = async record => { - + try { + const res = await flowRun(record.uuid); + setStatus(r => { + const result = { ...r } + result[record.uuid] = {} + return result + }) + } catch (error) { + } + return true; } const stop = async record => { - + try { + const res = await flowStop(record.uuid); + setStatus(r => { + const result = { ...r } + result[record.uuid] = undefined + return result + }) + } catch (error) { + } + return true; } const columns = [ { @@ -32,8 +56,10 @@ export default () => { dataIndex: 'status', title: "状态", render: (_, record) => { - if (record?.is_running) { - return + if (isRunning(record)) { + return + 执行中 + } return "空闲" } @@ -57,13 +83,13 @@ export default () => { width: user?.level === 0 ? 120 : 60, fixed: 'right', render: (_, record) => [ - !record?.is_running && ( - run(record)}> + !isRunning(record) && ( + await run(record)}> ), - record?.is_running && ( - stop(record)}> + isRunning(record) && ( + await stop(record)}> ), @@ -73,7 +99,6 @@ export default () => { } ]; const onRequest = async (params) => { - console.log(params); try { const res = await tableList(); res?.data?.forEach((e, i) => { @@ -84,6 +109,42 @@ export default () => { return {} } } + // 状态检测 + useEffect(() => { + if (!pageInit.current) return; + if (!user?.token) return; + if (eventSourceRef.current) { + eventSourceRef.current.close(); + } + const url = "/api/flows/status?token=" + user?.token; + const es = new EventSource(url); + eventSourceRef.current = es; + es.onmessage = (e) => { + try { + const data = JSON.parse(e.data); + if (data?.code === 403) { + logout(); + } + setStatus({ ...data, time: new Date() }) + } catch (err) { + console.error("Failed to parse SSE data:", err); + } + }; + es.onerror = () => { + console.log(user); + es.close(); + setTimeout(() => { + eventSourceRef.current = new EventSource(url); + }, 3000); + }; + return () => { + eventSourceRef?.current?.close(); + }; + }, [user]); + + useEffect(() => { + pageInit.current = true; + }, []) return ( { ) } -const RunningStatus = ({record}) => { +const RunningStatus = ({record, status}) => { const [open, setOpen] = useState(false); + + const current = record?.content?.findIndex(e => status?.[e.uuid]); + + console.log(status, current); + return <> setOpen(true)}> @@ -118,22 +184,9 @@ const RunningStatus = ({record}) => { > In Progress, - description: 'This is a description.', - }, - { - title: 'Waiting', - description: 'This is a description.', - }, - ]} + items={record?.content} />