link之家
链接快照平台
- 输入网页链接，自动生成快照
- 标签化管理网页链接
相关文章推荐
|
痴情的松鼠 · 【大川视界】关于选派研究生参加2025年暑期 ...· 6 月前 · |
|
|
眼睛小的木耳 · 立法意见征集_河北省司法厅· 6 月前 · |
|
|
乐观的签字笔 · 全部成交？洛克菲勒藏品拍卖创22项世界纪录-新华网· 6 月前 · |
|
|
重感情的地瓜 · AxureRP9运行启动失败？！An ...· 1 年前 · |
|
|
火爆的茴香 · 孩子爱看漫画书算不算爱看书？对于孩子阅读，妈 ...· 2 年前 · |
link之家
›
vscode 里 Import “numpy“ count not be resolved_import numpy could not be resolved
|
英勇无比的大脸猫
1 周前 |
我们分析一下这个问题，这里的问题。问题的翻译是：导入"numpy"不能被解决。
这可能有几个问题，1：vscode的python插件没有安装，2: vscode的python的解析器没有设置好。
按照这个思路，去解决问题吧，
1，看看python插件有没有安装好，如下图，如果没有python，请安装python的插件。
装好后，再次试试，看能不能运行。
如果不行的话，再设置一下python的解释器。
如图，点击vscode左下角，会在顶端弹出可选择的解析器，选择你需要的解释器即可。
也可以用快捷键，出现如下图，选择python解析器。
Ctrl + shift + P
然后就会出现可用的解释器选择，选择你需要的解释器，即可
再试着运行一下。
更新 2022-05-14
如果按上述方法还是会报这个提示的，请看下面
今天学cython，自己编绎了一个函数名称是fibo_cy()，解释器选择了，而且还能正常功能，不报错。但还是出现提示的下划线与提示。
Import "fibo_cy" could not be resolved
下面是大体的结果，运行没有问题。（这里没有显示报错了，是我已经做了配置了）
因为这个报的fibo_cy()是编译的，而且pylance的分析也不会分析当前的编译输出的目录下。所以我们要做的事情，应该是要找到pylance
步骤如下：
setting -> 输入 python analysis extra paths -> 添加当前的编译输出的目录即可（其它目录，你也可以添加）
也可以在setting.json修改
本文主要介绍了import numpy出现ImportError: DLL load failed: 找不到指定的模块的解决方案，希望能对使用numpy的同学们有所帮助。
1. 问题描述
2. 解决方案
已安装numpy 1.14.2，又conda 命令安装numpy，自动检测最新版本安装了numpy1.13.1运行import numpy as np 报错，如下：---------------------------------------------------------------------------
ImportError ...
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[logging.StreamHandler()]
logger = logging.getLogger("PDFExtractor")
logger.setLevel(logging.INFO)
class ConstructionDrawingExtractor:
def __init__(self):
# 初始化OCR引擎 - 简化兼容性处理
logger.info("正在初始化OCR引擎...")
# 使用最简参数初始化OCR
self.ocr = PaddleOCR(lang='ch', use_gpu=False, show_log=False)
logger.info("OCR初始化成功")
except Exception as e:
logger.error(f"OCR初始化失败: {str(e)}，尝试无参数初始化")
self.ocr = PaddleOCR()
logger.info("使用无参数初始化OCR成功")
# 增强提取规则
self.rules = {
'drawing_title': r'(图纸名称|图名)[：:\s]*(.+)',
'drawing_number': r'(图纸编号|图号)[：:\s]*([A-Za-z0-9-]+)',
'reinforcement': [
r'(HPB235|HRB335|HRB400|一级钢|二级钢|三级钢)[\s\S]*?直径\s*(\d+\.?\d*)[^\d]*?数量\s*(\d+)',
r'Ø(\d+)[^\d]*?@(\d+)mm[^\d]*?(\d+)根',
r'(钢筋|筋)[\s\S]*?直径\s*(\d+\.?\d*)[^\d]*?数量\s*(\d+)',
r'(\w+)\s*(钢筋|筋)\s*直径\s*(\d+\.?\d*)\s*数量\s*(\d+)'
'beam': r'框架梁\s*(\d+)[×x](\d+)\s*(\d+)根\s*位置[：:\s]*(轴[A-Z])-?(轴[A-Z])?',
'axis': r'轴线尺寸[：:\s]*(\d+\.?\d*)\s*米[×x](\d+\.?\d*)\s*米',
'column': r'柱尺寸[：:\s]*(\d+\.?\d*)\s*米[×x](\d+\.?\d*)\s*米',
'shear_wall': r'剪力墙[厚度]*[：:\s]*(\d+\.?\d*)\s*毫米',
'concrete': r'混凝土总量[：:\s]*(\d+\.?\d*)\s*立方米'
# 钢筋品种映射表
self.steel_grade_map = {
'HPB235': '一级钢',
'HRB335': '二级钢',
'HRB400': '三级钢',
'一级钢': '一级钢',
'二级钢': '二级钢',
'三级钢': '三级钢'
def resolve_file_path(self, file_path: str) -> str:
"""智能解析文件路径，解决相对路径问题"""
# 如果是绝对路径且存在
if os.path.isabs(file_path) and os.path.exists(file_path):
return file_path
# 尝试在当前工作目录查找
current_dir = os.getcwd()
current_path = os.path.join(current_dir, file_path)
if os.path.exists(current_path):
return current_path
# 尝试在脚本目录查找
script_dir = os.path.dirname(os.path.abspath(__file__))
script_path = os.path.join(script_dir, file_path)
if os.path.exists(script_path):
return script_path
# 尝试在项目根目录查找
project_root = os.path.abspath(os.path.join(script_dir, ".."))
project_path = os.path.join(project_root, file_path)
if os.path.exists(project_path):
return project_path
# 尝试在测试数据目录查找
test_data_dir = os.path.abspath(os.path.join(script_dir, "test_data"))
test_data_path = os.path.join(test_data_dir, file_path)
if os.path.exists(test_data_path):
return test_data_path
# 所有尝试失败
logger.error(f"文件路径解析失败: {file_path}")
logger.info(f"当前工作目录: {current_dir}")
logger.info(f"脚本所在目录: {script_dir}")
logger.info(f"项目根目录: {project_root}")
logger.info(f"测试数据目录: {test_data_dir}")
raise FileNotFoundError(f"文件不存在: {file_path}")
def extract_text_from_pdf(self, pdf_path: str) -> Dict[int, str]:
"""从PDF提取文字（优先使用文本层，失败则使用OCR）"""
text_dict = {}
logger.info(f"开始处理PDF文件: {pdf_path}")
doc = fitz.open(pdf_path)
for page_num in range(len(doc)):
page = doc.load_page(page_num)
text = page.get_text()
# 如果文本层内容过少（<100字符），则使用OCR
if len(text) < 100:
logger.info(f"页面 {page_num} 文本层内容不足，启用OCR")
# 获取页面图像
pix = page.get_pixmap(matrix=fitz.Matrix(300 / 72, 300 / 72))
img_data = pix.samples
# 使用内存中的图像数据避免文件系统操作
img_array = np.frombuffer(img_data, dtype=np.uint8).reshape(
pix.height, pix.width, pix.n
# 使用通用的OCR调用方式
ocr_result = self.ocr.ocr(img_array)
recognized_text = ""
if ocr_result:
for line in ocr_result[0]:
if line and len(line) > 1 and isinstance(line[1], tuple):
recognized_text += line[1][0] + "\n"
elif isinstance(line, list) and len(line) > 1:
recognized_text += line[1][0] + "\n"
text_dict[page_num] = recognized_text
logger.info(f"页面 {page_num} OCR完成，识别字符数: {len(recognized_text)}")
except Exception as e:
logger.error(f"OCR处理失败: {str(e)}，跳过此页面")
text_dict[page_num] = "" # 返回空文本
else:
text_dict[page_num] = text
logger.info(f"页面 {page_num} 使用文本层，字符数: {len(text)}")
return text_dict
except Exception as e:
logger.error(f"PDF处理失败: {str(e)}", exc_info=True)
raise RuntimeError(f"PDF处理失败: {str(e)}")
def parse_reinforcement(self, text: str) -> List[Dict]:
"""提取钢筋信息（考虑品种分类）"""
results = []
# 匹配钢筋直径和数量
for pattern in self.rules['reinforcement']:
matches = re.findall(pattern, text, re.IGNORECASE)
for match in matches:
# 处理不同匹配模式
if len(match) == 3:
grade, diameter, count = match
elif len(match) == 4:
# 处理包含间距的匹配
grade = match[0] if match[0] else "三级钢"
diameter = match[1] if match[1] else match[3]
count = match[2] if match[2] else match[4]
elif len(match) == 2:
# 无品种信息，默认三级钢
grade = "三级钢"
diameter, count = match
else:
continue
# 标准化钢筋品种
grade = grade.strip()
grade = self.steel_grade_map.get(grade, "未知")
# 计算截面积 (mm²)
diameter = float(diameter)
cross_area = math.pi * (diameter/2)**2
# 添加到结果
results.append({
"grade": grade,
"diameter": diameter,
"count": int(count),
"cross_area": cross_area
logger.debug(f"匹配到钢筋: {grade}, Ø{diameter}mm, {count}根")
except (ValueError, TypeError) as ve:
logger.warning(f"无效的钢筋数据: {match} - {str(ve)}")
return results
def parse_beams(self, text: str) -> List[Dict]:
"""提取框架梁信息（单位为毫米）"""
results = []
matches = re.findall(self.rules['beam'], text, re.IGNORECASE)
for match in matches:
width = float(match[0])
height = float(match[1])
count = int(match[2])
start_axis = match[3]
end_axis = match[4] if len(match) > 4 and match[4] else start_axis
# 单位为毫米
results.append({
"width": width,
"height": height,
"cross_area": width * height,
"count": count,
"location": f"{start_axis}-{end_axis}" if start_axis != end_axis else start_axis
logger.debug(f"匹配到框架梁: {width}x{height}mm, {count}根, 位置: {start_axis}-{end_axis}")
except (ValueError, IndexError, TypeError) as e:
logger.warning(f"无效的梁数据: {match} - {str(e)}")
return results
def extract_drawing_info(self, text_dict: Dict[int, str]) -> Dict[str, Any]:
"""从提取的文本中解析结构化信息"""
full_text = "\n".join(text_dict.values())
logger.info(f"提取的完整文本长度: {len(full_text)}字符")
results = {
"drawing_title": "未识别",
"drawing_number": "未识别",
"reinforcement": [],
"reinforcement_summary": {},
"beams": [],
"axis_dimensions": None,
"column_dimensions": None,
"shear_wall": {"present": False, "thickness": 0},
"concrete_volume": 0,
"max_beam": None
# 图纸标题和编号
title_match = re.search(self.rules['drawing_title'], full_text, re.IGNORECASE)
number_match = re.search(self.rules['drawing_number'], full_text, re.IGNORECASE)
if title_match:
results["drawing_title"] = title_match.group(2).strip()
if number_match:
results["drawing_number"] = number_match.group(2).strip()
# 轴线尺寸（单位为米）
axis_match = re.search(self.rules['axis'], full_text, re.IGNORECASE)
if axis_match:
results["axis_dimensions"] = {
"width": float(axis_match.group(1)) * 1000, # 米转毫米
"length": float(axis_match.group(2)) * 1000 # 米转毫米
except (ValueError, IndexError):
logger.warning("无法解析轴线尺寸")
# 柱子尺寸（单位为米）
column_match = re.search(self.rules['column'], full_text, re.IGNORECASE)
if column_match:
results["column_dimensions"] = {
"width": float(column_match.group(1)) * 1000, # 米转毫米
"height": float(column_match.group(2)) * 1000 # 米转毫米
except (ValueError, IndexError):
logger.warning("无法解析柱子尺寸")
# 剪力墙（单位为毫米）
shear_wall_match = re.search(self.rules['shear_wall'], full_text, re.IGNORECASE)
if shear_wall_match:
results["shear_wall"] = {
"present": True,
"thickness": float(shear_wall_match.group(1))
except (ValueError, IndexError):
logger.warning("无法解析剪力墙厚度")
# 混凝土总量（单位为立方米）
concrete_match = re.search(self.rules['concrete'], full_text, re.IGNORECASE)
if concrete_match:
results["concrete_volume"] = float(concrete_match.group(1))
except ValueError:
logger.warning("无法解析混凝土总量")
# 提取钢筋和梁的信息
results["reinforcement"] = self.parse_reinforcement(full_text)
results["beams"] = self.parse_beams(full_text)
# 按钢筋品种分类
reinforcement_summary = {}
for item in results["reinforcement"]:
grade = item['grade']
if grade not in reinforcement_summary:
reinforcement_summary[grade] = {
'total_count': 0,
'total_cross_area': 0.0,
'diameters': {}
reinforcement_summary[grade]['total_count'] += item['count']
reinforcement_summary[grade]['total_cross_area'] += item['cross_area'] * item['count']
# 记录不同直径
diameter = item['diameter']
if diameter not in reinforcement_summary[grade]['diameters']:
reinforcement_summary[grade]['diameters'][diameter] = {
'count': 0,
'cross_area': item['cross_area']
reinforcement_summary[grade]['diameters'][diameter]['count'] += item['count']
results["reinforcement_summary"] = reinforcement_summary
# 计算最大截面的梁
if results["beams"]:
max_area_beam = max(results["beams"], key=lambda x: x['cross_area'])
results["max_beam"] = max_area_beam
logger.info("图纸信息提取完成")
return results
def process(self, pdf_path: str) -> Dict[str, Any]:
"""处理PDF并返回结构化数据"""
# 步骤1: 解析文件路径
resolved_path = self.resolve_file_path(pdf_path)
logger.info(f"使用解析后的路径: {resolved_path}")
# 步骤2: 提取PDF文本
text_dict = self.extract_text_from_pdf(resolved_path)
# 步骤3: 解析关键信息
return self.extract_drawing_info(text_dict)
except Exception as e:
logger.error(f"处理过程中出错: {str(e)}", exc_info=True)
raise
def main(params: Dict) -> Dict:
"""Dify插件入口函数"""
# 获取输入文件路径
pdf_path = params.get("input_file", "")
if not pdf_path:
return {
"error": "未提供文件路径",
"code": 400,
"message": "input_file参数为空"
# 初始化提取器
extractor = ConstructionDrawingExtractor()
# 处理PDF
result = extractor.process(pdf_path)
# 格式化为Dify需要的输出
# 轴线尺寸（毫米）
axis_info = "未识别"
if result.get('axis_dimensions'):
axis_info = (
f"宽度: {result['axis_dimensions']['width']:.0f}mm, "
f"长度: {result['axis_dimensions']['length']:.0f}mm"
# 柱尺寸（毫米）
column_info = "未识别"
if result.get('column_dimensions'):
column_info = (
f"宽度: {result['column_dimensions']['width']:.0f}mm, "
f"高度: {result['column_dimensions']['height']:.0f}mm"
# 剪力墙
shear_wall_info = "无"
if result.get('shear_wall', {}).get('present'):
shear_wall_info = f"存在, 厚度: {result['shear_wall']['thickness']:.0f}毫米"
# 最大梁
max_beam_info = "无"
if result.get('max_beam'):
max_beam = result['max_beam']
max_beam_info = (
f"尺寸: {max_beam['width']:.0f}×{max_beam['height']:.0f}mm, "
f"数量: {max_beam['count']}, 位置: {max_beam['location']}"
# 钢筋分类汇总信息
reinforcement_summary = []
reinforcement_data = result.get('reinforcement_summary', {})
for grade, data in reinforcement_data.items():
grade_info = {
"品种": grade,
"总数量": data['total_count'],
"总截面积(mm²)": f"{data['total_cross_area']:.2f}",
"直径分布": []
for diameter, details in data['diameters'].items():
grade_info["直径分布"].append({
"直径(mm)": diameter,
"数量": details['count'],
"单根截面积(mm²)": f"{details['cross_area']:.2f}"
reinforcement_summary.append(grade_info)
# 生成梁尺寸统计表格
beam_table = []
for beam in result.get('beams', []):
beam_table.append({
"宽度(mm)": beam['width'],
"高度(mm)": beam['height'],
"截面面积(mm²)": f"{beam['cross_area']:.2f}",
"数量": beam
报错：D:\dify-1.8.0\dify\pdf_blueprint_drawing\venv\Scripts\python.exe "D:/Program Files/JetBrains/PyCharm 2025.2.1.1/plugins/python-ce/helpers/pydev/pydevd.py" --multiprocess --qt-support=auto --client 127.0.0.1 --port 55011 --file D:\dify-1.8.0\dify\pdf_blueprint_drawing\tools\pdf_blueprint_drawing.py
Connected to: <socket.socket fd=676, family=2, type=1, proto=0, laddr=('127.0.0.1', 55012), raddr=('127.0.0.1', 55011)>.
已连接到 pydev 调试器(内部版本号 252.25557.178)Traceback (most recent call last):
File "D:\Program Files\JetBrains\PyCharm 2025.2.1.1\plugins\python-ce\helpers\pydev\pydevd.py", line 1648, in _exec
pydev_imports.execfile(file, globals, locals) # execute the script
~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^
File "D:\Program Files\JetBrains\PyCharm 2025.2.1.1\plugins\python-ce\helpers\pydev\_pydev_imps\_pydev_execfile.py", line 18, in execfile
exec(compile(contents+"\n", file, 'exec'), glob, loc)
~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\dify-1.8.0\dify\pdf_blueprint_drawing\tools\pdf_blueprint_drawing.py", line 422
beam_table.append({
SyntaxError: '{' was never closed
进程已结束，退出代码为 1
import shutil
from typing import List, Dict, Tuple, Optional, Set
from threading import Lock, Semaphore, RLock
from datetime import datetime
from pydub import AudioSegment
from pydub.silence import split_on_silence
from modelscope.pipelines import pipeline
from modelscope.utils.constant import Tasks
from transformers import AutoModelForSequenceClassification, AutoTokenizer
from torch.utils.data import TensorDataset, DataLoader
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
QPushButton, QLabel, QLineEdit, QTextEdit, QFileDialog, QProgressBar,
QGroupBox, QMessageBox, QListWidget, QSplitter, QTabWidget, QTableWidget,
QTableWidgetItem, QHeaderView, QAction, QMenu, QToolBar, QCheckBox,
QComboBox, QSpinBox, QDialog, QDialogButtonBox, QStatusBar)
from PyQt5.QtCore import QThread, pyqtSignal, Qt, QTimer, QSize
from PyQt5.QtGui import QFont, QTextCursor, QColor, QIcon
# ====================== 资源监控器 ======================
class ResourceMonitor:
"""统一资源监控器（增强版）"""
def __init__(self):
self.gpu_available = torch.cuda.is_available()
def memory_percent(self) -> Dict[str, float]:
"""获取内存使用百分比，同时返回CPU和GPU信息"""
result = {
"cpu": psutil.virtual_memory().percent
if self.gpu_available:
allocated = torch.cuda.memory_allocated() / (1024 ** 3)
total = torch.cuda.get_device_properties(0).total_memory / (1024 ** 3)
result["gpu"] = (allocated / total) * 100 if total > 0 else 0
return result
except Exception as e:
print(f"获取内存使用百分比失败: {str(e)}")
return {"cpu": 0, "gpu": 0}
# ====================== 方言配置中心（优化版） ======================
class DialectConfig:
"""集中管理方言配置，便于维护和扩展（带缓存）"""
# 标准关键词
STANDARD_KEYWORDS = {
"opening": ["您好", "很高兴为您服务", "请问有什么可以帮您"],
"closing": ["感谢来电", "祝您生活愉快", "再见"],
"forbidden": ["不知道", "没办法", "你投诉吧", "随便你"]
# 贵州方言关键词
GUIZHOU_KEYWORDS = {
"opening": ["麻烦您喽", "请问搞哪样", "有咋个可以帮您", "多谢喽"],
"closing": ["搞归一喽", "麻烦您喽", "再见喽", "慢走喽"],
"forbidden": ["搞不成", "没得法", "随便你喽", "你投诉吧喽"]
# 方言到标准表达的映射（扩展更多贵州方言）
DIALECT_MAPPING = {
"恼火得很": "非常生气",
"鬼火戳": "很愤怒",
"搞不成": "无法完成",
"没得": "没有",
"搞哪样嘛": "做什么呢",
"归一喽": "完成了",
"咋个": "怎么",
"克哪点": "去哪里",
"麻烦您喽": "麻烦您了",
"多谢喽": "多谢了",
"憨包": "傻瓜",
"归一": "结束",
"板扎": "很好",
"鬼火冒": "非常生气",
"背时": "倒霉",
"吃豁皮": "占便宜"
# 类属性缓存
_combined_keywords = None
_compiled_opening = None
_compiled_closing = None
_hotwords = None
_dialect_trie = None # 使用Trie树替换正则表达式
class TrieNode:
"""Trie树节点类"""
def __init__(self):
self.children = {}
self.is_end = False
self.value = ""
@classmethod
def _build_dialect_trie(cls):
"""构建方言Trie树"""
root = cls.TrieNode()
# 按长度降序添加关键词
for dialect, standard in sorted(cls.DIALECT_MAPPING.items(),
key=lambda x: len(x[0]), reverse=True):
node = root
for char in dialect:
if char not in node.children:
node.children[char] = cls.TrieNode()
node = node.children[char]
node.is_end = True
node.value = standard
return root
@classmethod
def get_combined_keywords(cls) -> Dict[str, List[str]]:
"""获取合并后的关键词集（带缓存）"""
if cls._combined_keywords is None:
cls._combined_keywords = {
"opening": cls.STANDARD_KEYWORDS["opening"] + cls.GUIZHOU_KEYWORDS["opening"],
"closing": cls.STANDARD_KEYWORDS["closing"] + cls.GUIZHOU_KEYWORDS["closing"],
"forbidden": cls.STANDARD_KEYWORDS["forbidden"] + cls.GUIZHOU_KEYWORDS["forbidden"]
return cls._combined_keywords
@classmethod
def get_compiled_opening(cls) -> List[re.Pattern]:
"""获取预编译的开场关键词正则表达式（带缓存）"""
if cls._compiled_opening is None:
keywords = cls.get_combined_keywords()["opening"]
cls._compiled_opening = [re.compile(re.escape(kw)) for kw in keywords]
return cls._compiled_opening
@classmethod
def get_compiled_closing(cls) -> List[re.Pattern]:
"""获取预编译的结束关键词正则表达式（带缓存）"""
if cls._compiled_closing is None:
keywords = cls.get_combined_keywords()["closing"]
cls._compiled_closing = [re.compile(re.escape(kw)) for kw in keywords]
return cls._compiled_closing
@classmethod
def get_asr_hotwords(cls) -> List[str]:
"""获取ASR热词列表（带缓存）"""
if cls._hotwords is None:
combined = cls.get_combined_keywords()
cls._hotwords = sorted(set(
combined["opening"] + combined["closing"]
return cls._hotwords
@classmethod
def preprocess_text(cls, texts: List[str]) -> List[str]:
"""将方言文本转换为标准表达（使用Trie树优化）"""
if cls._dialect_trie is None:
cls._dialect_trie = cls._build_dialect_trie()
processed_texts = []
for text in texts:
# 使用Trie树进行高效替换
processed = []
i = 0
n = len(text)
while i < n:
node = cls._dialect_trie
j = i
found = False
# 查找最长匹配
while j < n and text[j] in node.children:
node = node.children[text[j]]
j += 1
if node.is_end:
processed.append(node.value)
i = j
found = True
break
if not found:
processed.append(text[i])
i += 1
processed_texts.append(''.join(processed))
return processed_texts
# ====================== 系统配置管理器 ======================
class ConfigManager:
"""管理应用程序配置"""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._init_config()
return cls._instance
def _init_config(self):
"""初始化默认配置"""
self.config = {
"model_paths": {
"asr": "./models/iic-speech_paraformer-large-vad-punc-spk_asr_nat-zh-cn",
"sentiment": "./models/IDEA-CCNL-Erlangshen-Roberta-110M-Sentiment"
"sample_rate": 16000,
"silence_thresh": -40,
"min_silence_len": 1000,
"max_concurrent": 1,
"dialect_config": "guizhou",
"max_audio_duration": 3600 # 最大音频时长（秒）
self.load_config()
def load_config(self):
"""从文件加载配置"""
if os.path.exists("config.json"):
with open("config.json", "r") as f:
self.config.update(json.load(f))
except:
def save_config(self):
"""保存配置到文件"""
with open("config.json", "w") as f:
json.dump(self.config, f, indent=2)
except:
def get(self, key: str, default=None):
"""获取配置值"""
return self.config.get(key, default)
def set(self, key: str, value):
"""设置配置值"""
self.config[key] = value
self.save_config()
# ====================== 音频处理工具（优化版） ======================
class AudioProcessor:
"""处理音频转换和特征提取（避免重复加载）"""
SUPPORTED_FORMATS = ('.mp3', '.wav', '.amr', '.m4a')
@staticmethod
def convert_to_wav(input_path: str, temp_dir: str) -> Optional[List[str]]:
"""将音频转换为WAV格式（在静音处分割）"""
os.makedirs(temp_dir, exist_ok=True)
# 检查文件格式
if not any(input_path.lower().endswith(ext) for ext in AudioProcessor.SUPPORTED_FORMATS):
raise ValueError(f"不支持的音频格式: {os.path.splitext(input_path)[1]}")
if input_path.lower().endswith('.wav'):
return [input_path] # 已经是WAV格式
# 检查ffmpeg是否可用
AudioSegment.converter = "ffmpeg" # 显式指定ffmpeg
audio = AudioSegment.from_file(input_path)
except FileNotFoundError:
print("错误: 未找到ffmpeg，请安装并添加到环境变量")
return None
# 检查音频时长是否超过限制
max_duration = ConfigManager().get("max_audio_duration", 3600) * 1000 # 毫秒
if len(audio) > max_duration:
return AudioProcessor._split_long_audio(audio, input_path, temp_dir)
else:
return AudioProcessor._convert_single_audio(audio, input_path, temp_dir)
except Exception as e:
print(f"格式转换失败: {str(e)}")
return None
@staticmethod
def _split_long_audio(audio: AudioSegment, input_path: str, temp_dir: str) -> List[str]:
"""分割长音频文件"""
wav_paths = []
# 在静音处分割音频
chunks = split_on_silence(
audio,
min_silence_len=ConfigManager().get("min_silence_len", 1000),
silence_thresh=ConfigManager().get("silence_thresh", -40),
keep_silence=500
# 合并小片段，避免分段过多
merged_chunks = []
current_chunk = AudioSegment.empty()
for chunk in chunks:
if len(current_chunk) + len(chunk) < 5 * 60 * 1000: # 5分钟
current_chunk += chunk
else:
if len(current_chunk) > 0:
merged_chunks.append(current_chunk)
current_chunk = chunk
if len(current_chunk) > 0:
merged_chunks.append(current_chunk)
# 导出分段音频
sample_rate = ConfigManager().get("sample_rate", 16000)
for i, chunk in enumerate(merged_chunks):
chunk = chunk.set_frame_rate(sample_rate).set_channels(1)
chunk_path = os.path.join(
temp_dir,
f"{os.path.splitext(os.path.basename(input_path))[0]}_part{i + 1}.wav"
chunk.export(chunk_path, format="wav")
wav_paths.append(chunk_path)
return wav_paths
@staticmethod
def _convert_single_audio(audio: AudioSegment, input_path: str, temp_dir: str) -> List[str]:
"""转换单个短音频文件"""
sample_rate = ConfigManager().get("sample_rate", 16000)
audio = audio.set_frame_rate(sample_rate).set_channels(1)
wav_path = os.path.join(temp_dir, os.path.splitext(os.path.basename(input_path))[0] + ".wav")
audio.export(wav_path, format="wav")
return [wav_path]
@staticmethod
def extract_features_from_audio(y: np.ndarray, sr: int) -> Dict[str, float]:
"""从音频数据中提取特征（流式处理优化）"""
duration = librosa.get_duration(y=y, sr=sr)
segment_length = 60 # 60秒分段
total_segments = max(1, int(np.ceil(duration / segment_length)))
syllable_rates = []
volume_stabilities = []
total_samples = len(y)
samples_per_segment = int(segment_length * sr)
# 流式处理每个分段
for i in range(total_segments):
start = i * samples_per_segment
end = min((i + 1) * samples_per_segment, total_samples)
y_segment = y[start:end]
if len(y_segment) == 0:
continue
# 语速计算（使用VAD检测语音段）
intervals = librosa.effects.split(y_segment, top_db=20)
speech_samples = sum(end - start for start, end in intervals)
speech_duration = speech_samples / sr
if speech_duration > 0.1:
syllable_rate = len(intervals) / speech_duration
else:
syllable_rate = 0
syllable_rates.append(syllable_rate)
# 音量稳定性（使用RMS能量）
rms = librosa.feature.rms(y=y_segment, frame_length=2048, hop_length=512)[0]
if len(rms) > 0 and np.mean(rms) > 0:
volume_stability = np.std(rms) / np.mean(rms)
volume_stabilities.append(volume_stability)
# 计算加权平均值（按时长加权）
valid_syllable = [r for r in syllable_rates if r > 0]
valid_volume = [v for v in volume_stabilities if v > 0]
return {
"duration": duration,
"syllable_rate": round(np.mean(valid_syllable) if valid_syllable else 0, 2),
"volume_stability": round(np.mean(valid_volume) if valid_volume else 0, 4)
except Exception as e:
print(f"特征提取错误: {str(e)}")
return {"duration": 0, "syllable_rate": 0, "volume_stability": 0}
# ====================== 模型加载器（优化版） ======================
class ModelLoader:
"""加载和管理AI模型（使用RLock）"""
asr_pipeline = None
sentiment_model = None
sentiment_tokenizer = None
model_lock = RLock() # 使用RLock代替Lock
models_loaded = False # 添加模型加载状态标志
@classmethod
def load_models(cls):
"""加载所有模型"""
config = ConfigManager()
# 加载ASR模型
if not cls.asr_pipeline:
with cls.model_lock:
if not cls.asr_pipeline: # 双重检查锁定
cls.load_asr_model(config.get("model_paths")["asr"])
# 加载情感分析模型
if not cls.sentiment_model:
with cls.model_lock:
if not cls.sentiment_model: # 双重检查锁定
cls.load_sentiment_model(config.get("model_paths")["sentiment"])
cls.models_loaded = True
@classmethod
def reload_models(cls):
"""重新加载模型（配置变更后）"""
with cls.model_lock:
cls.asr_pipeline = None
cls.sentiment_model = None
cls.sentiment_tokenizer = None
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
cls.load_models()
@classmethod
def load_asr_model(cls, model_path: str):
"""加载语音识别模型"""
if not os.path.exists(model_path):
raise FileNotFoundError(f"ASR模型路径不存在: {model_path}")
asr_kwargs = {}
if hasattr(torch, 'quantization'):
asr_kwargs['quantize'] = 'int8'
print("启用ASR模型量化")
cls.asr_pipeline = pipeline(
task=Tasks.auto_speech_recognition,
model=model_path,
device='cuda' if torch.cuda.is_available() else 'cpu',
**asr_kwargs
print("ASR模型加载完成")
except Exception as e:
print(f"加载ASR模型失败: {str(e)}")
raise
@classmethod
def load_sentiment_model(cls, model_path: str):
"""加载情感分析模型"""
if not os.path.exists(model_path):
raise FileNotFoundError(f"情感分析模型路径不存在: {model_path}")
cls.sentiment_model = AutoModelForSequenceClassification.from_pretrained(model_path)
cls.sentiment_tokenizer = AutoTokenizer.from_pretrained(model_path)
if torch.cuda.is_available():
cls.sentiment_model = cls.sentiment_model.cuda()
print("情感分析模型加载完成")
except Exception as e:
print(f"加载情感分析模型失败: {str(e)}")
raise
# ====================== 核心分析线程（优化版） ======================
class AnalysisThread(QThread):
progress_updated = pyqtSignal(int, str, str)
result_ready = pyqtSignal(dict)
finished_all = pyqtSignal()
error_occurred = pyqtSignal(str, str)
memory_warning = pyqtSignal()
resource_cleanup = pyqtSignal()
def __init__(self, audio_paths: List[str], temp_dir: str = "temp_wav"):
super().__init__()
self.audio_paths = audio_paths
self.temp_dir = temp_dir
self.is_running = True
self.current_file = ""
self.max_concurrent = min(
ConfigManager().get("max_concurrent", 1),
self.get_max_concurrent_tasks()
self.resource_monitor = ResourceMonitor()
self.semaphore = Semaphore(self.max_concurrent)
os.makedirs(temp_dir, exist_ok=True)
def run(self):
if not ModelLoader.models_loaded:
self.error_occurred.emit("模型未加载", "请等待模型加载完成后再开始分析")
return
self.progress_updated.emit(0, f"最大并行任务数: {self.max_concurrent}", "")
# 使用线程池并行处理
with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_concurrent) as executor:
# 创建任务
future_to_path = {}
for path in self.audio_paths:
if not self.is_running:
break
# 使用信号量控制并发
self.semaphore.acquire()
batch_size = self.get_available_batch_size()
future = executor.submit(self.analyze_audio, path, batch_size)
future_to_path[future] = path
future.add_done_callback(lambda f: self.semaphore.release())
# 处理完成的任务
for i, future in enumerate(concurrent.futures.as_completed(future_to_path)):
if not self.is_running:
break
path = future_to_path[future]
self.current_file = os.path.basename(path)
# 内存检查
if self.check_memory_usage():
self.memory_warning.emit()
self.is_running = False
break
result = future.result()
if result:
self.result_ready.emit(result)
# 更新进度
progress = int((i + 1) / len(self.audio_paths) * 100)
self.progress_updated.emit(
progress,
f"完成: {self.current_file} ({i + 1}/{len(self.audio_paths)})",
self.current_file
except Exception as e:
result = {
"file_name": self.current_file,
"status": "error",
"error": f"分析失败: {str(e)}"
self.result_ready.emit(result)
# 分析完成后
if self.is_running:
self.finished_all.emit()
except Exception as e:
self.error_occurred.emit("系统错误", str(e))
traceback.print_exc()
finally:
# 确保资源清理
self.resource_cleanup.emit()
self.cleanup_resources()
def analyze_audio(self, audio_path: str, batch_size: int) -> Dict:
"""分析单个音频文件（整合所有优化）"""
result = {
"file_name": os.path.basename(audio_path),
"status": "processing"
wav_paths = []
# 1. 音频格式转换
wav_paths = AudioProcessor.convert_to_wav(audio_path, self.temp_dir)
if not wav_paths:
result["error"] = "格式转换失败（请检查ffmpeg是否安装）"
result["status"] = "error"
return result
# 2. 提取音频特征（合并所有分段）
audio_features = self._extract_audio_features(wav_paths)
result.update(audio_features)
result["duration_str"] = self._format_duration(audio_features["duration"])
# 3. 语音识别与处理（使用批处理优化）
all_segments, full_text = self._process_asr_segments(wav_paths)
# 4. 说话人区分（使用优化后的方法）
agent_segments, customer_segments = self.identify_speakers(all_segments)
# 5. 生成带说话人标签的文本
labeled_text = self._generate_labeled_text(all_segments, agent_segments, customer_segments)
result["asr_text"] = labeled_text.strip()
# 6. 文本分析（包含方言预处理）
text_analysis = self._analyze_text(agent_segments, customer_segments, batch_size)
result.update(text_analysis)
# 7. 服务规范检查（使用方言适配的关键词）
service_check = self._check_service_rules(agent_segments)
result.update(service_check)
# 8. 问题解决率（上下文关联）
result["issue_resolved"] = self._check_issue_resolution(customer_segments, agent_segments)
result["status"] = "success"
except Exception as e:
result["error"] = f"分析失败: {str(e)}"
result["status"] = "error"
finally:
# 清理临时文件（使用优化后的清理方法）
self._cleanup_temp_files(wav_paths)
# 显式内存清理
self.cleanup_resources()
return result
def identify_speakers(self, segments: List[Dict]) -> Tuple[List[Dict], List[Dict]]:
"""区分客服与客户（增强版）"""
if not segments:
return [], []
# 1. 基于关键词的识别
agent_id = self._identify_by_keywords(segments)
# 2. 基于说话模式的识别（如果关键词识别失败）
if agent_id is None and len(segments) >= 4:
agent_id = self._identify_by_speech_patterns(segments)
# 3. 使用说话频率最高的作为客服（最后手段）
if agent_id is None:
spk_counts = {}
for seg in segments:
spk_id = seg["spk_id"]
spk_counts[spk_id] = spk_counts.get(spk_id, 0) + 1
agent_id = max(spk_counts, key=spk_counts.get) if spk_counts else None
if agent_id is None:
return [], []
# 使用集合存储agent的spk_id
agent_spk_ids = {agent_id}
return (
[seg for seg in segments if seg["spk_id"] in agent_spk_ids],
[seg for seg in segments if seg["spk_id"] not in agent_spk_ids]
def _identify_by_keywords(self, segments: List[Dict]) -> Optional[str]:
"""基于关键词识别客服"""
opening_patterns = DialectConfig.get_compiled_opening()
closing_patterns = DialectConfig.get_compiled_closing()
# 策略1：在前3段中查找开场白关键词
for seg in segments[:3]:
text = seg["text"]
for pattern in opening_patterns:
if pattern.search(text):
return seg["spk_id"]
# 策略2：在后3段中查找结束语关键词
for seg in reversed(segments[-3:] if len(segments) >= 3 else segments):
text = seg["text"]
for pattern in closing_patterns:
if pattern.search(text):
return seg["spk_id"]
return None
def _identify_by_speech_patterns(self, segments: List[Dict]) -> Optional[str]:
"""基于说话模式识别客服"""
# 分析说话模式特征
speaker_features = {}
for seg in segments:
spk_id = seg["spk_id"]
if spk_id not in speaker_features:
speaker_features[spk_id] = {
"total_duration": 0.0,
"turn_count": 0,
"question_count": 0
features = speaker_features[spk_id]
features["total_duration"] += (seg["end"] - seg["start"])
features["turn_count"] += 1
# 检测问题（包含疑问词）
if any(q_word in seg["text"] for q_word in ["吗", "呢", "？", "?", "如何", "怎样"]):
features["question_count"] += 1
# 客服通常说话时间更长、提问更多
if speaker_features:
# 计算说话时间占比
max_duration = max(f["total_duration"] for f in speaker_features.values())
# 计算提问频率
question_rates = {
spk_id: features["question_count"] / features["turn_count"]
for spk_id, features in speaker_features.items()
# 综合评分
candidates = []
for spk_id, features in speaker_features.items():
score = (
0.6 * (features["total_duration"] / max_duration) +
0.4 * question_rates[spk_id]
candidates.append((spk_id, score))
# 返回得分最高的说话人
return max(candidates, key=lambda x: x[1])[0]
return None
def _analyze_text(self, agent_segments: List[Dict],
customer_segments: List[Dict],
batch_size: int) -> Dict:
"""文本情感分析（优化版：向量化批处理）"""
def analyze_speaker(segments: List[Dict], speaker_type: str) -> Dict:
if not segments:
return {
f"{speaker_type}_negative": 0.0,
f"{speaker_type}_neutral": 1.0,
f"{speaker_type}_positive": 0.0,
f"{speaker_type}_emotions": "无"
# 方言预处理 - 使用优化的一次性替换
texts = [seg["text"] for seg in segments]
processed_texts = DialectConfig.preprocess_text(texts)
# 使用DataLoader进行批处理
with ModelLoader.model_lock:
inputs = ModelLoader.sentiment_tokenizer(
processed_texts,
padding=True,
truncation=True,
max_length=128,
return_tensors="pt"
# 创建TensorDataset和DataLoader
dataset = TensorDataset(inputs['input_ids'], inputs['attention_mask'])
dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=False)
device = "cuda" if torch.cuda.is_available() else "cpu"
sentiment_dist = []
emotions = []
# 批量处理
for batch in dataloader:
input_ids, attention_mask = batch
inputs = {
'input_ids': input_ids.to(device),
'attention_mask': attention_mask.to(device)
with torch.no_grad():
outputs = ModelLoader.sentiment_model(**inputs)
batch_probs = torch.nn.functional.softmax(outputs.logits, dim=-1)
sentiment_dist.append(batch_probs.cpu())
# 情绪识别（批量）
emotion_keywords = ["愤怒", "生气", "鬼火", "不耐烦", "搞哪样嘛", "恼火", "背时"]
for text in processed_texts:
if any(kw in text for kw in emotion_keywords):
if any(kw in text for kw in ["愤怒", "生气", "鬼火", "恼火"]):
emotions.append("愤怒")
elif any(kw in text for kw in ["不耐烦", "搞哪样嘛"]):
emotions.append("不耐烦")
elif "背时" in text:
emotions.append("沮丧")
# 合并结果
if sentiment_dist:
all_probs = torch.cat(sentiment_dist, dim=0)
avg_sentiment = torch.mean(all_probs, dim=0).tolist()
else:
avg_sentiment = [0.0, 1.0, 0.0] # 默认值
return {
f"{speaker_type}_negative": round(avg_sentiment[0], 4),
f"{speaker_type}_neutral": round(avg_sentiment[1], 4),
f"{speaker_type}_positive": round(avg_sentiment[2], 4),
f"{speaker_type}_emotions": ",".join(set(emotions)) if emotions else "无"
return {
**analyze_speaker(agent_segments, "agent"),
**analyze_speaker(customer_segments, "customer")
# ====================== 辅助方法 ======================
def get_available_batch_size(self) -> int:
"""根据GPU内存动态调整batch size（考虑并行）"""
if not torch.cuda.is_available():
return 4 # CPU默认批次
total_mem = torch.cuda.get_device_properties(0).total_memory / (1024 ** 3) # GB
per_task_mem = total_mem / self.max_concurrent
# 修正批次大小逻辑：显存越少，批次越小
if per_task_mem < 2:
return 2
elif per_task_mem < 4:
return 4
else:
return 8
def get_max_concurrent_tasks(self) -> int:
"""根据系统资源计算最大并行任务数"""
if torch.cuda.is_available():
total_mem = torch.cuda.get_device_properties(0).total_memory / (1024 ** 3)
if total_mem < 6:
return 1
elif total_mem < 12:
return 2
else:
return 3
else:
# CPU模式下根据核心数设置
return max(1, os.cpu_count() // 2)
def check_memory_usage(self) -> bool:
mem_percent = self.resource_monitor.memory_percent()
return mem_percent.get("cpu", 0) > 85 or mem_percent.get("gpu", 0) > 85
except:
return False
def _extract_audio_features(self, wav_paths: List[str]) -> Dict[str, float]:
"""提取音频特征（合并所有分段）"""
combined_y = np.array([], dtype=np.float32)
sr = ConfigManager().get("sample_rate", 16000)
for path in wav_paths:
y, _ = librosa.load(path, sr=sr)
combined_y = np.concatenate((combined_y, y))
return AudioProcessor.extract_features_from_audio(combined_y, sr)
def _process_asr_segments(self, wav_paths: List[str]) -> Tuple[List[Dict], str]:
"""处理ASR分段（批处理优化）"""
segments = []
full_text = ""
# 分批处理（根据GPU内存动态调整批次大小）
batch_size = min(4, len(wav_paths), self.get_available_batch_size())
for i in range(0, len(wav_paths), batch_size):
if not self.is_running:
break
batch_paths = wav_paths[i:i + batch_size]
# 批处理调用ASR模型
results = ModelLoader.asr_pipeline(
batch_paths,
hotwords=DialectConfig.get_asr_hotwords(),
output_dir=None,
batch_size=batch_size
for result in results:
for seg in result[0]["sentences"]:
segments.append({
"start": seg["start"],
"end": seg["end"],
"text": seg["text"],
"spk_id": seg.get("spk_id", "0")
full_text += seg["text"] + " "
except Exception as e:
print(f"ASR批处理错误: {str(e)}")
# 失败时回退到单文件处理
for path in batch_paths:
result = ModelLoader.asr_pipeline(
path,
hotwords=DialectConfig.get_asr_hotwords(),
output_dir=None
for seg in result[0]["sentences"]:
segments.append({
"start": seg["start"],
"end": seg["end"],
"text": seg["text"],
"spk_id": seg.get("spk_id", "0")
full_text += seg["text"] + " "
except:
continue
return segments, full_text.strip()
def _generate_labeled_text(self, all_segments: List[Dict],
agent_segments: List[Dict],
customer_segments: List[Dict]) -> str:
"""生成带说话人标签的文本"""
agent_spk_id = agent_segments[0]["spk_id"] if agent_segments else None
customer_spk_id = customer_segments[0]["spk_id"] if customer_segments else None
labeled_text = []
for seg in all_segments:
if seg["spk_id"] == agent_spk_id:
speaker = "客服"
elif seg["spk_id"] == customer_spk_id:
speaker = "客户"
else:
speaker = f"说话人{seg['spk_id']}"
labeled_text.append(f"[{speaker}]: {seg['text']}")
return "\n".join(labeled_text)
def _check_service_rules(self, agent_segments: List[Dict]) -> Dict:
"""检查服务规范"""
forbidden_keywords = DialectConfig.get_combined_keywords()["forbidden"]
found_forbidden = []
found_opening = False
found_closing = False
# 检查开场白（前3段）
for seg in agent_segments[:3]:
text = seg["text"]
if any(kw in text for kw in DialectConfig.get_combined_keywords()["opening"]):
found_opening = True
break
# 检查结束语（后3段）
for seg in reversed(agent_segments[-3:] if len(agent_segments) >= 3 else agent_segments):
text = seg["text"]
if any(kw in text for kw in DialectConfig.get_combined_keywords()["closing"]):
found_closing = True
break
# 检查禁用词
for seg in agent_segments:
text = seg["text"]
for kw in forbidden_keywords:
if kw in text:
found_forbidden.append(kw)
break
return {
"opening_found": found_opening,
"closing_found": found_closing,
"forbidden_words": ", ".join(set(found_forbidden)) if found_forbidden else "无"
def _check_issue_resolution(self, customer_segments: List[Dict],
agent_segments: List[Dict]) -> bool:
"""检查问题是否解决（增强版）"""
if not customer_segments or not agent_segments:
return False
# 提取所有文本
customer_texts = [seg["text"] for seg in customer_segments]
agent_texts = [seg["text"] for seg in agent_segments]
full_conversation = " ".join(customer_texts + agent_texts)
# 问题解决关键词
resolution_keywords = ["解决", "处理", "完成", "已", "好了", "可以了", "没问题"]
thank_keywords = ["谢谢", "感谢", "多谢"]
negative_keywords = ["没解决", "不行", "不对", "还是", "仍然", "再"]
# 检查是否有负面词汇
has_negative = any(kw in full_conversation for kw in negative_keywords)
if has_negative:
return False
# 检查客户最后是否表达感谢
last_customer_text = customer_segments[-1]["text"]
if any(kw in last_customer_text for kw in thank_keywords):
return True
# 检查是否有解决关键词
if any(kw in full_conversation for kw in resolution_keywords):
return True
# 检查客服是否确认解决
for agent_text in reversed(agent_texts[-3:]): # 检查最后3段
if any(kw in agent_text for kw in resolution_keywords):
return True
return False
def _cleanup_temp_files(self, paths: List[str]):
"""清理临时文件（增强兼容性）"""
def safe_remove(path):
"""安全删除文件（多平台兼容）"""
if os.path.exists(path):
if sys.platform == 'win32':
# Windows系统需要特殊处理
os.chmod(path, 0o777) # 确保有权限
for _ in range(5): # 最多尝试5次
os.remove(path)
break
except PermissionError:
time.sleep(0.2)
else:
os.remove(path)
except Exception:
# 使用线程池并行删除
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
executor.map(safe_remove, paths)
# 额外清理：删除超过1小时的临时文件
now = time.time()
for file in os.listdir(self.temp_dir):
file_path = os.path.join(self.temp_dir, file)
if os.path.isfile(file_path):
file_age = now - os.path.getmtime(file_path)
if file_age > 3600: # 1小时
safe_remove(file_path)
def _format_duration(self, seconds: float) -> str:
"""将秒转换为时分秒格式"""
minutes, seconds = divmod(int(seconds), 60)
hours, minutes = divmod(minutes, 60)
return f"{hours:02d}:{minutes:02d}:{seconds:02d}"
def cleanup_resources(self):
"""显式清理资源"""
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
def stop(self):
"""停止分析"""
self.is_running = False
# ====================== 模型加载线程 ======================
class ModelLoadThread(QThread):
progress_updated = pyqtSignal(int, str)
finished = pyqtSignal(bool, str)
def run(self):
# 检查模型路径
config = ConfigManager().get("model_paths")
if not os.path.exists(config["asr"]):
self.finished.emit(False, "ASR模型路径不存在")
return
if not os.path.exists(config["sentiment"]):
self.finished.emit(False, "情感分析模型路径不存在")
return
self.progress_updated.emit(20, "加载语音识别模型...")
ModelLoader.load_asr_model(config["asr"])
self.progress_updated.emit(60, "加载情感分析模型...")
ModelLoader.load_sentiment_model(config["sentiment"])
self.progress_updated.emit(100, "模型加载完成")
self.finished.emit(True, "模型加载成功。建议：可通过设置界面修改模型路径")
except Exception as e:
self.finished.emit(False, f"模型加载失败: {str(e)}。建议：检查模型路径是否正确，或重新下载模型文件")
# ====================== GUI主界面 ======================
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("贵州方言客服质检系统")
self.setGeometry(100, 100, 1200, 800)
self.setup_ui()
self.setup_menu()
self.analysis_thread = None
self.model_load_thread = None
self.temp_dir = "temp_wav"
os.makedirs(self.temp_dir, exist_ok=True)
self.model_loaded = False
def setup_ui(self):
"""设置用户界面"""
# 主布局
main_widget = QWidget()
main_layout = QVBoxLayout()
main_widget.setLayout(main_layout)
self.setCentralWidget(main_widget)
# 工具栏
toolbar = QToolBar("主工具栏")
toolbar.setIconSize(QSize(24, 24))
self.addToolBar(toolbar)
# 添加文件按钮
add_file_action = QAction(QIcon("icons/add.png"), "添加文件", self)
add_file_action.triggered.connect(self.add_files)
toolbar.addAction(add_file_action)
# 开始分析按钮
analyze_action = QAction(QIcon("icons/start.png"), "开始分析", self)
analyze_action.triggered.connect(self.start_analysis)
toolbar.addAction(analyze_action)
# 停止按钮
stop_action = QAction(QIcon("icons/stop.png"), "停止分析", self)
stop_action.triggered.connect(self.stop_analysis)
toolbar.addAction(stop_action)
# 设置按钮
settings_action = QAction(QIcon("icons/settings.png"), "设置", self)
settings_action.triggered.connect(self.open_settings)
toolbar.addAction(settings_action)
# 分割布局
splitter = QSplitter(Qt.Horizontal)
main_layout.addWidget(splitter)
# 左侧文件列表
left_widget = QWidget()
left_layout = QVBoxLayout()
left_widget.setLayout(left_layout)
file_list_label = QLabel("待分析文件列表")
file_list_label.setFont(QFont("Arial", 12, QFont.Bold))
left_layout.addWidget(file_list_label)
self.file_list = QListWidget()
self.file_list.setSelectionMode(QListWidget.ExtendedSelection)
left_layout.addWidget(self.file_list)
# 右侧结果区域
right_widget = QWidget()
right_layout = QVBoxLayout()
right_widget.setLayout(right_layout)
# 进度条
progress_label = QLabel("分析进度")
progress_label.setFont(QFont("Arial", 12, QFont.Bold))
right_layout.addWidget(progress_label)
self.progress_bar = QProgressBar()
self.progress_bar.setRange(0, 100)
self.progress_bar.setTextVisible(True)
right_layout.addWidget(self.progress_bar)
# 当前文件标签
self.current_file_label = QLabel("当前文件: 无")
right_layout.addWidget(self.current_file_label)
# 结果标签页
self.tab_widget = QTabWidget()
right_layout.addWidget(self.tab_widget, 1)
# 文本结果标签页
text_tab = QWidget()
text_layout = QVBoxLayout()
text_tab.setLayout(text_layout)
self.text_result = QTextEdit()
self.text_result.setReadOnly(True)
text_layout.addWidget(self.text_result)
self.tab_widget.addTab(text_tab, "文本结果")
# 详细结果标签页
detail_tab = QWidget()
detail_layout = QVBoxLayout()
detail_tab.setLayout(detail_layout)
self.result_table = QTableWidget()
self.result_table.setColumnCount(10)
self.result_table.setHorizontalHeaderLabels([
"文件名", "时长", "语速", "音量稳定性",
"客服情感", "客户情感", "开场白", "结束语",
"禁用词", "问题解决"
self.result_table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
detail_layout.addWidget(self.result_table)
self.tab_widget.addTab(detail_tab, "详细结果")
# 添加左右部件到分割器
splitter.addWidget(left_widget)
splitter.addWidget(right_widget)
splitter.setSizes([300, 900])
def setup_menu(self):
"""设置菜单栏"""
menu_bar = self.menuBar()
# 文件菜单
file_menu = menu_bar.addMenu("文件")
add_file_action = QAction("添加文件", self)
add_file_action.triggered.connect(self.add_files)
file_menu.addAction(add_file_action)
export_action = QAction("导出结果", self)
export_action.triggered.connect(self.export_results)
file_menu.addAction(export_action)
exit_action = QAction("退出", self)
exit_action.triggered.connect(self.close)
file_menu.addAction(exit_action)
# 分析菜单
analysis_menu = menu_bar.addMenu("分析")
start_action = QAction("开始分析", self)
start_action.triggered.connect(self.start_analysis)
analysis_menu.addAction(start_action)
stop_action = QAction("停止分析", self)
stop_action.triggered.connect(self.stop_analysis)
analysis_menu.addAction(stop_action)
# 设置菜单
settings_menu = menu_bar.addMenu("设置")
config_action = QAction("系统配置", self)
config_action.triggered.connect(self.open_settings)
settings_menu.addAction(config_action)
model_action = QAction("加载模型", self)
model_action.triggered.connect(self.load_models)
settings_menu.addAction(model_action)
def add_files(self):
"""添加文件到分析列表"""
files, _ = QFileDialog.getOpenFileNames(
self,
"选择音频文件",
"音频文件 (*.mp3 *.wav *.amr *.m4a)"
if files:
for file in files:
self.file_list.addItem(file)
def start_analysis(self):
"""开始分析"""
if self.file_list.count() == 0:
QMessageBox.warning(self, "警告", "请先添加要分析的音频文件")
return
if not self.model_loaded:
QMessageBox.warning(self, "警告", "模型未加载，请先加载模型")
return
# 获取文件路径
audio_paths = [self.file_list.item(i).text() for i in range(self.file_list.count())]
# 清空结果
self.text_result.clear()
self.result_table.setRowCount(0)
# 创建分析线程
self.analysis_thread = AnalysisThread(audio_paths, self.temp_dir)
# 连接信号
self.analysis_thread.progress_updated.connect(self.update_progress)
self.analysis_thread.result_ready.connect(self.handle_result)
self.analysis_thread.finished_all.connect(self.analysis_finished)
self.analysis_thread.error_occurred.connect(self.show_error)
self.analysis_thread.memory_warning.connect(self.handle_memory_warning)
self.analysis_thread.resource_cleanup.connect(self.cleanup_resources)
# 启动线程
self.analysis_thread.start()
def stop_analysis(self):
"""停止分析"""
if self.analysis_thread and self.analysis_thread.isRunning():
self.analysis_thread.stop()
self.analysis_thread.wait()
QMessageBox.information(self, "信息", "分析已停止")
def load_models(self):
"""加载模型"""
if self.model_load_thread and self.model_load_thread.isRunning():
return
self.model_load_thread = ModelLoadThread()
self.model_load_thread.progress_updated.connect(
lambda value, msg: self.progress_bar.setValue(value)
self.model_load_thread.finished.connect(self.handle_model_load_result)
self.model_load_thread.start()
def update_progress(self, progress: int, message: str, current_file: str):
"""更新进度"""
self.progress_bar.setValue(progress)
self.current_file_label.setText(f"当前文件: {current_file}")
def handle_result(self, result: Dict):
"""处理分析结果"""
# 添加到文本结果
self.text_result.append(f"文件: {result['file_name']}")
self.text_result.append(f"状态: {result['status']}")
if result["status"] == "success":
self.text_result.append(f"时长: {result['duration_str']}")
self.text_result.append(f"语速: {result['syllable_rate']} 音节/秒")
self.text_result.append(f"音量稳定性: {result['volume_stability']}")
self.text_result.append(f"客服情感: 负面({result['agent_negative']:.2%}) "
f"中性({result['agent_neutral']:.2%}) "
f"正面({result['agent_positive']:.2%})")
self.text_result.append(f"客服情绪: {result['agent_emotions']}")
self.text_result.append(f"客户情感: 负面({result['customer_negative']:.2%}) "
f"中性({result['customer_neutral']:.2%}) "
f"正面({result['customer_positive']:.2%})")
self.text_result.append(f"客户情绪: {result['customer_emotions']}")
self.text_result.append(f"开场白: {'有' if result['opening_found'] else '无'}")
self.text_result.append(f"结束语: {'有' if result['closing_found'] else '无'}")
self.text_result.append(f"禁用词: {result['forbidden_words']}")
self.text_result.append(f"问题解决: {'是' if result['issue_resolved'] else '否'}")
self.text_result.append("\n=== 对话文本 ===\n")
self.text_result.append(result["asr_text"])
self.text_result.append("\n" + "=" * 50 + "\n")
# 添加到结果表格
row = self.result_table.rowCount()
self.result_table.insertRow(row)
self.result_table.setItem(row, 0, QTableWidgetItem(result["file_name"]))
self.result_table.setItem(row, 1, QTableWidgetItem(result["duration_str"]))
self.result_table.setItem(row, 2, QTableWidgetItem(str(result["syllable_rate"])))
self.result_table.setItem(row, 3, QTableWidgetItem(str(result["volume_stability"])))
self.result_table.setItem(row, 4, QTableWidgetItem(
f"负:{result['agent_negative']:.2f} 中:{result['agent_neutral']:.2f} 正:{result['agent_positive']:.2f}"
self.result_table.setItem(row, 5, QTableWidgetItem(
f"负:{result['customer_negative']:.2f} 中:{result['customer_neutral']:.2f} 正:{result['customer_positive']:.2f}"
self.result_table.setItem(row, 6, QTableWidgetItem("是" if result["opening_found"] else "否"))
self.result_table.setItem(row, 7, QTableWidgetItem("是" if result["closing_found"] else "否"))
self.result_table.setItem(row, 8, QTableWidgetItem(result["forbidden_words"]))
self.result_table.setItem(row, 9, QTableWidgetItem("是" if result["issue_resolved"] else "否"))
# 根据结果着色
if not result["opening_found"]:
self.result_table.item(row, 6).setBackground(QColor(255, 200, 200))
if not result["closing_found"]:
self.result_table.item(row, 7).setBackground(QColor(255, 200, 200))
if result["forbidden_words"] != "无":
self.result_table.item(row, 8).setBackground(QColor(255, 200, 200))
if not result["issue_resolved"]:
self.result_table.item(row, 9).setBackground(QColor(255, 200, 200))
def analysis_finished(self):
"""分析完成"""
QMessageBox.information(self, "完成", "所有音频分析完成")
self.progress_bar.setValue(100)
def show_error(self, title: str, message: str):
"""显示错误信息"""
QMessageBox.critical(self, title, message)
def handle_memory_warning(self):
"""处理内存警告"""
QMessageBox.warning(self, "内存警告", "内存使用过高，分析已停止。请关闭其他应用程序后重试")
def cleanup_resources(self):
"""清理资源"""
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
def handle_model_load_result(self, success: bool, message: str):
"""处理模型加载结果"""
if success:
self.model_loaded = True
QMessageBox.information(self, "成功", message)
else:
QMessageBox.critical(self, "错误", message)
def open_settings(self):
"""打开设置对话框"""
settings_dialog = QDialog(self)
settings_dialog.setWindowTitle("系统设置")
settings_dialog.setFixedSize(500, 400)
layout = QVBoxLayout()
# ASR模型路径
asr_layout = QHBoxLayout()
asr_label = QLabel("ASR模型路径:")
asr_line = QLineEdit(ConfigManager().get("model_paths")["asr"])
asr_browse = QPushButton("浏览...")
def browse_asr():
path = QFileDialog.getExistingDirectory(self, "选择ASR模型目录")
if path:
asr_line.setText(path)
asr_browse.clicked.connect(browse_asr)
asr_layout.addWidget(asr_label)
asr_layout.addWidget(asr_line)
asr_layout.addWidget(asr_browse)
layout.addLayout(asr_layout)
# 情感分析模型路径
sentiment_layout = QHBoxLayout()
sentiment_label = QLabel("情感模型路径:")
sentiment_line = QLineEdit(ConfigManager().get("model_paths")["sentiment"])
sentiment_browse = QPushButton("浏览...")
def browse_sentiment():
path = QFileDialog.getExistingDirectory(self, "选择情感模型目录")
if path:
sentiment_line.setText(path)
sentiment_browse.clicked.connect(browse_sentiment)
sentiment_layout.addWidget(sentiment_label)
sentiment_layout.addWidget(sentiment_line)
sentiment_layout.addWidget(sentiment_browse)
layout.addLayout(sentiment_layout)
# 并发设置
concurrent_layout = QHBoxLayout()
concurrent_label = QLabel("最大并发任务:")
concurrent_spin = QSpinBox()
concurrent_spin.setRange(1, 8)
concurrent_spin.setValue(ConfigManager().get("max_concurrent", 1))
concurrent_layout.addWidget(concurrent_label)
concurrent_layout.addWidget(concurrent_spin)
layout.addLayout(concurrent_layout)
# 方言设置
dialect_layout = QHBoxLayout()
dialect_label = QLabel("方言设置:")
dialect_combo = QComboBox()
dialect_combo.addItems(["标准普通话", "贵州方言"])
dialect_combo.setCurrentIndex(1 if ConfigManager().get("dialect_config") == "guizhou" else 0)
dialect_layout.addWidget(dialect_label)
dialect_layout.addWidget(dialect_combo)
layout.addLayout(dialect_layout)
# 音频时长限制
duration_layout = QHBoxLayout()
duration_label = QLabel("最大音频时长(秒):")
duration_spin = QSpinBox()
duration_spin.setRange(60, 86400) # 1分钟到24小时
duration_spin.setValue(ConfigManager().get("max_audio_duration", 3600))
duration_layout.addWidget(duration_label)
duration_layout.addWidget(duration_spin)
layout.addLayout(duration_layout)
button_box = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
button_box.accepted.connect(settings_dialog.accept)
button_box.rejected.connect(settings_dialog.reject)
layout.addWidget(button_box)
settings_dialog.setLayout(layout)
if settings_dialog.exec_() == QDialog.Accepted:
# 保存设置
ConfigManager().set("model_paths", {
"asr": asr_line.text(),
"sentiment": sentiment_line.text()
ConfigManager().set("max_concurrent", concurrent_spin.value())
ConfigManager().set("dialect_config", "guizhou" if dialect_combo.currentIndex() == 1 else "standard")
ConfigManager().set("max_audio_duration", duration_spin.value())
# 重新加载模型
ModelLoader.reload_models()
def export_results(self):
"""导出结果"""
if self.result_table.rowCount() == 0:
QMessageBox.warning(self, "警告", "没有可导出的结果")
return
path, _ = QFileDialog.getSaveFileName(
self,
"保存结果",
"CSV文件 (*.csv)"
if path:
with open(path, "w", encoding="utf-8") as f:
# 写入表头
headers = []
for col in range(self.result_table.columnCount()):
headers.append(self.result_table.horizontalHeaderItem(col).text())
f.write(",".join(headers) + "\n")
# 写入数据
for row in range(self.result_table.rowCount()):
row_data = []
for col in range(self.result_table.columnCount()):
item = self.result_table.item(row, col)
row_data.append(item.text() if item else "")
f.write(",".join(row_data) + "\n")
QMessageBox.information(self, "成功", f"结果已导出到: {path}")
except Exception as e:
QMessageBox.critical(self, "错误", f"导出失败: {str(e)}")
def closeEvent(self, event):
"""关闭事件处理"""
if self.analysis_thread and self.analysis_thread.isRunning():
self.analysis_thread.stop()
self.analysis_thread.wait()
# 清理临时目录（增强兼容性）
for file in os.listdir(self.temp_dir):
file_path = os.path.join(self.temp_dir, file)
if os.path.isfile(file_path):
# Windows系统可能需要多次尝试
for _ in range(3):
os.remove(file_path)
break
except PermissionError:
time.sleep(0.1)
os.rmdir(self.temp_dir)
except:
event.accept()
# ====================== 程序入口 ======================
if __name__ == "__main__":
torch.set_num_threads(4) # 限制CPU线程数
app = QApplication(sys.argv)
# 设置应用样式
app.setStyle('Fusion')
window = MainWindow()
window.show()
sys.exit(app.exec_())
之后，系统报错“Import numpy could not be resolved”，原因可能有两个
未下载此包，打开命令行，输入 pip list，可以看到你下载过的所有包，如果未下载，则下载后重启vscode就可以了。
你有多个python的编译环境，而你在vscode里使用的那个编译环境中没有下载该包。
解决办法：
键盘上按快捷键：Ctrl + shift + P
输入：Python:Select Interpret
推荐文章
|
|
眼睛小的木耳 · 立法意见征集_河北省司法厅 6 月前 |
|
|
乐观的签字笔 · 全部成交？洛克菲勒藏品拍卖创22项世界纪录-新华网 6 月前 |