Add Telegram Desktop tdata import script

.gitignore

@@ -4,6 +4,8 @@ __pycache__/
 *.db
 *.sqlite
 *.sqlite3
+*.session
+*.session-journal
 
 .venv/
 env/

README.md

@@ -4,8 +4,17 @@ OverUB is a modular Telegram userbot built around a plugin-first architecture.
 
 ## Quick Start
 1. Install dependencies: `pip install -r requirements.txt`
+   - Optional for QR display: `pip install qrcode`
 2. Edit `config/config.yml`
 3. Run: `python -m __main__`
+4. Optional login modes:
+   - `phone` (default)
+   - `qr` (QR login)
+   - `sms` (force SMS code)
+5. Optional: set `bot.session_string` or `OVERUB_SESSION_STRING` to use a Telethon StringSession
+6. Optional: import Telegram Desktop session:
+   - `pip install opentele`
+   - `python scripts/import-tdata.py --tdata /path/to/tdata`
 
 ## CLI
 - `python -m __main__ create-plugin <name>`

core/app.py

@@ -112,7 +112,14 @@ class OverUB:
         await self.events.emit("on_startup")
         await self.database.connect()
         await self.migrations.apply(self)
-        await self.client.connect()
+        bot_cfg = self.config.get().get("bot", {})
+        await self.client.connect(
+            login_mode=str(bot_cfg.get("login_mode", "phone")),
+            phone=str(bot_cfg.get("phone", "")),
+            qr_path=bot_cfg.get("login_qr_path"),
+            qr_open=bool(bot_cfg.get("login_qr_open", False)),
+            session_string=bot_cfg.get("session_string"),
+        )
         await self.client.attach_handlers(self)
         await self._load_builtin_modules()
         await self._load_plugins()

core/client.py

@@ -1,6 +1,12 @@
 from __future__ import annotations
 
-from typing import Any
+import asyncio
+import subprocess
+import sys
+import threading
+import webbrowser
+from pathlib import Path
+from typing import Any, Optional
 
 from core.logger import get_logger
 
@@ -15,14 +21,53 @@ class ClientWrapper:
         self.session_name = session_name
         self.client: Any = None
 
-    async def connect(self) -> None:
+    async def connect(
+        self,
+        login_mode: str = "phone",
+        phone: Optional[str] = None,
+        qr_path: Optional[str] = None,
+        qr_open: bool = False,
+        session_string: Optional[str] = None,
+    ) -> None:
         try:
             from telethon import TelegramClient
+            from telethon.errors import SessionPasswordNeededError
+            from telethon.sessions import StringSession
+            from telethon.tl import functions, types
         except ImportError as exc:
             raise RuntimeError("Telethon not installed") from exc
 
+        if session_string:
+            self.client = TelegramClient(StringSession(session_string), self.api_id, self.api_hash)
+        else:
             self.client = TelegramClient(self.session_name, self.api_id, self.api_hash)
-        await self.client.start()
+        if login_mode == "qr":
+            await self.client.connect()
+            if await self.client.is_user_authorized():
+                logger.info("Telethon client already authorized")
+                return
+            try:
+                qr = await self.client.qr_login()
+            except AttributeError:
+                logger.warning("QR login not supported by this Telethon version, falling back to phone login")
+                await self.client.start(phone=phone)
+                return
+            self._print_qr(qr.url, qr_path=qr_path, qr_open=qr_open)
+            try:
+                await qr.wait()
+            except SessionPasswordNeededError:
+                password = await self._prompt("Two-step verification enabled. Enter password: ")
+                await self.client.sign_in(password=password)
+        elif login_mode in {"sms", "code", "phone"}:
+            if not phone:
+                raise RuntimeError("Phone number required for login")
+            await self.client.connect()
+            if await self.client.is_user_authorized():
+                logger.info("Telethon client already authorized")
+                return
+            await self._login_by_code(phone, force_sms=login_mode == "sms", functions=functions, types=types)
+        else:
+            await self.client.start(phone=phone)
         logger.info("Telethon client connected")
 
     async def attach_handlers(self, app: Any) -> None:
