将成语接龙还原为内存存储,空调优化为部分内存存储且具有过期期限,避免频繁数据库查询
All checks were successful
continuous-integration/drone/push Build is passing

This commit is contained in:
2025-11-19 11:04:13 +08:00
parent 00bdb90e3c
commit 1233677eea
2 changed files with 191 additions and 67 deletions

View File

@ -117,18 +117,22 @@ class IdiomGameLLM:
@classmethod
async def storage_idiom(cls, idiom: str):
# 将 idiom 存入数据库
await db_manager.execute_by_sql_file(
ROOT_PATH / "sql" / "insert_custom_word.sql",
(idiom,)
)
# await db_manager.execute_by_sql_file(
# ROOT_PATH / "sql" / "insert_custom_word.sql",
# (idiom,)
# )
# 将 idiom 存入本地文件以备后续分析
with open(DATA_DIR / "idiom_llm_storage.txt", "a", encoding="utf-8") as f:
f.write(idiom + "\n")
IdiomGame.append_into_word_list(idiom)
class IdiomGame:
# ALL_WORDS = [] # 所有四字词语
# ALL_IDIOMS = [] # 所有成语
ALL_WORDS = [] # 所有四字词语
ALL_IDIOMS = [] # 所有成语
INSTANCE_LIST: dict[str, "IdiomGame"] = {} # 群号对应的游戏实例
# IDIOM_FIRST_CHAR = {} # 所有成语包括词语的首字字典
# AVALIABLE_IDIOM_FIRST_CHAR = {} # 真正有效的成语首字字典
IDIOM_FIRST_CHAR = {} # 所有成语包括词语的首字字典
AVALIABLE_IDIOM_FIRST_CHAR = {} # 真正有效的成语首字字典
__inited = False
@ -153,10 +157,15 @@ class IdiomGame:
'''
将一个新词加入到词语列表中
'''
await db_manager.execute_by_sql_file(
ROOT_PATH / "sql" / "insert_custom_word.sql",
(word,)
)
if word not in cls.ALL_WORDS:
cls.ALL_WORDS.append(word)
if word[0] not in cls.IDIOM_FIRST_CHAR:
cls.IDIOM_FIRST_CHAR[word[0]] = []
cls.IDIOM_FIRST_CHAR[word[0]].append(word)
# await db_manager.execute_by_sql_file(
# ROOT_PATH / "sql" / "insert_custom_word.sql",
# (word,)
# )
def be_able_to_play(self) -> bool:
if self.last_play_date != datetime.date.today():
@ -169,10 +178,11 @@ class IdiomGame:
@staticmethod
async def random_idiom() -> str:
result = await db_manager.query_by_sql_file(
ROOT_PATH / "sql" / "random_choose_idiom.sql"
)
return result[0]["idiom"]
# result = await db_manager.query_by_sql_file(
# ROOT_PATH / "sql" / "random_choose_idiom.sql"
# )
# return result[0]["idiom"]
return secrets.choice(IdiomGame.ALL_IDIOMS)
async def choose_start_idiom(self) -> str:
"""
@ -257,11 +267,12 @@ class IdiomGame:
"""
判断是否有成语可以接
"""
result = await db_manager.query_by_sql_file(
ROOT_PATH / "sql" / "is_nextable.sql",
(last_char,)
)
return result[0]["DEED"] == 1
# result = await db_manager.query_by_sql_file(
# ROOT_PATH / "sql" / "is_nextable.sql",
# (last_char,)
# )
# return result[0]["DEED"] == 1
return last_char in IdiomGame.AVALIABLE_IDIOM_FIRST_CHAR
def add_already_idiom(self, idiom: str):
if idiom in self.already_idioms:
@ -293,12 +304,13 @@ class IdiomGame:
state.append(TryVerifyState.WRONG_FIRST_CHAR)
return state
# 成语是否存在
result = await db_manager.query_by_sql_file(
ROOT_PATH / "sql" / "query_idiom.sql",
(idiom, idiom, idiom)
)
status_result = result[0]["status"]
if status_result == -1:
# result = await db_manager.query_by_sql_file(
# ROOT_PATH / "sql" / "query_idiom.sql",
# (idiom, idiom, idiom)
# )
# status_result = result[0]["status"]
# if status_result == -1:
if idiom not in IdiomGame.ALL_IDIOMS and idiom not in IdiomGame.ALL_WORDS:
logger.info(f"用户 {user_id} 发送了未知词语 {idiom},正在使用 LLM 进行验证")
try:
if not await IdiomGameLLM.verify_idiom_with_llm(idiom):
@ -320,7 +332,8 @@ class IdiomGame:
self.last_idiom = idiom
self.last_char = idiom[-1]
self.add_score(user_id, 1 * score_k) # 先加 1 分
if status_result == 1:
# if status_result == 1:
if idiom in IdiomGame.ALL_IDIOMS:
state.append(TryVerifyState.VERIFIED_AND_REAL)
self.add_score(user_id, 4 * score_k) # 再加 4 分
self.remain_rounds -= 1
@ -357,22 +370,26 @@ class IdiomGame:
@classmethod
async def random_idiom_starting_with(cls, first_char: str) -> Optional[str]:
# await cls.init_lexicon()
# result = await db_manager.query_by_sql_file(
# ROOT_PATH / "sql" / "query_idiom_start_with.sql",
# (first_char,)
# )
# if len(result) == 0:
# return None
# return result[0]["idiom"]
await cls.init_lexicon()
result = await db_manager.query_by_sql_file(
ROOT_PATH / "sql" / "query_idiom_start_with.sql",
(first_char,)
)
if len(result) == 0:
if first_char not in cls.AVALIABLE_IDIOM_FIRST_CHAR:
return None
return result[0]["idiom"]
return secrets.choice(cls.AVALIABLE_IDIOM_FIRST_CHAR[first_char])
@classmethod
async def init_lexicon(cls):
if cls.__inited:
return
await db_manager.execute_by_sql_file(
ROOT_PATH / "sql" / "create_table.sql"
) # 确保数据库初始化
# await db_manager.execute_by_sql_file(
# ROOT_PATH / "sql" / "create_table.sql"
# ) # 确保数据库初始化
cls.__inited = True
# 成语大表
@ -439,10 +456,10 @@ class IdiomGame:
ALL_IDIOMS = [idiom["word"] for idiom in ALL_IDIOMS_INFOS] + THUOCL_IDIOMS
ALL_IDIOMS = list(set(ALL_IDIOMS)) # 去重
# 批量插入数据库
await db_manager.execute_many_values_by_sql_file(
ROOT_PATH / "sql" / "insert_idiom.sql",
[(idiom,) for idiom in ALL_IDIOMS]
)
# await db_manager.execute_many_values_by_sql_file(
# ROOT_PATH / "sql" / "insert_idiom.sql",
# [(idiom,) for idiom in ALL_IDIOMS]
# )
# 其他四字词语表,仅表示可以有这个词
@ -451,29 +468,33 @@ class IdiomGame:
+ THUOCL_WORDS
+ COMMON_WORDS
)
cls.ALL_WORDS = ALL_WORDS + LOCAL_LLM_WORDS
cls.ALL_IDIOMS = ALL_IDIOMS
# 插入数据库
await db_manager.execute_many_values_by_sql_file(
ROOT_PATH / "sql" / "insert_word.sql",
[(word,) for word in ALL_WORDS]
)
# await db_manager.execute_many_values_by_sql_file(
# ROOT_PATH / "sql" / "insert_word.sql",
# [(word,) for word in ALL_WORDS]
# )
# 自定义词语 LOCAL_LLM_WORDS 插入数据库,兼容用
await db_manager.execute_many_values_by_sql_file(
ROOT_PATH / "sql" / "insert_custom_word.sql",
[(word,) for word in LOCAL_LLM_WORDS]
)
# await db_manager.execute_many_values_by_sql_file(
# ROOT_PATH / "sql" / "insert_custom_word.sql",
# [(word,) for word in LOCAL_LLM_WORDS]
# )
# # 根据成语大表,划分出成语首字字典
# for idiom in cls.ALL_IDIOMS + cls.ALL_WORDS:
# if idiom[0] not in cls.IDIOM_FIRST_CHAR:
# cls.IDIOM_FIRST_CHAR[idiom[0]] = []
# cls.IDIOM_FIRST_CHAR[idiom[0]].append(idiom)
# 根据成语大表,划分出成语首字字典
for idiom in cls.ALL_IDIOMS + cls.ALL_WORDS:
if idiom[0] not in cls.IDIOM_FIRST_CHAR:
cls.IDIOM_FIRST_CHAR[idiom[0]] = []
cls.IDIOM_FIRST_CHAR[idiom[0]].append(idiom)
# # 根据真正的成语大表,划分出有效成语首字字典
# for idiom in cls.ALL_IDIOMS:
# if idiom[0] not in cls.AVALIABLE_IDIOM_FIRST_CHAR:
# cls.AVALIABLE_IDIOM_FIRST_CHAR[idiom[0]] = []
# cls.AVALIABLE_IDIOM_FIRST_CHAR[idiom[0]].append(idiom)
# 根据真正的成语大表,划分出有效成语首字字典
for idiom in cls.ALL_IDIOMS:
if idiom[0] not in cls.AVALIABLE_IDIOM_FIRST_CHAR:
cls.AVALIABLE_IDIOM_FIRST_CHAR[idiom[0]] = []
cls.AVALIABLE_IDIOM_FIRST_CHAR[idiom[0]].append(idiom)
evt = on_alconna(