tg-spam-guard/handlers/welcome.py

"""入群欢迎 / 退群告别"""
import asyncio
import logging
from datetime import datetime, timedelta

from telegram import ChatMember, InlineKeyboardButton, InlineKeyboardMarkup, Update
from telegram.ext import ChatMemberHandler, ContextTypes, MessageHandler, filters

import config
from services.message_utils import record_bot_message
from db.models import Action, Ban, Message
from services.spam_detector import classify_username_spam
from services.verify_service import create_verify
from web.api import broadcast_sse, json_safe

logger = logging.getLogger("spam_guard")
JOIN_SERVICE_TEXT = "__service__:new_chat_member"


async def handle_new_chat_members_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """记录 Telegram 入群系统消息，供后台删除用户消息时联动清理。"""
    msg = update.effective_message
    if not msg or not msg.new_chat_members:
        return
    chat_id = update.effective_chat.id if update.effective_chat else msg.chat_id
    for user in msg.new_chat_members:
        if user.is_bot:
            continue
        username = user.username or ""
        first_name = " ".join(x for x in [user.first_name, user.last_name] if x) or user.first_name or ""
        exists = await Message.get_or_none(chat_id=chat_id, msg_id=msg.message_id, user_id=user.id, text=JOIN_SERVICE_TEXT)
        if exists:
            continue
        await Message.create(
            msg_id=msg.message_id,
            chat_id=chat_id,
            user_id=user.id,
            username=username,
            first_name=first_name,
            text=JOIN_SERVICE_TEXT,
        )
        logger.info(f"🧾 已记录入群系统消息: user={user.id}, msg_id={msg.message_id}")


WELCOME_TEXT = (
    "👋 欢迎新成员 **{username}** 加入收割机群组！\n"
    "━━━━━━━━━━━━━━━━━━━━\n"
    "📖 请先阅读群规，文明交流\n"
    "❓ 有问题用 /ask 向知识库提问\n"
    "🛡️ 自动垃圾检测已开启\n\n"
    "⚠️ 请在 {timeout} 分钟内点击按钮完成验证，否则可能被移出"
)

GOODBYE_TEXT = "👋 {username} 离开了群组"


def verify_keyboard() -> InlineKeyboardMarkup:
    return InlineKeyboardMarkup([
        [InlineKeyboardButton("✅ 开始验证", callback_data="start_verify")],
        [InlineKeyboardButton("💡 验证帮助", callback_data="verify_help")],
    ])


async def delete_join_service_message(bot, chat_id: int, user_id: int) -> dict:
    """删除指定用户最近的 Telegram 入群系统提示。"""
    rows = await Message.filter(
        chat_id=chat_id,
        user_id=user_id,
        text=JOIN_SERVICE_TEXT,
        manually_deleted=False,
    ).order_by("-created_at").limit(3)
    result = {"attempted": 0, "deleted": 0, "errors": []}
    seen_msg_ids: set[int] = set()
    for row in rows:
        if not row.msg_id or int(row.msg_id) in seen_msg_ids:
            row.manually_deleted = True
            row.deleted_at = datetime.now()
            row.delete_reason = "welcome_cleanup_join_duplicate"
            await row.save()
            continue
        seen_msg_ids.add(int(row.msg_id))
        result["attempted"] += 1
        try:
            await bot.delete_message(chat_id=chat_id, message_id=row.msg_id)
            result["deleted"] += 1
            row.delete_reason = "welcome_cleanup_join_service"
            logger.info(f"🧹 已删除入群系统消息: user={user_id}, msg_id={row.msg_id}")
        except Exception as e:
            msg = str(e)
            if "message to delete not found" in msg.lower():
                row.delete_reason = f"welcome_cleanup_join_not_found: {msg[:120]}"
            else:
                result["errors"].append(f"{type(e).__name__}: {msg}")
                continue
        row.manually_deleted = True
        row.deleted_at = datetime.now()
        await row.save()
    if result["attempted"]:
        await Action.create(action="WELCOME_CLEAN_JOIN_MESSAGE", chat_id=chat_id, user_id=user_id, details=result)
    return result


async def auto_delete_welcome_and_join(bot, welcome_msg, chat_id: int, user_id: int, seconds: int) -> None:
    """欢迎消息到期销毁时，同步删除该用户的入群系统记录。"""
    await asyncio.sleep(seconds)
    try:
        await welcome_msg.delete()
        logger.info(f"🧹 已删除入群欢迎消息: user={user_id}, msg_id={welcome_msg.message_id}")
    except Exception as e:
        logger.debug(f"自动删除入群欢迎消息失败: {e}")
    try:
        await delete_join_service_message(bot, chat_id, user_id)
    except Exception as e:
        logger.warning(f"删除入群系统消息失败: user={user_id}, {type(e).__name__}: {e}")


