159 lines
7.0 KiB
Python
159 lines
7.0 KiB
Python
|
import openai
|
||
|
|
import logging
|
||
|
|
import random
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
import asyncio
|
||
|
|
|
||
|
|
class IndependenceChat:
|
||
|
|
def __init__(self, bot):
|
||
|
|
"""IndependenceChatクラスの初期化を行う"""
|
||
|
|
self.bot = bot
|
||
|
|
openai.api_key = self.bot.config["openai_api_key"]
|
||
|
|
self.voice_client = None
|
||
|
|
self.independence_character_name = None
|
||
|
|
self.system = None
|
||
|
|
self.input_list = []
|
||
|
|
self.logs = []
|
||
|
|
self.independence_timers = []
|
||
|
|
self.log_file_path = "independence_chat_log.txt"
|
||
|
|
|
||
|
|
def enable(self, system_setting, voice_client, independence_character_name):
|
||
|
|
"""独立チャット機能を有効にする"""
|
||
|
|
self.system = {"role": "system", "content": system_setting}
|
||
|
|
self.input_list = [self.system]
|
||
|
|
self.voice_client = voice_client
|
||
|
|
self.independence_character_name = independence_character_name
|
||
|
|
self.log("独立チャットモードを有効にしました。")
|
||
|
|
|
||
|
|
def disable(self):
|
||
|
|
"""独立チャット機能を無効にする"""
|
||
|
|
self.voice_client = None
|
||
|
|
self.independence_character_name = None
|
||
|
|
self.system = None
|
||
|
|
self.input_list = []
|
||
|
|
self.log("独立チャットモードを無効にしました。")
|
||
|
|
|
||
|
|
async def input_message(self, input_text):
|
||
|
|
"""ユーザーからの入力メッセージを処理し、応答を生成する"""
|
||
|
|
self.input_list.append({"role": "user", "content": input_text})
|
||
|
|
self.logs.append(input_text)
|
||
|
|
self.log(input_text)
|
||
|
|
|
||
|
|
response = openai.ChatCompletion.create(
|
||
|
|
model="gpt-3.5-turbo",
|
||
|
|
messages=self.input_list
|
||
|
|
)
|
||
|
|
|
||
|
|
# 応答の最初の選択肢を取得し、メッセージ内容を取り出す
|
||
|
|
result = response.choices[0].message['content']
|
||
|
|
surgery_result = f"{self.independence_character_name}: {result}"
|
||
|
|
self.logs.append(surgery_result)
|
||
|
|
self.log(surgery_result)
|
||
|
|
|
||
|
|
self.input_list.append({"role": "assistant", "content": result})
|
||
|
|
|
||
|
|
logging.info("最新の3件の会話ログ:")
|
||
|
|
for log_entry in self.input_list[-3:]:
|
||
|
|
logging.info(log_entry)
|
||
|
|
|
||
|
|
await self.bot.message_reader.read_message(
|
||
|
|
result, self.bot.word_dictionary, self.bot.config["ffmpeg_executable_path"], self.bot.volume
|
||
|
|
)
|
||
|
|
|
||
|
|
async def start_independence_mode_check(self):
|
||
|
|
"""独立チャットモードでランダムにVCに参加する処理を開始"""
|
||
|
|
if self.independence_timers:
|
||
|
|
for timer in self.independence_timers:
|
||
|
|
timer.cancel()
|
||
|
|
self.independence_timers = []
|
||
|
|
|
||
|
|
# 24時間を時間帯によってランダムにチェックする時間を設定
|
||
|
|
now = datetime.now()
|
||
|
|
for hour in range(24):
|
||
|
|
if 18 <= hour or hour < 2: # 18時から2時の間
|
||
|
|
checks_per_hour = 2 # 1時間に2回チェック
|
||
|
|
else:
|
||
|
|
checks_per_hour = 1 # 2時間に1回チェック
|
||
|
|
|
||
|
|
for _ in range(checks_per_hour):
|
||
|
|
target_time = now.replace(hour=hour, minute=random.randint(0, 59), second=0, microsecond=0)
|
||
|
|
if target_time < now:
|
||
|
|
target_time += timedelta(days=1)
|
||
|
|
delay = (target_time - now).total_seconds()
|
||
|
|
timer = self.bot.loop.call_later(delay, self.bot.loop.create_task, self.check_independence_mode())
|
||
|
|
self.independence_timers.append(timer)
|
||
|
|
|
||
|
|
async def check_independence_mode(self):
|
||
|
|
"""独立チャットモードでランダムにVCに参加する"""
|
||
|
|
try:
|
||
|
|
if self.bot.voice_channel_manager.get_connected_channel(self.bot.guilds[0]) is not None:
|
||
|
|
logging.info("すでにVCに参加しています。自動参加をキャンセルします。")
|
||
|
|
self.log("VCに参加しようとしましたが、既に他のVCに参加していました。")
|
||
|
|
return
|
||
|
|
|
||
|
|
channel = random.choice(self.bot.independence_channel_ids)
|
||
|
|
voice_channel = self.bot.get_channel(channel)
|
||
|
|
if voice_channel is None or not hasattr(voice_channel, 'members'):
|
||
|
|
logging.warning("Voice channel not found or has no members attribute")
|
||
|
|
return
|
||
|
|
if len(voice_channel.members) == 0:
|
||
|
|
dummy_message = type('Dummy', (object,), {
|
||
|
|
"author": type('Author', (object,), {
|
||
|
|
"voice": type('VoiceState', (object,), {"channel": voice_channel})(),
|
||
|
|
"display_name": "IndependenceBot"
|
||
|
|
})(),
|
||
|
|
"channel": type('Channel', (object,), {"id": None})(),
|
||
|
|
"guild": voice_channel.guild
|
||
|
|
})()
|
||
|
|
await self.bot.command_handler.handle_ijoin(dummy_message, [])
|
||
|
|
self.bot.loop.create_task(self.leave_if_empty(voice_channel))
|
||
|
|
else:
|
||
|
|
logging.info("Voice channel has members")
|
||
|
|
self.log("VCに参加しようとしましたが、既に誰かが参加していました。")
|
||
|
|
finally:
|
||
|
|
await self.start_independence_mode_check() # タイマーを再設定
|
||
|
|
|
||
|
|
async def leave_if_empty(self, voice_channel):
|
||
|
|
"""一定時間誰も来なければVCを退出する"""
|
||
|
|
await asyncio.sleep(600) # 10分待つ
|
||
|
|
if len(voice_channel.members) == 0:
|
||
|
|
await self.bot.voice_channel_manager.set_vc_status(self.bot, None)
|
||
|
|
|
||
|
|
def start_debug_mode_check(self):
|
||
|
|
"""デバッグモードでランダムにVCに参加する処理を開始"""
|
||
|
|
if self.independence_timers:
|
||
|
|
for timer in self.independence_timers:
|
||
|
|
timer.cancel()
|
||
|
|
self.independence_timers = []
|
||
|
|
|
||
|
|
# 即時にVCに参加する
|
||
|
|
self.bot.loop.create_task(self.check_independence_mode())
|
||
|
|
|
||
|
|
async def generate_greeting(self, username):
|
||
|
|
"""ユーザーに対する挨拶メッセージを生成する"""
|
||
|
|
self.log(f"{username}が会話に参加しました。")
|
||
|
|
self.input_list.append({"role": "user", "content": f"{username}が会話に参加しました。挨拶メッセージを生成してください。"})
|
||
|
|
|
||
|
|
response = openai.ChatCompletion.create(
|
||
|
|
model="gpt-3.5-turbo",
|
||
|
|
messages=self.input_list
|
||
|
|
)
|
||
|
|
|
||
|
|
# 応答の最初の選択肢を取得し、メッセージ内容を取り出す
|
||
|
|
greeting = response.choices[0].message['content']
|
||
|
|
self.logs.append(f"{self.independence_character_name}: {greeting}")
|
||
|
|
|
||
|
|
self.input_list.append({"role": "assistant", "content": greeting})
|
||
|
|
self.log(greeting)
|
||
|
|
|
||
|
|
logging.info("最新の3件の会話ログ:")
|
||
|
|
for log_entry in self.input_list[-3:]:
|
||
|
|
logging.info(log_entry)
|
||
|
|
|
||
|
|
return greeting
|
||
|
|
|
||
|
|
def log(self, message):
|
||
|
|
"""ログメッセージをファイルに出力する"""
|
||
|
|
with open(self.log_file_path, "a", encoding="utf-8") as log_file:
|
||
|
|
log_file.write(f"{datetime.now()}: {message}\n")
|