discord_peak_bot_gen2/app/classes/database.py

90 lines
3.0 KiB
Python
Raw Permalink Normal View History

2024-08-03 10:51:18 +00:00
from peewee import *
import os
import random
import logging
from .voicepeak import Narrator
# 環境変数からデータベースの設定を取得
db_name = os.getenv('POSTGRES_DB', 'mydatabase')
db_user = os.getenv('POSTGRES_USER', 'myuser')
db_password = os.getenv('POSTGRES_PASSWORD', 'mysecurepassword')
db_host = os.getenv('POSTGRES_HOST', 'db')
db_port = os.getenv('POSTGRES_PORT', '5432')
# データベースの設定
db = PostgresqlDatabase(db_name, user=db_user, password=db_password, host=db_host, port=db_port)
class BaseModel(Model):
class Meta:
database = db
class ReadingEntry(BaseModel):
word = CharField(unique=True)
reading = CharField()
class Meta:
table_name = 'reading_entries'
class UserSetting(BaseModel):
user_id = BigIntegerField(unique=True)
reading_speaker = CharField(default="SEKAI")
reading_speed = IntegerField(default=100)
reading_pitch = IntegerField(default=100)
reading_emotion_happy = IntegerField(default=0)
reading_emotion_sad = IntegerField(default=0)
reading_emotion_angry = IntegerField(default=0)
reading_emotion_fun = IntegerField(default=0)
class Meta:
table_name = 'user_setting'
def initialize_database():
"""データベースの初期化"""
if db.is_closed():
db.connect()
db.create_tables([ReadingEntry, UserSetting])
def get_all_words():
"""現在の辞書情報を辞書型配列で取得する"""
try:
words = ReadingEntry.select()
dictionary = {word.word: word.reading for word in words}
return dictionary
except Exception as e:
logging.error(f"辞書情報取得中にエラーが発生しました: {e}")
return {}
def get_user_setting(user_id):
"""ユーザーごとの読み上げ設定を取得し、存在しない場合は生成する"""
try:
user_setting = UserSetting.get_or_none(UserSetting.user_id == user_id)
if user_setting:
return user_setting
else:
new_user_setting = UserSetting(
user_id=user_id,
reading_speaker=Narrator.random().name,
reading_speed=random.randint(50, 200),
reading_pitch=random.randint(-300, 300),
reading_emotion_happy=random.randint(0, 100),
reading_emotion_sad=random.randint(0, 100),
reading_emotion_angry=random.randint(0, 100),
reading_emotion_fun=random.randint(0, 100)
)
new_user_setting.save()
return new_user_setting
except Exception as e:
logging.error(f"ユーザー設定取得中にエラーが発生しました: {e}")
def delete_word(word):
"""辞書から指定された単語を削除する"""
try:
query = ReadingEntry.get_or_none(ReadingEntry.word == word)
if query:
query.delete_instance()
return True
return False
except Exception as e:
logging.error(f"単語削除中にエラーが発生しました: {e}")
return False