from __future__ import annotations

from datetime import date, datetime, timezone
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from io import BytesIO
from typing import Any

from openpyxl import Workbook, load_workbook
from sqlalchemy import and_, or_, select
from sqlalchemy.orm import Session, selectinload

from app.modules.billing.models import (
    BillingFeeGroup,
    BillingFeeGroupService,
    BillingInvoice,
    BillingInvoiceLine,
    BillingInvoiceGenerationBatch,
    BillingPayment,
    BillingSettings,
    BillingOnlinePaymentTransaction,
)
from app.modules.clients.models import Client
from app.modules.email_integration.event_service import send_invoice_issued_email, send_payment_receipt_email
from app.modules.services.models import ClientServiceSubscription, ServiceCatalogue

TAX_TYPES = ["CGST_SGST", "IGST", "NO_GST"]
BILLING_MODES = ["PACKAGE", "SERVICE_WISE"]
FREQUENCIES = ["Monthly", "Quarterly", "Half-Yearly", "Yearly", "One-time"]
INVOICE_STATUSES = ["DRAFT", "ISSUED", "PARTLY_PAID", "PAID", "OVERDUE", "CANCELLED", "WRITTEN_OFF"]
PAYMENT_MODES = ["CASH", "BANK", "UPI", "CHEQUE", "ONLINE", "ADJUSTMENT"]
PAYMENT_STATUSES = ["RECEIVED", "CANCELLED", "REFUNDED"]


