#!/usr/bin/env python3 """ 🌐 وب‌سایت ورود خودکار به روبیکا - تک فایل """ import subprocess import sys import os # -------------------- نصب خودکار پیش‌نیازها -------------------- def install_if_missing(): packages = { "fastapi": "fastapi", "uvicorn": "uvicorn[standard]", "websockets": "websockets", "rubpy": "rubpy", } for module, pkg in packages.items(): try: __import__(module) except ImportError: print(f"📦 نصب {pkg}...") try: subprocess.check_call([sys.executable, "-m", "pip", "install", pkg]) except Exception as e: print(f"⚠️ خطا در نصب {pkg}: {e}") install_if_missing() import asyncio import builtins import queue import threading from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse import uvicorn try: from rubpy import Client LIBRARY_NAME = "rubpy" except ImportError: try: from pyshad import Client LIBRARY_NAME = "pyshad" except ImportError: print("❌ هیچ کتابخانه‌ای نصب نشد.") sys.exit(1) # -------------------- HTML جاسازی‌شده -------------------- HTML_PAGE = """ ورود خودکار به روبیکا

ورود خودکار به روبیکا

با شماره تلفن خود وارد شوید

در حال اتصال...
🔒 اطلاعات شما به‌صورت محلی ذخیره می‌شود
""" # -------------------- اپلیکیشن FastAPI -------------------- app = FastAPI(title="Rubika Auto Login") _ORIGINAL_INPUT = builtins.input class LoginSession: def __init__(self, websocket, loop): self.ws = websocket self.loop = loop self.code_queue = queue.Queue() self.active = True def send_sync(self, data): if not self.active: return try: asyncio.run_coroutine_threadsafe( self.ws.send_json(data), self.loop ) except Exception: pass def ask_input(self, prompt=""): self.send_sync({"type": "input_required", "prompt": str(prompt)}) return self.code_queue.get() def extract_user_info(me): info = {} if me is None: return info if isinstance(me, dict): for k, v in me.items(): info[str(k)] = v if isinstance(v, (str, int, float, bool, type(None))) else str(v) return info if hasattr(me, "to_dict") and callable(getattr(me, "to_dict")): try: d = me.to_dict() for k, v in d.items(): info[str(k)] = v if isinstance(v, (str, int, float, bool, type(None))) else str(v) return info except Exception: pass if hasattr(me, "__dict__"): for k, v in vars(me).items(): if isinstance(v, (str, int, float, bool, type(None))): info[k] = v elif isinstance(v, (list, tuple)): info[k] = [str(x) for x in v] elif isinstance(v, dict): info[k] = {str(kk): str(vv) for kk, vv in v.items()} else: try: info[k] = str(v) except Exception: info[k] = "" for attr in dir(me): if attr.startswith("_") or attr in info: continue try: v = getattr(me, attr) if callable(v): continue if isinstance(v, (str, int, float, bool, type(None))): info[attr] = v else: info[attr] = str(v) except Exception: continue return info def run_login_thread(session, phone): def patched_input(prompt=""): return session.ask_input(prompt) builtins.input = patched_input loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: loop.run_until_complete(_login_flow(session, phone)) except Exception as e: session.send_sync({"type": "error", "message": f"خطا: {str(e)}"}) finally: builtins.input = _ORIGINAL_INPUT try: loop.close() except Exception: pass session.active = False async def _login_flow(session, phone): session.send_sync({"type": "log", "message": "🔌 در حال اتصال به سرور روبیکا..."}) session.send_sync({"type": "log", "message": f"📱 شماره: {phone}"}) try: bot = Client(name="web_rubika_session", phone_number=phone) except TypeError: bot = Client(name="web_rubika_session") async with bot: session.send_sync({"type": "log", "message": "✅ ورود با موفقیت انجام شد!"}) session.send_sync({"type": "log", "message": "📋 در حال دریافت اطلاعات حساب..."}) try: me = await bot.get_me() user_info = extract_user_info(me) session.send_sync({"type": "user_info", "data": user_info}) except Exception as e: session.send_sync({"type": "log", "message": f"⚠️ خطا در دریافت اطلاعات: {e}"}) try: await bot.send_message("me", "🎉 ورود با موفقیت انجام شد!") session.send_sync({"type": "log", "message": "📤 پیام تست ارسال شد"}) except Exception as e: session.send_sync({"type": "log", "message": f"⚠️ خطا در ارسال پیام: {e}"}) session.send_sync({"type": "success", "message": "ورود کامل شد"}) @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() loop = asyncio.get_event_loop() session = LoginSession(websocket, loop) try: await websocket.send_json({"type": "ready", "library": LIBRARY_NAME}) while True: data = await websocket.receive_json() msg_type = data.get("type") if msg_type == "start_login": phone = (data.get("phone") or "").strip() if not phone: await websocket.send_json({ "type": "error", "message": "شماره تلفن خالی است" }) continue t = threading.Thread( target=run_login_thread, args=(session, phone), daemon=True ) t.start() elif msg_type == "code_input": session.code_queue.put(data.get("code", "")) except WebSocketDisconnect: session.active = False except Exception: session.active = False @app.get("/") async def index(): return HTMLResponse(HTML_PAGE) if __name__ == "__main__": print("=" * 60) print(" 🌐 سرور وب ورود به روبیکا") print(f" 📚 کتابخانه: {LIBRARY_NAME}") print(" 🚀 آدرس: http://localhost:8000") print("=" * 60) uvicorn.run(app, host="0.0.0.0", port=8000)