"""RBS REGAL — Secure Invoice PDF Link System. Generates encrypted short tokens for invoices so customers can view + download PDF without logging in. The token URL is what gets shared via WhatsApp / Email. Endpoints (mounted under /api/invoice-share): - POST /token/{invoice_id} — generate (or rotate) a share token - GET /public/{token} — returns the invoice JSON (PUBLIC, no auth) - GET /public/{token}/pdf — streams a watermarked PDF (PUBLIC, no auth) - GET /public/{token}/verify — returns verification hash for QR check - GET /settings — Super Admin: read share settings - PUT /settings — Super Admin: update share settings """ from __future__ import annotations import hashlib import io import os import secrets from datetime import datetime, timedelta, timezone from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.responses import StreamingResponse from pydantic import BaseModel from auth import get_current_user, require_admin from bson import ObjectId router = APIRouter(prefix="/api/invoice-share", tags=["invoice-share"]) # ---------------------------------------------------------------------------- # Settings — stored in db.invoice_share_settings (singleton _id="active") # ---------------------------------------------------------------------------- DEFAULT_SETTINGS = { "_id": "active", "enable_share": True, "auto_download": True, "expiry_days": 30, "watermark": True, "watermark_text": "Generated by RGE REGALGOA ERP AI", "qr_verify": True, "digital_signature": False, "custom_domain": "", # e.g. "invoice.regalmarketing.com" — must be DNS-pointed via Emergent Support "brand_name": "RGE REGALGOA ERP AI", "wa_templates": { "sale": "Namaste {customer_name},\n\nREGAL MARKETING se aapka invoice tayyar hai.\n\nInvoice No: {invoice_no}\nAmount: ₹{total}\n\nInvoice Download: {link}\n\nDhanyavaad\nRGE REGALGOA ERP AI", "purchase": "Namaste,\n\nPurchase invoice {invoice_no} ka link:\n{link}\n\nAmount: ₹{total}\n\nRGE REGALGOA ERP AI", "payment": "Namaste {customer_name},\n\nReceipt no. {invoice_no} for ₹{total} attached.\n\n{link}\n\nDhanyavaad\nRGE REGALGOA ERP AI", }, "updated_at": "", "updated_by": "", } async def _get_settings(db) -> dict: doc = await db.invoice_share_settings.find_one({"_id": "active"}) if not doc: doc = dict(DEFAULT_SETTINGS) await db.invoice_share_settings.insert_one(doc) # Merge in any new defaults the user hasn't saved yet merged = {**DEFAULT_SETTINGS, **doc} return merged class ShareSettingsUpdate(BaseModel): enable_share: bool | None = None auto_download: bool | None = None expiry_days: int | None = None watermark: bool | None = None watermark_text: str | None = None qr_verify: bool | None = None digital_signature: bool | None = None custom_domain: str | None = None brand_name: str | None = None wa_templates: dict | None = None @router.get("/settings") async def read_settings(request: Request, user=Depends(require_admin)): return await _get_settings(request.app.state.db) @router.put("/settings") async def update_settings(payload: ShareSettingsUpdate, request: Request, user=Depends(require_admin)): db = request.app.state.db patch = {k: v for k, v in payload.model_dump(exclude_none=True).items()} patch["updated_at"] = datetime.now(timezone.utc).isoformat() patch["updated_by"] = user.get("email", "") await db.invoice_share_settings.update_one( {"_id": "active"}, {"$set": patch}, upsert=True ) return await _get_settings(db) # ---------------------------------------------------------------------------- # Token generation # ---------------------------------------------------------------------------- def _new_token() -> str: """Cryptographically-strong, URL-safe, 24-char token.""" return secrets.token_urlsafe(18)[:24] @router.post("/token/{invoice_id}") async def issue_token(invoice_id: str, request: Request, user=Depends(get_current_user)): db = request.app.state.db s = await _get_settings(db) if not s["enable_share"]: raise HTTPException(status_code=403, detail="Public share is disabled by Super Admin") # Confirm invoice exists & user has access (same company as user's session) try: inv = await db.invoices.find_one({"_id": ObjectId(invoice_id)}) except Exception: raise HTTPException(status_code=400, detail="Invalid invoice id") if not inv: raise HTTPException(status_code=404, detail="Invoice not found") # Re-use a non-expired token if one exists for the same invoice now = datetime.now(timezone.utc) existing = await db.invoice_share_tokens.find_one({ "invoice_id": str(invoice_id), "expires_at": {"$gt": now.isoformat()}, "revoked": {"$ne": True}, }) if existing: token = existing["token"] expires_at_iso = existing.get("expires_at", "") else: token = _new_token() expires_at = now + timedelta(days=int(s["expiry_days"] or 30)) expires_at_iso = expires_at.isoformat() await db.invoice_share_tokens.insert_one({ "token": token, "invoice_id": str(invoice_id), "invoice_no": inv.get("invoice_no", ""), "created_at": now.isoformat(), "created_by": user.get("email", ""), "expires_at": expires_at_iso, "revoked": False, "view_count": 0, "download_count": 0, }) # Build public URL — prefer custom domain, else app origin from request custom = (s.get("custom_domain") or "").strip() origin = "" if custom: origin = f"https://{custom}" else: origin = str(request.base_url).rstrip("/") # If running behind ingress, base_url is the app domain public_path = f"/p/inv/{token}" return { "token": token, "url": f"{origin}{public_path}", "path": public_path, "expires_at": expires_at_iso, "settings": { "auto_download": s["auto_download"], "watermark": s["watermark"], "qr_verify": s["qr_verify"], "brand_name": s["brand_name"], }, } @router.delete("/token/{token}") async def revoke_token(token: str, request: Request, user=Depends(require_admin)): db = request.app.state.db res = await db.invoice_share_tokens.update_one( {"token": token}, {"$set": {"revoked": True, "revoked_at": datetime.now(timezone.utc).isoformat()}} ) return {"revoked": bool(res.modified_count)} # ---------------------------------------------------------------------------- # Public endpoints # ---------------------------------------------------------------------------- async def _resolve_invoice_by_token(db, token: str) -> tuple[dict, dict, dict]: """Resolve (token_doc, invoice_doc, company_doc); raises 404 / 410 on issues.""" tdoc = await db.invoice_share_tokens.find_one({"token": token}) if not tdoc: raise HTTPException(status_code=404, detail="Invalid or expired link") if tdoc.get("revoked"): raise HTTPException(status_code=410, detail="This link has been revoked") now = datetime.now(timezone.utc).isoformat() if tdoc.get("expires_at") and tdoc["expires_at"] < now: raise HTTPException(status_code=410, detail="Link expired") try: inv = await db.invoices.find_one({"_id": ObjectId(tdoc["invoice_id"])}) except Exception: inv = None if not inv: raise HTTPException(status_code=404, detail="Invoice no longer exists") company = None if inv.get("company_id"): try: company = await db.companies.find_one({"_id": ObjectId(inv["company_id"])}) except Exception: company = None # Strip ObjectId from token doc — callers only need scalar fields tdoc_safe = dict(tdoc) tdoc_safe.pop("_id", None) return tdoc_safe, inv, (company or {}) def _safe_invoice(inv: dict) -> dict: """Strip MongoDB ObjectId + audit fields for public consumption.""" inv = dict(inv) inv.pop("_id", None) inv.pop("created_by", None) inv.pop("updated_by", None) inv.pop("deleted_by", None) # Keep enough for full PDF render return inv def _safe_company(c: dict) -> dict: c = dict(c or {}) c.pop("_id", None) return c def _verification_hash(inv: dict, token: str) -> str: """SHA-256 of invoice number + total + token — printed under QR for verification.""" raw = f"{inv.get('invoice_no')}|{inv.get('total')}|{token}".encode("utf-8") return hashlib.sha256(raw).hexdigest()[:16].upper() @router.get("/public/{token}") async def public_view(token: str, request: Request): db = request.app.state.db s = await _get_settings(db) if not s["enable_share"]: raise HTTPException(status_code=403, detail="Public share is currently disabled") tdoc, inv, company = await _resolve_invoice_by_token(db, token) # Bump view count (fire-and-forget) await db.invoice_share_tokens.update_one({"token": token}, {"$inc": {"view_count": 1}}) return { "invoice": _safe_invoice(inv), "company": _safe_company(company), "settings": { "auto_download": s["auto_download"], "watermark": s["watermark"], "watermark_text": s["watermark_text"], "qr_verify": s["qr_verify"], "brand_name": s["brand_name"], }, "verification_hash": _verification_hash(inv, token), "expires_at": tdoc.get("expires_at"), } @router.get("/public/{token}/verify") async def public_verify(token: str, request: Request): """Lightweight endpoint hit by the QR-scan flow to confirm authenticity.""" db = request.app.state.db tdoc, inv, company = await _resolve_invoice_by_token(db, token) return { "valid": True, "invoice_no": inv.get("invoice_no"), "total": inv.get("total"), "issued_by": company.get("name", ""), "verification_hash": _verification_hash(inv, token), "issued_on": inv.get("invoice_date"), } # ---------------------------------------------------------------------------- # PDF generation (reportlab) # ---------------------------------------------------------------------------- def _build_pdf(inv: dict, company: dict, settings: dict, token: str, verification_hash: str, base_url: str = "") -> bytes: from reportlab.lib.pagesizes import A4 from reportlab.lib import colors from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle from reportlab.lib.units import mm from reportlab.pdfgen import canvas as rl_canvas from reportlab.platypus import Paragraph, Table, TableStyle, SimpleDocTemplate, Spacer, KeepTogether import qrcode buf = io.BytesIO() # Watermark + footer via onPage callback brand = settings.get("brand_name") or "RGE REGALGOA ERP AI" watermark_text = settings.get("watermark_text") or f"Generated by {brand}" def _on_page(c: "rl_canvas.Canvas", _doc): c.saveState() if settings.get("watermark"): c.setFillColor(colors.Color(0.85, 0.85, 0.85, alpha=0.35)) c.setFont("Helvetica-Bold", 38) c.translate(A4[0] / 2, A4[1] / 2) c.rotate(45) c.drawCentredString(0, 0, watermark_text) c.restoreState() # Footer band c.saveState() c.setFillColor(colors.HexColor("#0C7C59")) c.rect(0, 0, A4[0], 14, fill=1, stroke=0) c.setFillColor(colors.white) c.setFont("Helvetica", 7) c.drawString(10 * mm, 4, f"{brand} · Verify: {verification_hash}") c.drawRightString(A4[0] - 10 * mm, 4, f"Page {_doc.page}") c.restoreState() styles = getSampleStyleSheet() p_small = ParagraphStyle("p_sm", parent=styles["Normal"], fontSize=8, leading=10) p_norm = ParagraphStyle("p_norm", parent=styles["Normal"], fontSize=9.5, leading=12) p_right = ParagraphStyle("p_right", parent=styles["Normal"], fontSize=9.5, alignment=2) doc = SimpleDocTemplate( buf, pagesize=A4, leftMargin=14 * mm, rightMargin=14 * mm, topMargin=14 * mm, bottomMargin=16 * mm, ) story = [] # Header — company info + invoice meta title = "TAX INVOICE" if inv.get("type") == "sale" else (inv.get("type", "INVOICE").upper()) inv_no = inv.get("invoice_no", "") inv_dt = inv.get("invoice_date", "") status = (inv.get("status") or "").upper() co_lines = [] if company.get("name"): co_lines.append(f"{company.get('name')}") if company.get("address"): co_lines.append(company["address"]) bits = [] if company.get("gstin"): bits.append(f"GSTIN: {company['gstin']}") if company.get("phone"): bits.append(f"Phone: {company['phone']}") if company.get("email"): bits.append(f"Email: {company['email']}") if bits: co_lines.append(" · ".join(bits)) left_html = "
".join(co_lines) right_html = ( f"{title}
" f"{inv_no}
" f"Date: {inv_dt}
" + (f"Status: {status}" if status else "") ) t = Table([[Paragraph(left_html, p_norm), Paragraph(right_html, p_right)]], colWidths=[doc.width * 0.6, doc.width * 0.4]) t.setStyle(TableStyle([("VALIGN", (0, 0), (-1, -1), "TOP")])) story.append(t) story.append(Spacer(1, 6)) # Divider story.append(Table([[""]], colWidths=[doc.width], style=TableStyle([ ("LINEABOVE", (0, 0), (-1, -1), 1.2, colors.HexColor("#0C7C59")), ]))) story.append(Spacer(1, 6)) # Bill-to party_block = [] party_block.append(f"Bill To: {inv.get('party_name','')}") if inv.get("party_gstin"): party_block.append(f"GSTIN: {inv['party_gstin']}") if inv.get("party_state"): party_block.append(f"State: {inv['party_state']}") if inv.get("billing_address"): party_block.append(inv["billing_address"]) story.append(Paragraph("
".join(party_block), p_norm)) story.append(Spacer(1, 8)) # Line items head = ["#", "Item", "HSN", "Qty", "Rate", "Disc%", "GST%", "Amount"] rows = [head] for idx, ln in enumerate(inv.get("lines", []), 1): rows.append([ str(idx), ln.get("name", ""), ln.get("hsn", ""), f"{ln.get('qty', 0)} {ln.get('unit') or ''}", f"₹{float(ln.get('rate', 0)):,.2f}", f"{float(ln.get('discount', 0)):.0f}%", f"{float(ln.get('gst_rate', 0)):.0f}%", f"₹{float(ln.get('amount', 0)):,.2f}", ]) line_tbl = Table(rows, repeatRows=1, colWidths=[ doc.width * 0.04, doc.width * 0.34, doc.width * 0.08, doc.width * 0.10, doc.width * 0.10, doc.width * 0.08, doc.width * 0.08, doc.width * 0.18, ]) line_tbl.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#0C7C59")), ("TEXTCOLOR", (0, 0), (-1, 0), colors.white), ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"), ("FONTSIZE", (0, 0), (-1, -1), 8.5), ("ALIGN", (3, 1), (-1, -1), "RIGHT"), ("ALIGN", (0, 0), (0, -1), "CENTER"), ("ALIGN", (0, 0), (-1, 0), "CENTER"), ("VALIGN", (0, 0), (-1, -1), "MIDDLE"), ("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.HexColor("#F6F8F5")]), ("GRID", (0, 0), (-1, -1), 0.4, colors.HexColor("#CCD8CC")), ("LEFTPADDING", (0, 0), (-1, -1), 4), ("RIGHTPADDING", (0, 0), (-1, -1), 4), ("TOPPADDING", (0, 0), (-1, -1), 3), ("BOTTOMPADDING", (0, 0), (-1, -1), 3), ])) story.append(line_tbl) story.append(Spacer(1, 8)) # Totals + QR def _money(v): try: return f"₹{float(v):,.2f}" except Exception: return f"₹{v}" totals_rows = [ ["Subtotal", _money(inv.get("subtotal", 0))], ["CGST", _money(inv.get("cgst", 0))], ["SGST", _money(inv.get("sgst", 0))], ["IGST", _money(inv.get("igst", 0))], ] totals_rows.append(["Total", _money(inv.get("total", 0))]) totals_rows.append(["Paid", _money(inv.get("payment_received", 0))]) balance = max(0.0, float(inv.get("total", 0)) - float(inv.get("payment_received", 0))) totals_rows.append(["Balance Due", _money(balance)]) totals_tbl = Table(totals_rows, colWidths=[doc.width * 0.20, doc.width * 0.20]) totals_tbl.setStyle(TableStyle([ ("FONTSIZE", (0, 0), (-1, -1), 9.5), ("ALIGN", (1, 0), (1, -1), "RIGHT"), ("LINEBELOW", (0, 0), (-1, 3), 0.3, colors.HexColor("#CCD8CC")), ("LINEABOVE", (0, 4), (-1, 4), 1.2, colors.HexColor("#0C7C59")), ("FONTNAME", (0, 4), (-1, 4), "Helvetica-Bold"), ("BACKGROUND", (0, 4), (-1, 4), colors.HexColor("#E8F2EC")), ("BACKGROUND", (0, 6), (-1, 6), colors.HexColor("#FFF6E5")), ("FONTNAME", (0, 6), (-1, 6), "Helvetica-Bold"), ("TOPPADDING", (0, 0), (-1, -1), 3), ("BOTTOMPADDING", (0, 0), (-1, -1), 3), ])) qr_img_data = None if settings.get("qr_verify"): # Build verify URL — prefer custom domain, else base_url, else relative path custom = (settings.get("custom_domain") or "").strip() if custom: verify_url = f"https://{custom}/p/inv/{token}/verify" elif base_url: verify_url = f"{base_url.rstrip('/')}/api/invoice-share/public/{token}/verify" else: verify_url = f"/api/invoice-share/public/{token}/verify" qr_img = qrcode.make(verify_url, box_size=4, border=1) qr_buf = io.BytesIO() qr_img.save(qr_buf, format="PNG") qr_img_data = qr_buf.getvalue() if qr_img_data: from reportlab.platypus import Image as RLImage qr_img_obj = RLImage(io.BytesIO(qr_img_data), width=22 * mm, height=22 * mm) verify_para = Paragraph( f"Scan to verify
Hash: {verification_hash}", p_small, ) composite = Table([[qr_img_obj, verify_para]], colWidths=[24 * mm, doc.width * 0.20 - 24 * mm]) composite.setStyle(TableStyle([("VALIGN", (0, 0), (-1, -1), "MIDDLE")])) bottom = Table( [[composite, totals_tbl]], colWidths=[doc.width * 0.40, doc.width * 0.40], ) else: bottom = Table([["", totals_tbl]], colWidths=[doc.width * 0.40, doc.width * 0.40]) bottom.setStyle(TableStyle([("VALIGN", (0, 0), (-1, -1), "TOP")])) story.append(KeepTogether(bottom)) # Terms if inv.get("terms"): story.append(Spacer(1, 8)) story.append(Paragraph("Terms & Conditions:", p_small)) story.append(Paragraph(str(inv["terms"]).replace("\n", "
"), p_small)) # Notice story.append(Spacer(1, 8)) story.append(Paragraph( f"This is a system-generated read-only document. Verify authenticity at the QR link or via verification hash {verification_hash}.", p_small, )) doc.build(story, onFirstPage=_on_page, onLaterPages=_on_page) return buf.getvalue() @router.get("/public/{token}/pdf") async def public_pdf(token: str, request: Request): db = request.app.state.db s = await _get_settings(db) if not s["enable_share"]: raise HTTPException(status_code=403, detail="Public share is disabled") tdoc, inv, company = await _resolve_invoice_by_token(db, token) vhash = _verification_hash(inv, token) base_url = str(request.base_url).rstrip("/") pdf_bytes = _build_pdf(inv, company, s, token, vhash, base_url=base_url) # Bump download count await db.invoice_share_tokens.update_one({"token": token}, {"$inc": {"download_count": 1}}) filename = f"{inv.get('invoice_no','invoice').replace('/', '-')}.pdf" headers = { "Content-Disposition": f'inline; filename="{filename}"', "Cache-Control": "no-store", } return StreamingResponse(io.BytesIO(pdf_bytes), media_type="application/pdf", headers=headers) @router.get("/tokens") async def list_tokens(request: Request, user=Depends(require_admin), limit: int = 200): db = request.app.state.db cursor = db.invoice_share_tokens.find({}, {"_id": 0}).sort("created_at", -1).limit(limit) return [d async for d in cursor]