机器人,知识库标签权限修改

api/services/dataset_service.py

@@ -834,16 +834,16 @@ class DatasetService:
         # 添加创建人部门过滤
         if creator_dept:
             from sqlalchemy.orm import aliased
-            CreatorAccount = aliased(Account)
-            union_query = union_query.join(CreatorAccount, Dataset.created_by == CreatorAccount.id)
-            union_query = union_query.filter(CreatorAccount.dept_id == literal(str(creator_dept)))
+            creator_account = aliased(Account)
+            union_query = union_query.join(creator_account, Dataset.created_by == creator_account.id)
+            union_query = union_query.filter(creator_account.dept_id == literal(str(creator_dept)))
 
         # 添加创建人过滤
         if creator:
             from sqlalchemy.orm import aliased
-            CreatorAccount = aliased(Account)
-            union_query = union_query.join(CreatorAccount, Dataset.created_by == CreatorAccount.id)
-            union_query = union_query.filter(CreatorAccount.name.ilike(f"%{creator}%"))
+            creator_account = aliased(Account)
+            union_query = union_query.join(creator_account, Dataset.created_by == creator_account.id)
+            union_query = union_query.filter(creator_account.name.ilike(f"%{creator}%"))
 
         # 其它过滤
         if search:

api/services/tag_service.py

@@ -1,14 +1,17 @@
 import uuid
 from typing import Optional
+import logging
 
 from flask_login import current_user  # type: ignore
 from sqlalchemy import func
-from werkzeug.exceptions import NotFound
+from werkzeug.exceptions import NotFound, Unauthorized
 
 from extensions.ext_database import db
 from models.dataset import Dataset
-from models.model import App, Tag, TagBinding
+from models.model import App, Tag, TagBinding, AppPermissionAll
 from services.errors.tag import TagNameDuplicateError
+from services.errors.account import NoPermissionError
+from models.account import TenantAccountRole
 
 
 class TagService:
@@ -145,6 +148,8 @@ class TagService:
 
     @staticmethod
     def save_tag_binding(args):
+        # 1.智能体设置可见授权的编辑权限一致，2.知识库的标签都能设置--修改为随设置权限一致
+        TagService.check_target_edit_auth(args["type"], args["target_id"])
         # check if target exists
         TagService.check_target_exists(args["type"], args["target_id"])
         # save tag binding
@@ -167,6 +172,7 @@ class TagService:
 
     @staticmethod
     def delete_tag_binding(args):
+        TagService.check_target_edit_auth(args["type"], args["target_id"])
         # check if target exists
         TagService.check_target_exists(args["type"], args["target_id"])
         # delete tag binding
@@ -199,3 +205,29 @@ class TagService:
                 raise NotFound("App not found")
         else:
             raise NotFound("Invalid binding type")
+
+    @staticmethod
+    def check_target_edit_auth(type: str, target_id: str):
+        if type in {"knowledge", "knowledge_category"}:
+            dataset = (
+                db.session.query(Dataset)
+                .filter(Dataset.id == target_id)
+                .first()
+            )
+            if (
+                    current_user.current_role not in [TenantAccountRole.ADMIN, TenantAccountRole.OWNER]
+                    and dataset.created_by != current_user.id
+            ):
+                raise NoPermissionError("You do not have permission to operate this dataset.")
+        elif type == "app":
+            app = (
+                db.session.query(AppPermissionAll)
+                .filter(AppPermissionAll.has_read_permission == True,
+                        AppPermissionAll.account_id == current_user.id,
+                        AppPermissionAll.app_id == target_id)
+                .first()
+            )
+            if not app:
+                raise NoPermissionError("You do not have permission to operate this app.")
+        else:
+            raise NotFound("Invalid binding type")