TelegramBotAdmin/wbb/modules/points.py

from __future__ import annotations

import asyncio
import re
from html import escape

from pyrogram import filters
from pyrogram.enums import ParseMode
from pyrogram.types import Message, User

from wbb import app, log
from wbb.core.decorators.permissions import adminsOnly
from wbb.services.member_identity import non_mention_account_name
from wbb.services.message_cleanup import schedule_group_cleanup
from wbb.services.point_settings import (
    CHECKIN_BUTTON_TEXT,
    apply_point_rules,
    checkin_button_active,
    refresh_active_checkin_keyboards,
)
from wbb.utils.dbpoints import (
    DEFAULT_CHECKIN_ALREADY_MESSAGE,
    DEFAULT_CHECKIN_SUCCESS_MESSAGE,
    SOURCE_ADMIN,
    InsufficientPoints,
    PointsError,
    adjust_points,
    award_activity,
    award_checkin,
    get_checkin_metrics,
    get_point_account,
    get_point_rules,
    list_point_accounts,
    list_point_transactions,
    reconcile_all_accounts,
    set_points,
    utc_now,
)
from wbb.utils.i18n import point_source_label

__MODULE__ = "积分管理"
__HELP__ = """/points - 查看自己或被回复成员的积分余额。
/checkin - 领取每日签到积分。
/points_rank - 查看本群积分排行榜，也可直接发送“积分排行”。
/points_history - 查看自己或被回复成员的最近积分流水。
/points_add <成员> <数值> <原因> - 增加积分。
/points_deduct <成员> <数值> <原因> - 扣减积分。
/points_set <成员> <余额> <原因> - 通过新流水设置余额。
/points_toggle <开启|关闭> - 开关本群积分系统。
/checkin_button <开启|关闭> - 开关输入框底部的签到按钮。
群成员也可以直接发送“签到”“积分”“流水”触发对应功能。"""

_VOTE_RE = re.compile(
    r"^(\++|\+1|thx|tnx|tq|ty|thankyou|thank you|thanx|thanks|pro|cool|good|agree|👍|-+|-1|not cool|disagree|worst|bad|👎)$",
    re.IGNORECASE,
)
_reconciler_started = False
RECONCILE_INTERVAL_SECONDS = 3600
_TOGGLE_CHOICES = {
    "enable": True,
    "enabled": True,
    "on": True,
    "开启": True,
    "打开": True,
    "disable": False,
    "disabled": False,
    "off": False,
    "关闭": False,
}


def _mention(user_id: int, label: str | None) -> str:
    return f'<a href="tg://user?id={user_id}">{escape(label or str(user_id))}</a>'


def _user_display_name(user: User) -> str:
    return " ".join(
        value for value in (user.first_name, user.last_name) if value
    ) or (f"@{user.username}" if user.username else str(user.id))


def _message_content(message: Message) -> str:
    text = (message.text or message.caption or "").strip()
    if text:
        return text
    media = (
        message.photo
        or message.animation
        or message.video
        or message.document
        or message.audio
        or message.voice
        or message.sticker
    )
    unique_id = getattr(media, "file_unique_id", None)
    return f"media:{unique_id}" if unique_id else ""


async def _reply_checkin(message: Message, text: str) -> Message:
    response = await message.reply_text(text, parse_mode=ParseMode.DISABLED)
    schedule_group_cleanup(message, response)
    return response


async def _reply_balance(message: Message, text: str) -> Message:
    response = await message.reply_text(text, parse_mode=ParseMode.HTML)
    schedule_group_cleanup(message, response)
    return response


def _checkin_message(
    template: str,
    *,
    points: int,
    balance: int,
    rank: int,
    streak: int,
    month_days: int,
) -> str:
    return (
        template.replace("{points}", str(points))
        .replace("{balance}", str(balance))
        .replace("{rank}", str(rank))
        .replace("{streak}", str(streak))
        .replace("{month_days}", str(month_days))
    )


async def _resolve_target(message: Message, raw: str | None = None) -> User | None:
    if message.reply_to_message and message.reply_to_message.from_user:
        return message.reply_to_message.from_user
    if not raw:
        return message.from_user
    try:
        return await app.get_users(int(raw) if raw.lstrip("-").isdigit() else raw)
    except Exception:
        return None


def _adjustment_parts(message: Message) -> tuple[str | None, str | None, str]:
    parts = message.text.split() if message.text else []
    if message.reply_to_message:
        amount = parts[1] if len(parts) > 1 else None
        reason = " ".join(parts[2:]).strip()
        return None, amount, reason
    target = parts[1] if len(parts) > 1 else None
    amount = parts[2] if len(parts) > 2 else None
    reason = " ".join(parts[3:]).strip()
    return target, amount, reason


