活动介绍

pd.json_normalize函数

时间: 2023-08-04 20:07:17 浏览: 346
pd.json_normalize函数是Pandas库中的一个函数，它可以将嵌套的JSON数据规范化为平面（flattened）的表格形式，使得数据更易于处理和分析。它的语法如下： ```python pd.json_normalize(data, record_path=None, meta=None, meta_prefix=None, record_prefix=None, errors='raise', sep='.') ``` 其中，参数data是需要规范化的JSON数据，record_path是指定嵌套JSON数据的路径，meta是额外的列名，meta_prefix和record_prefix是对应列名的前缀，errors是处理JSON解析错误的方式，sep是指定列名的分隔符。通过调用pd.json_normalize函数，可以将嵌套的JSON数据转换为Pandas的DataFrame对象，方便进行数据分析和处理。
相关问题

pd.json_normalize（）用法

`pd.json_normalize()` 是 Pandas 库中的一个函数，用于将 JSON 数据规范化为 Pandas 数据帧。 该函数主要有以下参数： - `data`：要规范化的 JSON 数据。 - `record_path`：规范化的 JSON 数据路径。 - `meta`：要添加到结果数据帧的元数据。 - `meta_prefix`：元数据列名的前缀。 - `record_prefix`：记录列名的前缀。 - `errors`：处理错误的方法。 以下是一个简单的示例： ```python import pandas as pd import json data = [ {"name": "John", "age": 30, "city": "New York"}, {"name": "Mike", "age": 25, "city": "Los Angeles"}, {"name": "Sarah", "age": 32, "city": "Chicago"} ] json_data = json.dumps(data) df = pd.json_normalize(json.loads(json_data)) print(df) ``` 输出结果： ``` name age city 0 John 30 New York 1 Mike 25 Los Angeles 2 Sarah 32 Chicago ``` 在该示例中，我们首先将 Python 列表转换为 JSON 字符串，然后使用 `json.loads()` 函数将其转换回 Python 字典。接下来，我们使用 `pd.json_normalize()` 函数将字典转换为 Pandas 数据帧，并将其打印出来。

pandas.json_normalize

pandas的json_normalize函数是一个非常实用的工具，它用于将JSON格式的数据转换成更标准、更容易处理的DataFrame结构。当面对嵌套的JSON数据，例如一个包含多个层级键值对的对象列表，这个函数能够扁平化这些数据，使得每一项的属性成为一个独立的列。 它接受一个JSON对象、字典列表或包含JSON内容的Pandas Series作为输入，并返回一个新的DataFrame。这个新的DataFrame通常会有一个名为“columns”的列，包含了原始JSON对象的所有路径，其他列则是沿着这些路径展开后的值。 例如，如果你有如下的JSON数据： ```json data = [ { "id": 1, "name": "Alice", "address": {"city": "New York", "country": "USA"} }, { "id": 2, "name": "Bob", "address": {"city": "London", "country": "UK"} } ] ``` 你可以通过`pd.json_normalize(data)`来将其转换为： ``` id name city country 0 1 Alice New York USA 1 2 Bob London UK ```
阅读全文

相关推荐

.zip

前端开源库-geojson-normalize

前端开源库-geojson-normalizegeojson规范化，将不同的geojson表单规范化为FeatureCollections

df=pd.json_normalize(response_json['data']['list'],errors='ignore')。如何将这句代码中的.json_normalize改为使用json.load来实现这句代码的功能

pd.json_normalize 是 pandas 的一个函数，用于从 JSON 格式的数据结构中提取嵌套数据并将其转换为 DataFrame。这个函数方便地展开 JSON 对象，使其每一级键都变成 DataFrame 的一列。 如果要用 json.load 替代...

“使用的评论列: 评论详情（用户：内容【点赞数】） Traceback (most recent call last): File "C:\Users\LENOVO\AppData\Local\Programs\Python\Python38\lib\site-packages\pandas\core\indexes\base.py", line 3653, in get_loc return self._engine.get_loc(casted_key) File "pandas\_libs\index.pyx", line 147, in pandas._libs.index.IndexEngine.get_loc File "pandas\_libs\index.pyx", line 176, in pandas._libs.index.IndexEngine.get_loc File "pandas\_libs\hashtable_class_helper.pxi", line 7080, in pandas._libs.hashtable.PyObjectHashTable.get_item File "pandas\_libs\hashtable_class_helper.pxi", line 7088, in pandas._libs.hashtable.PyObjectHashTable.get_item KeyError: ' 解析后评论' The above exception was the direct cause of the following exception: Traceback (most recent call last): File "C:/Users/LENOVO/Desktop/1835.py", line 41, in <module> df = df.join(pd.json_normalize(df[' 解析后评论'])) File "C:\Users\LENOVO\AppData\Local\Programs\Python\Python38\lib\site-packages\pandas\core\frame.py", line 3761, in __getitem__ indexer = self.columns.get_loc(key) File "C:\Users\LENOVO\AppData\Local\Programs\Python\Python38\lib\site-packages\pandas\core\indexes\base.py", line 3655, in get_loc raise KeyError(key) from err KeyError: ' 解析后评论'”这是什么问题？应该如何解决

- **确保列存在**：在调用pd.json_normalize之前，确保df中确实存在该列。例如，检查之前的代码是否正确生成了该列，或者是否有条件导致该列未被创建。 另外，用户可能在处理JSON数据时，解析后的结果可能为空或者...

ImportError: cannot import name 'json_normalize' from 'pandas.io.json' (D:\Python\PY\lib\site-packages\pandas\io\json\__init__.py)

print(pd.__version__) 如果没有安装pandas或者版本较低，可以使用以下命令安装最新版本： pip install pandas --upgrade 2. 检查是否缺少必要的依赖项。有时候，pandas需要其他库的支持才能正常...

json_normalize

json_normalize是pandas库中的一个函数，用于将嵌套的JSON数据转换为平面表格形式。它可以将JSON数据中的每个嵌套层级展开为一列，并将每个嵌套层级的键作为列名。这个函数可以方便地将JSON数据转换为pandas ...

pd.read_json(data)报错，arrays must all be same length，如何修改

当 pd.read_json 函数报错 "arrays must all be same length" 时，通常表示你在尝试将不同长度的数组合并成一个 DataFrame。这是因为 JSON 数据在转换为 DataFrame 时，每一行的所有列必须有相同的长度才能组成...

