feat: 状态检测
This commit is contained in:
parent
63b1f0f10b
commit
fb646cd3f7
|
|
@ -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()
|
||||
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
},
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
|
|
|
|||
|
|
@ -41,6 +41,7 @@ const EditNode = forwardRef(({ onFinish }, ref) => {
|
|||
<ProFormGroup>
|
||||
<ProFormText name='uuid' disabled label='UUID' width='md'/>
|
||||
<ProFormSelect label="节点" rules={[{ required: true }]} name={["detail", "client"]} options={clients} request={onRequest} width='md' />
|
||||
<ProFormSelect label="是否 Root" rules={[{ required: true }]} name={["detail", "sudo"]} options={[{ "label": "True", "value": true }, { "label": "False", "value": false, }]} request={onRequest} width='md' />
|
||||
<ProFormText name='title' rules={[{ required: true }]} label='标题' fieldProps={{ maxLength: 20 }} width='md' />
|
||||
<ProFormText fieldProps={{ maxLength: 50 }} name='description' label='描述' width="md" />
|
||||
<div style={{ display: 'none'}}>
|
||||
|
|
|
|||
Loading…
Reference in New Issue