tg-spam-guard/handlers/admin.py

"""管理命令 handler"""
import asyncio
import logging
from datetime import datetime, timedelta

from telegram import Update
from telegram.ext import CommandHandler, ContextTypes, MessageHandler, filters

import config
from db.models import Action, Ban, Message, Warning
from services.admin_utils import require_admin, resolve_target_user
from services.stats_service import get_daily_stats
from services.message_utils import record_bot_message, safe_reply

logger = logging.getLogger("spam_guard")


async def ban_user_with_record(chat, bot, user_id: int, target: str | None, operator_id: int | None, reason: str, method: str = "command"):
    unban_at = datetime.now() + timedelta(minutes=config.BAN_DURATION_MINUTES)
    await chat.ban_member(user_id=user_id, until_date=int(unban_at.timestamp()))
    ban = await Ban.create(
        chat_id=chat.id,
        user_id=user_id,
        username=target,
        reason=reason,
        method=method,
        duration_minutes=config.BAN_DURATION_MINUTES,
        unban_at=unban_at,
    )
    await Action.create(action="BAN", chat_id=chat.id, user_id=user_id, username=target, operator_id=operator_id, details={"method": method})
    return ban


async def mark_reply_message_deleted(reply_msg, reason: str = "command_ban_reply"):
    rec = await Message.get_or_none(chat_id=reply_msg.chat_id, msg_id=reply_msg.message_id)
    if rec:
        rec.manually_deleted = True
        rec.deleted_at = datetime.now()
        rec.delete_reason = reason
        await rec.save()


async def delete_replied_message(update: Update, reason: str) -> None:
    """删除管理命令引用的目标消息，并同步标记本地消息记录。"""
    reply_msg = update.message.reply_to_message if update.message else None
    if not reply_msg:
        return
    try:
        await reply_msg.delete()
        await mark_reply_message_deleted(reply_msg, reason=reason)
    except Exception as e:
        logger.debug(f"删除引用消息失败: {type(e).__name__}: {e}")


async def delete_command_message(update: Update) -> None:
    """尽量删除管理员命令消息，失败不影响主流程。"""
    try:
        await update.message.delete()
    except Exception as e:
        logger.debug(f"删除命令消息失败: {type(e).__name__}: {e}")


async def log_admin_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    logger.info(f"⌨️ 收到指令: {update.message.text} from={update.effective_user.id}")


