316 lines
8.5 KiB
Python
316 lines
8.5 KiB
Python
"""Analytics & Insights — scan aggregates with date-range filtering + CSV export."""
|
|
import csv
|
|
import io
|
|
from collections import Counter, defaultdict
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, Response
|
|
|
|
from app.core.deps import require_org, Principal
|
|
from app.models import ScanLog, Product
|
|
|
|
router = APIRouter(prefix="/analytics", tags=["analytics"])
|
|
|
|
DOW = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
|
|
|
|
|
|
def _aware(dt: datetime) -> datetime:
|
|
return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
|
|
|
|
|
|
def _range(days: int):
|
|
end = now_utc()
|
|
start = end - timedelta(days=days)
|
|
return start, end
|
|
|
|
|
|
def now_utc() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional
|
|
|
|
from typing import Optional
|
|
|
|
async def _scans(
|
|
org_id: str,
|
|
days: Optional[int] = None,
|
|
from_date: Optional[datetime] = None,
|
|
to_date: Optional[datetime] = None,
|
|
brand_id: Optional[str] = None,
|
|
):
|
|
items = await ScanLog.find(
|
|
ScanLog.org_id == org_id
|
|
).to_list()
|
|
|
|
if from_date:
|
|
from_date = _aware(from_date)
|
|
|
|
if to_date:
|
|
to_date = _aware(to_date) + timedelta(days=1)
|
|
|
|
if from_date and to_date:
|
|
items = [
|
|
s for s in items
|
|
if from_date <= _aware(s.scanned_at) < to_date
|
|
]
|
|
elif days:
|
|
start = now_utc() - timedelta(days=days)
|
|
items = [
|
|
s for s in items
|
|
if _aware(s.scanned_at) >= start
|
|
]
|
|
|
|
if brand_id:
|
|
products = await Product.find(
|
|
Product.org_id == org_id,
|
|
Product.brand_id == brand_id,
|
|
).to_list()
|
|
|
|
product_ids = {
|
|
str(p.id)
|
|
for p in products
|
|
}
|
|
|
|
items = [
|
|
s
|
|
for s in items
|
|
if s.product_id in product_ids
|
|
]
|
|
|
|
return items
|
|
|
|
|
|
@router.get("/summary")
|
|
async def summary(p: Principal = Depends(require_org), days: Optional[int] = None,
|
|
from_date: Optional[datetime] = None,
|
|
to_date: Optional[datetime] = None,brand_id: Optional[str] = None,):
|
|
oid = str(p.org.id)
|
|
scans = await _scans(oid, days,
|
|
from_date,
|
|
to_date,brand_id,
|
|
)
|
|
all_scans = await ScanLog.find(ScanLog.org_id == oid).to_list()
|
|
products = await Product.find(Product.org_id == oid).to_list()
|
|
today = now_utc().date()
|
|
unique = len({(s.ip, s.product_id) for s in scans})
|
|
avg_trust = round(sum(x.trust_score for x in products) / len(products)) if products else 0
|
|
return {
|
|
"total_scans": len(scans),
|
|
"all_time_scans": len(all_scans),
|
|
"unique_consumers": unique,
|
|
"today_scans": sum(1 for s in scans if _aware(s.scanned_at).date() == today),
|
|
"active_products": sum(1 for x in products if x.status == "active"),
|
|
"avg_trust_score": avg_trust,
|
|
"countries": len({s.country for s in scans if s.country}),
|
|
}
|
|
|
|
|
|
@router.get("/trend")
|
|
async def trend(
|
|
p: Principal = Depends(require_org),
|
|
days: Optional[int] = None,
|
|
from_date: Optional[datetime] = None,
|
|
to_date: Optional[datetime] = None,
|
|
brand_id: Optional[str] = None,
|
|
):
|
|
scans = await _scans(
|
|
str(p.org.id),
|
|
days,
|
|
from_date,
|
|
to_date,brand_id,
|
|
)
|
|
|
|
buckets = {}
|
|
|
|
for s in scans:
|
|
d = _aware(s.scanned_at).date()
|
|
buckets[d] = buckets.get(d, 0) + 1
|
|
|
|
return [
|
|
{"date": str(k), "scans": v}
|
|
for k, v in sorted(buckets.items())
|
|
]
|
|
|
|
|
|
@router.get("/top-products")
|
|
async def top_products(
|
|
p: Principal = Depends(require_org),
|
|
limit: int = 8,
|
|
days: Optional[int] = None,
|
|
from_date: Optional[datetime] = None,
|
|
to_date: Optional[datetime] = None,brand_id: Optional[str] = None,
|
|
):
|
|
scans = await _scans(
|
|
str(p.org.id),
|
|
days,
|
|
from_date,
|
|
to_date,brand_id,
|
|
)
|
|
|
|
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
|
|
}
|
|
|
|
return [
|
|
{
|
|
"name": lookup.get(pid, "Unknown Product"),
|
|
"scans": count,
|
|
}
|
|
for pid, count in counter.most_common(limit)
|
|
]
|
|
|
|
|
|
@router.get("/top-cities")
|
|
async def top_cities(p: Principal = Depends(require_org), limit: int = 10, days: Optional[int] = None,
|
|
from_date: Optional[datetime] = None,
|
|
to_date: Optional[datetime] = None,brand_id: Optional[str] = None,):
|
|
scans = await _scans(str(p.org.id), days,
|
|
from_date,
|
|
to_date,brand_id,
|
|
)
|
|
c = Counter(s.city for s in scans if s.city)
|
|
return [{"city": city, "scans": n} for city, n in c.most_common(limit)]
|
|
|
|
|
|
@router.get("/devices")
|
|
async def devices(
|
|
p: Principal = Depends(require_org),
|
|
days: Optional[int] = None,
|
|
from_date: Optional[datetime] = None,
|
|
to_date: Optional[datetime] = None,brand_id: Optional[str] = None,
|
|
):
|
|
scans = await _scans(
|
|
str(p.org.id),
|
|
days,
|
|
from_date,
|
|
to_date,brand_id,
|
|
)
|
|
|
|
c = Counter(s.device or "Other" for s in scans)
|
|
|
|
return [
|
|
{"device": d, "scans": n}
|
|
for d, n in c.most_common()
|
|
]
|
|
|
|
|
|
@router.get("/geography")
|
|
async def geography(p: Principal = Depends(require_org), days: Optional[int] = None,
|
|
from_date: Optional[datetime] = None,
|
|
to_date: Optional[datetime] = None,brand_id: Optional[str] = None,):
|
|
scans = await _scans(str(p.org.id), days,
|
|
from_date,
|
|
to_date,brand_id,
|
|
)
|
|
c = Counter(s.country for s in scans if s.country)
|
|
return [{"country": k, "scans": n} for k, n in c.most_common()]
|
|
|
|
@router.get("/time-distribution")
|
|
async def time_distribution(p: Principal = Depends(require_org), days: Optional[int] = None,
|
|
from_date: Optional[datetime] = None,
|
|
to_date: Optional[datetime] = None,brand_id: Optional[str] = None,):
|
|
"""Heatmap: day-of-week (0=Mon) x hour (0-23) scan counts."""
|
|
scans = await _scans(str(p.org.id), days,
|
|
from_date,
|
|
to_date,brand_id,
|
|
)
|
|
grid = defaultdict(int)
|
|
for s in scans:
|
|
dt = _aware(s.scanned_at)
|
|
grid[(dt.weekday(), dt.hour)] += 1
|
|
return [{"dow": d, "day": DOW[d], "hour": h, "count": grid.get((d, h), 0)}
|
|
for d in range(7) for h in range(24)]
|
|
|
|
|
|
@router.get("/recent-scans")
|
|
async def recent_scans(
|
|
p: Principal = Depends(require_org),
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
days: Optional[int] = None,
|
|
from_date: Optional[datetime] = None,
|
|
to_date: Optional[datetime] = None,
|
|
brand_id: Optional[str] = None,
|
|
):
|
|
scans = await _scans(
|
|
str(p.org.id),
|
|
days,
|
|
from_date,
|
|
to_date,
|
|
brand_id,
|
|
)
|
|
|
|
scans.sort(key=lambda s: s.scanned_at, reverse=True)
|
|
|
|
total = len(scans)
|
|
|
|
start = (page - 1) * page_size
|
|
end = start + page_size
|
|
|
|
rows = scans[start:end]
|
|
|
|
return {
|
|
"items": [
|
|
{
|
|
"at": s.scanned_at,
|
|
"product_id": s.product_id,
|
|
"city": s.city,
|
|
"country": s.country,
|
|
"device": s.device,
|
|
"code": s.qr_code,
|
|
"ip": s.ip,
|
|
}
|
|
for s in rows
|
|
],
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"total": total,
|
|
"pages": (total + page_size - 1) // page_size,
|
|
}
|
|
|
|
|
|
@router.get("/export")
|
|
async def export_csv(p: Principal = Depends(require_org), days: Optional[int] = None,
|
|
from_date: Optional[datetime] = None,
|
|
to_date: Optional[datetime] = None,brand_id: Optional[str] = None,):
|
|
scans = await _scans(str(p.org.id), days,
|
|
from_date,
|
|
to_date,brand_id,
|
|
)
|
|
scans.sort(key=lambda s: s.scanned_at, reverse=True)
|
|
buf = io.StringIO()
|
|
w = csv.writer(buf)
|
|
w.writerow(["Timestamp", "QR Code", "Product ID", "City", "Country", "Device", "IP"])
|
|
for s in scans:
|
|
w.writerow([_aware(s.scanned_at).isoformat(), s.qr_code, s.product_id,
|
|
s.city or "", s.country or "", s.device or "", s.ip or ""])
|
|
return Response(content=buf.getvalue(), media_type="text/csv",
|
|
headers={"Content-Disposition": 'attachment; filename="scans.csv"'})
|
|
|
|
from app.models import Brand
|
|
|
|
@router.get("/brands")
|
|
async def analytics_brands(
|
|
p: Principal = Depends(require_org),
|
|
):
|
|
brands = await Brand.find(
|
|
Brand.org_id == str(p.org.id)
|
|
).to_list()
|
|
|
|
return [
|
|
{
|
|
"id": str(b.id),
|
|
"name": b.name
|
|
}
|
|
for b in brands
|
|
] |