更新 qiandao/scripts/qiandao.py
This commit is contained in:
parent
195d8eca99
commit
22336380ad
@ -1,249 +1,246 @@
|
||||
#!/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 <user_id> [nickname]
|
||||
python qiandao.py query <user_id> [nickname]
|
||||
python qiandao.py rank [top_n]
|
||||
python qiandao.py reset <user_id>
|
||||
"""
|
||||
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)
|
||||
#!/usr/bin/env python3
|
||||
# pyright: reportExplicitAny=false, reportAny=false, reportUnknownMemberType=false, reportUnknownArgumentType=false, reportUnknownVariableType=false, reportUnusedVariable=false, reportUnusedImport=false
|
||||
"""签到游戏插件 - 支持每日签到、连续签到奖励、排行榜"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import traceback
|
||||
from datetime import date, timedelta
|
||||
from typing import Any
|
||||
|
||||
# ---------- 数据存储 ----------
|
||||
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()
|
||||
cutoff = (today - timedelta(days=MAX_HISTORY_DAYS)).isoformat()
|
||||
for uid, user in data["users"].items():
|
||||
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 <user_id> [nickname]
|
||||
python qiandao.py query <user_id> [nickname]
|
||||
python qiandao.py rank [top_n]
|
||||
python qiandao.py reset <user_id>
|
||||
"""
|
||||
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)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user