pqai_agent_server/api_server.py

@@ -1,6 +1,7 @@
 #! /usr/bin/env python
 # -*- coding: utf-8 -*-
 # vim:fenc=utf-8
+import json
 import time
 import logging
 import werkzeug.exceptions
@@ -11,11 +12,12 @@ from pqai_agent import configs
 
 from pqai_agent import logging_service, chat_service, prompt_templates
 from pqai_agent.agents.message_reply_agent import MessageReplyAgent
+from pqai_agent.configs import apollo_config
 from pqai_agent.history_dialogue_service import HistoryDialogueService
 from pqai_agent.user_manager import MySQLUserManager, MySQLUserRelationManager
 from pqai_agent.utils.prompt_utils import format_agent_profile, format_user_profile
 from pqai_agent_server.const import AgentApiConst
-from pqai_agent_server.models import MySQLSessionManager
+from pqai_agent_server.models import MySQLSessionManager, MySQLStaffManager
 from pqai_agent_server.utils import wrap_response, quit_human_intervention_status
 from pqai_agent_server.utils import (
     run_extractor_prompt,
@@ -36,11 +38,55 @@ def list_staffs():
 @app.route('/api/getStaffProfile', methods=['GET'])
 def get_staff_profile():
     staff_id = request.args['staff_id']
-    profile = app.user_manager.get_staff_profile(staff_id)
-    if not profile:
+    if not staff_id:
+        return wrap_response(400, msg='staff_id is required')
+    agent_profile = app.staff_manager.get_staff_profile(staff_id)
+    if not agent_profile:
         return wrap_response(404, msg='staff not found')
     else:
-        return wrap_response(200, data=profile)
+        field_map_list = apollo_config.get_json_value("field_map_list", [])
+        field_map = {
+            item["field_name"]: item["display_name"] for item in field_map_list
+        }
+        profile_info = [
+            {
+                "field_name": key,
+                "display_name": field_map[key],
+                "field_value": value,
+            }
+            for key, value in agent_profile.items()
+            if agent_profile.get(key)
+        ]
+        return wrap_response(200, data=profile_info)
+
+
+@app.route('/api/saveStaffProfile', methods=['POST'])
+def save_staff_profile():
+    staff_id = request.json.get('staff_id')
+    staff_profile = request.json.get('staff_profile')
+    if not staff_id:
+        return wrap_response(400, msg='staff id is required')
+
+    if not staff_profile:
+        return wrap_response(400, msg='profile is required')
+    else:
+        try:
+            profile_info_list = json.loads(staff_profile)
+            profile_dict = {item['field_name']: item['field_value'] for item in profile_info_list}
+            affected_rows = app.staff_manager.save_staff_profile(staff_id, profile_dict)
+
+            if not affected_rows:
+                return wrap_response(500, msg='save staff profile failed')
+            else:
+                return wrap_response(200, msg='save staff profile success')
+
+        except json.decoder.JSONDecodeError:
+            return wrap_response(400, msg='profile is not a valid json')
+
+
+@app.route('/api/getProfileFields', methods=['GET'])
+def get_profile_fields():
+    return wrap_response(200, data=apollo_config.get_json_value("field_map_list", []))
 
 
 @app.route('/api/getUserProfile', methods=['GET'])
@@ -282,7 +328,7 @@ def get_conversation_list():
     if not staff_id or not user_id:
         return wrap_response(404, msg="staff_id and user_id are required")
 
-    page = request.args.get("page")
+    page = request.args.get("page_id")
     response = app.session_manager.get_conversation_list(staff_id, user_id, page, const.DEFAULT_CONVERSATION_SIZE)
     return wrap_response(200, data=response)
 
@@ -346,6 +392,15 @@ if __name__ == '__main__':
     )
     app.session_manager = session_manager
 
+    # init staff manager
+    staff_manager = MySQLStaffManager(
+        db_config=user_db_config['mysql'],
+        staff_table=staff_db_config['table'],
+        user_table=user_db_config['table']
+    )
+    app.staff_manager = staff_manager
+
+    # init wecom manager
     wecom_db_config = config['storage']['user_relation']
     user_relation_manager = MySQLUserRelationManager(
         user_db_config['mysql'], wecom_db_config['mysql'],

pqai_agent_server/models/__init__.py

@@ -1 +1,2 @@
-from .mysql_session_manager import MySQLSessionManager
+from .mysql_session_manager import MySQLSessionManager
+from .mysql_staff_manager import MySQLStaffManager

pqai_agent_server/models/mysql_staff_manager.py

@@ -0,0 +1,85 @@
+import abc
+import json
+import pymysql.cursors
+
+from typing import Dict, List
+from pqai_agent.database import MySQLManager
+
+
+class StaffManager(abc.ABC):
+
+    @abc.abstractmethod
+    def list_all_staffs(self, page_id: int, page_size: int) -> List[Dict]:
+        pass
+
+    @abc.abstractmethod
+    def get_staff_profile(self, staff_id) -> Dict:
+        pass
+
+    @abc.abstractmethod
+    def save_staff_profile(self, staff_id: str, staff_profile: Dict):
+        pass
+
+
+class MySQLStaffManager(StaffManager):
+
+    def __init__(self, db_config, staff_table, user_table):
+        self.db = MySQLManager(db_config)
+        self.staff_table = staff_table
+        self.user_table = user_table
+
+    def save_staff_profile(self, staff_id: str, staff_profile: Dict):
+        update_query = f"""
+            update {self.staff_table} set agent_profile = %s where third_party_user_id = %s;
+        """
+        affected_rows = self.db.execute(
+            update_query, (json.dumps(staff_profile), staff_id)
+        )
+        return affected_rows
+
+    def get_staff_profile(self, staff_id) -> Dict:
+        empty_profile = {}
+        sql = f"""select agent_profile from {self.staff_table} where third_party_user_id = %s;"""
+        response = self.db.select(
+            sql=sql,
+            cursor_type=pymysql.cursors.DictCursor,
+            args=(staff_id,),
+        )
+        if not response:
+            return empty_profile
+
+        agent_profile = response[0]["agent_profile"]
+        profile = json.loads(agent_profile)
+        if not profile:
+            return empty_profile
+        return profile
+
+    def list_all_staffs(self, page_id: int, page_size: int) -> Dict:
+        """
+        :param page_size:
+        :param page_id:
+        :return:
+        """
+        sql = f"""
+            select t1.third_party_user_id as staff_id, t1.name as staff_name, t2.iconurl as avatar
+            from {self.staff_table} t1 left join {self.user_table} t2
+            on t1.third_party_user_id = t2.third_party_user_id
+            limit %s offset %s;
+        """
+        staff_list = self.db.select(
+            sql=sql,
+            cursor_type=pymysql.cursors.DictCursor,
+            args=(page_size + 1, page_size * (page_id - 1)),
+        )
+        if len(staff_list) > page_size:
+            has_next_page = True
+            next_page_id = page_id + 1
+            staff_list = staff_list[:page_size]
+        else:
+            has_next_page = False
+            next_page_id = None
+        return {
+            "has_next_page": has_next_page,
+            "next_page": next_page_id,
+            "data": staff_list,
+        }