async def handle_chat_member(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """入群欢迎 / 退群告别"""
    cmu = update.chat_member
    if not cmu or not cmu.new_chat_member:
        return

    user = cmu.new_chat_member.user
    if user.is_bot:
        return

    old_status = cmu.old_chat_member.status if cmu.old_chat_member else "unknown"
    new_status = cmu.new_chat_member.status

    # 入群
    if old_status in (ChatMember.LEFT, ChatMember.BANNED) and new_status in (ChatMember.MEMBER, ChatMember.RESTRICTED):
        username = user.username or user.first_name or str(user.id)

        full_name = " ".join(x for x in [user.first_name, user.last_name] if x)
        await Action.create(action="JOIN", chat_id=update.effective_chat.id, user_id=user.id, username=username, details={"chat_id": update.effective_chat.id})
        logger.info(f"🆕 入群: @{username} ({user.id})")

        # 新入群用户名称广告检测：命中后不再进入欢迎/验证流程
        is_name_spam, name_reason, name_score = classify_username_spam(
            username=user.username or "",
            first_name=user.first_name or "",
            full_name=full_name,
        )
        if is_name_spam:
            logger.warning(f"🚨 新用户名称疑似广告 [{username}] ({user.id}): {name_reason}")
            if user.id in config.ADMIN_USER_IDS:
                await Action.create(
                    action="SPAM_NAME_ADMIN_TEST",
                    user_id=user.id,
                    username=username,
                    details={"reason": name_reason, "score": name_score},
                )
                return
            try:
                unban_at = datetime.now() + timedelta(minutes=config.BAN_DURATION_MINUTES)
                await update.effective_chat.ban_member(user_id=user.id, until_date=int(unban_at.timestamp()))
                ban = await Ban.create(
                    chat_id=update.effective_chat.id,
                    user_id=user.id,
                    username=user.username or username,
                    first_name=user.first_name or "",
                    reason=name_reason,
                    method="join_name_spam",
                    duration_minutes=config.BAN_DURATION_MINUTES,
                    unban_at=unban_at,
                )
                await Action.create(
                    action="BAN_NAME_SPAM",
                    user_id=user.id,
                    username=username,
                    details={"reason": name_reason, "score": name_score},
                )
                await broadcast_sse("ban", json_safe({
                    "id": ban.id,
                    "user_id": user.id,
                    "username": user.username or username,
                    "first_name": user.first_name or "",
                    "reason": name_reason,
                    "method": "join_name_spam",
                    "duration_minutes": config.BAN_DURATION_MINUTES,
                    "auto_unban": True,
                    "unban_at": unban_at,
                    "time": ban.created_at,
                }))
                logger.info(f"🔨 已因广告用户名封禁 [{username}] {config.BAN_DURATION_MINUTES} 分钟")
            except Exception as e:
                logger.error(f"❌ 广告用户名封禁失败: {e}")
            return

        # 先创建验证题，避免按钮点下去无 pending 记录
        try:
            await create_verify(user.id, username)
        except Exception as e:
            logger.error(f"⚠️ 创建验证失败: {e}")

        # 限制新人（允许文本，禁止媒体/链接预览/拉人）
        try:
            from telegram import ChatPermissions

            await update.effective_chat.restrict_member(
                user_id=user.id,
                permissions=ChatPermissions(
                    can_send_messages=True,
                    can_send_audios=False,
                    can_send_documents=False,
                    can_send_photos=False,
                    can_send_videos=False,
                    can_send_video_notes=False,
                    can_send_voice_notes=False,
                    can_send_polls=False,
                    can_send_other_messages=False,
                    can_add_web_page_previews=False,
                    can_invite_users=False,
                ),
            )
        except Exception as e:
            logger.error(f"⚠️ 限制新人失败: {e}")

        # 发欢迎+验证按钮
        try:
            timeout_min = max(1, config.VERIFY_TIMEOUT_SECONDS // 60)
            msg = await update.effective_chat.send_message(
                WELCOME_TEXT.format(username=f"@{username}", timeout=timeout_min),
                parse_mode="Markdown",
                reply_markup=verify_keyboard(),
            )
            asyncio.create_task(auto_delete_welcome_and_join(context.bot, msg, update.effective_chat.id, user.id, config.VERIFY_TIMEOUT_SECONDS))
        except Exception as e:
            logger.error(f"❌ 欢迎消息发送失败: {e}")

    # 退群
    elif new_status in (ChatMember.LEFT, ChatMember.BANNED):
        username = user.username or user.first_name or str(user.id)
        await Action.create(action="LEAVE", chat_id=update.effective_chat.id, user_id=user.id, username=username)
        logger.info(f"🚪 离群: @{username} ({user.id})")


def register_welcome(app):
    """注册入群欢迎/退群告别"""
    app.add_handler(MessageHandler(filters.StatusUpdate.NEW_CHAT_MEMBERS, handle_new_chat_members_message), group=-1)
    app.add_handler(ChatMemberHandler(handle_chat_member, ChatMemberHandler.CHAT_MEMBER))