from __future__ import annotations

import hashlib
import io
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import pandas as pd
from sqlalchemy import func, select

from app.modules.accounting.sales_models import AccountingSalesIncomeTransaction
from app.modules.accounting.sales_taxonomy import ensure_sales_income_taxonomy, sales_income_natures
from app.modules.accounting.taxonomy_models import AccountingNature


COLUMN_ALIASES = {
    "invoice_number": [
        "invoice number", "invoice no", "invoice no.", "inv no", "document number",
        "document no", "voucher number", "bill no", "bill number",
    ],
    "invoice_date": ["invoice date", "document date", "date", "voucher date", "bill date"],
    "customer_name": [
        "customer name", "recipient name", "party name", "buyer name", "customer",
        "recipient", "party",
    ],
    "customer_gstin": [
        "customer gstin", "recipient gstin", "gstin", "gstin/uin", "gstin uin",
        "buyer gstin",
    ],
    "document_type": ["document type", "doc type", "invoice type", "voucher type"],
    "return_period": ["return period", "tax period", "period", "month"],
    "place_of_supply": ["place of supply", "pos", "state"],
    "supply_type": ["supply type", "nature of supply", "transaction type"],
    "export_type": ["export type", "exp type"],
    "hsn_code": ["hsn", "hsn code", "hsn/sac", "sac", "sac code"],
    "description_text": [
        "description", "item description", "product description", "particulars",
        "narration", "item name", "service description",
    ],
    "taxable_value": ["taxable value", "taxable amount", "assessable value", "basic amount"],
    "cgst": ["cgst", "cgst amount"],
    "sgst": ["sgst", "sgst amount", "utgst", "utgst amount"],
    "igst": ["igst", "igst amount"],
    "cess": ["cess", "cess amount"],
    "invoice_value": [
        "invoice value", "invoice amount", "total invoice value", "total amount",
        "gross amount", "bill amount", "voucher amount",
    ],
}


def _utcnow():
    return datetime.now(timezone.utc)


def _key(value: Any) -> str:
    return re.sub(r"[^a-z0-9]+", " ", str(value or "").strip().lower()).strip()


def _s(value: Any) -> str:
    if value is None:
        return ""
    try:
        if pd.isna(value):
            return ""
    except Exception:
        pass
    return str(value).strip()


def _f(value: Any) -> float:
    if value is None:
        return 0.0
    try:
        if pd.isna(value):
            return 0.0
    except Exception:
        pass
    text = str(value).replace(",", "").replace("₹", "").strip()
    if text.startswith("(") and text.endswith(")"):
        text = "-" + text[1:-1]
    try:
        return round(float(text or 0), 2)
    except Exception:
        return 0.0


def _date(value: Any) -> str:
    dt = pd.to_datetime(value, errors="coerce", dayfirst=True)
    if pd.isna(dt):
        return _s(value)[:20]
    return dt.strftime("%Y-%m-%d")


def _columns(frame: pd.DataFrame) -> dict[str, str]:
    keyed = {_key(column): column for column in frame.columns}
    result = {}
    for target, aliases in COLUMN_ALIASES.items():
        for alias in aliases:
            if _key(alias) in keyed:
                result[target] = keyed[_key(alias)]
                break
    return result


def _read_upload(filename: str, payload: bytes) -> pd.DataFrame:
    suffix = Path(filename or "").suffix.lower()
    if suffix == ".csv":
        return pd.read_csv(io.BytesIO(payload))
    if suffix in {".xlsx", ".xlsm", ".xls"}:
        book = pd.ExcelFile(io.BytesIO(payload))
        candidates = []
        for sheet in book.sheet_names:
            frame = pd.read_excel(book, sheet_name=sheet)
            mapping = _columns(frame)
            score = len(mapping)
            if "invoice_number" in mapping:
                score += 3
            if "customer_name" in mapping or "customer_gstin" in mapping:
                score += 2
            if "taxable_value" in mapping or "invoice_value" in mapping:
                score += 2
            candidates.append((score, sheet, frame))
        candidates.sort(key=lambda item: (-item[0], item[1]))
        if not candidates or candidates[0][0] < 3:
            raise ValueError(
                "No usable outward-sales sheet was found. Expected invoice/customer/value columns."
            )
        return candidates[0][2]
    raise ValueError("Sales & Income import supports CSV, XLSX, XLSM and XLS files.")