检查并优化代码： import sys import os import json import time import wave import numpy as np import pandas as pd import matplotlib.pyplot as plt import soundfile as sf from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel, QLineEdit, QTextEdit, QFileDialog, QProgressBar, QGroupBox, QComboBox, QCheckBox) from PyQt5.QtCore import QThread, pyqtSignal from pydub import AudioSegment from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification import whisper from pyannote.audio import Pipeline from docx import Document from docx.shared import Inches import librosa import tempfile from collections import defaultdict import re class AnalysisThread(QThread): progress = pyqtSignal(int) message = pyqtSignal(str) analysis_complete = pyqtSignal(dict) error = pyqtSignal(str) def __init__(self, audio_files, keyword_file, whisper_model_path, pyannote_model_path, emotion_model_path): super().__init__() self.audio_files = audio_files self.keyword_file = keyword_file self.whisper_model_path = whisper_model_path self.pyannote_model_path = pyannote_model_path self.emotion_model_path = emotion_model_path self.running = True self.cached_models = {} def run(self): try: # 加载关键词 self.message.emit("正在加载关键词...") keywords = self.load_keywords() # 预加载模型 self.message.emit("正在预加载模型...") self.preload_models() results = [] total_files = len(self.audio_files) for idx, audio_file in enumerate(self.audio_files): if not self.running: self.message.emit("分析已停止") return self.message.emit(f"正在处理文件: {os.path.basename(audio_file)} ({idx + 1}/{total_files})") file_result = self.analyze_file(audio_file, keywords) if file_result: results.append(file_result) self.progress.emit(int((idx + 1) / total_files * 100)) self.analysis_complete.emit({"results": results, "keywords": keywords}) self.message.emit("分析完成！") except Exception as e: import traceback self.error.emit(f"分析过程中发生错误: {str(e)}\n{traceback.format_exc()}") def preload_models(self): """预加载所有模型到缓存""" # 检查是否已加载模型 if hasattr(self, 'cached_models') and self.cached_models: return self.cached_models = {} # 加载语音识别模型 if 'whisper' not in self.cached_models: self.message.emit("正在加载语音识别模型...") self.cached_models['whisper'] = whisper.load_model(self.whisper_model_path) # 加载说话人分离模型 if 'pyannote' not in self.cached_models: self.message.emit("正在加载说话人分离模型...") self.cached_models['pyannote'] = Pipeline.from_pretrained(self.pyannote_model_path) # 加载情感分析模型 if 'emotion_classifier' not in self.cached_models: self.message.emit("正在加载情感分析模型...") tokenizer = AutoTokenizer.from_pretrained(self.emotion_model_path) model = AutoModelForSequenceClassification.from_pretrained(self.emotion_model_path) self.cached_models['emotion_classifier'] = pipeline( "text-classification", model=model, tokenizer=tokenizer, device=0 if torch.cuda.is_available() else -1 # 使用GPU如果可用 ) def analyze_file(self, audio_file, keywords): """分析单个音频文件""" try: # 确保音频为WAV格式 wav_file = self.convert_to_wav(audio_file) # 获取音频信息 duration, sample_rate, channels = self.get_audio_info(wav_file) # 说话人分离 diarization = self.cached_models['pyannote'](wav_file) # 识别客服和客户（使用改进的方法） agent_segments, customer_segments = self.identify_speakers(wav_file, diarization, keywords['opening']) # 语音识别（使用优化后的方法） whisper_model = self.cached_models['whisper'] agent_text = self.transcribe_audio(wav_file, agent_segments, whisper_model) customer_text = self.transcribe_audio(wav_file, customer_segments, whisper_model) # 情感分析 emotion_classifier = self.cached_models['emotion_classifier'] agent_emotion = self.analyze_emotion(agent_text, emotion_classifier) customer_emotion = self.analyze_emotion(customer_text, emotion_classifier) # 服务规范检查 opening_check = self.check_opening(agent_text, keywords['opening']) closing_check = self.check_closing(agent_text, keywords['closing']) forbidden_check = self.check_forbidden(agent_text, keywords['forbidden']) # 沟通技巧分析（使用改进的方法） speech_rate = self.analyze_speech_rate(wav_file, agent_segments) volume_analysis = self.analyze_volume(wav_file, agent_segments) # 问题解决率分析 resolution_rate = self.analyze_resolution(agent_text, customer_text, keywords['resolution']) # 构建结果 return { "file_name": os.path.basename(audio_file), "duration": duration, "agent_text": agent_text, "customer_text": customer_text, "opening_check": opening_check, "closing_check": closing_check, "forbidden_check": forbidden_check, "agent_emotion": agent_emotion, "customer_emotion": customer_emotion, "speech_rate": speech_rate, "volume_mean": volume_analysis['mean'], "volume_std": volume_analysis['std'], "resolution_rate": resolution_rate } except Exception as e: self.error.emit(f"处理文件 {os.path.basename(audio_file)} 时出错: {str(e)}") return None def load_keywords(self): """从Excel文件加载关键词""" try: df = pd.read_excel(self.keyword_file) keywords = { "opening": [str(k).strip() for k in df['opening'].dropna().tolist()], "closing": [str(k).strip() for k in df['closing'].dropna().tolist()], "forbidden": [str(k).strip() for k in df['forbidden'].dropna().tolist()], "resolution": [str(k).strip() for k in df['resolution'].dropna().tolist()] } return keywords except Exception as e: raise Exception(f"加载关键词文件失败: {str(e)}") def convert_to_wav(self, audio_file): """将音频文件转换为WAV格式（如果需要）""" try: if not audio_file.lower().endswith('.wav'): # 使用临时文件避免磁盘IO with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as tmpfile: output_file = tmpfile.name audio = AudioSegment.from_file(audio_file) audio.export(output_file, format='wav') return output_file return audio_file except Exception as e: raise Exception(f"音频转换失败: {str(e)}") def get_audio_info(self, wav_file): """获取音频文件信息""" try: with wave.open(wav_file, 'rb') as wf: frames = wf.getnframes() rate = wf.getframerate() channels = wf.getnchannels() duration = frames / float(rate) return duration, rate, channels except Exception as e: raise Exception(f"获取音频信息失败: {str(e)}") def identify_speakers(self, wav_file, diarization, opening_keywords): """改进的客服识别方法 - 检查前三个片段是否有开场白关键词""" speaker_segments = defaultdict(list) for segment, _, speaker in diarization.itertracks(yield_label=True): speaker_segments[speaker].append((segment.start, segment.end)) # 如果没有说话人 if not speaker_segments: return [], [] # 如果只有一个说话人 if len(speaker_segments) == 1: speaker = list(speaker_segments.keys())[0] return speaker_segments[speaker], [] # 检查每个说话人的前三个片段是否有开场白 speaker_scores = {} whisper_model = self.cached_models['whisper'] for speaker, segments in speaker_segments.items(): score = 0 # 取前三个片段（或所有片段如果少于3个） check_segments = segments[:3] for start, end in check_segments: # 转录片段 text = self.transcribe_audio_segment(wav_file, [(start, end)], whisper_model) # 检查开场白关键词 for keyword in opening_keywords: if keyword and keyword in text: score += 1 break speaker_scores[speaker] = score # 找到得分最高的说话人作为客服 agent_speaker = max(speaker_scores, key=speaker_scores.get) agent_segments = [] customer_segments = [] for speaker, segments in speaker_segments.items(): if speaker == agent_speaker: agent_segments = segments else: customer_segments.extend(segments) return agent_segments, customer_segments def transcribe_audio_segment(self, wav_file, segments, model): """转录单个音频片段 - 用于客服识别""" if not segments: return "" # 使用pydub加载音频 audio = AudioSegment.from_wav(wav_file) start, end = segments[0] # 转换为毫秒 start_ms = int(start * 1000) end_ms = int(end * 1000) segment_audio = audio[start_ms:end_ms] # 使用临时文件 with tempfile.NamedTemporaryFile(suffix='.wav') as tmpfile: segment_audio.export(tmpfile.name, format="wav") result = model.transcribe(tmpfile.name) return result['text'] def transcribe_audio(self, wav_file, segments, model): """优化后的转录方法 - 按片段转录""" if not segments: return "" # 使用pydub加载音频 audio = AudioSegment.from_wav(wav_file) full_text = "" # 只处理指定片段 for start, end in segments: # 转换为毫秒 start_ms = int(start * 1000) end_ms = int(end * 1000) segment_audio = audio[start_ms:end_ms] # 使用临时文件避免内存占用 with tempfile.NamedTemporaryFile(suffix='.wav') as tmpfile: segment_audio.export(tmpfile.name, format="wav") result = model.transcribe(tmpfile.name) full_text += result['text'] + " " return full_text.strip() def analyze_emotion(self, text, classifier): """分析文本情感""" if not text.strip(): return {"label": "中性", "score": 0.0} # 截断长文本以提高性能 if len(text) > 500: text = text[:500] result = classifier(text, truncation=True, max_length=512) return { "label": result[0]['label'], "score": result[0]['score'] } def check_opening(self, text, opening_keywords): """检查开场白""" return any(keyword in text for keyword in opening_keywords if keyword) def check_closing(self, text, closing_keywords): """检查结束语""" return any(keyword in text for keyword in closing_keywords if keyword) def check_forbidden(self, text, forbidden_keywords): """检查服务禁语""" return any(keyword in text for keyword in forbidden_keywords if keyword) def analyze_speech_rate(self, wav_file, segments): """改进的语速分析 - 基于实际识别文本""" if not segments: return 0 # 加载音频 y, sr = librosa.load(wav_file, sr=None) total_chars = 0 total_duration = 0 whisper_model = self.cached_models['whisper'] for start, end in segments: # 计算片段时长（秒） duration = end - start total_duration += duration # 转录片段 text = self.transcribe_audio_segment(wav_file, [(start, end)], whisper_model) # 计算中文字符数（去除标点和空格） chinese_chars = sum(1 for char in text if '\u4e00' <= char <= '\u9fff') total_chars += chinese_chars if total_duration == 0: return 0 # 语速 = 总字数 / 总时长(分钟) return total_chars / (total_duration / 60) def analyze_volume(self, wav_file, segments): """改进的音量分析 - 使用librosa计算RMS分贝值""" if not segments: return {"mean": -60, "std": 0} # 加载音频 y, sr = librosa.load(wav_file, sr=None) all_dB = [] for start, end in segments: start_sample = int(start * sr) end_sample = int(end * sr) segment_audio = y[start_sample:end_sample] # 计算RMS并转换为dB rms = librosa.feature.rms(y=segment_audio)[0] dB = librosa.amplitude_to_db(rms, ref=np.max) all_dB.extend(dB) if not all_dB: return {"mean": -60, "std": 0} return { "mean": float(np.mean(all_dB)), "std": float(np.std(all_dB)) } def analyze_resolution(self, agent_text, customer_text, resolution_keywords): """分析问题解决率""" return any(keyword in agent_text for keyword in resolution_keywords if keyword) def stop(self): """停止分析""" self.running = False class MainWindow(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle("外呼电话录音包质检分析系统") self.setGeometry(100, 100, 1000, 700) # 初始化变量 self.audio_files = [] self.keyword_file = "" self.whisper_model_path = "./models/whisper-small" self.pyannote_model_path = "./models/pyannote-speaker-diarization" self.emotion_model_path = "./models/Erlangshen-Roberta-110M-Sentiment" self.output_dir = "./reports" # 创建主控件 central_widget = QWidget() self.setCentralWidget(central_widget) main_layout = QVBoxLayout(central_widget) # 文件选择区域 file_group = QGroupBox("文件选择") file_layout = QVBoxLayout(file_group) # 音频文件选择 audio_layout = QHBoxLayout() self.audio_label = QLabel("音频文件/文件夹:") audio_layout.addWidget(self.audio_label) self.audio_path_edit = QLineEdit() audio_layout.addWidget(self.audio_path_edit) self.audio_browse_btn = QPushButton("浏览...") self.audio_browse_btn.clicked.connect(self.browse_audio) audio_layout.addWidget(self.audio_browse_btn) file_layout.addLayout(audio_layout) # 关键词文件选择 keyword_layout = QHBoxLayout() self.keyword_label = QLabel("关键词文件:") keyword_layout.addWidget(self.keyword_label) self.keyword_path_edit = QLineEdit() keyword_layout.addWidget(self.keyword_path_edit) self.keyword_browse_btn = QPushButton("浏览...") self.keyword_browse_btn.clicked.connect(self.browse_keyword) keyword_layout.addWidget(self.keyword_browse_btn) file_layout.addLayout(keyword_layout) main_layout.addWidget(file_group) # 模型设置区域 model_group = QGroupBox("模型设置") model_layout = QVBoxLayout(model_group) # Whisper模型路径 whisper_layout = QHBoxLayout() whisper_layout.addWidget(QLabel("Whisper模型路径:")) self.whisper_edit = QLineEdit(self.whisper_model_path) whisper_layout.addWidget(self.whisper_edit) model_layout.addLayout(whisper_layout) # Pyannote模型路径 pyannote_layout = QHBoxLayout() pyannote_layout.addWidget(QLabel("Pyannote模型路径:")) self.pyannote_edit = QLineEdit(self.pyannote_model_path) pyannote_layout.addWidget(self.pyannote_edit) model_layout.addLayout(pyannote_layout) # 情感分析模型路径 emotion_layout = QHBoxLayout() emotion_layout.addWidget(QLabel("情感分析模型路径:")) self.emotion_edit = QLineEdit(self.emotion_model_path) emotion_layout.addWidget(self.emotion_edit) model_layout.addLayout(emotion_layout) # 输出目录 output_layout = QHBoxLayout() output_layout.addWidget(QLabel("输出目录:")) self.output_edit = QLineEdit(self.output_dir) output_layout.addWidget(self.output_edit) self.output_browse_btn = QPushButton("浏览...") self.output_browse_btn.clicked.connect(self.browse_output) output_layout.addWidget(self.output_browse_btn) model_layout.addLayout(output_layout) main_layout.addWidget(model_group) # 控制按钮区域 control_layout = QHBoxLayout() self.start_btn = QPushButton("开始分析") self.start_btn.clicked.connect(self.start_analysis) control_layout.addWidget(self.start_btn) self.stop_btn = QPushButton("停止分析") self.stop_btn.clicked.connect(self.stop_analysis) self.stop_btn.setEnabled(False) control_layout.addWidget(self.stop_btn) self.clear_btn = QPushButton("清空") self.clear_btn.clicked.connect(self.clear_all) control_layout.addWidget(self.clear_btn) main_layout.addLayout(control_layout) # 进度条 self.progress_bar = QProgressBar() self.progress_bar.setValue(0) main_layout.addWidget(self.progress_bar) # 日志输出区域 log_group = QGroupBox("分析日志") log_layout = QVBoxLayout(log_group) self.log_text = QTextEdit() self.log_text.setReadOnly(True) log_layout.addWidget(self.log_text) main_layout.addWidget(log_group) # 状态区域 status_layout = QHBoxLayout() self.status_label = QLabel("就绪") status_layout.addWidget(self.status_label) self.file_count_label = QLabel("已选择0个音频文件") status_layout.addWidget(self.file_count_label) main_layout.addLayout(status_layout) # 初始化分析线程 self.analysis_thread = None def browse_audio(self): """浏览音频文件或文件夹""" options = QFileDialog.Options() files, _ = QFileDialog.getOpenFileNames( self, "选择音频文件", "", "音频文件 (*.mp3 *.wav *.amr *.ogg *.flac);;所有文件 (*)", options=options ) if files: self.audio_files = files self.audio_path_edit.setText("; ".join(files)) self.file_count_label.setText(f"已选择{len(files)}个音频文件") self.log_text.append(f"已选择{len(files)}个音频文件") def browse_keyword(self): """浏览关键词文件""" options = QFileDialog.Options() file, _ = QFileDialog.getOpenFileName( self, "选择关键词文件", "", "Excel文件 (*.xlsx *.xls);;所有文件 (*)", options=options ) if file: self.keyword_file = file self.keyword_path_edit.setText(file) self.log_text.append(f"已选择关键词文件: {file}") def browse_output(self): """浏览输出目录""" options = QFileDialog.Options() directory = QFileDialog.getExistingDirectory( self, "选择输出目录", "", options=options ) if directory: self.output_dir = directory self.output_edit.setText(directory) self.log_text.append(f"输出目录设置为: {directory}") def start_analysis(self): """开始分析""" if not self.audio_files: self.log_text.append("错误: 请先选择音频文件") return if not self.keyword_file: self.log_text.append("错误: 请先选择关键词文件") return # 更新模型路径 self.whisper_model_path = self.whisper_edit.text() self.pyannote_model_path = self.pyannote_edit.text() self.emotion_model_path = self.emotion_edit.text() self.output_dir = self.output_edit.text() # 创建输出目录 os.makedirs(self.output_dir, exist_ok=True) self.log_text.append("开始分析...") self.start_btn.setEnabled(False) self.stop_btn.setEnabled(True) self.status_label.setText("分析中...") self.progress_bar.setValue(0) # 创建并启动分析线程 self.analysis_thread = AnalysisThread( self.audio_files, self.keyword_file, self.whisper_model_path, self.pyannote_model_path, self.emotion_model_path ) self.analysis_thread.progress.connect(self.progress_bar.setValue) self.analysis_thread.message.connect(self.log_text.append) self.analysis_thread.analysis_complete.connect(self.on_analysis_complete) self.analysis_thread.error.connect(self.on_analysis_error) self.analysis_thread.finished.connect(self.on_analysis_finished) self.analysis_thread.start() def stop_analysis(self): """停止分析""" if self.analysis_thread and self.analysis_thread.isRunning(): self.analysis_thread.stop() self.log_text.append("正在停止分析...") self.stop_btn.setEnabled(False) def clear_all(self): """清空所有内容""" self.audio_files = [] self.keyword_file = "" self.audio_path_edit.clear() self.keyword_path_edit.clear() self.log_text.clear() self.progress_bar.setValue(0) self.status_label.setText("就绪") self.file_count_label.setText("已选择0个音频文件") self.log_text.append("已清空所有内容") def on_analysis_complete(self, result): """分析完成处理""" try: self.log_text.append("正在生成报告...") if not result.get("results"): self.log_text.append("警告: 没有生成任何分析结果") return # 生成Excel报告 excel_path = os.path.join(self.output_dir, "质检分析报告.xlsx") self.generate_excel_report(result, excel_path) # 生成Word报告 word_path = os.path.join(self.output_dir, "质检分析报告.docx") self.generate_word_report(result, word_path) self.log_text.append(f"分析报告已保存至: {excel_path}") self.log_text.append(f"可视化报告已保存至: {word_path}") self.log_text.append("分析完成！") self.status_label.setText(f"分析完成！报告保存至: {self.output_dir}") except Exception as e: import traceback self.log_text.append(f"生成报告时出错: {str(e)}\n{traceback.format_exc()}") def on_analysis_error(self, message): """分析错误处理""" self.log_text.append(f"错误: {message}") self.status_label.setText("发生错误") def on_analysis_finished(self): """分析线程结束处理""" self.start_btn.setEnabled(True) self.stop_btn.setEnabled(False) def generate_excel_report(self, result, output_path): """生成Excel报告""" # 从结果中提取数据 data = [] for res in result['results']: data.append({ "文件名": res['file_name'], "音频时长(秒)": res['duration'], "开场白检查": "通过" if res['opening_check'] else "未通过", "结束语检查": "通过" if res['closing_check'] else "未通过", "服务禁语检查": "通过" if not res['forbidden_check'] else "未通过", "客服情感": res['agent_emotion']['label'], "客服情感得分": res['agent_emotion']['score'], "客户情感": res['customer_emotion']['label'], "客户情感得分": res['customer_emotion']['score'], "语速(字/分)": res['speech_rate'], "平均音量(dB)": res['volume_mean'], "音量标准差": res['volume_std'], "问题解决率": "是" if res['resolution_rate'] else "否" }) # 创建DataFrame并保存 df = pd.DataFrame(data) df.to_excel(output_path, index=False) # 添加汇总统计 try: with pd.ExcelWriter(output_path, engine='openpyxl', mode='a', if_sheet_exists='replace') as writer: summary_data = { "统计项": ["总文件数", "开场白通过率", "结束语通过率", "服务禁语通过率", "问题解决率"], "数值": [ len(result['results']), df['开场白检查'].value_counts().get('通过', 0) / len(df), df['结束语检查'].value_counts().get('通过', 0) / len(df), df['服务禁语检查'].value_counts().get('通过', 0) / len(df), df['问题解决率'].value_counts().get('是', 0) / len(df) ] } summary_df = pd.DataFrame(summary_data) summary_df.to_excel(writer, sheet_name='汇总统计', index=False) except Exception as e: self.log_text.append(f"添加汇总统计时出错: {str(e)}") def generate_word_report(self, result, output_path): """生成Word报告""" doc = Document() # 添加标题 doc.add_heading('外呼电话录音质检分析报告', 0) # 添加基本信息 doc.add_heading('分析概况', level=1) doc.add_paragraph(f"分析时间: {time.strftime('%Y-%m-%d %H:%M:%S')}") doc.add_paragraph(f"分析文件数量: {len(result['results'])}") doc.add_paragraph(f"关键词文件: {os.path.basename(self.keyword_file)}") # 添加汇总统计 doc.add_heading('汇总统计', level=1) # 创建汇总表格 table = doc.add_table(rows=5, cols=2) table.style = 'Table Grid' # 表头 hdr_cells = table.rows[0].cells hdr_cells[0].text = '统计项' hdr_cells[1].text = '数值' # 计算统计数据 df = pd.DataFrame(result['results']) pass_rates = { "开场白通过率": df['opening_check'].mean() if not df.empty else 0, "结束语通过率": df['closing_check'].mean() if not df.empty else 0, "服务禁语通过率": (1 - df['forbidden_check']).mean() if not df.empty else 0, "问题解决率": df['resolution_rate'].mean() if not df.empty else 0 } # 填充表格 rows = [ ("总文件数", len(result['results'])), ("开场白通过率", f"{pass_rates['开场白通过率']:.2%}"), ("结束语通过率", f"{pass_rates['结束语通过率']:.2%}"), ("服务禁语通过率", f"{pass_rates['服务禁语通过率']:.2%}"), ("问题解决率", f"{pass_rates['问题解决率']:.2%}") ] for i, row_data in enumerate(rows): if i < len(table.rows): row_cells = table.rows[i].cells row_cells[0].text = row_data[0] row_cells[1].text = str(row_data[1]) # 添加情感分析图表 if result['results']: doc.add_heading('情感分析', level=1) # 客服情感分布 agent_emotions = [res['agent_emotion']['label'] for res in result['results']] agent_emotion_counts = pd.Series(agent_emotions).value_counts() if not agent_emotion_counts.empty: fig, ax = plt.subplots(figsize=(6, 4)) agent_emotion_counts.plot.pie(autopct='%1.1f%%', ax=ax) ax.set_title('客服情感分布') plt.tight_layout() # 保存图表到临时文件 chart_path = os.path.join(self.output_dir, "agent_emotion_chart.png") plt.savefig(chart_path, dpi=100) plt.close() doc.add_picture(chart_path, width=Inches(4)) doc.add_paragraph('图1: 客服情感分布') # 客户情感分布 customer_emotions = [res['customer_emotion']['label'] for res in result['results']] customer_emotion_counts = pd.Series(customer_emotions).value_counts() if not customer_emotion_counts.empty: fig, ax = plt.subplots(figsize=(6, 4)) customer_emotion_counts.plot.pie(autopct='%1.1f%%', ax=ax) ax.set_title('客户情感分布') plt.tight_layout() chart_path = os.path.join(self.output_dir, "customer_emotion_chart.png") plt.savefig(chart_path, dpi=100) plt.close() doc.add_picture(chart_path, width=Inches(4)) doc.add_paragraph('图2: 客户情感分布') # 添加详细分析结果 doc.add_heading('详细分析结果', level=1) # 创建详细表格 table = doc.add_table(rows=1, cols=6) table.style = 'Table Grid' # 表头 hdr_cells = table.rows[0].cells headers = ['文件名', '开场白', '结束语', '禁语', '客服情感', '问题解决'] for i, header in enumerate(headers): hdr_cells[i].text = header # 填充数据 for res in result['results']: row_cells = table.add_row().cells row_cells[0].text = res['file_name'] row_cells[1].text = "✓" if res['opening_check'] else "✗" row_cells[2].text = "✓" if res['closing_check'] else "✗" row_cells[3].text = "✗" if res['forbidden_check'] else "✓" row_cells[4].text = res['agent_emotion']['label'] row_cells[5].text = "✓" if res['resolution_rate'] else "✗" # 保存文档 doc.save(output_path) if __name__ == "__main__": # 检查是否安装了torch try: import torch except ImportError: print("警告: PyTorch 未安装，情感分析可能无法使用GPU加速") app = QApplication(sys.argv) window = MainWindow() window.show() sys.exit(app.exec_())

