From 8fc015acf8fa67fa7ba16d8358c35df68f0a4165 Mon Sep 17 00:00:00 2001 From: lj091715 <1091062319@qq.com> Date: Wed, 20 May 2026 20:16:53 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=8A=E4=BC=A0=E6=96=87=E4=BB=B6=E8=87=B3?= =?UTF-8?q?=20qiandao/scripts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- qiandao/scripts/qiandao.py | 249 +++++++++++++++++++++++++++++++++++++ 1 file changed, 249 insertions(+) create mode 100644 qiandao/scripts/qiandao.py diff --git a/qiandao/scripts/qiandao.py b/qiandao/scripts/qiandao.py new file mode 100644 index 0000000..b913eae --- /dev/null +++ b/qiandao/scripts/qiandao.py @@ -0,0 +1,249 @@ +#!/usr/bin/env python3 +"""签到游戏插件 - 支持每日签到、连续签到奖励、排行榜""" +from __future__ import annotations + +import json +import os +import sys +import time +import traceback +from datetime import date, datetime +from typing import Any + +sys.stderr = sys.stdout + +# ---------- 数据存储 ---------- +DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data") +DATA_FILE = os.path.join(DATA_DIR, "qiandao.json") +os.makedirs(DATA_DIR, exist_ok=True) + +# ---------- 签到奖励配置 ---------- +BASE_POINTS = 10 # 每日基础签到积分 +CONSECUTIVE_BONUS = { # 连续签到额外奖励 + 3: 5, # 连续3天额外+5 + 7: 15, # 连续7天额外+15 + 15: 30, # 连续15天额外+30 + 30: 80, # 连续30天额外+80 +} +MAX_HISTORY_DAYS = 365 # 保留签到历史的天数 + + +def load_data() -> dict[str, Any]: + """从 JSON 文件加载用户数据。""" + if os.path.exists(DATA_FILE): + try: + with open(DATA_FILE, encoding="utf-8") as f: + return json.load(f) + except (json.JSONDecodeError, OSError): + pass + return {"users": {}, "last_cleanup": ""} + + +def save_data(data: dict[str, Any]) -> None: + """保存用户数据到 JSON 文件。""" + with open(DATA_FILE, "w", encoding="utf-8") as f: + json.dump(data, f, ensure_ascii=False, indent=2) + + +def today_str() -> str: + return date.today().isoformat() + + +def get_or_create_user(data: dict[str, Any], user_id: str, nickname: str) -> dict[str, Any]: + """获取或创建用户数据。""" + users = data["users"] + if user_id not in users: + users[user_id] = { + "nickname": nickname, + "total_points": 0, + "consecutive_days": 0, + "last_checkin": "", + "history": [], # 签到日期列表 ["2026-05-01", ...] + "points_history": [], # [(日期, 获得积分), ...] + } + else: + # 更新昵称 + if nickname: + users[user_id]["nickname"] = nickname + return users[user_id] + + +def do_checkin(user_id: str, nickname: str = "") -> str: + """执行签到，返回回复文本。""" + data = load_data() + user = get_or_create_user(data, user_id, nickname) + today = today_str() + + # ---------- 已签到 ---------- + if today in user["history"]: + return ( + f"⚠️ {user['nickname']}，你今天已经签过到啦！\n" + f"📅 总积分：{user['total_points']} 分\n" + f"🔥 连续签到：{user['consecutive_days']} 天" + ) + + # ---------- 计算连续签到 ---------- + yesterday = date.fromisoformat(today) + yesterday_str = yesterday.isoformat() + + if user["last_checkin"] == yesterday_str: + user["consecutive_days"] += 1 + else: + user["consecutive_days"] = 1 + + # ---------- 计算积分 ---------- + earned = BASE_POINTS + bonus_reason = [] + + # 连续签到额外奖励 + for threshold, bonus in sorted(CONSECUTIVE_BONUS.items()): + if user["consecutive_days"] == threshold: + earned += bonus + bonus_reason.append(f"🎉 连续签到 {threshold} 天奖励 +{bonus} 分") + break # 只触发最近的里程碑 + + # 如果连续天数超过最大里程碑但又不是正好等于，奖励固定值 + milestones = sorted(CONSECUTIVE_BONUS.keys()) + if user["consecutive_days"] > milestones[-1]: + # 超过最大里程碑的，每多一天额外+1 + extra_days = user["consecutive_days"] - milestones[-1] + earned += extra_days + bonus_reason.append(f"🔥 长期签到奖励 +{extra_days} 分") + + user["total_points"] += earned + user["last_checkin"] = today + user["history"].append(today) + user["points_history"].append(f"{today}|{earned}") + + # ---------- 清理超期历史 ---------- + cleanup_old_history(data) + + save_data(data) + + # ---------- 构建回复 ---------- + lines = [ + f"✅ 签到成功！{user['nickname']}", + f"📅 获得 {earned} 积分（基础 {BASE_POINTS} 分）", + ] + lines.extend(bonus_reason) + lines.append(f"🔥 已连续签到 {user['consecutive_days']} 天") + lines.append(f"💰 总积分：{user['total_points']}") + return "\n".join(lines) + + +def query_user(user_id: str, nickname: str = "") -> str: + """查询用户签到状态。""" + data = load_data() + if user_id not in data["users"]: + return f"❌ {nickname or user_id} 还没有签到记录，发送「签到」开始吧！" + user = data["users"][user_id] + if nickname: + user["nickname"] = nickname + today = today_str() + checked = "✅ 已签到" if today in user["history"] else "❌ 未签到" + return ( + f"📊 {user['nickname']} 的签到信息\n" + f"{'━' * 20}\n" + f"📅 今日状态：{checked}\n" + f"💰 总积分：{user['total_points']} 分\n" + f"🔥 连续签到：{user['consecutive_days']} 天\n" + f"📆 累计签到：{len(user['history'])} 天" + ) + + +def leaderboard(top_n: int = 10) -> str: + """返回积分排行榜。""" + data = load_data() + sorted_users = sorted( + data["users"].values(), + key=lambda u: u["total_points"], + reverse=True, + ) + if not sorted_users: + return "📭 还没有人签到，快发送「签到」成为第一名吧！" + + lines = ["🏆 签到排行榜 🏆", "━" * 20] + for rank, user in enumerate(sorted_users[:top_n], 1): + medal = {1: "🥇", 2: "🥈", 3: "🥉"}.get(rank, f"{rank}.") + lines.append(f"{medal} {user['nickname']} — {user['total_points']} 分（连续{user['consecutive_days']}天）") + return "\n".join(lines) + + +def cleanup_old_history(data: dict[str, Any]) -> None: + """清理超过 MAX_HISTORY_DAYS 的历史记录。""" + today = date.today() + threshold = today.isoformat() + for uid, user in data["users"].items(): + cutoff = (today - __import__("datetime").timedelta(days=MAX_HISTORY_DAYS)).isoformat() + user["history"] = [d for d in user["history"] if d >= cutoff] + user["points_history"] = [p for p in user["points_history"] if p.split("|")[0] >= cutoff] + + +def reset_user(user_id: str) -> str: + """重置用户数据（管理员功能）。""" + data = load_data() + if user_id in data["users"]: + del data["users"][user_id] + save_data(data) + return f"✅ 已重置用户 {user_id} 的签到数据" + return f"❌ 用户 {user_id} 不存在" + + +def main() -> int: + """主入口，根据命令行参数执行不同操作。 + + 用法: + python qiandao.py checkin [nickname] + python qiandao.py query [nickname] + python qiandao.py rank [top_n] + python qiandao.py reset + """ + args = sys.argv[1:] + if not args: + print("用法: python qiandao.py <命令> [参数...]") + print("命令: checkin, query, rank, reset") + return 1 + + cmd = args[0] + + if cmd == "checkin": + if len(args) < 2: + print("请提供 user_id") + return 1 + uid = args[1] + nickname = args[2] if len(args) > 2 else uid + print(do_checkin(uid, nickname)) + + elif cmd == "query": + if len(args) < 2: + print("请提供 user_id") + return 1 + uid = args[1] + nickname = args[2] if len(args) > 2 else uid + print(query_user(uid, nickname)) + + elif cmd == "rank": + top_n = int(args[1]) if len(args) > 1 else 10 + print(leaderboard(top_n)) + + elif cmd == "reset": + if len(args) < 2: + print("请提供 user_id") + return 1 + print(reset_user(args[1])) + + else: + print(f"未知命令: {cmd}") + return 1 + + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except SystemExit: + raise + except Exception: + traceback.print_exc(file=sys.stdout) + raise SystemExit(1)