TelegramBotAdmin/wbb/services/point_settings.py

from __future__ import annotations

from typing import Any

from pyrogram.types import KeyboardButton, ReplyKeyboardMarkup, ReplyKeyboardRemove

from wbb import app, log
from wbb.services.message_cleanup import schedule_group_cleanup
from wbb.utils.dbpoints import (
    PointsError,
    get_point_rules,
    list_checkin_keyboard_migrations,
    mark_checkin_keyboard_version,
    normalize_point_rules,
    set_point_rules,
)

CHECKIN_BUTTON_TEXT = "签到"
CHECKIN_KEYBOARD_VERSION = 1


def checkin_button_active(rules: dict[str, Any]) -> bool:
    return bool(
        rules.get("enabled")
        and rules.get("checkin_enabled")
        and rules.get("checkin_button_enabled")
    )


def checkin_keyboard() -> ReplyKeyboardMarkup:
    return ReplyKeyboardMarkup(
        [[KeyboardButton(CHECKIN_BUTTON_TEXT)]],
        is_persistent=False,
        resize_keyboard=True,
        one_time_keyboard=True,
        selective=False,
        placeholder="输入消息",
    )


async def _sync_checkin_keyboard(chat_id: int, enabled: bool) -> None:
    messages = []
    try:
        if enabled:
            messages.append(
                await app.send_message(
                    chat_id,
                    "正在更新签到快捷按钮。",
                    reply_markup=ReplyKeyboardRemove(selective=False),
                )
            )
            messages.append(
                await app.send_message(
                    chat_id,
                    "签到快捷按钮已开启。点击下方“签到”即可领取每日积分。",
                    reply_markup=checkin_keyboard(),
                )
            )
        else:
            messages.append(
                await app.send_message(
                    chat_id,
                    "签到快捷按钮已关闭。",
                    reply_markup=ReplyKeyboardRemove(selective=False),
                )
            )
    except Exception as exc:
        raise PointsError(
            "无法更新群底部签到按钮，请确认机器人可以在群内发言。"
        ) from exc
    schedule_group_cleanup(*messages)


async def refresh_active_checkin_keyboards() -> int:
    chat_ids = await list_checkin_keyboard_migrations(CHECKIN_KEYBOARD_VERSION)
    refreshed = 0
    for chat_id in chat_ids:
        try:
            await _sync_checkin_keyboard(chat_id, True)
            await mark_checkin_keyboard_version(
                chat_id,
                CHECKIN_KEYBOARD_VERSION,
            )
            refreshed += 1
        except Exception as exc:
            log.error(f"刷新群 {chat_id} 的签到快捷按钮失败：{exc}")
    return refreshed


async def apply_point_rules(chat_id: int, values: dict[str, Any]) -> dict[str, Any]:
    current = await get_point_rules(chat_id)
    target = normalize_point_rules({**current, **values})
    current_active = checkin_button_active(current)
    target_active = checkin_button_active(target)

    saved = await set_point_rules(chat_id, target)
    if current_active == target_active:
        return saved
    try:
        await _sync_checkin_keyboard(chat_id, target_active)
        if target_active:
            await mark_checkin_keyboard_version(
                chat_id,
                CHECKIN_KEYBOARD_VERSION,
            )
    except Exception:
        await set_point_rules(chat_id, current)
        raise
    return saved