"""
telegram_worker.py — runs Telethon on its own persistent asyncio event loop
in a background thread, and exposes plain (blocking-safe) methods that the
synchronous Flask routes can call.

Read-only by design: the bot only listens for new messages in the group you
pick. It never sends, forwards, or reacts to anything.
"""
import asyncio
import base64
import io
import os
import queue
import threading
import time
import uuid
from concurrent.futures import TimeoutError as FuturesTimeoutError

from telethon import TelegramClient, events
from telethon.errors import SessionPasswordNeededError, FloodWaitError
from telethon.tl.types import Chat, Channel

import qrcode

from parsers import parse_payment_message
import storage

try:
    import edge_tts
    EDGE_TTS_AVAILABLE = True
except ImportError:
    EDGE_TTS_AVAILABLE = False

# Published Telegram Desktop API credentials — shipped with every official
# Telegram Desktop install (see https://telegra.ph/telegraph-01-31-6).
# Only the user's own phone number / login code / session stay private.
PUBLIC_API_ID = 2040
PUBLIC_API_HASH = "b18441a1ff607e10a989891a5462e627"

SESSION_PATH = os.path.join(storage.DATA_DIR, "aba_userbot_session")
TTS_DIR = os.path.join(os.path.dirname(__file__), "static", "tts")
os.makedirs(TTS_DIR, exist_ok=True)

RECONNECT_BACKOFF = [2, 5, 10, 20, 30, 60]