其次，在transcribe_audio_segment和transcribe_audio方法中，也创建了临时文件，这些临时文件在函数返回后就可以删除（因为已经转录完成）。我们可以使用delete=True（默认）来让临时文件在使用后自动删除。...

json_normalize 筛选输出列

json_normalize 是 pandas 库中的一个函数，用于将 JSON 格式的数据规范化成表格形式，通常是从字典列表中提取嵌套的数据结构。当你想要筛选输出列时，你可以传递一个 columns 参数给 json_normalize，这个...

json_normalize只输出部分列

json_normalize 是 pandas 中的一个函数，用于将 JSON 格式的数据转换成更易分析的表格形式。如果你只想输出JSON对象的部分列，你可以通过指定 record_path 参数来选择要包含的嵌套层级。 例如，假设你有一个 ...

json_normalize处理列中数据

df_courses = pd.json_normalize( data, record_path=['students', 'courses'], # 指定嵌套记录的位置 meta=[ ['students', 'name'], ['students', 'age'] ], # 添加额外元数据到每一行 errors='ignore' # ...

json_normalize 只取部分列

json_normalize 是 pandas 库中的一个函数，用于将 JSON 对象转换为 DataFrame 格式，通常是从字典嵌套结构中提取数据。如果你只想从 JSON 数据中取特定的列，可以在调用 json_normalize 时指定需要的键路径...

根据修改建议进行修改： 1、AnalysisThread.add_charts 中生成图表后虽删除临时文件，但未显式释放图表对象，可能导致内存泄漏（尤其批量处理时）。使用 plt.close(fig) 显式关闭图表，释放内存 2、AnalysisThread.stop() 仅设置 stop_requested=True，但线程池中的任务仍会继续运行，可能导致资源占用。结合 concurrent.futures 的 Executor.shutdown(wait=False) 强制终止线程池。 3、AudioAnalyzer.convert_audio 中若临时目录创建失败，temp_dir 为 None，后续 temp_dir.cleanup() 会报错。添加容错处理，确保临时目录安全清理。 4、ModelLoader 中加载的 Whisper 模型未充分利用硬件性能（如未指定 language 参数，可能增加语言检测耗时）。指定 language=“chinese” 减少推理时间。 5、convert_audio 中无论原始音频格式如何，均强制转换为 WAV，部分格式（如 16kHz 单声道 WAV）可跳过转换。 6、短时间内同一说话人的片段可能被拆分（如停顿导致），影响后续文本关联准确性。合并连续相同说话人的片段（如间隔 < 1 秒）。 7、当前情感分析仅基于文本内容，未结合对话上下文（如客户抱怨后客服回应的情感）。增加上下文权重，如客户表达消极情绪后，客服的回应情感权重提升。 8、模型加载失败后无重试逻辑，用户需重启程序。添加重试按钮，允许用户重新加载模型。 9、未校验音频文件的有效性（如损坏文件、非音频文件），可能导致分析线程崩溃。添加文件合法性校验，过滤无效文件。 代码： import os import sys import time import json import traceback import numpy as np import pandas as pd import torch import librosa import jieba import tempfile from pydub import AudioSegment from transformers import pipeline, AutoModelForSequenceClassification, AutoTokenizer from pyannote.audio import Pipeline from concurrent.futures import ThreadPoolExecutor, as_completed from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit, QPushButton, QFileDialog, QTextEdit, QProgressBar, QGroupBox, QCheckBox, QListWidget, QMessageBox) from PyQt5.QtCore import QThread, pyqtSignal, Qt, QTimer from PyQt5.QtGui import QFont from docx import Document from docx.shared import Inches import matplotlib.pyplot as plt from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas from collections import Counter # 全局配置 MODEL_CONFIG = { "whisper_model": "openai/whisper-small", "diarization_model": "pyannote/[email protected]", # 使用更轻量模型 "sentiment_model": "IDEA-CCNL/Erlangshen-Roberta-110M-Sentiment", "chunk_size": 10, # 强制10秒分块 "sample_rate": 16000, "device": "cuda" if torch.cuda.is_available() else "cpu", "max_workers": 2 if torch.cuda.is_available() else 4, # GPU模式下并行度降低 "batch_size": 8 # 批处理大小 } # 初始化分词器 jieba.initialize() class ModelLoader(QThread): """模型加载线程""" progress = pyqtSignal(str) finished = pyqtSignal(bool, str) def __init__(self): super().__init__() self.models = {} self.error = None def run(self): try: self.progress.emit("正在加载语音识别模型...") # 语音识别模型 self.models["asr_pipeline"] = pipeline( "automatic-speech-recognition", model=MODEL_CONFIG["whisper_model"], torch_dtype=torch.float16, device=MODEL_CONFIG["device"], batch_size=MODEL_CONFIG["batch_size"] # 添加批处理支持 ) self.progress.emit("正在加载说话人分离模型...") # 说话人分离模型 - 使用更轻量版本 self.models["diarization_pipeline"] = Pipeline.from_pretrained( MODEL_CONFIG["diarization_model"], use_auth_token=True ).to(torch.device(MODEL_CONFIG["device"]), torch.float16) self.progress.emit("正在加载情感分析模型...") # 情感分析模型 self.models["sentiment_tokenizer"] = AutoTokenizer.from_pretrained( MODEL_CONFIG["sentiment_model"] ) self.models["sentiment_model"] = AutoModelForSequenceClassification.from_pretrained( MODEL_CONFIG["sentiment_model"], torch_dtype=torch.float16 ).to(MODEL_CONFIG["device"]) self.finished.emit(True, "模型加载完成!") except Exception as e: self.error = str(e) traceback.print_exc() self.finished.emit(False, f"模型加载失败: {str(e)}") class AudioAnalyzer: """深度优化的核心音频分析类""" def __init__(self, models): self.keywords = { "opening": ["您好", "请问是", "先生/女士", "很高兴为您服务"], "closing": ["感谢接听", "祝您生活愉快", "再见", "有问题随时联系"], "forbidden": ["不可能", "没办法", "我不管", "随便你", "投诉也没用"], "solution": ["解决", "处理好了", "已完成", "满意吗", "还有问题吗"] } self.synonyms = { "不可能": ["不可能", "没可能", "做不到", "无法做到"], "解决": ["解决", "处理", "完成", "搞定", "办妥"] } self.models = models self.models_loaded = True if models else False def load_keywords(self, excel_path): """从Excel加载关键词和同义词""" try: # 使用更健壮的Excel读取方式 df = pd.read_excel(excel_path, sheet_name=None) if "开场白" in df: self.keywords["opening"] = df["开场白"].dropna()["关键词"].tolist() if "结束语" in df: self.keywords["closing"] = df["结束语"].dropna()["关键词"].tolist() if "禁语" in df: self.keywords["forbidden"] = df["禁语"].dropna()["关键词"].tolist() if "解决关键词" in df: self.keywords["solution"] = df["解决关键词"].dropna()["关键词"].tolist() # 加载同义词表 if "同义词" in df: for _, row in df["同义词"].iterrows(): main_word = row["主词"] synonyms = row["同义词"].split("、") self.synonyms[main_word] = synonyms return True, "关键词加载成功" except Exception as e: error_msg = f"加载关键词失败: {str(e)}" return False, error_msg def convert_audio(self, input_path): """转换音频为WAV格式并分块，使用临时目录管理""" try: # 创建临时目录 temp_dir = tempfile.TemporaryDirectory() # 读取音频文件 audio = AudioSegment.from_file(input_path) # 转换为单声道16kHz audio = audio.set_frame_rate(MODEL_CONFIG["sample_rate"]) audio = audio.set_channels(1) # 计算总时长 duration = len(audio) / 1000.0 # 毫秒转秒 # 分块处理（10秒） chunks = [] chunk_size = MODEL_CONFIG["chunk_size"] * 1000 # 毫秒 for i in range(0, len(audio), chunk_size): chunk = audio[i:i + chunk_size] chunk_path = os.path.join(temp_dir.name, f"chunk_{i // chunk_size}.wav") chunk.export(chunk_path, format="wav") chunks.append({ "path": chunk_path, "start_time": i / 1000.0, # 全局起始时间（秒） "end_time": (i + len(chunk)) / 1000.0 # 全局结束时间（秒） }) return chunks, duration, temp_dir except Exception as e: error_msg = f"音频转换失败: {str(e)}" return [], 0, None def diarize_speakers(self, audio_path): """说话人分离""" try: diarization = self.models["diarization_pipeline"](audio_path) segments = [] for turn, _, speaker in diarization.itertracks(yield_label=True): segments.append({ "start": turn.start, "end": turn.end, "speaker": speaker, "text": "" }) return segments except Exception as e: error_msg = f"说话人分离失败: {str(e)}" raise Exception(error_msg) from e def transcribe_audio_batch(self, chunk_paths): """批量语音识别多个分块""" try: # 批量处理音频分块 results = self.models["asr_pipeline"]( chunk_paths, chunk_length_s=MODEL_CONFIG["chunk_size"], stride_length_s=(4, 2), batch_size=MODEL_CONFIG["batch_size"], return_timestamps=True ) # 整理结果 transcribed_data = [] for result in results: text = result["text"] chunks = result["chunks"] transcribed_data.append((text, chunks)) return transcribed_data except Exception as e: error_msg = f"语音识别失败: {str(e)}" raise Exception(error_msg) from e def analyze_sentiment_batch(self, texts): """批量情感分析 - 支持长文本处理""" try: if not texts: return [] # 预处理文本 - 截断并添加特殊token inputs = self.models["sentiment_tokenizer"]( texts, padding=True, truncation=True, max_length=512, return_tensors="pt" ).to(MODEL_CONFIG["device"]) # 批量推理 with torch.no_grad(): outputs = self.models["sentiment_model"](**inputs) # 计算概率 probs = torch.softmax(outputs.logits, dim=-1).cpu().numpy() # 处理结果 results = [] labels = ["积极", "消极", "中性"] for i, text in enumerate(texts): sentiment = labels[np.argmax(probs[i])] # 情感强度检测 strong_negative = probs[i][1] > 0.7 # 消极概率超过70% strong_positive = probs[i][0] > 0.7 # 积极概率超过70% # 特定情绪检测 specific_emotion = "无" if "生气" in text or "愤怒" in text or "气死" in text: specific_emotion = "愤怒" elif "不耐烦" in text or "快点" in text or "急死" in text: specific_emotion = "不耐烦" elif "失望" in text or "无奈" in text: specific_emotion = "失望" # 如果有强烈情感则覆盖平均结果 if strong_negative: sentiment = "强烈消极" elif strong_positive: sentiment = "强烈积极" results.append({ "sentiment": sentiment, "emotion": specific_emotion, "scores": probs[i].tolist() }) return results except Exception as e: error_msg = f"情感分析失败: {str(e)}" raise Exception(error_msg) from e def match_keywords(self, text, keyword_type): """高级关键词匹配 - 使用分词和同义词""" # 获取关键词列表 keywords = self.keywords.get(keyword_type, []) if not keywords: return False # 分词处理 words = jieba.lcut(text) # 检查每个关键词 for keyword in keywords: # 检查直接匹配 if keyword in text: return True # 检查同义词 synonyms = self.synonyms.get(keyword, []) for synonym in synonyms: if synonym in text: return True # 检查分词匹配（全词匹配） if keyword in words: return True return False def identify_agent(self, segments, full_text): """智能客服身份识别""" # 候选客服信息 candidates = {} # 特征1：开场白关键词 for i, segment in enumerate(segments[:5]): # 检查前5个片段 if self.match_keywords(segment["text"], "opening"): speaker = segment["speaker"] candidates.setdefault(speaker, {"score": 0, "segments": []}) candidates[speaker]["score"] += 3 # 开场白权重高 candidates[speaker]["segments"].append(i) # 特征2：结束语关键词 for i, segment in enumerate(segments[-3:]): # 检查最后3个片段 if self.match_keywords(segment["text"], "closing"): speaker = segment["speaker"] candidates.setdefault(speaker, {"score": 0, "segments": []}) candidates[speaker]["score"] += 2 # 结束语权重中等 candidates[speaker]["segments"].append(len(segments) - 3 + i) # 特征3：说话时长 speaker_durations = {} for segment in segments: duration = segment["end"] - segment["start"] speaker_durations[segment["speaker"]] = speaker_durations.get(segment["speaker"], 0) + duration # 为说话时长最长的加分 if speaker_durations: max_duration = max(speaker_durations.values()) for speaker, duration in speaker_durations.items(): candidates.setdefault(speaker, {"score": 0, "segments": []}) if duration == max_duration: candidates[speaker]["score"] += 1 # 特征4：客服特定词汇出现频率 agent_keywords = ["客服", "代表", "专员", "先生", "女士"] speaker_keyword_count = {} for segment in segments: text = segment["text"] speaker = segment["speaker"] for word in agent_keywords: if word in text: speaker_keyword_count[speaker] = speaker_keyword_count.get(speaker, 0) + 1 # 为关键词出现最多的加分 if speaker_keyword_count: max_count = max(speaker_keyword_count.values()) for speaker, count in speaker_keyword_count.items(): if count == max_count: candidates.setdefault(speaker, {"score": 0, "segments": []}) candidates[speaker]["score"] += 1 # 选择得分最高的作为客服 if candidates: best_speaker = max(candidates.items(), key=lambda x: x[1]["score"])[0] return best_speaker # 默认选择第一个说话人 return segments[0]["speaker"] if segments else None def associate_speaker_text(self, segments, full_text_chunks): """基于时间重叠度的说话人-文本关联""" for segment in segments: segment_text = "" segment_start = segment["start"] segment_end = segment["end"] for word_info in full_text_chunks: if "global_start" not in word_info: continue word_start = word_info["global_start"] word_end = word_info["global_end"] # 计算重叠度 overlap_start = max(segment_start, word_start) overlap_end = min(segment_end, word_end) overlap = max(0, overlap_end - overlap_start) # 计算重叠比例 word_duration = word_end - word_start segment_duration = segment_end - segment_start if overlap > 0: # 如果重叠超过50%或单词完全在片段内 if (overlap / word_duration > 0.5) or (overlap / segment_duration > 0.5): segment_text += word_info["text"] + " " segment["text"] = segment_text.strip() def analyze_audio(self, audio_path): """完整分析单个音频文件 - 优化版本""" try: # 步骤1: 转换音频并分块（使用临时目录） chunks, duration, temp_dir = self.convert_audio(audio_path) if not chunks or not temp_dir: raise Exception("音频转换失败或未生成分块") try: # 步骤2: 说话人分离 segments = self.diarize_speakers(audio_path) # 步骤3: 批量语音识别 chunk_paths = [chunk["path"] for chunk in chunks] transcribed_data = self.transcribe_audio_batch(chunk_paths) # 步骤4: 处理识别结果 full_text_chunks = [] for idx, (text, chunk_data) in enumerate(transcribed_data): chunk = chunks[idx] # 调整时间戳为全局时间 for word_info in chunk_data: if "timestamp" in word_info: start, end = word_info["timestamp"] word_info["global_start"] = chunk["start_time"] + start word_info["global_end"] = chunk["start_time"] + end else: word_info["global_start"] = chunk["start_time"] word_info["global_end"] = chunk["end_time"] full_text_chunks.extend(chunk_data) # 步骤5: 基于时间重叠度关联说话人和文本 self.associate_speaker_text(segments, full_text_chunks) # 步骤6: 智能识别客服身份 agent_id = self.identify_agent(segments, full_text_chunks) # 步骤7: 提取客服和客户文本 agent_text = "" customer_text = "" opening_found = False closing_found = False forbidden_found = False for segment in segments: if segment["speaker"] == agent_id: agent_text += segment["text"] + " " else: customer_text += segment["text"] + " " # 使用高级关键词匹配 if not opening_found and self.match_keywords(segment["text"], "opening"): opening_found = True if not closing_found and self.match_keywords(segment["text"], "closing"): closing_found = True if not forbidden_found and self.match_keywords(segment["text"], "forbidden"): forbidden_found = True # 步骤8: 批量情感分析 sentiment_results = self.analyze_sentiment_batch([agent_text, customer_text]) if sentiment_results: agent_sentiment = sentiment_results[0]["sentiment"] agent_emotion = sentiment_results[0]["emotion"] customer_sentiment = sentiment_results[1]["sentiment"] customer_emotion = sentiment_results[1]["emotion"] else: agent_sentiment = "未知" agent_emotion = "无" customer_sentiment = "未知" customer_emotion = "无" # 问题解决率分析 solution_found = self.match_keywords(agent_text, "solution") # 语速分析 agent_words = len(agent_text.split()) agent_duration = sum([s["end"] - s["start"] for s in segments if s["speaker"] == agent_id]) agent_speed = agent_words / (agent_duration / 60) if agent_duration > 0 else 0 # 词/分钟 # 音量分析（简单版） try: y, sr = librosa.load(audio_path, sr=MODEL_CONFIG["sample_rate"]) rms = librosa.feature.rms(y=y) avg_volume = np.mean(rms) volume_stability = np.std(rms) / avg_volume if avg_volume > 0 else 0 except: avg_volume = 0 volume_stability = 0 # 构建结果 result = { "file_name": os.path.basename(audio_path), "duration": round(duration, 2), "opening_check": "是" if opening_found else "否", "closing_check": "是" if closing_found else "否", "forbidden_check": "是" if forbidden_found else "否", "agent_sentiment": agent_sentiment, "agent_emotion": agent_emotion, "customer_sentiment": customer_sentiment, "customer_emotion": customer_emotion, "agent_speed": round(agent_speed, 1), "volume_level": round(avg_volume, 4), "volume_stability": round(volume_stability, 2), "solution_rate": "是" if solution_found else "否", "agent_text": agent_text[:500] + "..." if len(agent_text) > 500 else agent_text, "customer_text": customer_text[:500] + "..." if len(customer_text) > 500 else customer_text } return result finally: # 自动清理临时目录 temp_dir.cleanup() except Exception as e: error_msg = f"分析文件 {os.path.basename(audio_path)} 时出错: {str(e)}" raise Exception(error_msg) from e class AnalysisThread(QThread): """分析线程 - 并行优化版本""" progress = pyqtSignal(int, str) result_ready = pyqtSignal(dict) finished_all = pyqtSignal() error_occurred = pyqtSignal(str, str) def __init__(self, audio_files, keywords_file, output_dir, models): super().__init__() self.audio_files = audio_files self.keywords_file = keywords_file self.output_dir = output_dir self.stop_requested = False self.analyzer = AudioAnalyzer(models) self.completed_count = 0 def run(self): try: total = len(self.audio_files) # 加载关键词 if self.keywords_file: success, msg = self.analyzer.load_keywords(self.keywords_file) if not success: self.error_occurred.emit("关键词加载", msg) results = [] errors = [] # 使用线程池进行并行处理 with ThreadPoolExecutor(max_workers=MODEL_CONFIG["max_workers"]) as executor: # 提交所有任务 future_to_file = { executor.submit(self.analyzer.analyze_audio, audio_file): audio_file for audio_file in self.audio_files } # 处理完成的任务 for future in as_completed(future_to_file): if self.stop_requested: break audio_file = future_to_file[future] try: result = future.result() if result: results.append(result) self.result_ready.emit(result) except Exception as e: error_msg = str(e) errors.append({ "file": audio_file, "error": error_msg }) self.error_occurred.emit(os.path.basename(audio_file), error_msg) # 更新进度 self.completed_count += 1 progress = int(self.completed_count / total * 100) self.progress.emit( progress, f"已完成 {self.completed_count}/{total} ({progress}%)" ) # 生成报告 if results: self.generate_reports(results, errors) self.finished_all.emit() except Exception as e: self.error_occurred.emit("全局错误", str(e)) def stop(self): self.stop_requested = True def generate_reports(self, results, errors): """生成Excel和Word报告 - 优化版本""" try: # 生成Excel报告 df = pd.DataFrame(results) excel_path = os.path.join(self.output_dir, "质检分析报告.xlsx") # 创建Excel写入器 with pd.ExcelWriter(excel_path, engine='xlsxwriter') as writer: df.to_excel(writer, sheet_name='详细结果', index=False) # 添加统计摘要 stats_data = { "指标": ["分析文件总数", "成功分析文件数", "分析失败文件数", "开场白合格率", "结束语合格率", "禁语出现率", "客服积极情绪占比", "客户消极情绪占比", "问题解决率"], "数值": [ len(results) + len(errors), len(results), len(errors), f"{df['opening_check'].value_counts(normalize=True).get('是', 0) * 100:.1f}%", f"{df['closing_check'].value_counts(normalize=True).get('是', 0) * 100:.1f}%", f"{df['forbidden_check'].value_counts(normalize=True).get('是', 0) * 100:.1f}%", f"{df[df['agent_sentiment'] == '积极'].shape[0] / len(df) * 100:.1f}%", f"{df[df['customer_sentiment'] == '消极'].shape[0] / len(df) * 100:.1f}%", f"{df['solution_rate'].value_counts(normalize=True).get('是', 0) * 100:.1f}%" ] } stats_df = pd.DataFrame(stats_data) stats_df.to_excel(writer, sheet_name='统计摘要', index=False) # 生成Word报告 doc = Document() doc.add_heading('外呼电话质检分析汇总报告', 0) # 添加统计信息 doc.add_heading('整体统计', level=1) stats = [ f"分析文件总数: {len(results) + len(errors)}", f"成功分析文件数: {len(results)}", f"分析失败文件数: {len(errors)}", f"开场白合格率: {stats_data['数值'][3]}", f"结束语合格率: {stats_data['数值'][4]}", f"禁语出现率: {stats_data['数值'][5]}", f"客服积极情绪占比: {stats_data['数值'][6]}", f"客户消极情绪占比: {stats_data['数值'][7]}", f"问题解决率: {stats_data['数值'][8]}" ] for stat in stats: doc.add_paragraph(stat) # 添加图表 self.add_charts(doc, df) # 添加错误列表 if errors: doc.add_heading('分析失败文件', level=1) table = doc.add_table(rows=1, cols=2) hdr_cells = table.rows[0].cells hdr_cells[0].text = '文件' hdr_cells[1].text = '错误原因' for error in errors: row_cells = table.add_row().cells row_cells[0].text = os.path.basename(error['file']) row_cells[1].text = error['error'] word_path = os.path.join(self.output_dir, "可视化分析报告.docx") doc.save(word_path) return True, f"报告已保存到: {self.output_dir}" except Exception as e: return False, f"生成报告失败: {str(e)}" def add_charts(self, doc, df): """在Word文档中添加图表""" try: # 客服情感分布 fig1, ax1 = plt.subplots(figsize=(6, 4)) sentiment_counts = df['agent_sentiment'].value_counts() sentiment_counts.plot(kind='bar', ax=ax1, color=['green', 'red', 'blue', 'darkred', 'darkgreen']) ax1.set_title('客服情感分布') ax1.set_xlabel('情感类型') ax1.set_ylabel('数量') fig1.tight_layout() fig1.savefig('agent_sentiment.png') doc.add_picture('agent_sentiment.png', width=Inches(5)) os.remove('agent_sentiment.png') # 客户情感分布 fig2, ax2 = plt.subplots(figsize=(6, 4)) df['customer_sentiment'].value_counts().plot(kind='bar', ax=ax2, color=['green', 'red', 'blue', 'darkred', 'darkgreen']) ax2.set_title('客户情感分布') ax2.set_xlabel('情感类型') ax2.set_ylabel('数量') fig2.tight_layout() fig2.savefig('customer_sentiment.png') doc.add_picture('customer_sentiment.png', width=Inches(5)) os.remove('customer_sentiment.png') # 合规性检查 fig3, ax3 = plt.subplots(figsize=(6, 4)) compliance = df[['opening_check', 'closing_check', 'forbidden_check']].apply( lambda x: x.value_counts().get('是', 0)) compliance.plot(kind='bar', ax=ax3, color=['blue', 'green', 'red']) ax3.set_title('合规性检查') ax3.set_xlabel('检查项') ax3.set_ylabel('合格数量') fig3.tight_layout() fig3.savefig('compliance.png') doc.add_picture('compliance.png', width=Inches(5)) os.remove('compliance.png') except Exception as e: print(f"生成图表失败: {str(e)}") class MainWindow(QMainWindow): """主界面 - 优化版本""" def __init__(self): super().__init__() self.setWindowTitle("外呼电话录音质检分析系统") self.setGeometry(100, 100, 1000, 800) # 初始化变量 self.audio_files = [] self.keywords_file = "" self.output_dir = os.getcwd() self.analysis_thread = None self.model_loader = None self.models = {} self.models_loaded = False # 初始化为False # 设置全局字体 app_font = QFont("Microsoft YaHei", 10) QApplication.setFont(app_font) # 创建主布局 main_widget = QWidget() main_layout = QVBoxLayout() main_layout.setSpacing(10) main_layout.setContentsMargins(15, 15, 15, 15) # 状态栏 self.status_label = QLabel("准备就绪") self.status_label.setAlignment(Qt.AlignCenter) self.status_label.setStyleSheet("background-color: #f0f0f0; padding: 5px; border-radius: 5px;") # 文件选择区域 file_group = QGroupBox("文件选择") file_layout = QVBoxLayout() file_layout.setSpacing(10) # 音频选择 audio_layout = QHBoxLayout() self.audio_label = QLabel("音频文件/文件夹:") self.audio_path_edit = QLineEdit() self.audio_path_edit.setReadOnly(True) self.audio_path_edit.setPlaceholderText("请选择音频文件或文件夹") self.audio_browse_btn = QPushButton("浏览...") self.audio_browse_btn.setFixedWidth(80) self.audio_browse_btn.clicked.connect(self.browse_audio) audio_layout.addWidget(self.audio_label) audio_layout.addWidget(self.audio_path_edit, 1) audio_layout.addWidget(self.audio_browse_btn) # 关键词选择 keyword_layout = QHBoxLayout() self.keyword_label = QLabel("关键词文件:") self.keyword_path_edit = QLineEdit() self.keyword_path_edit.setReadOnly(True) self.keyword_path_edit.setPlaceholderText("可选：选择关键词Excel文件") self.keyword_browse_btn = QPushButton("浏览...") self.keyword_browse_btn.setFixedWidth(80) self.keyword_browse_btn.clicked.connect(self.browse_keywords) keyword_layout.addWidget(self.keyword_label) keyword_layout.addWidget(self.keyword_path_edit, 1) keyword_layout.addWidget(self.keyword_browse_btn) # 输出目录 output_layout = QHBoxLayout() self.output_label = QLabel("输出目录:") self.output_path_edit = QLineEdit(os.getcwd()) self.output_path_edit.setReadOnly(True) self.output_browse_btn = QPushButton("浏览...") self.output_browse_btn.setFixedWidth(80) self.output_browse_btn.clicked.connect(self.browse_output) output_layout.addWidget(self.output_label) output_layout.addWidget(self.output_path_edit, 1) output_layout.addWidget(self.output_browse_btn) file_layout.addLayout(audio_layout) file_layout.addLayout(keyword_layout) file_layout.addLayout(output_layout) file_group.setLayout(file_layout) # 控制按钮区域 control_layout = QHBoxLayout() control_layout.setSpacing(15) self.start_btn = QPushButton("开始分析") self.start_btn.setFixedHeight(40) self.start_btn.setStyleSheet("background-color: #4CAF50; color: white; font-weight: bold;") self.start_btn.clicked.connect(self.start_analysis) self.stop_btn = QPushButton("停止分析") self.stop_btn.setFixedHeight(40) self.stop_btn.setStyleSheet("background-color: #f44336; color: white; font-weight: bold;") self.stop_btn.clicked.connect(self.stop_analysis) self.stop_btn.setEnabled(False) self.clear_btn = QPushButton("清空") self.clear_btn.setFixedHeight(40) self.clear_btn.setStyleSheet("background-color: #2196F3; color: white; font-weight: bold;") self.clear_btn.clicked.connect(self.clear_all) control_layout.addWidget(self.start_btn) control_layout.addWidget(self.stop_btn) control_layout.addWidget(self.clear_btn) # 进度条 self.progress_bar = QProgressBar() self.progress_bar.setRange(0, 100) self.progress_bar.setTextVisible(True) self.progress_bar.setStyleSheet("QProgressBar {border: 1px solid grey; border-radius: 5px; text-align: center;}" "QProgressBar::chunk {background-color: #4CAF50; width: 10px;}") # 结果展示区域 result_group = QGroupBox("分析结果") result_layout = QVBoxLayout() result_layout.setSpacing(10) # 结果标签 result_header = QHBoxLayout() self.result_label = QLabel("分析结果:") self.result_count_label = QLabel("0/0") self.result_count_label.setAlignment(Qt.AlignRight) result_header.addWidget(self.result_label) result_header.addWidget(self.result_count_label) self.result_text = QTextEdit() self.result_text.setReadOnly(True) self.result_text.setStyleSheet("font-family: Consolas, 'Microsoft YaHei';") # 错误列表 error_header = QHBoxLayout() self.error_label = QLabel("错误信息:") self.error_count_label = QLabel("0") self.error_count_label.setAlignment(Qt.AlignRight) error_header.addWidget(self.error_label) error_header.addWidget(self.error_count_label) self.error_list = QListWidget() self.error_list.setFixedHeight(120) self.error_list.setStyleSheet("color: #d32f2f;") result_layout.addLayout(result_header) result_layout.addWidget(self.result_text) result_layout.addLayout(error_header) result_layout.addWidget(self.error_list) result_group.setLayout(result_layout) # 添加到主布局 main_layout.addWidget(file_group) main_layout.addLayout(control_layout) main_layout.addWidget(self.progress_bar) main_layout.addWidget(self.status_label) main_layout.addWidget(result_group) main_widget.setLayout(main_layout) self.setCentralWidget(main_widget) # 启动模型加载 self.load_models() def load_models(self): """后台加载模型""" self.status_label.setText("正在加载AI模型，请稍候...") self.start_btn.setEnabled(False) self.model_loader = ModelLoader() self.model_loader.progress.connect(self.update_model_loading_status) self.model_loader.finished.connect(self.handle_model_loading_finished) self.model_loader.start() def update_model_loading_status(self, message): """更新模型加载状态""" self.status_label.setText(message) def handle_model_loading_finished(self, success, message): """处理模型加载完成""" if success: self.models = self.model_loader.models self.models_loaded = True # 修复标志位 self.status_label.setText(message) self.start_btn.setEnabled(True) else: self.status_label.setText(message) QMessageBox.critical(self, "模型加载失败", message) def browse_audio(self): """选择音频文件或文件夹""" options = QFileDialog.Options() files, _ = QFileDialog.getOpenFileNames( self, "选择音频文件", "", "音频文件 (*.mp3 *.wav *.amr *.flac *.m4a);;所有文件 (*)", options=options ) if files: self.audio_files = files self.audio_path_edit.setText(f"已选择 {len(files)} 个文件") self.result_count_label.setText(f"0/{len(files)}") def browse_keywords(self): """选择关键词文件""" options = QFileDialog.Options() file, _ = QFileDialog.getOpenFileName( self, "选择关键词文件", "", "Excel文件 (*.xlsx);;所有文件 (*)", options=options ) if file: self.keywords_file = file self.keyword_path_edit.setText(os.path.basename(file)) def browse_output(self): """选择输出目录""" options = QFileDialog.Options() directory = QFileDialog.getExistingDirectory( self, "选择输出目录", options=options ) if directory: self.output_dir = directory self.output_path_edit.setText(directory) def start_analysis(self): """开始分析""" if not self.audio_files: self.show_message("错误", "请先选择音频文件!") return if not self.models_loaded: # 使用修复后的标志位 self.show_message("错误", "AI模型尚未加载完成!") return # 检查输出目录 if not os.path.exists(self.output_dir): try: os.makedirs(self.output_dir) except Exception as e: self.show_message("错误", f"无法创建输出目录: {str(e)}") return # 更新UI状态 self.start_btn.setEnabled(False) self.stop_btn.setEnabled(True) self.result_text.clear() self.error_list.clear() self.error_count_label.setText("0") self.result_text.append("开始分析音频文件...") self.progress_bar.setValue(0) # 创建并启动分析线程 self.analysis_thread = AnalysisThread( self.audio_files, self.keywords_file, self.output_dir, self.models ) # 连接信号 self.analysis_thread.progress.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.handle_error) 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() self.result_text.append("分析已停止") self.status_label.setText("分析已停止") self.start_btn.setEnabled(True) self.stop_btn.setEnabled(False) def clear_all(self): """清空所有内容""" self.audio_files = [] self.keywords_file = "" self.audio_path_edit.clear() self.keyword_path_edit.clear() self.result_text.clear() self.error_list.clear() self.progress_bar.setValue(0) self.status_label.setText("准备就绪") self.result_count_label.setText("0/0") self.error_count_label.setText("0") def update_progress(self, value, message): """更新进度""" self.progress_bar.setValue(value) self.status_label.setText(message) # 更新结果计数 if "已完成" in message: parts = message.split() if len(parts) >= 2: self.result_count_label.setText(parts[1]) def handle_result(self, result): """处理单个结果""" summary = f""" 文件: {result['file_name']} 时长: {result['duration']}秒 ---------------------------------------- 开场白: {result['opening_check']} | 结束语: {result['closing_check']} | 禁语: {result['forbidden_check']} 客服情感: {result['agent_sentiment']} ({result['agent_emotion']}) | 语速: {result['agent_speed']}词/分 客户情感: {result['customer_sentiment']} ({result['customer_emotion']}) 问题解决: {result['solution_rate']} 音量水平: {result['volume_level']} | 稳定性: {result['volume_stability']} ---------------------------------------- """ self.result_text.append(summary) def handle_error(self, file_name, error): """处理错误""" self.error_list.addItem(f"{file_name}: {error}") self.error_count_label.setText(str(self.error_list.count())) def analysis_finished(self): """分析完成""" self.start_btn.setEnabled(True) self.stop_btn.setEnabled(False) self.status_label.setText(f"分析完成! 报告已保存到: {self.output_dir}") self.result_text.append("分析完成!") # 显示完成消息 self.show_message("完成", f"分析完成! 报告已保存到: {self.output_dir}") def show_message(self, title, message): """显示消息对话框""" msg = QMessageBox(self) msg.setWindowTitle(title) msg.setText(message) msg.setStandardButtons(QMessageBox.Ok) msg.exec_() if __name__ == "__main__": app = QApplication(sys.argv) # 检查GPU可用性 if MODEL_CONFIG["device"] == "cuda": try: gpu_mem = torch.cuda.get_device_properties(0).total_memory / (1024 ** 3) print(f"GPU内存: {gpu_mem:.2f}GB") # 根据GPU内存调整并行度 if gpu_mem < 4: # 确保有足够内存 MODEL_CONFIG["device"] = "cpu" MODEL_CONFIG["max_workers"] = 4 print("GPU内存不足，切换到CPU模式") elif gpu_mem < 8: MODEL_CONFIG["max_workers"] = 2 else: MODEL_CONFIG["max_workers"] = 4 except: MODEL_CONFIG["device"] = "cpu" MODEL_CONFIG["max_workers"] = 4 print("无法获取GPU信息，切换到CPU模式") window = MainWindow() window.show() sys.exit(app.exec_())

