107 lines
3.9 KiB
Python
107 lines
3.9 KiB
Python
"""Dashboard + analytics aggregates that feed the client dashboard UI."""
|
|
from collections import Counter
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from fastapi import APIRouter, Depends
|
|
|
|
from app.core.deps import require_org, Principal
|
|
from app.models import Product, Batch, QRCode, ScanLog, Plan, StorageUsage
|
|
|
|
router = APIRouter(prefix="/dashboard", tags=["dashboard"])
|
|
|
|
|
|
@router.get("/summary")
|
|
async def summary(p: Principal = Depends(require_org)):
|
|
oid = str(p.org.id)
|
|
products = await Product.find(Product.org_id == oid).to_list()
|
|
batches = await Batch.find(Batch.org_id == oid).to_list()
|
|
qrs = await QRCode.find(QRCode.org_id == oid).to_list()
|
|
scans = await ScanLog.find(ScanLog.org_id == oid).to_list()
|
|
plan = await Plan.get(p.org.plan_id) if p.org.plan_id else None
|
|
storage = await StorageUsage.find_one(StorageUsage.org_id == oid)
|
|
|
|
today = datetime.now(timezone.utc).date()
|
|
scans_today = sum(1 for s in scans if s.scanned_at.date() == today)
|
|
active_qr = sum(1 for q in qrs if q.status == "active")
|
|
avg_trust = round(sum(x.trust_score for x in products) / len(products)) if products else 0
|
|
|
|
return {
|
|
"products": len(products),
|
|
"product_limit": plan.product_limit if plan else None,
|
|
"qr_codes": len(qrs),
|
|
"active_qr": active_qr,
|
|
"batches": len(batches),
|
|
"scans_today": scans_today,
|
|
"total_scans": len(scans),
|
|
"trust_score_avg": avg_trust,
|
|
"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),
|
|
"plan_name": plan.name if plan else "Free",
|
|
"user_count": p.org.user_count,
|
|
"user_limit": plan.user_limit if plan else None,
|
|
}
|
|
|
|
|
|
@router.get("/scan-trend")
|
|
async def scan_trend(p: Principal = Depends(require_org), days: int = 7):
|
|
oid = str(p.org.id)
|
|
scans = await ScanLog.find(ScanLog.org_id == oid).to_list()
|
|
start = datetime.now(timezone.utc).date() - timedelta(days=days - 1)
|
|
buckets = {start + timedelta(days=i): 0 for i in range(days)}
|
|
for s in scans:
|
|
d = s.scanned_at.date()
|
|
if d in buckets:
|
|
buckets[d] += 1
|
|
return [{"date": str(d), "scans": c} for d, c in sorted(buckets.items())]
|
|
|
|
|
|
@router.get("/top-products")
|
|
async def top_products(
|
|
p: Principal = Depends(require_org),
|
|
limit: int = 8,
|
|
):
|
|
scans = await ScanLog.find(
|
|
ScanLog.org_id == str(p.org.id)
|
|
).to_list()
|
|
|
|
counter = Counter(s.product_id for s in scans if s.product_id)
|
|
|
|
products = await Product.find(
|
|
Product.org_id == str(p.org.id)
|
|
).to_list()
|
|
|
|
lookup = {str(prod.id): prod.name for prod in products}
|
|
|
|
result = []
|
|
|
|
for product_id, count in counter.most_common(limit):
|
|
result.append({
|
|
"name": lookup.get(product_id, "Unknown Product"),
|
|
"scans": count,
|
|
})
|
|
|
|
return result
|
|
|
|
|
|
@router.get("/top-locations")
|
|
async def top_locations(p: Principal = Depends(require_org), limit: int = 5):
|
|
scans = await ScanLog.find(ScanLog.org_id == str(p.org.id)).to_list()
|
|
counter = Counter(s.city for s in scans if s.city)
|
|
return [{"city": city, "scans": n} for city, n in counter.most_common(limit)]
|
|
|
|
|
|
@router.get("/recent-activity")
|
|
async def recent_activity(p: Principal = Depends(require_org), limit: int = 8):
|
|
from app.models import AuditLog
|
|
logs = await AuditLog.find(AuditLog.org_id == str(p.org.id)).to_list()
|
|
logs.sort(key=lambda x: x.created_at, reverse=True)
|
|
return [{"action": l.action, "details": l.details, "user": l.user_name,
|
|
"module": l.module, "at": l.created_at} for l in logs[:limit]]
|
|
|
|
|
|
@router.get("/devices")
|
|
async def devices(p: Principal = Depends(require_org)):
|
|
scans = await ScanLog.find(ScanLog.org_id == str(p.org.id)).to_list()
|
|
counter = Counter(s.device or "Other" for s in scans)
|
|
return [{"device": d, "scans": n} for d, n in counter.most_common()]
|