AgentCoreService/pqai_agent/user_manager.py

#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vim:fenc=utf-8

from pqai_agent.logging_service import logger
from typing import Dict, Optional, List
import json
import time
import os
import abc

import pymysql.cursors

from pqai_agent import configs
from pqai_agent.database import MySQLManager


class UserManager(abc.ABC):
    @abc.abstractmethod
    def get_user_profile(self, user_id) -> Dict:
        pass

    @abc.abstractmethod
    def save_user_profile(self, user_id, profile: Dict) -> None:
        pass

    @abc.abstractmethod
    def list_all_users(self):
        pass

    @abc.abstractmethod
    def get_staff_profile(self, staff_id) -> Dict:
        #FIXME(zhoutian): 重新设计用户和员工数据管理模型
        pass

    @staticmethod
    def get_default_profile(**kwargs) -> Dict:
        default_profile = {
            "name": "",
            "nickname": "",
            "avatar": "",
            "preferred_nickname": "",
            "gender": "未知",
            "age": 0,
            "region": '',
            "interests": [],
            "family_members": {},
            "health_conditions": [],
            "medications": [],
            "reminder_preferences": {
                "medication": True,
                "health": True,
                "weather": True,
                "news": False
            },
            "interaction_style": "standard",  # standard, verbose, concise
            "interaction_frequency": "medium",  # low, medium, high
            "last_topics": [],
            "created_at": int(time.time() * 1000),
            "human_intervention_history": []
        }
        for key, value in kwargs.items():
            if key in default_profile:
                default_profile[key] = value
        return default_profile

    def list_users(self, **kwargs) -> List[Dict]:
        pass

class UserRelationManager(abc.ABC):
    @abc.abstractmethod
    def list_staffs(self):
        pass

    @abc.abstractmethod
    def list_users(self, staff_id: str, page: int = 1, page_size: int = 100):
        pass

    @abc.abstractmethod
    def list_staff_users(self, staff_id: str = None, tag_id: int = None) -> List[Dict]:
        pass

    @abc.abstractmethod
    def get_user_tags(self, user_id: str) -> List[str]:
        pass

    @abc.abstractmethod
    def stop_user_daily_push(self, user_id: str) -> bool:
        pass

class LocalUserManager(UserManager):
    def get_user_profile(self, user_id) -> Dict:
        """加载用户个人资料，如不存在则创建默认资料。主要用于本地调试"""
        default_profile = self.get_default_profile()
        try:
            with open(f"user_profiles/{user_id}.json", "r", encoding="utf-8") as f:
                profile = json.load(f)
            entry_added = False
            for key, value in default_profile.items():
                if key not in profile:
                    logger.debug(f"user[{user_id}] add profile key[{key}] value[{value}]")
                    profile[key] = value
                    entry_added = True
            if entry_added:
                self.save_user_profile(user_id, profile)
            return profile
        except FileNotFoundError:
            # 创建默认用户资料
            self.save_user_profile(user_id, default_profile)
            return default_profile

    def save_user_profile(self, user_id, profile: Dict) -> None:
        if not user_id:
            raise Exception("Invalid user_id: {}".format(user_id))
        with open(f"user_profiles/{user_id}.json", "w", encoding="utf-8") as f:
            json.dump(profile, f, ensure_ascii=False, indent=2)

    def list_all_users(self):
        user_ids = []
        for root, dirs, files in os.walk('../user_profiles/'):
            for file in files:
                if file.endswith('.json'):
                    user_ids.append(os.path.splitext(file)[0])
        return user_ids

    def get_staff_profile(self, staff_id) -> Dict:
        try:
            with open(f"user_profiles/{staff_id}.json", "r", encoding="utf-8") as f:
                profile = json.load(f)
            entry_added = False
            if entry_added:
                self.save_user_profile(staff_id, profile)
            return profile
        except Exception as e:
            logger.error("staff profile not found: {}".format(e))
            return {}

    def list_users(self, **kwargs) -> List[Dict]:
        pass