5. 在 convert_audio 中，如果原始音频已经是 16kHz 单声道 WAV 格式，则跳过转换（这里我们简化处理，实际中需要判断格式，但为了代码简洁，我们暂时不添加复杂判断，因为转换函数已经能够处理多种格式）。...

AttributeError: 'module' object has no attribute 'json_normalize'

如果你已经安装了pandas并且导入了正确的模块（通常是import pandas as pd），你应该可以直接使用pd.json_normalize()。如果没有导入，你需要先导入pandas.io.json： python from pandas.io.json import ...

import os import pandas as pd import tkinter as tk from tkinter import ttk, filedialog, scrolledtext, messagebox from tkinter.colorchooser import askcolor from difflib import SequenceMatcher import re import openpyxl import threading import numpy as np from openpyxl.utils import get_column_letter import xlrd import gc import hashlib import json import tempfile from concurrent.futures import ThreadPoolExecutor, as_completed import unicodedata class EnhancedSignalComparator: def __init__(self, root): self.root = root self.root.title("增强版信号功能对比工具") self.root.geometry("1200x800") self.root.configure(bg="#f0f0f0") # 初始化变量 self.folder_path = tk.StringVar() self.search_text = tk.StringVar() self.files = [] self.results = {} # 存储信号对比结果 self.highlight_color = "#FFD700" # 默认高亮色 self.search_running = False self.stop_requested = False self.cache_dir = os.path.join(tempfile.gettempdir(), "excel_cache") self.file_cache = {} # 文件缓存 self.column_cache = {} # 列名缓存 self.max_workers = 4 # 最大并发线程数 # 创建缓存目录 os.makedirs(self.cache_dir, exist_ok=True) # 创建界面 self.create_widgets() def create_widgets(self): # 顶部控制面板 control_frame = ttk.Frame(self.root, padding=10) control_frame.pack(fill=tk.X) # 文件夹选择 ttk.Label(control_frame, text="选择文件夹:").grid(row=0, column=0, sticky=tk.W) folder_entry = ttk.Entry(control_frame, textvariable=self.folder_path, width=50) folder_entry.grid(row=0, column=1, padx=5, sticky=tk.EW) ttk.Button(control_frame, text="浏览...", command=self.browse_folder).grid(row=0, column=2) # 搜索输入 ttk.Label(control_frame, text="搜索信号:").grid(row=1, column=0, sticky=tk.W, pady=(10,0)) search_entry = ttk.Entry(control_frame, textvariable=self.search_text, width=50) search_entry.grid(row=1, column=1, padx=5, pady=(10,0), sticky=tk.EW) search_entry.bind("<Return>", lambda event: self.start_search_thread()) ttk.Button(control_frame, text="搜索", command=self.start_search_thread).grid(row=1, column=2, pady=(10,0)) ttk.Button(control_frame, text="停止", command=self.stop_search).grid(row=1, column=3, pady=(10,0), padx=5) # 高级选项 ttk.Label(control_frame, text="并发线程:").grid(row=2, column=0, sticky=tk.W, pady=(10,0)) self.thread_var = tk.StringVar(value="4") ttk.Combobox(control_frame, textvariable=self.thread_var, values=["1", "2", "4", "8"], width=5).grid(row=2, column=1, sticky=tk.W, padx=5, pady=(10,0)) # 文件过滤 ttk.Label(control_frame, text="文件过滤:").grid(row=2, column=2, sticky=tk.W, pady=(10,0)) self.filter_var = tk.StringVar(value="*.xlsx;*.xlsm;*.xls") ttk.Entry(control_frame, textvariable=self.filter_var, width=20).grid(row=2, column=3, sticky=tk.W, padx=5, pady=(10,0)) # 高亮颜色选择 ttk.Label(control_frame, text="高亮颜色:").grid(row=3, column=0, sticky=tk.W, pady=(10,0)) self.color_btn = tk.Button(control_frame, bg=self.highlight_color, width=3, command=self.choose_color) self.color_btn.grid(row=3, column=1, sticky=tk.W, padx=5, pady=(10,0)) # 进度条 self.progress = ttk.Progressbar(control_frame, orient="horizontal", length=200, mode="determinate") self.progress.grid(row=3, column=2, columnspan=2, sticky=tk.EW, padx=5, pady=(10,0)) # 结果标签 self.result_label = ttk.Label(control_frame, text="") self.result_label.grid(row=3, column=4, sticky=tk.W, padx=5, pady=(10,0)) # 对比面板 notebook = ttk.Notebook(self.root) notebook.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) # 表格视图 self.table_frame = ttk.Frame(notebook) notebook.add(self.table_frame, text="表格视图") # 文本对比视图 self.text_frame = ttk.Frame(notebook) notebook.add(self.text_frame, text="行内容对比") # 状态栏 self.status_var = tk.StringVar() status_bar = ttk.Label(self.root, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W) status_bar.pack(side=tk.BOTTOM, fill=tk.X) # 初始化表格和文本区域 self.init_table_view() self.init_text_view() def init_table_view(self): """初始化表格视图""" # 创建树状表格 columns = ("信号", "文件", "行内容摘要") self.tree = ttk.Treeview(self.table_frame, columns=columns, show="headings") # 设置列标题 for col in columns: self.tree.heading(col, text=col) self.tree.column(col, width=200, anchor=tk.W) # 添加滚动条 scrollbar = ttk.Scrollbar(self.table_frame, orient=tk.VERTICAL, command=self.tree.yview) self.tree.configure(yscrollcommand=scrollbar.set) self.tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) scrollbar.pack(side=tk.RIGHT, fill=tk.Y) # 绑定选择事件 self.tree.bind("<<TreeviewSelect>>", self.on_table_select) def init_text_view(self): """初始化文本对比视图""" self.text_panes = {} self.text_frame.columnconfigure(0, weight=1) self.text_frame.rowconfigure(0, weight=1) # 创建对比容器 self.compare_container = ttk.Frame(self.text_frame) self.compare_container.grid(row=0, column=0, sticky="nsew", padx=5, pady=5) # 添加差异高亮按钮 btn_frame = ttk.Frame(self.text_frame) btn_frame.grid(row=1, column=0, sticky="ew", padx=5, pady=5) ttk.Button(btn_frame, text="高亮显示差异", command=self.highlight_differences).pack(side=tk.LEFT) ttk.Button(btn_frame, text="导出差异报告", command=self.export_report).pack(side=tk.LEFT, padx=5) ttk.Button(btn_frame, text="清除缓存", command=self.clear_cache).pack(side=tk.LEFT, padx=5) ttk.Button(btn_frame, text="手动指定列名", command=self.manual_column_select).pack(side=tk.LEFT, padx=5) def browse_folder(self): """选择文件夹""" folder = filedialog.askdirectory(title="选择包含Excel文件的文件夹") if folder: self.folder_path.set(folder) self.load_files() def load_files(self): """加载文件夹中的Excel文件（优化特殊字符处理）""" folder = self.folder_path.get() if not folder or not os.path.isdir(folder): return # 获取文件过滤模式 filter_patterns = self.filter_var.get().split(';') self.files = [] for file in os.listdir(folder): file_path = os.path.join(folder, file) # 跳过临时文件 if file.startswith('~$'): continue # 检查文件扩展名 file_lower = file.lower() matched = False for pattern in filter_patterns: # 移除通配符并转换为小写 ext = pattern.replace('*', '').lower() if file_lower.endswith(ext): matched = True break if matched: # 规范化文件名处理特殊字符 normalized_path = self.normalize_file_path(file_path) if normalized_path and os.path.isfile(normalized_path): self.files.append(normalized_path) self.status_var.set(f"找到 {len(self.files)} 个Excel文件") def normalize_file_path(self, path): """规范化文件路径，处理特殊字符""" try: # 尝试直接访问文件 if os.path.exists(path): return path # 尝试Unicode规范化 normalized = unicodedata.normalize('NFC', path) if os.path.exists(normalized): return normalized # 尝试不同编码方案 encodings = ['utf-8', 'shift_jis', 'euc-jp', 'cp932'] for encoding in encodings: try: decoded = path.encode('latin1').decode(encoding) if os.path.exists(decoded): return decoded except: continue # 最终尝试原始路径 return path except Exception as e: self.status_var.set(f"文件路径处理错误: {str(e)}") return path def get_file_hash(self, file_path): """计算文件哈希值用于缓存""" try: hash_md5 = hashlib.md5() with open(file_path, "rb") as f: for chunk in iter(lambda: f.read(4096), b""): hash_md5.update(chunk) return hash_md5.hexdigest() except Exception as e: self.status_var.set(f"计算文件哈希失败: {str(e)}") return str(os.path.getmtime(file_path)) def get_cache_filename(self, file_path): """获取缓存文件名""" file_hash = self.get_file_hash(file_path) return os.path.join(self.cache_dir, f"{os.path.basename(file_path)}_{file_hash}.cache") def load_header_cache(self, file_path): """加载列名缓存""" cache_file = self.get_cache_filename(file_path) if os.path.exists(cache_file): try: with open(cache_file, "r", encoding='utf-8') as f: return json.load(f) except: return None return None def save_header_cache(self, file_path, header_info): """保存列名缓存""" cache_file = self.get_cache_filename(file_path) try: with open(cache_file, "w", encoding='utf-8') as f: json.dump(header_info, f) return True except: return False def find_header_row(self, file_path): """查找列名行（增强版）""" # 检查缓存 cache = self.load_header_cache(file_path) if cache: return cache.get("header_row"), cache.get("signal_col") # 没有缓存则重新查找 if file_path.lower().endswith((".xlsx", ".xlsm")): return self.find_header_row_openpyxl(file_path) elif file_path.lower().endswith(".xls"): return self.find_header_row_xlrd(file_path) return None, None def find_header_row_openpyxl(self, file_path): """使用openpyxl查找列名行（增强版）""" try: wb = openpyxl.load_workbook(file_path, read_only=True, data_only=True) ws = wb.active # 尝试多种列名匹配模式 patterns = [ r'ﾃﾞｰﾀ名', # 半角片假名 r'データ名', # 全角片假名 r'信号名', # 中文 r'Signal Name', # 英文 r'Data Name', r'信号名称', r'データ名称' ] # 扩大搜索范围：前100行和前100列 for row_idx in range(1, 101): # 1-100行 # 扩大列搜索范围到100列 for col_idx in range(1, 101): # 1-100列 try: cell = ws.cell(row=row_idx, column=col_idx) cell_value = cell.value if not cell_value: continue # 尝试所有匹配模式 cell_str = str(cell_value) for pattern in patterns: if re.search(pattern, cell_str, re.IGNORECASE): # 找到列名行后，尝试确定信号列 signal_col = None # 在同行中查找信号列 for col_idx2 in range(1, 101): # 1-100列 try: cell2 = ws.cell(row=row_idx, column=col_idx2) cell2_value = cell2.value if not cell2_value: continue cell2_str = str(cell2_value) if re.search(pattern, cell2_str, re.IGNORECASE): signal_col = col_idx2 break except: continue # 保存缓存 if signal_col is not None: header_info = {"header_row": row_idx, "signal_col": signal_col} self.save_header_cache(file_path, header_info) wb.close() return row_idx, signal_col except: continue wb.close() except Exception as e: self.status_var.set(f"查找列名行出错: {str(e)}") return None, None def find_header_row_xlrd(self, file_path): """使用xlrd查找列名行（增强版）""" try: wb = xlrd.open_workbook(file_path) ws = wb.sheet_by_index(0) # 尝试多种列名匹配模式 patterns = [ r'ﾃﾞｰﾀ名', # 半角片假名 r'データ名', # 全角片假名 r'信号名', # 中文 r'Signal Name', # 英文 r'Data Name', r'信号名称', r'データ名称' ] # 扩大搜索范围：前100行和前100列 for row_idx in range(0, 100): # 0-99行 # 扩大列搜索范围到100列 for col_idx in range(0, 100): # 0-99列 try: cell_value = ws.cell_value(row_idx, col_idx) if not cell_value: continue # 尝试所有匹配模式 cell_str = str(cell_value) for pattern in patterns: if re.search(pattern, cell_str, re.IGNORECASE): # 找到列名行后，尝试确定信号列 signal_col = None # 在同行中查找信号列 for col_idx2 in range(0, 100): # 0-99列 try: cell2_value = ws.cell_value(row_idx, col_idx2) if not cell2_value: continue cell2_str = str(cell2_value) if re.search(pattern, cell2_str, re.IGNORECASE): signal_col = col_idx2 break except: continue # 保存缓存 if signal_col is not None: header_info = {"header_row": row_idx, "signal_col": signal_col} self.save_header_cache(file_path, header_info) return row_idx, signal_col except: continue except Exception as e: self.status_var.set(f"查找列名行出错: {str(e)}") return None, None def extract_row_content(self, ws, row_idx, header_row, max_cols=100): """高效提取行内容（最多到100列）""" content = [] # 扩展到100列 for col_idx in range(1, max_cols + 1): try: cell = ws.cell(row=row_idx, column=col_idx) if cell.value is not None and str(cell.value).strip() != '': # 使用列名缓存 col_key = f"{header_row}-{col_idx}" if col_key in self.column_cache: col_name = self.column_cache[col_key] else: col_name_cell = ws.cell(row=header_row, column=col_idx) col_name = col_name_cell.value if col_name_cell.value else f"列{get_column_letter(col_idx)}" self.column_cache[col_key] = col_name content.append(f"{col_name}: {str(cell.value).strip()}") except: continue return "\n".join(content) def start_search_thread(self): """启动搜索线程""" if self.search_running: return self.search_running = True self.stop_requested = False self.max_workers = int(self.thread_var.get()) threading.Thread(target=self.search_files, daemon=True).start() def stop_search(self): """停止搜索""" self.stop_requested = True self.status_var.set("正在停止搜索...") def search_files(self): """在文件中搜索内容（优化特殊文件处理）""" search_term = self.search_text.get().strip() if not search_term: self.status_var.set("请输入搜索内容") self.search_running = False return if not self.files: self.status_var.set("请先选择文件夹") self.search_running = False return # 重置结果和UI self.results = {} for item in self.tree.get_children(): self.tree.delete(item) total_files = len(self.files) processed_files = 0 found_signals = 0 # 使用线程池处理文件 with ThreadPoolExecutor(max_workers=self.max_workers) as executor: futures = {} for file_path in self.files: if self.stop_requested: break future = executor.submit(self.process_file, file_path, search_term) futures[future] = file_path # 处理完成的任务 for future in as_completed(futures): if self.stop_requested: break file_path = futures[future] try: found = future.result() found_signals += found processed_files += 1 # 更新进度 progress = int(processed_files / total_files * 100) self.progress["value"] = progress self.status_var.set(f"已处理 {processed_files}/{total_files} 个文件") self.root.update_idletasks() except Exception as e: self.status_var.set(f"处理文件 {os.path.basename(file_path)} 出错: {str(e)}") # 更新结果 if self.stop_requested: self.status_var.set(f"搜索已停止，已处理 {processed_files}/{total_files} 个文件") elif found_signals == 0: self.status_var.set(f"未找到包含 '{search_term}' 的信号") else: self.status_var.set(f"找到 {len(self.results)} 个匹配信号，共 {found_signals} 处匹配") self.update_text_view() self.progress["value"] = 0 self.search_running = False gc.collect() # 强制垃圾回收释放内存 def process_file(self, file_path, search_term): """处理单个文件（增强异常处理）""" found = 0 try: # 获取列名行和信号列 header_row, signal_col = self.find_header_row(file_path) # 如果自动查找失败，尝试手动模式 if header_row is None or signal_col is None: self.status_var.set(f"文件 {os.path.basename(file_path)} 未找到列名行，尝试手动查找...") header_row, signal_col = self.manual_find_header_row(file_path) if header_row is None or signal_col is None: self.status_var.set(f"文件 {os.path.basename(file_path)} 无法确定列名行，已跳过") return found # 根据文件类型处理 if file_path.lower().endswith((".xlsx", ".xlsm")): found = self.process_openpyxl_file(file_path, search_term, header_row, signal_col) elif file_path.lower().endswith(".xls"): found = self.process_xlrd_file(file_path, search_term, header_row, signal_col) except Exception as e: self.status_var.set(f"处理文件 {os.path.basename(file_path)} 出错: {str(e)}") return found def manual_find_header_row(self, file_path): """手动查找列名行（当自动查找失败时使用）""" try: # 尝试打开文件 if file_path.lower().endswith((".xlsx", ".xlsm")): wb = openpyxl.load_workbook(file_path, read_only=True, data_only=True) ws = wb.active # 扫描整个工作表（最多1000行） for row_idx in range(1, 1001): for col_idx in range(1, 101): try: cell = ws.cell(row=row_idx, column=col_idx) if cell.value and "データ" in str(cell.value): # 找到可能的列名行 return row_idx, col_idx except: continue wb.close() elif file_path.lower().endswith(".xls"): wb = xlrd.open_workbook(file_path) ws = wb.sheet_by_index(0) # 扫描整个工作表（最多1000行） for row_idx in range(0, 1000): for col_idx in range(0, 100): try: cell_value = ws.cell_value(row_idx, col_idx) if cell_value and "データ" in str(cell_value): # 找到可能的列名行 return row_idx, col_idx except: continue except: pass return None, None def process_openpyxl_file(self, file_path, search_term, header_row, signal_col): """高效处理xlsx/xlsm文件（增强异常处理）""" found = 0 try: # 使用缓存 file_key = f"{file_path}_{header_row}_{signal_col}" if file_key in self.file_cache: ws = self.file_cache[file_key] else: wb = openpyxl.load_workbook(file_path, data_only=True, read_only=True) ws = wb.active self.file_cache[file_key] = ws # 优化搜索范围 min_row = header_row + 1 max_row = min(ws.max_row, min_row + 5000) # 限制搜索行数 # 批量读取信号列数据 signal_values = [] for row in ws.iter_rows(min_row=min_row, max_row=max_row, min_col=signal_col, max_col=signal_col, values_only=True): signal_values.append(row[0]) # 搜索匹配的信号 for idx, value in enumerate(signal_values): if self.stop_requested: break if value is None: continue cell_str = str(value) if search_term.lower() in cell_str.lower(): signal = cell_str row_idx = min_row + idx # 提取整行内容 row_content = self.extract_row_content(ws, row_idx, header_row, max_cols=100) # 添加到结果集 if signal not in self.results: self.results[signal] = {} # 使用短文件名避免特殊字符问题 short_name = os.path.basename(file_path) self.results[signal][short_name] = row_content # 添加到表格（显示前50字符摘要） summary = row_content[:50] + "..." if len(row_content) > 50 else row_content self.tree.insert("", tk.END, values=(signal, short_name, summary)) found += 1 except Exception as e: self.status_var.set(f"处理文件 {os.path.basename(file_path)} 出错: {str(e)}") return found def process_xlrd_file(self, file_path, search_term, header_row, signal_col): """高效处理xls文件（增强异常处理）""" found = 0 try: # 使用缓存 file_key = f"{file_path}_{header_row}_{signal_col}" if file_key in self.file_cache: ws = self.file_cache[file_key] else: wb = xlrd.open_workbook(file_path) ws = wb.sheet_by_index(0) self.file_cache[file_key] = ws # 优化搜索范围 min_row = header_row + 1 max_row = min(ws.nrows, min_row + 5000) # 限制搜索行数 # 批量读取信号列数据 signal_values = ws.col_values(signal_col, min_row, max_row) # 搜索匹配的信号 for idx, value in enumerate(signal_values): if self.stop_requested: break if not value: continue cell_str = str(value) if search_term.lower() in cell_str.lower(): signal = cell_str row_idx = min_row + idx # 提取整行内容 row_content = self.extract_xlrd_row_content(ws, row_idx, header_row) # 添加到结果集 if signal not in self.results: self.results[signal] = {} # 使用短文件名避免特殊字符问题 short_name = os.path.basename(file_path) self.results[signal][short_name] = row_content # 添加到表格（显示前50字符摘要） summary = row_content[:50] + "..." if len(row_content) > 50 else row_content self.tree.insert("", tk.END, values=(signal, short_name, summary)) found += 1 except Exception as e: self.status_var.set(f"处理文件 {os.path.basename(file_path)} 出错: {str(e)}") return found def extract_xlrd_row_content(self, ws, row_idx, header_row): """为xls文件高效提取行内容""" content = [] try: row_values = ws.row_values(row_idx) except: return "" # 扩展到100列 for col_idx in range(min(len(row_values), 100)): try: cell_value = row_values[col_idx] if cell_value is not None and str(cell_value).strip() != '': # 使用列名缓存 col_key = f"{header_row}-{col_idx}" if col_key in self.column_cache: col_name = self.column_cache[col_key] else: try: col_name = ws.cell_value(header_row, col_idx) if not col_name: col_name = f"列{col_idx+1}" except: col_name = f"列{col_idx+1}" self.column_cache[col_key] = col_name content.append(f"{col_name}: {str(cell_value).strip()}") except: continue return "\n".join(content) def update_text_view(self): """更新文本对比视图""" # 清除现有文本区域 for widget in self.compare_container.winfo_children(): widget.destroy() if not self.results: return # 获取第一个信号作为默认显示 first_signal = next(iter(self.results.keys())) self.display_signal_comparison(first_signal) def on_table_select(self, event): """表格选择事件处理""" selected = self.tree.selection() if not selected: return item = self.tree.item(selected[0]) signal = item["values"][0] self.display_signal_comparison(signal) def display_signal_comparison(self, signal): """显示指定信号的对比""" # 清除现有文本区域 for widget in self.compare_container.winfo_children(): widget.destroy() if signal not in self.results: return signal_data = self.results[signal] files = list(signal_data.keys()) contents = list(signal_data.values()) # 创建列框架 for i, (file, content) in enumerate(zip(files, contents)): col_frame = ttk.Frame(self.compare_container) col_frame.grid(row=0, column=i, sticky="nsew", padx=5, pady=5) self.compare_container.columnconfigure(i, weight=1) # 文件名标签 file_label = ttk.Label(col_frame, text=file, font=("Arial", 10, "bold")) file_label.pack(fill=tk.X, pady=(0, 5)) # 文本区域 text_area = scrolledtext.ScrolledText(col_frame, wrap=tk.WORD, width=30, height=15) text_area.insert(tk.INSERT, content) text_area.configure(state="disabled") text_area.pack(fill=tk.BOTH, expand=True) # 保存引用 self.text_panes[file] = text_area def highlight_differences(self): """高亮显示文本差异""" if not self.text_panes: return # 获取所有行内容 all_contents = [] for text_area in self.text_panes.values(): text_area.configure(state="normal") text = text_area.get("1.0", tk.END).strip() text_area.configure(state="disabled") all_contents.append(text) # 如果所有内容相同，则不需要高亮 if len(set(all_contents)) == 1: self.status_var.set("所有文件行内容完全一致") return # 使用第一个文件作为基准 base_text = all_contents[0] # 对比并高亮差异 for i, (file, text_area) in enumerate(self.text_panes.items()): if i == 0: # 基准文件不需要处理 continue text_area.configure(state="normal") text_area.tag_configure("diff", background=self.highlight_color) # 清除之前的高亮 text_area.tag_remove("diff", "1.0", tk.END) # 获取当前文本 compare_text = text_area.get("1.0", tk.END).strip() # 使用序列匹配器查找差异 s = SequenceMatcher(None, base_text, compare_text) # 高亮差异部分 for tag in s.get_opcodes(): opcode = tag[0] start = tag[3] end = tag[4] if opcode != "equal": # 添加高亮标签 text_area.tag_add("diff", f"1.0+{start}c", f"1.0+{end}c") text_area.configure(state="disabled") self.status_var.set("差异已高亮显示") def choose_color(self): """选择高亮颜色""" color = askcolor(title="选择高亮颜色", initialcolor=self.highlight_color) if color[1]: self.highlight_color = color[1] self.color_btn.configure(bg=self.highlight_color) def export_report(self): """导出差异报告""" if not self.results: messagebox.showwarning("警告", "没有可导出的结果") return try: # 创建报告数据结构 report_data = [] for signal, files_data in self.results.items(): for file, content in files_data.items(): report_data.append({ "信号": signal, "文件": file, "行内容": content }) # 转换为DataFrame df = pd.DataFrame(report_data) # 保存到Excel save_path = filedialog.asksaveasfilename( defaultextension=".xlsx", filetypes=[("Excel文件", "*.xlsx")], title="保存差异报告" ) if save_path: df.to_excel(save_path, index=False) self.status_var.set(f"报告已保存到: {save_path}") except Exception as e: messagebox.showerror("错误", f"导出报告失败: {str(e)}") def clear_cache(self): """清除缓存""" try: for file in os.listdir(self.cache_dir): if file.endswith(".cache"): os.remove(os.path.join(self.cache_dir, file)) self.file_cache = {} self.column_cache = {} self.status_var.set("缓存已清除") except Exception as e: self.status_var.set(f"清除缓存失败: {str(e)}") def manual_column_select(self): """手动指定列名位置""" if not self.files: messagebox.showinfo("提示", "请先选择文件夹") return # 创建手动选择窗口 manual_window = tk.Toplevel(self.root) manual_window.title("手动指定列名位置") manual_window.geometry("400x300") # 文件选择 ttk.Label(manual_window, text="选择文件:").pack(pady=(10, 5)) file_var = tk.StringVar() file_combo = ttk.Combobox(manual_window, textvariable=file_var, values=[os.path.basename(f) for f in self.files]) file_combo.pack(fill=tk.X, padx=20, pady=5) file_combo.current(0) # 行号输入 ttk.Label(manual_window, text="列名行号:").pack(pady=(10, 5)) row_var = tk.StringVar(value="1") row_entry = ttk.Entry(manual_window, textvariable=row_var) row_entry.pack(fill=tk.X, padx=20, pady=5) # 列号输入 ttk.Label(manual_window, text="信号列号:").pack(pady=(10, 5)) col_var = tk.StringVar(value="1") col_entry = ttk.Entry(manual_window, textvariable=col_var) col_entry.pack(fill=tk.X, padx=20, pady=5) # 确认按钮 def confirm_selection(): try: file_idx = file_combo.current() file_path = self.files[file_idx] header_row = int(row_var.get()) signal_col = int(col_var.get()) # 保存到缓存 header_info = {"header_row": header_row, "signal_col": signal_col} self.save_header_cache(file_path, header_info) messagebox.showinfo("成功", f"已为 {os.path.basename(file_path)} 设置列名位置：行{header_row} 列{signal_col}") manual_window.destroy() except Exception as e: messagebox.showerror("错误", f"无效输入: {str(e)}") ttk.Button(manual_window, text="确认", command=confirm_selection).pack(pady=20) if __name__ == "__main__": root = tk.Tk() app = EnhancedSignalComparator(root) root.mainloop() 1、目前发现在处理大文件，且处理的行数不一致时，后一个文件会将前一个文件的缓存覆盖，导致两个文件的显示的结果都不对，比如A文件CGWHV:T，B文件MET(LO):T、MET(MID/HI):T，现在却变成了送信初期値：T 2、优化一下处理逻辑，现在对于大文件搜索还是太慢了，不需要处理额外的单元格，只需要找到信号值在的那一行，并且将那一行的内容与列名相对应起来。 3、在上述两个都完成的情况下，试着增加多信号处理的功能