@@ -98,3 +143,111 @@ class ClientWrapper:
     async def wait_until_disconnected(self) -> None:
         if self.client:
             await self.client.disconnected
+
+    async def _login_by_code(self, phone: str, force_sms: bool, functions, types) -> None:
+        sent = await self.client.send_code_request(phone, force_sms=force_sms)
+        code_type = getattr(sent, "type", None)
+        logger.info("Login code type: %s", type(code_type).__name__ if code_type else "unknown")
+        if isinstance(code_type, types.auth.SentCodeTypeSetUpEmailRequired):
+            logger.warning("Telegram requires login email verification")
+            try:
+                sent = await self.client(functions.auth.ResetLoginEmailRequest(phone, sent.phone_code_hash))
+            except Exception as exc:
+                logger.error("Email reset failed: %s", exc)
+                raise RuntimeError(
+                    "Set a recovery email in Telegram (Settings -> Privacy and Security -> Two-Step Verification) "
+                    "then retry login."
+                ) from exc
+            code_type = getattr(sent, "type", None)
+            logger.info("Login code type: %s", type(code_type).__name__ if code_type else "unknown")
+        if isinstance(code_type, types.auth.SentCodeTypeEmailCode):
+            logger.warning("Check your email for the login code")
+        code = await self._prompt("Please enter the code you received: ")
+        try:
+            await self.client.sign_in(phone=phone, code=code, phone_code_hash=sent.phone_code_hash)
+        except Exception as exc:
+            from telethon.errors import SessionPasswordNeededError
+
+            if isinstance(exc, SessionPasswordNeededError):
+                password = await self._prompt("Two-step verification enabled. Enter password: ")
+                await self.client.sign_in(password=password)
+                return
+            raise
+
+    def _print_qr(self, url: str, qr_path: Optional[str] = None, qr_open: bool = False) -> None:
+        logger.info("QR login URL: %s", url)
+        print("Telegram app -> Settings -> Devices -> Scan QR")
+        print(f"If you cannot scan, try opening this URL on your phone: {url}")
+        saved_path = None
+        if qr_path:
+            saved_path = self._save_qr_image(url, qr_path)
+        try:
+            import qrcode
+
+            qr = qrcode.QRCode(border=1)
+            qr.add_data(url)
+            qr.make(fit=True)
+            qr.print_ascii(invert=True)
+        except Exception:
+            pass
+        if qr_open and saved_path:
+            self._open_qr_window(saved_path)
+
+    def _save_qr_image(self, url: str, qr_path: str) -> Optional[Path]:
+        try:
+            import qrcode
+        except Exception:
+            logger.warning("qrcode package not installed; cannot save QR image")
+            return None
+        path = Path(qr_path)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        img = qrcode.make(url)
+        img.save(path)
+        logger.info("QR image saved to %s", path)
+        return path
+
+    def _open_qr_window(self, path: Path) -> None:
+        if self._try_tk_window(path):
+            return
+        self._open_file(path)
+
+    def _try_tk_window(self, path: Path) -> bool:
+        try:
+            import tkinter as tk
+        except Exception:
+            return False
+
+        def run_window() -> None:
+            try:
+                root = tk.Tk()
+                root.title("OverUB QR Login")
+                root.attributes("-topmost", True)
+                image = tk.PhotoImage(file=str(path))
+                label = tk.Label(root, image=image)
+                label.image = image
+                label.pack()
+                root.mainloop()
+            except Exception:
+                return
+
+        thread = threading.Thread(target=run_window, daemon=True)
+        thread.start()
+        return True
+
+    def _open_file(self, path: Path) -> None:
+        try:
+            if sys.platform.startswith("darwin"):
+                subprocess.run(["open", str(path)], check=False)
+            elif sys.platform.startswith("win"):
+                subprocess.run(["cmd", "/c", "start", "", str(path)], check=False)
+            else:
+                subprocess.run(["xdg-open", str(path)], check=False)
+        except Exception:
+            try:
+                webbrowser.open(path.as_uri())
+            except Exception:
+                logger.warning("Failed to open QR image automatically")
+
+    async def _prompt(self, prompt: str) -> str:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, input, prompt)

