Telegram关键词监听机器人/关键词转发

yuanma.it定制开发          测试机器人  @cmzt_bot

Telegram关键词监听机器人/关键词转发

多重去除广告机制

class BotManager:
def __init__(self):
self.clients = {}
self.users_data = self.load_users_data()
self.client_dialogs = {}
self.bot = None # 初始化bot属性
self.admin_id = ADMIN_ID # 保存管理员ID

# 防广告机制
self.sender_blacklist = self.load_blacklist() # 从文件加载黑名单
self.user_blacklist = self.load_user_blacklist() # 从文件加载用户黑名单
self.recent_messages = [] # 最近的消息内容
self.sender_message_times = {} # 发送者消息频率记录
self.message_cache = {} # 消息缓存

# 消息格式限制
self.max_message_length = 50 # 最大消息长度
self.max_links_count = 3 # 最大链接数量
self.max_mentions_count = 3 # 最大@数量

def is_spam_sender(self, sender_id):
“””检查发送者是否在黑名单中”””
return sender_id in self.sender_blacklist

def record_spam_sender(self, sender_id):
“””记录广告发送者”””
self.sender_blacklist.add(sender_id)
self.save_blacklist() # 保存到文件
logger.info(f”已将发送者 {sender_id} 加入黑名单”)

def is_similar_message(self, text1, text2, threshold=0.85):
“””检查两条消息的相似度”””
from difflib import SequenceMatcher
ratio = SequenceMatcher(None, text1, text2).ratio()
return ratio > threshold

def is_frequent_sender(self, sender_id):
“””检查发送者是否频繁发送消息”””
current_time = datetime.now()
if sender_id not in self.sender_message_times:
self.sender_message_times[sender_id] = []

# 清理1小时前的记录
self.sender_message_times[sender_id] = [
t for t in self.sender_message_times[sender_id]
if current_time – t < timedelta(hours=1)
]

# 添加当前时间
self.sender_message_times[sender_id].append(current_time)

# 如果1小时内发送超过10条消息，认为是广告发送者
if len(self.sender_message_times[sender_id]) > 10:
self.record_spam_sender(sender_id)
return True
return False

def is_message_recently_sent(self, message_id, group_id, message_text):
“””检查消息是否在30分钟内发送过”””
cache_key = f”{message_text}_{group_id}”
if cache_key in self.message_cache:
send_time = self.message_cache[cache_key]
if datetime.now() – send_time < timedelta(minutes=30):
return True
return False

def record_message_sent(self, message_id, group_id, message_text):
“””记录消息发送时间”””
cache_key = f”{message_text}_{group_id}”
self.message_cache[cache_key] = datetime.now()

# 清理30分钟前的缓存
current_time = datetime.now()
self.message_cache = {
k: v for k, v in self.message_cache.items()
if current_time – v < timedelta(minutes=30)
}

def check_message_format(self, message_text):
“””检查消息格式是否符合要求”””
# 检查消息长度
if len(message_text) > self.max_message_length:
logger.info(f”消息过长（{len(message_text)}字），已忽略”)
return False

# 检查链接数量
import re
links_count = len(re.findall(r’http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+’, message_text))
if links_count > self.max_links_count:
logger.info(f”消息包含过多链接（{links_count}个），已忽略”)
return False

# 检查@数量
mentions_count = message_text.count(‘@’)
if mentions_count > self.max_mentions_count:
logger.info(f”消息包含过多@（{mentions_count}个），已忽略”)
return False

return True

async def handle_new_message(self, event):
“””处理新消息”””
try:
message = event.message
if not message.text:
return

logger.info(f”收到新消息: {message.text[:50]}…”)

# 0. 检查消息格式
if not self.check_message_format(message.text):
return

# 获取发送者信息
sender = await message.get_sender()

# 1. 检查发送者是否在黑名单中
if self.is_spam_sender(sender.id):
logger.info(f”发送者 {sender.id} 在黑名单中，跳过消息”)
return

# 2. 检查发送频率
if self.is_frequent_sender(sender.id):
logger.info(f”发送者 {sender.id} 发送频率过高，已加入黑名单”)
return

# 3. 检查消息相似度
for recent_message in self.recent_messages:
if self.is_similar_message(message.text, recent_message):
logger.info(f”发现相似消息，记录发送者 {sender.id}”)
self.record_spam_sender(sender.id)
return

# 4. 检查30分钟内是否发送过相同内容
if self.is_message_recently_sent(message.id, message.chat_id, message.text):
logger.info(f”相同内容的消息已在30分钟内发送过，跳过”)
return

# 记录当前消息内容
self.recent_messages.append(message.text)
# 只保留最近100条消息
if len(self.recent_messages) > 100:
self.recent_messages.pop(0)