1. 修改search_files函数，将search_term拆分成列表：search_terms = [term.strip() for term in self.search_text.get().split(';') if term.strip()] 2. 然后，在文件处理循环中，对每个文件，我们依次使用...
zip

langchain4j-anthropic-spring-boot-starter-0.31.0.jar中文文档.zip

1、压缩文件中包含： 中文文档、jar包下载地址、Maven依赖、Gradle依赖、源代码下载地址。 2、使用方法： 解压最外层zip，再解压其中的zip包，双击 【index.html】 文件，即可用浏览器打开、进行查看。 3、特殊说明： （1）本文档为人性化翻译，精心制作，请放心使用； （2）只翻译了该翻译的内容，如：注释、说明、描述、用法讲解 等； （3）不该翻译的内容保持原样，如：类名、方法名、包名、类型、关键字、代码 等。 4、温馨提示： （1）为了防止解压后路径太长导致浏览器无法打开，推荐在解压时选择“解压到当前文件夹”（放心，自带文件夹，文件不会散落一地）； （2）有时，一套Java组件会有多个jar，所以在下载前，请仔细阅读本篇描述，以确保这就是你需要的文件。 5、本文件关键字： jar中文文档.zip,java,jar包,Maven,第三方jar包,组件,开源组件,第三方组件,Gradle,中文API文档,手册,开发手册,使用手册,参考手册。
zip