async def cmd_stats(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """今日统计"""
    stats = await get_daily_stats()
    text = (
        f"📊 今日群组统计 ({stats['date']})\n"
        f"━━━━━━━━━━━━━━━━━━━━\n"
        f"💬 消息: {stats['messages']} 条\n"
        f"🗑️ 删除垃圾: {stats['deleted']} 条\n"
        f"🚨 垃圾判定: {stats['spam_messages']} 条\n"
        f"🆕 入群: {stats['joins']} 人\n"
        f"🚪 退群: {stats['leaves']} 人\n"
        f"🔨 封禁: {stats['today_bans']} 人\n"
        f"🔓 解封: {stats['unbans']} 人\n"
        f"📦 历史封禁: {stats['total_bans']} 条\n"
    )
    reply = await safe_reply(update, context, text)
    await record_bot_message(reply)
    asyncio.create_task(auto_delete(reply))


async def cmd_ban(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """手动封禁。支持回复消息、用户ID、历史 username。"""
    if not await require_admin(update, context):
        return

    user_id, target = await resolve_target_user(update, context.args)
    if not user_id:
        reply = await safe_reply(update, context, "用法: 回复用户消息 /ban，或 /ban 用户ID/@历史用户名")
        await record_bot_message(reply)
        asyncio.create_task(auto_delete(reply))
        return

    try:
        await ban_user_with_record(
            update.effective_chat,
            context.bot,
            user_id,
            target,
            update.effective_user.id,
            f"手动封禁 by {update.effective_user.username or update.effective_user.id}",
            method="reply_command" if update.message.reply_to_message else "command",
        )
        if update.message.reply_to_message:
            await delete_replied_message(update, reason="command_ban_reply")
        await delete_command_message(update)
        reply = await update.effective_chat.send_message(f"🔨 已封禁 {target} ({user_id}) {config.BAN_DURATION_MINUTES} 分钟")
    except Exception as e:
        logger.error(f"手动封禁失败: {type(e).__name__}: {e}")
        reply = await safe_reply(update, context, "❌ 封禁失败，请检查 Bot 权限或目标用户状态")

    await record_bot_message(reply)
    asyncio.create_task(auto_delete(reply))


async def cmd_unban(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """手动解封"""
    if not await require_admin(update, context):
        return

    user_id, target = await resolve_target_user(update, context.args)
    if not user_id:
        reply = await safe_reply(update, context, "用法: /unban 用户ID/@历史用户名")
        await record_bot_message(reply)
        asyncio.create_task(auto_delete(reply))
        return

    try:
        await context.bot.unban_chat_member(chat_id=config.TG_CHAT_ID, user_id=user_id)
        await Ban.filter(user_id=user_id).update(auto_unban=False, unban_at=datetime.now())
        await Action.create(action="UNBAN", chat_id=update.effective_chat.id, user_id=user_id, username=target, operator_id=update.effective_user.id)
        reply = await safe_reply(update, context, f"🔓 已解封用户 {target} ({user_id})")
    except Exception as e:
        logger.error(f"手动解封失败: {type(e).__name__}: {e}")
        reply = await safe_reply(update, context, "❌ 解封失败，请检查 Bot 权限或目标用户状态")

    await record_bot_message(reply)
    asyncio.create_task(auto_delete(reply))


async def cmd_warn(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """警告用户"""
    if not await require_admin(update, context):
        return

    user_id, target = await resolve_target_user(update, context.args)
    if not user_id:
        reply = await safe_reply(update, context, "用法: 回复用户消息 /warn [原因]，或 /warn 用户ID/@历史用户名 [原因]")
        await record_bot_message(reply)
        asyncio.create_task(auto_delete(reply))
        return

    # 回复模式下 args 全部是原因；非回复模式下 args[1:] 是原因
    if update.message.reply_to_message:
        reason = " ".join(context.args) if context.args else "未说明原因"
    else:
        reason = " ".join(context.args[1:]) if len(context.args) > 1 else "未说明原因"

    try:
        warn_count = await Warning.filter(user_id=user_id).count() + 1
        await Warning.create(chat_id=update.effective_chat.id, user_id=user_id, username=target, reason=reason, operator_id=update.effective_user.id)
        await Action.create(action="WARN", chat_id=update.effective_chat.id, user_id=user_id, username=target, operator_id=update.effective_user.id, details={"reason": reason})
        if update.message.reply_to_message:
            await delete_replied_message(update, reason="command_warn_reply")
            await delete_command_message(update)
            reply = await update.effective_chat.send_message(f"⚠️ 已警告 {target} ({user_id})\n原因: {reason}\n累计警告: {warn_count}")
        else:
            reply = await safe_reply(update, context, f"⚠️ 已警告 {target} ({user_id})\n原因: {reason}\n累计警告: {warn_count}")
    except Exception as e:
        logger.error(f"警告失败: {type(e).__name__}: {e}")
        reply = await safe_reply(update, context, "❌ 警告失败，请检查 Bot 权限或数据库状态")

    await record_bot_message(reply)
    asyncio.create_task(auto_delete(reply))

async def cmd_mute(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """禁言用户"""
    if not await require_admin(update, context):
        return

    user_id, target = await resolve_target_user(update, context.args)
    if not user_id:
        reply = await safe_reply(update, context, "用法: 回复用户消息 /mute [分钟]，或 /mute 用户ID/@历史用户名 [分钟]")
        await record_bot_message(reply)
        asyncio.create_task(auto_delete(reply))
        return

    minutes = 10
    minute_arg = context.args[0] if update.message.reply_to_message and context.args else (context.args[1] if len(context.args) > 1 else "")
    if minute_arg.isdigit():
        minutes = max(1, min(int(minute_arg), 1440))

    try:
        from telegram import ChatPermissions

        until_date = int((datetime.now() + timedelta(minutes=minutes)).timestamp())
        await update.effective_chat.restrict_member(
            user_id=user_id,
            permissions=ChatPermissions(can_send_messages=False),
            until_date=until_date,
        )
        await Action.create(action="MUTE", chat_id=update.effective_chat.id, user_id=user_id, username=target, operator_id=update.effective_user.id, details={"minutes": minutes})
        if update.message.reply_to_message:
            await delete_replied_message(update, reason="command_mute_reply")
            await delete_command_message(update)
            reply = await update.effective_chat.send_message(f"🔇 已禁言 {target} ({user_id}) {minutes} 分钟")
        else:
            reply = await safe_reply(update, context, f"🔇 已禁言 {target} ({user_id}) {minutes} 分钟")
    except Exception as e:
        logger.error(f"禁言失败: {type(e).__name__}: {e}")
        reply = await safe_reply(update, context, "❌ 禁言失败，请检查 Bot 权限或目标用户状态")

    await record_bot_message(reply)
    asyncio.create_task(auto_delete(reply))


async def require_reply_shortcut(update: Update, context: ContextTypes.DEFAULT_TYPE, usage: str) -> bool:
    """引用消息快捷管理命令必须回复目标消息。"""
    if update.message and update.message.reply_to_message:
        return True
    reply = await safe_reply(update, context, f"用法: 回复目标消息后使用 {usage}")
    await record_bot_message(reply)
    asyncio.create_task(auto_delete(reply))
    return False


async def cmd_dban(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """删除被引用消息并封禁用户。"""
    if not await require_admin(update, context):
        return
    if not await require_reply_shortcut(update, context, "/dban"):
        return
    await cmd_ban(update, context)


async def cmd_dmute(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """删除被引用消息并禁言用户。"""
    if not await require_admin(update, context):
        return
    if not await require_reply_shortcut(update, context, "/dmute [分钟]"):
        return
    await cmd_mute(update, context)


async def cmd_dwarn(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """删除被引用消息并警告用户。"""
    if not await require_admin(update, context):
        return
    if not await require_reply_shortcut(update, context, "/dwarn [原因]"):
        return
    await cmd_warn(update, context)


async def cmd_purge(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """批量删除消息。Telegram Bot API 无法拉历史，只支持回复某条消息后向前按已知 message_id 尝试删除。"""
    if not await require_admin(update, context):
        return

    if not context.args or not context.args[0].isdigit():
        reply = await safe_reply(update, context, "用法: 回复一条消息后 /purge 数量（最多50）")
        await record_bot_message(reply)
        asyncio.create_task(auto_delete(reply))
        return

    if not update.message.reply_to_message:
        reply = await safe_reply(update, context, "❌ /purge 需要回复一条消息作为删除起点")
        await record_bot_message(reply)
        asyncio.create_task(auto_delete(reply))
        return

    count = min(max(int(context.args[0]), 1), 50)
    start_id = update.message.reply_to_message.message_id
    deleted = 0

    for mid in range(start_id, start_id - count, -1):
        try:
            await context.bot.delete_message(chat_id=update.effective_chat.id, message_id=mid)
            deleted += 1
        except Exception:
            pass

    try:
        await update.message.delete()
    except Exception:
        pass

    reply = await update.effective_chat.send_message(f"🧹 已尝试清理 {count} 条，成功删除 {deleted} 条")
    await Action.create(action="PURGE", chat_id=update.effective_chat.id, operator_id=update.effective_user.id, details={"count": count, "deleted": deleted})
    await record_bot_message(reply)
    asyncio.create_task(auto_delete(reply))


async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """帮助"""
    text = (
        "📖 收割机群管机器人帮助\n"
        "━━━━━━━━━━━━━━━━━━━━\n"
        "普通操作：\n"
        "🤖 /ask <问题> - 向知识库提问\n"
        "📊 /stats - 查看今日群组统计\n"
        "✅ /checkin - 每日签到\n"
        "🎮 /trivia - 知识答题\n"
        "🏆 /rank - 查看积分排行\n"
        "🔗 /bind - 绑定账号\n\n"
        "管理员操作：\n"
        "🎁 /lottery - 发起群内抽奖，进入私聊设置奖品和条件\n"
        "🎛️ /panel - 打开管理操作面板\n"
        "🌙 /night on/off - 开启/关闭夜间模式\n"
        "🔨 /ban 用户ID/@历史用户名 - 封禁用户\n"
        "🔇 /mute 用户ID/@历史用户名 [分钟] - 禁言用户，默认 10 分钟\n"
        "⚠️ /warn 用户ID/@历史用户名 [原因] - 警告用户\n"
        "🔓 /unban 用户ID/@历史用户名 - 解封用户\n\n"
        "引用消息操作（管理员）：\n"
        "🔨 回复消息 /dban - 删除被引用消息并封禁发送者\n"
        "🔇 回复消息 /dmute [分钟] - 删除被引用消息并禁言发送者\n"
        "⚠️ 回复消息 /dwarn [原因] - 删除被引用消息并警告发送者\n"
        "↪️ 回复消息 /ban /mute /warn 也支持同样处理\n"
        "🧹 回复消息 /purge 数量 - 从被引用消息开始尝试批量删除，最多 50 条\n"
    )
    reply = await safe_reply(update, context, text)
    await record_bot_message(reply)
    asyncio.create_task(auto_delete(reply, seconds=60))


async def auto_delete(message, seconds=60):
    try:
        await asyncio.sleep(seconds)
        await message.delete()
    except Exception:
        pass


def register_admin(app):
    app.add_handler(MessageHandler(filters.COMMAND & filters.ChatType.GROUPS, log_admin_command), group=-10)
    app.add_handler(CommandHandler("help", cmd_help, filters.ChatType.GROUPS))
    app.add_handler(CommandHandler("stats", cmd_stats, filters.ChatType.GROUPS))
    app.add_handler(CommandHandler("ban", cmd_ban, filters.ChatType.GROUPS))
    app.add_handler(CommandHandler("dban", cmd_dban, filters.ChatType.GROUPS))
    app.add_handler(CommandHandler("unban", cmd_unban, filters.ChatType.GROUPS))
    app.add_handler(CommandHandler("warn", cmd_warn, filters.ChatType.GROUPS))
    app.add_handler(CommandHandler("dwarn", cmd_dwarn, filters.ChatType.GROUPS))
    app.add_handler(CommandHandler("mute", cmd_mute, filters.ChatType.GROUPS))
    app.add_handler(CommandHandler("dmute", cmd_dmute, filters.ChatType.GROUPS))
    app.add_handler(CommandHandler("purge", cmd_purge, filters.ChatType.GROUPS))