from __future__ import annotations

import os
import subprocess
import threading
from contextlib import contextmanager
from typing import Any, Iterator

try:
    import pymysql
    from pymysql.constants import CLIENT
except ImportError:
    pymysql = None  # type: ignore[assignment]
    CLIENT = None  # type: ignore[assignment]
from dataclasses import dataclass, field


def sql_quote(value: str | None) -> str:
    if value is None:
        return "NULL"
    escaped = value.replace("\\", "\\\\").replace("'", "''")
    return "'" + escaped + "'"


def sql_blob_hex(data: bytes | None) -> str:
    if data is None:
        return "NULL"
    return "0x" + data.hex()


def result_cell_text(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, bytes):
        return value.decode("utf-8")
    return str(value)


@dataclass
class MysqlCli:
    database: str
    _connection_local: Any = field(default_factory=threading.local, init=False, repr=False)

    @classmethod
    def from_env(cls) -> "MysqlCli":
        return cls(database=os.environ.get("NFSE_DB_NAME", "maspernf"))

    def _base_cmd(self) -> list[str]:
        container = os.environ.get("NFSE_DB_DOCKER_CONTAINER")
        user = os.environ.get("NFSE_DB_USER", "root")
        password = os.environ.get("NFSE_DB_PASSWORD", "root")
        host = os.environ.get("NFSE_DB_HOST", "127.0.0.1")
        port = os.environ.get("NFSE_DB_PORT", "3307")

        mysql = [
            "mysql",
            "-N",
            "-B",
            "--default-character-set=utf8mb4",
            f"-u{user}",
            f"-p{password}",
            "-D",
            self.database,
        ]
        if container:
            return ["docker", "exec", "-i", container] + mysql
        return mysql[:1] + [f"-h{host}", f"-P{port}"] + mysql[1:]

    def _connection_or_none(self) -> Any:
        if os.environ.get("NFSE_DB_DOCKER_CONTAINER") or pymysql is None:
            return None
        connection = getattr(self._connection_local, "connection", None)
        if connection is None:
            connection = pymysql.connect(
                host=os.environ.get("NFSE_DB_HOST", "127.0.0.1"),
                port=int(os.environ.get("NFSE_DB_PORT", "3307")),
                user=os.environ.get("NFSE_DB_USER", "root"),
                password=os.environ.get("NFSE_DB_PASSWORD", "root"),
                database=self.database,
                charset="utf8mb4",
                autocommit=True,
                client_flag=CLIENT.MULTI_STATEMENTS,
            )
            self._connection_local.connection = connection
        else:
            connection.ping(reconnect=True)
        return connection

    def close(self) -> None:
        connection = getattr(self._connection_local, "connection", None)
        if connection is not None:
            connection.close()
            del self._connection_local.connection

    def require_persistent_connection(self) -> None:
        if self._connection_or_none() is None:
            raise RuntimeError(
                "Esta operacao exige PyMySQL e uma conexao persistente para garantir transacoes."
            )

    @contextmanager
    def transaction(self) -> Iterator[None]:
        connection = self._connection_or_none()
        if connection is None:
            raise RuntimeError(
                "Esta operacao exige PyMySQL e uma conexao persistente para garantir transacoes."
            )
        connection.begin()
        try:
            yield
        except Exception:
            connection.rollback()
            raise
        else:
            connection.commit()

    def execute(self, sql: str) -> str:
        connection = self._connection_or_none()
        if connection is not None:
            try:
                with connection.cursor() as cursor:
                    cursor.execute(sql)
                    result: list[tuple[Any, ...]] = []
                    while True:
                        if cursor.description:
                            result = cursor.fetchall()
                        if not cursor.nextset():
                            break
                return "\n".join("\t".join(result_cell_text(item) for item in row) for row in result)
            except Exception as exc:
                connection.rollback()
                raise RuntimeError(str(exc)) from exc
        proc = subprocess.run(
            self._base_cmd(),
            input=sql,
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=False,
        )
        if proc.returncode != 0:
            raise RuntimeError(proc.stderr.strip() or "Erro ao executar SQL.")
        return proc.stdout

    def execute_file_sql(self, sql: str) -> None:
        self.execute(sql)

    def query_rows(self, sql: str) -> list[list[str]]:
        out = self.execute(sql)
        rows: list[list[str]] = []
        for line in out.splitlines():
            rows.append(line.split("\t"))
        return rows