TMS320F28335电机控制程序详解：BLDC、PMSM无感有感及异步VF源代码与开发资料

TMS320F28335这款高性能数字信号处理器（DSP）在电机控制领域的应用，涵盖了BLDC（无刷直流电机）、PMSM（永磁同步电机）的无感有感控制以及异步VF（变频调速）程序。文章不仅解释了各类型的电机控制原理，还提供了完整的开发资料，包括源代码、原理图和说明文档，帮助读者深入了解其工作原理和编程技巧。 适合人群：从事电机控制系统开发的技术人员，尤其是对TMS320F28335感兴趣的工程师。 使用场景及目标：适用于需要掌握TMS320F28335在不同电机控制应用场景下具体实现方法的专业人士，旨在提高他们对该微控制器的理解和实际操作能力。 其他说明：文中提供的开发资料为读者提供了从硬件到软件的全面支持，有助于加速项目开发进程并提升系统性能。
zip

基于爬山搜索法的风力发电MPPT控制Simulink仿真：定步长与变步长算法性能对比 - 爬山搜索法 最新版

基于爬山搜索法的风力发电最大功率点追踪（MPPT）控制的Simulink仿真模型，重点比较了定步长和变步长算法在不同风速条件下的表现。文中展示了两种算法的具体实现方法及其优缺点。定步长算法虽然结构简单、计算量小，但在风速突变时响应较慢，存在明显的稳态振荡。相比之下，变步长算法能够根据功率变化动态调整步长，表现出更快的响应速度和更高的精度，尤其在风速突变时优势明显。实验数据显示，变步长算法在风速从8m/s突增至10m/s的情况下，仅用0.3秒即可稳定，功率波动范围仅为±15W，而定步长算法则需要0.8秒，功率波动达到±35W。 适合人群：从事风力发电研究的技术人员、对MPPT控制感兴趣的工程技术人员以及相关专业的高校师生。 使用场景及目标：适用于风力发电系统的设计与优化，特别是需要提高系统响应速度和精度的场合。目标是在不同风速条件下，选择合适的MPPT算法以最大化风能利用率。 其他说明：文章还讨论了定步长算法在风速平稳情况下的优势，提出了根据不同应用场景灵活选择或组合使用这两种算法的建议。