core/config.py

@@ -38,6 +38,7 @@ class ConfigManager:
             "OVERUB_API_ID": ("bot", "api_id"),
             "OVERUB_API_HASH": ("bot", "api_hash"),
             "OVERUB_SESSION": ("bot", "session_name"),
+            "OVERUB_SESSION_STRING": ("bot", "session_string"),
             "OVERUB_PREFIX": ("bot", "command_prefix"),
             "OVERUB_LOG_LEVEL": ("logging", "level"),
             "OVERUB_GIT_REMOTE": ("updates", "git", "remote"),

scripts/generate-session-string.py

@@ -0,0 +1,90 @@
+#!/usr/bin/env python3
+"""
+Generate a Telethon StringSession for OverUB.
+"""
+
+import argparse
+import asyncio
+import os
+
+from telethon import TelegramClient
+from telethon.errors import SessionPasswordNeededError
+from telethon.tl import functions, types
+from telethon.sessions import StringSession
+
+
+async def generate(args: argparse.Namespace) -> None:
+    if os.path.exists(".env"):
+        with open(".env", "r", encoding="utf-8") as handle:
+            for line in handle:
+                line = line.strip()
+                if not line or line.startswith("#") or "=" not in line:
+                    continue
+                key, value = line.split("=", 1)
+                os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))
+
+    api_id = os.getenv("OVERUB_API_ID") or os.getenv("API_ID")
+    api_hash = os.getenv("OVERUB_API_HASH") or os.getenv("API_HASH")
+    if not api_id or not api_hash:
+        print("API_ID/API_HASH required (set OVERUB_API_ID/OVERUB_API_HASH or API_ID/API_HASH).")
+        return
+    try:
+        api_id = int(api_id)
+    except ValueError:
+        print("API_ID must be an integer")
+        return
+
+    print("Starting session generation. You will be prompted to log in.")
+    client = TelegramClient(StringSession(), api_id, api_hash)
+    await client.connect()
+    if await client.is_user_authorized():
+        session_string = client.session.save()
+        await client.disconnect()
+        print("\nSession string:")
+        print(session_string)
+        print("\nAdd to .env as OVERUB_SESSION_STRING=" + session_string)
+        return
+    phone = args.phone or os.getenv("OVERUB_PHONE") or os.getenv("PHONE")
+    if not phone:
+        phone = input("Phone number (+123456789): ").strip()
+    sent = await client.send_code_request(phone, force_sms=args.sms)
+    code_type = getattr(sent, "type", None)
+    print(f"Code delivery type: {type(code_type).__name__ if code_type else 'unknown'}")
+    if isinstance(code_type, types.auth.SentCodeTypeSetUpEmailRequired):
+        print("Telegram requires login email verification.")
+        try:
+            sent = await client(functions.auth.ResetLoginEmailRequest(phone, sent.phone_code_hash))
+        except Exception as exc:
+            print(f"Email reset failed: {exc}")
+            print("Open Telegram -> Settings -> Privacy and Security -> Two-Step Verification")
+            print("Add and verify a recovery email, then retry.")
+            await client.disconnect()
+            return
+        code_type = getattr(sent, "type", None)
+        print(f"Code delivery type: {type(code_type).__name__ if code_type else 'unknown'}")
+    if isinstance(code_type, types.auth.SentCodeTypeEmailCode):
+        print("Check your email for the login code.")
+    code = input("Enter the code you received: ").strip()
+    try:
+        await client.sign_in(phone=phone, code=code, phone_code_hash=sent.phone_code_hash)
+    except SessionPasswordNeededError:
+        password = input("Two-step verification password: ").strip()
+        await client.sign_in(password=password)
+    session_string = client.session.save()
+    await client.disconnect()
+
+    print("\nSession string:")
+    print(session_string)
+    print("\nAdd to .env as OVERUB_SESSION_STRING=" + session_string)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Generate Telethon StringSession for OverUB")
+    parser.add_argument("--phone", default="")
+    parser.add_argument("--sms", action="store_true", help="Force SMS delivery")
+    args = parser.parse_args()
+    asyncio.run(generate(args))
+
+
+if __name__ == "__main__":
+    main()

