"""Learning system handler — teach the bot to respond.""" from __future__ import annotations import asyncio import json import logging import random import re import time from telegram import Update from telegram.constants import ParseMode from telegram.error import BadRequest, Forbidden from telegram.ext import ContextTypes from bot import db from bot.decorators import admin_only from bot.persian_router import register_mixed logger = logging.getLogger(__name__) @admin_only async def learn_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: msg = update.message if not msg: return chat = update.effective_chat text = msg.text or "" user_id = update.effective_user.id if update.effective_user else 0 # Reply-based: learn from replied message if msg.reply_to_message and "=" not in text.split("--")[0]: replied = msg.reply_to_message trigger = " ".join(context.args).strip().lower() if context.args else "" if not trigger: await msg.reply_text( "ℹ️ دو حالت:\n" "1. روی پیام ریپلی کن + یادبگیر [trigger]\n" "2. یادبگیر [trigger] = [response]", parse_mode=ParseMode.HTML, ) return response = replied.text or replied.caption or "" media_type = "text" file_id = "" if replied.sticker: media_type, file_id = "sticker", replied.sticker.file_id elif replied.photo: media_type, file_id = "photo", replied.photo[-1].file_id elif replied.voice: media_type, file_id = "voice", replied.voice.file_id elif replied.video: media_type, file_id = "video", replied.video.file_id elif replied.animation: media_type, file_id = "animation", replied.animation.file_id if not response and not file_id: await msg.reply_text("این پیام محتوای قابل یادگیری ندارد.") return resp = await db.brain.add( chat.id, trigger, response or file_id, media_type=media_type, file_id=file_id, created_by=user_id, ) icon = {"sticker": "📦", "photo": "📸", "voice": "🎤", "video": "🎬"}.get(media_type, "💬") await msg.reply_text(f"{icon} یاد گرفتم: «{trigger}»") return # Inline: یادبگیر [trigger] = [response] if "=" in text: weight, delay, category = 1.0, 0.0, "general" wm = re.search(r'--weight=(\d+\.?\d*)', text) if wm: weight = max(0.1, min(10.0, float(wm.group(1)))) text = text.replace(wm.group(0), "") dm = re.search(r'--delay=(\d+\.?\d*)', text) if dm: delay = max(0.0, min(5.0, float(dm.group(1)))) text = text.replace(dm.group(0), "") cm = re.search(r'--cat=(\S+)', text) if cm: category = cm.group(1).strip().lower() text = text.replace(cm.group(0), "") parts = text.split("=", 1) if len(parts) < 2: await msg.reply_text("مثال: یادبگیر سلام = سلام علیکم", parse_mode=ParseMode.HTML) return cmd_part = parts[0].strip() response = parts[1].strip() trigger = " ".join(context.args).strip().lower() if context.args else "" if not trigger: cmd_words = cmd_part.split(maxsplit=1) trigger = cmd_words[1].strip().lower() if len(cmd_words) > 1 else "" if not trigger or not response: await msg.reply_text("هم trigger و هم response الزامی هستند.") return resp = await db.brain.add( chat.id, trigger, response, category=category, weight=weight, delay=delay, created_by=user_id, ) flags = [] if category != "general": flags.append(f"📁 {category}") if weight != 1.0: flags.append(f"⚖️ {weight}x") if delay > 0: flags.append(f"⏱ {delay}s") flag_str = " | ".join(flags) await msg.reply_text( f"✅ یاد گرفتم: «{trigger}»\n" f" → {response[:80]}\n" f"{' ' + flag_str if flag_str else ''}" ) return await msg.reply_text( "🧠 سیستم یادگیری\n\n" "یادگیری:\n" "• یادبگیر [trigger] = [response]\n" "• روی پیام ریپلی کن + یادبگیر [trigger]\n\n" "پرچم‌ها:\n" "• --cat=name — دسته‌بندی\n" "• --weight=2 — اولویت (1-10)\n" "• --delay=1 — تأخیر (ثانیه)\n\n" "مدیریت:\n" "• فراموش کن [trigger] — حذف با trigger\n" "• حذف脑 [id] — حذف با شناسه\n" "• لیست脑 — نمایش همه\n" "• جستجو脑 [query] — جستجو\n" "• آمار脑 — آمار\n" "• دسته脑 — دسته‌ها\n" "• پاک脑 — پاک کردن همه\n\n" "متغیرها: {user} {name} {username} {chat} {time} {day} {random} {reaction}\n" "چندپ réponses: پاسخ1 || پاسخ2 || پاسخ3", parse_mode=ParseMode.HTML, ) @admin_only async def unlearn_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: msg = update.message if not msg: return trigger = " ".join(context.args).strip().lower() if context.args else "" if not trigger: await msg.reply_text("مثال: فراموش کن [trigger]", parse_mode=ParseMode.HTML) return count = await db.brain.remove(update.effective_chat.id, trigger) await msg.reply_text(f"🗑 {count} پاسخ برای «{trigger}» حذف شد." if count else f"چیزی برای «{trigger}» پیدا نشد.") @admin_only async def unlearn_id_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: msg = update.message if not msg: return resp_id = None for arg in (context.args or []): try: resp_id = int(arg) break except ValueError: continue if resp_id is None: await msg.reply_text("مثال: حذف脑 [id]", parse_mode=ParseMode.HTML) return count = await db.brain.remove(update.effective_chat.id, "", response_id=resp_id) if count: await msg.reply_text(f"✅ پاسخ #{resp_id} حذف شد.") else: await msg.reply_text("پاسخی با این شناسه پیدا نشد.") async def list_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: msg = update.message if not msg: return chat_id = update.effective_chat.id responses = await db.brain.get_all(chat_id) if not responses: await msg.reply_text("📭 هنوز چیزی یاد نگرفتم.") return groups: dict[str, list] = {} for r in responses: groups.setdefault(r.category, []).append(r) text = f"🧠 مغز: {len(responses)} پاسخ در {len(groups)} دسته\n\n" for cat, items in sorted(groups.items()): text += f"📁 {cat} ({len(items)})\n" for r in items[:3]: media_icon = {"sticker": "📦", "photo": "📸", "voice": "🎤", "video": "🎬"}.get(r.media_type, "💬") text += f" {media_icon} #{r.id} «{r.trigger[:20]}» → {r.response[:30]}\n" if len(items) > 3: text += f" ... و {len(items) - 3} تا بیشتر\n" text += "\n🔍 جستجو脑 [query] برای جستجو" await msg.reply_text(text, parse_mode=ParseMode.HTML) async def search_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: msg = update.message if not msg: return query = " ".join(context.args).strip() if context.args else "" if not query: await msg.reply_text("مثال: جستجو脑 [query]", parse_mode=ParseMode.HTML) return results = await db.brain.search(update.effective_chat.id, query) if not results: await msg.reply_text(f"🔍 نتیجه‌ای برای «{query}» پیدا نشد.") return text = f"🔍 نتایج «{query}»:\n\n" for r in results[:15]: media_icon = {"sticker": "📦", "photo": "📸", "voice": "🎤", "video": "🎬"}.get(r.media_type, "💬") text += f"{media_icon} #{r.id} «{r.trigger[:25]}» → {r.response[:40]} ({r.uses_count}x)\n" if len(results) > 15: text += f"\n... و {len(results) - 15} تا بیشتر" await msg.reply_text(text, parse_mode=ParseMode.HTML) async def stats_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: msg = update.message if not msg: return stats = await db.brain.get_stats(update.effective_chat.id) text = ( f"📊 آمار مغز\n\n" f"📝 مجموع پاسخ‌ها: {stats['total']}\n" f"📁 دسته‌ها: {stats['categories']}\n" f"📸 رسانه‌ای: {stats['media']}\n" ) if stats["most_used"]: text += f"🔥 پرکاربردترین: «{stats['most_used']['trigger_text']}» ({stats['most_used']['uses_count']} بار)\n" await msg.reply_text(text, parse_mode=ParseMode.HTML) async def categories_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: msg = update.message if not msg: return cats = await db.brain.get_categories(update.effective_chat.id) if not cats: await msg.reply_text("هیچ دسته‌ای وجود ندارد.") return text = "📁 دسته‌ها:\n\n" for c in cats: text += f"• {c['category']} — {c['cnt']} پاسخ\n" await msg.reply_text(text, parse_mode=ParseMode.HTML) @admin_only async def clear_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: msg = update.message if not msg: return text = (msg.text or "").strip().lower() if "تأیید" not in text and "confirm" not in text: await msg.reply_text( "⚠️ همه پاسخ‌های یادگرفته شده حذف خواهند شد!\n" "برای تأیید: پاک脑 تأیید", parse_mode=ParseMode.HTML, ) return count = await db.brain.remove_all(update.effective_chat.id) await msg.reply_text(f"🧹 همه {count} پاسخ حذف شدند.") @admin_only async def export_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: msg = update.message if not msg: return data = await db.brain.export_chat(update.effective_chat.id) if not data: await msg.reply_text("چیزی برای خروجی وجود ندارد.") return await msg.reply_text(f"📦 خروجی: {len(data)} پاسخ") import io content = json.dumps(data, ensure_ascii=False, indent=2) file = io.BytesIO(content.encode("utf-8")) file.name = f"brain_{update.effective_chat.id}_{int(time.time())}.json" await msg.reply_document(document=file) @admin_only async def import_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: msg = update.message if not msg or not msg.document: await msg.reply_text("روی یک فایل JSON ریپلی کن با ورودی脑", parse_mode=ParseMode.HTML) return try: file = await context.bot.get_file(msg.document.file_id) content = await file.download_as_bytearray() data = json.loads(content.decode("utf-8")) count = await db.brain.import_chat( update.effective_chat.id, data, created_by=update.effective_user.id, ) await msg.reply_text(f"✅ {count} پاسخ وارد شد.") except Exception as e: await msg.reply_text(f"❌ خطا در ورودی: {e}") async def check_learned(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool: """Check if message matches a learned response.""" message = update.message if not message or not message.text: return False chat_id = update.effective_chat.id if update.effective_chat.type == "private": return False # Save context await db.brain.remember(chat_id, update.effective_user.id, message.text) # Cooldown if await db.brain.check_cooldown(chat_id, "_global", 3): return False # Match result = await db.brain.match(chat_id, message.text) if not result: return False # Per-trigger cooldown if await db.brain.check_cooldown(chat_id, result.response.trigger, 8): return False # Apply template response = db.brain.apply_template( result.response.response, user=update.effective_user, chat=update.effective_chat, ) r = result.response # Multi-response if "||" in response and r.media_type == "text": response = random.choice([p.strip() for p in response.split("||")]) # Delay if r.delay > 0: await asyncio.sleep(r.delay) elif random.random() < 0.3 and len(response) > 15: await asyncio.sleep(random.uniform(0.3, 0.8)) # Send try: if r.media_type == "sticker" and r.file_id: await message.reply_sticker(r.file_id) elif r.media_type == "photo" and r.file_id: await message.reply_photo(r.file_id, caption=response if response and response != r.file_id else None) elif r.media_type == "voice" and r.file_id: await message.reply_voice(r.file_id) elif r.media_type == "video" and r.file_id: await message.reply_video(r.file_id) elif r.media_type == "animation" and r.file_id: await message.reply_animation(r.file_id) else: await message.reply_text(response, parse_mode=ParseMode.HTML) except (BadRequest, Forbidden): try: await message.reply_text(response) except (BadRequest, Forbidden): pass return True def register(app) -> None: register_mixed( app, { "brain": learn_cmd, "learn": learn_cmd, "unbrain": unlearn_cmd, "unlearn": unlearn_cmd, "unlearnid": unlearn_id_cmd, "listbrain": list_cmd, "brains": list_cmd, "searchbrain": search_cmd, "brainstats": stats_cmd, "braincats": categories_cmd, "clearbrain": clear_cmd, "exportbrain": export_cmd, "importbrain": import_cmd, }, { "یادبگیر": learn_cmd, "یاد بگیر": learn_cmd, "به من یاد بده": learn_cmd, "فراموش کن": unlearn_cmd, "حذف یادگیری": unlearn_cmd, "حذف脑": unlearn_id_cmd, "لیست脑": list_cmd, "لیست یادگیری": list_cmd, "یادگرفته‌ها": list_cmd, "چی یاد گرفتی": list_cmd, "جستجو脑": search_cmd, "جستجوی یادگیری": search_cmd, "آمار脑": stats_cmd, "آمار یادگیری": stats_cmd, "دسته脑": categories_cmd, "دسته‌ها": categories_cmd, "پاک脑": clear_cmd, "پاک کردن همه": clear_cmd, "خروجی脑": export_cmd, "ورودی脑": import_cmd, }, )