mttu-developers/konabot
9
2
Fork 4
Code Issues 13 Pull Requests Actions Packages Releases 79 Activity

我拿 AI 改坏枪代码！

Browse Source
This commit is contained in:
钟晓帕
2025-11-18 23:55:31 +08:00
parent f21da657db
commit 0d540eea4c
16 changed files with 367 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
konabot/plugins/idiomgame/__init__.py
View File

@ -8,6 +8,7 @@ from typing import Optional
from loguru import logger
from nonebot import on_message
import nonebot
from nonebot.adapters import Event as BaseEvent
from nonebot.adapters.discord.event import MessageEvent as DiscordMessageEvent
from nonebot_plugin_alconna import (
@ -32,6 +33,9 @@ DATA_FILE_PATH = (
DATA_DIR / "idiom_banned.json"
)
# 创建全局数据库管理器实例
db_manager = DatabaseManager()
def load_banned_ids() -> list[str]:
if not DATA_FILE_PATH.exists():
return []
@ -61,6 +65,15 @@ def remove_banned_id(group_id: str):
DATA_FILE_PATH.write_text(json.dumps(banned_ids, ensure_ascii=False, indent=4), "utf-8")
driver = nonebot.get_driver()
@driver.on_startup
async def register_startup_hook():
"""注册启动时需要执行的函数"""
await IdiomGame.init_lexicon()
class TryStartState(Enum):
STARTED = 0
ALREADY_PLAYING = 1
@ -98,7 +111,7 @@ class IdiomGameLLM:
@classmethod
async def storage_idiom(cls, idiom: str):
# 将 idiom 存入数据库
DatabaseManager.execute_by_sql_file(
await db_manager.execute_by_sql_file(
ROOT_PATH / "sql" / "insert_custom_word.sql",
(idiom,)
)
@ -130,11 +143,11 @@ class IdiomGame:
IdiomGame.INSTANCE_LIST[group_id] = self
@classmethod
def append_into_word_list(cls, word: str):
async def append_into_word_list(cls, word: str):
'''
将一个新词加入到词语列表中
'''
DatabaseManager.execute_by_sql_file(
await db_manager.execute_by_sql_file(
ROOT_PATH / "sql" / "insert_custom_word.sql",
(word,)
)
@ -149,26 +162,27 @@ class IdiomGame:
return False
@staticmethod
def random_idiom() -> str:
return DatabaseManager.query_by_sql_file(
async def random_idiom() -> str:
result = await db_manager.query_by_sql_file(
ROOT_PATH / "sql" / "random_choose_idiom.sql"
)[0]["idiom"]
)
return result[0]["idiom"]
def choose_start_idiom(self) -> str:
async def choose_start_idiom(self) -> str:
"""
随机选择一个成语作为起始成语
"""
self.last_idiom = IdiomGame.random_idiom()
self.last_idiom = await IdiomGame.random_idiom()
self.last_char = self.last_idiom[-1]
if not self.is_nextable(self.last_char):
self.choose_start_idiom()
if not await self.is_nextable(self.last_char):
await self.choose_start_idiom()
else:
self.add_history_idiom(self.last_idiom, new_chain=True)
return self.last_idiom
@classmethod
def try_start_game(cls, group_id: str, force: bool = False) -> TryStartState:
cls.init_lexicon()
async def try_start_game(cls, group_id: str, force: bool = False) -> TryStartState:
await cls.init_lexicon()
if not cls.INSTANCE_LIST.get(group_id):
cls(group_id)
instance = cls.INSTANCE_LIST[group_id]
@ -179,10 +193,10 @@ class IdiomGame:
instance.now_playing = True
return TryStartState.STARTED
def start_game(self, rounds: int = 100):
async def start_game(self, rounds: int = 100):
self.now_playing = True
self.remain_rounds = rounds
self.choose_start_idiom()
await self.choose_start_idiom()
@classmethod
def try_stop_game(cls, group_id: str) -> TryStopState:
@ -212,20 +226,20 @@ class IdiomGame:
跳过当前成语，选择下一个成语
"""
async with self.lock:
self._skip_idiom_async()
await self._skip_idiom_async()
self.add_buff_score(buff_score)
return self.last_idiom
def _skip_idiom_async(self) -> str:
self.last_idiom = IdiomGame.random_idiom()
async def _skip_idiom_async(self) -> str:
self.last_idiom = await IdiomGame.random_idiom()
self.last_char = self.last_idiom[-1]
if not self.is_nextable(self.last_char):
self._skip_idiom_async()
if not await self.is_nextable(self.last_char):
await self._skip_idiom_async()
else:
self.add_history_idiom(self.last_idiom, new_chain=True)
return self.last_idiom
async def try_verify_idiom(self, idiom: str, user_id: str) -> TryVerifyState:
async def try_verify_idiom(self, idiom: str, user_id: str) -> list[TryVerifyState]:
"""
用户发送成语
"""
@ -233,14 +247,15 @@ class IdiomGame:
state = await self._verify_idiom(idiom, user_id)
return state
def is_nextable(self, last_char: str) -> bool:
async def is_nextable(self, last_char: str) -> bool:
"""
判断是否有成语可以接
"""
return DatabaseManager.query_by_sql_file(
result = await db_manager.query_by_sql_file(
ROOT_PATH / "sql" / "is_nextable.sql",
(last_char,)
)[0]["DEED"] == 1
)
return result[0]["DEED"] == 1
def add_already_idiom(self, idiom: str):
if idiom in self.already_idioms:
@ -272,11 +287,12 @@ class IdiomGame:
state.append(TryVerifyState.WRONG_FIRST_CHAR)
return state
# 成语是否存在
result = DatabaseManager.query_by_sql_file(
result = await db_manager.query_by_sql_file(
ROOT_PATH / "sql" / "query_idiom.sql",
(idiom, idiom, idiom)
)[0]["status"]
if result == -1:
)
status_result = result[0]["status"]
if status_result == -1:
logger.info(f"用户 {user_id} 发送了未知词语 {idiom}，正在使用 LLM 进行验证")
try:
if not await IdiomGameLLM.verify_idiom_with_llm(idiom):
@ -298,16 +314,16 @@ class IdiomGame:
self.last_idiom = idiom
self.last_char = idiom[-1]
self.add_score(user_id, 1 * score_k) # 先加 1 分
if result == 1:
if status_result == 1:
state.append(TryVerifyState.VERIFIED_AND_REAL)
self.add_score(user_id, 4 * score_k) # 再加 4 分
self.remain_rounds -= 1
if self.remain_rounds <= 0:
self.now_playing = False
state.append(TryVerifyState.GAME_END)
if not self.is_nextable(self.last_char):
if not await self.is_nextable(self.last_char):
# 没有成语可以接了，自动跳过
self._skip_idiom_async()
await self._skip_idiom_async()
self.add_buff_score(-100)
state.append(TryVerifyState.BUT_NO_NEXT)
return state
@ -334,9 +350,9 @@ class IdiomGame:
return self.last_char
@classmethod
def random_idiom_starting_with(cls, first_char: str) -> Optional[str]:
cls.init_lexicon()
result = DatabaseManager.query_by_sql_file(
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,)
)
@ -345,10 +361,10 @@ class IdiomGame:
return result[0]["idiom"]
@classmethod
def init_lexicon(cls):
async def init_lexicon(cls):
if cls.__inited:
return
DatabaseManager.execute_by_sql_file(
await db_manager.execute_by_sql_file(
ROOT_PATH / "sql" / "create_table.sql"
) # 确保数据库初始化
cls.__inited = True
@ -417,7 +433,7 @@ class IdiomGame:
ALL_IDIOMS = [idiom["word"] for idiom in ALL_IDIOMS_INFOS] + THUOCL_IDIOMS
ALL_IDIOMS = list(set(ALL_IDIOMS)) # 去重
# 批量插入数据库
DatabaseManager.execute_many_values_by_sql_file(
await db_manager.execute_many_values_by_sql_file(
ROOT_PATH / "sql" / "insert_idiom.sql",
[(idiom,) for idiom in ALL_IDIOMS]
)
@ -430,13 +446,13 @@ class IdiomGame:
+ COMMON_WORDS
)
# 插入数据库
DatabaseManager.execute_many_values_by_sql_file(
await db_manager.execute_many_values_by_sql_file(
ROOT_PATH / "sql" / "insert_word.sql",
[(word,) for word in ALL_WORDS]
)
# 自定义词语 LOCAL_LLM_WORDS 插入数据库，兼容用
DatabaseManager.execute_many_values_by_sql_file(
await db_manager.execute_many_values_by_sql_file(
ROOT_PATH / "sql" / "insert_custom_word.sql",
[(word,) for word in LOCAL_LLM_WORDS]
)
@ -483,7 +499,7 @@ async def play_game(
if rounds <= 0:
await evt.send(await UniMessage().text("干什么！你想玩负数局吗？").export())
return
state = IdiomGame.try_start_game(group_id, force)
state = await IdiomGame.try_start_game(group_id, force)
if state == TryStartState.ALREADY_PLAYING:
await evt.send(
await UniMessage()
@ -502,7 +518,7 @@ async def play_game(
.export()
)
instance = IdiomGame.INSTANCE_LIST[group_id]
instance.start_game(rounds)
await instance.start_game(rounds)
# 发布成语
await evt.send(
await UniMessage()
@ -595,7 +611,7 @@ async def _(target: DepLongTaskTarget):
instance = IdiomGame.INSTANCE_LIST.get(group_id)
if not instance or not instance.get_playing_state():
return
avaliable_idiom = IdiomGame.random_idiom_starting_with(instance.get_last_char())
avaliable_idiom = await IdiomGame.random_idiom_starting_with(instance.get_last_char())
# 发送哈哈狗图片
with open(ASSETS_PATH / "img" / "dog" / "haha_dog.jpg", "rb") as f:
img_data = f.read()