def _nature_map(db, tenant_id: int):
    ensure_sales_income_taxonomy(db, tenant_id)
    rows = sales_income_natures(db, tenant_id)
    return {row.code: row for row in rows}


def _classify(data: dict[str, Any], nature_by_code: dict[str, AccountingNature]):
    haystack = " ".join([
        _s(data.get("document_type")),
        _s(data.get("supply_type")),
        _s(data.get("export_type")),
        _s(data.get("description_text")),
        _s(data.get("customer_name")),
    ]).upper()
    hsn = re.sub(r"\D", "", _s(data.get("hsn_code")))

    def result(code, confidence, reason):
        nature = nature_by_code.get(code) or nature_by_code["SALES_REVIEW_REQUIRED"]
        return nature, confidence, reason

    if any(token in haystack for token in ("EXPORT", "SEZ", "WITH PAYMENT", "WITHOUT PAYMENT")):
        return result("EXPORT_SALES", 92, "Export/SEZ wording in outward transaction.")
    if any(token in haystack for token in ("EXEMPT", "NIL RATED", "NIL-RATED", "NON GST", "NON-GST")):
        return result("EXEMPT_NIL_SALES", 92, "Exempt/nil/non-GST supply wording.")
    if any(token in haystack for token in ("SCRAP", "WASTE", "OLD NEWSPAPER", "METAL WASTE")):
        return result("SCRAP_SALES", 90, "Scrap/waste sale wording.")
    if any(token in haystack for token in ("COMMISSION", "BROKERAGE")):
        return result("COMMISSION_INCOME", 88, "Commission/brokerage income wording.")
    if any(token in haystack for token in ("RENTAL", "RENT ", "HIRE CHARGES", "LEASE INCOME")):
        return result("RENTAL_INCOME", 86, "Rental/hire/lease income wording.")
    if any(token in haystack for token in ("INTEREST", "FD INTEREST", "LOAN INTEREST")):
        return result("INTEREST_INCOME", 90, "Interest income wording.")
    if any(token in haystack for token in ("JOB WORK", "JOBWORK", "PROCESSING CHARGES", "CONVERSION CHARGES")):
        return result("JOB_WORK_INCOME", 88, "Job-work/processing income wording.")
    if any(token in haystack for token in (
        "PROFESSIONAL", "CONSULTANCY", "SERVICE CHARGES", "SERVICE FEE",
        "AUDIT FEE", "ACCOUNTING FEE", "MANAGEMENT FEE", "LABOUR CHARGES",
    )):
        return result("SERVICE_INCOME", 86, "Service/professional income wording.")
    if any(token in haystack for token in (
        "SALE OF VEHICLE", "SALE OF MACHINERY", "SALE OF ASSET",
        "OLD VEHICLE SALE", "OLD MACHINERY SALE",
    )):
        return result("ASSET_SALE", 90, "Capital-asset disposal wording.")

    # SAC/service HSN chapters commonly begin 99. This is only a first-pass signal.
    if hsn.startswith("99"):
        return result("SERVICE_INCOME", 78, "SAC/HSN 99xx indicates service supply.")

    # Generic outward invoice with goods HSN is treated conservatively as trading sales.
    if hsn and not hsn.startswith("99"):
        return result(
            "TRADING_SALES",
            68,
            "Goods HSN present; trading/manufacturing distinction requires business context or review.",
        )

    if _s(data.get("description_text")):
        return result(
            "OTHER_OPERATING_INCOME",
            55,
            "Outward transaction has description but insufficient evidence for a narrower nature.",
        )
    return result("SALES_REVIEW_REQUIRED", 30, "Insufficient evidence for reliable income classification.")


