mirror of
https://github.com/xinnan-tech/xiaozhi-esp32-server.git
synced 2026-07-29 04:13:56 +08:00
重新划分目录 (#214)
* 🦄 refactor(web): 修改zhikongtaiweb到web * 🦄 refactor: 重写前端 路由守护尚未写完 * 🦄 refactor: 标准化路由 * update:前端重写，保留后端 * update:添加前端代码 * update:pip转成poetry启动 * update:增加mem0ai包依赖 * update:文档增加mem0ai的描述 * feat: play online mp3 (#181) Co-authored-by: 欣南科技 <huangrongzhuang@xin-nan.com> * 修改前端代码 * update:调整项目目录 * update:优化 * update:配置文件去除8002端口 * update:增加开发说明 * update:更新开发协议 --------- Co-authored-by: kalicyh <34980061+kaliCYH@users.noreply.github.com> Co-authored-by: hrz <1710360675@qq.com> Co-authored-by: freshlife001 <talent@mises.site> Co-authored-by: CGD <3030332422@qq.com>
This commit is contained in:
欣南科技
co-authored by
kalicyh
hrz
freshlife001
CGD
kalicyh
hrz
freshlife001
CGD
parent
8b151d07c2
commit
0e43748fdc
@@ -0,0 +1,29 @@
|
||||
import os
|
||||
import sys
|
||||
from loguru import logger
|
||||
from config.settings import load_config
|
||||
|
||||
def setup_logging():
|
||||
"""从配置文件中读取日志配置，并设置日志输出格式和级别"""
|
||||
config = load_config()
|
||||
log_config = config["log"]
|
||||
log_format = log_config.get("log_format", "<green>{time:YY-MM-DD HH:mm:ss}</green>[<light-blue>{extra[tag]}</light-blue>] - <level>{level}</level> - <light-green>{message}</light-green>")
|
||||
log_format_simple = log_config.get("log_format_file", "{time:YYYY-MM-DD HH:mm:ss} - {name} - {level} - {extra[tag]} - {message}")
|
||||
log_level = log_config.get("log_level", "INFO")
|
||||
log_dir = log_config.get("log_dir", "tmp")
|
||||
log_file = log_config.get("log_file", "server.log")
|
||||
data_dir = log_config.get("data_dir", "data")
|
||||
|
||||
os.makedirs(log_dir, exist_ok=True)
|
||||
os.makedirs(data_dir, exist_ok=True)
|
||||
|
||||
# 配置日志输出
|
||||
logger.remove()
|
||||
|
||||
# 输出到控制台
|
||||
logger.add(sys.stdout, format=log_format, level=log_level)
|
||||
|
||||
# 输出到文件
|
||||
logger.add(os.path.join(log_dir, log_file), format=log_format_simple, level=log_level)
|
||||
|
||||
return logger
|
||||
@@ -0,0 +1,241 @@
|
||||
import os
|
||||
import time
|
||||
import yaml
|
||||
from config.logger import setup_logging
|
||||
from typing import Dict, Any, Optional
|
||||
from copy import deepcopy
|
||||
from core.utils.util import get_project_dir
|
||||
from core.utils import llm, tts
|
||||
from core.utils.lock_manager import FileLockManager
|
||||
|
||||
TAG = __name__
|
||||
|
||||
class PrivateConfig:
|
||||
def __init__(self, device_id: str, default_config: Dict[str, Any], auth_code_gen=None):
|
||||
self.device_id = device_id
|
||||
self.default_config = default_config
|
||||
self.config_path = get_project_dir() + 'data/.private_config.yaml'
|
||||
self.logger = setup_logging()
|
||||
self.private_config = {}
|
||||
self.auth_code_gen = auth_code_gen
|
||||
self.lock_manager = FileLockManager()
|
||||
|
||||
async def load_or_create(self):
|
||||
try:
|
||||
await self.lock_manager.acquire_lock(self.config_path)
|
||||
try:
|
||||
if os.path.exists(self.config_path):
|
||||
with open(self.config_path, 'r', encoding='utf-8') as f:
|
||||
all_configs = yaml.safe_load(f) or {}
|
||||
else:
|
||||
all_configs = {}
|
||||
|
||||
if self.device_id not in all_configs:
|
||||
# Get selected module names
|
||||
selected_modules = self.default_config['selected_module']
|
||||
selected_tts = selected_modules['TTS']
|
||||
selected_llm = selected_modules['LLM']
|
||||
selected_asr = selected_modules['ASR']
|
||||
selected_vad = selected_modules['VAD']
|
||||
|
||||
# 生成认证码
|
||||
auth_code = None
|
||||
if self.auth_code_gen:
|
||||
auth_code = self.auth_code_gen.generate_code()
|
||||
|
||||
# Initialize device config with only necessary configurations
|
||||
device_config = {
|
||||
'selected_module': deepcopy(selected_modules),
|
||||
'prompt': self.default_config['prompt'],
|
||||
'LLM': {
|
||||
selected_llm: deepcopy(self.default_config['LLM'][selected_llm])
|
||||
},
|
||||
'TTS': {
|
||||
selected_tts: deepcopy(self.default_config['TTS'][selected_tts])
|
||||
},
|
||||
'ASR': {
|
||||
selected_asr: deepcopy(self.default_config['ASR'][selected_asr])
|
||||
},
|
||||
'VAD': {
|
||||
selected_vad: deepcopy(self.default_config['VAD'][selected_vad])
|
||||
},
|
||||
'auth_code': auth_code # 添加认证码字段
|
||||
}
|
||||
|
||||
all_configs[self.device_id] = device_config
|
||||
|
||||
# Save updated configs
|
||||
with open(self.config_path, 'w', encoding='utf-8') as f:
|
||||
yaml.dump(all_configs, f, allow_unicode=True)
|
||||
|
||||
self.private_config = all_configs[self.device_id]
|
||||
|
||||
finally:
|
||||
self.lock_manager.release_lock(self.config_path)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.bind(tag=TAG).error(f"Error handling private config: {e}")
|
||||
self.private_config = {}
|
||||
|
||||
async def update_config(self, selected_modules: Dict[str, str], prompt: str, nickname: str) -> bool:
|
||||
"""更新设备配置
|
||||
Args:
|
||||
selected_modules: 选择的模块配置，格式如 {'LLM': 'AliLLM', 'TTS': 'EdgeTTS',...}
|
||||
prompt: 提示词配置
|
||||
Returns:
|
||||
bool: 更新是否成功
|
||||
"""
|
||||
try:
|
||||
await self.lock_manager.acquire_lock(self.config_path)
|
||||
try:
|
||||
# Read main config to get full module configurations
|
||||
main_config = self.default_config
|
||||
|
||||
# Create new device config
|
||||
device_config = {
|
||||
'selected_module': selected_modules,
|
||||
'prompt': prompt,
|
||||
'nickname': nickname,
|
||||
}
|
||||
if self.private_config.get('last_chat_time'):
|
||||
device_config['last_chat_time'] = self.private_config['last_chat_time']
|
||||
if self.private_config.get('owner'):
|
||||
device_config['owner'] = self.private_config['owner']
|
||||
|
||||
# Copy full module configurations from main config
|
||||
for module_type, selected_name in selected_modules.items():
|
||||
if selected_name and selected_name in main_config.get(module_type, {}):
|
||||
device_config[module_type] = {
|
||||
selected_name: main_config[module_type][selected_name]
|
||||
}
|
||||
|
||||
# Read all configs
|
||||
if os.path.exists(self.config_path):
|
||||
with open(self.config_path, 'r', encoding='utf-8') as f:
|
||||
all_configs = yaml.safe_load(f) or {}
|
||||
else:
|
||||
all_configs = {}
|
||||
|
||||
# Update device config
|
||||
all_configs[self.device_id] = device_config
|
||||
self.private_config = device_config
|
||||
|
||||
# Save back to file
|
||||
with open(self.config_path, 'w', encoding='utf-8') as f:
|
||||
yaml.dump(all_configs, f, allow_unicode=True)
|
||||
|
||||
return True
|
||||
finally:
|
||||
self.lock_manager.release_lock(self.config_path)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.bind(tag=TAG).error(f"Error updating config: {e}")
|
||||
return False
|
||||
|
||||
async def delete_config(self) -> bool:
|
||||
"""删除设备配置
|
||||
Returns:
|
||||
bool: 删除是否成功
|
||||
"""
|
||||
try:
|
||||
await self.lock_manager.acquire_lock(self.config_path)
|
||||
try:
|
||||
# 读取所有配置
|
||||
if os.path.exists(self.config_path):
|
||||
with open(self.config_path, 'r', encoding='utf-8') as f:
|
||||
all_configs = yaml.safe_load(f) or {}
|
||||
else:
|
||||
return False
|
||||
|
||||
# 删除设备配置
|
||||
if self.device_id in all_configs:
|
||||
del all_configs[self.device_id]
|
||||
|
||||
# 保存更新后的配置
|
||||
with open(self.config_path, 'w', encoding='utf-8') as f:
|
||||
yaml.dump(all_configs, f, allow_unicode=True)
|
||||
|
||||
self.private_config = {}
|
||||
return True
|
||||
|
||||
return False
|
||||
finally:
|
||||
self.lock_manager.release_lock(self.config_path)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.bind(tag=TAG).error(f"Error deleting config: {e}")
|
||||
return False
|
||||
|
||||
def create_private_instances(self):
|
||||
# 判断存在私有配置，并且self.device_id在私有配置中
|
||||
if not self.private_config:
|
||||
self.logger.bind(tag=TAG).error("Private config not found for device_id: {}", self.device_id)
|
||||
return None, None
|
||||
|
||||
"""创建私有处理模块实例"""
|
||||
config = self.private_config
|
||||
selected_modules = config['selected_module']
|
||||
return (
|
||||
llm.create_instance(
|
||||
selected_modules["LLM"]
|
||||
if not 'type' in config["LLM"][selected_modules["LLM"]]
|
||||
else
|
||||
config["LLM"][selected_modules["LLM"]]['type'],
|
||||
config["LLM"][selected_modules["LLM"]],
|
||||
),
|
||||
tts.create_instance(
|
||||
selected_modules["TTS"]
|
||||
if not 'type' in config["TTS"][selected_modules["TTS"]]
|
||||
else
|
||||
config["TTS"][selected_modules["TTS"]]["type"],
|
||||
config["TTS"][selected_modules["TTS"]],
|
||||
self.default_config.get("delete_audio", True) # Using default_config for global settings
|
||||
)
|
||||
)
|
||||
|
||||
async def update_last_chat_time(self, timestamp=None):
|
||||
"""更新设备最近一次的聊天时间
|
||||
Args:
|
||||
timestamp: 指定的时间戳,不传则使用当前时间
|
||||
"""
|
||||
if not self.private_config:
|
||||
self.logger.bind(tag=TAG).error("Private config not found")
|
||||
return False
|
||||
|
||||
try:
|
||||
await self.lock_manager.acquire_lock(self.config_path)
|
||||
try:
|
||||
if timestamp is None:
|
||||
timestamp = int(time.time())
|
||||
|
||||
self.private_config['last_chat_time'] = timestamp
|
||||
|
||||
# 读取所有配置
|
||||
with open(self.config_path, 'r', encoding='utf-8') as f:
|
||||
all_configs = yaml.safe_load(f) or {}
|
||||
|
||||
# 更新当前设备配置
|
||||
all_configs[self.device_id] = self.private_config
|
||||
|
||||
# 保存回文件
|
||||
with open(self.config_path, 'w', encoding='utf-8') as f:
|
||||
yaml.dump(all_configs, f, allow_unicode=True)
|
||||
|
||||
return True
|
||||
finally:
|
||||
self.lock_manager.release_lock(self.config_path)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.bind(tag=TAG).error(f"Error updating last chat time: {e}")
|
||||
return False
|
||||
|
||||
def get_auth_code(self) -> str:
|
||||
"""获取设备的认证码
|
||||
Returns:
|
||||
str: 认证码，如果没有返回空字符串
|
||||
"""
|
||||
return self.private_config.get('auth_code', '')
|
||||
|
||||
def get_owner(self) -> Optional[str]:
|
||||
"""获取设备当前所有者"""
|
||||
return self.private_config.get('owner')
|
||||
@@ -0,0 +1,84 @@
|
||||
import os
|
||||
import argparse
|
||||
from ruamel.yaml import YAML
|
||||
from collections.abc import Mapping
|
||||
from core.utils.util import read_config, get_project_dir
|
||||
|
||||
default_config_file = "config.yaml"
|
||||
|
||||
|
||||
def get_config_file():
|
||||
global default_config_file
|
||||
# 判断是否存在私有的配置文件
|
||||
config_file = default_config_file
|
||||
if os.path.exists(get_project_dir() + "data/." + default_config_file):
|
||||
config_file = "data/." + default_config_file
|
||||
return config_file
|
||||
|
||||
|
||||
def load_config():
|
||||
"""加载配置文件"""
|
||||
parser = argparse.ArgumentParser(description="Server configuration")
|
||||
config_file = get_config_file()
|
||||
parser.add_argument("--config_path", type=str, default=config_file)
|
||||
args = parser.parse_args()
|
||||
return read_config(args.config_path)
|
||||
|
||||
|
||||
def update_config(config):
|
||||
yaml = YAML()
|
||||
yaml.preserve_quotes = True
|
||||
"""将配置保存到YAML文件"""
|
||||
with open(get_config_file(), 'w') as f:
|
||||
yaml.dump(config, f)
|
||||
|
||||
|
||||
def find_missing_keys(new_config, old_config, parent_key=''):
|
||||
"""
|
||||
递归查找缺失的配置项
|
||||
返回格式：[缺失配置路径]
|
||||
"""
|
||||
missing_keys = []
|
||||
|
||||
if not isinstance(new_config, Mapping):
|
||||
return missing_keys
|
||||
|
||||
for key, value in new_config.items():
|
||||
# 构建当前配置路径
|
||||
full_path = f"{parent_key}.{key}" if parent_key else key
|
||||
|
||||
# 检查键是否存在
|
||||
if key not in old_config:
|
||||
missing_keys.append(full_path)
|
||||
continue
|
||||
|
||||
# 递归检查嵌套字典
|
||||
if isinstance(value, Mapping):
|
||||
sub_missing = find_missing_keys(
|
||||
value,
|
||||
old_config[key],
|
||||
parent_key=full_path
|
||||
)
|
||||
missing_keys.extend(sub_missing)
|
||||
|
||||
return missing_keys
|
||||
|
||||
|
||||
def check_config_file():
|
||||
old_config_file = get_config_file()
|
||||
global default_config_file
|
||||
if not old_config_file.startswith('data'):
|
||||
return
|
||||
old_config = read_config(get_project_dir() + old_config_file)
|
||||
new_config = read_config(get_project_dir() + default_config_file)
|
||||
# 查找缺失的配置项
|
||||
missing_keys = find_missing_keys(new_config, old_config)
|
||||
|
||||
if missing_keys:
|
||||
error_msg = "您的配置文件太旧了，缺少了：\n"
|
||||
error_msg += "\n".join(f"- {key}" for key in missing_keys)
|
||||
error_msg += "\n建议您：\n"
|
||||
error_msg += "1、备份data/.config.yaml文件\n"
|
||||
error_msg += "2、将根目录的config.yaml文件复制到data下，重命名为.config.yaml\n"
|
||||
error_msg += "3、将密钥逐个复制到新的配置文件中\n"
|
||||
raise ValueError(error_msg)
|
||||
Reference in New Issue
Block a user