ngfchl
311 lines
13 KiB
Python
311 lines
13 KiB
Python
"""小游戏服务"""
|
||
import random
|
||
import logging
|
||
from datetime import datetime, timedelta
|
||
|
||
from db.models import GameScore, LotteryEvent, LotteryParticipant, ShopItem, ShopTransaction
|
||
|
||
logger = logging.getLogger("spam_guard")
|
||
|
||
TRIVIA_QUESTIONS = [
|
||
{"q": "收割机项目的 GitHub 地址是？", "a": "github.com/ngfchl/go-harvest", "opts": ["github.com/ngfchl/go-harvest", "github.com/ngfchl/harvest", "github.com/harvest/go-harvest"]},
|
||
{"q": "以下哪个不是 PT 站的常见等级？", "a": "Diamond", "opts": ["User", "Power User", "Diamond", "Elite"]},
|
||
{"q": "Docker 默认的日志驱动是？", "a": "json-file", "opts": ["json-file", "syslog", "journald", "fluentd"]},
|
||
]
|
||
|
||
SHOP_ITEMS = {
|
||
"1": {"name": "补签卡", "cost": 50, "desc": "预留权益：后续可用于补签"},
|
||
"2": {"name": "群内称号券", "cost": 80, "desc": "预留权益：联系管理员兑换称号"},
|
||
"3": {"name": "好运盲盒", "cost": 30, "desc": "兑换后记录一次好运盲盒"},
|
||
}
|
||
|
||
|
||
async def get_or_create_score(chat_id: int, user_id: int, username: str, game_type: str) -> GameScore:
|
||
score, _ = await GameScore.get_or_create(
|
||
chat_id=chat_id, user_id=user_id, game_type=game_type,
|
||
defaults={"username": username},
|
||
)
|
||
if username and score.username != username:
|
||
score.username = username
|
||
await score.save()
|
||
return score
|
||
|
||
|
||
async def get_trivia_question() -> dict:
|
||
return random.choice(TRIVIA_QUESTIONS)
|
||
|
||
|
||
async def check_trivia_answer(chat_id: int, user_id: int, username: str, answer: str, correct_answer: str) -> dict:
|
||
score = await get_or_create_score(chat_id, user_id, username, "trivia")
|
||
score.total += 1
|
||
is_correct = answer.strip().lower() in correct_answer.strip().lower()
|
||
if is_correct:
|
||
score.score += 10
|
||
score.wins += 1
|
||
score.streak += 1
|
||
if score.streak > 1:
|
||
score.score += 5
|
||
else:
|
||
score.streak = 0
|
||
score.last_played = datetime.now()
|
||
await score.save()
|
||
return {"correct": is_correct, "score": score.score, "streak": score.streak}
|
||
|
||
|
||
async def get_leaderboard(chat_id: int, game_type: str = "trivia", limit: int = 10) -> list:
|
||
return await GameScore.filter(chat_id=chat_id, game_type=game_type).order_by("-score").limit(limit).all()
|
||
|
||
|
||
async def daily_checkin(chat_id: int, user_id: int, username: str) -> dict:
|
||
score = await get_or_create_score(chat_id, user_id, username, "checkin")
|
||
today = datetime.now().strftime("%Y-%m-%d")
|
||
if score.last_played and score.last_played.strftime("%Y-%m-%d") == today:
|
||
return {"ok": False, "reason": "already_checked", "score": score.score, "streak": score.streak}
|
||
score.total += 1
|
||
base = random.randint(5, 12)
|
||
streak_bonus = min(score.streak + 1, 7)
|
||
bonus = base + streak_bonus
|
||
score.score += bonus
|
||
score.streak += 1
|
||
score.wins += 1
|
||
score.last_played = datetime.now()
|
||
await score.save()
|
||
return {"ok": True, "score": score.score, "streak": score.streak, "bonus": bonus, "base": base, "streak_bonus": streak_bonus}
|
||
|
||
|
||
async def get_scores_summary(user_id: int, chat_id: int = 0) -> dict:
|
||
scores = await GameScore.filter(chat_id=chat_id, user_id=user_id).all()
|
||
total = sum((s.score or 0) for s in scores)
|
||
by_type = {s.game_type: s for s in scores}
|
||
return {"scores": scores, "total": total, "by_type": by_type}
|
||
|
||
|
||
async def deduct_points(chat_id: int, user_id: int, cost: int) -> bool:
|
||
summary = await get_scores_summary(user_id, chat_id)
|
||
if summary["total"] < cost:
|
||
return False
|
||
remaining = cost
|
||
for score in sorted(summary["scores"], key=lambda x: x.score or 0, reverse=True):
|
||
if remaining <= 0:
|
||
break
|
||
take = min(score.score, remaining)
|
||
score.score -= take
|
||
remaining -= take
|
||
await score.save()
|
||
return True
|
||
|
||
|
||
async def lottery_draw(chat_id: int, user_id: int, username: str) -> dict:
|
||
cost = 10
|
||
if not await deduct_points(chat_id, user_id, cost):
|
||
return {"ok": False, "reason": "积分不足", "cost": cost}
|
||
roll = random.random()
|
||
reward = 0
|
||
prize = "谢谢参与"
|
||
if roll < 0.08:
|
||
reward, prize = 60, "一等奖 +60"
|
||
elif roll < 0.25:
|
||
reward, prize = 25, "二等奖 +25"
|
||
elif roll < 0.55:
|
||
reward, prize = 12, "三等奖 +12"
|
||
score = await get_or_create_score(chat_id, user_id, username, "lottery")
|
||
score.total += 1
|
||
if reward:
|
||
score.wins += 1
|
||
score.score += reward
|
||
score.last_played = datetime.now()
|
||
await score.save()
|
||
summary = await get_scores_summary(user_id, chat_id)
|
||
return {"ok": True, "cost": cost, "reward": reward, "prize": prize, "total": summary["total"]}
|
||
|
||
|
||
async def list_shop_items(chat_id: int, include_disabled: bool = False) -> list[dict]:
|
||
rows = await ShopItem.filter(chat_id=chat_id).order_by("item_key").all()
|
||
if not rows and chat_id != 0:
|
||
rows = await ShopItem.filter(chat_id=0).order_by("item_key").all()
|
||
if not rows:
|
||
return [{"item_key": k, "name": v["name"], "cost": v["cost"], "desc": v["desc"], "enabled": True} for k, v in SHOP_ITEMS.items()]
|
||
result = []
|
||
for r in rows:
|
||
if include_disabled or r.enabled:
|
||
result.append({"id": r.id, "chat_id": r.chat_id, "item_key": r.item_key, "name": r.name, "cost": r.cost, "desc": r.desc, "enabled": r.enabled})
|
||
return result
|
||
|
||
|
||
async def upsert_shop_item(chat_id: int, item_key: str, name: str, cost: int, desc: str = "", enabled: bool = True) -> ShopItem:
|
||
item, _ = await ShopItem.get_or_create(chat_id=chat_id, item_key=item_key, defaults={"name": name, "cost": cost, "desc": desc, "enabled": enabled})
|
||
item.name = name
|
||
item.cost = cost
|
||
item.desc = desc
|
||
item.enabled = enabled
|
||
await item.save()
|
||
return item
|
||
|
||
|
||
async def delete_shop_item(chat_id: int, item_key: str) -> bool:
|
||
deleted = await ShopItem.filter(chat_id=chat_id, item_key=item_key).delete()
|
||
return bool(deleted)
|
||
|
||
|
||
async def list_shop_transactions(chat_id: int, limit: int = 100) -> list[dict]:
|
||
rows = await ShopTransaction.filter(chat_id=chat_id).order_by("-created_at").limit(limit).all()
|
||
return [{"id": r.id, "chat_id": r.chat_id, "user_id": r.user_id, "username": r.username, "item_key": r.item_key, "item_name": r.item_name, "cost": r.cost, "status": r.status, "note": r.note, "created_at": r.created_at} for r in rows]
|
||
|
||
|
||
async def buy_shop_item(chat_id: int, user_id: int, item_id: str, username: str = "") -> dict:
|
||
items = await list_shop_items(chat_id)
|
||
item = next((x for x in items if str(x.get("item_key")) == str(item_id) and x.get("enabled", True)), None)
|
||
if not item:
|
||
return {"ok": False, "reason": "商品不存在"}
|
||
if not await deduct_points(chat_id, user_id, int(item["cost"])):
|
||
return {"ok": False, "reason": "积分不足", "item": item}
|
||
await ShopTransaction.create(
|
||
chat_id=chat_id,
|
||
user_id=user_id,
|
||
username=username,
|
||
item_key=str(item.get("item_key")),
|
||
item_name=item.get("name") or "",
|
||
cost=int(item.get("cost") or 0),
|
||
status="success",
|
||
)
|
||
summary = await get_scores_summary(user_id, chat_id)
|
||
return {"ok": True, "item": item, "total": summary["total"]}
|
||
|
||
|
||
LOTTERY_CONDITION_HELP = "参与条件：all=所有人；checkin=今日已签到；points:N=总积分不少于 N"
|
||
LOTTERY_END_HELP = "结束条件：people:N=满 N 人开奖；minutes:N=N 分钟后开奖；manual=手动开奖（预留）"
|
||
|
||
|
||
def parse_lottery_condition(raw: str) -> tuple[str, int, str]:
|
||
text = (raw or "all").strip().lower()
|
||
if text in {"all", "所有人", "不限"}:
|
||
return "all", 0, "所有人可参与"
|
||
if text in {"checkin", "签到", "今日签到"}:
|
||
return "checkin", 0, "今日已签到用户"
|
||
if text.startswith("points:") or text.startswith("积分:"):
|
||
value = int(text.split(":", 1)[1])
|
||
return "min_points", value, f"总积分 ≥ {value}"
|
||
raise ValueError(LOTTERY_CONDITION_HELP)
|
||
|
||
|
||
def parse_lottery_end(raw: str) -> tuple[str, int, str]:
|
||
text = (raw or "people:3").strip().lower()
|
||
if text in {"instant", "即开", "即开即中"}:
|
||
return "instant", 1, "即开即中：首位参与者立即中奖"
|
||
if text.startswith("people:") or text.startswith("人数:"):
|
||
value = max(1, int(text.split(":", 1)[1]))
|
||
return "people", value, f"满 {value} 人自动开奖"
|
||
if text.startswith("minutes:") or text.startswith("分钟:"):
|
||
value = max(1, int(text.split(":", 1)[1]))
|
||
return "time", value, f"{value} 分钟后自动开奖"
|
||
if text.startswith("time:") or text.startswith("时间:"):
|
||
return "time_at", 0, f"指定时间开奖：{raw.split(':', 1)[1].strip()}"
|
||
if text in {"manual", "手动"}:
|
||
return "manual", 0, "手动开奖"
|
||
raise ValueError(LOTTERY_END_HELP)
|
||
|
||
|
||
def parse_lottery_spec(text: str) -> dict:
|
||
parts = [p.strip() for p in text.replace("｜", "|").split("|")]
|
||
if len(parts) != 3:
|
||
raise ValueError("格式：奖品 | 参与条件 | 结束条件")
|
||
prize = parts[0]
|
||
if not prize:
|
||
raise ValueError("奖品不能为空")
|
||
condition_type, condition_value, condition_text = parse_lottery_condition(parts[1])
|
||
end_type, end_value, end_text = parse_lottery_end(parts[2])
|
||
return {
|
||
"prize": prize,
|
||
"condition_type": condition_type,
|
||
"condition_value": condition_value,
|
||
"condition_text": condition_text,
|
||
"end_type": end_type,
|
||
"end_value": end_value,
|
||
"end_text": end_text,
|
||
}
|
||
|
||
|
||
async def create_lottery_event(chat_id: int, creator_id: int, creator_name: str, spec: dict) -> LotteryEvent:
|
||
return await LotteryEvent.create(
|
||
chat_id=chat_id,
|
||
creator_id=creator_id,
|
||
creator_name=creator_name,
|
||
title=spec.get("title") or spec["prize"],
|
||
description=spec.get("description") or "",
|
||
prize=spec["prize"],
|
||
condition_type=spec["condition_type"],
|
||
condition_value=spec["condition_value"],
|
||
draw_type=spec.get("draw_type") or spec["end_type"],
|
||
draw_at=spec.get("draw_at"),
|
||
max_participants=spec.get("max_participants") or (spec["end_value"] if spec["end_type"] == "people" else 0),
|
||
end_type=spec["end_type"],
|
||
end_value=spec["end_value"],
|
||
)
|
||
|
||
|
||
async def lottery_condition_text(event: LotteryEvent) -> str:
|
||
if event.condition_type == "checkin":
|
||
return "今日已签到用户"
|
||
if event.condition_type == "min_points":
|
||
return f"总积分 ≥ {event.condition_value}"
|
||
return "所有人可参与"
|
||
|
||
|
||
async def lottery_end_text(event: LotteryEvent) -> str:
|
||
if event.end_type == "people":
|
||
return f"满 {event.end_value} 人自动开奖"
|
||
if event.end_type == "minutes":
|
||
return f"{event.end_value} 分钟后自动开奖"
|
||
return "手动开奖"
|
||
|
||
|
||
async def count_lottery_participants(event_id: int) -> int:
|
||
return await LotteryParticipant.filter(event_id=event_id).count()
|
||
|
||
|
||
async def check_lottery_eligible(event: LotteryEvent, user_id: int) -> tuple[bool, str]:
|
||
if event.condition_type == "checkin":
|
||
score = await GameScore.filter(user_id=user_id, game_type="checkin").first()
|
||
today = datetime.now().strftime("%Y-%m-%d")
|
||
if not score or not score.last_played or score.last_played.strftime("%Y-%m-%d") != today:
|
||
return False, "需要今日已签到后才能参与"
|
||
elif event.condition_type == "min_points":
|
||
summary = await get_scores_summary(user_id, event.chat_id)
|
||
if summary["total"] < event.condition_value:
|
||
return False, f"总积分需要 ≥ {event.condition_value}，当前 {summary['total']}"
|
||
return True, "ok"
|
||
|
||
|
||
async def join_lottery_event(event_id: int, user_id: int, username: str = "", first_name: str = "") -> dict:
|
||
event = await LotteryEvent.filter(id=event_id).first()
|
||
if not event or event.status != "active":
|
||
return {"ok": False, "reason": "抽奖不存在或已结束"}
|
||
ok, reason = await check_lottery_eligible(event, user_id)
|
||
if not ok:
|
||
return {"ok": False, "reason": reason}
|
||
exists = await LotteryParticipant.filter(event_id=event_id, user_id=user_id).exists()
|
||
if exists:
|
||
return {"ok": False, "reason": "你已经参与过了"}
|
||
await LotteryParticipant.create(event_id=event_id, user_id=user_id, username=username, first_name=first_name)
|
||
count = await count_lottery_participants(event_id)
|
||
return {"ok": True, "event": event, "count": count, "should_draw": event.end_type == "people" and count >= event.end_value}
|
||
|
||
|
||
async def draw_lottery_event(event_id: int) -> dict:
|
||
event = await LotteryEvent.filter(id=event_id).first()
|
||
if not event or event.status != "active":
|
||
return {"ok": False, "reason": "抽奖不存在或已结束"}
|
||
participants = await LotteryParticipant.filter(event_id=event_id).all()
|
||
if not participants:
|
||
event.status = "cancelled"
|
||
event.ended_at = datetime.now()
|
||
await event.save()
|
||
return {"ok": False, "reason": "无人参与，抽奖已取消", "event": event}
|
||
winner = random.choice(participants)
|
||
event.status = "ended"
|
||
event.winner_id = winner.user_id
|
||
event.winner_name = winner.username or winner.first_name or str(winner.user_id)
|
||
event.ended_at = datetime.now()
|
||
await event.save()
|
||
return {"ok": True, "event": event, "winner": winner, "count": len(participants)}
|