# coding:utf-8 from base64 import b64encode from hashlib import md5, sha1 from hmac import new import json import re import sys from threading import Thread import time from urllib.parse import quote from pyaudio import paInt16, PyAudio import websocket from PyQt5.QtCore import Qt, QSize, QUrl, pyqtSlot, pyqtSignal from PyQt5.QtGui import QIcon, QFont, QColor from PyQt5.QtWidgets import QApplication, QWidget, QHBoxLayout, QVBoxLayout, QSizePolicy, QScrollArea from qfluentwidgets import (SmoothScrollArea, IconWidget, BodyLabel, CaptionLabel, TransparentToolButton, FluentIcon, ImageLabel, SimpleCardWidget, LineEdit, HeaderCardWidget, InfoBar, InfoBarPosition, InfoBarIcon, HyperlinkLabel, PrimaryPushButton, TitleLabel, setFont, ScrollArea, VerticalSeparator, MSFluentWindow) import ahocorasick import pandas as pd def isWin11(): return sys.platform == 'win32' and sys.getwindowsversion().build >= 22000 class StatisticsWidget(QWidget): """ Statistics widget """ def __init__(self, title: str, value: str, parent=None): super().__init__(parent=parent) self.titleLabel = CaptionLabel(title, self) self.valueLabel = BodyLabel(value, self) self.vBoxLayout = QVBoxLayout(self) self.vBoxLayout.setContentsMargins(16, 0, 16, 0) self.vBoxLayout.addWidget(self.valueLabel, 0, Qt.AlignTop) self.vBoxLayout.addWidget(self.titleLabel, 0, Qt.AlignBottom) setFont(self.valueLabel, 18, QFont.DemiBold) self.titleLabel.setTextColor(QColor(96, 96, 96), QColor(206, 206, 206)) class ConfigCard(HeaderCardWidget): """ Config card for API settings """ configChanged = pyqtSignal(str, str) # Signal for when config changes (app_id, api_key) def __init__(self, parent=None): super().__init__(parent) self.setTitle('API 配置') self.setBorderRadius(8) # Create layout self.mainLayout = QVBoxLayout() # App ID input self.appIdLayout = QHBoxLayout() self.appIdLabel = BodyLabel('App ID:', self) self.appIdInput = LineEdit(self) self.appIdInput.setPlaceholderText('输入 App ID') self.appIdLayout.addWidget(self.appIdLabel) self.appIdLayout.addWidget(self.appIdInput) # API Key input self.apiKeyLayout = QHBoxLayout() self.apiKeyLabel = BodyLabel('API Key:', self) self.apiKeyInput = LineEdit(self) self.apiKeyInput.setPlaceholderText('输入 API Key') self.apiKeyLayout.addWidget(self.apiKeyLabel) self.apiKeyLayout.addWidget(self.apiKeyInput) # Test connection button self.buttonLayout = QHBoxLayout() self.testButton = PrimaryPushButton('测试连接', self) self.saveButton = PrimaryPushButton('保存配置', self) self.buttonLayout.addWidget(self.testButton) self.buttonLayout.addWidget(self.saveButton) # Add all layouts to main layout self.mainLayout.addLayout(self.appIdLayout) self.mainLayout.addLayout(self.apiKeyLayout) self.mainLayout.addLayout(self.buttonLayout) # Set main layout self.viewLayout.addLayout(self.mainLayout) # Connect signals self.testButton.clicked.connect(self.test_connection) self.saveButton.clicked.connect(self.save_config) # Load existing config if any self.load_config() def load_config(self): """Load existing configuration if available""" # Here you might want to load from a config file # For now, we'll use default values self.appIdInput.setText("f3a9a1bc") self.apiKeyInput.setText("3ba5a497a68a930546fb95ef750abf90") def save_config(self): """Save the configuration""" app_id = self.appIdInput.text() api_key = self.apiKeyInput.text() if not app_id or not api_key: InfoBar.error( title='错误', content="App ID 和 API Key 不能为空", orient=Qt.Horizontal, isClosable=True, position=InfoBarPosition.TOP, duration=2000, parent=self ) return # Emit the config changed signal self.configChanged.emit(app_id, api_key) InfoBar.success( title='成功', content="配置已保存", orient=Qt.Horizontal, isClosable=True, position=InfoBarPosition.TOP, duration=2000, parent=self ) def test_connection(self): """Test the API connection""" app_id = self.appIdInput.text() api_key = self.apiKeyInput.text() if not app_id or not api_key: InfoBar.error( title='错误', content="请先填写 App ID 和 API Key", orient=Qt.Horizontal, isClosable=True, position=InfoBarPosition.TOP, duration=2000, parent=self ) return try: # Prepare connection parameters base_url = "ws://rtasr.xfyun.cn/v1/ws" ts = str(int(time.time())) tt = (app_id + ts).encode('utf-8') md5_ = md5() md5_.update(tt) baseString = md5_.hexdigest() baseString = bytes(baseString, encoding='utf-8') apiKey = api_key.encode('utf-8') signa = new(apiKey, baseString, sha1).digest() signa = b64encode(signa) signa = str(signa, 'utf-8') # Create WebSocket connection ws_url = f"{base_url}?appid={app_id}&ts={ts}&signa={quote(signa)}" ws = websocket.create_connection(ws_url) # Check the response result = ws.recv() ws.close() if "success" in result: InfoBar.success( title='成功', content="API连接测试成功", orient=Qt.Horizontal, isClosable=True, position=InfoBarPosition.TOP, duration=2000, parent=self ) else: InfoBar.error( title='错误', content="API连接测试失败", orient=Qt.Horizontal, isClosable=True, position=InfoBarPosition.TOP, duration=2000, parent=self ) except Exception as e: InfoBar.error( title='错误', content=f"连接测试失败: {str(e)}", orient=Qt.Horizontal, isClosable=True, position=InfoBarPosition.TOP, duration=2000, parent=self ) class AppInfoCard(SimpleCardWidget): """ App information card """ def __init__(self, parent=None): super().__init__(parent) self.iconLabel = ImageLabel(":/qfluentwidgets/images/logo.png", self) self.iconLabel.setBorderRadius(8, 8, 8, 8) self.iconLabel.scaledToWidth(120) self.nameLabel = TitleLabel('实时语音识别', self) self.installButton = PrimaryPushButton('开始', self) self.stopButton = PrimaryPushButton('停止', self) self.companyLabel = HyperlinkLabel( QUrl('https://blog.csdn.net/fjh1997/article/details/104449110'), '电脑音频设置教程', self) self.installButton.setFixedWidth(160) self.stopButton.setFixedWidth(160) self.scoreWidget = StatisticsWidget('版本', '1.0', self) self.separator = VerticalSeparator(self) self.descriptionLabel = BodyLabel( '一个实时语音识别工具，能够实时识别语言内容，并提供帮助提示。', self) self.descriptionLabel.setWordWrap(True) self.shareButton = TransparentToolButton(FluentIcon.ROBOT, self) self.shareButton.setFixedSize(32, 32) self.shareButton.setIconSize(QSize(14, 14)) self.hBoxLayout = QHBoxLayout(self) self.vBoxLayout = QVBoxLayout() self.topLayout = QHBoxLayout() self.statisticsLayout = QHBoxLayout() self.buttonLayout = QHBoxLayout() self.initLayout() self.setBorderRadius(8) def initLayout(self): self.hBoxLayout.setSpacing(30) self.hBoxLayout.setContentsMargins(34, 24, 24, 24) self.hBoxLayout.addWidget(self.iconLabel) self.hBoxLayout.addLayout(self.vBoxLayout) self.vBoxLayout.setContentsMargins(0, 0, 0, 0) self.vBoxLayout.setSpacing(0) # name label and install button self.vBoxLayout.addLayout(self.topLayout) self.topLayout.setContentsMargins(0, 0, 0, 0) self.topLayout.addWidget(self.nameLabel) self.topLayout.addWidget(self.installButton, 0, Qt.AlignRight) self.topLayout.addWidget(self.stopButton, 0, Qt.AlignRight) # company label self.vBoxLayout.addSpacing(3) self.vBoxLayout.addWidget(self.companyLabel) # statistics widgets self.vBoxLayout.addSpacing(20) self.vBoxLayout.addLayout(self.statisticsLayout) self.statisticsLayout.setContentsMargins(0, 0, 0, 0) self.statisticsLayout.setSpacing(10) self.statisticsLayout.addWidget(self.scoreWidget) self.statisticsLayout.addWidget(self.separator) self.statisticsLayout.setAlignment(Qt.AlignLeft) # description label self.vBoxLayout.addSpacing(20) self.vBoxLayout.addWidget(self.descriptionLabel) # button self.vBoxLayout.addSpacing(12) self.buttonLayout.setContentsMargins(0, 0, 0, 0) self.vBoxLayout.addLayout(self.buttonLayout) self.buttonLayout.addWidget(self.shareButton, 0, Qt.AlignRight) class InnerAudioCard(HeaderCardWidget): """ Inner audio card """ # 定义自定义信号 textChanged = pyqtSignal(str) def __init__(self, parent=None): super().__init__(parent) self.descriptionLabel = BodyLabel(self) self.descriptionLabel.setWordWrap(True) self.viewLayout.addWidget(self.descriptionLabel) self.setTitle('音频识别结果') self.setBorderRadius(8) self.init_asr() def init_asr(self): self.app_id = "f3a9a1bc" self.api_key = "3ba5a497a68a930546fb95ef750abf90" self.base_url = "ws://rtasr.xfyun.cn/v1/ws" self.ts = str(int(time.time())) tt = (self.app_id + self.ts).encode('utf-8') md5_ = md5() md5_.update(tt) baseString = md5_.hexdigest() baseString = bytes(baseString, encoding='utf-8') apiKey = self.api_key.encode('utf-8') signa = new(apiKey, baseString, sha1).digest() signa = b64encode(signa) self.signa = str(signa, 'utf-8') self.end_tag = "{\"end\": true}" self.ws = None self.p = None self.stream = None self.tsend = None self.trecv = None self.running = False def findInternalRecordingDevice(self, p): target = 'CABLE Output' for i in range(p.get_device_count()): devInfo = p.get_device_info_by_index(i) if devInfo['name'].find(target) >= 0 and devInfo['hostApi'] == 0: return i print('无法找到内录设备!') return -1 def start_asr(self): if not self.running: self.running = True # 创建WebSocket连接 self.ws = websocket.create_connection( self.base_url + "?appid=" + self.app_id + "&ts=" + self.ts + "&signa=" + quote(self.signa)) # 启动发送和接收线程 self.tsend = Thread(target=self.send) self.tsend.start() self.trecv = Thread(target=self.recv) self.trecv.start() def stop_asr(self): if self.running: self.running = False try: # 发送结束标签 if self.ws and self.ws.connected: self.ws.send(self.end_tag.encode('utf-8')) print("send end tag success") # 停止并关闭音频流 if self.stream: self.stream.stop_stream() self.stream.close() self.stream = None if self.p: self.p.terminate() self.p = None # 等待线程结束 if self.tsend and self.tsend.is_alive(): self.tsend.join(timeout=2) # 给2秒钟超时 if self.trecv and self.trecv.is_alive(): self.trecv.join(timeout=2) # 给2秒钟超时 # 关闭WebSocket连接 if self.ws: self.ws.close() self.ws = None print("ASR stopped successfully") except Exception as e: print(f"Error stopping ASR: {str(e)}") # 确保资源被释放 self.stream = None self.p = None self.ws = None def send(self): chunk_size = 1280 audio_format = paInt16 channels = 1 rate = 16000 self.p = PyAudio() dev_idx = self.findInternalRecordingDevice(self.p) if dev_idx < 0: return self.stream = self.p.open(input_device_index=dev_idx, format=audio_format, channels=channels, rate=rate, input=True, frames_per_buffer=chunk_size) try: while self.running and self.ws and self.ws.connected: data = self.stream.read(chunk_size) self.ws.send(data, opcode=websocket.ABNF.OPCODE_BINARY) time.sleep(0.04) except Exception as e: print(f"Send thread error: {e}") finally: if self.stream: self.stream.stop_stream() self.stream.close() if self.p: self.p.terminate() def recv(self): try: while self.running and self.ws and self.ws.connected: result = self.ws.recv() if len(result) == 0: print("receive result end") break result_dict = json.loads(result) if result_dict["action"] == "started": print("handshake success, result: " + result) if result_dict["action"] == "result": pattern = r'"w":"(.*?)"' extracted_words = ''.join(re.findall(pattern, result_dict["data"])) self.updateLabel(extracted_words) if result_dict["action"] == "error": print("rtasr error: " + result) self.ws.close() self.running = False return except Exception as e: print(f"Recv thread error: {e}") def updateLabel(self, text): pre = self.descriptionLabel.text() pre_list = pre.split("\n") if len(text) > len(pre_list[-1]): pre_list[-1] = text else: pre_list.append(text) new_text = "\n".join(pre_list) self.descriptionLabel.setText(new_text) # 发射信号，传递新的文本内容 self.textChanged.emit(new_text) class OuterAudioCard(HeaderCardWidget): """ Outer audio card """ def __init__(self, parent=None): super().__init__(parent) self.descriptionLabel = BodyLabel(self) self.descriptionLabel.setWordWrap(True) self.viewLayout.addWidget(self.descriptionLabel) self.setTitle('帮助提示') self.setBorderRadius(8) df = pd.read_csv('tips.csv') self.automaton = ahocorasick.Automaton() for index, row in df.iterrows(): keyword = row['关键字'] self.automaton.add_word(keyword, (index, keyword, row['解决方案'])) self.automaton.make_automaton() def updateLabel(self, text): text = text.split("\n")[-1] results = set() for end_index, (index, keyword, solution) in self.automaton.iter(text): results.add((keyword, solution)) print(results) if len(results) != 0: content = [] for keyword, solution in results: content.append(f"关键字: {keyword}, 解决方案: {solution}") pre = self.descriptionLabel.text() content_str = pre + "\n" + "\n".join(content) self.descriptionLabel.setText(content_str) class SystemRequirementCard(HeaderCardWidget): """ System requirements card """ def __init__(self, parent=None): super().__init__(parent) self.setTitle('系统要求') self.setBorderRadius(8) self.infoLabel = BodyLabel('此产品适用于你的设备。具有复选标记的项目符合开发人员的系统要求。', self) self.successIcon = IconWidget(InfoBarIcon.SUCCESS, self) self.detailButton = HyperlinkLabel('详细信息', self) self.vBoxLayout = QVBoxLayout() self.hBoxLayout = QHBoxLayout() self.successIcon.setFixedSize(16, 16) self.hBoxLayout.setSpacing(10) self.vBoxLayout.setSpacing(16) self.hBoxLayout.setContentsMargins(0, 0, 0, 0) self.vBoxLayout.setContentsMargins(0, 0, 0, 0) self.hBoxLayout.addWidget(self.successIcon) self.hBoxLayout.addWidget(self.infoLabel) self.vBoxLayout.addLayout(self.hBoxLayout) self.vBoxLayout.addWidget(self.detailButton) self.viewLayout.addLayout(self.vBoxLayout) class ConfigInterface(ScrollArea): """ Configuration interface """ def __init__(self, parent=None): super().__init__(parent=parent) self.view = QWidget(self) self.vBoxLayout = QVBoxLayout(self.view) # Create config card self.configCard = ConfigCard(self) # Setup layout self.vBoxLayout.setSpacing(30) self.vBoxLayout.setContentsMargins(36, 20, 36, 36) self.vBoxLayout.addWidget(self.configCard) self.vBoxLayout.addStretch(1) # Setup scroll area self.setWidget(self.view) self.setWidgetResizable(True) self.setObjectName("configInterface") self.enableTransparentBackground() class AppInterface(ScrollArea): def __init__(self, parent=None): super().__init__(parent) self.view = QWidget(self) self.vBoxLayout = QVBoxLayout(self.view) # Create all cards (removed config card) self.appCard = AppInfoCard(self) self.systemCard = SystemRequirementCard(self) # Create scroll areas for audio cards self.innerAudioScrollArea = ScrollArea(self) self.outerAudioScrollArea = ScrollArea(self) self.innerAudioCard = InnerAudioCard(self) self.outerAudioCard = OuterAudioCard(self) # Setup scroll areas self.innerAudioScrollArea.setWidget(self.innerAudioCard) self.outerAudioScrollArea.setWidget(self.outerAudioCard) self.innerAudioScrollArea.setWidgetResizable(True) self.outerAudioScrollArea.setWidgetResizable(True) self.innerAudioScrollArea.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed) self.outerAudioScrollArea.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed) self.innerAudioScrollArea.setFixedHeight(200) self.outerAudioScrollArea.setFixedHeight(200) # Create horizontal layout for audio cards self.hBoxLayout = QHBoxLayout() self.hBoxLayout.addWidget(self.innerAudioScrollArea, 1) self.hBoxLayout.addWidget(self.outerAudioScrollArea, 1) # Setup main layout self.vBoxLayout.setSpacing(10) self.vBoxLayout.setContentsMargins(0, 0, 10, 30) self.vBoxLayout.addWidget(self.appCard, 0, Qt.AlignTop) self.vBoxLayout.addLayout(self.hBoxLayout) self.vBoxLayout.addWidget(self.systemCard, 0, Qt.AlignTop) self.setWidget(self.view) self.setWidgetResizable(True) self.setObjectName("appInterface") # Connect signals self.appCard.installButton.clicked.connect(self.start_asr) self.appCard.stopButton.clicked.connect(self.stop_asr) self.innerAudioCard.textChanged.connect(self.onTextChanged) self.enableTransparentBackground() def start_asr(self): self.innerAudioCard.start_asr() def stop_asr(self): self.innerAudioCard.stop_asr() def resizeEvent(self, e): super().resizeEvent(e) @pyqtSlot(str) def onTextChanged(self, text): self.outerAudioCard.updateLabel(text) print(f"QLabel 内容已更新为: {text}") def closeEvent(self, event): # 确保所有子组件的线程被正确关闭 self.innerAudioCard.stop_asr() event.accept() class Demo3(MSFluentWindow): def __init__(self, parent=None): super().__init__(parent) # 创建界面 self.appInterface = AppInterface(self) self.configInterface = ConfigInterface(self) # 添加子界面到侧边栏，只使用基础图标 self.addSubInterface(self.appInterface, FluentIcon.HOME, "主页", FluentIcon.HOME, isTransparent=True) self.addSubInterface(self.configInterface, FluentIcon.SETTING, "配置", FluentIcon.SETTING, isTransparent=True) # 连接配置信号 self.configInterface.configCard.configChanged.connect(self.onConfigChanged) self.resize(880, 760) self.setWindowTitle('语音识别工具') self.setWindowIcon(QIcon(':/qfluentwidgets/images/logo.png')) self.titleBar.raise_() def onConfigChanged(self, app_id, api_key): """处理配置变更""" # 将配置更新传递给主页面的InnerAudioCard self.appInterface.innerAudioCard.app_id = app_id self.appInterface.innerAudioCard.api_key = api_key self.appInterface.innerAudioCard.init_asr() if __name__ == '__main__': # enable dpi scale QApplication.setHighDpiScaleFactorRoundingPolicy( Qt.HighDpiScaleFactorRoundingPolicy.PassThrough) QApplication.setAttribute(Qt.AA_EnableHighDpiScaling) QApplication.setAttribute(Qt.AA_UseHighDpiPixmaps) app = QApplication(sys.argv) w3 = Demo3() w3.show() app.exec_()