# --- START OF FILE core.py ---
import asyncio
import html
import logging
import re
import httpx
import io
from datetime import datetime, timedelta, timezone
from urllib.parse import quote
from telegram import Update, MessageOriginChannel, ChatMember, Chat, User, InlineKeyboardButton, InlineKeyboardMarkup, ChatPermissions, ReactionTypeEmoji
from telegram.ext import ContextTypes
from telegram.constants import ParseMode, MessageEntityType
from telegram.error import Forbidden, BadRequest
import config
import db
from . import utils
from .batch import BatchDeleter
logger = logging.getLogger(__name__)
# --- ১. হেল্পার ফাংশনসমূহ ---
async def extract_text_with_own_api(image_bytes: bytes, api_url: str) -> str:
try:
files = {'file': ('image.jpg', image_bytes, 'image/jpeg')}
async with httpx.AsyncClient() as client:
response = await client.post(api_url, files=files, timeout=60.0)
if response.status_code == 200:
return response.json().get("extracted_text", "").strip()
return ""
except Exception as e:
logger.error(f"OCR API Error: {e}")
return ""
async def scheduled_delete(context, chat_id, message_id, delay):
try:
await asyncio.sleep(delay)
await context.bot.delete_message(chat_id=chat_id, message_id=message_id)
except: pass
# 🔥 আপডেট করা স্মার্ট লক (ট্রিগার মেসেজ আইডি সেভ করবে)
async def alert_missing_permissions(context, chat_id, user_id, trigger_msg_id=None):
redis = utils.get_redis_client_for_chat(context, chat_id)
if not redis: return
if await redis.get(f"perm_warn:{chat_id}"): return
try:
lang = await db.get_user_lang_from_db(context, user_id)
msg_text = utils.get_text("deletion_fail_group_warning", lang)
sent = await context.bot.send_message(chat_id, msg_text, parse_mode='HTML')
await redis.set(f"perm_warn:{chat_id}", "1", ex=3600)
await redis.set(f"perm_warn_msg_id:{chat_id}", str(sent.message_id), ex=3600)
# যে লিংকটি ডিলিট করতে পারেনি সেটিও সেভ রাখা হলো
if trigger_msg_id:
await redis.set(f"perm_trigger_msg_id:{chat_id}", str(trigger_msg_id), ex=3600)
except: pass
# --- ২. ডিলিট করার মূল লজিক ---
async def process_deletion_and_tasks(context, user, chat, msg_id, settings, reason_keys, content):
batch_deleter = context.bot_data.get('batch_deleter')
if batch_deleter:
await batch_deleter.delete(chat.id, msg_id)
else:
try:
await context.bot.delete_message(chat.id, msg_id)
except BadRequest as e:
if "not found" not in str(e).lower():
asyncio.create_task(alert_missing_permissions(context, chat.id, user.id, msg_id))
except Forbidden:
asyncio.create_task(alert_missing_permissions(context, chat.id, user.id, msg_id))
asyncio.create_task(_background_processing(context, user, chat, settings, reason_keys, content))
async def _background_processing(context, user, chat, settings, reason_keys, content):
redis_client = utils.get_redis_client_for_chat(context, chat.id)
is_join_leave = 'censor_reason_join_leave' in reason_keys
is_bot_link = False
if 'censor_reason_link' in reason_keys or 'censor_reason_bio_link' in reason_keys:
if re.search(r'(@[\w_]+bot\b|(?:t\.me|telegram\.me)\/[\w_]+bot\b)', content, re.IGNORECASE):
is_bot_link = True
if not is_join_leave and not is_bot_link:
if re.search(r'(?:t\.me|telegram\.me)', content, re.IGNORECASE) or 'censor_reason_word' in reason_keys:
asyncio.create_task(utils.send_deletion_report(context, user, chat, reason_keys[0], content))
await utils.update_user_activity_score(context, chat.id, user.id, 'deletion', reason_keys, content)
if settings.get('mute_on_link_24_h', False) and ('censor_reason_link' in reason_keys or 'censor_reason_bio_link' in reason_keys):
try:
mute_until = datetime.now(timezone.utc) + timedelta(hours=24)
await context.bot.restrict_chat_member(chat.id, user.id, ChatPermissions(can_send_messages=False), until_date=mute_until)
except: pass
if redis_client:
cooldown_key = f"warn_cooldown:{chat.id}:{user.id}"
if not await redis_client.get(cooldown_key):
try:
lang = await db.get_user_lang_from_db(context, user.id)
reason_msg = utils.get_text(reason_keys[0], lang)
warn_text = utils.get_text("censor_warning", lang).format(user_mention=user.mention_html(), reason=reason_msg, bot_username=context.bot.username)
sent_warn = await context.bot.send_message(chat.id, warn_text, parse_mode=ParseMode.HTML)
await redis_client.set(cooldown_key, "1", ex=600)
asyncio.create_task(utils.delete_message_after_delay(context, chat.id, sent_warn.message_id, config.CENSOR_WARNING_DELETE_SECONDS))
except: pass
async def process_image_with_ocr_in_background(context, chat_id, user_id, message_id, image_bytes):
try:
ocr_semaphore = context.bot_data.get('ocr_semaphore')
async with ocr_semaphore:
ocr_text = await extract_text_with_own_api(image_bytes, config.OCR_API_URL)
if ocr_text:
normalized_text = re.sub(r'\s+', '', ocr_text).lower()
if config.LINK_PATTERN_COMPILED.search(ocr_text) or any(indicator in normalized_text for indicator in config.LINK_INDICATOR_KEYWORDS):
settings = await utils.get_settings_from_cache_and_update_regex(context, chat_id)
try:
chat_obj = await context.bot.get_chat(chat_id)
member = await context.bot.get_chat_member(chat_id, user_id)
asyncio.create_task(process_deletion_and_tasks(context, member.user, chat_obj, message_id, settings, ['censor_reason_link'], ocr_text))
except: pass
except: pass
# --- ৩. প্রধান মেসেজ হ্যান্ডেলার ---
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
msg, chat, user = update.effective_message, update.effective_chat, update.effective_user
if not all([msg, chat, user]): return
redis = utils.get_redis_client_for_chat(context, chat.id)
if redis and await redis.get(f"perm_warn:{chat.id}"): return
is_bot_admin = await utils.is_group_admin(context, chat.id, context.bot.id)
if not is_bot_admin and chat.type != 'private':
asyncio.create_task(alert_missing_permissions(context, chat.id, user.id, msg.message_id))
return
if chat.type == 'private':
if not user.is_bot and not msg.new_chat_members:
asyncio.create_task(db.buffer_user_stats(context, user.id, user.first_name, user.username))
return
if msg.new_chat_members:
is_adder_admin = await utils.is_group_admin(context, chat.id, user.id)
for member in msg.new_chat_members:
if member.is_bot and member.id != context.bot.id and not is_adder_admin:
try:
await context.bot.ban_chat_member(chat.id, member.id)
await msg.delete()
return
except: pass
is_edit = update.edited_message is not None
asyncio.create_task(db.track_chat_member(context.bot_data.get('db_pool'), chat.id, user.id))
if not is_edit:
asyncio.create_task(db.buffer_user_stats(context, user.id, user.first_name, user.username))
settings = await utils.get_settings_from_cache_and_update_regex(context, chat.id)
owner_id = await utils.get_group_owner_id(context, chat.id)
is_exempt = (user.id == owner_id) or (user.id == context.bot_data.get('BOT_OWNER_ID')) or (settings.get('allow_admins_manage') and await utils.is_group_admin(context, chat.id, user.id))
if is_exempt and not settings.get('enable_traffic_control', False): return
if settings.get('enable_traffic_control', False): asyncio.create_task(utils.check_traffic_control(context, chat.id))
if is_exempt: return
# --- লিংক এবং মেনশন স্মার্ট চেক ---
text_content = (msg.text or msg.caption or "").strip()
# --- লিংক এবং মেনশন স্মার্ট চেক ---
# আগে শুধু text_content চেক হতো, এখন পুরো 'msg' অবজেক্ট পাঠাবো
is_link = utils.has_link_in_text(msg)
# মেনশন বা ইউজারনেম চেক
if not is_link:
text_content = (msg.text or msg.caption or "").strip()
entities = msg.entities or msg.caption_entities or []
# এনটিটির ভেতর মেনশন আছে কি না দেখা
has_mention_entity = any(ent.type in [MessageEntityType.MENTION, MessageEntityType.TEXT_MENTION] for ent in entities)
if has_mention_entity:
if not settings.get('allow_usernames', False):
is_link = True # ইউজারনেম এলাউ না থাকলে ডিলিট হবে
else:
# যদি এলাউ থাকে কিন্তু সেটি চ্যানেল কি না চেক করা
mentions = config.MENTION_PATTERN_COMPILED.findall(text_content)
for mention in mentions:
if await utils.is_channel_or_group_username(context, mention):
is_link = True
break
# --- ১৮৫-১৯২ এর পর যেখানে violations চেক শুরু হবে ---
violations = []
# ১. লিংক ভায়োলেশন অ্যাড করা (এটি যোগ করা জরুরি)
if is_link:
violations.append(('link', 'censor_reason_link'))
# ২. নিষিদ্ধ শব্দ চেক
if (fw_regex := settings.get('fw_regex')) and fw_regex.search(text_content):
violations.append(('word', 'censor_reason_word'))
# ৩. ফরোয়ার্ড মেসেজ চেক
if settings.get('block_channel_forwards') and isinstance(msg.forward_origin, MessageOriginChannel):
violations.append(('forward', 'censor_reason_forward'))
# ৪. কাউন্টার এবং মিউট লজিক (সবার জন্য)
if violations:
if redis:
spam_key = f"spam_violation_counter:{chat.id}:{user.id}"
current_count = await redis.incr(spam_key)
if current_count == 1:
await redis.expire(spam_key, 600) # ১০ মিনিট
if current_count >= 11:
try:
# পারমানেন্ট মিউট
await context.bot.restrict_chat_member(chat.id, user.id, ChatPermissions(can_send_messages=False))
except: pass
# ডিলিট এবং রিপোর্ট পাঠানোর টাস্ক
asyncio.create_task(process_deletion_and_tasks(context, user, chat, msg.message_id, settings, [v[1] for v in violations], text_content))
return
# --- স্প্যাম স্ক্যানার (ওনার মেনশন ও বাটন জাম্প লিঙ্ক) ---
if settings.get('enable_spamscan') and config.SPAM_PHRASES_COMPILED.search(text_content):
try:
# ১. স্মার্ট জাম্প লিঙ্ক তৈরি (সরাসরি আসল স্প্যাম মেসেজে যাওয়ার জন্য)
if chat.username:
jump_url = f"https://t.me/{chat.username}/{msg.message_id}"
else:
chat_id_clean = str(chat.id).replace("-100", "")
jump_url = f"https://t.me/c/{chat_id_clean}/{msg.message_id}"
# ২. গ্রুপ ওনারকে খুঁজে বের করা (যাতে নোটিফিকেশন যায়)
owner_id = await utils.get_group_owner_id(context, chat.id)
mention_text = ""
if owner_id:
# ওনারের আইডি দিয়ে লিঙ্ক তৈরি, এতে 100% নোটিফিকেশন বাজবে
mention_text = f"📢 Attention: Admin\n"
# ৩. ক্লিন অ্যালার্ট টেক্সট
warn_text = f"🚨 Spam Alert!\n{mention_text}⚠️ একটি স্প্যাম মেসেজ পাওয়া গেছে।"
# ৪. বাটন সেটআপ (এখানে ব্র্যাকেট ফিক্স করা হয়েছে)
btns = [[InlineKeyboardButton("👀 View Message", url=jump_url)],[InlineKeyboardButton("🔇 Mute & Delete", callback_data=f"spam_act_mute_{user.id}_{msg.message_id}"),
InlineKeyboardButton("❌ Cancel", callback_data="spam_act_cancel")]
]
# ৫. মেসেজ পাঠানো (কোনো রিপ্লাই বা নীল বক্স আসবে না)
await context.bot.send_message(
chat_id=chat.id,
text=warn_text,
parse_mode=ParseMode.HTML,
reply_markup=InlineKeyboardMarkup(btns),
disable_web_page_preview=True
)
except Exception as e:
logger.error(f"Spam Scan Error: {e}")
return
if settings.get('enable_sentiment', False) and text_content:
senti = await asyncio.to_thread(utils.analyze_sentiment, text_content)
if senti:
try: await msg.set_reaction(reaction=[ReactionTypeEmoji("❤️" if senti == 'positive' else "👎")])
except: pass
if settings.get('block_bio_links', True):
bio_status_key = f"bio_status:{user.id}"
cached_bio = await redis.get(bio_status_key) if redis else None
if cached_bio == "bad":
asyncio.create_task(process_deletion_and_tasks(context, user, chat, msg.message_id, settings, ['censor_reason_bio_link'], "Bio contains links (cached)"))
return
if not cached_bio:
async with context.bot_data.get('api_semaphore'):
try:
profile = await context.bot.get_chat(user.id)
if profile.bio and utils.has_link_in_text(profile.bio):
if redis: await redis.set(bio_status_key, "bad", ex=600)
asyncio.create_task(process_deletion_and_tasks(context, user, chat, msg.message_id, settings, ['censor_reason_bio_link'], f"Bio: {profile.bio}"))
return
if redis: await redis.set(bio_status_key, "safe", ex=600)
except: pass
if settings.get('delete_join_messages') and (msg.new_chat_members or msg.left_chat_member):
asyncio.create_task(process_deletion_and_tasks(context, user, chat, msg.message_id, settings, ['censor_reason_join_leave'], "Join/Leave message"))
return
if text_content:
filters_cache = await db.get_chat_filters(context, chat.id)
for keyword, fdata in filters_cache.items():
if keyword in text_content.lower():
args = {'chat_id': chat.id, 'reply_to_message_id': msg.message_id}
if fdata['type'] == 'photo': await context.bot.send_photo(photo=fdata['file_id'], caption=fdata['text'], **args)
else: await context.bot.send_message(text=fdata['text'], parse_mode=ParseMode.HTML, **args)
return
if text_content.startswith('#'):
if note_data := await db.get_note_from_db(context, chat.id, text_content.split(' ')[0][1:].lower()):
await context.bot.send_message(chat.id, note_data['text'], parse_mode=ParseMode.HTML, reply_to_message_id=msg.message_id)
auto_del = settings.get('auto_delete_seconds', 0)
if auto_del > 0:
is_media = (msg.photo or msg.video or msg.document or msg.voice or msg.audio or msg.sticker or msg.animation or msg.video_note)
if is_media: asyncio.create_task(scheduled_delete(context, chat.id, msg.message_id, auto_del))
if settings.get('enable_ocr_scan') and msg.photo and config.OCR_API_URL:
try:
photo_file = await msg.photo[-1].get_file()
photo_bytes = await photo_file.download_as_bytearray()
asyncio.create_task(process_image_with_ocr_in_background(context, chat_id, user.id, msg.message_id, bytes(photo_bytes)))
except: pass
# --- ৮. মেম্বার ও বটের স্ট্যাটাস হ্যান্ডেলার ---
async def handle_member_status_change(update: Update, context: ContextTypes.DEFAULT_TYPE):
if not update.chat_member: return
chat = update.chat_member.chat
user = update.chat_member.new_chat_member.user
new_status = update.chat_member.new_chat_member.status
old_status = update.chat_member.old_chat_member.status
redis = utils.get_redis_client_for_chat(context, chat.id)
if redis:
if new_status in [ChatMember.ADMINISTRATOR, ChatMember.OWNER]:
await redis.set(f"is_admin:{chat.id}:{user.id}", "1", ex=86400)
else:
await redis.set(f"is_admin:{chat.id}:{user.id}", "0", ex=86400)
await redis.delete(f"bio_status:{user.id}")
if new_status == ChatMember.OWNER:
await redis.set(f"group_owner:{chat.id}", str(user.id), ex=86400)
if new_status == ChatMember.MEMBER and old_status not in [ChatMember.MEMBER, ChatMember.ADMINISTRATOR, ChatMember.OWNER] and not user.is_bot:
await db._add_user_to_db_core(context.bot_data.get('db_pool'), user.id, user.first_name, user.username)
try:
settings = await utils.get_settings_from_cache_and_update_regex(context, chat.id)
welcome_data = settings.get('welcome_data') or {}
if redis:
last_msg_id = await redis.get(f"last_welcome_msg:{chat.id}")
if last_msg_id:
try: await context.bot.delete_message(chat.id, int(last_msg_id))
except: pass
raw_text = welcome_data.get('text') or (
"👋 Welcome {name} to our Group!\n\n"
"🛡️ I am your Group Protector. I will keep this chat safe from spam.\n"
"📝 Please follow the rules to avoid being muted or banned."
)
welcome_text = raw_text.replace("{name}", user.mention_html())
# --- স্মার্ট বাটন জেনারেট লজিক (এনকোডিং ফিক্স সহ) ---
buttons_list = welcome_data.get('buttons', [])
keyboard = []
for i, btn in enumerate(buttons_list):
btn_name = btn['name']
btn_content = btn['content'].strip()
if btn_content.startswith('share:'):
# এনকোডিং ফিক্সের জন্য quote ইমপোর্ট করা হলো
from urllib.parse import quote
link_to_share = btn_content.replace('share:', '').strip()
share_url = f"https://t.me/share/url?url={quote(link_to_share)}"
keyboard.append([InlineKeyboardButton(btn_name, url=share_url)])
elif btn_content.startswith('http'):
keyboard.append([InlineKeyboardButton(btn_name, url=btn_content)])
else:
keyboard.append([InlineKeyboardButton(btn_name, callback_data=f"w_pop_{i}")])
reply_markup = InlineKeyboardMarkup(keyboard) if keyboard else None
sent = None
if settings.get('welcome_mode', 'text') == 'card':
photos = await user.get_profile_photos(limit=1)
pfp = None
if photos.total_count > 0:
try:
pfp_file = await photos.photos[0][-1].get_file()
pfp = bytes(await pfp_file.download_as_bytearray())
except: pfp = None
card = await utils.generate_welcome_card(user.first_name, pfp)
if card:
sent = await context.bot.send_photo(chat.id, photo=card, caption=welcome_text, parse_mode=ParseMode.HTML, reply_markup=reply_markup)
if not sent:
sent = await context.bot.send_message(chat.id, text=welcome_text, parse_mode=ParseMode.HTML, disable_web_page_preview=True, reply_markup=reply_markup)
if sent and redis:
await redis.set(f"last_welcome_msg:{chat.id}", sent.message_id)
ds = welcome_data.get('delete_seconds', 0)
if ds > 0:
asyncio.create_task(utils.delete_message_after_delay(context, chat.id, sent.message_id, ds))
except Exception as e:
logger.error(f"Welcome Error: {e}")
# 🔥 অটো আনলক ও অটো-ডিলিট ট্রিগার মেসেজ
async def handle_bot_status_change(update: Update, context: ContextTypes.DEFAULT_TYPE):
result = update.my_chat_member
if not result: return
chat = result.chat
new_status = result.new_chat_member.status
pool = context.bot_data.get('db_pool')
if new_status in [ChatMember.MEMBER, ChatMember.ADMINISTRATOR]:
await db._add_chat_to_db_core(pool, chat.id)
if new_status == ChatMember.ADMINISTRATOR:
redis = utils.get_redis_client_for_chat(context, chat.id)
if redis:
# ১. বটের নিজের Admin Status Cache আপডেট করা (যাতে সাথে সাথে কাজ শুরু করে)
await redis.set(f"is_admin:{chat.id}:{context.bot.id}", "1", ex=86400)
# ২. পারমিশন ওয়ার্নিং লক রিলিজ করা
await redis.delete(f"perm_warn:{chat.id}")
# ৩. আগের ওয়ার্নিং মেসেজ ডিলিট করা
if (last_warn := await redis.get(f"perm_warn_msg_id:{chat.id}")):
try:
await context.bot.delete_message(chat.id, int(last_warn))
await redis.delete(f"perm_warn_msg_id:{chat.id}")
except: pass
# ৪. 🔥 যে লিংকটির কারণে পারমিশন এরর এসেছিল সেটি ডিলিট করা
if (trigger_link_id := await redis.get(f"perm_trigger_msg_id:{chat.id}")):
try:
await context.bot.delete_message(chat.id, int(trigger_link_id))
await redis.delete(f"perm_trigger_msg_id:{chat.id}")
except: pass
elif new_status in [ChatMember.LEFT, ChatMember.KICKED, ChatMember.BANNED]:
if pool:
try:
async with pool.acquire() as conn: await conn.execute("DELETE FROM chats WHERE chat_id = $1", chat.id)
except: pass
async def handle_edited_message(update, context):
await handle_message(update, context)
# --- END OF FILE ---