arrr-erp/app/modules/accounting/gstr2b_ui.py

from __future__ import annotations

import json
from urllib.parse import urlencode

from fastapi import APIRouter, File, Form, Request, UploadFile
from fastapi.responses import RedirectResponse
from sqlalchemy import select

from app.core.db.common import CommonSessionLocal
from app.core.security.csrf import get_or_create_csrf_token, validate_csrf
from app.core.templating import templates
from app.modules.accounting.gstr2b_models import AccountingGSTR2BImportBatch, AccountingGSTR2BPurchase
from app.modules.accounting.gstr2b_service import (
    analyze_batch,
    batches_for_client,
    import_gstr2b,
    nature_lookup,
    purchases_for_client,
    review_purchase,
)
from app.modules.accounting.historical_learning_service import active_natures, ledger_mappings
from app.modules.accounting.ledger_learning_service import available_tally_guids
from app.modules.accounting.ui import _find_visible_client, _require_partner, _visible_clients
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles

router = APIRouter(prefix="/tools/accounting/gstr2b", tags=["accounting-gstr2b-ui"])


def _redirect(client_id: int, *, batch_id: int | None = None, message: str = "", error: str = ""):
    params = {"client_id": client_id}
    if batch_id:
        params["batch_id"] = batch_id
    if message:
        params["message"] = message
    if error:
        params["error"] = error[:180]
    return RedirectResponse(url="/tools/accounting/gstr2b?" + urlencode(params), status_code=303)


@router.get("")
def gstr2b_page(
    request: Request,
    client_id: int | None = None,
    batch_id: int | None = None,
    message: str = "",
    error: str = "",
):
    db = CommonSessionLocal()
    try:
        user, response = _require_partner(request, db, "accounting.learning.view")
        if response:
            return response
        clients, scope = _visible_clients(db, request, user)
        selected = next((c for c in clients if client_id and int(c.id) == int(client_id)), None)

        batches = []
        purchases = []
        mappings = []
        companies = []
        natures = []
        nature_by_id = {}
        selected_batch = None

        if selected:
            batches = batches_for_client(db, scope.tenant_id, selected.id)
            selected_batch = next((b for b in batches if batch_id and b.id == batch_id), None)
            if not selected_batch and batches:
                selected_batch = batches[0]
            purchases = purchases_for_client(
                db, scope.tenant_id, selected.id,
                batch_id=selected_batch.id if selected_batch else None
            )
            mappings = ledger_mappings(
                db, scope.tenant_id, selected.id,
                selected_batch.tally_guid if selected_batch else ""
            )
            companies = available_tally_guids(db, scope.tenant_id, selected.id)
            natures = active_natures(db, scope.tenant_id)
            ids = []
            for row in purchases:
                ids.extend([row.suggested_nature_id, row.final_nature_id])
            nature_by_id = nature_lookup(db, ids)

        explanations = {}
        for row in purchases:
            try:
                explanations[row.id] = json.loads(row.suggestion_explanation_json or "[]")
            except Exception:
                explanations[row.id] = []

        return templates.TemplateResponse("modules/accounting/templates/accounting/gstr2b.html", {
            "request": request,
            "current_user": user,
            "current_user_roles": get_user_roles(db, user.id),
            "current_user_permissions": get_user_permissions(db, user.id),
            "csrf_token": get_or_create_csrf_token(request),
            "title": "GSTR-2B Purchase Intelligence",
            "clients": clients,
            "selected_client": selected,
            "batches": batches,
            "selected_batch": selected_batch,
            "purchases": purchases,
            "mappings": mappings,
            "tally_companies": companies,
            "natures": natures,
            "nature_by_id": nature_by_id,
            "explanations": explanations,
            "message": message,
            "error": error,
        })
    finally:
        db.close()


