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