scripts/import-tdata.py

@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+"""
+Import Telegram Desktop tdata into a Telethon StringSession.
+Requires `opentele` (pip install opentele).
+"""
+
+import argparse
+import asyncio
+import os
+from pathlib import Path
+
+from telethon.sessions import StringSession
+
+
+def load_env() -> None:
+    if not os.path.exists(".env"):
+        return
+    with open(".env", "r", encoding="utf-8") as handle:
+        for line in handle:
+            line = line.strip()
+            if not line or line.startswith("#") or "=" not in line:
+                continue
+            key, value = line.split("=", 1)
+            os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))
+
+
+def find_tdata_path() -> Path:
+    candidates = [
+        Path.home() / ".local" / "share" / "TelegramDesktop" / "tdata",
+        Path.home() / "AppData" / "Roaming" / "Telegram Desktop" / "tdata",
+        Path.home() / "Library" / "Application Support" / "Telegram Desktop" / "tdata",
+    ]
+    for candidate in candidates:
+        if candidate.exists():
+            return candidate
+    return candidates[0]
+
+
+async def import_tdata(args: argparse.Namespace) -> None:
+    load_env()
+    api_id = args.api_id or os.getenv("OVERUB_API_ID") or os.getenv("API_ID")
+    api_hash = args.api_hash or os.getenv("OVERUB_API_HASH") or os.getenv("API_HASH")
+    if not api_id or not api_hash:
+        print("API_ID/API_HASH required (set in .env or pass --api-id/--api-hash).")
+        return
+    try:
+        api_id = int(api_id)
+    except ValueError:
+        print("API_ID must be an integer")
+        return
+
+    tdata = Path(args.tdata or find_tdata_path())
+    if not tdata.exists():
+        print(f"tdata not found: {tdata}")
+        return
+
+    try:
+        from opentele.td import TDesktop  # type: ignore
+    except Exception:
+        try:
+            from opentele import TDesktop  # type: ignore
+        except Exception as exc:
+            print("opentele not installed. Install with: pip install opentele")
+            print(f"Import error: {exc}")
+            return
+
+    client = None
+    td = TDesktop(str(tdata))
+    if hasattr(td, "to_telethon"):
+        try:
+            client = td.to_telethon(session=StringSession(), api_id=api_id, api_hash=api_hash)
+        except TypeError:
+            client = td.to_telethon(StringSession(), api_id, api_hash)
+    if client is None:
+        try:
+            from opentele.tl import TelegramClient as OTClient  # type: ignore
+        except Exception:
+            print("Unsupported opentele version. Try upgrading opentele.")
+            return
+        client = OTClient(td, api_id, api_hash)
+
+    await client.connect()
+    session_string = client.session.save()
+    await client.disconnect()
+
+    print("\nSession string:")
+    print(session_string)
+    print("\nAdd to .env as OVERUB_SESSION_STRING=" + session_string)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Import Telegram Desktop tdata to Telethon StringSession")
+    parser.add_argument("--tdata", default="", help="Path to Telegram Desktop tdata directory")
+    parser.add_argument("--api-id", default="")
+    parser.add_argument("--api-hash", default="")
+    args = parser.parse_args()
+    asyncio.run(import_tdata(args))
+
+
+if __name__ == "__main__":
+    main()