@router.post("/upload")
async def upload_gstr2b(
    request: Request,
    client_id: int = Form(...),
    tally_guid: str = Form(""),
    return_period: str = Form(""),
    upload: UploadFile = File(...),
    csrf_token: str = Form(...),
):
    validate_csrf(request, csrf_token)
    db = CommonSessionLocal()
    try:
        user, response = _require_partner(request, db, "accounting.learning.manage")
        if response:
            return response
        client, _, scope = _find_visible_client(db, request, user, client_id)
        if not client:
            from app.core.http_responses import ui_access_denied
            return ui_access_denied()
        content = await upload.read()
        batch, duplicate_file = import_gstr2b(
            db,
            tenant_id=scope.tenant_id,
            client_id=client.id,
            tally_guid=tally_guid,
            return_period=return_period,
            filename=upload.filename or "gstr2b.xlsx",
            content=content,
            user_id=user.id,
        )
        if duplicate_file:
            msg = f"This exact GSTR-2B file was already imported as batch #{batch.id}."
        else:
            msg = (
                f"Imported {batch.rows_imported} purchase document(s); "
                f"{batch.rows_skipped_duplicate} duplicate(s) and "
                f"{batch.rows_skipped_invalid} invalid row(s) skipped."
            )
        return _redirect(client.id, batch_id=batch.id, message=msg)
    except Exception as exc:
        db.rollback()
        return _redirect(client_id, error=str(exc))
    finally:
        db.close()


@router.post("/batch/{batch_id}/analyze")
def analyze(
    request: Request,
    batch_id: int,
    client_id: int = Form(...),
    csrf_token: str = Form(...),
):
    validate_csrf(request, csrf_token)
    db = CommonSessionLocal()
    try:
        user, response = _require_partner(request, db, "accounting.learning.manage")
        if response:
            return response
        client, _, scope = _find_visible_client(db, request, user, client_id)
        if not client:
            from app.core.http_responses import ui_access_denied
            return ui_access_denied()
        batch = db.execute(select(AccountingGSTR2BImportBatch).where(
            AccountingGSTR2BImportBatch.id == batch_id,
            AccountingGSTR2BImportBatch.tenant_id == scope.tenant_id,
            AccountingGSTR2BImportBatch.client_id == client.id,
        )).scalar_one_or_none()
        if not batch:
            return _redirect(client_id, error="GSTR-2B import batch was not found.")
        count = analyze_batch(db, batch)
        return _redirect(client_id, batch_id=batch.id, message=f"Analyzed {count} purchase document(s).")
    except Exception as exc:
        db.rollback()
        return _redirect(client_id, batch_id=batch_id, error=str(exc))
    finally:
        db.close()


@router.post("/purchase/{purchase_id}/review")
def review(
    request: Request,
    purchase_id: int,
    client_id: int = Form(...),
    final_nature_id: int = Form(...),
    final_ledger_name: str = Form(""),
    csrf_token: str = Form(...),
):
    validate_csrf(request, csrf_token)
    db = CommonSessionLocal()
    try:
        user, response = _require_partner(request, db, "accounting.learning.manage")
        if response:
            return response
        client, _, scope = _find_visible_client(db, request, user, client_id)
        if not client:
            from app.core.http_responses import ui_access_denied
            return ui_access_denied()
        row = db.execute(select(AccountingGSTR2BPurchase).where(
            AccountingGSTR2BPurchase.id == purchase_id,
            AccountingGSTR2BPurchase.tenant_id == scope.tenant_id,
            AccountingGSTR2BPurchase.client_id == client.id,
        )).scalar_one_or_none()
        if not row:
            return _redirect(client_id, error="GSTR-2B purchase record was not found.")

        # If a ledger is selected, ensure it is one of the Phase 5 mappings for this nature/company.
        if final_ledger_name.strip():
            allowed = ledger_mappings(db, scope.tenant_id, client.id, row.tally_guid)
            valid = any(
                m.ledger_name == final_ledger_name.strip() and int(m.nature_id) == int(final_nature_id)
                for m in allowed
            )
            if not valid:
                return _redirect(
                    client_id, batch_id=row.batch_id,
                    error="Selected Tally ledger is not mapped to the selected accounting nature for this client/company."
                )

        review_purchase(
            db,
            row=row,
            final_nature_id=final_nature_id,
            final_ledger_name=final_ledger_name,
            user_id=user.id,
        )
        return _redirect(client_id, batch_id=row.batch_id, message=f"Invoice {row.invoice_number} reviewed and learned.")
    except Exception as exc:
        db.rollback()
        return _redirect(client_id, error=str(exc))
    finally:
        db.close()