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

from __future__ import annotations

import json
from datetime import datetime, timezone

from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError

from app.modules.accounting.gstr2b_models import AccountingGSTR2BImportBatch, AccountingGSTR2BPurchase
from app.modules.accounting.gstr2b_parser import file_sha256, parse_gstr2b
from app.modules.accounting.ledger_learning_service import rank_suggestions, record_review
from app.modules.accounting.taxonomy_models import AccountingNature


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


def import_gstr2b(
    db, *,
    tenant_id: int,
    client_id: int,
    tally_guid: str,
    return_period: str,
    filename: str,
    content: bytes,
    user_id: int,
):
    digest = file_sha256(content)
    existing_batch = db.execute(select(AccountingGSTR2BImportBatch).where(
        AccountingGSTR2BImportBatch.tenant_id == tenant_id,
        AccountingGSTR2BImportBatch.client_id == client_id,
        AccountingGSTR2BImportBatch.file_sha256 == digest,
    )).scalar_one_or_none()
    if existing_batch:
        return existing_batch, True

    parsed = parse_gstr2b(content, filename)
    batch = AccountingGSTR2BImportBatch(
        tenant_id=tenant_id,
        client_id=client_id,
        tally_guid=(tally_guid or "").strip(),
        original_filename=(filename or "gstr2b.xlsx")[:260],
        file_sha256=digest,
        return_period=(return_period or "").strip()[:20],
        source_kind="gstr2b_upload",
        status="importing",
        rows_read=len(parsed),
        imported_by_user_id=user_id,
    )
    db.add(batch)
    db.flush()

    imported = duplicate = invalid = 0
    for row in parsed:
        if not row.get("invoice_number") or not row.get("invoice_date"):
            invalid += 1
            continue

        exists = db.execute(select(AccountingGSTR2BPurchase.id).where(
            AccountingGSTR2BPurchase.tenant_id == tenant_id,
            AccountingGSTR2BPurchase.client_id == client_id,
            AccountingGSTR2BPurchase.supplier_gstin == row.get("supplier_gstin", ""),
            AccountingGSTR2BPurchase.invoice_number == row["invoice_number"],
            AccountingGSTR2BPurchase.invoice_date == row["invoice_date"],
            AccountingGSTR2BPurchase.document_type == row["document_type"],
        )).scalar_one_or_none()
        if exists:
            duplicate += 1
            continue

        db.add(AccountingGSTR2BPurchase(
            tenant_id=tenant_id,
            client_id=client_id,
            batch_id=batch.id,
            tally_guid=(tally_guid or "").strip(),
            return_period=(return_period or "").strip()[:20],
            **row,
        ))
        imported += 1

    batch.rows_imported = imported
    batch.rows_skipped_duplicate = duplicate
    batch.rows_skipped_invalid = invalid
    batch.status = "imported"
    batch.completed_at_utc = _utcnow()
    db.commit()
    db.refresh(batch)
    return batch, False


def batches_for_client(db, tenant_id: int, client_id: int, limit: int = 20):
    return list(db.execute(select(AccountingGSTR2BImportBatch).where(
        AccountingGSTR2BImportBatch.tenant_id == tenant_id,
        AccountingGSTR2BImportBatch.client_id == client_id,
    ).order_by(AccountingGSTR2BImportBatch.id.desc()).limit(limit)).scalars().all())


def purchases_for_client(db, tenant_id: int, client_id: int, *, batch_id: int | None = None, limit: int = 300):
    stmt = select(AccountingGSTR2BPurchase).where(
        AccountingGSTR2BPurchase.tenant_id == tenant_id,
        AccountingGSTR2BPurchase.client_id == client_id,
    )
    if batch_id:
        stmt = stmt.where(AccountingGSTR2BPurchase.batch_id == batch_id)
    return list(db.execute(stmt.order_by(
        AccountingGSTR2BPurchase.invoice_date.desc(),
        AccountingGSTR2BPurchase.id.desc(),
    ).limit(limit)).scalars().all())