class MySQLUserManager(UserManager):
    PROFILE_EXCLUDE_ITEMS = ['avatar', ]

    def __init__(self, db_config, table_name, staff_table):
        self.db = MySQLManager(db_config)
        self.table_name = table_name
        self.staff_table = staff_table

    def get_user_profile(self, user_id) -> Dict:
        sql = f"SELECT name, wxid, profile_data_v1, gender, iconurl as avatar" \
              f" FROM {self.table_name} WHERE third_party_user_id = {user_id}"
        data = self.db.select(sql, pymysql.cursors.DictCursor)
        if not data:
            logger.error(f"user[{user_id}] not found")
            return {}
        data = data[0]
        gender_map = {0: '未知', 1: '男', 2: '女', None: '未知'}
        gender = gender_map[data['gender']]
        default_profile = self.get_default_profile(nickname=data['name'], gender=gender, avatar=data['avatar'])
        if not data['profile_data_v1']:
            logger.warning(f"user[{user_id}] profile not found, create a default one")
            self.save_user_profile(user_id, default_profile)
            return default_profile
        else:
            profile = json.loads(data['profile_data_v1'])
            # 资料条目有增加时，需合并更新
            entry_added = False
            for key, value in default_profile.items():
                if key not in profile:
                    # logger.debug(f"user[{user_id}] add profile key[{key}] value[{value}]")
                    profile[key] = value
                    entry_added = True
            if entry_added:
                self.save_user_profile(user_id, profile)
            return profile

    def save_user_profile(self, user_id, profile: Dict) -> None:
        if not user_id:
            raise Exception("Invalid user_id: {}".format(user_id))
        if configs.get().get('debug_flags', {}).get('disable_database_write', False):
            return
        profile = profile.copy()
        for name in self.PROFILE_EXCLUDE_ITEMS:
            profile.pop(name, None)
        sql = f"UPDATE {self.table_name} SET profile_data_v1 = %s WHERE third_party_user_id = {user_id}"
        self.db.execute(sql, (json.dumps(profile),))

    def list_all_users(self):
        sql = f"SELECT third_party_user_id FROM {self.table_name}"
        data = self.db.select(sql, pymysql.cursors.DictCursor)
        return [user['third_party_user_id'] for user in data]

    def get_staff_profile(self, staff_id) -> Dict:
        if not self.staff_table:
            raise Exception("staff_table is not set")
        return self.get_staff_profile_v3(staff_id)

    def get_staff_profile_v1(self, staff_id) -> Dict:
        sql = f"SELECT agent_name, agent_gender, agent_age, agent_region, agent_profile " \
              f"FROM {self.staff_table} WHERE third_party_user_id = '{staff_id}'"
        data = self.db.select(sql, pymysql.cursors.DictCursor)
        if not data:
            logger.error(f"staff[{staff_id}] not found")
            return {}
        profile = data[0]
        # 转换性别格式
        gender_map = {0: '未知', 1: '男', 2: '女', None: '未知'}
        profile['agent_gender'] = gender_map[profile['agent_gender']]
        return profile

    def get_staff_profile_v2(self, staff_id) -> Dict:
        sql = f"SELECT agent_name as name, agent_gender as gender, agent_age as age, agent_region as region, agent_profile " \
              f"FROM {self.staff_table} WHERE third_party_user_id = '{staff_id}'"
        data = self.db.select(sql, pymysql.cursors.DictCursor)
        if not data:
            logger.error(f"staff[{staff_id}] not found")
            return {}
        profile = data[0]
        # 转换性别格式
        gender_map = {0: '未知', 1: '男', 2: '女', None: '未知'}
        profile['gender'] = gender_map[profile['gender']]

        # 合并JSON字段（新版本）数据
        if profile['agent_profile']:
            detail_profile = json.loads(profile['agent_profile'])
            profile.update(detail_profile)

        # 去除原始字段
        profile.pop('agent_profile', None)
        return profile

    def get_staff_profile_v3(self, staff_id) -> Dict:
        sql = f"SELECT agent_profile " \
              f"FROM {self.staff_table} WHERE third_party_user_id = '{staff_id}'"
        data = self.db.select(sql)
        if not data:
            logger.error(f"staff[{staff_id}] not found")
            return {}
        profile_str = data[0][0]
        if not profile_str:
            return {}
        profile = json.loads(profile_str)
        return profile

    def save_staff_profile(self, staff_id: str, profile: Dict):
        # 正常情况下不应该有此操作
        if not self.staff_table:
            raise Exception("staff_table is not set")
        if not staff_id:
            raise Exception("Invalid staff_id: {}".format(staff_id))
        sql = f"UPDATE {self.staff_table} SET agent_profile = %s WHERE third_party_user_id = '{staff_id}'"
        self.db.execute(sql, (json.dumps(profile),))

    def list_users(self, **kwargs) -> List[Dict]:
        user_union_id = kwargs.get('user_union_id', None)
        user_name = kwargs.get('user_name', None)
        if not user_union_id and not user_name:
            raise Exception("user_union_id or user_name is required")
        sql = f"SELECT third_party_user_id, wxid, name, iconurl, gender FROM {self.table_name} WHERE 1=1 "
        if user_name:
            sql += f"AND name = '{user_name}' COLLATE utf8mb4_bin "
        if user_union_id:
            sql += f"AND wxid = '{user_union_id}' "
        data = self.db.select(sql, pymysql.cursors.DictCursor)
        return data

    def get_staff_sessions(self, staff_id, page_id: int = 1, page_size: int = 10, session_type: str = 'default') -> List[Dict]:
        """
        :param page_size:
        :param page_id:
        :param session_type:
        :param staff_id:
        :return:
        """
        match session_type:
            case 'active':
                sql = f"""
                    select staff_id, current_state, user_id
                    from agent_state
                    where staff_id = %s and update_timestamp >= DATE_SUB(NOW(), INTERVAL 2 HOUR)
                    order by update_timestamp desc;
                """
            case 'human_intervention':
                sql = f"""
                    select staff_id, current_state, user_id
                    from agent_state
                    where staff_id = %s and current_state = 5 order by update_timestamp desc;
                """
            case _:
                sql = f"""
                    select t1.staff_id, t1.current_state, t1.user_id, t2.name, t2.iconurl
                    from agent_state t1 join third_party_user t2 
                        on t1.user_id = t2.third_party_user_id
                    where t1.staff_id = %s
                    order by 
                        IF(t1.current_state = 5, 0, 1),
                        t1.update_timestamp desc
                    limit {page_size + 1} offset {page_size * (page_id - 1)};
                """

        staff_sessions = self.db.select(sql, cursor_type=pymysql.cursors.DictCursor, args=(staff_id, ))
        return staff_sessions

    def get_staff_sessions_summary_v1(self, staff_id, page_id: int, page_size: int, status: int) -> Dict:
        """
        :param status: staff status(0: unemployed, 1: employed)
        :param staff_id: staff
        :param page_id: page id
        :param page_size: page size
        :return:
        :todo: 未使用 Mysql 连接池，每次查询均需要与 MySQL 建立连接，性能较低，需要优化
        """
        if not staff_id:
            get_staff_query = f"""
                select third_party_user_id, name from {self.staff_table} where status = %s
                limit %s offset %s;
            """
            staff_id_list = self.db.select(
                sql=get_staff_query,
                cursor_type=pymysql.cursors.DictCursor,
                args=(status, page_size + 1, (page_id - 1) * page_size)
            )
            if not staff_id_list:
                return {}

            if len(staff_id_list) > page_size:
                has_next_page = True
                next_page_id = page_id + 1
                staff_id_list = staff_id_list[:page_size]
            else:
                has_next_page = False
                next_page_id = None
        else:
            get_staff_query = f"""
                select third_party_user_id, name from {self.staff_table} 
                where status = %s and third_party_user_id = %s;
            """
            staff_id_list = self.db.select(
                sql=get_staff_query,
                cursor_type=pymysql.cursors.DictCursor,
                args=(status, staff_id)
            )
            if not staff_id_list:
                return {}
            has_next_page = False
            next_page_id = None
        response_data = [
            {
                'staff_id': staff['third_party_user_id'],
                'staff_name': staff['name'],
                'active_sessions': len(self.get_staff_sessions(staff['third_party_user_id'], session_type='active')),
                'human_intervention_sessions': len(self.get_staff_sessions(staff['third_party_user_id'], session_type='human_intervention'))
            }
            for staff in staff_id_list
        ]
        return {
            'has_next_page': has_next_page,
            'next_page_id': next_page_id,
            'data': response_data
        }

    def get_staff_session_list_v1(self, staff_id, page_id: int, page_size: int) -> Dict:
        """
        :param page_size:
        :param page_id:
        :param staff_id:
        :return:
        """
        session_list = self.get_staff_sessions(staff_id, page_id, page_size)
        if len(session_list) > page_size:
            has_next_page = True
            next_page_id = page_id + 1
            session_list = session_list[:page_size]
        else:
            has_next_page = False
            next_page_id = None
        response_data = []
        for session in session_list:
            temp_obj = {}
            user_id = session['user_id']
            room_id = ':'.join(['private', staff_id, user_id])
            select_query = f"""select content, max(sendtime) as max_timestamp from qywx_chat_history where roomid = %s;"""
            last_message = self.db.select(
                sql=select_query,
                cursor_type=pymysql.cursors.DictCursor,
                args=(room_id,)
            )
            if not last_message:
                temp_obj['message'] = ''
                temp_obj['timestamp'] = 0
            else:
                temp_obj['message'] = last_message[0]['content']
                temp_obj['timestamp'] = last_message[0]['max_timestamp']
            temp_obj['customer_id'] = user_id
            temp_obj['customer_name'] = session['name']
            temp_obj['avatar'] = session['iconurl']
            response_data.append(temp_obj)
        return {
            "staff_id": staff_id,
            "has_next_page": has_next_page,
            "next_page_id": next_page_id,
            "data": response_data
        }

    def get_staff_list(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 qywx_employee t1 left join third_party_user 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
        }

    def get_conversation_list_v1(self, staff_id: str, customer_id: str, page: Optional[int]):
        """
        :param staff_id:
        :param customer_id:
        :param page: timestamp
        :return:
        """
        room_id = ':'.join(['private', staff_id, customer_id])
        page_size = 20
        if not page:
            fetch_query = f"""
                select t1.sender, t2.name, t1.sendtime, t1.content, t2.iconurl
                from qywx_chat_history t1
                join third_party_user t2 on t1.sender = t2.third_party_user_id
                where roomid = %s
                order by sendtime desc
                limit %s;
            """
            messages = self.db.select(
                sql=fetch_query,
                cursor_type=pymysql.cursors.DictCursor,
                args=(room_id, page_size + 1)
            )
        else:
            fetch_query = f"""
                select t1.sender, t2.name, t1.sendtime, t1.content, t2.iconurl
                from qywx_chat_history t1
                join third_party_user t2 on t1.sender = t2.third_party_user_id
                where t1.roomid = %s and t1.sendtime <= %s
                order by sendtime desc
                limit %s;
            """
            messages = self.db.select(
                sql=fetch_query,
                cursor_type=pymysql.cursors.DictCursor,
                args=(room_id, page, page_size + 1)
            )
        if messages:
            if len(messages) > page_size:
                has_next_page = True
                next_page = messages[-1]['sendtime']
            else:
                has_next_page = False
                next_page = None
            response_data = [
                {
                    "sender_id": message['sender'],
                    "sender_name": message['name'],
                    "avatar": message['iconurl'],
                    "content": message['content'],
                    "timestamp": message['sendtime'],
                    "role": "customer" if message['sender'] == customer_id else "staff"
                }
                for message in messages
            ]
            return {
                "staff_id": staff_id,
                "customer_id": customer_id,
                "has_next_page": has_next_page,
                "next_page": next_page,
                "data": response_data
            }
        else:
            has_next_page = False
            next_page = None
            return {
                "staff_id": staff_id,
                "customer_id": customer_id,
                "has_next_page": has_next_page,
                "next_page": next_page,
                "data": []
            }


