Hass/xiaozhi-esp32-server
1
0
Fork 0
mirror of https://github.com/xinnan-tech/xiaozhi-esp32-server.git synced 2026-07-25 08:33:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into tts-response

Browse Source 
# Conflicts:
#	main/manager-web/.env.development
This commit is contained in:
hrz
2025-04-09 21:46:59 +08:00
parent 4359f1d5b3 55be189772
commit d7aac5e10f
129 changed files with 8298 additions and 2083 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/core/utils/util.py
+67 -38
View File
@@ -10,7 +10,10 @@ import requests
def get_project_dir():
"""获取项目根目录"""
return os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + '/'
return (
os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+ "/"
)
def get_local_ip():
@@ -24,6 +27,7 @@ def get_local_ip():
except Exception as e:
return "127.0.0.1"
def is_private_ip(ip_addr):
"""
Check if an IP address is a private IP address (compatible with IPv4 and IPv6).
@@ -33,46 +37,48 @@ def is_private_ip(ip_addr):
"""
try:
# Validate IPv4 or IPv6 address format
if not re.match(r"^(\d{1,3}\.){3}\d{1,3}$|^([0-9a-fA-F]{1,4}:){7}[0-9a-fA-F]{1,4}$", ip_addr):
if not re.match(
r"^(\d{1,3}\.){3}\d{1,3}$|^([0-9a-fA-F]{1,4}:){7}[0-9a-fA-F]{1,4}$", ip_addr
):
return False # Invalid IP address format
# IPv4 private address ranges
if '.' in ip_addr: # IPv4 address
ip_parts = list(map(int, ip_addr.split('.')))
if "." in ip_addr: # IPv4 address
ip_parts = list(map(int, ip_addr.split(".")))
if ip_parts[0] == 10:
return True # 10.0.0.0/8 range
elif ip_parts[0] == 172 and 16 <= ip_parts[1] <= 31:
return True # 172.16.0.0/12 range
elif ip_parts[0] == 192 and ip_parts[1] == 168:
return True # 192.168.0.0/16 range
elif ip_addr == '127.0.0.1':
elif ip_addr == "127.0.0.1":
return True # Loopback address
elif ip_parts[0] == 169 and ip_parts[1] == 254:
return True # Link-local address 169.254.0.0/16
return True # Link-local address 169.254.0.0/16
else:
return False # Not a private IPv4 address
else: # IPv6 address
ip_addr = ip_addr.lower()
if ip_addr.startswith('fc00:') or ip_addr.startswith('fd00:'):
if ip_addr.startswith("fc00:") or ip_addr.startswith("fd00:"):
return True # Unique Local Addresses (FC00::/7)
elif ip_addr == '::1':
elif ip_addr == "::1":
return True # Loopback address
elif ip_addr.startswith('fe80:'):
return True # Link-local unicast addresses (FE80::/10)
elif ip_addr.startswith("fe80:"):
return True # Link-local unicast addresses (FE80::/10)
else:
return False # Not a private IPv6 address
except (ValueError, IndexError):
return False # IP address format error or insufficient segments
def get_ip_info(ip_addr):
try:
url = "https://whois.pconline.com.cn/ipJson.jsp?json=true"
if is_private_ip(ip_addr):
ip_addr = ""
url = f"https://whois.pconline.com.cn/ipJson.jsp?json=true&ip={ip_addr}"
resp = requests.get(url).json()
ip_info = {
"city": resp.get("city")
}
ip_info = {"city": resp.get("city")}
return ip_info
except Exception as e:
logging.error(f"Error getting client ip info: {e}")
@@ -87,7 +93,7 @@ def read_config(config_path):
def write_json_file(file_path, data):
"""将数据写入 JSON 文件"""
with open(file_path, 'w', encoding='utf-8') as file:
with open(file_path, "w", encoding="utf-8") as file:
json.dump(data, file, ensure_ascii=False, indent=4)
@@ -95,21 +101,28 @@ def is_punctuation_or_emoji(char):
"""检查字符是否为空格、指定标点或表情符号"""
# 定义需要去除的中英文标点（包括全角/半角）
punctuation_set = {
'，', ',', # 中文逗号 + 英文逗号
'。', '.', # 中文句号 + 英文句号
'！', '!', # 中文感叹号 + 英文感叹号
'-', '－', # 英文连字符 + 中文全角横线
'、' # 中文顿号
"，",
",", # 中文逗号 + 英文逗号
"。",
".", # 中文句号 + 英文句号
"！",
"!", # 中文感叹号 + 英文感叹号
"-",
"－", # 英文连字符 + 中文全角横线
"、", # 中文顿号
}
if char.isspace() or char in punctuation_set:
return True
# 检查表情符号（保留原有逻辑）
code_point = ord(char)
emoji_ranges = [
(0x1F600, 0x1F64F), (0x1F300, 0x1F5FF),
(0x1F680, 0x1F6FF), (0x1F900, 0x1F9FF),
(0x1FA70, 0x1FAFF), (0x2600, 0x26FF),
(0x2700, 0x27BF)
(0x1F600, 0x1F64F),
(0x1F300, 0x1F5FF),
(0x1F680, 0x1F6FF),
(0x1F900, 0x1F9FF),
(0x1FA70, 0x1FAFF),
(0x2600, 0x26FF),
(0x2700, 0x27BF),
]
return any(start <= code_point <= end for start, end in emoji_ranges)
@@ -125,27 +138,42 @@ def get_string_no_punctuation_or_emoji(s):
end = len(chars) - 1
while end >= start and is_punctuation_or_emoji(chars[end]):
end -= 1
return ''.join(chars[start:end + 1])
return "".join(chars[start : end + 1])
def remove_punctuation_and_length(text):
# 全角符号和半角符号的Unicode范围
full_width_punctuations = '！＂＃＄％＆＇（）＊＋，－。／：；＜＝＞？＠［＼］＾＿｀｛｜｝～'
full_width_punctuations = (
"！＂＃＄％＆＇（）＊＋，－。／：；＜＝＞？＠［＼］＾＿｀｛｜｝～"
)
half_width_punctuations = r'!"#$%&\'()*+,-./:;<=>?@[\]^_`{|}~'
space = ' ' # 半角空格
full_width_space = '　' # 全角空格
space = " " # 半角空格
full_width_space = "　" # 全角空格
# 去除全角和半角符号以及空格
result = ''.join([char for char in text if
char not in full_width_punctuations and char not in half_width_punctuations and char not in space and char not in full_width_space])
result = "".join(
[
char
for char in text
if char not in full_width_punctuations
and char not in half_width_punctuations
and char not in space
and char not in full_width_space
]
)
if result == "Yeah":
return 0, ""
return len(result), result
def check_model_key(modelType, modelKey):
if "你" in modelKey:
logging.error("你还没配置" + modelType + "的密钥，请在配置文件中配置密钥，否则无法正常工作")
logging.error(
"你还没配置"
+ modelType
+ "的密钥，请在配置文件中配置密钥，否则无法正常工作"
)
return False
return True
@@ -155,15 +183,15 @@ def check_ffmpeg_installed():
try:
# 执行ffmpeg -version命令，并捕获输出
result = subprocess.run(
['ffmpeg', '-version'],
["ffmpeg", "-version"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=True # 如果返回码非零则抛出异常
check=True, # 如果返回码非零则抛出异常
)
# 检查输出中是否包含版本信息（可选）
output = result.stdout + result.stderr
if 'ffmpeg version' in output.lower():
if "ffmpeg version" in output.lower():
ffmpeg_installed = True
return False
except (subprocess.CalledProcessError, FileNotFoundError):
@@ -175,11 +203,12 @@ def check_ffmpeg_installed():
error_msg += "1、按照项目的安装文档，正确进入conda环境\n"
error_msg += "2、查阅安装文档，如何在conda环境中安装ffmpeg\n"
raise ValueError(error_msg)
def extract_json_from_string(input_string):
"""提取字符串中的 JSON 部分"""
pattern = r'(\{.*\})'
pattern = r"(\{.*\})"
match = re.search(pattern, input_string)
if match:
return match.group(1) # 返回提取的 JSON 字符串
return None
return None
main/xiaozhi-server/core/utils/vad.py
+1 -1
View File
@@ -31,7 +31,7 @@ class SileroVAD(VAD):
def is_vad(self, conn, opus_packet):
try:
pcm_frame = self.decoder.decode(opus_packet, 960)
conn.client_audio_buffer += pcm_frame # 将新数据加入缓冲区
conn.client_audio_buffer.extend(pcm_frame) # 将新数据加入缓冲区
# 处理缓冲区中的完整帧（每次处理512采样点）
client_have_voice = False