# -*- coding: utf-8 -*- # GAMA Bot - Rubika (corrected per official docs) import os, json, random, logging, traceback from datetime import datetime import urllib.request, urllib.error # ==================== تنظیمات ==================== BOT_TOKEN = "CGHABG0SKVLELHZXPXYFYXIGURNQGGSQGXFXDXWGWMHZLLVPMLDZWHORXEUIBVUO" DEEPSEEK_KEY = "sk-4d3fd3ac7e2c4bafabb93ed7ae7672a0" CHANNEL_USERNAME = "@GammaRobot" CHANNEL_LINK = "https://rubika.ir/GammaRobot" # طبق مستندات: https://botapi.rubika.ir/v3/{token}/{method} BASE_URL = "https://botapi.rubika.ir/v3/" + BOT_TOKEN DEEPSEEK_API = "https://api.deepseek.com/chat/completions" BASE_DIR = os.path.dirname(os.path.abspath(__file__)) STATE_FILE = os.path.join(BASE_DIR, "state.json") LOG_FILE = os.path.join(BASE_DIR, "bot.log") logging.basicConfig( filename=LOG_FILE, level=logging.INFO, format='%(asctime)s | %(levelname)s | %(message)s', encoding='utf-8' ) PERSONALITIES = { "عادی": "تو یه ربات دوستانه توی گروه روبیکا هستی. با کاربرا گرم و صمیمی حرف می‌زنی. جواب‌هات کوتاه و طبیعی باشه، حداکثر ۲ جمله.", "بی ادب": "تو یه ربات بی‌ادب و شوخ توی گروه روبیکا هستی. رک و راست حرف می‌زنی، مسخره می‌کنی. جواب‌هات کوتاه و بامزه باشه.", "مؤدب": "تو یه ربات فوق‌العاده مؤدب و محترم هستی. با احترام کامل حرف می‌زنی.", "شوخ": "تو یه ربات شوخ و طنزپرداز هستی. هر حرفی رو با شوخی جواب می‌دی.", "فلسفی": "تو یه ربات فلسفی و عمیق هستی. با جملات فلسفی جواب می‌دی.", "عاشق": "تو یه ربات رمانتیک و عاشق هستی. با کلمات عاشقانه جواب می‌دی.", } JOKES = [ "یه روز یه برنامه‌نویس میره رستوران، گارسون میگه چی میل دارید؟ میگه یه while(true) بیار! 😂", "چرا برنامه‌نویسا عینک می‌زنن؟ چون C# نمی‌بینن! 🤓", "دیتابیس به برنامه‌نویس گفت: چرا اینقدر دیر میای؟ گفت: index نداشتم! 😅", "چرا برنامه‌نویسا از تاریکی میترسن؟ چون توی debug گیر میکنن! 👻", ] TRUTHS = [ "آخرین دروغی که گفتی چی بود؟ 😏", "تا حالا کسی رو دوست داشتی و بهش نگفتی؟ 💘", "یه راز که کسی نمیدونه رو بگو! 🤫", "آخرین بار کی گریه کردی و چرا؟ 😢", ] DARES = [ "اسم ۵ تا از دوستات رو بدون مکث بگو! 🗣", "صدای یه حیوون رو تقلید کن! 🐶", "نظرت رو راجع به نفر بالایی بگو! 👀", ] # ==================== HTTP ==================== def api_post(method, data=None): url = BASE_URL + "/" + method if data is None: data = {} try: body = json.dumps(data).encode('utf-8') req = urllib.request.Request(url, data=body, method='POST') req.add_header('Content-Type', 'application/json') with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read().decode('utf-8')) except urllib.error.HTTPError as e: try: body = e.read().decode('utf-8') logging.error("API %s HTTP %s: %s", method, e.code, body[:500]) except Exception: logging.error("API %s HTTP %s", method, e.code) return None except Exception as e: logging.error("API %s err: %s", method, e) return None def send_message(chat_id, text, reply_to=None, keypad=None): data = {"chat_id": chat_id, "text": text[:4000]} if reply_to: data["reply_to_message_id"] = reply_to if keypad: data["chat_keypad"] = keypad data["chat_keypad_type"] = "New" return api_post("sendMessage", data) def get_updates(start_id=None): data = {} if start_id: data["start_id"] = str(start_id) r = api_post("getUpdates", data) if not r: return [] try: # طبق مستندات: result.updates احتمالاً result = r.get("result", {}) if isinstance(result, dict): return result.get("updates", []) if isinstance(result, list): return result except Exception: pass return [] # ==================== DeepSeek ==================== def ai_chat(user_message, user_name, personality_key, history=None): personality = PERSONALITIES.get(personality_key, PERSONALITIES["عادی"]) messages = [ {"role": "system", "content": personality + "\nاسم کاربر: " + user_name + "\nفقط فارسی جواب بده."}, ] if history: messages.extend(history[-6:]) messages.append({"role": "user", "content": user_message}) body = { "model": "deepseek-chat", "messages": messages, "max_tokens": 200, "temperature": 0.9, } try: b = json.dumps(body).encode('utf-8') req = urllib.request.Request(DEEPSEEK_API, data=b, method='POST') req.add_header('Content-Type', 'application/json') req.add_header('Authorization', 'Bearer ' + DEEPSEEK_KEY) with urllib.request.urlopen(req, timeout=30) as r: res = json.loads(r.read().decode('utf-8')) return res["choices"][0]["message"]["content"].strip() except Exception as e: logging.error("ai_chat: %s", e) return None # ==================== State ==================== def load_state(): if not os.path.exists(STATE_FILE): return {"groups": {}, "last_update_id": None} try: with open(STATE_FILE, 'r', encoding='utf-8') as f: return json.load(f) except Exception: return {"groups": {}, "last_update_id": None} def save_state(state): try: with open(STATE_FILE, 'w', encoding='utf-8') as f: json.dump(state, f, ensure_ascii=False, indent=2) except Exception as e: logging.error("save_state: %s", e) # ==================== Messages ==================== def help_message(): return ( "📚 راهنمای کامل ربات گاما\n" "━━━━━━━━━━━━━━━━\n\n" "🎭 چت با هوش مصنوعی:\n" ".سخنگو — فعال‌سازی\n" ".سخنگو بی ادب — لحن بی‌ادب\n" ".سخنگو مؤدب — لحن محترمانه\n" ".سخنگو شوخ — لحن طنز\n" ".سخنگو فلسفی — لحن عمیق\n" ".سخنگو عاشق — لحن رمانتیک\n" ".سخنگو قفل — ساکت\n" ".سخنگو روشن — فعال\n\n" "🎮 سرگرمی:\n" ".جوک / .تاس / .سکه\n" ".حقیقت / .جرئت / .ساعت / .پینگ\n\n" "📊 عمومی:\n" "فعال / نصب — فعال‌سازی بات\n" ".وضعیت — وضعیت گروه\n" ".راهنما — همین متن\n\n" "🌐 " + CHANNEL_USERNAME ) def force_join_message(): return ( "کاربر گرامی 💯\n" "برای استفاده از بات بصورت دائمی و رایگان حتما در کانال زیر عضو شوید.\n\n" "🟡 " + CHANNEL_USERNAME + "\n\n" "پس از عضویت روی /help1 🔜 کلیک کنید" ) def welcome_message(): return ( "به بات خوش آمدید 🙏❤️\n\n" "من بات گاما مدیریت گروه حرفه‌ای و رایگان هستم.\n\n" "✅👇 نحوه فعالسازی:\n" + CHANNEL_LINK + "\n\n" "🌐 " + CHANNEL_USERNAME + " :چنل اطلاع رسانی\n\n" "امیدوارم از گاما لذت ببرید 🎭❤️" ) # ==================== Handlers ==================== def handle_group_command(text, chat_id, user_name, user_id, msg_id, state): g = state["groups"].setdefault(chat_id, {"chat": False, "mode": "عادی", "history": []}) parts = text.strip().split() cmd = parts[0] if parts else "" arg = " ".join(parts[1:]).strip() if text.strip() in ["فعال", "نصب", "فعال کن", "نصب کن", "/active"]: g["active"] = True send_message(chat_id, "✅ بات گاما با موفقیت فعال شد!\n\n🎭 برای شروع: .سخنگو\n📚 راهنما: .راهنما", msg_id) return True if cmd in [".راهنما", "/help", ".help", "راهنما"]: send_message(chat_id, help_message(), msg_id) return True if cmd == ".وضعیت": info = ( "📊 وضعیت گروه:\n\n" "⚙️ بات: " + ("✅ فعال" if g.get("active") else "❌ غیرفعال") + "\n" "🎭 سخنگو: " + ("روشن ✅" if g.get("chat") else "خاموش ❌") + "\n" "💬 حالت: " + g.get("mode", "عادی") ) send_message(chat_id, info, msg_id) return True if cmd == ".سخنگو": if not arg: g["chat"] = True; g["mode"] = "عادی"; g["history"] = [] send_message(chat_id, "سلام بچه‌ها! 😊 چه خبرا؟", msg_id) return True if arg in ["قفل", "خاموش", "بسه"]: g["chat"] = False; g["history"] = [] send_message(chat_id, "باشه، ساکت شدم 🤐", msg_id) return True if arg in ["روشن", "فعال", "بیدار"]: g["chat"] = True; g["history"] = [] send_message(chat_id, "برگشتم! 😎", msg_id) return True modes = {"بی ادب": "بی ادب", "مؤدب": "مؤدب", "مودب": "مؤدب", "شوخ": "شوخ", "فلسفی": "فلسفی", "عاشق": "عاشق", "عادی": "عادی"} if arg in modes: g["chat"] = True; g["mode"] = modes[arg]; g["history"] = [] starters = { "بی ادب": "خب بابا! 😏 چیه؟ چرا ساکتید؟", "مؤدب": "سلام دوستان عزیز 🌹", "شوخ": "سلام رفقا! 😂 یه جوک بگم؟", "فلسفی": "زندگی چیست؟ 🧠", "عاشق": "سلام عزیزانم 💘", "عادی": "سلام بچه‌ها! 😊", } send_message(chat_id, starters.get(modes[arg], "سلام!"), msg_id) return True if cmd == ".جوک": send_message(chat_id, random.choice(JOKES), msg_id); return True if cmd == ".تاس": send_message(chat_id, "🎲 " + str(random.randint(1, 6)), msg_id); return True if cmd == ".سکه": send_message(chat_id, "🪙 " + random.choice(["شیر 🦁", "خط ✋"]), msg_id); return True if cmd == ".حقیقت": send_message(chat_id, "🎯 " + random.choice(TRUTHS), msg_id); return True if cmd == ".جرئت": send_message(chat_id, "🔥 " + random.choice(DARES), msg_id); return True if cmd == ".ساعت": send_message(chat_id, "🕐 " + datetime.now().strftime("%H:%M:%S"), msg_id); return True if cmd == ".پینگ": send_message(chat_id, "🏓 پونگ!", msg_id); return True if g.get("chat") and not text.startswith("."): reply = ai_chat(text, user_name, g.get("mode", "عادی"), g.get("history", [])) if reply: send_message(chat_id, reply, msg_id) h = g.setdefault("history", []) h.append({"role": "user", "content": text}) h.append({"role": "assistant", "content": reply}) g["history"] = h[-8:] return False def handle_private_command(text, chat_id, user_name, user_id, msg_id, state): t = text.strip().lower() if t in ["/start", "شروع", "استارت"]: send_message(chat_id, welcome_message(), msg_id) return True if t in ["/help", "help", "راهنما"]: send_message(chat_id, help_message(), msg_id) return True return False # ==================== Update Processor ==================== def process_update(upd, state): try: # طبق مستندات: {"update": {"type":"NewMessage", "chat_id":"...", "new_message": {...}}} u = upd.get("update") or upd u_type = u.get("type", "") if u_type != "NewMessage": return msg = u.get("new_message") or {} if not msg: return text = msg.get("text", "") or "" if not text.strip(): return chat_id = u.get("chat_id") or msg.get("chat_id") or "" user_id = msg.get("sender_id") or "" user_name = msg.get("sender_name") or "کاربر" msg_id = msg.get("message_id") or "" if not chat_id: return # پیوی یا گروه is_private = str(chat_id).startswith("u") or str(chat_id).startswith("U") if is_private: handle_private_command(text, chat_id, user_name, user_id, msg_id, state) else: if text.startswith(".") or text.strip() in ["فعال", "نصب", "فعال کن", "نصب کن", "راهنما"]: handle_group_command(text, chat_id, user_name, user_id, msg_id, state) else: g = state["groups"].get(chat_id, {}) if g.get("chat"): reply = ai_chat(text, user_name, g.get("mode", "عادی"), g.get("history", [])) if reply: send_message(chat_id, reply, msg_id) h = g.setdefault("history", []) h.append({"role": "user", "content": text}) h.append({"role": "assistant", "content": reply}) g["history"] = h[-8:] except Exception: logging.error("process_update: %s", traceback.format_exc()) # ==================== Main ==================== def main(): state = load_state() last_id = state.get("last_update_id") updates = get_updates(last_id) logging.info("Got %d updates (last_id=%s)", len(updates), last_id) if not updates: return max_id = last_id for upd in updates: try: # هر update ممکنه update_id داشته باشه u = upd.get("update") or upd uid = upd.get("update_id") or u.get("update_id") if uid: max_id = str(uid) if max_id is None else str(max(int(max_id), int(uid))) except Exception: pass process_update(upd, state) if max_id is not None: state["last_update_id"] = max_id save_state(state) if __name__ == "__main__": try: main() except Exception: logging.error("main: %s", traceback.format_exc())