diff --git a/backend/app/main.py b/backend/app/main.py
index eb9fbce..d172408 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -1,9 +1,12 @@
import asyncio
+from hashlib import md5
+import json
import uvicorn
import uuid
import time
import models.db as DB
import nodes as F
+from tortoise.transactions import in_transaction
from fastapi import FastAPI, Depends, HTTPException
from models.schemas.system import InitStatusResponse, InitConfigRequest, BaseResponse
from models.schemas.user import LoginRequest, LoginResponse
@@ -46,10 +49,123 @@ async def is_init():
return init_status
+async def _create_demo_records(app: FastAPI):
+ """创建测试用的"""
+ if await DB.Work.exists():
+ return
+ async with in_transaction():
+ work = DB.Work(
+ uuid=uuid.uuid4().hex,
+ creator_id=1,
+ work_title="pai-beta",
+ work_desc="大白智护beta环境",
+ )
+
+ await work.save()
+ work_content = DB.WorkContent(
+ user_id=work.creator_id,
+ work_id=work.id,
+ content=[
+ {
+ "type": "git",
+ "title": "同步前端仓库",
+ "description": "clone 分支 pai-beta-build",
+ "uuid": uuid.uuid4().hex,
+ "detail": {
+ "host": "58.214.239.10",
+ "user": "dgx07",
+ "password": "f=bHt/.7kRo:KIXq",
+ "port": 6378,
+ "sudo": True,
+ "git_path": "/data/wrok-flow/dbpai-web",
+ "git_user": "liruwei",
+ "git_password": "Wsad.8246",
+ "git_url": "https://tencentgit.dabby.com.cn/dbpai/dbpai-web.git",
+ "git_branch": "pai-beta-build",
+ "git_clone_type": "http/https",
+ "git_clone_depth": 1,
+ },
+ },
+ {
+ "type": "git",
+ "title": "同步后端仓库",
+ "description": "clone 分支 v3.4.2",
+ "uuid": uuid.uuid4().hex,
+ "detail": {
+ "host": "58.214.239.10",
+ "user": "dgx07",
+ "password": "f=bHt/.7kRo:KIXq",
+ "port": 6378,
+ "sudo": True,
+ "git_path": "/data/wrok-flow/dbpai-server",
+ "git_user": "liruwei",
+ "git_password": "Wsad.8246",
+ "git_url": "https://tencentgit.dabby.com.cn/dbpai/dbpai-server.git",
+ "git_branch": "v3.4.2",
+ "git_clone_type": "http/https",
+ "git_clone_depth": 1,
+ },
+ },
+ {
+ "type": "git",
+ "title": "同步项目集合仓库",
+ "description": "clone 分支 v3.4.2",
+ "uuid": uuid.uuid4().hex,
+ "detail": {
+ "host": "58.214.239.10",
+ "user": "dgx07",
+ "password": "f=bHt/.7kRo:KIXq",
+ "port": 6378,
+ "sudo": True,
+ "git_path": "/data/wrok-flow/dbpai-jenkins",
+ "git_user": "liruwei",
+ "git_password": "Wsad.8246",
+ "git_url": "https://tencentgit.dabby.com.cn/dbpai/dbpai-jenkins.git",
+ "git_branch": "v3.4.2",
+ "git_clone_type": "http/https",
+ "git_clone_depth": 1,
+ },
+ },
+ {
+ "type": "cmd",
+ "title": "更新项目集合仓库",
+ "description": "更新 server/ html/ 内容",
+ "uuid": uuid.uuid4().hex,
+ "detail": {
+ "host": "58.214.239.10",
+ "user": "dgx07",
+ "password": "f=bHt/.7kRo:KIXq",
+ "port": 6378,
+ "sudo": True,
+ "cmd": ("cd /data/wrok-flow/dbpai-jenkins" " && rm -fr ./server" " && cp -r ../dbpai-server ./server" " && rm -fr ./html" " && cp -r ../dbpai-web/dist ./html" " && git add ." ' && git commit -m "build"' " && git push"),
+ },
+ },
+ {
+ "type": "cmd",
+ "title": "部署",
+ "description": "重新运行 pai-beta 分支 v3.4.2",
+ "uuid": uuid.uuid4().hex,
+ "detail": {
+ "host": "58.214.239.10",
+ "user": "dgx07",
+ "password": "f=bHt/.7kRo:KIXq",
+ "port": 6378,
+ "sudo": True,
+ "cmd": ("cd /data/PAI/dbpai-beta/dbpai-jenkins;" " git stash;" " git pull origin v3.4.2;" " git stash pop;" " docker-compose restart;"),
+ },
+ },
+ ],
+ )
+ await work_content.save()
+ logger.debug("创建")
+
+
@asynccontextmanager
async def lifespan(app: FastAPI):
tasks = {}
+ task_nodes = {}
app.state.tasks = tasks
+ app.state.task_nodes = task_nodes
# 数据库
if await is_init():
init_config = await load_init_config()
@@ -60,6 +176,8 @@ async def lifespan(app: FastAPI):
scheduler.start()
logger.info("初始化定时器")
+ await _create_demo_records(app)
+
yield
for _, v in tasks.items():
@@ -161,109 +279,94 @@ def create_app():
description="执行命令",
response_model=BaseResponse,
)
- async def flow_run(id: int):
+ async def flow_run(work_uuid: str):
+ work = await DB.Work.filter(uuid=work_uuid).first()
+ if not work:
+ return BaseResponse(success=False, message="任务不存在")
- task_id = uuid.uuid4().hex
+ task_id = work_uuid
+ if task_id in app.state.tasks:
+ return BaseResponse(success=False, message="已有任务正在执行...")
- async def _task(task_id):
+ async def _update_node_status(task_id: str, node_id: str, status: str):
+ task_nodes = app.state.task_nodes
+ logger.info(f"task:{task_id} node:{node_id} status:{status}")
+ if task_id in task_nodes and node_id in task_nodes[task_id]:
+ task_nodes[task_id][node_id] = status
+
+ async def _task(task_id: str, work: DB.Work):
try:
+ work_content = await DB.WorkContent.filter(work_id=work.id).first()
+ if not work_content:
+ logger.warning(f"没有找到对应的内容: {work.uuid}")
+ return
nodes = []
- node_0 = F.GitNode()
- node_0.set_id("01K3JHSEGBHHAVQ92RAH8M6GPX")
- nodes.append(node_0)
- node_1 = F.GitNode()
- node_1.set_id("01K3JHSHKESMWYEGSY4VNCJY4D")
- nodes.append(node_1)
- node_2 = F.GitNode()
- node_2.set_id("01K3T4GTWZJF3RT5FFH25D3T9C")
- nodes.append(node_2)
- node_3 = F.CmdNode()
- node_3.set_id("01K3T4S2HN97T1B0TPFBCFSB00")
- nodes.append(node_3)
+ params = {}
+ status = {}
+ for x in work_content.content:
+ node_id = x["uuid"]
+ node_type = x["type"]
+ node_info = x["detail"]
+ params[node_id] = node_info
+ status[node_id] = "wait"
+ if node_type == "git":
+ node = F.GitNode()
+ node.set_info(node_id, lambda k, v: _update_node_status(task_id, k, v))
+ nodes.append(node)
+ elif node_type == "cmd":
+ node = F.CmdNode()
+ node.set_info(node_id, lambda k, v: _update_node_status(task_id, k, v))
+ nodes.append(node)
- logger.debug(app.state.tasks)
+ nodes.pop()
- # 设置依赖关系: node0 - "true" >> node1 - "true" >> node2
for i in range(len(nodes) - 1):
nodes[i] - "true" >> nodes[i + 1]
-
- flow = F.AsyncFlow(node_0)
- flow.set_params(
- {
- "01K3JHSEGBHHAVQ92RAH8M6GPX": {
- "host": "58.214.239.10",
- "user": "dgx07",
- "password": "f=bHt/.7kRo:KIXq",
- "port": 6378,
- "sudo": True,
- "git_path": "/data/wrok-flow/dbpai-web",
- "git_user": "liruwei",
- "git_password": "Wsad.8246",
- "git_url": "https://tencentgit.dabby.com.cn/dbpai/dbpai-web.git",
- "git_branch": "pai-beta-build",
- "git_clone_type": "http/https",
- "git_clone_depth": 1,
- },
- "01K3JHSHKESMWYEGSY4VNCJY4D": {
- "host": "58.214.239.10",
- "user": "dgx07",
- "password": "f=bHt/.7kRo:KIXq",
- "port": 6378,
- "sudo": True,
- "git_path": "/data/wrok-flow/dbpai-server",
- "git_user": "liruwei",
- "git_password": "Wsad.8246",
- "git_url": "https://tencentgit.dabby.com.cn/dbpai/dbpai-server.git",
- "git_branch": "v3.4.2",
- "git_clone_type": "http/https",
- "git_clone_depth": 1,
- },
- "01K3T4GTWZJF3RT5FFH25D3T9C": {
- "host": "58.214.239.10",
- "user": "dgx07",
- "password": "f=bHt/.7kRo:KIXq",
- "port": 6378,
- "sudo": True,
- "git_path": "/data/wrok-flow/dbpai-jenkins",
- "git_user": "liruwei",
- "git_password": "Wsad.8246",
- "git_url": "https://tencentgit.dabby.com.cn/dbpai/dbpai-jenkins.git",
- "git_branch": "v3.4.2",
- "git_clone_type": "http/https",
- "git_clone_depth": 1,
- },
- "01K3T4S2HN97T1B0TPFBCFSB00": {
- "host": "58.214.239.10",
- "user": "dgx07",
- "password": "f=bHt/.7kRo:KIXq",
- "port": 6378,
- "sudo": True,
- "cmd": "\n".join(
- [
- "cd /data/wrok-flow/dbpai-jenkins",
- "git log",
- ]
- ),
- },
- }
- )
+ flow = F.AsyncFlow(nodes[0])
+ flow.set_params(params)
shared = {}
+ app.state.task_nodes[task_id] = status
await flow.run_async(shared)
app.state.tasks.pop(task_id, None)
+ app.state.task_nodes.pop(task_id, None)
+
except asyncio.CancelledError:
logger.debug(f"任务被取消: {task_id}")
app.state.tasks.pop(task_id, None)
- app.state.tasks[task_id] = asyncio.create_task(_task(task_id))
+ 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)
if task:
task.cancel()
return BaseResponse()
+ @app.get(
+ "/api/tasks",
+ tags=["Flow"],
+ description="任务列表",
+ response_model=BaseResponse,
+ )
+ async def task_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_content_list = await DB.WorkContent.filter(work_id__in=[x.id for x in work_list]).all()
+ res = []
+ for work in work_list:
+ obj = work.json()
+ content = next((x for x in work_content_list if x.work_id == work.id), None)
+ if content:
+ obj["content"] = [{k: v for k, v in _c.items() if k != "detail"} for _c in (content.content or [])]
+ obj["is_running"] = work.uuid in global_tasks
+ obj["status"] = global_status[work.uuid] if work.uuid in global_status else {}
+ res.append(obj)
+ return BaseResponse(data=res)
+
return app
diff --git a/backend/app/nodes/cmd_node.py b/backend/app/nodes/cmd_node.py
index ec9c034..25c913c 100644
--- a/backend/app/nodes/cmd_node.py
+++ b/backend/app/nodes/cmd_node.py
@@ -1,3 +1,4 @@
+from fastapi import FastAPI
from core.pocket_flow import AsyncNode
from core.logger import logger
import asyncio
@@ -107,10 +108,14 @@ async def async_run_cmd(
class CmdNode(AsyncNode):
- def set_id(self, id):
+ def set_info(self, id, callback):
self.id = id
+ self.callback = callback
async def prep_async(self, shared):
+ if self.callback:
+ await self.callback(self.id, "exec")
+
return {
"cmd": self.params[self.id].get("cmd"),
"host": self.params[self.id].get("host"),
@@ -121,6 +126,9 @@ class CmdNode(AsyncNode):
}
async def exec_async(self, prep_res):
+ if self.callback:
+ await self.callback(self.id, "exec")
+
err = []
out = []
@@ -144,6 +152,11 @@ class CmdNode(AsyncNode):
async def post_async(self, shared, prep_res, exec_res):
shared[self.id] = exec_res
+ result = "true"
if exec_res.get("err"):
- return "false"
- return "true"
+ result = "false"
+
+ if self.callback:
+ await self.callback(self.id, "post")
+
+ return result
diff --git a/backend/app/nodes/git_node.py b/backend/app/nodes/git_node.py
index d5bf9f0..539ea4d 100644
--- a/backend/app/nodes/git_node.py
+++ b/backend/app/nodes/git_node.py
@@ -1,13 +1,18 @@
+from fastapi import FastAPI
from core.pocket_flow import AsyncNode
from core.logger import logger
from .cmd_node import async_run_cmd
from pathlib import Path
class GitNode(AsyncNode):
- def set_id(self, id):
+ def set_info(self, id, callback):
self.id = id
+ self.callback = callback
async def prep_async(self, shared):
+ if self.callback:
+ await self.callback(self.id, "prep")
+
return {
"client": {
"host": self.params[self.id].get("host"),
@@ -28,6 +33,9 @@ class GitNode(AsyncNode):
}
async def exec_async(self, prep_res):
+ if self.callback:
+ await self.callback(self.id, "exec")
+
client:dict = prep_res.get("client")
git:dict = prep_res.get("git")
@@ -85,11 +93,14 @@ class GitNode(AsyncNode):
return {"err": err_str, "out": out_str}
- return None
-
async def post_async(self, shared, prep_res, exec_res):
shared[self.id] = exec_res
+ result = "true"
if exec_res.get("err"):
- return "false"
- return "true"
+ result = "false"
+
+ if self.callback:
+ await self.callback(self.id, "post")
+
+ return result
diff --git a/frontend/src/pages/Flow/index.jsx b/frontend/src/pages/Flow/index.jsx
index 7ec3ea2..4985733 100644
--- a/frontend/src/pages/Flow/index.jsx
+++ b/frontend/src/pages/Flow/index.jsx
@@ -40,11 +40,11 @@ export default () => {
title: '操作',
valueType: 'option',
key: 'option',
- width: user.level === 0 ? 120 : 60,
+ width: user?.level === 0 ? 120 : 60,
fixed: 'right',
render: (_, record) => [
run(record)}>
-
+
,
user?.level === 0 && ,
user?.level === 0 && ,
diff --git a/frontend/src/pages/Layout.jsx b/frontend/src/pages/Layout.jsx
index 388cd84..88de368 100644
--- a/frontend/src/pages/Layout.jsx
+++ b/frontend/src/pages/Layout.jsx
@@ -100,7 +100,7 @@ const HeaderRight = ({ logout, navigate, user }) => {
style={{ border: 'none' }}
icon={}
>
- {user.username}
+ {user?.username}