AgentCoreService/user_manager.py

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

from typing import Dict, Optional, Tuple, Any
import json
import time
import os
import abc

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

class LocalUserManager(UserManager):
    def get_user_profile(self, user_id) -> Dict:
        """加载用户个人资料，如不存在则创建默认资料。主要用于本地调试"""
        try:
            with open(f"user_profiles/{user_id}.json", "r", encoding="utf-8") as f:
                return json.load(f)
        except FileNotFoundError:
            # 创建默认用户资料
            default_profile = {
                "name": "",
                "nickname": "",
                "preferred_nickname": "",
                "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": []
            }
            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):
        json_files = []
        for root, dirs, files in os.walk('user_profiles/'):
            for file in files:
                if file.endswith('.json'):
                    json_files.append(os.path.splitext(file)[0])
        return json_files