438 lines
19 KiB
Python
438 lines
19 KiB
Python
"""Super Admin Panel — Product Admin roles only. Org utilization, plans, revenue."""
|
|
import csv
|
|
import io
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from fastapi import APIRouter, Body, Depends, HTTPException, Response, status
|
|
|
|
from app.core.config import settings
|
|
from app.core.deps import require_admin_roles, Principal
|
|
from app.models import (
|
|
Organization, Plan, Product, QRCode, ScanLog, StorageUsage, Subscription,
|
|
Invoice, AuditLog, OrgMember, User, SupportNote, PlanHistory,
|
|
)
|
|
from app.models.documents import now
|
|
from app.models.enums import ProductAdminRole, SubscriptionStatus
|
|
|
|
router = APIRouter(prefix="/admin", tags=["admin"])
|
|
|
|
|
|
@router.get("/config-status")
|
|
async def config_status(p: Principal = Depends(require_admin_roles())):
|
|
"""Env-only credential status surfaced on the Product Admin settings page."""
|
|
return {
|
|
"google_oauth": settings.google_configured,
|
|
"razorpay": settings.razorpay_configured,
|
|
"aws_ses": settings.ses_configured,
|
|
"mock_db": settings.mock_db,
|
|
"mock_email": settings.mock_email,
|
|
"mock_storage": settings.mock_storage,
|
|
}
|
|
|
|
|
|
async def _utilization(org: Organization) -> dict:
|
|
oid = str(org.id)
|
|
plan = await Plan.get(org.plan_id) if org.plan_id else None
|
|
storage = await StorageUsage.find_one(StorageUsage.org_id == oid)
|
|
sub = await Subscription.find_one(Subscription.org_id == oid)
|
|
qr_count = await QRCode.find(QRCode.org_id == oid).count()
|
|
scans = await ScanLog.find(ScanLog.org_id == oid).count()
|
|
return {
|
|
"org_id": oid, "name": org.name, "logo_url": org.logo_url,
|
|
"plan": plan.name if plan else None,
|
|
"plan_price": plan.price if plan else 0,
|
|
"products_used": org.product_count,
|
|
"product_limit": plan.product_limit if plan else None,
|
|
"users_used": org.user_count,
|
|
"user_limit": plan.user_limit if plan else None,
|
|
"storage_used_gb": round((storage.used_bytes if storage else 0) / 1024 ** 3, 1),
|
|
"storage_total_gb": round((storage.total_bytes if storage else 0) / 1024 ** 3, 1),
|
|
"qr_codes": qr_count,
|
|
"total_scans": scans,
|
|
"subscription_status": org.subscription_status,
|
|
"renewal_date": sub.renewal_date if sub else None,
|
|
"payment_status": sub.payment_status if sub else None,
|
|
}
|
|
|
|
|
|
@router.get("/organizations")
|
|
async def list_orgs(p: Principal = Depends(require_admin_roles())):
|
|
orgs = await Organization.find_all().to_list()
|
|
return [await _utilization(o) for o in orgs]
|
|
|
|
|
|
@router.get("/organizations/{org_id}")
|
|
async def org_detail(org_id: str, p: Principal = Depends(require_admin_roles())):
|
|
org = await Organization.get(org_id)
|
|
if not org:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Org not found")
|
|
util = await _utilization(org)
|
|
members = await OrgMember.find(OrgMember.org_id == org_id).to_list()
|
|
admins = []
|
|
for m in members:
|
|
u = await User.get(m.user_id)
|
|
if u:
|
|
admins.append({"name": u.name, "email": u.email, "role": m.role,
|
|
"last_login": u.last_login})
|
|
return {**util, "admins": admins}
|
|
|
|
|
|
@router.get("/organizations/{org_id}/utilization")
|
|
async def org_utilization(org_id: str, p: Principal = Depends(require_admin_roles())):
|
|
org = await Organization.get(org_id)
|
|
if not org:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Org not found")
|
|
return await _utilization(org)
|
|
|
|
|
|
@router.post("/organizations/{org_id}/impersonate")
|
|
async def impersonate(org_id: str,
|
|
p: Principal = Depends(require_admin_roles(ProductAdminRole.super_admin))):
|
|
org = await Organization.get(org_id)
|
|
if not org:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Org not found")
|
|
await AuditLog(org_id=org_id, user_id=str(p.user.id), user_name=p.user.name,
|
|
action="org.impersonate", module="admin",
|
|
details=f"Impersonation started for {org.name}",
|
|
impersonator_id=str(p.user.id)).insert()
|
|
return {"message": f"Impersonating {org.name}", "org_id": org_id}
|
|
|
|
|
|
@router.get("/plans")
|
|
async def list_plans(p: Principal = Depends(require_admin_roles())):
|
|
plans = await Plan.find_all().to_list()
|
|
return [{"id": str(pl.id), "name": pl.name, "price": pl.price,
|
|
"product_limit": pl.product_limit, "user_limit": pl.user_limit,
|
|
"storage_gb": pl.storage_gb, "features": pl.features,
|
|
"enabled": pl.enabled} for pl in plans]
|
|
|
|
|
|
@router.post("/plans", status_code=201)
|
|
async def create_plan(body: dict = Body(...),
|
|
p: Principal = Depends(require_admin_roles(ProductAdminRole.super_admin))):
|
|
if await Plan.find_one(Plan.name == body.get("name")):
|
|
raise HTTPException(status.HTTP_409_CONFLICT, "Plan name already exists")
|
|
plan = Plan(
|
|
name=body.get("name", "New Plan"), price=int(body.get("price", 0)),
|
|
product_limit=body.get("product_limit"), user_limit=body.get("user_limit"),
|
|
storage_gb=int(body.get("storage_gb", 5)),
|
|
grace_days=int(body.get("grace_days", 30)),
|
|
archive_years=int(body.get("archive_years", 1)),
|
|
features=body.get("features", []), enabled=body.get("enabled", True),
|
|
)
|
|
await plan.insert()
|
|
await AuditLog(user_id=str(p.user.id), user_name=p.user.name,
|
|
action="plan.create", module="admin",
|
|
details=f"Created plan {plan.name}").insert()
|
|
return {"id": str(plan.id)}
|
|
|
|
|
|
@router.put("/plans/{plan_id}")
|
|
async def update_plan(plan_id: str, body: dict = Body(...),
|
|
p: Principal = Depends(require_admin_roles(
|
|
ProductAdminRole.super_admin, ProductAdminRole.finance))):
|
|
plan = await Plan.get(plan_id)
|
|
if not plan:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Plan not found")
|
|
for k in ("name", "price", "product_limit", "user_limit", "storage_gb",
|
|
"grace_days", "archive_years", "features", "enabled"):
|
|
if k in body:
|
|
setattr(plan, k, body[k])
|
|
await plan.save()
|
|
await AuditLog(user_id=str(p.user.id), user_name=p.user.name,
|
|
action="plan.update", module="admin",
|
|
details=f"Updated plan {plan.name}").insert()
|
|
return {"message": "updated"}
|
|
|
|
|
|
@router.get("/dashboard")
|
|
async def admin_dashboard(p: Principal = Depends(require_admin_roles())):
|
|
"""Platform-wide overview KPIs for the Super Admin landing page."""
|
|
orgs = await Organization.find_all().to_list()
|
|
plans = {str(pl.id): pl for pl in await Plan.find_all().to_list()}
|
|
invoices = await Invoice.find_all().to_list()
|
|
total_qr = await QRCode.find_all().count()
|
|
total_scans = await ScanLog.find_all().count()
|
|
|
|
def lc(s): return sum(1 for o in orgs if o.subscription_status == s)
|
|
|
|
# MRR = sum of active orgs' plan prices
|
|
mrr = 0
|
|
paying = 0
|
|
for o in orgs:
|
|
if o.subscription_status == SubscriptionStatus.active and o.plan_id in plans:
|
|
price = plans[o.plan_id].price
|
|
if price > 0:
|
|
mrr += price
|
|
paying += 1
|
|
|
|
paid = [i for i in invoices if i.status == "paid"]
|
|
total_revenue = sum(i.total for i in paid)
|
|
|
|
# revenue trend — last 6 months by invoice month
|
|
from collections import defaultdict
|
|
buckets = defaultdict(int)
|
|
for i in paid:
|
|
key = i.created_at.strftime("%Y-%m")
|
|
buckets[key] += i.total
|
|
months = sorted(buckets.keys())[-6:]
|
|
trend = [{"month": m, "revenue": buckets[m]} for m in months]
|
|
|
|
return {
|
|
"total_orgs": len(orgs),
|
|
"active": lc(SubscriptionStatus.active),
|
|
"grace": lc(SubscriptionStatus.grace),
|
|
"archived": lc(SubscriptionStatus.archived),
|
|
"deleted": lc(SubscriptionStatus.deleted),
|
|
"mrr": mrr, "arr": mrr * 12, "paying_orgs": paying,
|
|
"arpo": round(mrr / paying) if paying else 0,
|
|
"total_revenue": total_revenue,
|
|
"total_qr": total_qr, "total_scans": total_scans,
|
|
"revenue_trend": trend,
|
|
}
|
|
|
|
|
|
@router.get("/revenue")
|
|
async def revenue(p: Principal = Depends(require_admin_roles(
|
|
ProductAdminRole.super_admin, ProductAdminRole.finance))):
|
|
orgs = await Organization.find_all().to_list()
|
|
plans = {str(pl.id): pl for pl in await Plan.find_all().to_list()}
|
|
invoices = await Invoice.find_all().to_list()
|
|
invoices.sort(key=lambda i: i.created_at, reverse=True)
|
|
paid = [i for i in invoices if i.status == "paid"]
|
|
|
|
mrr = sum(plans[o.plan_id].price for o in orgs
|
|
if o.subscription_status == SubscriptionStatus.active
|
|
and o.plan_id in plans and plans[o.plan_id].price > 0)
|
|
|
|
# plan-wise breakdown of active paying orgs
|
|
from collections import defaultdict
|
|
plan_rev = defaultdict(int)
|
|
for o in orgs:
|
|
if o.subscription_status == SubscriptionStatus.active and o.plan_id in plans:
|
|
pl = plans[o.plan_id]
|
|
if pl.price > 0:
|
|
plan_rev[pl.name] += pl.price
|
|
|
|
org_names = {str(o.id): o.name for o in orgs}
|
|
return {
|
|
"mrr": mrr, "arr": mrr * 12,
|
|
"total_revenue": sum(i.total for i in paid),
|
|
"invoices_count": len(invoices), "paid_count": len(paid),
|
|
"plan_breakdown": [{"plan": k, "revenue": v} for k, v in plan_rev.items()],
|
|
"invoices": [{"id": str(i.id), "invoice_number": i.invoice_number,
|
|
"org": org_names.get(i.org_id, "—"), "plan": i.plan_name,
|
|
"subtotal": i.subtotal, "gst": i.gst_amount, "total": i.total,
|
|
"status": i.status, "at": i.created_at} for i in invoices[:100]],
|
|
}
|
|
|
|
|
|
@router.get("/revenue/export")
|
|
async def revenue_export(p: Principal = Depends(require_admin_roles(
|
|
ProductAdminRole.super_admin, ProductAdminRole.finance))):
|
|
orgs = {str(o.id): o.name for o in await Organization.find_all().to_list()}
|
|
invoices = await Invoice.find_all().to_list()
|
|
invoices.sort(key=lambda i: i.created_at, reverse=True)
|
|
buf = io.StringIO()
|
|
w = csv.writer(buf)
|
|
w.writerow(["Invoice", "Organization", "Plan", "Subtotal", "GST", "Total", "Status", "Date"])
|
|
for i in invoices:
|
|
w.writerow([i.invoice_number, orgs.get(i.org_id, ""), i.plan_name,
|
|
i.subtotal, i.gst_amount, i.total, i.status, i.created_at.isoformat()])
|
|
return Response(content=buf.getvalue(), media_type="text/csv",
|
|
headers={"Content-Disposition": 'attachment; filename="revenue.csv"'})
|
|
|
|
|
|
@router.get("/audit-logs")
|
|
async def audit_logs(p: Principal = Depends(require_admin_roles(
|
|
ProductAdminRole.super_admin, ProductAdminRole.admin)), limit: int = 100):
|
|
logs = await AuditLog.find_all().to_list()
|
|
logs.sort(key=lambda x: x.created_at, reverse=True)
|
|
return [{"at": l.created_at, "user": l.user_name, "role": l.user_role,
|
|
"action": l.action, "module": l.module, "details": l.details,
|
|
"ip": l.ip} for l in logs[:limit]]
|
|
|
|
|
|
# ---------- org detail extras: history, payments, support notes ----------
|
|
|
|
@router.get("/organizations/{org_id}/history")
|
|
async def org_history(org_id: str, p: Principal = Depends(require_admin_roles())):
|
|
subs = await Subscription.find(Subscription.org_id == org_id).to_list()
|
|
invs = await Invoice.find(Invoice.org_id == org_id).to_list()
|
|
invs.sort(key=lambda i: i.created_at, reverse=True)
|
|
plan_hist = await PlanHistory.find(PlanHistory.org_id == org_id).to_list()
|
|
plan_hist.sort(key=lambda h: h.created_at, reverse=True)
|
|
return {
|
|
"subscriptions": [{"plan": s.plan_name, "status": s.status,
|
|
"start": s.start_date, "renewal": s.renewal_date,
|
|
"payment_status": s.payment_status} for s in subs],
|
|
"invoices": [{"invoice_number": i.invoice_number, "total": i.total,
|
|
"status": i.status, "at": i.created_at} for i in invs],
|
|
"plan_history": [{"from": h.from_plan, "to": h.to_plan,
|
|
"reason": h.reason, "at": h.created_at} for h in plan_hist],
|
|
}
|
|
|
|
|
|
@router.get("/organizations/{org_id}/notes")
|
|
async def list_notes(org_id: str, p: Principal = Depends(require_admin_roles())):
|
|
notes = await SupportNote.find(SupportNote.org_id == org_id).to_list()
|
|
notes.sort(key=lambda n: n.created_at, reverse=True)
|
|
return [{"id": str(n.id), "note": n.note, "flag": n.flag,
|
|
"author": n.author_name, "at": n.created_at} for n in notes]
|
|
|
|
|
|
@router.post("/organizations/{org_id}/notes", status_code=201)
|
|
async def add_note(org_id: str, body: dict = Body(...),
|
|
p: Principal = Depends(require_admin_roles())):
|
|
note = SupportNote(org_id=org_id, author_id=str(p.user.id),
|
|
author_name=p.user.name, note=body.get("note", ""),
|
|
flag=body.get("flag"))
|
|
await note.insert()
|
|
return {"id": str(note.id)}
|
|
|
|
|
|
# ---------- subscription override / extend / reminder ----------
|
|
|
|
@router.put("/organizations/{org_id}/subscription")
|
|
async def override_subscription(org_id: str, body: dict = Body(...),
|
|
p: Principal = Depends(require_admin_roles(
|
|
ProductAdminRole.super_admin, ProductAdminRole.admin))):
|
|
org = await Organization.get(org_id)
|
|
sub = await Subscription.find_one(Subscription.org_id == org_id)
|
|
if not org or not sub:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Org or subscription not found")
|
|
if "status" in body and body["status"] in [s.value for s in SubscriptionStatus]:
|
|
org.subscription_status = SubscriptionStatus(body["status"])
|
|
sub.status = org.subscription_status
|
|
if body.get("extend_days"):
|
|
base = sub.renewal_date or now()
|
|
if base.tzinfo is None:
|
|
base = base.replace(tzinfo=timezone.utc)
|
|
sub.renewal_date = base + timedelta(days=int(body["extend_days"]))
|
|
if body.get("plan_id"):
|
|
plan = await Plan.get(body["plan_id"])
|
|
if plan:
|
|
org.plan_id = str(plan.id)
|
|
sub.plan_id = str(plan.id)
|
|
sub.plan_name = plan.name
|
|
await org.save()
|
|
await sub.save()
|
|
await AuditLog(org_id=org_id, user_id=str(p.user.id), user_name=p.user.name,
|
|
action="subscription.override", module="admin",
|
|
details=f"Admin override on {org.name}").insert()
|
|
return {"message": "updated", "status": org.subscription_status}
|
|
|
|
|
|
@router.post("/organizations/{org_id}/send-reminder")
|
|
async def send_reminder(org_id: str, p: Principal = Depends(require_admin_roles())):
|
|
from app.services.email import send_email
|
|
org = await Organization.get(org_id)
|
|
if not org:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Org not found")
|
|
to = org.contact_email or "admin@example.com"
|
|
await send_email(to, "VerifyPack subscription renewal reminder",
|
|
f"Hi {org.name}, your subscription is due for renewal.", "renewal_reminder")
|
|
await AuditLog(org_id=org_id, user_id=str(p.user.id), user_name=p.user.name,
|
|
action="reminder.send", module="admin",
|
|
details=f"Renewal reminder to {to}").insert()
|
|
return {"sent": True, "to": to}
|
|
|
|
|
|
# ---------- subscription lifecycle + utilization report ----------
|
|
|
|
@router.get("/organizations/{org_id}/support-view")
|
|
async def support_view(org_id: str, p: Principal = Depends(require_admin_roles())):
|
|
"""Read-only debugging view for Support staff."""
|
|
org = await Organization.get(org_id)
|
|
if not org:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Org not found")
|
|
products = await Product.find(Product.org_id == org_id).to_list()
|
|
qrs = await QRCode.find(QRCode.org_id == org_id).to_list()
|
|
scans = await ScanLog.find(ScanLog.org_id == org_id).to_list()
|
|
scans.sort(key=lambda s: s.scanned_at, reverse=True)
|
|
return {
|
|
"org": {"name": org.name, "slug": org.slug, "status": org.subscription_status,
|
|
"gstin": org.gstin, "contact_email": org.contact_email},
|
|
"products": [{"name": x.name, "sku": x.sku, "status": x.status,
|
|
"qr_count": x.qr_count} for x in products],
|
|
"qr_codes": [{"code": q.code, "status": q.status, "scans": q.scan_count}
|
|
for q in qrs[:50]],
|
|
"recent_scans": [{"at": s.scanned_at, "code": s.qr_code, "city": s.city,
|
|
"device": s.device} for s in scans[:25]],
|
|
}
|
|
|
|
|
|
@router.get("/subscription-lifecycle")
|
|
async def lifecycle(p: Principal = Depends(require_admin_roles())):
|
|
orgs = await Organization.find_all().to_list()
|
|
def count(s): return sum(1 for o in orgs if o.subscription_status == s)
|
|
upcoming = []
|
|
for o in orgs:
|
|
sub = await Subscription.find_one(Subscription.org_id == str(o.id))
|
|
if sub and sub.renewal_date:
|
|
rd = sub.renewal_date
|
|
if rd.tzinfo is None:
|
|
rd = rd.replace(tzinfo=timezone.utc)
|
|
days = (rd - now()).days
|
|
if 0 <= days <= 30:
|
|
upcoming.append({"org": o.name, "org_id": str(o.id),
|
|
"renewal": sub.renewal_date, "days": days})
|
|
upcoming.sort(key=lambda x: x["days"])
|
|
return {
|
|
"active": count(SubscriptionStatus.active),
|
|
"grace": count(SubscriptionStatus.grace),
|
|
"archived": count(SubscriptionStatus.archived),
|
|
"deleted": count(SubscriptionStatus.deleted),
|
|
"upcoming_expiries": upcoming,
|
|
}
|
|
|
|
|
|
@router.get("/utilization-report")
|
|
async def utilization_report(format: str = "csv",
|
|
p: Principal = Depends(require_admin_roles())):
|
|
orgs = await Organization.find_all().to_list()
|
|
rows = [await _utilization(o) for o in orgs]
|
|
if format == "pdf":
|
|
# Simple PDF table via reportlab
|
|
import io as _io
|
|
from reportlab.lib.pagesizes import A4
|
|
from reportlab.pdfgen import canvas
|
|
from reportlab.lib.units import mm
|
|
buf = _io.BytesIO()
|
|
c = canvas.Canvas(buf, pagesize=A4)
|
|
w, h = A4
|
|
y = h - 25 * mm
|
|
c.setFont("Helvetica-Bold", 14)
|
|
c.drawString(20 * mm, y, "VerifyPack — Utilization Report")
|
|
c.setFont("Helvetica", 8)
|
|
y -= 12 * mm
|
|
c.drawString(20 * mm, y, "Org / Plan / Products / Users / QRs / Scans / Status")
|
|
y -= 6 * mm
|
|
for r in rows:
|
|
line = (f"{r['name'][:24]:24} {str(r['plan'] or '-')[:10]:10} "
|
|
f"{r['products_used']}/{r['product_limit'] or '∞'} "
|
|
f"{r['users_used']}/{r['user_limit'] or '∞'} "
|
|
f"{r['qr_codes']} {r['total_scans']} {r['subscription_status']}")
|
|
c.drawString(20 * mm, y, line)
|
|
y -= 5 * mm
|
|
if y < 20 * mm:
|
|
c.showPage(); y = h - 25 * mm
|
|
c.showPage(); c.save()
|
|
return Response(content=buf.getvalue(), media_type="application/pdf",
|
|
headers={"Content-Disposition": 'attachment; filename="utilization.pdf"'})
|
|
# CSV
|
|
buf = io.StringIO()
|
|
wr = csv.writer(buf)
|
|
wr.writerow(["Organization", "Plan", "Price", "Products", "Product Limit",
|
|
"Users", "User Limit", "Storage GB", "QR Codes", "Total Scans", "Status"])
|
|
for r in rows:
|
|
wr.writerow([r["name"], r["plan"], r["plan_price"], r["products_used"],
|
|
r["product_limit"], r["users_used"], r["user_limit"],
|
|
r["storage_used_gb"], r["qr_codes"], r["total_scans"],
|
|
r["subscription_status"]])
|
|
return Response(content=buf.getvalue(), media_type="text/csv",
|
|
headers={"Content-Disposition": 'attachment; filename="utilization.csv"'})
|
|
|
|
|
|
|