diff --git a/backend/app/main.py b/backend/app/main.py index 607355b..6ac5936 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -2,6 +2,7 @@ import asyncio import datetime import json import os +from pydoc import cli import uvicorn import uuid import time @@ -23,6 +24,7 @@ from contextlib import asynccontextmanager from fastapi.security import OAuth2PasswordBearer from fastapi.responses import StreamingResponse from apscheduler.schedulers.asyncio import AsyncIOScheduler +import traceback # 简单的登陆状态存储 sessions: dict[str, tuple[float, DB.User]] = {} @@ -42,7 +44,7 @@ def clear_session(): logger.info(f"清理过期 now: {now} session: {expired_keys}") -def save_session_to_file(app:FastAPI): +def save_session_to_file(app: FastAPI): if sessions: login_json = {} for k, v in sessions.items(): @@ -52,9 +54,9 @@ def save_session_to_file(app:FastAPI): json.dump(login_json, f, indent=4, ensure_ascii=False) -async def load_session_from_file(app:FastAPI): +async def load_session_from_file(app: FastAPI): if os.path.exists(".login.json"): - with open('.login.json', 'r', encoding='utf-8') as f: + with open(".login.json", "r", encoding="utf-8") as f: login_json = json.load(f) if login_json: users = await DB.User.filter(is_deleted=0).all() @@ -180,6 +182,16 @@ def create_app(): if task_id in app.state.tasks: return BaseResponse(success=False, message="已有任务正在执行...") + async def _append_host_config(detail: dict): + result = {**detail} + client = await DB.Client.filter(id=result.get("client", 0)).first() + if client: + result["host"] = client.host + result["user"] = client.user + result["password"] = client.password + result["port"] = client.port + return result + 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}") @@ -200,45 +212,42 @@ def create_app(): if not work_content: logger.warning(f"没有找到对应的内容: {work.uuid}") return + + # 找出从 “开始” 到最后的节点 + node_dict = {x["uuid"]: x for x in work_content.content} + node_list = [] + current = "1" + while current in node_dict: + t = node_dict[current] + node_list.append(t) + if "edge" in t and "target" in t["edge"]: + current = t["edge"]["target"] + else: + break + + # pocket flow 节点创建 nodes = [] params = {} + shared = {} status = {} - uuids = [] - first_uuid = next((x['uuid'] for x in work_content.content if x.get("type") == "first"), "") - if not first_uuid: - work_log.status = DB.WorkLogsStatus.FAILED - await work_log.save() - app.state.tasks.pop(task_id, None) - - - - for x in work_content.content: + for x in node_list: node_id = x["uuid"] node_type = x["type"] - node_info = x["detail"] - node_edge = x.get("edge") - params[node_id] = node_info + if node_id == "1": + continue + params[node_id] = await _append_host_config(x["detail"]) status[node_id] = "wait" - node_class_map = { - "git": F.GitNode, - "cmd": F.CmdNode, - "status": F.StatusNode - } + node_class_map = {"git": F.GitNode, "cmd": F.CmdNode, "status": F.StatusNode} node_cls = node_class_map[node_type] node = node_cls() node.set_info(node_id, lambda k, v: _update_node_status(task_id, k, v)) nodes.append(node) - # 测试,过滤部署节点 - build_index = next((i for i, x in enumerate(work_content.content) if x["title"] == "部署"), -1) - if build_index >= 0: - del nodes[build_index] - for i in range(len(nodes) - 1): nodes[i] - "true" >> nodes[i + 1] 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) @@ -257,6 +266,7 @@ def create_app(): logger.debug(f"任务被取消: {task_id}") app.state.tasks.pop(task_id, None) except Exception as e: + traceback.print_exc() logger.error(e) work_log.status = DB.WorkLogsStatus.FAILED await work_log.save() @@ -290,8 +300,8 @@ def create_app(): 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 [])] - log:DB.WorkLogs = next((x for x in work_logs_list if x.work_id == work.id), DB.WorkLogs(user_id=-1)) - log_user:DB.User = next((u for u in user_list if u.id==log.user_id), DB.User()) + log: DB.WorkLogs = next((x for x in work_logs_list if x.work_id == work.id), DB.WorkLogs(user_id=-1)) + log_user: DB.User = next((u for u in user_list if u.id == log.user_id), DB.User()) obj["last_user"] = log_user.username or "" obj["last_run_time"] = log.create_time.timestamp() if log.create_time else 0 obj["last_status"] = log.status or "" @@ -363,7 +373,7 @@ def create_app(): return BaseResponse() @app.put("/api/clients", tags=["Client"], description="终端修改", response_model=BaseResponse) - async def update_client(id:int, body: ClientRequest, user: DB.User = Depends(get_current_user)): + async def update_client(id: int, body: ClientRequest, user: DB.User = Depends(get_current_user)): obj = await DB.Client.filter(id=id).first() if obj: obj.update_from_dict(body.model_dump()) @@ -372,7 +382,7 @@ def create_app(): return BaseResponse() @app.delete("/api/clients", tags=["Client"], description="终端删除", response_model=BaseResponse) - async def del_client(id: int, user: DB.User = Depends(get_current_user)): + async def del_client(id: int, user: DB.User = Depends(get_current_user)): await DB.Client.filter(id=id).delete() return BaseResponse() diff --git a/backend/app/nodes/git_node.py b/backend/app/nodes/git_node.py index c82199d..4648008 100644 --- a/backend/app/nodes/git_node.py +++ b/backend/app/nodes/git_node.py @@ -28,7 +28,7 @@ class GitNode(AsyncNode): "branch": self.params[self.id].get("git_branch"), "path": self.params[self.id].get("git_path"), "clone_type": self.params[self.id].get("git_clone_type", "http/https"), - "depth": self.params[self.id].get("git_clone_depth", 1) + "depth": self.params[self.id].get("git_clone_depth", 1), }, } diff --git a/backend/app/nodes/status_node.py b/backend/app/nodes/status_node.py index 1edfc84..dbb2a33 100644 --- a/backend/app/nodes/status_node.py +++ b/backend/app/nodes/status_node.py @@ -21,7 +21,7 @@ class StatusNode(AsyncNode): "password": self.params[self.id].get("password"), "port": self.params[self.id].get("port"), "sudo": self.params[self.id].get("sudo", False), - "cmd": "\n".join( + "cmd": " \ \n".join( [ f'status_code=$(curl -s -o /dev/null -w "%{{http_code}}" {status_url})', 'if [ "$status_code" -eq 200 ]; then', diff --git a/frontend/src/pages/Flow/components/EditNode.jsx b/frontend/src/pages/Flow/components/EditNode.jsx index c351bd4..226a87b 100644 --- a/frontend/src/pages/Flow/components/EditNode.jsx +++ b/frontend/src/pages/Flow/components/EditNode.jsx @@ -41,6 +41,7 @@ const EditNode = forwardRef(({ onFinish }, ref) => { +