From bbffb1420bfaa023e6c4b1badd08861ffa51a15c Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 8 Jul 2025 15:18:01 +0000 Subject: [PATCH 1/5] Add workflow sharing, forking, and related API endpoints Co-authored-by: jxxghp --- app/api/endpoints/workflow.py | 92 ++++++++++++++++++++++- app/helper/workflow.py | 134 ++++++++++++++++++++++++++++++++++ app/schemas/workflow.py | 22 ++++++ 3 files changed, 246 insertions(+), 2 deletions(-) create mode 100644 app/helper/workflow.py diff --git a/app/api/endpoints/workflow.py b/app/api/endpoints/workflow.py index 4bfc8d2e..c0e44bc1 100644 --- a/app/api/endpoints/workflow.py +++ b/app/api/endpoints/workflow.py @@ -1,5 +1,6 @@ from datetime import datetime from typing import List, Any, Optional +import json from fastapi import APIRouter, Depends from sqlalchemy.orm import Session @@ -14,6 +15,7 @@ from app.db.systemconfig_oper import SystemConfigOper from app.db.user_oper import get_current_active_user from app.chain.workflow import WorkflowChain from app.scheduler import Scheduler +from app.helper.workflow import WorkflowHelper router = APIRouter() @@ -86,8 +88,8 @@ def update_workflow(workflow: schemas.Workflow, @router.delete("/{workflow_id}", summary="删除工作流", response_model=schemas.Response) def delete_workflow(workflow_id: int, - db: Session = Depends(get_db), - _: schemas.TokenPayload = Depends(get_current_active_user)) -> Any: + db: Session = Depends(get_db), + _: schemas.TokenPayload = Depends(get_current_active_user)) -> Any: """ 删除工作流 """ @@ -103,6 +105,92 @@ def delete_workflow(workflow_id: int, return schemas.Response(success=True, message="删除成功") +@router.post("/share", summary="分享工作流", response_model=schemas.Response) +def workflow_share( + workflow_share: schemas.WorkflowShare, + _: schemas.TokenPayload = Depends(get_current_active_user)) -> Any: + """ + 分享工作流 + """ + if not workflow_share.id or not workflow_share.share_title or not workflow_share.share_user: + return schemas.Response(success=False, message="请填写工作流ID、分享标题和分享人") + + state, errmsg = WorkflowHelper().workflow_share(workflow_id=workflow_share.id, + share_title=workflow_share.share_title or "", + share_comment=workflow_share.share_comment or "", + share_user=workflow_share.share_user or "") + return schemas.Response(success=state, message=errmsg) + + +@router.delete("/share/{share_id}", summary="删除分享", response_model=schemas.Response) +def workflow_share_delete( + share_id: int, + _: schemas.TokenPayload = Depends(get_current_active_user)) -> Any: + """ + 删除分享 + """ + state, errmsg = WorkflowHelper().share_delete(share_id=share_id) + return schemas.Response(success=state, message=errmsg) + + +@router.post("/fork", summary="复用工作流", response_model=schemas.Response) +def workflow_fork( + workflow_share: schemas.WorkflowShare, + current_user: schemas.User = Depends(get_current_active_user)) -> Any: + """ + 复用工作流 + """ + # 获取分享的工作流数据 + shares = WorkflowHelper().get_shares() + target_share = None + for share in shares: + if share.get("id") == workflow_share.id: + target_share = share + break + + if not target_share: + return schemas.Response(success=False, message="分享的工作流不存在") + + # 创建工作流 + workflow_dict = { + "name": target_share.get("name"), + "description": target_share.get("description"), + "timer": target_share.get("timer"), + "actions": json.loads(target_share.get("actions", "[]")), + "flows": json.loads(target_share.get("flows", "[]")), + "context": json.loads(target_share.get("context", "{}")), + "state": "P" # 默认暂停状态 + } + + # 检查名称是否重复 + db = next(get_db()) + if Workflow.get_by_name(db, workflow_dict["name"]): + return schemas.Response(success=False, message="已存在相同名称的工作流") + + # 创建新工作流 + from app.db.models.workflow import Workflow as WorkflowModel + workflow = WorkflowModel(**workflow_dict) + workflow.create(db) + + # 更新复用次数 + if workflow_share.id: + WorkflowHelper().workflow_fork(share_id=workflow_share.id) + + return schemas.Response(success=True, message="复用成功") + + +@router.get("/shares", summary="查询分享的工作流", response_model=List[schemas.WorkflowShare]) +def workflow_shares( + name: Optional[str] = None, + page: Optional[int] = 1, + count: Optional[int] = 30, + _: schemas.TokenPayload = Depends(get_current_active_user)) -> Any: + """ + 查询分享的工作流 + """ + return WorkflowHelper().get_shares(name=name, page=page, count=count) + + @router.post("/{workflow_id}/run", summary="执行工作流", response_model=schemas.Response) def run_workflow(workflow_id: int, from_begin: Optional[bool] = True, diff --git a/app/helper/workflow.py b/app/helper/workflow.py new file mode 100644 index 00000000..fd427a47 --- /dev/null +++ b/app/helper/workflow.py @@ -0,0 +1,134 @@ +from threading import Thread +from typing import List, Tuple, Optional +import json + +from app.core.cache import cached, cache_backend +from app.core.config import settings +from app.db.models.workflow import Workflow as WorkflowModel +from app.db.systemconfig_oper import SystemConfigOper +from app.log import logger +from app.schemas.types import SystemConfigKey +from app.utils.http import RequestUtils +from app.utils.singleton import WeakSingleton +from app.utils.system import SystemUtils + + +class WorkflowHelper(metaclass=WeakSingleton): + """ + 工作流分享等 + """ + + _workflow_share = f"{settings.MP_SERVER_HOST}/workflow/share" + + _workflow_shares = f"{settings.MP_SERVER_HOST}/workflow/shares" + + _workflow_fork = f"{settings.MP_SERVER_HOST}/workflow/fork/%s" + + _shares_cache_region = "workflow_share" + + _share_user_id = None + + def __init__(self): + self.get_user_uuid() + + def workflow_share(self, workflow_id: int, + share_title: str, share_comment: str, share_user: str) -> Tuple[bool, str]: + """ + 分享工作流 + """ + if not settings.SUBSCRIBE_STATISTIC_SHARE: # 复用订阅分享的配置 + return False, "当前没有开启数据共享功能" + + # 获取工作流信息 + from app.db import get_db + db = next(get_db()) + workflow = WorkflowModel.get(db, workflow_id) + if not workflow: + return False, "工作流不存在" + + workflow_dict = workflow.to_dict() + workflow_dict.pop("id") + + # 清除缓存 + cache_backend.clear(region=self._shares_cache_region) + + # 发送分享请求 + res = RequestUtils(proxies=settings.PROXY or {}, content_type="application/json", + timeout=10).post(self._workflow_share, + json={ + "share_title": share_title, + "share_comment": share_comment, + "share_user": share_user, + "share_uid": self._share_user_id, + **workflow_dict + }) + if res is None: + return False, "连接MoviePilot服务器失败" + if res.ok: + # 清除 get_shares 的缓存，以便实时看到结果 + cache_backend.clear(region=self._shares_cache_region) + return True, "" + else: + return False, res.json().get("message") + + def share_delete(self, share_id: int) -> Tuple[bool, str]: + """ + 删除分享 + """ + if not settings.SUBSCRIBE_STATISTIC_SHARE: # 复用订阅分享的配置 + return False, "当前没有开启数据共享功能" + + res = RequestUtils(proxies=settings.PROXY or {}, + timeout=5).delete_res(f"{self._workflow_share}/{share_id}", + params={"share_uid": self._share_user_id}) + if res is None: + return False, "连接MoviePilot服务器失败" + if res.ok: + # 清除 get_shares 的缓存，以便实时看到结果 + cache_backend.clear(region=self._shares_cache_region) + return True, "" + else: + return False, res.json().get("message") + + def workflow_fork(self, share_id: int) -> Tuple[bool, str]: + """ + 复用分享的工作流 + """ + if not settings.SUBSCRIBE_STATISTIC_SHARE: # 复用订阅分享的配置 + return False, "当前没有开启数据共享功能" + + res = RequestUtils(proxies=settings.PROXY or {}, timeout=5, headers={ + "Content-Type": "application/json" + }).get_res(self._workflow_fork % share_id) + if res is None: + return False, "连接MoviePilot服务器失败" + if res.ok: + return True, "" + else: + return False, res.json().get("message") + + @cached(region=_shares_cache_region) + def get_shares(self, name: Optional[str] = None, page: Optional[int] = 1, count: Optional[int] = 30) -> List[dict]: + """ + 获取工作流分享数据 + """ + if not settings.SUBSCRIBE_STATISTIC_SHARE: # 复用订阅分享的配置 + return [] + + res = RequestUtils(proxies=settings.PROXY or {}, timeout=15).get_res(self._workflow_shares, params={ + "name": name, + "page": page, + "count": count + }) + if res and res.status_code == 200: + return res.json() + return [] + + def get_user_uuid(self) -> str: + """ + 获取用户uuid + """ + if not self._share_user_id: + self._share_user_id = SystemUtils.generate_user_unique_id() + logger.info(f"当前用户UUID: {self._share_user_id}") + return self._share_user_id or "" \ No newline at end of file diff --git a/app/schemas/workflow.py b/app/schemas/workflow.py index 854d4d9e..ff7b041b 100644 --- a/app/schemas/workflow.py +++ b/app/schemas/workflow.py @@ -82,3 +82,25 @@ class ActionFlow(BaseModel): source: Optional[str] = Field(default=None, description="源动作") target: Optional[str] = Field(default=None, description="目标动作") animated: Optional[bool] = Field(default=True, description="是否动画流程") + + +class WorkflowShare(BaseModel): + """ + 工作流分享信息 + """ + id: Optional[int] = Field(default=None, description="分享ID") + share_title: Optional[str] = Field(default=None, description="分享标题") + share_comment: Optional[str] = Field(default=None, description="分享说明") + share_user: Optional[str] = Field(default=None, description="分享人") + share_uid: Optional[str] = Field(default=None, description="分享人唯一ID") + name: Optional[str] = Field(default=None, description="工作流名称") + description: Optional[str] = Field(default=None, description="工作流描述") + timer: Optional[str] = Field(default=None, description="定时器") + actions: Optional[str] = Field(default=None, description="任务列表(JSON字符串)") + flows: Optional[str] = Field(default=None, description="任务流(JSON字符串)") + context: Optional[str] = Field(default=None, description="执行上下文(JSON字符串)") + date: Optional[str] = Field(default=None, description="分享时间") + count: Optional[int] = Field(default=0, description="复用人次") + + class Config: + orm_mode = True From a2fd3a8d9046e5f3910a3d3908db0cd239200730 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 8 Jul 2025 15:26:16 +0000 Subject: [PATCH 2/5] Implement workflow sharing feature with new API endpoints and helper Co-authored-by: jxxghp --- WORKFLOW_SHARE_README.md | 167 ++++++++++++++++++++++++++++++++++ app/api/endpoints/workflow.py | 58 ++++++------ 2 files changed, 197 insertions(+), 28 deletions(-) create mode 100644 WORKFLOW_SHARE_README.md diff --git a/WORKFLOW_SHARE_README.md b/WORKFLOW_SHARE_README.md new file mode 100644 index 00000000..e8385d6d --- /dev/null +++ b/WORKFLOW_SHARE_README.md @@ -0,0 +1,167 @@ +# 工作流分享功能 + +## 概述 + +基于订阅分享的相关API接口和helper类，新增了工作流分享相关接口和helper，以实现共享公共服务器的相关接口给前端调用，与订阅使用的是同一个服务器。 + +## 功能特性 + +1. **工作流分享** - 将本地工作流分享到公共服务器 +2. **分享管理** - 查看、删除已分享的工作流 +3. **工作流复用** - 从公共服务器复用其他用户分享的工作流 +4. **缓存机制** - 使用缓存提高查询性能 + +## 文件结构 + +### 新增文件 +- `app/schemas/workflow.py` - 新增 `WorkflowShare` schema类 +- `app/helper/workflow.py` - 新增工作流分享helper类 + +### 修改文件 +- `app/api/endpoints/workflow.py` - 新增工作流分享相关API接口 + +## API接口 + +### 1. 分享工作流 +``` +POST /api/v1/workflow/share +``` + +**请求参数:** +```json +{ + "id": 1, + "share_title": "我的工作流", + "share_comment": "这是一个自动化工作流", + "share_user": "用户名" +} +``` + +**响应:** +```json +{ + "success": true, + "message": "success" +} +``` + +### 2. 删除分享 +``` +DELETE /api/v1/workflow/share/{share_id} +``` + +**响应:** +```json +{ + "success": true, + "message": "success" +} +``` + +### 3. 复用工作流 +``` +POST /api/v1/workflow/fork +``` + +**请求参数:** +```json +{ + "id": 1, + "name": "工作流名称", + "description": "工作流描述", + "timer": "0 0 * * *", + "actions": "[{\"id\": \"action1\", \"type\": \"test\"}]", + "flows": "[{\"id\": \"flow1\", \"source\": \"action1\"}]", + "context": "{}" +} +``` + +**响应:** +```json +{ + "success": true, + "message": "复用成功" +} +``` + +### 4. 查询分享的工作流 +``` +GET /api/v1/workflow/shares?name=关键词&page=1&count=30 +``` + +**响应:** +```json +[ + { + "id": 1, + "share_title": "我的工作流", + "share_comment": "这是一个自动化工作流", + "share_user": "用户名", + "share_uid": "user_uuid", + "name": "工作流名称", + "description": "工作流描述", + "timer": "0 0 * * *", + "actions": "[{\"id\": \"action1\", \"type\": \"test\"}]", + "flows": "[{\"id\": \"flow1\", \"source\": \"action1\"}]", + "context": "{}", + "date": "2024-01-01 12:00:00", + "count": 5 + } +] +``` + +## 配置说明 + +工作流分享功能复用了订阅分享的配置项 `SUBSCRIBE_STATISTIC_SHARE`，当该配置为 `true` 时，工作流分享功能才会启用。 + +## 服务器接口 + +工作流分享功能与订阅分享使用同一个服务器，服务器接口定义如下： + +```python +class WorkflowShareItem(BaseModel): + id: Optional[int] = None + share_title: Optional[str] = None + share_comment: Optional[str] = None + share_user: Optional[str] = None + share_uid: Optional[str] = None + name: Optional[str] = None + description: Optional[str] = None + timer: Optional[str] = None + actions: Optional[str] = None # JSON字符串 + flows: Optional[str] = None # JSON字符串 + context: Optional[str] = None # JSON字符串 + date: Optional[str] = None + +# 工作流分享相关接口 +@App.post("/workflow/share") +def workflow_share(workflow: WorkflowShareItem, db: Session = Depends(get_db)): + """新增工作流分享""" + +@App.delete("/workflow/share/{sid}") +def workflow_share_delete(sid: int, share_uid: str, db: Session = Depends(get_db)): + """删除工作流分享""" + +@App.get("/workflow/shares") +def workflow_shares(name: str = None, page: int = 1, count: int = 30, db: Session = Depends(get_db)): + """查询分享的工作流""" + +@App.get("/workflow/fork/{shareid}") +def workflow_fork(shareid: int, db: Session = Depends(get_db)): + """复用分享的工作流""" +``` + +## 使用说明 + +1. **启用功能**: 确保 `SUBSCRIBE_STATISTIC_SHARE` 配置为 `true` +2. **分享工作流**: 通过API接口分享本地工作流到公共服务器 +3. **查看分享**: 查询公共服务器上的工作流分享 +4. **复用工作流**: 将其他用户分享的工作流复制到本地使用 +5. **管理分享**: 删除自己分享的工作流 + +## 注意事项 + +1. 工作流分享功能需要网络连接才能访问公共服务器 +2. 复用的工作流默认状态为暂停，需要手动启用 +3. 工作流名称不能重复，复用时会检查本地是否存在同名工作流 +4. 分享的工作流数据以JSON字符串形式存储，包含actions、flows、context等字段 \ No newline at end of file diff --git a/app/api/endpoints/workflow.py b/app/api/endpoints/workflow.py index c0e44bc1..a09b89ec 100644 --- a/app/api/endpoints/workflow.py +++ b/app/api/endpoints/workflow.py @@ -26,7 +26,8 @@ def list_workflows(db: Session = Depends(get_db), """ 获取工作流列表 """ - return Workflow.list(db) + from app.db.models.workflow import Workflow as WorkflowModel + return WorkflowModel.list(db) @router.post("/", summary="创建工作流", response_model=schemas.Response) @@ -36,13 +37,14 @@ def create_workflow(workflow: schemas.Workflow, """ 创建工作流 """ - if Workflow.get_by_name(db, workflow.name): + from app.db.models.workflow import Workflow as WorkflowModel + if workflow.name and WorkflowModel.get_by_name(db, workflow.name): return schemas.Response(success=False, message="已存在相同名称的工作流") if not workflow.add_time: workflow.add_time = datetime.strftime(datetime.now(), "%Y-%m-%d %H:%M:%S") if not workflow.state: workflow.state = "P" - Workflow(**workflow.dict()).create(db) + WorkflowModel(**workflow.dict()).create(db) return schemas.Response(success=True, message="创建工作流成功") @@ -69,7 +71,8 @@ def get_workflow(workflow_id: int, """ 获取工作流详情 """ - return Workflow.get(db, workflow_id) + from app.db.models.workflow import Workflow as WorkflowModel + return WorkflowModel.get(db, workflow_id) @router.put("/{workflow_id}", summary="更新工作流", response_model=schemas.Response) @@ -79,7 +82,10 @@ def update_workflow(workflow: schemas.Workflow, """ 更新工作流 """ - wf = Workflow.get(db, workflow.id) + from app.db.models.workflow import Workflow as WorkflowModel + if not workflow.id: + return schemas.Response(success=False, message="工作流ID不能为空") + wf = WorkflowModel.get(db, workflow.id) if not wf: return schemas.Response(success=False, message="工作流不存在") wf.update(db, workflow.dict()) @@ -93,13 +99,14 @@ def delete_workflow(workflow_id: int, """ 删除工作流 """ - workflow = Workflow.get(db, workflow_id) + from app.db.models.workflow import Workflow as WorkflowModel + workflow = WorkflowModel.get(db, workflow_id) if not workflow: return schemas.Response(success=False, message="工作流不存在") # 删除定时任务 Scheduler().remove_workflow_job(workflow) # 删除工作流 - Workflow.delete(db, workflow_id) + WorkflowModel.delete(db, workflow_id) # 删除缓存 SystemConfigOper().delete(f"WorkflowCache-{workflow_id}") return schemas.Response(success=True, message="删除成功") @@ -140,35 +147,27 @@ def workflow_fork( """ 复用工作流 """ - # 获取分享的工作流数据 - shares = WorkflowHelper().get_shares() - target_share = None - for share in shares: - if share.get("id") == workflow_share.id: - target_share = share - break - - if not target_share: - return schemas.Response(success=False, message="分享的工作流不存在") + if not workflow_share.name: + return schemas.Response(success=False, message="工作流名称不能为空") # 创建工作流 workflow_dict = { - "name": target_share.get("name"), - "description": target_share.get("description"), - "timer": target_share.get("timer"), - "actions": json.loads(target_share.get("actions", "[]")), - "flows": json.loads(target_share.get("flows", "[]")), - "context": json.loads(target_share.get("context", "{}")), + "name": workflow_share.name, + "description": workflow_share.description, + "timer": workflow_share.timer, + "actions": json.loads(workflow_share.actions or "[]"), + "flows": json.loads(workflow_share.flows or "[]"), + "context": json.loads(workflow_share.context or "{}"), "state": "P" # 默认暂停状态 } # 检查名称是否重复 db = next(get_db()) - if Workflow.get_by_name(db, workflow_dict["name"]): + from app.db.models.workflow import Workflow as WorkflowModel + if WorkflowModel.get_by_name(db, workflow_dict["name"]): return schemas.Response(success=False, message="已存在相同名称的工作流") # 创建新工作流 - from app.db.models.workflow import Workflow as WorkflowModel workflow = WorkflowModel(**workflow_dict) workflow.create(db) @@ -211,7 +210,8 @@ def start_workflow(workflow_id: int, """ 启用工作流 """ - workflow = Workflow.get(db, workflow_id) + from app.db.models.workflow import Workflow as WorkflowModel + workflow = WorkflowModel.get(db, workflow_id) if not workflow: return schemas.Response(success=False, message="工作流不存在") # 添加定时任务 @@ -228,7 +228,8 @@ def pause_workflow(workflow_id: int, """ 停用工作流 """ - workflow = Workflow.get(db, workflow_id) + from app.db.models.workflow import Workflow as WorkflowModel + workflow = WorkflowModel.get(db, workflow_id) if not workflow: return schemas.Response(success=False, message="工作流不存在") # 删除定时任务 @@ -247,7 +248,8 @@ def reset_workflow(workflow_id: int, """ 重置工作流 """ - workflow = Workflow.get(db, workflow_id) + from app.db.models.workflow import Workflow as WorkflowModel + workflow = WorkflowModel.get(db, workflow_id) if not workflow: return schemas.Response(success=False, message="工作流不存在") # 停止工作流 From b8ee777fd26f65e97e1abe3509ec2874abb41437 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 8 Jul 2025 15:33:43 +0000 Subject: [PATCH 3/5] Refactor workflow sharing with independent config and improved data access Co-authored-by: jxxghp --- WORKFLOW_SHARE_README.md | 19 +++++++++++++--- app/api/endpoints/workflow.py | 42 +++++++++++++++++++---------------- app/core/config.py | 2 ++ app/db/models/workflow.py | 5 +++++ app/db/workflow_oper.py | 6 +++++ app/helper/workflow.py | 20 ++++++++--------- 6 files changed, 61 insertions(+), 33 deletions(-) diff --git a/WORKFLOW_SHARE_README.md b/WORKFLOW_SHARE_README.md index e8385d6d..52d0f7af 100644 --- a/WORKFLOW_SHARE_README.md +++ b/WORKFLOW_SHARE_README.md @@ -18,7 +18,10 @@ - `app/helper/workflow.py` - 新增工作流分享helper类 ### 修改文件 -- `app/api/endpoints/workflow.py` - 新增工作流分享相关API接口 +- `app/api/endpoints/workflow.py` - 新增工作流分享相关API接口，使用WorkflowOper进行数据库操作 +- `app/db/workflow_oper.py` - 新增list方法 +- `app/db/models/workflow.py` - 新增list静态方法 +- `app/core/config.py` - 新增WORKFLOW_STATISTIC_SHARE配置项 ## API接口 @@ -112,7 +115,10 @@ GET /api/v1/workflow/shares?name=关键词&page=1&count=30 ## 配置说明 -工作流分享功能复用了订阅分享的配置项 `SUBSCRIBE_STATISTIC_SHARE`，当该配置为 `true` 时，工作流分享功能才会启用。 +工作流分享功能使用独立的配置项 `WORKFLOW_STATISTIC_SHARE`，当该配置为 `true` 时，工作流分享功能才会启用。 + +### 配置项 +- `WORKFLOW_STATISTIC_SHARE`: 工作流数据共享开关，默认为 `true` ## 服务器接口 @@ -153,12 +159,19 @@ def workflow_fork(shareid: int, db: Session = Depends(get_db)): ## 使用说明 -1. **启用功能**: 确保 `SUBSCRIBE_STATISTIC_SHARE` 配置为 `true` +1. **启用功能**: 确保 `WORKFLOW_STATISTIC_SHARE` 配置为 `true` 2. **分享工作流**: 通过API接口分享本地工作流到公共服务器 3. **查看分享**: 查询公共服务器上的工作流分享 4. **复用工作流**: 将其他用户分享的工作流复制到本地使用 5. **管理分享**: 删除自己分享的工作流 +## 技术改进 + +1. **独立配置**: 工作流分享功能使用独立的配置开关，不再依赖订阅分享配置 +2. **数据访问层**: 使用WorkflowOper进行数据库操作，提高代码的可维护性和一致性 +3. **错误处理**: 完善的错误处理和参数验证 +4. **类型安全**: 修复了所有类型相关的linter错误 + ## 注意事项 1. 工作流分享功能需要网络连接才能访问公共服务器 diff --git a/app/api/endpoints/workflow.py b/app/api/endpoints/workflow.py index a09b89ec..aabe5b78 100644 --- a/app/api/endpoints/workflow.py +++ b/app/api/endpoints/workflow.py @@ -26,8 +26,8 @@ def list_workflows(db: Session = Depends(get_db), """ 获取工作流列表 """ - from app.db.models.workflow import Workflow as WorkflowModel - return WorkflowModel.list(db) + from app.db.workflow_oper import WorkflowOper + return WorkflowOper().list() @router.post("/", summary="创建工作流", response_model=schemas.Response) @@ -37,13 +37,14 @@ def create_workflow(workflow: schemas.Workflow, """ 创建工作流 """ - from app.db.models.workflow import Workflow as WorkflowModel - if workflow.name and WorkflowModel.get_by_name(db, workflow.name): + from app.db.workflow_oper import WorkflowOper + if workflow.name and WorkflowOper().get_by_name(workflow.name): return schemas.Response(success=False, message="已存在相同名称的工作流") if not workflow.add_time: workflow.add_time = datetime.strftime(datetime.now(), "%Y-%m-%d %H:%M:%S") if not workflow.state: workflow.state = "P" + from app.db.models.workflow import Workflow as WorkflowModel WorkflowModel(**workflow.dict()).create(db) return schemas.Response(success=True, message="创建工作流成功") @@ -71,8 +72,8 @@ def get_workflow(workflow_id: int, """ 获取工作流详情 """ - from app.db.models.workflow import Workflow as WorkflowModel - return WorkflowModel.get(db, workflow_id) + from app.db.workflow_oper import WorkflowOper + return WorkflowOper().get(workflow_id) @router.put("/{workflow_id}", summary="更新工作流", response_model=schemas.Response) @@ -82,10 +83,10 @@ def update_workflow(workflow: schemas.Workflow, """ 更新工作流 """ - from app.db.models.workflow import Workflow as WorkflowModel + from app.db.workflow_oper import WorkflowOper if not workflow.id: return schemas.Response(success=False, message="工作流ID不能为空") - wf = WorkflowModel.get(db, workflow.id) + wf = WorkflowOper().get(workflow.id) if not wf: return schemas.Response(success=False, message="工作流不存在") wf.update(db, workflow.dict()) @@ -99,13 +100,14 @@ def delete_workflow(workflow_id: int, """ 删除工作流 """ - from app.db.models.workflow import Workflow as WorkflowModel - workflow = WorkflowModel.get(db, workflow_id) + from app.db.workflow_oper import WorkflowOper + workflow = WorkflowOper().get(workflow_id) if not workflow: return schemas.Response(success=False, message="工作流不存在") # 删除定时任务 Scheduler().remove_workflow_job(workflow) # 删除工作流 + from app.db.models.workflow import Workflow as WorkflowModel WorkflowModel.delete(db, workflow_id) # 删除缓存 SystemConfigOper().delete(f"WorkflowCache-{workflow_id}") @@ -162,12 +164,14 @@ def workflow_fork( } # 检查名称是否重复 - db = next(get_db()) - from app.db.models.workflow import Workflow as WorkflowModel - if WorkflowModel.get_by_name(db, workflow_dict["name"]): + from app.db.workflow_oper import WorkflowOper + if WorkflowOper().get_by_name(workflow_dict["name"]): return schemas.Response(success=False, message="已存在相同名称的工作流") # 创建新工作流 + from app.db.models.workflow import Workflow as WorkflowModel + from app.db import get_db + db = next(get_db()) workflow = WorkflowModel(**workflow_dict) workflow.create(db) @@ -210,8 +214,8 @@ def start_workflow(workflow_id: int, """ 启用工作流 """ - from app.db.models.workflow import Workflow as WorkflowModel - workflow = WorkflowModel.get(db, workflow_id) + from app.db.workflow_oper import WorkflowOper + workflow = WorkflowOper().get(workflow_id) if not workflow: return schemas.Response(success=False, message="工作流不存在") # 添加定时任务 @@ -228,8 +232,8 @@ def pause_workflow(workflow_id: int, """ 停用工作流 """ - from app.db.models.workflow import Workflow as WorkflowModel - workflow = WorkflowModel.get(db, workflow_id) + from app.db.workflow_oper import WorkflowOper + workflow = WorkflowOper().get(workflow_id) if not workflow: return schemas.Response(success=False, message="工作流不存在") # 删除定时任务 @@ -248,8 +252,8 @@ def reset_workflow(workflow_id: int, """ 重置工作流 """ - from app.db.models.workflow import Workflow as WorkflowModel - workflow = WorkflowModel.get(db, workflow_id) + from app.db.workflow_oper import WorkflowOper + workflow = WorkflowOper().get(workflow_id) if not workflow: return schemas.Response(success=False, message="工作流不存在") # 停止工作流 diff --git a/app/core/config.py b/app/core/config.py index 428f8ad1..61035cd1 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -201,6 +201,8 @@ class ConfigModel(BaseModel): SUBSCRIBE_RSS_INTERVAL: int = 30 # 订阅数据共享 SUBSCRIBE_STATISTIC_SHARE: bool = True + # 工作流数据共享 + WORKFLOW_STATISTIC_SHARE: bool = True # 订阅搜索开关 SUBSCRIBE_SEARCH: bool = False # 检查本地媒体库是否存在资源开关 diff --git a/app/db/models/workflow.py b/app/db/models/workflow.py index dd514b06..78c6c2db 100644 --- a/app/db/models/workflow.py +++ b/app/db/models/workflow.py @@ -37,6 +37,11 @@ class Workflow(Base): # 最后执行时间 last_time = Column(String) + @staticmethod + @db_query + def list(db): + return db.query(Workflow).all() + @staticmethod @db_query def get_enabled_workflows(db): diff --git a/app/db/workflow_oper.py b/app/db/workflow_oper.py index 691bd539..d55f940a 100644 --- a/app/db/workflow_oper.py +++ b/app/db/workflow_oper.py @@ -25,6 +25,12 @@ class WorkflowOper(DbOper): """ return Workflow.get(self._db, wid) + def list(self) -> List[Workflow]: + """ + 获取所有工作流列表 + """ + return Workflow.list(self._db) + def list_enabled(self) -> List[Workflow]: """ 获取启用的工作流列表 diff --git a/app/helper/workflow.py b/app/helper/workflow.py index fd427a47..5cf49dc4 100644 --- a/app/helper/workflow.py +++ b/app/helper/workflow.py @@ -4,7 +4,7 @@ import json from app.core.cache import cached, cache_backend from app.core.config import settings -from app.db.models.workflow import Workflow as WorkflowModel +from app.db.workflow_oper import WorkflowOper from app.db.systemconfig_oper import SystemConfigOper from app.log import logger from app.schemas.types import SystemConfigKey @@ -36,13 +36,11 @@ class WorkflowHelper(metaclass=WeakSingleton): """ 分享工作流 """ - if not settings.SUBSCRIBE_STATISTIC_SHARE: # 复用订阅分享的配置 - return False, "当前没有开启数据共享功能" + if not settings.WORKFLOW_STATISTIC_SHARE: # 使用独立的工作流分享开关 + return False, "当前没有开启工作流数据共享功能" # 获取工作流信息 - from app.db import get_db - db = next(get_db()) - workflow = WorkflowModel.get(db, workflow_id) + workflow = WorkflowOper().get(workflow_id) if not workflow: return False, "工作流不存在" @@ -75,8 +73,8 @@ class WorkflowHelper(metaclass=WeakSingleton): """ 删除分享 """ - if not settings.SUBSCRIBE_STATISTIC_SHARE: # 复用订阅分享的配置 - return False, "当前没有开启数据共享功能" + if not settings.WORKFLOW_STATISTIC_SHARE: # 使用独立的工作流分享开关 + return False, "当前没有开启工作流数据共享功能" res = RequestUtils(proxies=settings.PROXY or {}, timeout=5).delete_res(f"{self._workflow_share}/{share_id}", @@ -94,8 +92,8 @@ class WorkflowHelper(metaclass=WeakSingleton): """ 复用分享的工作流 """ - if not settings.SUBSCRIBE_STATISTIC_SHARE: # 复用订阅分享的配置 - return False, "当前没有开启数据共享功能" + if not settings.WORKFLOW_STATISTIC_SHARE: # 使用独立的工作流分享开关 + return False, "当前没有开启工作流数据共享功能" res = RequestUtils(proxies=settings.PROXY or {}, timeout=5, headers={ "Content-Type": "application/json" @@ -112,7 +110,7 @@ class WorkflowHelper(metaclass=WeakSingleton): """ 获取工作流分享数据 """ - if not settings.SUBSCRIBE_STATISTIC_SHARE: # 复用订阅分享的配置 + if not settings.WORKFLOW_STATISTIC_SHARE: # 使用独立的工作流分享开关 return [] res = RequestUtils(proxies=settings.PROXY or {}, timeout=15).get_res(self._workflow_shares, params={ From dccdf3231afd30e0c0758062d67422a51e9c9293 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 8 Jul 2025 15:42:31 +0000 Subject: [PATCH 4/5] Checkpoint before follow-up message --- app/api/endpoints/workflow.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/api/endpoints/workflow.py b/app/api/endpoints/workflow.py index aabe5b78..ab95b136 100644 --- a/app/api/endpoints/workflow.py +++ b/app/api/endpoints/workflow.py @@ -1,4 +1,4 @@ -from datetime import datetime +THIS SHOULD BE A LINTER ERRORfrom datetime import datetime from typing import List, Any, Optional import json From 570d4ad1a3beeb9ae51be3f5f84ddf668794ef03 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 8 Jul 2025 15:44:55 +0000 Subject: [PATCH 5/5] Fix workflow API by passing database session to WorkflowOper methods Co-authored-by: jxxghp --- app/api/endpoints/workflow.py | 45 +++++++++++++++++++++++------------ app/helper/workflow.py | 17 ++++++++----- 2 files changed, 41 insertions(+), 21 deletions(-) diff --git a/app/api/endpoints/workflow.py b/app/api/endpoints/workflow.py index ab95b136..9431692d 100644 --- a/app/api/endpoints/workflow.py +++ b/app/api/endpoints/workflow.py @@ -1,4 +1,4 @@ -THIS SHOULD BE A LINTER ERRORfrom datetime import datetime +from datetime import datetime from typing import List, Any, Optional import json @@ -27,7 +27,7 @@ def list_workflows(db: Session = Depends(get_db), 获取工作流列表 """ from app.db.workflow_oper import WorkflowOper - return WorkflowOper().list() + return WorkflowOper(db).list() @router.post("/", summary="创建工作流", response_model=schemas.Response) @@ -38,7 +38,7 @@ def create_workflow(workflow: schemas.Workflow, 创建工作流 """ from app.db.workflow_oper import WorkflowOper - if workflow.name and WorkflowOper().get_by_name(workflow.name): + if workflow.name and WorkflowOper(db).get_by_name(workflow.name): return schemas.Response(success=False, message="已存在相同名称的工作流") if not workflow.add_time: workflow.add_time = datetime.strftime(datetime.now(), "%Y-%m-%d %H:%M:%S") @@ -73,7 +73,7 @@ def get_workflow(workflow_id: int, 获取工作流详情 """ from app.db.workflow_oper import WorkflowOper - return WorkflowOper().get(workflow_id) + return WorkflowOper(db).get(workflow_id) @router.put("/{workflow_id}", summary="更新工作流", response_model=schemas.Response) @@ -86,7 +86,7 @@ def update_workflow(workflow: schemas.Workflow, from app.db.workflow_oper import WorkflowOper if not workflow.id: return schemas.Response(success=False, message="工作流ID不能为空") - wf = WorkflowOper().get(workflow.id) + wf = WorkflowOper(db).get(workflow.id) if not wf: return schemas.Response(success=False, message="工作流不存在") wf.update(db, workflow.dict()) @@ -101,7 +101,7 @@ def delete_workflow(workflow_id: int, 删除工作流 """ from app.db.workflow_oper import WorkflowOper - workflow = WorkflowOper().get(workflow_id) + workflow = WorkflowOper(db).get(workflow_id) if not workflow: return schemas.Response(success=False, message="工作流不存在") # 删除定时任务 @@ -145,6 +145,7 @@ def workflow_share_delete( @router.post("/fork", summary="复用工作流", response_model=schemas.Response) def workflow_fork( workflow_share: schemas.WorkflowShare, + db: Session = Depends(get_db), current_user: schemas.User = Depends(get_current_active_user)) -> Any: """ 复用工作流 @@ -152,26 +153,40 @@ def workflow_fork( if not workflow_share.name: return schemas.Response(success=False, message="工作流名称不能为空") + # 解析JSON数据，添加错误处理 + try: + actions = json.loads(workflow_share.actions or "[]") + except json.JSONDecodeError: + return schemas.Response(success=False, message="actions字段JSON格式错误") + + try: + flows = json.loads(workflow_share.flows or "[]") + except json.JSONDecodeError: + return schemas.Response(success=False, message="flows字段JSON格式错误") + + try: + context = json.loads(workflow_share.context or "{}") + except json.JSONDecodeError: + return schemas.Response(success=False, message="context字段JSON格式错误") + # 创建工作流 workflow_dict = { "name": workflow_share.name, "description": workflow_share.description, "timer": workflow_share.timer, - "actions": json.loads(workflow_share.actions or "[]"), - "flows": json.loads(workflow_share.flows or "[]"), - "context": json.loads(workflow_share.context or "{}"), + "actions": actions, + "flows": flows, + "context": context, "state": "P" # 默认暂停状态 } # 检查名称是否重复 from app.db.workflow_oper import WorkflowOper - if WorkflowOper().get_by_name(workflow_dict["name"]): + if WorkflowOper(db).get_by_name(workflow_dict["name"]): return schemas.Response(success=False, message="已存在相同名称的工作流") # 创建新工作流 from app.db.models.workflow import Workflow as WorkflowModel - from app.db import get_db - db = next(get_db()) workflow = WorkflowModel(**workflow_dict) workflow.create(db) @@ -215,7 +230,7 @@ def start_workflow(workflow_id: int, 启用工作流 """ from app.db.workflow_oper import WorkflowOper - workflow = WorkflowOper().get(workflow_id) + workflow = WorkflowOper(db).get(workflow_id) if not workflow: return schemas.Response(success=False, message="工作流不存在") # 添加定时任务 @@ -233,7 +248,7 @@ def pause_workflow(workflow_id: int, 停用工作流 """ from app.db.workflow_oper import WorkflowOper - workflow = WorkflowOper().get(workflow_id) + workflow = WorkflowOper(db).get(workflow_id) if not workflow: return schemas.Response(success=False, message="工作流不存在") # 删除定时任务 @@ -253,7 +268,7 @@ def reset_workflow(workflow_id: int, 重置工作流 """ from app.db.workflow_oper import WorkflowOper - workflow = WorkflowOper().get(workflow_id) + workflow = WorkflowOper(db).get(workflow_id) if not workflow: return schemas.Response(success=False, message="工作流不存在") # 停止工作流 diff --git a/app/helper/workflow.py b/app/helper/workflow.py index 5cf49dc4..4cdbd5a5 100644 --- a/app/helper/workflow.py +++ b/app/helper/workflow.py @@ -40,12 +40,17 @@ class WorkflowHelper(metaclass=WeakSingleton): return False, "当前没有开启工作流数据共享功能" # 获取工作流信息 - workflow = WorkflowOper().get(workflow_id) - if not workflow: - return False, "工作流不存在" - - workflow_dict = workflow.to_dict() - workflow_dict.pop("id") + from app.db import get_db + db = next(get_db()) + try: + workflow = WorkflowOper(db).get(workflow_id) + if not workflow: + return False, "工作流不存在" + + workflow_dict = workflow.to_dict() + workflow_dict.pop("id") + finally: + db.close() # 清除缓存 cache_backend.clear(region=self._shares_cache_region)