@app.on_message(
    filters.group & (filters.command("points") | filters.regex(r"^积分$"))
)
async def points_balance_command(_, message: Message):
    target = await _resolve_target(message)
    if not target:
        return await _reply_balance(message, "未找到该成员。")
    account = await get_point_account(message.chat.id, target.id)
    return await _reply_balance(
        message,
        f"{_mention(target.id, target.first_name)} 当前有 "
        f"<b>{int(account['balance'])}</b> 积分。",
    )


@app.on_message(
    filters.group
    & (filters.command("checkin") | filters.regex(r"^签到$"))
)
async def points_checkin_command(_, message: Message):
    if not message.from_user:
        return await _reply_checkin(message, "匿名管理员无法签到。")
    now = utc_now()
    try:
        account, created = await award_checkin(
            chat_id=message.chat.id,
            user_id=message.from_user.id,
            username=message.from_user.username,
            first_name=message.from_user.first_name,
            display_name=_user_display_name(message.from_user),
            now=now,
        )
        metrics = await get_checkin_metrics(
            chat_id=message.chat.id,
            user_id=message.from_user.id,
            now=now,
        )
    except PointsError as exc:
        return await _reply_checkin(message, str(exc))
    rules = await get_point_rules(message.chat.id)
    points = int(rules.get("checkin_points", 0))
    balance = int(account["balance"])
    if not created:
        return await _reply_checkin(
            message,
            _checkin_message(
                str(
                    rules.get("checkin_already_message")
                    or DEFAULT_CHECKIN_ALREADY_MESSAGE
                ),
                points=points,
                balance=balance,
                **metrics,
            ),
        )
    return await _reply_checkin(
        message,
        _checkin_message(
            str(
                rules.get("checkin_success_message")
                or DEFAULT_CHECKIN_SUCCESS_MESSAGE
            ),
            points=points,
            balance=balance,
            **metrics,
        ),
    )


@app.on_message(
    filters.group
    & (filters.command("points_rank") | filters.regex(r"^积分排行$"))
)
async def points_rank_command(_, message: Message):
    accounts, _ = await list_point_accounts(
        chat_id=message.chat.id, page=1, page_size=10
    )
    if not accounts:
        return await message.reply_text("本群尚无积分记录。")
    lines = ["<b>本群积分排行榜</b>"]
    for index, account in enumerate(accounts, 1):
        lines.append(
            f"{index}. {escape(non_mention_account_name(account))} - "
            f"<b>{int(account['balance'])}</b> 积分"
        )
    return await message.reply_text(
        "\n".join(lines), parse_mode=ParseMode.HTML, disable_web_page_preview=True
    )


@app.on_message(
    filters.group
    & (filters.command("points_history") | filters.regex(r"^流水$"))
)
async def points_history_command(_, message: Message):
    target = await _resolve_target(message)
    if not target:
        return await message.reply_text("未找到该成员。")
    items, _ = await list_point_transactions(
        chat_id=message.chat.id, user_id=target.id, page=1, page_size=10
    )
    if not items:
        return await message.reply_text("未找到积分流水。")
    lines = [f"<b>{escape(target.first_name)} 的最近积分流水</b>"]
    for item in items:
        sign = "+" if int(item["delta"]) > 0 else ""
        created_at = item["created_at"].strftime("%Y-%m-%d %H:%M UTC")
        lines.append(
            f"{sign}{int(item['delta'])} · "
            f"{escape(point_source_label(item['source']))} · "
            f"余额 {int(item['balance_after'])} · {created_at}"
        )
    return await message.reply_text("\n".join(lines), parse_mode=ParseMode.HTML)