大家在看

recommend-type

机械臂建模+MATLAB代码+六自由度.zip

机械臂建模+MATLAB代码+六自由度.zip
recommend-type

易语言WinSock模块应用

易语言WinSock模块应用源码,WinSock模块应用,启动,停止,监听,发送,接收,断开连接,取服务器端口,取服务器IP,取客户IP,取客户端口,异步选择,检查连接状态,连接,断开,关闭,创建,发送数据,接收数据,取本机名,取本机IP组,窗口1消息处理,客户进入,客户离开,数据到达
recommend-type

VxWorks和RTlinux的性能测试分析

VxWorks和RTlinux的性能测试分析
recommend-type

波特率任意设 串口调试助手

波特率任意设，很好用的串口调试助手
recommend-type

十几种水下图像增强算法源代码

水下增强方法代码，多种组合，matlab,传统方法

最新推荐

recommend-type

langchain4j-anthropic-spring-boot-starter-0.31.0.jar中文文档.zip

1、压缩文件中包含： 中文文档、jar包下载地址、Maven依赖、Gradle依赖、源代码下载地址。 2、使用方法： 解压最外层zip，再解压其中的zip包，双击 【index.html】 文件，即可用浏览器打开、进行查看。 3、特殊说明： （1）本文档为人性化翻译，精心制作，请放心使用； （2）只翻译了该翻译的内容，如：注释、说明、描述、用法讲解 等； （3）不该翻译的内容保持原样，如：类名、方法名、包名、类型、关键字、代码 等。 4、温馨提示： （1）为了防止解压后路径太长导致浏览器无法打开，推荐在解压时选择“解压到当前文件夹”（放心，自带文件夹，文件不会散落一地）； （2）有时，一套Java组件会有多个jar，所以在下载前，请仔细阅读本篇描述，以确保这就是你需要的文件。 5、本文件关键字： jar中文文档.zip,java,jar包,Maven,第三方jar包,组件,开源组件,第三方组件,Gradle,中文API文档,手册,开发手册,使用手册,参考手册。
recommend-type

TMS320F28335电机控制程序详解：BLDC、PMSM无感有感及异步VF源代码与开发资料

TMS320F28335这款高性能数字信号处理器（DSP）在电机控制领域的应用，涵盖了BLDC（无刷直流电机）、PMSM（永磁同步电机）的无感有感控制以及异步VF（变频调速）程序。文章不仅解释了各类型的电机控制原理，还提供了完整的开发资料，包括源代码、原理图和说明文档，帮助读者深入了解其工作原理和编程技巧。 适合人群：从事电机控制系统开发的技术人员，尤其是对TMS320F28335感兴趣的工程师。 使用场景及目标：适用于需要掌握TMS320F28335在不同电机控制应用场景下具体实现方法的专业人士，旨在提高他们对该微控制器的理解和实际操作能力。 其他说明：文中提供的开发资料为读者提供了从硬件到软件的全面支持，有助于加速项目开发进程并提升系统性能。
recommend-type

