完善权限系统

bbit_ai/app/db/postgres/aiprofile.py

@@ -0,0 +1,92 @@
+from typing import List, Dict
+
+from config.pgDb import pg_pool
+
+
+def get_ai_personality(ai_id: str):
+    with pg_pool.getConn() as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                "SELECT ai_personality FROM ai_chat_profiles WHERE id = %s", (ai_id,)
+            )
+            row = cur.fetchone()
+            if row:
+                return row[0]
+            else:
+                return "你是一个乐于助人的AI助手，请保持中文简洁回答用户。"
+
+
+def get_description(ai_id: str):
+    with pg_pool.getConn() as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                "SELECT description FROM ai_chat_profiles WHERE id = %s", (ai_id,)
+            )
+            row = cur.fetchone()
+            if row:
+                return row[0]
+            else:
+                return "你是一个乐于助人的AI助手，请保持中文简洁回答用户。"
+
+
+def get_ai_available_kn_bases(ai_id: str) -> List[str]:
+    with pg_pool.getConn() as conn:
+        result = conn.execute(
+            "SELECT available_kn_bases FROM ai_chat_profiles WHERE id = %s", (ai_id,)
+        )
+        return result.fetchone()[0]
+
+
+def get_all_ai_bot(user_id: str, module: str) -> List[Dict]:
+    with pg_pool.getConn() as conn:
+        with conn.cursor() as cur:
+            # 查询用户角色
+            cur.execute("SELECT roles FROM users WHERE id = %s", (user_id,))
+            role_row = cur.fetchone()
+            if not role_row:
+                return []  # 用户不存在
+            user_roles = role_row[0]
+
+            # 查询 AI 角色 JSON 字段包含用户角色
+
+            cur.execute(
+                """
+                SELECT id, title, description, welcome_words, ai_personality, available_report_tables, available_kn_bases
+                FROM ai_chat_profiles
+                WHERE available_module = %s
+                AND is_active = TRUE
+                AND available_roles::jsonb ?| %s
+                """,
+                (module, user_roles),
+            )
+            rows = cur.fetchall()
+            result = []
+            for row in rows:
+                # row 索引对应 SELECT 字段顺序
+                (
+                    id_,
+                    title,
+                    description,
+                    welcome_words,
+                    ai_personality,
+                    available_report_tables,
+                    available_kn_bases,
+                ) = row
+
+                # 解析 JSON
+                roles_json = ai_personality if ai_personality else {}
+                result.append(
+                    {
+                        "id": id_,
+                        "title": title,
+                        "description": description,
+                        "welcome_words": welcome_words,
+                        "name": roles_json.get("名字", ""),
+                        "role": roles_json.get("性格", ""),
+                        "service": roles_json.get("业务", ""),
+                        "available_report_tables": available_report_tables,
+                        "available_kn_bases": available_kn_bases,
+                    }
+                )
+
+            return result