async def _adjust_points_command(message: Message, operation: str):
    target_raw, amount_raw, reason = _adjustment_parts(message)
    if not amount_raw or not amount_raw.isdigit() or not reason:
        operation_labels = {
            "add": "add",
            "deduct": "deduct",
            "set": "set",
        }
        return await message.reply_text(
            f"用法：/points_{operation_labels[operation]} "
            "<成员> <积分数值> <原因>（也可以回复成员消息后使用）"
        )
    target = await _resolve_target(message, target_raw)
    if not target:
        return await message.reply_text("未找到该成员。")
    value = int(amount_raw)
    if operation == "deduct":
        value = -value
    actor_id = message.from_user.id if message.from_user else message.chat.id
    key = f"admin-points:{message.chat.id}:{message.id}:{operation}:{target.id}"
    try:
        if operation == "set":
            account, created = await set_points(
                chat_id=message.chat.id,
                user_id=target.id,
                balance=value,
                actor_id=actor_id,
                reason=reason,
                idempotency_key=key,
                username=target.username,
                first_name=target.first_name,
                display_name=_user_display_name(target),
            )
        else:
            account, created = await adjust_points(
                chat_id=message.chat.id,
                user_id=target.id,
                delta=value,
                source=SOURCE_ADMIN,
                idempotency_key=key,
                actor_id=actor_id,
                reason=reason,
                username=target.username,
                first_name=target.first_name,
                display_name=_user_display_name(target),
            )
    except (PointsError, InsufficientPoints) as exc:
        return await message.reply_text(str(exc))
    state = "已更新" if created else "此前已处理"
    return await message.reply_text(
        f"积分{state}。{target.first_name} 的当前余额："
        f"{int(account['balance'])}。"
    )


@app.on_message(filters.command("points_add") & ~filters.private)
@adminsOnly("can_change_info")
async def points_add_command(_, message: Message):
    return await _adjust_points_command(message, "add")


@app.on_message(filters.command("points_deduct") & ~filters.private)
@adminsOnly("can_change_info")
async def points_deduct_command(_, message: Message):
    return await _adjust_points_command(message, "deduct")


@app.on_message(filters.command("points_set") & ~filters.private)
@adminsOnly("can_change_info")
async def points_set_command(_, message: Message):
    return await _adjust_points_command(message, "set")


@app.on_message(filters.command("points_toggle") & ~filters.private)
@adminsOnly("can_change_info")
async def points_toggle_command(_, message: Message):
    choice = message.command[1].lower() if len(message.command) == 2 else ""
    if choice not in _TOGGLE_CHOICES:
        return await message.reply_text("用法：/points_toggle <开启|关闭>")
    enabled = _TOGGLE_CHOICES[choice]
    try:
        await apply_point_rules(message.chat.id, {"enabled": enabled})
    except PointsError as exc:
        return await message.reply_text(str(exc))
    return await message.reply_text(
        f"本群积分系统已{'开启' if enabled else '关闭'}。"
    )


@app.on_message(filters.command("checkin_button") & ~filters.private)
@adminsOnly("can_change_info")
async def checkin_button_command(_, message: Message):
    choice = message.command[1].lower() if len(message.command) == 2 else ""
    if choice not in _TOGGLE_CHOICES:
        return await message.reply_text("用法：/checkin_button <开启|关闭>")
    enabled = _TOGGLE_CHOICES[choice]
    try:
        rules = await apply_point_rules(
            message.chat.id, {"checkin_button_enabled": enabled}
        )
    except PointsError as exc:
        return await message.reply_text(str(exc))
    if enabled and not checkin_button_active(rules):
        return await message.reply_text(
            "底部签到按钮设置已开启；启用积分系统和签到规则后会自动显示。"
        )
    return await message.reply_text(
        f"群底部签到按钮已{'开启' if enabled else '关闭'}。"
    )


@app.on_message(filters.group & filters.incoming, group=13)
async def points_activity_listener(_, message: Message):
    if not message.from_user or message.from_user.is_bot:
        return
    if getattr(message, "service", None):
        return
    content = _message_content(message)
    if not content or content.startswith("/"):
        return
    if content == CHECKIN_BUTTON_TEXT:
        return
    if message.reply_to_message and _VOTE_RE.fullmatch(content.strip()):
        return
    await award_activity(
        chat_id=message.chat.id,
        user_id=message.from_user.id,
        message_id=message.id,
        content=content,
        username=message.from_user.username,
        first_name=message.from_user.first_name,
        display_name=_user_display_name(message.from_user),
    )


async def _reconcile_points_loop() -> None:
    try:
        refreshed = await refresh_active_checkin_keyboards()
        if refreshed:
            log.info(f"已刷新 {refreshed} 个群的签到快捷按钮")
    except Exception as exc:
        log.error(f"刷新签到快捷按钮失败：{exc}")
    await asyncio.sleep(60)
    while True:
        try:
            await reconcile_all_accounts(limit=10000)
        except Exception as exc:
            log.error(f"Points reconciliation failed: {exc}")
        await asyncio.sleep(RECONCILE_INTERVAL_SECONDS)


def _start_reconciler() -> None:
    global _reconciler_started
    if _reconciler_started:
        return
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        return
    _reconciler_started = True
    loop.create_task(_reconcile_points_loop(), name="points-reconciler")


_start_reconciler()