def _fingerprint(client_id: int, data: dict[str, Any]) -> str:
    raw = "|".join([
        str(client_id),
        _s(data.get("invoice_number")).upper(),
        _s(data.get("invoice_date")),
        _s(data.get("customer_gstin")).upper(),
        _s(data.get("customer_name")).upper(),
        f"{_f(data.get('taxable_value')):.2f}",
        f"{_f(data.get('invoice_value')):.2f}",
        _s(data.get("document_type")).upper(),
    ])
    return hashlib.sha256(raw.encode("utf-8", "ignore")).hexdigest()


def import_sales_file(
    db,
    *,
    tenant_id: int,
    client_id: int,
    filename: str,
    payload: bytes,
    user_id: int,
    source_type: str = "outward_import",
):
    frame = _read_upload(filename, payload)
    mapping = _columns(frame)

    required = {"invoice_number"}
    if not required.issubset(mapping):
        raise ValueError("Could not identify an Invoice Number column in the uploaded file.")
    if not ({"customer_name", "customer_gstin"} & set(mapping)):
        raise ValueError("Could not identify Customer Name or Customer GSTIN.")
    if not ({"taxable_value", "invoice_value"} & set(mapping)):
        raise ValueError("Could not identify Taxable Value or Invoice Value.")

    nature_by_code = _nature_map(db, tenant_id)
    inserted = 0
    duplicates = 0
    skipped_blank = 0

    for _, row in frame.iterrows():
        data = {
            target: row[column]
            for target, column in mapping.items()
        }
        data = {
            "invoice_number": _s(data.get("invoice_number")),
            "invoice_date": _date(data.get("invoice_date")),
            "customer_name": _s(data.get("customer_name")),
            "customer_gstin": re.sub(r"\s+", "", _s(data.get("customer_gstin")).upper()),
            "document_type": _s(data.get("document_type")) or "Invoice",
            "return_period": _s(data.get("return_period")),
            "place_of_supply": _s(data.get("place_of_supply")),
            "supply_type": _s(data.get("supply_type")),
            "export_type": _s(data.get("export_type")),
            "hsn_code": re.sub(r"\.0$", "", _s(data.get("hsn_code"))),
            "description_text": _s(data.get("description_text")),
            "taxable_value": _f(data.get("taxable_value")),
            "cgst": _f(data.get("cgst")),
            "sgst": _f(data.get("sgst")),
            "igst": _f(data.get("igst")),
            "cess": _f(data.get("cess")),
            "invoice_value": _f(data.get("invoice_value")),
        }

        if not data["invoice_number"] and not data["customer_name"] and not data["customer_gstin"]:
            skipped_blank += 1
            continue
        if data["invoice_value"] == 0:
            data["invoice_value"] = round(
                data["taxable_value"] + data["cgst"] + data["sgst"] + data["igst"] + data["cess"],
                2,
            )

        fp = _fingerprint(client_id, data)
        exists = db.execute(
            select(AccountingSalesIncomeTransaction.id).where(
                AccountingSalesIncomeTransaction.tenant_id == int(tenant_id),
                AccountingSalesIncomeTransaction.client_id == int(client_id),
                AccountingSalesIncomeTransaction.fingerprint == fp,
            )
        ).scalar_one_or_none()
        if exists:
            duplicates += 1
            continue

        nature, confidence, reason = _classify(data, nature_by_code)
        record = AccountingSalesIncomeTransaction(
            tenant_id=int(tenant_id),
            client_id=int(client_id),
            source_type=source_type,
            source_file_name=Path(filename or "").name[:255],
            fingerprint=fp,
            return_period=data["return_period"],
            document_type=data["document_type"],
            invoice_number=data["invoice_number"],
            invoice_date=data["invoice_date"],
            customer_name=data["customer_name"],
            customer_gstin=data["customer_gstin"],
            place_of_supply=data["place_of_supply"],
            supply_type=data["supply_type"],
            export_type=data["export_type"],
            hsn_code=data["hsn_code"],
            description_text=data["description_text"],
            taxable_value=data["taxable_value"],
            cgst=data["cgst"],
            sgst=data["sgst"],
            igst=data["igst"],
            cess=data["cess"],
            invoice_value=data["invoice_value"],
            suggested_nature_id=nature.id,
            suggested_confidence=int(confidence),
            suggestion_reason=reason,
            review_status="suggested" if confidence >= 60 else "review_required",
            created_by_user_id=user_id,
        )
        db.add(record)
        inserted += 1

    db.commit()
    return {
        "inserted": inserted,
        "duplicates": duplicates,
        "skipped_blank": skipped_blank,
        "source_rows": len(frame),
    }