def money(value: Any) -> Decimal:
    try:
        if value in (None, ""):
            return Decimal("0.00")
        return Decimal(str(value)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
    except (InvalidOperation, ValueError):
        return Decimal("0.00")


def parse_date(value: Any) -> date | None:
    if value in (None, ""):
        return None
    if isinstance(value, datetime):
        return value.date()
    if isinstance(value, date):
        return value
    value = str(value).strip()
    if not value:
        return None
    return date.fromisoformat(value)


def normalize_code(value: Any) -> str:
    return str(value or "").strip().upper().replace(" ", "-")


def normalize_yes_no(value: Any) -> bool:
    return str(value or "").strip().lower() in {"yes", "y", "true", "1", "active"}


def get_or_create_settings(db: Session, *, tenant_id: int, branch_id: int | None = None) -> BillingSettings:
    row = db.execute(
        select(BillingSettings).where(BillingSettings.tenant_id == tenant_id, BillingSettings.branch_id == branch_id)
    ).scalar_one_or_none()
    if row:
        return row
    row = BillingSettings(tenant_id=tenant_id, branch_id=branch_id)
    db.add(row)
    db.flush()
    return row


def _financial_year_label(value: date | None = None) -> str:
    value = value or date.today()
    start_year = value.year if value.month >= 4 else value.year - 1
    return f"{start_year}-{str(start_year + 1)[-2:]}"


def billing_financial_year(*, billing_period_from: date | None = None, invoice_date: date | None = None, fallback: date | None = None) -> str:
    return _financial_year_label(billing_period_from or invoice_date or fallback or date.today())


def next_invoice_number(db: Session, *, tenant_id: int, branch_id: int | None = None, financial_year: str | None = None) -> str:
    settings = get_or_create_settings(db, tenant_id=tenant_id, branch_id=branch_id)
    serial = str(settings.next_invoice_no).zfill(settings.padding or 4)
    fy = financial_year or _financial_year_label()
    fmt = (getattr(settings, "invoice_number_format", None) or "{prefix}/{fy}/{number}").strip()
    try:
        number = fmt.format(prefix=settings.invoice_prefix or "INV", fy=fy, number=serial, branch_id=branch_id or "")
    except Exception:
        number = f"{settings.invoice_prefix or 'INV'}/{fy}/{serial}"
    settings.next_invoice_no += 1
    settings.updated_at_utc = datetime.now(timezone.utc)
    return number


def preview_invoice_number(settings: BillingSettings, *, branch_id: int | None = None, financial_year: str | None = None) -> str:
    serial = str((settings.next_invoice_no or 1)).zfill(settings.padding or 4)
    fy = financial_year or _financial_year_label()
    fmt = (getattr(settings, "invoice_number_format", None) or "{prefix}/{fy}/{number}").strip()
    try:
        return fmt.format(prefix=settings.invoice_prefix or "INV", fy=fy, number=serial, branch_id=branch_id or "")
    except Exception:
        return f"{settings.invoice_prefix or 'INV'}/{fy}/{serial}"



_ONES = ["", "One", "Two", "Three", "Four", "Five", "Six", "Seven", "Eight", "Nine", "Ten", "Eleven", "Twelve", "Thirteen", "Fourteen", "Fifteen", "Sixteen", "Seventeen", "Eighteen", "Nineteen"]
_TENS = ["", "", "Twenty", "Thirty", "Forty", "Fifty", "Sixty", "Seventy", "Eighty", "Ninety"]


def _words_below_1000(n: int) -> str:
    parts: list[str] = []
    if n >= 100:
        parts.append(_ONES[n // 100] + " Hundred")
        n %= 100
    if n >= 20:
        parts.append(_TENS[n // 10])
        n %= 10
    if n > 0:
        parts.append(_ONES[n])
    return " ".join(parts)


def amount_to_indian_words(value: Any) -> str:
    amount = money(value)
    rupees = int(amount)
    paise = int((amount - Decimal(rupees)) * 100)
    if rupees == 0:
        words = "Zero"
    else:
        parts: list[str] = []
        crore, rupees = divmod(rupees, 10000000)
        lakh, rupees = divmod(rupees, 100000)
        thousand, rupees = divmod(rupees, 1000)
        if crore:
            parts.append(_words_below_1000(crore) + " Crore")
        if lakh:
            parts.append(_words_below_1000(lakh) + " Lakh")
        if thousand:
            parts.append(_words_below_1000(thousand) + " Thousand")
        if rupees:
            parts.append(_words_below_1000(rupees))
        words = " ".join(parts)
    result = f"Rupees {words} Only"
    if paise:
        result = f"Rupees {words} and Paise {_words_below_1000(paise)} Only"
    return result


def _client_address_snapshot(client: Client) -> str | None:
    parts = [
        getattr(client, "address_line_1", None),
        getattr(client, "address_line_2", None),
        getattr(client, "city", None),
        getattr(client, "state", None),
        getattr(client, "pincode", None),
        getattr(client, "country", None),
    ]
    return ", ".join([str(p).strip() for p in parts if str(p or "").strip()]) or None


def _state_code_from_gstin(gstin: str | None) -> str | None:
    value = str(gstin or "").strip()
    if len(value) >= 2 and value[:2].isdigit():
        return value[:2]
    return None


def get_effective_billing_settings(db: Session, *, tenant_id: int, branch_id: int | None = None) -> BillingSettings:
    if branch_id:
        row = db.execute(select(BillingSettings).where(BillingSettings.tenant_id == tenant_id, BillingSettings.branch_id == branch_id)).scalar_one_or_none()
        if row:
            return row
    return get_or_create_settings(db, tenant_id=tenant_id, branch_id=None)


def build_invoice_print_context(db: Session, invoice: BillingInvoice) -> dict[str, Any]:
    settings = get_effective_billing_settings(db, tenant_id=invoice.tenant_id, branch_id=invoice.branch_id)
    return {
        "invoice": invoice,
        "settings": settings,
        "invoice_title": invoice.invoice_title or settings.invoice_title or "Tax Invoice",
        "firm_name": settings.legal_name or "Audit Firm",
        "firm_address": settings.billing_address,
        "firm_gstin": settings.gstin,
        "firm_pan": settings.pan,
        "firm_state_code": settings.state_code,
        "firm_contact_email": settings.contact_email,
        "firm_contact_mobile": settings.contact_mobile,
        "firm_website": settings.website_url,
        "bank_name": settings.bank_name,
        "bank_account_name": settings.bank_account_name,
        "bank_account_number": settings.bank_account_number,
        "bank_ifsc": settings.bank_ifsc,
        "upi_id": settings.upi_id,
        "bank_details": settings.bank_details,
        "declaration": invoice.notes or settings.declaration,
        "terms": invoice.terms or settings.terms,
        "footer_note": settings.footer_note,
        "authorised_signatory_name": settings.authorised_signatory_name,
    }

def calculate_line(*, quantity: Decimal, rate: Decimal, discount: Decimal, gst_rate: Decimal, tax_type: str) -> dict[str, Decimal]:
    taxable = (quantity * rate - discount).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
    if taxable < 0:
        taxable = Decimal("0.00")
    cgst = sgst = igst = Decimal("0.00")
    if tax_type == "IGST":
        igst = (taxable * gst_rate / Decimal("100.00")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
    elif tax_type == "CGST_SGST":
        half = (taxable * gst_rate / Decimal("200.00")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
        cgst = half
        sgst = half
    total = taxable + cgst + sgst + igst
    return {"taxable": taxable, "cgst": cgst, "sgst": sgst, "igst": igst, "total": total}


def recalc_invoice(invoice: BillingInvoice) -> None:
    subtotal = Decimal("0.00")
    discount = Decimal("0.00")
    taxable = Decimal("0.00")
    cgst = Decimal("0.00")
    sgst = Decimal("0.00")
    igst = Decimal("0.00")
    total = Decimal("0.00")
    for line in invoice.lines:
        subtotal += money(line.quantity) * money(line.rate)
        discount += money(line.discount_amount)
        taxable += money(line.taxable_amount)
        cgst += money(line.cgst_amount)
        sgst += money(line.sgst_amount)
        igst += money(line.igst_amount)
        total += money(line.line_total)
    invoice.subtotal = money(subtotal)
    invoice.discount_amount = money(discount)
    invoice.taxable_amount = money(taxable)
    invoice.cgst_amount = money(cgst)
    invoice.sgst_amount = money(sgst)
    invoice.igst_amount = money(igst)
    invoice.total_amount = money(total + money(invoice.round_off))
    update_invoice_payment_totals(invoice)


def update_invoice_payment_totals(invoice: BillingInvoice) -> None:
    paid = Decimal("0.00")
    tds = Decimal("0.00")
    charges = Decimal("0.00")
    for payment in getattr(invoice, "payments", []) or []:
        if getattr(payment, "status", "RECEIVED") != "CANCELLED":
            paid += money(payment.amount_received)
            tds += money(payment.tds_deducted)
            charges += money(payment.bank_charges)
    invoice.amount_received = money(paid)
    invoice.tds_deducted = money(tds)
    invoice.bank_charges = money(charges)
    invoice.balance_amount = money(money(invoice.total_amount) - paid - tds)
    if invoice.balance_amount < Decimal("0.00"):
        invoice.balance_amount = Decimal("0.00")
    if invoice.status not in {"DRAFT", "CANCELLED", "WRITTEN_OFF"}:
        if invoice.balance_amount <= Decimal("0.00") and money(invoice.total_amount) > Decimal("0.00"):
            invoice.status = "PAID"
        elif paid > Decimal("0.00") or tds > Decimal("0.00"):
            invoice.status = "PARTLY_PAID"
        elif invoice.status == "PAID":
            invoice.status = "ISSUED"


def next_receipt_number(db: Session, *, tenant_id: int, branch_id: int | None = None, financial_year: str | None = None) -> str:
    fy = financial_year or _financial_year_label()
    prefix = "RCT"
    like_prefix = f"{prefix}/{fy}/%"
    last = db.execute(
        select(BillingPayment.receipt_no)
        .where(BillingPayment.tenant_id == tenant_id, BillingPayment.receipt_no.ilike(like_prefix))
        .order_by(BillingPayment.id.desc())
    ).scalar_one_or_none()
    next_no = 1
    if last:
        try:
            next_no = int(str(last).split("/")[-1]) + 1
        except Exception:
            next_no = 1
    return f"{prefix}/{fy}/{str(next_no).zfill(4)}"


def record_invoice_payment(
    db: Session,
    *,
    invoice: BillingInvoice,
    payment_date: date,
    amount_received: Decimal,
    tds_deducted: Decimal = Decimal("0.00"),
    bank_charges: Decimal = Decimal("0.00"),
    mode: str = "BANK",
    reference_no: str | None = None,
    remarks: str | None = None,
    created_by_user_id: int | None = None,
    payment_gateway: str | None = None,
    gateway_transaction_id: str | None = None,
) -> BillingPayment:
    if invoice.status in {"DRAFT", "CANCELLED"}:
        raise ValueError("Payment can be recorded only after invoice is issued.")
    mode = mode if mode in PAYMENT_MODES else "BANK"
    payment = BillingPayment(
        tenant_id=invoice.tenant_id,
        branch_id=invoice.branch_id,
        invoice_id=invoice.id,
        client_id=invoice.client_id,
        receipt_no=next_receipt_number(db, tenant_id=invoice.tenant_id, branch_id=invoice.branch_id, financial_year=getattr(invoice, "financial_year", None)),
        receipt_date=payment_date,
        payment_date=payment_date,
        financial_year=getattr(invoice, "financial_year", None) or billing_financial_year(invoice_date=payment_date),
        amount_received=money(amount_received),
        tds_deducted=money(tds_deducted),
        bank_charges=money(bank_charges),
        mode=mode,
        reference_no=(reference_no or "").strip() or None,
        payment_gateway=(payment_gateway or "").strip() or None,
        gateway_transaction_id=(gateway_transaction_id or "").strip() or None,
        remarks=(remarks or "").strip() or None,
        created_by_user_id=created_by_user_id,
        status="RECEIVED",
    )
    db.add(payment)
    db.flush()
    if payment not in invoice.payments:
        invoice.payments.append(payment)
    update_invoice_payment_totals(invoice)
    invoice.updated_at_utc = datetime.now(timezone.utc)
    db.flush()
    try:
        send_payment_receipt_email(db, payment)
    except Exception:
        # Email failure should not block payment posting or receipt generation.
        pass
    return payment


def list_payments(db: Session, *, tenant_id: int, branch_id: int | None = None, partner_id: int | None = None, financial_year: str | None = None, q: str = ""):
    stmt = select(BillingPayment).options(selectinload(BillingPayment.invoice), selectinload(BillingPayment.client)).where(BillingPayment.tenant_id == tenant_id)
    if branch_id:
        stmt = stmt.where(BillingPayment.branch_id == branch_id)
    if partner_id:
        stmt = stmt.join(Client, Client.id == BillingPayment.client_id).where(Client.partner_id == partner_id)
    if financial_year and financial_year.upper() != "ALL":
        stmt = stmt.where(BillingPayment.financial_year == financial_year)
    if q.strip():
        term = f"%{q.strip()}%"
        stmt = stmt.join(BillingInvoice, BillingInvoice.id == BillingPayment.invoice_id).join(Client, Client.id == BillingPayment.client_id).where(or_(BillingPayment.receipt_no.ilike(term), BillingInvoice.invoice_no.ilike(term), Client.client_name.ilike(term), Client.client_code.ilike(term)))
    return db.execute(stmt.order_by(BillingPayment.payment_date.desc(), BillingPayment.id.desc())).scalars().unique().all()


def get_payment(db: Session, *, payment_id: int, tenant_id: int, partner_id: int | None = None, financial_year: str | None = None) -> BillingPayment | None:
    stmt = select(BillingPayment).options(selectinload(BillingPayment.invoice).selectinload(BillingInvoice.lines), selectinload(BillingPayment.client)).where(BillingPayment.id == payment_id, BillingPayment.tenant_id == tenant_id)
    if partner_id:
        stmt = stmt.join(Client, Client.id == BillingPayment.client_id).where(Client.partner_id == partner_id)
    if financial_year and financial_year.upper() != "ALL":
        stmt = stmt.where(BillingPayment.financial_year == financial_year)
    return db.execute(stmt).scalars().unique().one_or_none()


def list_clients_for_billing(db: Session, *, tenant_id: int, branch_id: int | None = None, partner_id: int | None = None, q: str = ""):
    stmt = select(Client).where(Client.tenant_id == tenant_id, Client.is_archived.is_(False))
    if branch_id:
        stmt = stmt.where(Client.branch_id == branch_id)
    if partner_id:
        stmt = stmt.where(Client.partner_id == partner_id)
    if q.strip():
        term = f"%{q.strip()}%"
        stmt = stmt.where(or_(Client.client_name.ilike(term), Client.client_code.ilike(term), Client.pan.ilike(term), Client.gstin.ilike(term)))
    return db.execute(stmt.order_by(Client.client_name.asc())).scalars().all()


def list_services_for_billing(db: Session):
    return db.execute(
        select(ServiceCatalogue).where(ServiceCatalogue.is_active.is_(True)).order_by(ServiceCatalogue.service_name.asc())
    ).scalars().all()


def list_invoices(db: Session, *, tenant_id: int, branch_id: int | None = None, partner_id: int | None = None, financial_year: str | None = None, q: str = ""):
    stmt = select(BillingInvoice).options(selectinload(BillingInvoice.client), selectinload(BillingInvoice.lines), selectinload(BillingInvoice.payments)).where(BillingInvoice.tenant_id == tenant_id)
    if branch_id:
        stmt = stmt.where(BillingInvoice.branch_id == branch_id)
    if partner_id:
        stmt = stmt.join(Client, Client.id == BillingInvoice.client_id).where(Client.partner_id == partner_id)
    if financial_year and financial_year.upper() != "ALL":
        stmt = stmt.where(BillingInvoice.financial_year == financial_year)
    if q.strip():
        term = f"%{q.strip()}%"
        stmt = stmt.join(Client, Client.id == BillingInvoice.client_id).where(or_(BillingInvoice.invoice_no.ilike(term), Client.client_name.ilike(term), Client.client_code.ilike(term)))
    return db.execute(stmt.order_by(BillingInvoice.invoice_date.desc(), BillingInvoice.id.desc())).scalars().unique().all()


def build_billing_report_summary(db: Session, *, tenant_id: int, branch_id: int | None = None, partner_id: int | None = None, financial_year: str | None = None) -> dict[str, Any]:
    invoices = list_invoices(db, tenant_id=tenant_id, branch_id=branch_id, partner_id=partner_id, financial_year=financial_year)
    payments = list_payments(db, tenant_id=tenant_id, branch_id=branch_id, partner_id=partner_id, financial_year=financial_year)
    total_billed = sum((money(row.total_amount) for row in invoices if row.status != "CANCELLED"), Decimal("0.00"))
    total_received = sum((money(row.amount_received) for row in payments if row.status == "RECEIVED"), Decimal("0.00"))
    total_tds = sum((money(row.tds_deducted) for row in payments if row.status == "RECEIVED"), Decimal("0.00"))
    outstanding = sum((money(row.balance_amount) for row in invoices if row.status in {"ISSUED", "PARTLY_PAID", "OVERDUE"}), Decimal("0.00"))
    draft_count = sum(1 for row in invoices if row.status == "DRAFT")
    issued_count = sum(1 for row in invoices if row.status in {"ISSUED", "PARTLY_PAID", "OVERDUE"})
    paid_count = sum(1 for row in invoices if row.status == "PAID")
    return {
        "invoice_count": len(invoices),
        "draft_count": draft_count,
        "issued_count": issued_count,
        "paid_count": paid_count,
        "payment_count": len(payments),
        "total_billed": money(total_billed),
        "total_received": money(total_received),
        "total_tds": money(total_tds),
        "total_collected_with_tds": money(total_received + total_tds),
        "outstanding": money(outstanding),
    }


def get_invoice(db: Session, *, invoice_id: int, tenant_id: int, partner_id: int | None = None, financial_year: str | None = None) -> BillingInvoice | None:
    stmt = select(BillingInvoice).options(selectinload(BillingInvoice.client), selectinload(BillingInvoice.lines).selectinload(BillingInvoiceLine.service), selectinload(BillingInvoice.payments)).where(BillingInvoice.id == invoice_id, BillingInvoice.tenant_id == tenant_id)
    if partner_id:
        stmt = stmt.join(Client, Client.id == BillingInvoice.client_id).where(Client.partner_id == partner_id)
    if financial_year and financial_year.upper() != "ALL":
        stmt = stmt.where(BillingInvoice.financial_year == financial_year)
    return db.execute(stmt).scalars().unique().one_or_none()


def create_invoice(
    db: Session,
    *,
    tenant_id: int,
    branch_id: int | None,
    client_id: int,
    invoice_date: date,
    due_date: date | None,
    billing_period_from: date | None,
    billing_period_to: date | None,
    tax_type: str,
    notes: str | None,
    terms: str | None,
    place_of_supply: str | None = None,
    client_state_code: str | None = None,
    reverse_charge: bool = False,
    created_by_user_id: int,
    raw_lines: list[dict[str, Any]],
    generation_batch_id: int | None = None,
    engagement_id: int | None = None,
    financial_year: str | None = None,
) -> BillingInvoice:
    settings = get_effective_billing_settings(db, tenant_id=tenant_id, branch_id=branch_id)
    client = db.get(Client, client_id)
    if client is None:
        raise ValueError("Client not found.")
    snapshot_state_code = (client_state_code or _state_code_from_gstin(getattr(client, "gstin", None)) or "").strip()[:2] or None
    invoice_financial_year = financial_year or billing_financial_year(billing_period_from=billing_period_from, invoice_date=invoice_date)
    invoice = BillingInvoice(
        tenant_id=tenant_id,
        branch_id=branch_id,
        client_id=client_id,
        engagement_id=engagement_id,
        invoice_no=next_invoice_number(db, tenant_id=tenant_id, branch_id=branch_id, financial_year=invoice_financial_year),
        invoice_date=invoice_date,
        due_date=due_date,
        billing_period_from=billing_period_from,
        billing_period_to=billing_period_to,
        financial_year=invoice_financial_year,
        invoice_title=settings.invoice_title or "Tax Invoice",
        place_of_supply=(place_of_supply or getattr(client, "state", None) or "").strip() or None,
        reverse_charge=bool(reverse_charge),
        client_legal_name=getattr(client, "client_name", None),
        client_trade_name=getattr(client, "trade_name", None),
        client_gstin=(getattr(client, "gstin", None) or "").strip().upper() or None,
        client_pan=(getattr(client, "pan", None) or "").strip().upper() or None,
        client_billing_address=_client_address_snapshot(client),
        client_state=getattr(client, "state", None),
        client_state_code=snapshot_state_code,
        client_email=getattr(client, "email", None),
        client_mobile=getattr(client, "mobile", None),
        tax_type=tax_type if tax_type in TAX_TYPES else (settings.default_tax_type if settings.default_tax_type in TAX_TYPES else "CGST_SGST"),
        notes=(notes or "").strip() or None,
        terms=(terms or settings.terms or "").strip() or None,
        created_by_user_id=created_by_user_id,
        generation_batch_id=generation_batch_id,
        status="DRAFT",
        balance_amount=Decimal("0.00"),
    )
    db.add(invoice)
    db.flush()

    sort_order = 1
    for raw in raw_lines:
        description = str(raw.get("description") or "").strip()
        if not description:
            continue
        qty = money(raw.get("quantity") or 1)
        if qty <= 0:
            qty = Decimal("1.00")
        rate = money(raw.get("rate"))
        disc = money(raw.get("discount_amount"))
        gst_rate = money(raw.get("gst_rate") or settings.default_gst_rate)
        sac_code = str(raw.get("sac_code") or settings.default_sac_code or "").strip()[:20] or None
        calc = calculate_line(quantity=qty, rate=rate, discount=disc, gst_rate=gst_rate, tax_type=invoice.tax_type)
        service_id = raw.get("service_id") or None
        fee_group_id = raw.get("fee_group_id") or None
        raw_engagement_id = raw.get("engagement_id") or None
        line = BillingInvoiceLine(
            invoice_id=invoice.id,
            service_id=int(service_id) if service_id else None,
            fee_group_id=int(fee_group_id) if fee_group_id else None,
            engagement_id=int(raw_engagement_id) if raw_engagement_id else None,
            description=description,
            sac_code=sac_code,
            billing_period_from=billing_period_from,
            billing_period_to=billing_period_to,
            quantity=qty,
            rate=rate,
            discount_amount=disc,
            taxable_amount=calc["taxable"],
            gst_rate=gst_rate,
            cgst_amount=calc["cgst"],
            sgst_amount=calc["sgst"],
            igst_amount=calc["igst"],
            line_total=calc["total"],
            sort_order=sort_order,
        )
        db.add(line)
        invoice.lines.append(line)
        sort_order += 1

    if sort_order == 1:
        raise ValueError("At least one invoice line with description is required.")
    recalc_invoice(invoice)
    invoice.amount_in_words = amount_to_indian_words(invoice.total_amount)
    db.flush()
    return invoice


def issue_invoice(db: Session, invoice: BillingInvoice, *, user_id: int | None = None) -> BillingInvoice:
    issued_now = False
    if invoice.status == "DRAFT":
        invoice.status = "ISSUED"
        invoice.approved_by_user_id = user_id
        invoice.posted_at_utc = datetime.now(timezone.utc)
        update_invoice_payment_totals(invoice)
        invoice.updated_at_utc = datetime.now(timezone.utc)
        issued_now = True
    if issued_now:
        db.flush()
        try:
            send_invoice_issued_email(db, invoice)
        except Exception:
            # Email failure should be recorded in email logs and must not block invoice issue.
            pass
    return invoice


def list_fee_groups(db: Session, *, tenant_id: int, branch_id: int | None = None, partner_id: int | None = None, q: str = ""):
    stmt = select(BillingFeeGroup).options(selectinload(BillingFeeGroup.client), selectinload(BillingFeeGroup.services).selectinload(BillingFeeGroupService.service)).where(BillingFeeGroup.tenant_id == tenant_id)
    if branch_id:
        stmt = stmt.where(BillingFeeGroup.branch_id == branch_id)
    if partner_id:
        stmt = stmt.where(BillingFeeGroup.partner_id == partner_id)
    if q.strip():
        term = f"%{q.strip()}%"
        stmt = stmt.join(Client, Client.id == BillingFeeGroup.client_id).where(or_(BillingFeeGroup.group_code.ilike(term), BillingFeeGroup.group_name.ilike(term), Client.client_name.ilike(term), Client.client_code.ilike(term)))
    return db.execute(stmt.order_by(BillingFeeGroup.group_code.asc())).scalars().unique().all()



def period_label(period_from: date, period_to: date) -> str:
    if period_from.year == period_to.year and period_from.month == period_to.month:
        return period_from.strftime("%B %Y")
    return f"{period_from.isoformat()} to {period_to.isoformat()}"


def list_fee_groups_for_generation(
    db: Session,
    *,
    tenant_id: int,
    branch_id: int | None = None,
    partner_id: int | None = None,
    frequency: str | None = None,
    auto_generate_only: bool = True,
    q: str = "",
):
    stmt = (
        select(BillingFeeGroup)
        .options(
            selectinload(BillingFeeGroup.client),
            selectinload(BillingFeeGroup.services).selectinload(BillingFeeGroupService.service),
        )
        .where(BillingFeeGroup.tenant_id == tenant_id, BillingFeeGroup.is_active.is_(True))
    )
    if branch_id:
        stmt = stmt.where(BillingFeeGroup.branch_id == branch_id)
    if partner_id:
        stmt = stmt.where(BillingFeeGroup.partner_id == partner_id)
    if frequency:
        stmt = stmt.where(BillingFeeGroup.frequency == frequency)
    if auto_generate_only:
        stmt = stmt.where(BillingFeeGroup.auto_generate.is_(True))
    if q.strip():
        term = f"%{q.strip()}%"
        stmt = stmt.join(Client, Client.id == BillingFeeGroup.client_id).where(
            or_(BillingFeeGroup.group_code.ilike(term), BillingFeeGroup.group_name.ilike(term), Client.client_name.ilike(term), Client.client_code.ilike(term))
        )
    return db.execute(stmt.order_by(BillingFeeGroup.group_code.asc())).scalars().unique().all()


def fee_group_already_billed(db: Session, *, tenant_id: int, fee_group_id: int, period_from: date, period_to: date) -> BillingInvoice | None:
    stmt = (
        select(BillingInvoice)
        .join(BillingInvoiceLine, BillingInvoiceLine.invoice_id == BillingInvoice.id)
        .where(
            BillingInvoice.tenant_id == tenant_id,
            BillingInvoice.status != "CANCELLED",
            BillingInvoice.billing_period_from == period_from,
            BillingInvoice.billing_period_to == period_to,
            BillingInvoiceLine.fee_group_id == fee_group_id,
        )
        .order_by(BillingInvoice.id.desc())
    )
    return db.execute(stmt).scalars().first()


def _financial_year_from_period(period_from: date) -> str:
    return _financial_year_label(period_from)


def _billing_engagement_lookup(
    db: Session,
    *,
    tenant_id: int,
    branch_id: int | None,
    client_ids: set[int],
    service_ids: set[int],
    financial_year: str,
) -> dict[tuple[int, int], ClientServiceSubscription]:
    if not client_ids or not service_ids:
        return {}
    stmt = (
        select(ClientServiceSubscription)
        .options(selectinload(ClientServiceSubscription.client), selectinload(ClientServiceSubscription.catalogue))
        .where(
            ClientServiceSubscription.tenant_id == tenant_id,
            ClientServiceSubscription.client_id.in_(client_ids),
            ClientServiceSubscription.service_catalogue_id.in_(service_ids),
            ClientServiceSubscription.financial_year == financial_year,
            ClientServiceSubscription.is_active.is_(True),
        )
    )
    if branch_id:
        stmt = stmt.where(ClientServiceSubscription.branch_id == branch_id)
    rows = db.execute(stmt.order_by(ClientServiceSubscription.id.desc())).scalars().unique().all()
    lookup: dict[tuple[int, int], ClientServiceSubscription] = {}
    for row in rows:
        lookup.setdefault((row.client_id, row.service_catalogue_id), row)
    return lookup


def _annotate_invoice_lines_with_engagements(
    raw_lines: list[dict[str, Any]],
    *,
    client_id: int,
    lookup: dict[tuple[int, int], ClientServiceSubscription],
) -> tuple[list[dict[str, Any]], int | None, list[ClientServiceSubscription]]:
    engagement_ids: set[int] = set()
    linked: list[ClientServiceSubscription] = []
    for line in raw_lines:
        service_id = line.get("service_id")
        if not service_id:
            continue
        subscription = lookup.get((client_id, int(service_id)))
        if not subscription:
            continue
        line["engagement_id"] = subscription.id
        engagement_ids.add(subscription.id)
        linked.append(subscription)
    invoice_engagement_id = next(iter(engagement_ids)) if len(engagement_ids) == 1 else None
    return raw_lines, invoice_engagement_id, linked


def _fee_group_invoice_lines(fee_group: BillingFeeGroup, *, period_from: date, period_to: date) -> list[dict[str, Any]]:
    label = period_label(period_from, period_to)
    if fee_group.billing_mode == "SERVICE_WISE":
        lines: list[dict[str, Any]] = []
        for item in sorted(fee_group.services, key=lambda x: x.sort_order or 100):
            rate = money(item.line_amount)
            if rate <= 0 and item.percentage is not None:
                rate = (money(fee_group.fee_amount) * money(item.percentage) / Decimal("100.00")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
            description = item.line_description or (item.service.service_name if item.service else fee_group.group_name)
            lines.append({
                "description": f"{description} - {label}"[:500],
                "service_id": item.service_id,
                "fee_group_id": fee_group.id,
                "quantity": "1",
                "rate": rate,
                "discount_amount": "0",
                "gst_rate": fee_group.gst_rate,
            })
        if lines:
            return lines
    included = []
    for item in sorted(fee_group.services, key=lambda x: x.sort_order or 100):
        if item.service:
            included.append(item.line_description or item.service.service_name)
    desc = f"{fee_group.group_name} - {label}"
    if included:
        desc = desc + "\nIncluded services: " + ", ".join(included)
    return [{
        "description": desc[:500],
        "service_id": None,
        "fee_group_id": fee_group.id,
        "quantity": "1",
        "rate": fee_group.fee_amount,
        "discount_amount": "0",
        "gst_rate": fee_group.gst_rate,
    }]


def generate_draft_invoices_from_fee_groups(
    db: Session,
    *,
    tenant_id: int,
    branch_id: int | None,
    partner_id: int | None,
    generated_by_user_id: int,
    billing_period_from: date,
    billing_period_to: date,
    frequency: str | None,
    fee_group_ids: list[int],
    skip_duplicates: bool = True,
) -> dict[str, Any]:
    if billing_period_to < billing_period_from:
        raise ValueError("Billing Period To cannot be earlier than Billing Period From.")
    if not fee_group_ids:
        raise ValueError("Select at least one fee structure to generate invoices.")

    stmt = (
        select(BillingFeeGroup)
        .options(selectinload(BillingFeeGroup.client), selectinload(BillingFeeGroup.services).selectinload(BillingFeeGroupService.service))
        .where(BillingFeeGroup.tenant_id == tenant_id, BillingFeeGroup.id.in_(fee_group_ids), BillingFeeGroup.is_active.is_(True))
    )
    if branch_id:
        stmt = stmt.where(BillingFeeGroup.branch_id == branch_id)
    if partner_id:
        stmt = stmt.where(BillingFeeGroup.partner_id == partner_id)
    if frequency:
        stmt = stmt.where(BillingFeeGroup.frequency == frequency)
    fee_groups = db.execute(stmt.order_by(BillingFeeGroup.group_code.asc())).scalars().unique().all()

    batch = BillingInvoiceGenerationBatch(
        tenant_id=tenant_id,
        branch_id=branch_id,
        billing_period_from=billing_period_from,
        billing_period_to=billing_period_to,
        financial_year=billing_financial_year(billing_period_from=billing_period_from, invoice_date=billing_period_to),
        frequency=frequency or None,
        selected_count=len(fee_group_ids),
        generated_by_user_id=generated_by_user_id,
        status="DRAFT_CREATED",
    )
    db.add(batch)
    db.flush()

    created: list[BillingInvoice] = []
    skipped: list[str] = []
    errors: list[str] = []
    found_ids = {g.id for g in fee_groups}
    financial_year = _financial_year_from_period(billing_period_from)
    client_ids = {int(g.client_id) for g in fee_groups if g.client_id}
    service_ids = {int(item.service_id) for g in fee_groups for item in (g.services or []) if item.service_id}
    engagement_lookup = _billing_engagement_lookup(
        db,
        tenant_id=tenant_id,
        branch_id=branch_id,
        client_ids=client_ids,
        service_ids=service_ids,
        financial_year=financial_year,
    )
    for missing_id in sorted(set(fee_group_ids) - found_ids):
        skipped.append(f"Fee structure ID {missing_id} is not available in the active Audit Firm/Branch context.")

    for fee_group in fee_groups:
        try:
            existing = fee_group_already_billed(
                db,
                tenant_id=tenant_id,
                fee_group_id=fee_group.id,
                period_from=billing_period_from,
                period_to=billing_period_to,
            )
            if existing and skip_duplicates:
                skipped.append(f"{fee_group.group_code}: already billed in invoice {existing.invoice_no}.")
                continue
            raw_lines = _fee_group_invoice_lines(fee_group, period_from=billing_period_from, period_to=billing_period_to)
            raw_lines, invoice_engagement_id, linked_subscriptions = _annotate_invoice_lines_with_engagements(
                raw_lines,
                client_id=fee_group.client_id,
                lookup=engagement_lookup,
            )
            linked_note = ""
            if linked_subscriptions:
                linked_labels = []
                for sub in linked_subscriptions:
                    service_name = sub.catalogue.service_name if getattr(sub, "catalogue", None) else f"Service {sub.service_catalogue_id}"
                    linked_labels.append(f"{service_name} / {sub.financial_year}")
                linked_note = " Linked service subscriptions: " + "; ".join(linked_labels[:5]) + "."
            invoice = create_invoice(
                db,
                tenant_id=tenant_id,
                branch_id=fee_group.branch_id or branch_id,
                client_id=fee_group.client_id,
                invoice_date=date.today(),
                due_date=None,
                billing_period_from=billing_period_from,
                billing_period_to=billing_period_to,
                tax_type=fee_group.tax_type if fee_group.tax_type in TAX_TYPES else "CGST_SGST",
                notes=f"Draft generated from fee structure {fee_group.group_code}.{linked_note}",
                terms=None,
                created_by_user_id=generated_by_user_id,
                raw_lines=raw_lines,
                generation_batch_id=batch.id,
                engagement_id=invoice_engagement_id,
                financial_year=financial_year,
            )
            created.append(invoice)
        except Exception as exc:  # keep batch generation resilient per client/package
            errors.append(f"{fee_group.group_code}: {exc}")

    batch.created_invoice_count = len(created)
    batch.skipped_count = len(skipped)
    batch.error_count = len(errors)
    if errors and created:
        batch.status = "PARTIAL"
    elif errors and not created:
        batch.status = "FAILED"
    batch.remarks = "\n".join(skipped + errors) or None
    db.flush()
    return {"batch": batch, "created": created, "skipped": skipped, "errors": errors}

def build_fee_structure_template() -> bytes:
    wb = Workbook()
    ws = wb.active
    ws.title = "Fee_Structure"
    ws.append([
        "Client Code", "Billing Group Code", "Billing Group Name", "Billing Mode", "Frequency", "Fee Amount",
        "GST Rate", "Tax Type", "Effective From", "Effective To", "Auto Generate", "Notes"
    ])
    ws.append(["ABC001", "ABC-GST-MONTHLY", "Monthly GST Compliance", "PACKAGE", "Monthly", 2500, 18, "CGST_SGST", "2026-04-01", "", "Yes", "GSTR-1 and GSTR-3B package"])
    ws2 = wb.create_sheet("Fee_Services")
    ws2.append(["Billing Group Code", "Service Code", "Line Description", "Allocation Type", "Line Amount", "Percentage", "Sort Order"])
    ws2.append(["ABC-GST-MONTHLY", "GSTR1", "GSTR-1 Filing", "Included", 0, "", 1])
    ws2.append(["ABC-GST-MONTHLY", "GSTR3B", "GSTR-3B Filing", "Included", 0, "", 2])
    bio = BytesIO()
    wb.save(bio)
    return bio.getvalue()


def import_fee_structure_excel(db: Session, *, tenant_id: int, branch_id: int | None, created_by_user_id: int, file_bytes: bytes) -> dict[str, Any]:
    wb = load_workbook(BytesIO(file_bytes), data_only=True)
    if "Fee_Structure" not in wb.sheetnames or "Fee_Services" not in wb.sheetnames:
        raise ValueError("Excel must contain Fee_Structure and Fee_Services sheets.")

    clients = {c.client_code.strip().upper(): c for c in db.execute(select(Client).where(Client.tenant_id == tenant_id)).scalars().all()}
    services = {s.service_code.strip().upper(): s for s in db.execute(select(ServiceCatalogue)).scalars().all()}

    ws = wb["Fee_Structure"]
    header = [str(c.value or "").strip() for c in ws[1]]
    rows = []
    errors: list[str] = []
    for idx, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2):
        data = dict(zip(header, row))
        if not any(data.values()):
            continue
        client_code = normalize_code(data.get("Client Code"))
        group_code = normalize_code(data.get("Billing Group Code"))
        if not client_code or client_code not in clients:
            errors.append(f"Fee_Structure row {idx}: Client Code not found: {client_code}")
            continue
        if not group_code:
            errors.append(f"Fee_Structure row {idx}: Billing Group Code is required")
            continue
        mode = str(data.get("Billing Mode") or "PACKAGE").strip().upper().replace(" ", "_")
        if mode not in BILLING_MODES:
            errors.append(f"Fee_Structure row {idx}: Billing Mode must be PACKAGE or SERVICE_WISE")
            continue
        frequency = str(data.get("Frequency") or "Monthly").strip() or "Monthly"
        rows.append({
            "client": clients[client_code],
            "group_code": group_code,
            "group_name": str(data.get("Billing Group Name") or group_code).strip(),
            "billing_mode": mode,
            "frequency": frequency,
            "fee_amount": money(data.get("Fee Amount")),
            "gst_rate": money(data.get("GST Rate") or 18),
            "tax_type": str(data.get("Tax Type") or "CGST_SGST").strip().upper() if str(data.get("Tax Type") or "").strip().upper() in TAX_TYPES else "CGST_SGST",
            "effective_from": parse_date(data.get("Effective From")),
            "effective_to": parse_date(data.get("Effective To")),
            "auto_generate": normalize_yes_no(data.get("Auto Generate")),
            "notes": str(data.get("Notes") or "").strip() or None,
        })

    service_rows_by_group: dict[str, list[dict[str, Any]]] = {}
    ws2 = wb["Fee_Services"]
    header2 = [str(c.value or "").strip() for c in ws2[1]]
    for idx, row in enumerate(ws2.iter_rows(min_row=2, values_only=True), start=2):
        data = dict(zip(header2, row))
        if not any(data.values()):
            continue
        group_code = normalize_code(data.get("Billing Group Code"))
        service_code = normalize_code(data.get("Service Code"))
        if not group_code:
            errors.append(f"Fee_Services row {idx}: Billing Group Code is required")
            continue
        if not service_code or service_code not in services:
            errors.append(f"Fee_Services row {idx}: Service Code not found: {service_code}")
            continue
        service_rows_by_group.setdefault(group_code, []).append({
            "service": services[service_code],
            "line_description": str(data.get("Line Description") or services[service_code].service_name).strip(),
            "allocation_type": str(data.get("Allocation Type") or "Included").strip() or "Included",
            "line_amount": money(data.get("Line Amount")),
            "percentage": money(data.get("Percentage")) if data.get("Percentage") not in (None, "") else None,
            "sort_order": int(data.get("Sort Order") or 100),
        })

    if errors:
        return {"success": False, "created": 0, "updated": 0, "errors": errors}

    created = updated = 0
    for row in rows:
        existing = db.execute(select(BillingFeeGroup).where(BillingFeeGroup.tenant_id == tenant_id, BillingFeeGroup.group_code == row["group_code"])).scalar_one_or_none()
        if existing:
            fee_group = existing
            updated += 1
        else:
            fee_group = BillingFeeGroup(tenant_id=tenant_id, group_code=row["group_code"], created_by_user_id=created_by_user_id)
            db.add(fee_group)
            created += 1
        fee_group.branch_id = branch_id or row["client"].branch_id
        fee_group.client_id = row["client"].id
        fee_group.partner_id = row["client"].partner_id
        fee_group.group_name = row["group_name"]
        fee_group.billing_mode = row["billing_mode"]
        fee_group.frequency = row["frequency"]
        fee_group.fee_amount = row["fee_amount"]
        fee_group.gst_rate = row["gst_rate"]
        fee_group.tax_type = row["tax_type"]
        fee_group.effective_from = row["effective_from"]
        fee_group.effective_to = row["effective_to"]
        fee_group.auto_generate = row["auto_generate"]
        fee_group.notes = row["notes"]
        fee_group.updated_by_user_id = created_by_user_id
        db.flush()

        for old in list(fee_group.services):
            db.delete(old)
        db.flush()
        for service_row in service_rows_by_group.get(row["group_code"], []):
            db.add(BillingFeeGroupService(
                fee_group_id=fee_group.id,
                service_id=service_row["service"].id,
                line_description=service_row["line_description"],
                allocation_type=service_row["allocation_type"],
                line_amount=service_row["line_amount"],
                percentage=service_row["percentage"],
                sort_order=service_row["sort_order"],
            ))

    db.commit()
    return {"success": True, "created": created, "updated": updated, "errors": []}

# ---------------------------------------------------------------------------
# Phase 7R.6 - PayUMoney / PayU redirect integration helpers
# ---------------------------------------------------------------------------
import base64
import hashlib
import hmac
import json
from urllib.parse import urlencode
from urllib.request import Request as UrlRequest, urlopen
from urllib.error import HTTPError, URLError

from app.modules.billing.models import BillingOnlinePaymentTransaction

PAYUMONEY_PROVIDER = "PAYUMONEY"
PAYUMONEY_TEST_URL = "https://test.payu.in/_payment"
PAYUMONEY_PROD_URL = "https://secure.payu.in/_payment"
PAYUMONEY_MODES = ["TEST", "LIVE"]


def payumoney_checkout_url(settings: BillingSettings) -> str:
    return PAYUMONEY_PROD_URL if str(getattr(settings, "payumoney_mode", "TEST")).upper() == "LIVE" else PAYUMONEY_TEST_URL


def is_payumoney_ready(settings: BillingSettings | None) -> bool:
    return bool(
        settings
        and getattr(settings, "payumoney_enabled", False)
        and (getattr(settings, "payumoney_merchant_key", None) or "").strip()
        and (getattr(settings, "payumoney_merchant_salt", None) or "").strip()
    )


def generate_payumoney_hash(*, key: str, txnid: str, amount: str, productinfo: str, firstname: str, email: str, salt: str, udf1: str = "", udf2: str = "", udf3: str = "", udf4: str = "", udf5: str = "") -> str:
    hash_string = f"{key}|{txnid}|{amount}|{productinfo}|{firstname}|{email}|{udf1}|{udf2}|{udf3}|{udf4}|{udf5}||||||{salt}"
    return hashlib.sha512(hash_string.encode("utf-8")).hexdigest().lower()


def verify_payumoney_response_hash(*, response_data: dict[str, Any], salt: str, key: str) -> bool:
    # PayU redirect response hash: salt|status||||||udf5|udf4|udf3|udf2|udf1|email|firstname|productinfo|amount|txnid|key
    received_hash = str(response_data.get("hash") or "").strip().lower()
    if not received_hash:
        return False
    status = str(response_data.get("status") or "")
    txnid = str(response_data.get("txnid") or "")
    amount = str(response_data.get("amount") or "")
    productinfo = str(response_data.get("productinfo") or "")
    firstname = str(response_data.get("firstname") or "")
    email = str(response_data.get("email") or "")
    udf1 = str(response_data.get("udf1") or "")
    udf2 = str(response_data.get("udf2") or "")
    udf3 = str(response_data.get("udf3") or "")
    udf4 = str(response_data.get("udf4") or "")
    udf5 = str(response_data.get("udf5") or "")
    reverse = f"{salt}|{status}||||||{udf5}|{udf4}|{udf3}|{udf2}|{udf1}|{email}|{firstname}|{productinfo}|{amount}|{txnid}|{key}"
    expected = hashlib.sha512(reverse.encode("utf-8")).hexdigest().lower()
    return expected == received_hash


def create_payumoney_transaction(db: Session, *, invoice: BillingInvoice, settings: BillingSettings, base_url: str, client_ip: str | None = None) -> dict[str, Any]:
    if not is_payumoney_ready(settings):
        raise ValueError("PayUMoney is not enabled or merchant credentials are missing in Billing Settings.")
    if invoice.status not in {"ISSUED", "PARTLY_PAID", "OVERDUE"} or money(invoice.balance_amount) <= Decimal("0.00"):
        raise ValueError("Only issued invoices with balance can be paid online.")

    amount = f"{money(invoice.balance_amount):.2f}"
    txnid = f"AF{invoice.tenant_id}I{invoice.id}T{int(datetime.now(timezone.utc).timestamp())}"
    key = (settings.payumoney_merchant_key or "").strip()
    salt = (settings.payumoney_merchant_salt or "").strip()
    productinfo = (settings.payumoney_product_info or f"Invoice {invoice.invoice_no}").strip()[:250]
    firstname = (invoice.client_legal_name or invoice.client_trade_name or getattr(invoice.client, "client_name", None) or "Client").strip()[:120]
    email = (invoice.client_email or getattr(invoice.client, "email", None) or settings.contact_email or "no-reply@example.com").strip()
    phone = (invoice.client_mobile or getattr(invoice.client, "mobile", None) or settings.contact_mobile or "9999999999").strip()
    udf1, udf2, udf3, udf4, udf5 = str(invoice.id), str(invoice.client_id), str(invoice.tenant_id), str(invoice.branch_id or ""), "audit_firm_erp"
    hash_value = generate_payumoney_hash(key=key, txnid=txnid, amount=amount, productinfo=productinfo, firstname=firstname, email=email, salt=salt, udf1=udf1, udf2=udf2, udf3=udf3, udf4=udf4, udf5=udf5)

    transaction = BillingOnlinePaymentTransaction(
        tenant_id=invoice.tenant_id,
        branch_id=invoice.branch_id,
        invoice_id=invoice.id,
        client_id=invoice.client_id,
        provider=PAYUMONEY_PROVIDER,
        mode=(settings.payumoney_mode or "TEST").upper(),
        txnid=txnid,
        amount=money(amount),
        productinfo=productinfo,
        firstname=firstname,
        email=email,
        phone=phone,
        status="INITIATED",
        gateway_status="created",
    )
    db.add(transaction)
    db.flush()

    surl = f"{base_url.rstrip('/')}/client/billing/payumoney/success"
    furl = f"{base_url.rstrip('/')}/client/billing/payumoney/failure"
    payload = {
        "key": key,
        "txnid": txnid,
        "amount": amount,
        "productinfo": productinfo,
        "firstname": firstname,
        "email": email,
        "phone": phone,
        "surl": surl,
        "furl": furl,
        "hash": hash_value,
        "udf1": udf1,
        "udf2": udf2,
        "udf3": udf3,
        "udf4": udf4,
        "udf5": udf5,
    }
    if getattr(settings, "payumoney_merchant_id", None):
        payload["merchant_id"] = settings.payumoney_merchant_id
    return {"transaction": transaction, "payload": payload, "checkout_url": payumoney_checkout_url(settings)}


def get_online_transaction_by_txnid(db: Session, *, txnid: str) -> BillingOnlinePaymentTransaction | None:
    return db.execute(
        select(BillingOnlinePaymentTransaction)
        .options(selectinload(BillingOnlinePaymentTransaction.invoice).selectinload(BillingInvoice.payments))
        .where(BillingOnlinePaymentTransaction.txnid == txnid)
    ).scalars().unique().one_or_none()


def process_payumoney_response(db: Session, *, response_data: dict[str, Any]) -> BillingOnlinePaymentTransaction | None:
    txnid = str(response_data.get("txnid") or "").strip()
    if not txnid:
        return None
    transaction = get_online_transaction_by_txnid(db, txnid=txnid)
    if not transaction:
        return None
    invoice = transaction.invoice
    settings = get_effective_billing_settings(db, tenant_id=invoice.tenant_id, branch_id=invoice.branch_id)
    hash_ok = verify_payumoney_response_hash(response_data=response_data, salt=(settings.payumoney_merchant_salt or ""), key=(settings.payumoney_merchant_key or ""))
    gateway_status = str(response_data.get("status") or "").lower()
    transaction.gateway_status = gateway_status or None
    transaction.response_hash = str(response_data.get("hash") or "") or None
    transaction.payu_payment_id = str(response_data.get("payuMoneyId") or response_data.get("payu_money_id") or "") or None
    transaction.mihpayid = str(response_data.get("mihpayid") or "") or None
    transaction.bank_ref_num = str(response_data.get("bank_ref_num") or response_data.get("bank_ref_no") or "") or None
    transaction.raw_response = json.dumps({k: str(v) for k, v in response_data.items()}, ensure_ascii=False)
    transaction.updated_at_utc = datetime.now(timezone.utc)

    if not hash_ok:
        transaction.status = "HASH_FAILED"
        return transaction
    if gateway_status == "success":
        transaction.status = "SUCCESS"
        transaction.completed_at_utc = datetime.now(timezone.utc)
        if not transaction.receipt_payment_id:
            payment = record_invoice_payment(
                db,
                invoice=invoice,
                payment_date=date.today(),
                amount_received=money(response_data.get("amount") or transaction.amount),
                tds_deducted=Decimal("0.00"),
                bank_charges=Decimal("0.00"),
                mode="ONLINE",
                reference_no=transaction.bank_ref_num or transaction.mihpayid or transaction.txnid,
                remarks=f"Online payment received through PayUMoney. Txn ID: {transaction.txnid}",
                created_by_user_id=None,
                payment_gateway=PAYUMONEY_PROVIDER,
                gateway_transaction_id=transaction.mihpayid or transaction.payu_payment_id or transaction.txnid,
            )
            transaction.receipt_payment_id = payment.id
    else:
        transaction.status = "FAILED"
    return transaction


# ---------------------------------------------------------------------------
# Phase 7R.6A - Cashfree Payment Gateway Integration helpers
# ---------------------------------------------------------------------------
CASHFREE_PROVIDER = "CASHFREE"
CASHFREE_API_VERSION_DEFAULT = "2023-08-01"
CASHFREE_TEST_BASE_URL = "https://sandbox.cashfree.com/pg"
CASHFREE_PROD_BASE_URL = "https://api.cashfree.com/pg"
CASHFREE_MODES = ["TEST", "LIVE"]


def cashfree_base_url(settings: BillingSettings) -> str:
    return CASHFREE_PROD_BASE_URL if str(getattr(settings, "cashfree_mode", "TEST")).upper() == "LIVE" else CASHFREE_TEST_BASE_URL


def is_cashfree_ready(settings: BillingSettings | None) -> bool:
    return bool(
        settings
        and getattr(settings, "cashfree_enabled", False)
        and (getattr(settings, "cashfree_client_id", None) or "").strip()
        and (getattr(settings, "cashfree_client_secret", None) or "").strip()
    )


def _cashfree_headers(settings: BillingSettings) -> dict[str, str]:
    return {
        "Content-Type": "application/json",
        "Accept": "application/json",
        "x-api-version": (getattr(settings, "cashfree_api_version", None) or CASHFREE_API_VERSION_DEFAULT).strip() or CASHFREE_API_VERSION_DEFAULT,
        "x-client-id": (settings.cashfree_client_id or "").strip(),
        "x-client-secret": (settings.cashfree_client_secret or "").strip(),
    }


def _cashfree_api_request(settings: BillingSettings, *, method: str, path: str, payload: dict[str, Any] | None = None) -> dict[str, Any]:
    url = cashfree_base_url(settings).rstrip("/") + path
    body = json.dumps(payload or {}).encode("utf-8") if payload is not None else None
    req = UrlRequest(url, data=body, headers=_cashfree_headers(settings), method=method.upper())
    try:
        with urlopen(req, timeout=30) as response:
            raw = response.read().decode("utf-8")
            return json.loads(raw or "{}")
    except HTTPError as exc:
        raw = exc.read().decode("utf-8", errors="replace")
        raise ValueError(f"Cashfree API error {exc.code}: {raw}") from exc
    except URLError as exc:
        raise ValueError(f"Cashfree API connection error: {exc.reason}") from exc


def create_cashfree_transaction(db: Session, *, invoice: BillingInvoice, settings: BillingSettings, base_url: str, client_ip: str | None = None) -> dict[str, Any]:
    if not is_cashfree_ready(settings):
        raise ValueError("Cashfree is not enabled or client credentials are missing in Billing Settings.")
    if invoice.status not in {"ISSUED", "PARTLY_PAID", "OVERDUE"} or money(invoice.balance_amount) <= Decimal("0.00"):
        raise ValueError("Only issued invoices with balance can be paid online.")

    amount = money(invoice.balance_amount)
    order_id = f"AF{invoice.tenant_id}I{invoice.id}C{int(datetime.now(timezone.utc).timestamp())}"
    customer_name = (invoice.client_legal_name or invoice.client_trade_name or getattr(invoice.client, "client_name", None) or "Client").strip()[:120]
    customer_email = (invoice.client_email or getattr(invoice.client, "email", None) or settings.contact_email or "no-reply@example.com").strip()
    customer_phone = (invoice.client_mobile or getattr(invoice.client, "mobile", None) or settings.contact_mobile or "9999999999").strip()
    note = (settings.cashfree_order_note or f"Invoice {invoice.invoice_no}").strip()[:250]

    return_url = f"{base_url.rstrip('/')}/client/billing/cashfree/return?order_id={{order_id}}"
    notify_url = f"{base_url.rstrip('/')}/client/billing/cashfree/webhook"
    payload = {
        "order_id": order_id,
        "order_amount": float(amount),
        "order_currency": "INR",
        "customer_details": {
            "customer_id": str(invoice.client_id),
            "customer_name": customer_name,
            "customer_email": customer_email,
            "customer_phone": customer_phone,
        },
        "order_meta": {
            "return_url": return_url,
            "notify_url": notify_url,
        },
        "order_note": note,
        "order_tags": {
            "tenant_id": str(invoice.tenant_id),
            "branch_id": str(invoice.branch_id or ""),
            "invoice_id": str(invoice.id),
            "invoice_no": str(invoice.invoice_no),
            "source": "audit_firm_erp",
        },
    }
    response = _cashfree_api_request(settings, method="POST", path="/orders", payload=payload)
    payment_session_id = str(response.get("payment_session_id") or "").strip()
    if not payment_session_id:
        raise ValueError(f"Cashfree order created without payment_session_id: {response}")

    transaction = BillingOnlinePaymentTransaction(
        tenant_id=invoice.tenant_id,
        branch_id=invoice.branch_id,
        invoice_id=invoice.id,
        client_id=invoice.client_id,
        provider=CASHFREE_PROVIDER,
        mode=(settings.cashfree_mode or "TEST").upper(),
        txnid=order_id,
        amount=amount,
        productinfo=note,
        firstname=customer_name,
        email=customer_email,
        phone=customer_phone,
        status="INITIATED",
        gateway_status=str(response.get("order_status") or "ACTIVE"),
        cashfree_order_id=order_id,
        cashfree_cf_order_id=str(response.get("cf_order_id") or "") or None,
        cashfree_payment_session_id=payment_session_id,
        raw_response=json.dumps(response, ensure_ascii=False, default=str),
    )
    db.add(transaction)
    db.flush()
    return {"transaction": transaction, "payment_session_id": payment_session_id, "order_response": response}


def get_cashfree_transaction_by_order_id(db: Session, *, order_id: str) -> BillingOnlinePaymentTransaction | None:
    return db.execute(
        select(BillingOnlinePaymentTransaction)
        .options(selectinload(BillingOnlinePaymentTransaction.invoice).selectinload(BillingInvoice.payments))
        .where(
            BillingOnlinePaymentTransaction.provider == CASHFREE_PROVIDER,
            BillingOnlinePaymentTransaction.txnid == order_id,
        )
    ).scalars().unique().one_or_none()


def fetch_cashfree_order_status(settings: BillingSettings, *, order_id: str) -> dict[str, Any]:
    return _cashfree_api_request(settings, method="GET", path=f"/orders/{order_id}")


def _mark_cashfree_success(db: Session, transaction: BillingOnlinePaymentTransaction, *, amount: Any, reference_no: str | None, raw_payload: dict[str, Any] | None = None) -> BillingOnlinePaymentTransaction:
    invoice = transaction.invoice
    transaction.status = "SUCCESS"
    transaction.gateway_status = "PAID"
    transaction.completed_at_utc = datetime.now(timezone.utc)
    transaction.updated_at_utc = datetime.now(timezone.utc)
    if raw_payload is not None:
        transaction.raw_response = json.dumps(raw_payload, ensure_ascii=False, default=str)
    if not transaction.receipt_payment_id:
        payment = record_invoice_payment(
            db,
            invoice=invoice,
            payment_date=date.today(),
            amount_received=money(amount or transaction.amount),
            tds_deducted=Decimal("0.00"),
            bank_charges=Decimal("0.00"),
            mode="ONLINE",
            reference_no=reference_no or transaction.cashfree_payment_id or transaction.txnid,
            remarks=f"Online payment received through Cashfree. Order ID: {transaction.txnid}",
            created_by_user_id=None,
            payment_gateway=CASHFREE_PROVIDER,
            gateway_transaction_id=reference_no or transaction.cashfree_payment_id or transaction.txnid,
        )
        transaction.receipt_payment_id = payment.id
    return transaction


def process_cashfree_return(db: Session, *, order_id: str) -> BillingOnlinePaymentTransaction | None:
    transaction = get_cashfree_transaction_by_order_id(db, order_id=order_id)
    if not transaction:
        return None
    invoice = transaction.invoice
    settings = get_effective_billing_settings(db, tenant_id=invoice.tenant_id, branch_id=invoice.branch_id)
    order = fetch_cashfree_order_status(settings, order_id=order_id)
    transaction.gateway_status = str(order.get("order_status") or "") or transaction.gateway_status
    transaction.cashfree_cf_order_id = str(order.get("cf_order_id") or transaction.cashfree_cf_order_id or "") or None
    transaction.raw_response = json.dumps(order, ensure_ascii=False, default=str)
    transaction.updated_at_utc = datetime.now(timezone.utc)
    if str(order.get("order_status") or "").upper() == "PAID":
        return _mark_cashfree_success(db, transaction, amount=order.get("order_amount") or transaction.amount, reference_no=str(order.get("cf_order_id") or order_id), raw_payload=order)
    if str(order.get("order_status") or "").upper() in {"EXPIRED", "TERMINATED", "CANCELLED"}:
        transaction.status = "FAILED"
    return transaction


def verify_cashfree_webhook_signature(*, raw_body: bytes, timestamp: str, signature: str, secret: str) -> bool:
    if not raw_body or not timestamp or not signature or not secret:
        return False
    signed_payload = timestamp.encode("utf-8") + raw_body
    digest = hmac.new(secret.encode("utf-8"), signed_payload, hashlib.sha256).digest()
    expected = base64.b64encode(digest).decode("utf-8")
    return hmac.compare_digest(expected, signature)


def process_cashfree_webhook(db: Session, *, raw_body: bytes, headers: dict[str, str]) -> BillingOnlinePaymentTransaction | None:
    payload = json.loads(raw_body.decode("utf-8") or "{}")
    data = payload.get("data") or payload
    order = data.get("order") or data
    payment = data.get("payment") or {}
    order_id = str(order.get("order_id") or data.get("order_id") or "").strip()
    if not order_id:
        return None
    transaction = get_cashfree_transaction_by_order_id(db, order_id=order_id)
    if not transaction:
        return None
    invoice = transaction.invoice
    settings = get_effective_billing_settings(db, tenant_id=invoice.tenant_id, branch_id=invoice.branch_id)
    timestamp = headers.get("x-webhook-timestamp") or headers.get("X-Webhook-Timestamp") or ""
    signature = headers.get("x-webhook-signature") or headers.get("X-Webhook-Signature") or ""
    if not verify_cashfree_webhook_signature(raw_body=raw_body, timestamp=timestamp, signature=signature, secret=(settings.cashfree_client_secret or "")):
        transaction.status = "HASH_FAILED"
        transaction.gateway_status = "WEBHOOK_SIGNATURE_FAILED"
        transaction.raw_response = raw_body.decode("utf-8", errors="replace")
        transaction.updated_at_utc = datetime.now(timezone.utc)
        return transaction

    event_id = str(payload.get("event_id") or payload.get("cf_event_id") or "") or None
    if event_id and transaction.webhook_event_id == event_id and transaction.status == "SUCCESS":
        return transaction
    transaction.webhook_event_id = event_id
    transaction.cashfree_payment_id = str(payment.get("cf_payment_id") or payment.get("payment_id") or "") or transaction.cashfree_payment_id
    transaction.gateway_status = str(payment.get("payment_status") or order.get("order_status") or payload.get("type") or "") or None
    transaction.raw_response = raw_body.decode("utf-8", errors="replace")
    transaction.updated_at_utc = datetime.now(timezone.utc)

    status_text = (transaction.gateway_status or "").upper()
    if "SUCCESS" in status_text or status_text == "PAID" or str(order.get("order_status") or "").upper() == "PAID":
        amount = payment.get("payment_amount") or order.get("order_amount") or transaction.amount
        reference = transaction.cashfree_payment_id or str(payment.get("bank_reference") or order.get("cf_order_id") or order_id)
        return _mark_cashfree_success(db, transaction, amount=amount, reference_no=reference, raw_payload=payload)
    if "FAILED" in status_text or "CANCELLED" in status_text or "EXPIRED" in status_text:
        transaction.status = "FAILED"
    return transaction
