tg-spam-guard/handlers/welcome.py

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

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

import config
from db.models import Action, Ban
from services.message_utils import schedule_delete
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")

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 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", 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(
                    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(),
            )
            schedule_delete(msg, seconds=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", user_id=user.id, username=username)
        logger.info(f"🚪 离群: @{username} ({user.id})")


def register_welcome(app):
    """注册入群欢迎/退群告别"""
    app.add_handler(ChatMemberHandler(handle_chat_member, ChatMemberHandler.CHAT_MEMBER))