456 lines
17 KiB
Python
456 lines
17 KiB
Python
import json
|
||
import random
|
||
import schedule
|
||
import time
|
||
from datetime import datetime
|
||
import pymysql
|
||
|
||
from game_task.game_chatgpt_qa import game_question_json, game_answer_json
|
||
|
||
# 数据库连接配置
|
||
db_config = {
|
||
'host': '192.168.2.32', # 替换为你的MariaDB服务器地址
|
||
'user': 'root', # 替换为你的MariaDB用户名
|
||
'password': 'lw123456', # 替换为你的MariaDB密码
|
||
'database': 'message_archive',
|
||
'charset': 'utf8mb4',
|
||
'cursorclass': pymysql.cursors.DictCursor
|
||
}
|
||
|
||
|
||
# 连接数据库
|
||
def get_db_connection():
|
||
return pymysql.connect(**db_config)
|
||
|
||
|
||
# 添加群聊
|
||
def add_group(group_id, player_id):
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
try:
|
||
cursor.execute(
|
||
"INSERT INTO t_encyclopedia_groups (group_id) VALUES (%s)",
|
||
(group_id,)
|
||
)
|
||
conn.commit()
|
||
return (
|
||
f"🎉 好消息来啦！\n"
|
||
f"🌟 群 {group_id} 已成功加入\n"
|
||
f"🎈 大家快来一起玩吧！"
|
||
)
|
||
except pymysql.err.IntegrityError:
|
||
return (
|
||
f"😄 嘿，小伙伴\n"
|
||
f"🌈 群 {group_id} 已经在了\n"
|
||
f"🎶 快来一起开心吧！"
|
||
)
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
|
||
# 获取所有群聊ID
|
||
def get_group_ids():
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
try:
|
||
cursor.execute("SELECT group_id FROM t_encyclopedia_groups")
|
||
return [row['group_id'] for row in cursor.fetchall()]
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
|
||
# 开始游戏（使用 player_id）
|
||
def start_game(group_id, player_id, player_name="未知玩家"):
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
try:
|
||
# 确保群聊存在
|
||
cursor.execute("SELECT group_id FROM t_encyclopedia_groups WHERE group_id = %s", (group_id,))
|
||
if not cursor.fetchone():
|
||
return (
|
||
f"😅 哎呀，小朋友\n"
|
||
f"🌼 群 {group_id} 还没准备好\n"
|
||
f"🎀 先用 /addgroup 添加哦！"
|
||
)
|
||
|
||
cursor.execute(
|
||
"INSERT INTO t_encyclopedia_players (player_id, group_id, player_name) VALUES (%s, %s, %s)",
|
||
(player_id, group_id, player_name)
|
||
)
|
||
conn.commit()
|
||
return (
|
||
f"🎉 欢迎加入，{player_name}！\n"
|
||
f"🌟 ID: {player_id} 已就位\n"
|
||
f"🎈 群 {group_id} 等你来玩\n"
|
||
f"🌈 快来接任务吧！"
|
||
)
|
||
except pymysql.err.IntegrityError:
|
||
cursor.execute(
|
||
"SELECT player_name FROM t_encyclopedia_players WHERE group_id = %s AND player_id = %s",
|
||
(group_id, player_id)
|
||
)
|
||
existing_name = cursor.fetchone()['player_name']
|
||
return (
|
||
f"😄 你好呀，{existing_name}\n"
|
||
f"🌼 ID: {player_id} 已加入群 {group_id}\n"
|
||
f"🎶 快来一起乐呵吧！"
|
||
)
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
|
||
# 随机分配任务
|
||
def assign_random_task(group_id):
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
try:
|
||
# 获取当前群聊的玩家列表
|
||
cursor.execute("SELECT player_id FROM t_encyclopedia_players WHERE group_id = %s", (group_id,))
|
||
players = [row['player_id'] for row in cursor.fetchall()]
|
||
if not players:
|
||
return (
|
||
f"😔 哎呀，群 {group_id} 好安静\n"
|
||
f"🌟 没有小伙伴加入\n"
|
||
f"🎈 快叫朋友来玩吧！"
|
||
)
|
||
|
||
# 使用内部方法获取任务
|
||
task = game_question_json("请从不同的学科领域(历史学：古代历史、近现代史、战争与政治、文化遗产与考古学；生物学：进化论、动植物学、基因与遗传学、生态学；化学：有机化学、无机化学、生物化学、环境化学；地理学：人文地理、自然地理、地质学、气候变化；文学：古典文学、现代文学、小说与戏剧、诗歌与散文；艺术：美术、音乐、戏剧与舞蹈、电影与媒体；哲学：西方哲学、东方哲学、道德与伦理学、逻辑与认识论；社会学：人类学、心理学、政治学、经济学；计算机科学：编程语言、人工智能、数据科学、网络与安全；工程学：机械工程、电气工程、化学工程、土木工程；医学：解剖学、生理学、临床医学、药学与护理；体育：球类运动、奥林匹克运动、运动心理学、健身与营养；宗教与神话：世界宗教、神话与民间故事、宗教哲学；语言学：语法与词汇、语言习得、方言与语言变异；经济学：宏观经济学、微观经济学、国际贸易、金融与投资；法律：民法与刑法、国际法、知识产权法、环境法)生成一个全新问题。")
|
||
|
||
question = task["question"]
|
||
answer = task["answer"]
|
||
score = int(task["score"])
|
||
description = task.get("description", "")
|
||
|
||
# 分配给随机玩家
|
||
holder_id = random.choice(players)
|
||
cursor.execute(
|
||
"INSERT INTO t_encyclopedia_active_tasks (group_id, question, answer, score, description, holder_id) VALUES (%s, %s, %s, %s, %s, %s)",
|
||
(group_id, question, answer, score, description, holder_id)
|
||
)
|
||
conn.commit()
|
||
|
||
# 获取分配的任务ID和持有者名称
|
||
cursor.execute(
|
||
"SELECT active_task_id FROM t_encyclopedia_active_tasks WHERE group_id = %s AND question = %s AND holder_id = %s ORDER BY assigned_at DESC LIMIT 1",
|
||
(group_id, question, holder_id)
|
||
)
|
||
active_task_id = cursor.fetchone()['active_task_id']
|
||
cursor.execute(
|
||
"SELECT player_name FROM t_encyclopedia_players WHERE group_id = %s AND player_id = %s",
|
||
(group_id, holder_id)
|
||
)
|
||
holder_name = cursor.fetchone()['player_name']
|
||
|
||
return (
|
||
f"🎁 新任务来啦！\n"
|
||
f"🌟 幸运玩家：{holder_name} (ID: {holder_id})\n"
|
||
f"🎈 问题：{question}\n"
|
||
f"🌈 群 {group_id} 等你抢答\n"
|
||
f"🎀 任务ID: task_{active_task_id}\n"
|
||
f"🌼 积分：{score}"
|
||
)
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
|
||
# 提交答案并计分（错误扣1分）
|
||
def submit_answer(group_id, player_id, task_id, answer):
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
try:
|
||
# 检查玩家是否存在并获取名称
|
||
cursor.execute("SELECT player_name FROM t_encyclopedia_players WHERE group_id = %s AND player_id = %s",
|
||
(group_id, player_id))
|
||
player_row = cursor.fetchone()
|
||
if not player_row:
|
||
return (
|
||
f"😅 嘿，小朋友\n"
|
||
f"🌟 ID: {player_id} 还没加入群 {group_id}\n"
|
||
f"🎈 用 /start 先报名吧！"
|
||
)
|
||
|
||
player_name = player_row['player_name']
|
||
|
||
# 检查任务ID格式并提取数字部分
|
||
if not task_id.startswith("task_") or not task_id[5:].isdigit():
|
||
return (
|
||
f"😅 哎呀，小调皮\n"
|
||
f"🌟 任务ID格式不对哦\n"
|
||
f"🎈 应该是 task_[数字]\n"
|
||
f"🌈 比如：task_1\n"
|
||
f"🎀 再试一次吧！"
|
||
)
|
||
|
||
active_task_id = int(task_id.split('_')[1])
|
||
|
||
# 检查任务是否存在并获取任务详情
|
||
cursor.execute(
|
||
"SELECT question, answer, score, holder_id, status FROM t_encyclopedia_active_tasks WHERE group_id = %s AND active_task_id = %s",
|
||
(group_id, active_task_id)
|
||
)
|
||
task_data = cursor.fetchone()
|
||
if not task_data:
|
||
return (
|
||
f"😔 哎呀，小伙伴\n"
|
||
f"🌼 群 {group_id} 的任务 {task_id} 不见了\n"
|
||
f"🎀 可能已经被抢答啦！"
|
||
)
|
||
|
||
if task_data['status'] == 'completed':
|
||
return (
|
||
f"😄 哈哈，太慢啦！\n"
|
||
f"🌟 任务 {task_id} 已完成\n"
|
||
f"🎈 群 {group_id} 等你抢新任务哦！"
|
||
)
|
||
|
||
question, correct_answer_db, top_score, holder_id, _ = task_data['question'], task_data['answer'].lower(), \
|
||
task_data['score'], task_data['holder_id'], task_data['status']
|
||
|
||
# 获取持有者名称
|
||
cursor.execute("SELECT player_name FROM t_encyclopedia_players WHERE group_id = %s AND player_id = %s",
|
||
(group_id, holder_id))
|
||
holder_name = cursor.fetchone()['player_name']
|
||
|
||
# 调用内部方法校验答案
|
||
answer_json = {"question": question, "top_score": str(top_score), "answer": answer}
|
||
result = game_answer_json(answer_json)
|
||
user_answer = answer.strip().lower()
|
||
points = int(result["score"])
|
||
correct_answer = result["answer"].lower()
|
||
description = result["description"]
|
||
is_correct = points > 0
|
||
|
||
# 记录历史并更新积分
|
||
if is_correct:
|
||
cursor.execute(
|
||
"UPDATE t_encyclopedia_players SET points = points + %s WHERE group_id = %s AND player_id = %s",
|
||
(points, group_id, player_id)
|
||
)
|
||
cursor.execute(
|
||
"UPDATE t_encyclopedia_active_tasks SET status = 'completed' WHERE group_id = %s AND active_task_id = %s",
|
||
(group_id, active_task_id)
|
||
)
|
||
message = (
|
||
f"🎉 {player_name} (ID: {player_id}) 太棒啦！\n"
|
||
f"🌟 任务：{question}\n"
|
||
f"🎈 答对了，真厉害！\n"
|
||
f"🌈 奖励：{points} 分\n"
|
||
f"🎀 提示：{description}"
|
||
if player_id == holder_id
|
||
else
|
||
f"🎉 {player_name} (ID: {player_id}) 抢答成功！\n"
|
||
f"🌟 任务：{question}\n"
|
||
f"🎈 原主：{holder_name} (ID: {holder_id})\n"
|
||
f"🌈 获得 {points} 分，太棒了！\n"
|
||
f"🎀 提示：{description}"
|
||
)
|
||
else:
|
||
# 错误扣1分，确保积分不低于0
|
||
cursor.execute(
|
||
"UPDATE t_encyclopedia_players SET points = GREATEST(points - 1, 0) WHERE group_id = %s AND player_id = %s",
|
||
(group_id, player_id)
|
||
)
|
||
points = -1 # 记录惩罚分
|
||
message = (
|
||
f"😅 {player_name} (ID: {player_id}) 有点小遗憾！\n"
|
||
f"🌼 任务：{question}\n"
|
||
f"🎈 你答：{answer}\n"
|
||
f"🌟 正确答案：{correct_answer}\n"
|
||
f"🌈 扣 1 分，别灰心！\n"
|
||
f"🎀 提示：{description}"
|
||
)
|
||
|
||
cursor.execute(
|
||
"INSERT INTO t_encyclopedia_task_history (group_id, active_task_id, player_id, answer, is_correct, points_earned) VALUES (%s, %s, %s, %s, %s, %s)",
|
||
(group_id, active_task_id, player_id, answer, is_correct, points)
|
||
)
|
||
conn.commit()
|
||
|
||
return message
|
||
except ValueError:
|
||
return (
|
||
f"😅 哎呀，小调皮\n"
|
||
f"🌟 任务ID格式不对哦\n"
|
||
f"🎈 应该是 task_[数字]\n"
|
||
f"🌈 比如：task_1\n"
|
||
f"🎀 再试一次吧！"
|
||
)
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
|
||
# 显示排行榜
|
||
def show_rank(group_id, player_id):
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
try:
|
||
cursor.execute(
|
||
"SELECT player_name, points FROM t_encyclopedia_players WHERE group_id = %s ORDER BY points DESC LIMIT 10",
|
||
(group_id,)
|
||
)
|
||
ranks = cursor.fetchall()
|
||
if not ranks:
|
||
return (
|
||
f"😔 群 {group_id} 好安静\n"
|
||
f"🌟 排行榜还没有人\n"
|
||
f"🎈 快来一起玩吧！"
|
||
)
|
||
rank_text = f"🎉 群 {group_id} 排行榜（Top 10）\n"
|
||
for i, row in enumerate(ranks, 1):
|
||
rank_text += f"🌈 {i}. {row['player_name']}: {row['points']} 分\n"
|
||
return rank_text
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
|
||
# 显示当前活跃任务
|
||
def show_active_tasks(group_id, player_id):
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
try:
|
||
cursor.execute("""
|
||
SELECT a.active_task_id, a.question, p.player_name, p.player_id
|
||
FROM t_encyclopedia_active_tasks a
|
||
JOIN t_encyclopedia_players p ON a.holder_id = p.player_id AND a.group_id = p.group_id
|
||
WHERE a.group_id = %s AND a.status = 'pending'
|
||
""", (group_id,))
|
||
tasks = cursor.fetchall()
|
||
if not tasks:
|
||
return (
|
||
f"😄 群 {group_id} 现在很平静\n"
|
||
f"🌟 没有活跃任务哦\n"
|
||
f"🎈 用 /gettask 拿一个吧！"
|
||
)
|
||
task_text = f"🎉 群 {group_id} 当前活跃任务：\n"
|
||
for task in tasks:
|
||
task_text += (
|
||
f"🌈 任务ID: task_{task['active_task_id']}\n"
|
||
f"🎀 问题：{task['question']}\n"
|
||
f"🌼 持有者：{task['player_name']} (ID: {task['player_id']})\n"
|
||
)
|
||
return task_text
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
|
||
# 列举所有未完成任务及其所属者
|
||
def list_uncompleted_tasks(group_id, player_id):
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
try:
|
||
cursor.execute("""
|
||
SELECT a.active_task_id, a.question, p.player_name, p.player_id
|
||
FROM t_encyclopedia_active_tasks a
|
||
JOIN t_encyclopedia_players p ON a.holder_id = p.player_id AND a.group_id = p.group_id
|
||
WHERE a.group_id = %s AND a.status = 'pending'
|
||
""", (group_id,))
|
||
tasks = cursor.fetchall()
|
||
if not tasks:
|
||
return (
|
||
f"😄 群 {group_id} 任务都完成啦\n"
|
||
f"🌟 目前没有未完成任务\n"
|
||
f"🎈 用 /gettask 再来一个吧！"
|
||
)
|
||
task_text = f"🎉 群 {group_id} 所有未完成任务列表：\n"
|
||
for task in tasks:
|
||
task_text += (
|
||
f"🌈 任务ID: task_{task['active_task_id']}\n"
|
||
f"🎀 问题：{task['question']}\n"
|
||
f"🌼 所属：{task['player_name']} (ID: {task['player_id']})\n"
|
||
)
|
||
return task_text
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
|
||
# 定时任务：整点触发，排除23:00-08:00
|
||
def run_random_task_assignment(group_id):
|
||
current_hour = datetime.now().hour
|
||
if current_hour >= 23 or current_hour < 9: # 排除23:00-08:00
|
||
print(f"{datetime.now()} 群 {group_id} 当前时间 {current_hour}:00 在23:00-08:00区间，跳过任务发放")
|
||
return
|
||
result = assign_random_task(group_id)
|
||
print(f"{datetime.now()} {result}")
|
||
|
||
|
||
# 处理群聊消息
|
||
def game_process_message(group_id, player_id, message, player_name="未知玩家"):
|
||
if message == "/start":
|
||
return start_game(group_id, player_id, player_name)
|
||
elif message == "/tasks":
|
||
return show_active_tasks(group_id, player_id)
|
||
elif message == "/list":
|
||
return list_uncompleted_tasks(group_id, player_id)
|
||
elif message.startswith("/answer"):
|
||
parts = message.split(" ", 2)
|
||
if len(parts) < 3:
|
||
return (
|
||
f"😅 哎呀，小朋友\n"
|
||
f"🌟 格式不对哦\n"
|
||
f"🎈 正确用法：/answer [任务ID] [答案]\n"
|
||
f"🌈 例如：/answer task_1 钒（Vanadium）"
|
||
)
|
||
task_id, answer = parts[1], parts[2]
|
||
return submit_answer(group_id, player_id, task_id, answer)
|
||
elif message == "/rank":
|
||
return show_rank(group_id, player_id)
|
||
elif message.startswith("/addgroup"):
|
||
return add_group(group_id, player_id)
|
||
elif message == "/gettask":
|
||
current_hour = datetime.now().hour
|
||
if current_hour >= 23 or current_hour < 9:
|
||
return (
|
||
f"😴 现在是 {current_hour}:00\n"
|
||
f"🌙 太晚啦，小伙伴们休息中\n"
|
||
f"🌞 9点到22点再来玩吧！"
|
||
)
|
||
return assign_random_task(group_id)
|
||
else:
|
||
return (
|
||
f"😄 嘿，小调皮\n"
|
||
f"🌟 指令不对哦\n"
|
||
f"🎈 可用指令：\n"
|
||
f"🌈 /start - 加入游戏\n"
|
||
f"🎀 /tasks - 查看活跃任务\n"
|
||
f"🌼 /list - 未完成任务\n"
|
||
f"🌟 /answer [任务ID] [答案] - 提交答案\n"
|
||
f"🎈 /addgroup - 添加群聊\n"
|
||
f"🌈 /gettask - 主动拿任务\n"
|
||
f"🎀 /rank - 查看排行榜"
|
||
)
|
||
|
||
|
||
# 设置定时任务
|
||
def setup_schedule():
|
||
group_ids = get_group_ids()
|
||
for gid in group_ids:
|
||
schedule.every().hour.at(":00").do(run_random_task_assignment, group_id=gid)
|
||
|
||
|
||
# 主程序
|
||
if __name__ == "__main__":
|
||
# # 初始化群聊
|
||
# print(add_group("group1", "admin1"))
|
||
# print(add_group("group2", "admin1"))
|
||
#
|
||
# # 初始化玩家
|
||
# print(game_process_message("group1", "player1001", "/start", "玩家1"))
|
||
# print(game_process_message("group1", "player1002", "/start", "玩家2"))
|
||
# print(game_process_message("group2", "player2001", "/start", "玩家A"))
|
||
# print(game_process_message("group2", "player2002", "/start", "玩家B"))
|
||
#
|
||
# # 设置调度
|
||
# setup_schedule()
|
||
|
||
# 测试 /gettask
|
||
print(game_process_message("47530198896@chatroom", "player1001", "/gettask"))
|