def nature_lookup(db, ids):
    ids = {int(x) for x in ids if x}
    if not ids:
        return {}
    return {row.id: row for row in db.execute(
        select(AccountingNature).where(AccountingNature.id.in_(ids))
    ).scalars().all()}


def analyze_purchase(db, row: AccountingGSTR2BPurchase):
    description_parts = [
        row.description_text or "",
        f"Invoice type {row.invoice_type}" if row.invoice_type else "",
        f"Place of supply {row.place_of_supply}" if row.place_of_supply else "",
    ]
    suggestions = rank_suggestions(
        db,
        tenant_id=row.tenant_id,
        client_id=row.client_id,
        tally_guid=row.tally_guid,
        supplier_name=row.supplier_name,
        supplier_gstin=row.supplier_gstin,
        hsn_code=row.hsn_code,
        description=" | ".join(x for x in description_parts if x),
        amount=row.taxable_value,
    )
    if suggestions:
        top = suggestions[0]
        row.suggested_nature_id = top["nature"].id
        row.suggested_ledger_name = top.get("suggested_ledger") or ""
        row.suggested_confidence = int(top.get("confidence") or 0)
        row.suggestion_explanation_json = json.dumps(top.get("reasons") or [], ensure_ascii=False)
        row.review_status = "suggested"
    else:
        row.suggested_nature_id = None
        row.suggested_ledger_name = ""
        row.suggested_confidence = 0
        row.suggestion_explanation_json = json.dumps(
            ["No reliable classification evidence is available yet."], ensure_ascii=False
        )
        row.review_status = "review_required"
    return suggestions


def analyze_batch(db, batch: AccountingGSTR2BImportBatch):
    rows = list(db.execute(select(AccountingGSTR2BPurchase).where(
        AccountingGSTR2BPurchase.batch_id == batch.id
    )).scalars().all())
    analyzed = 0
    for row in rows:
        analyze_purchase(db, row)
        analyzed += 1
    batch.analyzed_rows = analyzed
    batch.analyzed_at_utc = _utcnow()
    batch.status = "analyzed"
    db.commit()
    return analyzed


def review_purchase(
    db, *,
    row: AccountingGSTR2BPurchase,
    final_nature_id: int,
    final_ledger_name: str,
    user_id: int,
):
    try:
        explanation = json.loads(row.suggestion_explanation_json or "[]")
    except Exception:
        explanation = []

    record_review(
        db,
        tenant_id=row.tenant_id,
        client_id=row.client_id,
        tally_guid=row.tally_guid,
        supplier_name=row.supplier_name,
        supplier_gstin=row.supplier_gstin,
        hsn_code=row.hsn_code,
        description=row.description_text or "",
        amount=row.taxable_value,
        suggested_nature_id=row.suggested_nature_id,
        suggested_ledger_name=row.suggested_ledger_name,
        suggested_confidence=row.suggested_confidence,
        final_nature_id=final_nature_id,
        final_ledger_name=final_ledger_name,
        user_id=user_id,
        explanation=explanation,
    )

    # record_review commits its learning event; refresh the purchase in this session.
    row.final_nature_id = final_nature_id
    row.final_ledger_name = (final_ledger_name or "").strip()
    row.review_status = "reviewed"
    row.reviewed_by_user_id = user_id
    row.reviewed_at_utc = _utcnow()
    db.add(row)

    batch = db.get(AccountingGSTR2BImportBatch, row.batch_id)
    if batch:
        reviewed_count = db.execute(select(func.count(AccountingGSTR2BPurchase.id)).where(
            AccountingGSTR2BPurchase.batch_id == batch.id,
            AccountingGSTR2BPurchase.review_status == "reviewed",
        )).scalar_one()
        # Include the current row if the database count was evaluated before flush.
        if row.review_status == "reviewed":
            db.flush()
            reviewed_count = db.execute(select(func.count(AccountingGSTR2BPurchase.id)).where(
                AccountingGSTR2BPurchase.batch_id == batch.id,
                AccountingGSTR2BPurchase.review_status == "reviewed",
            )).scalar_one()
        batch.reviewed_rows = int(reviewed_count or 0)
    db.commit()
    db.refresh(row)
    return row