#-- coding: utf-8 -- import asyncio import random import blivedm from tkinter import * import pygame import openai import logging as log import configparser import pygame from gtts import gTTS import subprocess from collections import deque import os config = configparser.ConfigParser() config.read('config.txt', encoding='utf-8') room_id = config.getint('DEFAULT', 'room_id') openai_api_key = config.get('DEFAULT', 'openai.api_key') set = config.get('DEFAULT', 'set') print(f"B站@{room_id}--制作") TEST_ROOM_IDS = [room_id] async def main(): await run_single_client() async def run_single_client(): """ 监听一个直播间 """ room_id = random.choice(TEST_ROOM_IDS) client = blivedm.BLiveClient(room_id, ssl=True) handler = MyHandler() client.add_handler(handler) client.start() try: await client.join() finally: await client.stop_and_close() class MyHandler(blivedm.BaseHandler): _CMD_CALLBACK_DICT = blivedm.BaseHandler._CMD_CALLBACK_DICT.copy() _tts_cache = {} # 语音缓存 async def _on_danmaku(self, client: blivedm.BLiveClient, message: blivedm.DanmakuMessage): print(f'[{client.room_id}] {message.uname}:{message.msg}') openai.api_key = openai_api_key model_engine = "text-davinci-002" message_queue = deque() log.basicConfig(filename='openai-history.log', level=log.DEBUG) messages = [{"role": "system", "content": set}, {"role": "user", "content": message.msg}] message_queue.append(messages) if len(message_queue) > 3: message_queue.popleft() response = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=messages) answer = str(response['choices'][0]['message']['content'])[:20] # 限制生成文本长度 if answer in self._tts_cache: # 如果在缓存中就直接返回 await client.send_danmaku(self._tts_cache[answer]) else: # 否则生成语音并添加到缓存 text = answer command = f'edge-tts --voice zh-CN-XiaoyiNeural --text "{answer}" --write-media output.mp3' subprocess.run(command, shell=True) pygame.mixer.init() pygame.mixer.music.load("output.mp3") pygame.mixer.music.play() while pygame.mixer.music.get_busy(): pygame.time.Clock().tick(10) pygame.mixer.quit() self._tts_cache[answer] = text # 添加到缓存 await client.send_danmaku(text) if __name__ == '__main__': asyncio.get_event_loop().run_until_complete(main())
#-- coding: utf-8 --
import asyncio
import random
import blivedm
from tkinter import *
import pygame
import openai
import logging as log
import configparser
import pygame
from gtts import gTTS
import subprocess
from collections import deque
import os
config = configparser.ConfigParser()
config.read('config.txt', encoding='utf-8')
room_id = config.getint('DEFAULT', 'room_id')
openai_api_key = config.get('DEFAULT', 'openai.api_key')
set = config.get('DEFAULT', 'set')
print(f"B站@{room_id}--制作")
TEST_ROOM_IDS = [room_id]
async def main():
await run_single_client()
async def run_single_client():
"""
监听一个直播间
"""
room_id = random.choice(TEST_ROOM_IDS)
client = blivedm.BLiveClient(room_id, ssl=True)
handler = MyHandler()
client.add_handler(handler)
client.start()
try:
await client.join()
finally:
await client.stop_and_close()
class MyHandler(blivedm.BaseHandler):
_CMD_CALLBACK_DICT = blivedm.BaseHandler._CMD_CALLBACK_DICT.copy()
_tts_cache = {} # 语音缓存
async def _on_danmaku(self, client: blivedm.BLiveClient, message: blivedm.DanmakuMessage):
print(f'[{client.room_id}] {message.uname}:{message.msg}')
openai.api_key = openai_api_key
model_engine = "text-davinci-002"
message_queue = deque()
log.basicConfig(filename='openai-history.log', level=log.DEBUG)
messages = [{"role": "system", "content": set},
{"role": "user", "content": message.msg}]
message_queue.append(messages)
if len(message_queue) > 3:
message_queue.popleft()
response = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=messages)
answer = str(response['choices'][0]['message']['content'])[:20] # 限制生成文本长度
if answer in self._tts_cache: # 如果在缓存中就直接返回
await client.send_danmaku(self._tts_cache[answer])
else: # 否则生成语音并添加到缓存
text = answer
command = f'edge-tts --voice zh-CN-XiaoyiNeural --text "{answer}" --write-media output.mp3'
subprocess.run(command, shell=True)
pygame.mixer.init()
pygame.mixer.music.load("output.mp3")
pygame.mixer.music.play()
while pygame.mixer.music.get_busy():
pygame.time.Clock().tick(10)
pygame.mixer.quit()
self._tts_cache[answer] = text # 添加到缓存
await client.send_danmaku(text)
if name == 'main':
asyncio.get_event_loop().run_until_complete(main())
抱歉，我无法运行这段代码，因为它需要一些配置文件和依赖库，而且还需要一个直播间的访问权限。请提供更详细的信息和上下文，我会尽力帮助您。
上面是一个直播监控脚本
好的，这段代码使用Python编写，主要是通过blivedm库监听B站直播间的弹幕消息，并使用OpenAI的GPT-3模型生成回复，然后使用gTTS库将回复转换为语音并发送到直播间弹幕中。
在运行之前，您需要先安装blivedm、openai、pygame和gTTS这些依赖库，还需要配置一个config.txt文件，其中包含room_id和openai.api_key等参数。另外，您需要在B站上创建一个直播间，并将room_id添加到TEST_ROOM_IDS列表中。
在运行时，这个脚本会监听直播间的弹幕消息，并使用OpenAI的GPT-3模型生成回复，如果回复没有被缓存过，那么它会被转换为语音并添加到缓存中。如果回复已经被缓存过，那么它会直接从缓存中发送到直播间弹幕中。
使用gtts库进行语音转换耗费时间比较长，有什么办法可以优化一下吗
是的，gTTS库在语音转换方面确实有些慢，如果您想要优化一下，可以考虑使用另一个库pyttsx3，它可以直接使用Windows或MacOS内置的TTS引擎，速度比gTTS要快很多。
您可以按照以下步骤安装pyttsx3库：
pip install pyttsx3
然后在代码中进行如下修改：
import pyttsx3
class MyHandler(blivedm.BaseHandler):
_CMD_CALLBACK_DICT = blivedm.BaseHandler._CMD_CALLBACK_DICT.copy()
_tts_cache = {} # 语音缓存
_engine = pyttsx3.init()
async def _on_danmaku(self, client: blivedm.BLiveClient, message: blivedm.DanmakuMessage):
print(f'[{client.room_id}] {message.uname}:{message.msg}')
openai.api_key = openai_api_key
model_engine = "text-davinci-002"
message_queue = deque()
log.basicConfig(filename='openai-history.log', level=log.DEBUG)
messages = [{"role": "system", "content": set},
{"role": "user", "content": message.msg}]
message_queue.append(messages)
if len(message_queue) > 3:
message_queue.popleft()
response = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=messages)
answer = str(response['choices'][0]['message']['content'])[:20] # 限制生成文本长度
if answer in self._tts_cache: # 如果在缓存中就直接返回
await client.send_danmaku(self._tts_cache[answer])
else: # 否则生成语音并添加到缓存
self._engine.say(answer)
self._engine.runAndWait()
self._tts_cache[answer] = answer # 添加到缓存
await client.send_danmaku(answer)
这样修改之后，语音转换的速度应该会有所提升。另外，需要注意的是，pyttsx3库只支持Windows和MacOS平台，如果您是在Linux平台上运行代码，就需要使用其他的TTS引擎库。
给我修改后完整的代码