def list_sales(
    db,
    *,
    tenant_id: int,
    client_id: int,
    status: str = "",
    customer: str = "",
    period: str = "",
    page: int = 1,
    per_page: int = 25,
):
    where = [
        AccountingSalesIncomeTransaction.tenant_id == int(tenant_id),
        AccountingSalesIncomeTransaction.client_id == int(client_id),
    ]
    if status and status != "all":
        where.append(AccountingSalesIncomeTransaction.review_status == status)
    if customer.strip():
        term = f"%{customer.strip()}%"
        where.append(
            (
                AccountingSalesIncomeTransaction.customer_name.ilike(term)
                | AccountingSalesIncomeTransaction.customer_gstin.ilike(term)
            )
        )
    if period.strip():
        where.append(AccountingSalesIncomeTransaction.return_period == period.strip())

    total = int(
        db.scalar(
            select(func.count(AccountingSalesIncomeTransaction.id)).where(*where)
        )
        or 0
    )
    per_page = max(10, min(100, int(per_page or 25)))
    pages = max(1, (total + per_page - 1) // per_page)
    page = max(1, min(int(page or 1), pages))

    rows = list(
        db.execute(
            select(AccountingSalesIncomeTransaction)
            .where(*where)
            .order_by(
                AccountingSalesIncomeTransaction.invoice_date.desc(),
                AccountingSalesIncomeTransaction.id.desc(),
            )
            .offset((page - 1) * per_page)
            .limit(per_page)
        ).scalars().all()
    )
    return rows, total, page, pages


def review_sales(
    db,
    *,
    tenant_id: int,
    client_id: int,
    row_id: int,
    final_nature_id: int,
    final_ledger_name: str,
    review_note: str,
    user_id: int,
):
    row = db.execute(
        select(AccountingSalesIncomeTransaction).where(
            AccountingSalesIncomeTransaction.id == int(row_id),
            AccountingSalesIncomeTransaction.tenant_id == int(tenant_id),
            AccountingSalesIncomeTransaction.client_id == int(client_id),
        )
    ).scalar_one_or_none()
    if not row:
        raise ValueError("Sales / income transaction was not found.")

    nature = db.execute(
        select(AccountingNature).where(
            AccountingNature.id == int(final_nature_id),
            AccountingNature.tenant_id == int(tenant_id),
            AccountingNature.classification_group == "income",
            AccountingNature.is_posting_nature.is_(True),
            AccountingNature.is_active.is_(True),
        )
    ).scalar_one_or_none()
    if not nature:
        raise ValueError("Select an active Sales & Income accounting nature.")

    row.final_nature_id = nature.id
    row.final_ledger_name = (final_ledger_name or "").strip()
    row.review_note = (review_note or "").strip() or None
    row.review_status = "reviewed"
    row.reviewed_by_user_id = user_id
    row.reviewed_at_utc = _utcnow()
    db.add(row)
    db.commit()
    db.refresh(row)
    return row