diff --git a/README.md b/README.md index 791de5d..189bf4f 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,11 @@ ## 环境准备 +```shell +# python +nvm local 3.13.0 +python -m venv venv +source venv/bin/activate + +# node +nvm use v20.18.0 +``` \ No newline at end of file diff --git a/backend/app/main.py b/backend/app/main.py index a36095e..eb9fbce 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -1,3 +1,4 @@ +import asyncio import uvicorn import uuid import time @@ -23,9 +24,7 @@ scheduler = AsyncIOScheduler() def clear_session(): TTL = 60 * 60 * 24 # 1天有效期 now = time.time() - expired_keys = [ - token for token, (expires_at, _) in sessions.items() if expires_at + TTL < now - ] + expired_keys = [token for token, (expires_at, _) in sessions.items() if expires_at + TTL < now] for token in expired_keys: sessions.pop(token, None) @@ -49,12 +48,12 @@ async def is_init(): @asynccontextmanager async def lifespan(app: FastAPI): + tasks = {} + app.state.tasks = tasks # 数据库 if await is_init(): init_config = await load_init_config() - await init_tortoise( - init_config["db_type"], init_config, init_config.get("admin_password") - ) + await init_tortoise(init_config["db_type"], init_config, init_config.get("admin_password")) logger.info("初始化数据库") # 定时器 scheduler.add_job(clear_session, "interval", seconds=30, misfire_grace_time=10) @@ -63,6 +62,9 @@ async def lifespan(app: FastAPI): yield + for _, v in tasks.items(): + v.cancel() + scheduler.shutdown() logger.info("关闭定时器") @@ -108,9 +110,7 @@ def create_app(): init_config["admin_password"] = request.admin_password init_config["is_init"] = True - init_success = await init_tortoise( - request.db_type, init_config, request.admin_password - ) + init_success = await init_tortoise(request.db_type, init_config, request.admin_password) if init_success: await save_init_config(init_config) @@ -127,16 +127,12 @@ def create_app(): response_model=LoginResponse, ) async def login(login_info: LoginRequest): - user = await DB.User.filter( - username=login_info.username, password=login_info.get_md5_password() - ).first() + user = await DB.User.filter(username=login_info.username, password=login_info.get_md5_password(), is_deleted=0).first() if user: token = uuid.uuid4().hex sessions[token] = time.time(), user logger.info(sessions) - return LoginResponse( - data={"token": token, "user": user.json(del_columns=["password"])} - ) + return LoginResponse(data={"token": token, "user": user.json(del_columns=["password"])}) return LoginResponse(success=False, message="帐号或密码错误") @app.post( @@ -160,79 +156,112 @@ def create_app(): return LoginResponse(data=user.json(del_columns=["password"])) @app.post( - "/api/run", tags=["Flow"], description="执行命令", response_model=BaseResponse + "/api/run", + tags=["Flow"], + description="执行命令", + response_model=BaseResponse, ) async def flow_run(id: int): - nodes = [] - git_node_0 = F.GitNode() - git_node_0.set_id("01K3JHSEGBHHAVQ92RAH8M6GPX") - nodes.append(git_node_0) + task_id = uuid.uuid4().hex - git_node_1 = F.GitNode() - git_node_1.set_id("01K3JHSHKESMWYEGSY4VNCJY4D") - nodes.append(git_node_1) + async def _task(task_id): + try: + 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) - cmd_node_2 = F.CmdNode() - cmd_node_2.set_id("01K3JHSMYM584C63QB74G3Q6X4") - nodes.append(cmd_node_2) + logger.debug(app.state.tasks) - cmd_node_3 = F.CmdNode() - cmd_node_3.set_id("01K3JHS5HTY78Z4EYWRFW5TA77") - nodes.append(cmd_node_3) + # 设置依赖关系: node0 - "true" >> node1 - "true" >> node2 + for i in range(len(nodes) - 1): + nodes[i] - "true" >> nodes[i + 1] - # 设置依赖关系: 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", + ] + ), + }, + } + ) + shared = {} + await flow.run_async(shared) + app.state.tasks.pop(task_id, None) + except asyncio.CancelledError: + logger.debug(f"任务被取消: {task_id}") + app.state.tasks.pop(task_id, None) - flow = F.AsyncFlow(git_node_0) - - flow.set_params( - { - "01K3JHSEGBHHAVQ92RAH8M6GPX": { - "git_path": "/Users/liruwei/Documents/code/project/dabai/dbpai-web", - "git_user": "liruwei", - "git_password": "Wsad.8246", - "git_url": "https://tencentgit.dabby.com.cn/dbpai/dbpai-web.git", - "git_branch": "v3.4.2", - "git_clone_type": "http/https", - "git_clone_depth": 0, - }, - "01K3JHSHKESMWYEGSY4VNCJY4D": { - "git_path": "/Users/liruwei/Documents/code/project/dabai/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": 0, - }, - "01K3JHSMYM584C63QB74G3Q6X4": { - "cmd": "\n".join( - [ - "cd /Users/liruwei/Documents/code/project/dabai", - "cp ./dbpai-web/public/dbpai-chat-widget.min.js ./dbpai-server/src/static", - "echo '开始打包'", - "sh build-dbpai-jenkisn.sh all 2> /dev/null", - "echo '打包结束'", - "pwd", - "cd ./dbpai-jenkins", - "git add .", - "git commit -m 'feat: build'", - "git push", - ] - ) - }, - "01K3JHS5HTY78Z4EYWRFW5TA77": { - "cmd": "\n".join(["pwd"]), - }, - } - ) - shared = {} - await flow.run_async(shared) - - logger.info(shared) + app.state.tasks[task_id] = asyncio.create_task(_task(task_id)) + 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) + if task: + task.cancel() return BaseResponse() return app diff --git a/backend/app/models/db/__init__.py b/backend/app/models/db/__init__.py index db523d3..f9f1d0a 100644 --- a/backend/app/models/db/__init__.py +++ b/backend/app/models/db/__init__.py @@ -1,2 +1,3 @@ from .system import * from .user import * +from .work import * \ No newline at end of file diff --git a/backend/app/models/db/user.py b/backend/app/models/db/user.py index 4ceee3a..6c8a348 100644 --- a/backend/app/models/db/user.py +++ b/backend/app/models/db/user.py @@ -5,6 +5,7 @@ class User(BaseModel): username = fields.CharField(max_length=50, null=True) password = fields.CharField(max_length=128, null=True) level = fields.IntField(default=1, null=True, description="0:超级用户 1:普通用户") + is_deleted = fields.IntField(default=0, null=True) class Meta: table = "tb_users" diff --git a/backend/app/models/db/work.py b/backend/app/models/db/work.py new file mode 100644 index 0000000..783fed0 --- /dev/null +++ b/backend/app/models/db/work.py @@ -0,0 +1,49 @@ +from tortoise import fields +from .system import BaseModel +from enum import Enum + +class DBAnalysisFileStatus(str, Enum): + PROCESSING = "PROCESSING" # 处理中 + ERROR = "ERROR" # 处理失败 + DONE = "DONE" # 处理完成 + + +class Work(BaseModel): + uuid = fields.CharField(max_length=128, index=True) + creator_id = fields.IntField(null=True) + work_title = fields.CharField(max_length=128, null=True) + work_desc = fields.CharField(max_length=255, null=True) + is_deleted = fields.IntField(default=0, null=True) + + class Meta: + table = "tb_work" + + +class WorkLogsStatus(str, Enum): + DEFAULT = "DEFAULT" + RUNNING = "RUNNING" + SUCCESS = "SUCCESS" + FAILED = "FAILED" + + +class WorkLogs(BaseModel): + user_id = fields.IntField(null=True) + work_id = fields.IntField(index=True) + status = fields.CharField(max_length=128, null=True) + result = fields.JSONField(default={}) + work_title = fields.CharField(max_length=128, null=True) + work_desc = fields.CharField(max_length=255, null=True) + is_deleted = fields.IntField(default=0) + + class Meta: + table = "tb_work_logs" + + +class WorkContent(BaseModel): + user_id = fields.IntField(null=True) + work_id = fields.IntField(index=True) + content = fields.JSONField(default={}) + is_deleted = fields.IntField(default=0) + + class Meta: + table = "tb_work_content" diff --git a/backend/app/models/schemas/system.py b/backend/app/models/schemas/system.py index 7a38a23..f2f37f4 100644 --- a/backend/app/models/schemas/system.py +++ b/backend/app/models/schemas/system.py @@ -1,11 +1,11 @@ from pydantic import BaseModel - +from typing import Any class BaseResponse(BaseModel): + data: Any | None = None success: bool = True message: str | None = None - class InitConfigRequest(BaseModel): db_type: str = "sqlite" admin_password: str = "admin" diff --git a/backend/app/nodes/cmd_node.py b/backend/app/nodes/cmd_node.py index 48394c5..ec9c034 100644 --- a/backend/app/nodes/cmd_node.py +++ b/backend/app/nodes/cmd_node.py @@ -1,4 +1,5 @@ from core.pocket_flow import AsyncNode +from core.logger import logger import asyncio import asyncssh @@ -22,7 +23,6 @@ async def async_run_cmd( text = line if not isinstance(text, str): text = line.decode() - print(text) yield stream_type, text.rstrip() if not host: @@ -127,8 +127,10 @@ class CmdNode(AsyncNode): async for stream_type, line in async_run_cmd(**prep_res): if stream_type == "stderr": err.append(str(line)) + logger.debug(f'err: {line}') else: out.append(str(line)) + logger.debug(f"out: {line}") err_str = "\n".join(err) out_str = "\n".join(out) diff --git a/frontend/index.html b/frontend/index.html index 67cb05e..ddb4168 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -4,7 +4,7 @@ -