class LocalUserRelationManager(UserRelationManager):
    def __init__(self):
        pass

    def list_staffs(self):
        return [
            {"third_party_user_id": '1688855931724582', "name": "", "wxid": "ShengHuoLeQu", "agent_name": "小芳"}
        ]

    def list_users(self, staff_id: str, page: int = 1, page_size: int = 100):
        return []

    def list_staff_users(self, staff_id: str = None, tag_id: int = None):
        user_ids = ['7881299453089278', '7881299453132630', '7881299454186909', '7881299455103430', '7881299455173476',
                    '7881299456216398', '7881299457990953', '7881299461167644', '7881299463002136', '7881299464081604',
                    '7881299465121735', '7881299465998082', '7881299466221881', '7881299467152300', '7881299470051791',
                    '7881299470112816', '7881299471149567', '7881299471168030', '7881299471277650', '7881299473321703']
        user_ids = user_ids[:5]
        return [
            {"staff_id": "1688855931724582", "user_id": "7881299670930896"},
            *[{"staff_id": "1688855931724582", "user_id": user_id} for user_id in user_ids]
        ]

    def get_user_tags(self, user_id: str):
        return []

    def stop_user_daily_push(self, user_id: str) -> bool:
        return True

class MySQLUserRelationManager(UserRelationManager):
    def __init__(self, agent_db_config, wecom_db_config,
                 agent_staff_table, agent_user_table,
                 staff_table, relation_table, user_table):
        # FIXME(zhoutian): 因为现在数据库表不统一，需要从两个库读取
        self.agent_db = MySQLManager(agent_db_config)
        self.wecom_db = MySQLManager(wecom_db_config)
        self.agent_staff_table = agent_staff_table
        self.staff_table = staff_table
        self.relation_table = relation_table
        self.agent_user_table = agent_user_table
        self.user_table = user_table

    def list_staffs(self):
        sql = f"SELECT third_party_user_id, name, wxid, agent_name FROM {self.agent_staff_table} WHERE status = 1"
        data = self.agent_db.select(sql, pymysql.cursors.DictCursor)
        return data

    def list_users(self, staff_id: str, page: int = 1, page_size: int = 100):
        return []

    def list_staff_users(self, staff_id: str = None, tag_id: int = None):
        sql = f"SELECT third_party_user_id, wxid FROM {self.agent_staff_table} WHERE status = 1"
        if staff_id:
            sql += f" AND third_party_user_id = '{staff_id}'"
        agent_staff_data = self.agent_db.select(sql, pymysql.cursors.DictCursor)
        if not agent_staff_data:
            return []
        ret = []
        for agent_staff in agent_staff_data:
            wxid = agent_staff['wxid']
            sql = f"SELECT id FROM {self.staff_table} WHERE carrier_id = '{wxid}'"
            staff_data = self.wecom_db.select(sql, pymysql.cursors.DictCursor)
            if not staff_data:
                logger.error(f"staff[{wxid}] not found in wecom database")
                continue
            staff_id = staff_data[0]['id']
            sql = f"SELECT user_id FROM {self.relation_table} WHERE staff_id = '{staff_id}' AND is_delete = 0"
            user_data = self.wecom_db.select(sql, pymysql.cursors.DictCursor)
            if not user_data:
                logger.warning(f"staff[{wxid}] has no user")
                continue
            user_ids = tuple(user['user_id'] for user in user_data)
            sql = f"SELECT union_id FROM {self.user_table} WHERE id IN {str(user_ids)} AND union_id is not null"
            if tag_id:
                sql += f" AND id in (SELECT distinct user_id FROM we_com_user_with_tag WHERE tag_id = {tag_id} and is_delete = 0)"
            user_data = self.wecom_db.select(sql, pymysql.cursors.DictCursor)
            if not user_data:
                logger.warning(f"staff[{wxid}] users not found in wecom database")
                continue
            user_union_ids = tuple(user['union_id'] for user in user_data)
            batch_size = 500
            n_batches = (len(user_union_ids) + batch_size - 1) // batch_size
            agent_user_data = []
            for i in range(n_batches):
                idx_begin = i * batch_size
                idx_end = min((i + 1) * batch_size, len(user_union_ids))
                batch_union_ids = user_union_ids[idx_begin:idx_end]
                sql = f"SELECT third_party_user_id, wxid FROM {self.agent_user_table} WHERE wxid IN {str(batch_union_ids)}"
                batch_agent_user_data = self.agent_db.select(sql, pymysql.cursors.DictCursor)
                if len(agent_user_data) != len(batch_union_ids):
                    # logger.debug(f"staff[{wxid}] some users not found in agent database")
                    pass
                agent_user_data.extend(batch_agent_user_data)
            staff_user_pairs = [
                {
                    'staff_id': agent_staff['third_party_user_id'],
                    'user_id': agent_user['third_party_user_id']
                }
                for agent_user in agent_user_data
            ]
            ret.extend(staff_user_pairs)
        return ret

    def get_user_union_id(self, user_id: str) -> Optional[str]:
        sql = f"SELECT wxid FROM {self.agent_user_table} WHERE third_party_user_id = '{user_id}' AND wxid is not null"
        user_data = self.agent_db.select(sql, pymysql.cursors.DictCursor)
        if not user_data:
            logger.error(f"user[{user_id}] has no union id")
            return None
        union_id = user_data[0]['wxid']
        return union_id

    def get_user_tags(self, user_id: str) -> List[str]:
        union_id = self.get_user_union_id(user_id)
        if not union_id:
            return []
        sql = f"""
            select b.tag_id, c.`tag_name`  from `we_com_user` as a
              join `we_com_user_with_tag` as b
              join `we_com_tag` as c
              on a.`id` = b.`user_id`
              and b.`tag_id` = c.id
              where a.union_id = '{union_id}' """
        tag_data = self.wecom_db.select(sql, pymysql.cursors.DictCursor)
        tag_names = [tag['tag_name'] for tag in tag_data]
        return tag_names

    def stop_user_daily_push(self, user_id: str) -> bool:
        try:
            union_id = self.get_user_union_id(user_id)
            if not union_id:
                return False
            sql = f"UPDATE {self.user_table} SET group_msg_disabled = 1 WHERE union_id = %s"
            rows = self.wecom_db.execute(sql, (union_id, ))
            if rows > 0:
                return True
            else:
                return False
        except Exception as e:
            logger.error(f"stop_user_daily_push failed: {e}")
            return False


if __name__ == '__main__':
    config = configs.get()
    user_db_config = config['storage']['user']
    staff_db_config = config['storage']['staff']
    user_manager = MySQLUserManager(user_db_config['mysql'], user_db_config['table'], staff_db_config['table'])
    user_profile = user_manager.get_user_profile('7881301263964433')
    print(user_profile)

    wecom_db_config = config['storage']['user_relation']
    user_relation_manager = MySQLUserRelationManager(
        user_db_config['mysql'], wecom_db_config['mysql'],
        config['storage']['staff']['table'],
        user_db_config['table'],
        wecom_db_config['table']['staff'],
        wecom_db_config['table']['relation'],
        wecom_db_config['table']['user']
    )
    # all_staff_users = user_relation_manager.list_staff_users()
    user_tags = user_relation_manager.get_user_tags('7881302078008656')
    print(user_tags)