feat: 执行脚本

This commit is contained in:
李如威 2025-09-13 17:12:43 +08:00
parent 2009f2b478
commit df7cb47cfd
3 changed files with 121 additions and 78 deletions

View File

@ -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

View File

@ -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)
}

View File

@ -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 <RunningStatus record={record} />
if (isRunning(record)) {
return <a>
<Space><LoadingOutlined />执行中</Space>
</a>
}
return "空闲"
}
@ -57,13 +83,13 @@ export default () => {
width: user?.level === 0 ? 120 : 60,
fixed: 'right',
render: (_, record) => [
!record?.is_running && (
<Popconfirm key='1' title="运行工作" onConfirm={() => run(record)}>
!isRunning(record) && (
<Popconfirm key='1' title="运行工作" onConfirm={async () => await run(record)}>
<Button type="link" style={{ padding: '0 0', border: 'none' }}>运行</Button>
</Popconfirm>
),
record?.is_running && (
<Popconfirm key='1' title="暂停工作" onConfirm={() => stop(record)}>
isRunning(record) && (
<Popconfirm key='1' title="暂停工作" onConfirm={async () => await stop(record)}>
<Button type="link" danger style={{ padding: '0 0', border: 'none' }}>暂停</Button>
</Popconfirm>
),
@ -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 (
<Card title="工作台">
<ProTable
@ -99,8 +160,13 @@ export default () => {
)
}
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 <>
<a onClick={() => setOpen(true)}>
<Space>
@ -118,22 +184,9 @@ const RunningStatus = ({record}) => {
>
<Steps
progressDot
current={1}
current={current}
direction="vertical"
items={[
{
title: 'Finished',
description: 'This is a description.',
},
{
title: <Space>In Progress<LoadingOutlined /></Space>,
description: 'This is a description.',
},
{
title: 'Waiting',
description: 'This is a description.',
},
]}
items={record?.content}
/>
</Modal>
</>