class TelegramManager:
    """Singleton-ish manager. Create one instance and share it across the app."""

    def __init__(self, emit_callback, log_callback):
        self.emit = emit_callback          # emit(event_name, payload_dict)
        self.log = log_callback            # log(message_str)

        self.loop = asyncio.new_event_loop()
        self._thread = threading.Thread(target=self._run_loop, daemon=True)
        self._thread.start()

        self.client = None
        self.status = "logged_out"         # logged_out | code_needed | password_needed
                                            # | logged_in | listening
        self.me_label = ""
        self.last_error = ""

        self._phone = None
        self._qr_login_obj = None
        self.qr_data_uri = None
        self.qr_expires_at = 0

        self._listener_registered = False
        self._active_handler = None
        self._current_group = None

        self.alert_queue = queue.Queue()
        self._pacer_thread = threading.Thread(target=self._pace_alerts, daemon=True)
        self._pacer_stop = False
        self._pacer_thread.start()

        # Try to resume an existing authorized session on startup. Network
        # issues here must never prevent the web app itself from starting.
        try:
            self._run(self._bootstrap(), timeout=15)
        except Exception as e:
            self.last_error = str(e)
            self.log(f"⚠️ Could not reach Telegram on startup ({e}). "
                      f"You can still log in from the dashboard once the network is available.")

    # ── loop plumbing ──────────────────────────────────────────────

    def _run_loop(self):
        asyncio.set_event_loop(self.loop)
        self.loop.run_forever()

    def _run(self, coro, timeout=30):
        """Run a coroutine on the manager's loop and block for the result."""
        fut = asyncio.run_coroutine_threadsafe(coro, self.loop)
        try:
            return fut.result(timeout=timeout)
        except FuturesTimeoutError:
            fut.cancel()
            raise TimeoutError(
                "Timed out waiting for Telegram — check your network connection.")

    def _fire(self, coro):
        """Schedule a coroutine on the loop without waiting for it."""
        return asyncio.run_coroutine_threadsafe(coro, self.loop)

    async def _get_client(self):
        if self.client is None:
            self.client = TelegramClient(SESSION_PATH, PUBLIC_API_ID, PUBLIC_API_HASH,
                                          loop=self.loop)
        if not self.client.is_connected():
            await self.client.connect()
        return self.client

    async def _bootstrap(self):
        try:
            client = await self._get_client()
            if await client.is_user_authorized():
                me = await client.get_me()
                self.me_label = f"@{me.username}" if me.username else (me.first_name or str(me.id))
                self.status = "logged_in"
                self.log(f"✅ Resumed session as {self.me_label}")
        except Exception as e:
            self.last_error = str(e)

    # ── phone login ─────────────────────────────────────────────────

    def start_phone_login(self, phone):
        try:
            return self._run(self._start_phone_login(phone))
        except Exception as e:
            self.log(f"❌ {e}")
            return {"ok": False, "error": str(e)}

    async def _start_phone_login(self, phone):
        try:
            client = await self._get_client()
            self._phone = phone
            try:
                await client.send_code_request(phone)
            except FloodWaitError as e:
                self.log(f"⏳ Telegram asked us to wait {e.seconds}s.")
                await asyncio.sleep(e.seconds + 1)
                await client.send_code_request(phone)
            self.status = "code_needed"
            self.log("📱 Login code sent to your Telegram app.")
            return {"ok": True, "status": self.status}
        except Exception as e:
            self.last_error = str(e)
            self.log(f"❌ Could not send login code: {e}")
            return {"ok": False, "error": str(e)}

    def submit_code(self, code):
        try:
            return self._run(self._submit_code(code))
        except Exception as e:
            self.log(f"❌ {e}")
            return {"ok": False, "error": str(e)}

    async def _submit_code(self, code):
        try:
            client = await self._get_client()
            try:
                await client.sign_in(self._phone, code)
            except SessionPasswordNeededError:
                self.status = "password_needed"
                self.log("🔒 2FA password required.")
                return {"ok": True, "status": self.status}
            me = await client.get_me()
            self.me_label = f"@{me.username}" if me.username else (me.first_name or str(me.id))
            self.status = "logged_in"
            self.log(f"✅ Logged in as {self.me_label}")
            return {"ok": True, "status": self.status, "me": self.me_label}
        except Exception as e:
            self.last_error = str(e)
            self.log(f"❌ Code sign-in failed: {e}")
            return {"ok": False, "error": str(e)}

    def submit_password(self, password):
        try:
            return self._run(self._submit_password(password))
        except Exception as e:
            self.log(f"❌ {e}")
            return {"ok": False, "error": str(e)}

    async def _submit_password(self, password):
        try:
            client = await self._get_client()
            await client.sign_in(password=password)
            me = await client.get_me()
            self.me_label = f"@{me.username}" if me.username else (me.first_name or str(me.id))
            self.status = "logged_in"
            self.log(f"✅ Logged in as {self.me_label}")
            return {"ok": True, "status": self.status, "me": self.me_label}
        except Exception as e:
            self.last_error = str(e)
            self.log(f"❌ Password sign-in failed: {e}")
            return {"ok": False, "error": str(e)}

    # ── QR login ────────────────────────────────────────────────────

    def start_qr_login(self):
        try:
            return self._run(self._start_qr_login())
        except Exception as e:
            self.log(f"❌ {e}")
            return {"ok": False, "error": str(e)}

    async def _start_qr_login(self):
        try:
            client = await self._get_client()
            self._qr_login_obj = await client.qr_login()
            self.qr_data_uri = self._render_qr(self._qr_login_obj.url)
            self.qr_expires_at = time.time() + 30
            self.status = "qr_pending"
            self._fire(self._wait_qr())
            return {"ok": True, "qr": self.qr_data_uri, "expires_in": 30}
        except Exception as e:
            self.last_error = str(e)
            self.log(f"❌ QR login failed to start: {e}")
            return {"ok": False, "error": str(e)}

    async def _wait_qr(self):
        try:
            await self._qr_login_obj.wait(timeout=30)
            client = await self._get_client()
            me = await client.get_me()
            self.me_label = f"@{me.username}" if me.username else (me.first_name or str(me.id))
            self.status = "logged_in"
            self.log(f"✅ Logged in via QR as {self.me_label}")
        except SessionPasswordNeededError:
            self.status = "password_needed"
            self.log("🔒 2FA password required (QR login).")
        except asyncio.TimeoutError:
            self.status = "qr_expired"
        except Exception as e:
            self.last_error = str(e)
            self.status = "qr_expired"
            self.log(f"⚠️ QR login error: {e}")

    @staticmethod
    def _render_qr(url):
        img = qrcode.make(url)
        buf = io.BytesIO()
        img.save(buf, format="PNG")
        b64 = base64.b64encode(buf.getvalue()).decode("ascii")
        return f"data:image/png;base64,{b64}"

    def qr_status(self):
        return {
            "status": self.status,
            "qr": self.qr_data_uri,
            "me": self.me_label,
            "error": self.last_error,
        }

    # ── logout ──────────────────────────────────────────────────────

    def logout(self):
        try:
            return self._run(self._logout())
        except Exception as e:
            self.log(f"⚠️ {e}")
            self.client = None
            self.status = "logged_out"
            self.me_label = ""

    async def _logout(self):
        try:
            self.stop_listener()
            if self.client:
                await self.client.log_out()
        except Exception as e:
            self.log(f"⚠️ Logout warning: {e}")
        finally:
            self.client = None
            self.status = "logged_out"
            self.me_label = ""

    # ── group listing ───────────────────────────────────────────────

    def list_groups(self):
        try:
            return self._run(self._list_groups(), timeout=60)
        except Exception as e:
            self.log(f"⚠️ Could not list groups: {e}")
            return []

    async def _list_groups(self):
        client = await self._get_client()
        out = []
        async for d in client.iter_dialogs():
            ent = d.entity
            if isinstance(ent, (Chat, Channel)):
                if isinstance(ent, Channel) and not (ent.megagroup or ent.broadcast is False):
                    # allow megagroups and regular channels/supergroups
                    pass
                out.append({
                    "id": d.id,
                    "title": d.name or getattr(ent, "title", str(d.id)),
                    "is_channel": isinstance(ent, Channel),
                })
        return out

    # ── listener control ────────────────────────────────────────────

    def start_listener(self, group_id, source_username, khr_rate, debug):
        try:
            return self._run(self._start_listener(group_id, source_username, khr_rate, debug), timeout=60)
        except Exception as e:
            self.log(f"❌ {e}")
            return {"ok": False, "error": str(e)}

    async def _start_listener(self, group_id, source_username, khr_rate, debug):
        try:
            client = await self._get_client()
            if not await client.is_user_authorized():
                return {"ok": False, "error": "Not logged in."}

            try:
                group_id_int = int(group_id)
            except ValueError:
                return {"ok": False, "error": "Group ID must be numeric."}

            try:
                entity = await client.get_entity(group_id_int)
            except Exception as e:
                self.log(f"⚠️ Could not pre-resolve group ({e}) — using raw ID.")
                entity = group_id_int

            source_username = (source_username or "").strip().lstrip("@").lower()

            if self._listener_registered and self._active_handler:
                client.remove_event_handler(self._active_handler)
                self._listener_registered = False

            async def handler(event):
                await self._on_new_message(event, source_username, float(khr_rate or 4000.0), bool(debug))

            self._active_handler = handler
            client.add_event_handler(handler, events.NewMessage(chats=entity))
            self._listener_registered = True
            self._current_group = group_id_int
            self.status = "listening"
            self.log(f"👂 Listening on group {group_id_int}"
                      + (f" (filtering @{source_username})" if source_username else " (all senders)"))
            return {"ok": True, "status": self.status}
        except Exception as e:
            self.last_error = str(e)
            self.log(f"❌ Could not start listener: {e}")
            return {"ok": False, "error": str(e)}

    def stop_listener(self):
        if self.client and self._listener_registered and self._active_handler:
            try:
                self.client.remove_event_handler(self._active_handler)
            except Exception:
                pass
            self._listener_registered = False
        if self.status == "listening":
            self.status = "logged_in"
        self.log("🛑 Listener stopped.")
        return {"ok": True, "status": self.status}

    async def _on_new_message(self, event, source_username, khr_rate, debug):
        try:
            text = (event.raw_text or "").strip()
            if not text:
                return
            try:
                sender = await event.get_sender()
            except Exception:
                sender = None
            actual_username = (getattr(sender, "username", "") or "").lower()

            if debug:
                preview = " ".join(text.split())[:80]
                self.log(f"📥 from=@{actual_username or 'unknown'} text=\"{preview}\"")

            if source_username and actual_username != source_username:
                return

            result = parse_payment_message(text)
            if result is None:
                return

            cfg = storage.load_config()
            stats = storage.update_top_donors(result["sender"], result["sender_key"],
                                                result["amount"], khr_rate)

            alert = {
                "id": int(time.time() * 1000),
                "sender": result["sender"],
                "amount": result["amount"],
                "message": "",
                "provider": result["provider"],
                "donor_total": stats["total_display"],
                "donor_count": stats["count"],
                "is_top_donor": stats["is_top"],
            }
            self.alert_queue.put(alert)

            crown = " 🏆 NEW #1 DONOR" if stats["is_top"] else ""
            self.log(f"✨ [{result['provider']}] {result['sender']} — {result['amount']}{crown} (queued)")
        except Exception as e:
            self.log(f"⚠️ Handler error: {e}")

    # ── alert pacing (delay / duration) ─────────────────────────────

    def queue_test_alert(self, sender="MEN SOKRAYUK", amount="$1.00", is_top=True):
        alert = {
            "id": int(time.time() * 1000),
            "sender": sender,
            "amount": amount,
            "message": "This is a test alert!",
            "provider": "ABA PayWay",
            "donor_total": "$12.75",
            "donor_count": 3,
            "is_top_donor": is_top,
        }
        self.alert_queue.put(alert)

    def _pace_alerts(self):
        while not self._pacer_stop:
            try:
                item = self.alert_queue.get(timeout=0.25)
            except queue.Empty:
                continue

            cfg = storage.load_config()
            delay = max(0.0, float(cfg.get("alert_delay", 0.0) or 0.0))
            duration = max(1.0, float(cfg.get("alert_duration", 8.0) or 8.0))

            if delay > 0:
                time.sleep(delay)

            item = dict(item)
            item["duration_ms"] = int(duration * 1000)
            item["theme"] = cfg.get("alert_theme", "gold")

            storage.push_recent({
                "sender": item["sender"], "amount": item["amount"],
                "provider": item["provider"], "ts": int(time.time() * 1000),
            })

            self.emit("donation_alert", item)
            self.emit("recent_alert", item)
            self.emit("leaderboard_update", {"donors": storage.load_donors()})

            if cfg.get("tts_enabled") and EDGE_TTS_AVAILABLE:
                self._spawn_tts(item, cfg)

            time.sleep(duration)

    def _spawn_tts(self, item, cfg):
        fmt = cfg.get("tts_format") or "New donation! {name} gave {amount} via {provider}!"
        text = fmt.replace("{name}", item["sender"]) \
                   .replace("{amount}", item["amount"]) \
                   .replace("{provider}", item["provider"])
        voice = cfg.get("tts_voice", "en-US-JennyNeural")
        rate = f"{int(cfg.get('tts_rate', 0)):+d}%"
        volume = f"{int(cfg.get('tts_volume', 10)):+d}%"
        pitch = f"{int(cfg.get('tts_pitch', 0)):+d}Hz"
        alert_id = item["id"]

        def worker():
            try:
                tts_loop = asyncio.new_event_loop()
                filename = f"tts_{alert_id}_{uuid.uuid4().hex[:6]}.mp3"
                path = os.path.join(TTS_DIR, filename)
                communicate = edge_tts.Communicate(text, voice, rate=rate, volume=volume, pitch=pitch)
                tts_loop.run_until_complete(communicate.save(path))
                tts_loop.close()
                self._cleanup_tts_files()
                self.emit("tts_ready", {"id": alert_id, "url": f"/static/tts/{filename}"})
            except Exception as e:
                self.log(f"⚠️ TTS error: {e}")

        threading.Thread(target=worker, daemon=True).start()

    @staticmethod
    def _cleanup_tts_files(keep=8):
        try:
            files = [f for f in os.listdir(TTS_DIR) if f.startswith("tts_") and f.endswith(".mp3")]
            files.sort(key=lambda f: os.path.getmtime(os.path.join(TTS_DIR, f)))
            for old in files[:-keep]:
                try:
                    os.remove(os.path.join(TTS_DIR, old))
                except OSError:
                    pass
        except Exception:
            pass

    def preview_tts(self, text, voice, rate, volume, pitch):
        if not EDGE_TTS_AVAILABLE:
            return {"ok": False, "error": "edge-tts not installed"}
        item = {"id": int(time.time() * 1000)}

        def worker():
            try:
                tts_loop = asyncio.new_event_loop()
                filename = f"preview_{item['id']}.mp3"
                path = os.path.join(TTS_DIR, filename)
                communicate = edge_tts.Communicate(text, voice, rate=f"{rate:+d}%",
                                                    volume=f"{volume:+d}%", pitch=f"{pitch:+d}Hz")
                tts_loop.run_until_complete(communicate.save(path))
                tts_loop.close()
                self.emit("tts_preview_ready", {"url": f"/static/tts/{filename}"})
            except Exception as e:
                self.log(f"⚠️ TTS preview error: {e}")

        threading.Thread(target=worker, daemon=True).start()
        return {"ok": True}