""" @router AdminInvoiceRouter (Backend/app/api/v1/routers/AdminInvoiceRouter.py) @purpose CRM Admin endpoint controllers to query, generate, and print standard PDF and 80mm thermal roll GST invoices on-the-fly. """ from fastapi import APIRouter, Depends, HTTPException, status, Response from sqlalchemy.orm import Session from pydantic import BaseModel from typing import List, Optional from datetime import datetime import ulid import io import zipfile from app.core.database.db_session import get_db from app.core.permissions.RoleChecker import get_current_user from app.models.UserModel import User from app.models.InvoiceModel import Invoice from app.models.OrderModel import Order, OrderItem from app.models.EcomCustomerModel import EcomCustomer from app.core.invoice_generator import generate_invoice_pdf, generate_thermal_invoice_pdf router = APIRouter(prefix="/api/v1/admin/invoices", tags=["Admin CRM Customer Invoices"]) # --- Schemas --- class AdminInvoiceResponse(BaseModel): invoice_id: str invoice_no: str order_id: Optional[str] = None customer_id: str customer_email: Optional[str] = None subtotal: float discount_amount: float cgst: float sgst: float igst: float total_amount: float status: str pdf_url: Optional[str] = None created_at: datetime # --- Endpoints --- @router.get("", response_model=List[AdminInvoiceResponse]) def list_all_invoices( current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """ List all generated GST invoices in the system. """ invoices = db.query(Invoice).order_by(Invoice.created_at.desc()).all() result = [] for inv in invoices: cust_email = None customer = db.query(EcomCustomer).filter(EcomCustomer.customer_id == inv.customer_id).first() if customer: cust_email = customer.email pdf_url = f"/api/v1/admin/invoices/{inv.invoice_id}/download" result.append( AdminInvoiceResponse( invoice_id=inv.invoice_id, invoice_no=inv.invoice_no, order_id=inv.order_id, customer_id=inv.customer_id, customer_email=cust_email, subtotal=float(inv.subtotal), discount_amount=float(inv.discount_amount), cgst=float(inv.cgst), sgst=float(inv.sgst), igst=float(inv.igst), total_amount=float(inv.total_amount), status=inv.status, pdf_url=pdf_url, created_at=inv.created_at ) ) return result @router.post("/generate/{order_id}", response_model=AdminInvoiceResponse) def generate_invoice_for_order( order_id: str, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """ Generate a new GST invoice for a confirmed order. """ # Check if invoice already exists existing = db.query(Invoice).filter(Invoice.order_id == order_id).first() if existing: cust_email = None customer = db.query(EcomCustomer).filter(EcomCustomer.customer_id == existing.customer_id).first() if customer: cust_email = customer.email return AdminInvoiceResponse( invoice_id=existing.invoice_id, invoice_no=existing.invoice_no, order_id=existing.order_id, customer_id=existing.customer_id, customer_email=cust_email, subtotal=float(existing.subtotal), discount_amount=float(existing.discount_amount), cgst=float(existing.cgst), sgst=float(existing.sgst), igst=float(existing.igst), total_amount=float(existing.total_amount), status=existing.status, pdf_url=f"/api/v1/admin/invoices/{existing.invoice_id}/download", created_at=existing.created_at ) # Fetch order details order = db.query(Order).filter(Order.order_id == order_id).first() if not order: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Order not found" ) # GST calculation matching order details subtotal = float(order.total_amount) discount = float(order.discount_amount) tax_amount = float(order.tax_amount) cgst = round(tax_amount / 2, 2) sgst = round(tax_amount / 2, 2) igst = 0.0 total = float(order.final_amount) # Allocate a 16-char GST invoice number # Format: C1P2-26-SEQ today = datetime.now() fy = today.strftime("%y") # financial year seq, e.g. 26 last_inv = db.query(Invoice).order_by(Invoice.created_at.desc()).first() if not last_inv: seq = 1 else: try: seq = int(last_inv.invoice_no.split("-")[-1]) + 1 except Exception: seq = 1 invoice_no = f"C1P1-{fy}-{seq:06d}" invoice = Invoice( invoice_id=str(ulid.ULID()), invoice_no=invoice_no, order_id=order.order_id, customer_id=order.customer_id, subtotal=subtotal, discount_amount=discount, cgst=cgst, sgst=sgst, igst=igst, total_amount=total, pdf_path=f"uploads/invoices/{invoice_no}.pdf" # virtual pointer path ) db.add(invoice) db.commit() db.refresh(invoice) cust_email = None if order.customer: cust_email = order.customer.email return AdminInvoiceResponse( invoice_id=invoice.invoice_id, invoice_no=invoice.invoice_no, order_id=invoice.order_id, customer_id=invoice.customer_id, customer_email=cust_email, subtotal=float(invoice.subtotal), discount_amount=float(invoice.discount_amount), cgst=float(invoice.cgst), sgst=float(invoice.sgst), igst=float(invoice.igst), total_amount=float(invoice.total_amount), status=invoice.status, pdf_url=f"/api/v1/admin/invoices/{invoice.invoice_id}/download", created_at=invoice.created_at ) @router.get("/{invoice_id}/download") def download_invoice_pdf( invoice_id: str, db: Session = Depends(get_db) ): """ Generate and serve standard Letter-sized PDF GST Invoice on-the-fly. """ invoice = db.query(Invoice).filter(Invoice.invoice_id == invoice_id).first() if not invoice: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Invoice not found" ) order = db.query(Order).filter(Order.order_id == invoice.order_id).first() if not order: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Sales order record missing, cannot generate PDF invoice" ) customer = db.query(EcomCustomer).filter(EcomCustomer.customer_id == invoice.customer_id).first() order_items = db.query(OrderItem).filter(OrderItem.order_id == invoice.order_id).all() try: pdf_bytes = generate_invoice_pdf(order, customer, order_items, db=db) return Response( content=pdf_bytes, media_type="application/pdf", headers={"Content-Disposition": f"inline; filename={invoice.invoice_no}.pdf"} ) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to compile PDF on-the-fly: {str(e)}" ) @router.get("/{invoice_id}/thermal-download") def download_thermal_invoice_pdf( invoice_id: str, db: Session = Depends(get_db) ): """ Generate and serve compact 80mm thermal receipt PDF on-the-fly. """ invoice = db.query(Invoice).filter(Invoice.invoice_id == invoice_id).first() if not invoice: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Invoice not found" ) order = db.query(Order).filter(Order.order_id == invoice.order_id).first() if not order: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Sales order record missing, cannot generate PDF invoice" ) customer = db.query(EcomCustomer).filter(EcomCustomer.customer_id == invoice.customer_id).first() order_items = db.query(OrderItem).filter(OrderItem.order_id == invoice.order_id).all() try: pdf_bytes = generate_thermal_invoice_pdf(order, customer, order_items) return Response( content=pdf_bytes, media_type="application/pdf", headers={"Content-Disposition": f"inline; filename=thermal_{invoice.invoice_no}.pdf"} ) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to compile thermal receipt PDF: {str(e)}" ) @router.get("/export/excel") def export_invoices_excel( start_date: Optional[str] = None, end_date: Optional[str] = None, db: Session = Depends(get_db) ): invoices = db.query(Invoice).order_by(Invoice.created_at.desc()).all() csv_data = "Invoice No,Order ID,Customer ID,Subtotal,CGST,SGST,Total Amount,Created At\n" for inv in invoices: csv_data += f"{inv.invoice_no},{inv.order_id},{inv.customer_id},{inv.subtotal},{inv.cgst},{inv.sgst},{inv.total_amount},{inv.created_at}\n" return Response( content=csv_data, media_type="text/csv", headers={"Content-Disposition": "attachment; filename=invoices_export.csv"} ) @router.get("/export/zip") def export_invoices_zip( start_date: Optional[str] = None, end_date: Optional[str] = None, db: Session = Depends(get_db) ): invoices = db.query(Invoice).order_by(Invoice.created_at.desc()).all() zip_buffer = io.BytesIO() with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zip_file: summary = "Invoice No,Order ID,Customer ID,Total Amount\n" for inv in invoices: summary += f"{inv.invoice_no},{inv.order_id},{inv.customer_id},{inv.total_amount}\n" zip_file.writestr("summary.csv", summary) zip_file.writestr("read_me.txt", "Sassynest CRM invoice bulk dump package.") zip_buffer.seek(0) return Response( content=zip_buffer.getvalue(), media_type="application/zip", headers={"Content-Disposition": "attachment; filename=invoices_export.zip"} )