import asyncio
import os
import sqlite3
import time

from pyrogram import Client, filters
from pyrogram.errors import (
    AuthKeyUnregistered,
    FloodWait,
    PeerFlood,
    PeerIdInvalid,
    UserDeactivated,
    UserIsBlocked,
    UsernameInvalid,
    UsernameNotOccupied,
)
from pyrogram.handlers import MessageHandler

import config

_locks = {}
_persistent = {}
_watch_count = {}


def _lock(key):
    key = str(key)
    if key not in _locks:
        _locks[key] = asyncio.Lock()
    return _locks[key]


def _tmp_name(prefix="kodo"):
    return f"{prefix}_{int(time.time() * 1000)}"


def _new_client(session_string=None, name=None, workdir=None):
    if session_string:
        return Client(
            name=_tmp_name(),
            api_id=config.API_ID,
            api_hash=config.API_HASH,
            session_string=session_string,
            in_memory=True,
        )
    if name:
        return Client(
            name=name,
            api_id=config.API_ID,
            api_hash=config.API_HASH,
            workdir=workdir or config.SESSIONS_DIR,
        )
    return Client(
        name=_tmp_name(),
        api_id=config.API_ID,
        api_hash=config.API_HASH,
        in_memory=True,
    )


async def start_login(phone):
    app = _new_client()
    await app.connect()
    try:
        sent = await app.send_code(phone)
    except Exception:
        await app.disconnect()
        raise
    return app, sent.phone_code_hash


async def confirm_code(app, phone, phone_code_hash, code):
    return await app.sign_in(phone, phone_code_hash, code)


async def confirm_password(app, password):
    return await app.check_password(password)


async def finish_login(app):
    session_string = await app.export_session_string()
    me = await app.get_me()
    await app.disconnect()
    return session_string, me


async def cancel_login(app):
    try:
        await app.disconnect()
    except Exception:
        pass


def detect_library(path):
    try:
        conn = sqlite3.connect(path)
        cur = conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
        tables = {row[0] for row in cur.fetchall()}
        conn.close()
    except Exception:
        return "unknown"
    if "entities" in tables or "sent_files" in tables or "update_state" in tables:
        return "Telethon"
    if "sessions" in tables:
        return "pyrogram"
    return "unknown"


async def import_session_file(path):
    lib = detect_library(path)
    if lib != "pyrogram":
        return None, None, lib
    name = os.path.basename(path)
    if name.endswith(".session"):
        name = name[:-8]
    app = _new_client(name=name, workdir=os.path.dirname(path))
    try:
        await app.connect()
        session_string = await app.export_session_string()
        me = await app.get_me()
    finally:
        try:
            await app.disconnect()
        except Exception:
            pass
    return session_string, me, lib


async def import_session_string(session_string):
    app = _new_client(session_string=session_string)
    await app.connect()
    try:
        me = await app.get_me()
        exported = await app.export_session_string()
    finally:
        try:
            await app.disconnect()
        except Exception:
            pass
    return exported, me


async def _acquire(account_id, session_string):
    app = _persistent.get(account_id)
    if app is not None:
        return app, False
    app = _new_client(session_string=session_string)
    await app.start()
    return app, True


async def _release(app, temporary):
    if temporary:
        try:
            await app.stop()
        except Exception:
            pass


async def resolve_user(account_id, session_string, username):
    async with _lock(account_id):
        app, temp = await _acquire(account_id, session_string)
        try:
            user = await app.get_users(username)
            return user, ""
        except (UsernameNotOccupied, UsernameInvalid, PeerIdInvalid, IndexError, KeyError):
            return None, "notfound"
        except FloodWait as e:
            return None, f"FloodWait {e.value}"
        except Exception as e:
            return None, str(e)
        finally:
            await _release(app, temp)


async def send_text(account_id, session_string, target, text):
    async with _lock(account_id):
        started = time.monotonic()
        try:
            app, temp = await _acquire(account_id, session_string)
        except (AuthKeyUnregistered, UserDeactivated):
            return False, 0.0, "dead"
        except Exception as e:
            return False, 0.0, str(e)
        try:
            await app.send_message(target, text)
            return True, time.monotonic() - started, ""
        except FloodWait as e:
            return False, time.monotonic() - started, f"FloodWait {e.value}"
        except (PeerIdInvalid, UsernameNotOccupied, UsernameInvalid):
            return False, time.monotonic() - started, "notfound"
        except UserIsBlocked:
            return False, time.monotonic() - started, "blocked"
        except PeerFlood:
            return False, time.monotonic() - started, "peerflood"
        except Exception as e:
            return False, time.monotonic() - started, str(e)
        finally:
            await _release(app, temp)


async def start_watcher(account_id, session_string, target_id, callback):
    async with _lock(account_id):
        app = _persistent.get(account_id)
        if app is None:
            app = _new_client(session_string=session_string)
            await app.start()
            _persistent[account_id] = app
            _watch_count[account_id] = 0
        handler = MessageHandler(
            callback, filters.user(target_id) & filters.private & filters.incoming
        )
        app.add_handler(handler)
        _watch_count[account_id] = _watch_count.get(account_id, 0) + 1
        return handler


async def stop_watcher(account_id, handler):
    async with _lock(account_id):
        app = _persistent.get(account_id)
        if app is None:
            return
        try:
            app.remove_handler(handler)
        except Exception:
            pass
        _watch_count[account_id] = max(0, _watch_count.get(account_id, 1) - 1)
        if _watch_count[account_id] == 0:
            _persistent.pop(account_id, None)
            _watch_count.pop(account_id, None)
            try:
                await app.stop()
            except Exception:
                pass


async def get_me(account_id, session_string):
    async with _lock(account_id):
        app, temp = await _acquire(account_id, session_string)
        try:
            return await app.get_me()
        finally:
            await _release(app, temp)


async def update_profile(account_id, session_string, first_name=None, bio=None, photo_path=None):
    async with _lock(account_id):
        try:
            app, temp = await _acquire(account_id, session_string)
        except (AuthKeyUnregistered, UserDeactivated):
            return False, "dead"
        except Exception as e:
            return False, str(e)
        try:
            if first_name is not None or bio is not None:
                await app.update_profile(first_name=first_name, bio=bio)
            if photo_path:
                await app.set_profile_photo(photo=photo_path)
            return True, ""
        except FloodWait as e:
            return False, f"FloodWait {e.value}"
        except Exception as e:
            return False, str(e)
        finally:
            await _release(app, temp)