from datetime import date, datetime
from typing import Any

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.customers.schemas import CustomerWriteRequest


STATUS_SQL = """
CASE
    WHEN UPPER(COALESCE(p.aktif, '')) IN ('AKTIF', 'Y', '1') THEN 'AKTIF'
    ELSE 'TIDAK'
END
"""


class CustomerRepository:
    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    async def list_customers(
        self,
        *,
        aktif: str | None,
        search: str | None,
        include_piutang_summary: bool,
        kode_sales: str | None = None,
        limit: int,
        offset: int,
    ) -> tuple[list[dict[str, Any]], int]:
        sales_piutang_filter = "AND kodesales = :kode_sales" if kode_sales else ""
        summary_join = f"""
        LEFT JOIN (
            SELECT
                kodepelanggan,
                SUM(CASE WHEN tanggaljatuhtempo <= CURRENT_DATE THEN sisa ELSE 0 END) AS jt,
                SUM(CASE WHEN tanggaljatuhtempo > CURRENT_DATE THEN sisa ELSE 0 END) AS belum_jt,
                SUM(sisa) AS piutang
            FROM vwlistpiutang
            WHERE sisa > 0.005 {sales_piutang_filter}
            GROUP BY kodepelanggan
        ) AS r ON r.kodepelanggan = p.kodepelanggan
        """ if include_piutang_summary else ""
        jt_column = "COALESCE(r.jt, 0)" if include_piutang_summary else "0"
        belum_jt_column = (
            "COALESCE(r.belum_jt, 0)" if include_piutang_summary else "0"
        )
        piutang_column = (
            "COALESCE(r.piutang, 0)" if include_piutang_summary else "0"
        )
        sales_filter = "AND p.salescreated = :kode_sales" if kode_sales else ""
        filters = f"""
            WHERE COALESCE(p.DLT, '0') = '0'
              AND (:aktif IS NULL OR {STATUS_SQL} = :aktif)
              {sales_filter}
              AND (:search IS NULL OR p.kodepelanggan LIKE :pattern
                   OR p.namapelanggan LIKE :pattern
                   OR p.alamat LIKE :pattern)
        """
        parameters = {
            "aktif": aktif,
            "search": search,
            "pattern": f"%{search}%" if search else None,
            "kode_sales": kode_sales,
        }
        count_result = await self.session.execute(
            text(f"SELECT COUNT(*) FROM mspelanggan AS p {filters}"), parameters
        )
        result = await self.session.execute(
            text(
                f"""
                SELECT
                    p.kodepelanggan AS kode_pelanggan,
                    p.namapelanggan AS nama_pelanggan,
                    p.alamat,
                    p.telp AS telepon,
                    j.namajalur AS nama_jalur,
                    o.jenisoutlet AS jenis_outlet,
                    t.tingkatharga AS tingkat_harga,
                    {jt_column} AS jt,
                    {belum_jt_column} AS belum_jt,
                    {piutang_column} AS piutang,
                    COALESCE(p.limitpiutang, 0) AS limit_piutang,
                    p.limithari AS limit_hari,
                    COALESCE(p.avgomset, 0) AS avg_omset,
                    COALESCE(p.avgpelunasan, 0) AS avg_pelunasan,
                    s.namasales AS nama_sales,
                    {STATUS_SQL} AS aktif
                FROM mspelanggan AS p
                LEFT JOIN msjalur AS j ON j.kodejalur = p.jalursales
                LEFT JOIN msjenisoutlet AS o ON o.kodeoutlet = p.kodeoutlet
                LEFT JOIN mstingkatharga AS t ON t.kodetingkat = p.kodetingkat
                LEFT JOIN mssales AS s ON s.kodesales = p.salescreated
                {summary_join}
                {filters}
                ORDER BY p.namapelanggan, p.kodepelanggan
                LIMIT :limit OFFSET :offset
                """
            ),
            {**parameters, "limit": limit, "offset": offset},
        )
        return [dict(row) for row in result.mappings().all()], int(
            count_result.scalar_one()
        )

    async def get_customer(
        self, kode_pelanggan: str, *, kode_sales: str | None = None
    ) -> dict[str, Any] | None:
        sales_filter = "AND p.salescreated = :kode_sales" if kode_sales else ""
        result = await self.session.execute(
            text(
                f"""
                SELECT
                    p.kodepelanggan AS kode_pelanggan,
                    p.namapelanggan AS nama_pelanggan,
                    p.alamat,
                    p.telp AS telepon,
                    COALESCE(p.limitpiutang, 0) AS limit_piutang,
                    p.limithari AS limit_hari,
                    p.namawajibpajak AS nama_wajib_pajak,
                    p.npwp, p.jalan, p.block, p.nomor, p.rt, p.rw,
                    p.kecamatan, p.kelurahan, p.kabupaten, p.propinsi,
                    p.kodepos AS kode_pos,
                    p.jalursales AS jalur_sales,
                    p.referensi,
                    p.kodetingkat AS kode_tingkat,
                    p.kodeoutlet AS kode_outlet,
                    p.wa, p.email, p.nik,
                    p.salescreated AS sales_created,
                    p.kall,
                    COALESCE(p.program, '0') IN ('1', 'Y') AS program,
                    p.keterangan,
                    {STATUS_SQL} AS aktif,
                    p.tanggalcreated AS tanggal_created,
                    p.tanggaledit AS tanggal_edit
                FROM mspelanggan AS p
                WHERE p.kodepelanggan = :kode_pelanggan
                  AND COALESCE(p.DLT, '0') = '0'
                  {sales_filter}
                """
            ),
            {"kode_pelanggan": kode_pelanggan, "kode_sales": kode_sales},
        )
        row = result.mappings().one_or_none()
        return dict(row) if row else None

    async def table_engine(self) -> str | None:
        result = await self.session.execute(
            text(
                """
                SELECT engine FROM information_schema.tables
                WHERE table_schema = DATABASE() AND table_name = 'mspelanggan'
                """
            )
        )
        return result.scalar_one_or_none()

    async def next_code(self, name: str) -> str:
        prefix = next((char for char in name.upper() if "A" <= char <= "Z"), "X")
        result = await self.session.execute(
            text(
                """
                SELECT COALESCE(MAX(CAST(SUBSTRING(kodepelanggan, 2) AS UNSIGNED)), 0) + 1
                FROM mspelanggan
                WHERE kodepelanggan REGEXP :pattern
                """
            ),
            {"pattern": f"^{prefix}[0-9]{{3}}$"},
        )
        number = int(result.scalar_one())
        if number > 999:
            raise ValueError(f"Nomor pelanggan awalan {prefix} sudah habis")
        return f"{prefix}{number:03d}"

    async def customer_exists(self, kode_pelanggan: str) -> bool:
        result = await self.session.execute(
            text("SELECT COUNT(*) FROM mspelanggan WHERE kodepelanggan=:kode"),
            {"kode": kode_pelanggan},
        )
        return bool(result.scalar_one())

    @staticmethod
    def _write_values(request: CustomerWriteRequest) -> dict[str, Any]:
        values = request.model_dump(exclude={"kode_pelanggan"})
        return {**values, "program": "Y" if request.program else "N"}

    async def insert_customer(
        self, kode_pelanggan: str, request: CustomerWriteRequest, now: date
    ) -> None:
        values = self._write_values(request)
        await self.session.execute(
            text(
                """
                INSERT INTO mspelanggan (
                    kodepelanggan, namapelanggan, alamat, telp, limitpiutang,
                    limithari, namawajibpajak, npwp, jalan, block, nomor, rt, rw,
                    kecamatan, kelurahan, kabupaten, propinsi, kodepos, jalursales,
                    referensi, kodetingkat, kodeoutlet, wa, email, nik,
                    salescreated, kall, program, keterangan, aktif,
                    tanggalcreated, tanggaledit, DLT
                ) VALUES (
                    :kode_pelanggan, :nama_pelanggan, :alamat, :telepon,
                    :limit_piutang, :limit_hari, :nama_wajib_pajak, :npwp,
                    :jalan, :block, :nomor, :rt, :rw, :kecamatan, :kelurahan,
                    :kabupaten, :propinsi, :kode_pos, :jalur_sales, :referensi,
                    :kode_tingkat, :kode_outlet, :wa, :email, :nik,
                    :sales_created, :kall, :program, :keterangan, :aktif,
                    :now, :now, '0'
                )
                """
            ),
            {**values, "kode_pelanggan": kode_pelanggan, "now": now},
        )

    async def update_customer(
        self, kode_pelanggan: str, request: CustomerWriteRequest, now: date
    ) -> bool:
        values = self._write_values(request)
        result = await self.session.execute(
            text(
                """
                UPDATE mspelanggan SET
                    namapelanggan=:nama_pelanggan, alamat=:alamat, telp=:telepon,
                    limitpiutang=:limit_piutang, limithari=:limit_hari,
                    namawajibpajak=:nama_wajib_pajak, npwp=:npwp, jalan=:jalan,
                    block=:block, nomor=:nomor, rt=:rt, rw=:rw,
                    kecamatan=:kecamatan, kelurahan=:kelurahan,
                    kabupaten=:kabupaten, propinsi=:propinsi, kodepos=:kode_pos,
                    jalursales=:jalur_sales, referensi=:referensi,
                    kodetingkat=:kode_tingkat, kodeoutlet=:kode_outlet, wa=:wa,
                    email=:email, nik=:nik, salescreated=:sales_created,
                    kall=:kall, program=:program, keterangan=:keterangan,
                    aktif=:aktif, tanggaledit=:now
                WHERE kodepelanggan=:kode_pelanggan AND COALESCE(DLT, '0')='0'
                """
            ),
            {**values, "kode_pelanggan": kode_pelanggan, "now": now},
        )
        return bool(result.rowcount)

    async def is_used_in_sales(self, kode_pelanggan: str) -> bool:
        result = await self.session.execute(
            text(
                """
                SELECT COUNT(*) FROM trsiheader
                WHERE kodepelanggan=:kode_pelanggan AND COALESCE(DLT, '0')='0'
                """
            ),
            {"kode_pelanggan": kode_pelanggan},
        )
        return bool(result.scalar_one())

    async def delete_customer(self, kode_pelanggan: str) -> bool:
        result = await self.session.execute(
            text("DELETE FROM mspelanggan WHERE kodepelanggan=:kode_pelanggan"),
            {"kode_pelanggan": kode_pelanggan},
        )
        return bool(result.rowcount)

    async def add_audit(
        self, *, user_code: str, user_name: str, kode: str, status: str
    ) -> None:
        await self.session.execute(
            text(
                """
                INSERT INTO xyz (
                    kodepengguna, namapengguna, kodetransaksi, status,
                    jumlahedit, tanggal
                ) VALUES (:user_code, :user_name, :kode, :status, 1, :tanggal)
                """
            ),
            {
                "user_code": user_code,
                "user_name": user_name,
                "kode": kode,
                "status": status,
                "tanggal": datetime.now(),
            },
        )

    async def list_lookup(self, kind: str, search: str | None) -> list[dict[str, str]]:
        definitions = {
            "jenis-outlet": ("msjenisoutlet", "kodeoutlet", "jenisoutlet"),
            "sales": ("mssales", "kodesales", "namasales"),
            "jalur": ("msjalur", "kodejalur", "namajalur"),
            "tingkat-harga": ("mstingkatharga", "kodetingkat", "tingkatharga"),
            "call-code": ("mscall", "kall", "keterangan"),
        }
        table, code_column, name_column = definitions[kind]
        active_filter = " AND aktif='AKTIF'" if kind in {"sales", "jalur"} else ""
        dlt_filter = "" if kind == "call-code" else " AND COALESCE(DLT, '0')='0'"
        result = await self.session.execute(
            text(
                f"""
                SELECT {code_column} AS kode, COALESCE({name_column}, '') AS nama
                FROM {table}
                WHERE (:search IS NULL OR {code_column} LIKE :pattern
                    OR {name_column} LIKE :pattern)
                    {active_filter}{dlt_filter}
                ORDER BY {name_column}, {code_column}
                LIMIT 100
                """
            ),
            {"search": search, "pattern": f"%{search}%" if search else None},
        )
        return [dict(row) for row in result.mappings().all()]

    async def receivable_summary(
        self, kode_pelanggan: str, *, kode_sales: str | None = None
    ) -> dict[str, Any]:
        sales_filter = "AND kodesales = :kode_sales" if kode_sales else ""
        result = await self.session.execute(
            text(
                f"""
                SELECT
                    COALESCE(SUM(CASE WHEN tanggaljatuhtempo <= CURRENT_DATE THEN sisa ELSE 0 END), 0) AS jt,
                    COALESCE(SUM(CASE WHEN tanggaljatuhtempo > CURRENT_DATE THEN sisa ELSE 0 END), 0) AS belum_jt,
                    COALESCE(SUM(sisa), 0) AS piutang
                FROM vwlistpiutang
                WHERE kodepelanggan=:kode_pelanggan AND sisa > 0.005
                  {sales_filter}
                """
            ),
            {"kode_pelanggan": kode_pelanggan, "kode_sales": kode_sales},
        )
        return dict(result.mappings().one())

    async def list_notifications(
        self,
        *,
        sampai_tanggal: date,
        kode_pelanggan: str | None,
        kode_sales: str | None,
        limit: int,
        offset: int,
    ) -> tuple[list[dict[str, Any]], int]:
        filters = """
            WHERE p.tanggalfaktur < DATE_ADD(:sampai_tanggal, INTERVAL 1 DAY)
              AND p.sisa > 0.005
              AND (:kode_pelanggan IS NULL OR p.kodepelanggan=:kode_pelanggan)
              AND (:kode_sales IS NULL OR p.kodesales=:kode_sales)
        """
        parameters = {
            "sampai_tanggal": sampai_tanggal,
            "kode_pelanggan": kode_pelanggan,
            "kode_sales": kode_sales,
        }
        count = await self.session.execute(
            text(f"SELECT COUNT(*) FROM vwlistpiutang AS p {filters}"), parameters
        )
        result = await self.session.execute(
            text(
                f"""
                SELECT DATE_FORMAT(p.tanggalfaktur, '%Y - %m') AS periode,
                    p.kodepelanggan AS kode_pelanggan,
                    p.namapelanggan AS nama_pelanggan,
                    p.kodesales AS kode_sales,
                    s.namasales AS nama_sales,
                    p.kodesi, p.tanggalfaktur AS tanggal_faktur,
                    p.tanggaljatuhtempo AS tanggal_jatuh_tempo,
                    COALESCE(p.total, 0) AS total,
                    COALESCE(p.returan, 0) AS returan,
                    COALESCE(p.bayar, 0) AS bayar,
                    COALESCE(p.sisa, 0) AS sisa
                FROM vwlistpiutang AS p
                LEFT JOIN mssales AS s ON s.kodesales=p.kodesales
                {filters}
                ORDER BY p.kodepelanggan, p.tanggalfaktur, p.kodesi
                LIMIT :limit OFFSET :offset
                """
            ),
            {**parameters, "limit": limit, "offset": offset},
        )
        return [dict(row) for row in result.mappings().all()], int(count.scalar_one())

    async def whatsapp_phone(self, kode_pelanggan: str) -> str | None:
        result = await self.session.execute(
            text(
                """
                SELECT NULLIF(TRIM(wa), '') FROM mspelanggan
                WHERE kodepelanggan=:kode_pelanggan AND COALESCE(DLT, '0')='0'
                """
            ),
            {"kode_pelanggan": kode_pelanggan},
        )
        return result.scalar_one_or_none()

    async def acquire_code_lock(self) -> bool:
        result = await self.session.execute(
            text("SELECT GET_LOCK('neoerp_customer_code', 10)")
        )
        return result.scalar_one() == 1

    async def release_code_lock(self) -> None:
        await self.session.execute(text("SELECT RELEASE_LOCK('neoerp_customer_code')"))