From c865132e1a5c5d4d17d3c42a8e479c92dc51a1a6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E5=A6=82=E5=A8=81?= Date: Wed, 10 Sep 2025 10:45:56 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8E=A5=E5=8F=A3=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/main.py | 261 +++++++++++++++++++++--------- backend/app/nodes/cmd_node.py | 19 ++- backend/app/nodes/git_node.py | 21 ++- frontend/src/pages/Flow/index.jsx | 4 +- frontend/src/pages/Layout.jsx | 2 +- 5 files changed, 217 insertions(+), 90 deletions(-) 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}