基于爬山搜索法的风力发电MPPT控制Simulink仿真：定步长与变步长算法性能对比 - 爬山搜索法 最新版

基于爬山搜索法的风力发电最大功率点追踪（MPPT）控制的Simulink仿真模型，重点比较了定步长和变步长算法在不同风速条件下的表现。文中展示了两种算法的具体实现方法及其优缺点。定步长算法虽然结构简单、计算量小，但在风速突变时响应较慢，存在明显的稳态振荡。相比之下，变步长算法能够根据功率变化动态调整步长，表现出更快的响应速度和更高的精度，尤其在风速突变时优势明显。实验数据显示，变步长算法在风速从8m/s突增至10m/s的情况下，仅用0.3秒即可稳定，功率波动范围仅为±15W，而定步长算法则需要0.8秒，功率波动达到±35W。 适合人群：从事风力发电研究的技术人员、对MPPT控制感兴趣的工程技术人员以及相关专业的高校师生。 使用场景及目标：适用于风力发电系统的设计与优化，特别是需要提高系统响应速度和精度的场合。目标是在不同风速条件下，选择合适的MPPT算法以最大化风能利用率。 其他说明：文章还讨论了定步长算法在风速平稳情况下的优势，提出了根据不同应用场景灵活选择或组合使用这两种算法的建议。
recommend-type

基于MatlabSimulink的风电场调频策略研究：虚拟惯性、超速减载与下垂控制的协调优化

内容概要：本文详细探讨了在Matlab/Simulink环境下，针对风电场调频的研究，尤其是双馈风机调频策略的应用及其与其他调频策略的协调工作。文中介绍了三种主要的调频策略——虚拟惯性、超速减载和下垂控制，并基于三机九节点系统进行了改进，模拟了四组风电机组的协同调频过程。研究指出，虚拟惯性的应用虽然可以提供短期频率支持，但也可能导致频率二次跌落的问题。因此，需要通过超速减载和下垂控制来进行补偿，以维持电网的稳定。此外，文章还展示了在变风速条件下，风电机组对电网频率支撑能力的显著提升，尤其是在高比例风电并网渗透的情况下，频率最低点提高了50%，验证了调频策略的有效性。 适合人群：从事电力系统、风电场运营管理和调频技术研发的专业人士，以及对风电调频感兴趣的科研人员和技术爱好者。 使用场景及目标：适用于希望深入理解风电场调频机制及其优化方法的人群。目标是掌握不同调频策略的工作原理及其协调工作的关键点，提高风电场的运行效率和稳定性。 其他说明：本文通过具体的案例研究和仿真数据，展示了调频策略的实际效果，强调了合理运用调频策略对于风电场稳定运行的重要意义。同时，也为未来的风电调频技术创新提供了理论依据和实践经验。
recommend-type

Visual C++.NET编程技术实战指南

根据提供的文件信息，可以生成以下知识点： ### Visual C++.NET编程技术体验 #### 第2章 定制窗口 - **设置窗口风格**：介绍了如何通过编程自定义窗口的外观和行为。包括改变窗口的标题栏、边框样式、大小和位置等。这通常涉及到Windows API中的`SetWindowLong`和`SetClassLong`函数。 - **创建六边形窗口**：展示了如何创建一个具有特殊形状边界的窗口，这类窗口不遵循标准的矩形形状。它需要使用`SetWindowRgn`函数设置窗口的区域。 - **创建异形窗口**：扩展了定制窗口的内容，提供了创建非标准形状窗口的方法。这可能需要创建一个不规则的窗口区域，并将其应用到窗口上。 #### 第3章 菜单和控制条高级应用 - **菜单编程**：讲解了如何创建和修改菜单项，处理用户与菜单的交互事件，以及动态地添加或删除菜单项。 - **工具栏编程**：阐述了如何使用工具栏，包括如何创建工具栏按钮、分配事件处理函数，并实现工具栏按钮的响应逻辑。 - **状态栏编程**：介绍了状态栏的创建、添加不同类型的指示器（如文本、进度条等）以及状态信息的显示更新。 - **为工具栏添加皮肤**：展示了如何为工具栏提供更加丰富的视觉效果，通常涉及到第三方的控件库或是自定义的绘图代码。 #### 第5章 系统编程 - **操作注册表**：解释了Windows注册表的结构和如何通过程序对其进行读写操作，这对于配置软件和管理软件设置非常关键。 - **系统托盘编程**：讲解了如何在系统托盘区域创建图标，并实现最小化到托盘、从托盘恢复窗口的功能。 - **鼠标钩子程序**：介绍了钩子（Hook）技术，特别是鼠标钩子，如何拦截和处理系统中的鼠标事件。 - **文件分割器**：提供了如何将文件分割成多个部分，并且能够重新组合文件的技术示例。 #### 第6章 多文档/多视图编程 - **单文档多视**：展示了如何在同一个文档中创建多个视图，这在文档编辑软件中非常常见。 #### 第7章 对话框高级应用 - **实现无模式对话框**：介绍了无模式对话框的概念及其应用场景，以及如何实现和管理无模式对话框。 - **使用模式属性表及向导属性表**：讲解了属性表的创建和使用方法，以及如何通过向导性质的对话框引导用户完成多步骤的任务。 - **鼠标敏感文字**：提供了如何实现点击文字触发特定事件的功能，这在阅读器和编辑器应用中很有用。 #### 第8章 GDI+图形编程 - **图像浏览器**：通过图像浏览器示例，展示了GDI+在图像处理和展示中的应用，包括图像的加载、显示以及基本的图像操作。 #### 第9章 多线程编程 - **使用全局变量通信**：介绍了在多线程环境下使用全局变量进行线程间通信的方法和注意事项。 - **使用Windows消息通信**：讲解了通过消息队列在不同线程间传递信息的技术，包括发送消息和处理消息。 - **使用CriticalSection对象**：阐述了如何使用临界区（CriticalSection）对象防止多个线程同时访问同一资源。 - **使用Mutex对象**：介绍了互斥锁（Mutex）的使用，用以同步线程对共享资源的访问，保证资源的安全。 - **使用Semaphore对象**：解释了信号量（Semaphore）对象的使用，它允许一个资源由指定数量的线程同时访问。 #### 第10章 DLL编程 - **创建和使用Win32 DLL**：介绍了如何创建和链接Win32动态链接库（DLL），以及如何在其他程序中使用这些DLL。 - **创建和使用MFC DLL**：详细说明了如何创建和使用基于MFC的动态链接库，适用于需要使用MFC类库的场景。 #### 第11章 ATL编程 - **简单的非属性化ATL项目**：讲解了ATL（Active Template Library）的基础使用方法，创建一个不使用属性化组件的简单项目。 - **使用ATL开发COM组件**：详细阐述了使用ATL开发COM组件的步骤，包括创建接口、实现类以及注册组件。 #### 第12章 STL编程 - **list编程**：介绍了STL（标准模板库）中的list容器的使用，讲解了如何使用list实现复杂数据结构的管理。 #### 第13章 网络编程 - **网上聊天应用程序**：提供了实现基本聊天功能的示例代码，包括客户端和服务器的通信逻辑。 - **简单的网页浏览器**：演示了如何创建一个简单的Web浏览器程序，涉及到网络通信和HTML解析。 - **ISAPI服务器扩展编程**：介绍了如何开发ISAPI（Internet Server API）服务器扩展来扩展IIS（Internet Information Services）的功能。 #### 第14章 数据库编程 - **ODBC数据库编程**：解释了ODBC（开放数据库互联）的概念，并提供了使用ODBC API进行数据库访问的示例。 - **ADO编程**：介绍了ADO（ActiveX Data Objects）技术，讲解了如何使用ADO进行数据库的增删改查等操作。 #### 第15章 HTML帮助系统编程 - **创建HTML帮助文件**：讲解了如何创建HTML帮助文件（.chm），这是微软帮助系统的一种形式。 - **区分上下文帮助**：提供了在应用程序中实现上下文敏感帮助的技术示例。 #### 第16章 安装和部署项目 - **用安装程序向导创建安装项目**：介绍了如何使用Visual Studio中的安装和部署项目向导，快速创建安装程序。 - **部署应用程序**：阐述了部署应用程序到目标系统的过程，以及如何确保应用程序的正确安装和运行。 以上内容涵盖了Visual C++.NET编程技术体验一书中涉及的各个技术点，并对每个章节中的示例做了详细的说明。这些知识点对于理解和掌握Visual C++.NET编程非常重要，并能够帮助读者在实际项目开发中应用这些技术。
recommend-type

HarmonyOS内核深度探秘：优化自由行旅游系统的策略

# 1. HarmonyOS内核概述与旅游系统需求分析 ## 1.1 HarmonyOS内核概述 HarmonyOS（鸿蒙操作系统）是由华为开发的一个分布式操作系统，旨在提供跨多种设备的无缝体验。其内核设计理念强调高效、安全和模块化，以微内核架构和分布式能力
recommend-type

tkinter模块所有控件

<think>好的，我现在需要回答用户关于Tkinter模块中所有控件列表的问题。首先，我需要回忆Tkinter的基础知识，确保自己了解各个控件的名称和用途。用户希望得到所有支持的控件列表，所以必须全面覆盖，不能遗漏。 首先，我应该从Tkinter的标准控件开始。常见的控件包括Label、Button、Entry这些基础部件。然后是Frame，用于布局，还有Canvas用于绘图。接下来是Checkbutton、Radiobutton，这些属于选择类控件。Listbox和Scrollbar通常一起使用，处理滚动内容。还有Scale（滑块）、Spinbox、Menu、Menubutton这些可能
recommend-type

局域网五子棋游戏：娱乐与聊天的完美结合

标题“网络五子棋”和描述“适合于局域网之间娱乐和聊天!”以及标签“五子棋 网络”所涉及的知识点主要围绕着五子棋游戏的网络版本及其在局域网中的应用。以下是详细的知识点： 1. 五子棋游戏概述： 五子棋是一种两人对弈的纯策略型棋类游戏，又称为连珠、五子连线等。游戏的目标是在一个15x15的棋盘上，通过先后放置黑白棋子，使得任意一方先形成连续五个同色棋子的一方获胜。五子棋的规则简单，但策略丰富，适合各年龄段的玩家。 2. 网络五子棋的意义： 网络五子棋是指可以在互联网或局域网中连接进行对弈的五子棋游戏版本。通过网络版本，玩家不必在同一地点即可进行游戏，突破了空间限制，满足了现代人们快节奏生活的需求，同时也为玩家们提供了与不同对手切磋交流的机会。 3. 局域网通信原理： 局域网（Local Area Network，LAN）是一种覆盖较小范围如家庭、学校、实验室或单一建筑内的计算机网络。它通过有线或无线的方式连接网络内的设备，允许用户共享资源如打印机和文件，以及进行游戏和通信。局域网内的计算机之间可以通过网络协议进行通信。 4. 网络五子棋的工作方式： 在局域网中玩五子棋，通常需要一个客户端程序（如五子棋.exe）和一个服务器程序。客户端负责显示游戏界面、接受用户输入、发送落子请求给服务器，而服务器负责维护游戏状态、处理玩家的游戏逻辑和落子请求。当一方玩家落子时，客户端将该信息发送到服务器，服务器确认无误后将更新后的棋盘状态传回给所有客户端，更新显示。 5. 五子棋.exe程序： 五子棋.exe是一个可执行程序，它使得用户可以在个人计算机上安装并运行五子棋游戏。该程序可能包含了游戏的图形界面、人工智能算法（如果支持单机对战AI的话）、网络通信模块以及游戏规则的实现。 6. put.wav文件： put.wav是一个声音文件，很可能用于在游戏进行时提供声音反馈，比如落子声。在网络环境中，声音文件可能被用于提升玩家的游戏体验，尤其是在局域网多人游戏场景中。当玩家落子时，系统会播放.wav文件中的声音，为游戏增添互动性和趣味性。 7. 网络五子棋的技术要求： 为了确保多人在线游戏的顺利进行，网络五子棋需要具备一些基本的技术要求，包括但不限于稳定的网络连接、高效的数据传输协议（如TCP/IP）、以及安全的数据加密措施（如果需要的话）。此外，还需要有一个良好的用户界面设计来提供直观和舒适的用户体验。 8. 社交与娱乐： 网络五子棋除了是一个娱乐游戏外，它还具有社交功能。玩家可以通过游戏内的聊天系统进行交流，分享经验和策略，甚至通过网络寻找新的朋友。这使得网络五子棋不仅是一个个人娱乐工具，同时也是一种社交活动。 总结来说，网络五子棋结合了五子棋游戏的传统魅力和现代网络技术，使得不同地区的玩家能够在局域网内进行娱乐和聊天，既丰富了人们的娱乐生活，又加强了人际交流。而实现这一切的基础在于客户端程序的设计、服务器端的稳定运行、局域网的高效通信，以及音效文件增强的游戏体验。
recommend-type

自由行旅游新篇章：HarmonyOS技术融合与系统架构深度解析

# 1. HarmonyOS技术概述 ## 1.1 HarmonyOS的起源与发展 HarmonyOS（鸿蒙操作系统）由华为公司开发，旨在构建全场景分布式OS，以应对不同设备间的互联问题。自从2019年首次发布以来，HarmonyOS迅速成长，并迅速应用于智能手机、平板、智能穿戴、车载设备等多种平台。该系
recommend-type

足底支撑相到达73%是什么问题

<think>好的，我现在要处理用户的问题：“足底支撑相达到73%可能的原因及影响”。首先，需要理解用户提到的“足底支撑相73%”具体指什么。足底支撑相通常指步态周期中脚与地面接触的时间占比，正常范围大约在60%-62%左右。如果达到73%，明显超出正常范围，可能意味着步态异常。 接下来，需要确定可能的原因。可能的原因包括生物力学异常，比如足弓异常（扁平足或高弓足）、踝关节活动度受限，或者肌肉力量不足，特别是小腿和足部肌肉。另外，神经系统疾病如脑卒中或帕金森病可能影响步态控制。骨骼关节问题如关节炎、髋膝关节病变也可能导致支撑时间延长。还有代偿性步态，比如因疼痛或受伤而改变步态模式。 然后是