ngfchl
66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
"""入群验证服务"""
|
|
import random
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
from db.models import VerifyQueue
|
|
|
|
logger = logging.getLogger("spam_guard")
|
|
|
|
# 验证题库
|
|
VERIFY_QUESTIONS = [
|
|
{"q": "收割机的英文名是什么？", "a": "harvest"},
|
|
{"q": "CookieCloud 的默认端口是？", "a": "80"},
|
|
{"q": "Docker 停止容器的命令是？", "a": "stop"},
|
|
{"q": "PT 站点的全称是？", "a": "private tracker"},
|
|
{"q": "收割机项目的作者是？", "a": "ngfchl"},
|
|
{"q": "收割机文档站在哪个域名？", "a": "ptools.fun"},
|
|
]
|
|
|
|
|
|
async def get_pending_verify(user_id: int) -> dict | None:
|
|
"""获取用户待验证题目。"""
|
|
vq = await VerifyQueue.filter(user_id=user_id, status="pending").order_by("-id").first()
|
|
if not vq:
|
|
return None
|
|
return {"id": vq.id, "question": vq.question}
|
|
|
|
|
|
async def create_verify(user_id: int, username: str = None) -> dict:
|
|
"""为新用户创建验证"""
|
|
existing = await get_pending_verify(user_id)
|
|
if existing:
|
|
return existing
|
|
|
|
q = random.choice(VERIFY_QUESTIONS)
|
|
vq = await VerifyQueue.create(
|
|
user_id=user_id,
|
|
username=username,
|
|
question=q["q"],
|
|
answer=q["a"],
|
|
status="pending",
|
|
)
|
|
logger.info(f"📝 验证已创建: user={user_id} q={q['q']}")
|
|
return {"id": vq.id, "question": q["q"]}
|
|
|
|
|
|
async def check_answer(user_id: int, answer: str) -> dict:
|
|
"""检查答案"""
|
|
vq = await VerifyQueue.filter(user_id=user_id, status="pending").order_by("-id").first()
|
|
if not vq:
|
|
return {"ok": False, "reason": "no_pending"}
|
|
|
|
vq.attempts += 1
|
|
if answer.strip().lower() == vq.answer.strip().lower():
|
|
vq.status = "verified"
|
|
vq.verified_at = datetime.now()
|
|
await vq.save()
|
|
return {"ok": True, "reason": "verified"}
|
|
else:
|
|
if vq.attempts >= 3:
|
|
vq.status = "failed"
|
|
await vq.save()
|
|
return {"ok": False, "reason": "max_attempts"}
|
|
await vq.save()
|
|
return {"ok": False, "reason": "wrong", "attempts": vq.attempts}
|