475 lines
18 KiB
Python
475 lines
18 KiB
Python
import base64
|
|
import binascii
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Body, Depends, HTTPException, status, Response
|
|
|
|
from app.core.deps import require_org, require_client_roles, require_write_access, Principal
|
|
from app.models import Batch, Product, Brand, QRCode, Compliance, ScanLog,AuditLog
|
|
from app.models.documents import QRStyle
|
|
from app.models.enums import ClientRole, QRType, QRStatus
|
|
from app.schemas.common import QRGenerateIn
|
|
from app.services import qr as qrsvc
|
|
from app.services.storage import upload_file
|
|
|
|
router = APIRouter(prefix="/qr", tags=["qr"])
|
|
|
|
|
|
async def _safe_get(model, oid):
|
|
"""Beanie .get() that returns None for empty/invalid ObjectIds instead of raising."""
|
|
if not oid:
|
|
return None
|
|
try:
|
|
return await model.get(oid)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
async def _build_snapshot(product: Product, brand: Brand, batch: Batch) -> dict:
|
|
return {
|
|
"product": {
|
|
"name": product.name, "sku": product.sku, "category": product.category,
|
|
"description": product.description, "image_url": product.image_url,
|
|
"trust_score": product.trust_score, "recalled": product.recalled,
|
|
"details": product.details.model_dump(),
|
|
"manufacturer": product.manufacturer.model_dump(),
|
|
"compliance": [c.model_dump() for c in product.compliance_badges],
|
|
"recycling_info": product.recycling_info,
|
|
},
|
|
"brand": {"name": brand.name, "logo_url": brand.logo_url} if brand else None,
|
|
"batch": {"number": batch.batch_number, "mfg_date": str(batch.mfg_date),
|
|
"expiry_date": str(batch.expiry_date)},
|
|
}
|
|
|
|
|
|
@router.post("/generate", status_code=201)
|
|
async def generate(body: QRGenerateIn,
|
|
p: Principal = Depends(require_client_roles(ClientRole.admin, ClientRole.user))):
|
|
"""Generate QR(s) for a batch. No credit deduction — free and unlimited."""
|
|
require_write_access(p)
|
|
batch = await Batch.get(body.batch_id)
|
|
if not batch or batch.org_id != str(p.org.id):
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Batch not found")
|
|
product = await Product.get(batch.product_id)
|
|
brand = await Brand.get(product.brand_id) if product else None
|
|
style = body.style or QRStyle()
|
|
snapshot = await _build_snapshot(product, brand, batch)
|
|
|
|
created: list[QRCode] = []
|
|
if batch.qr_type == QRType.batch:
|
|
# Option 1 — one QR for the whole batch
|
|
code = qrsvc.new_code()
|
|
qr = QRCode(org_id=str(p.org.id), product_id=str(product.id), batch_id=str(batch.id),
|
|
code=code, qr_type=QRType.batch, style=style, snapshot=snapshot,
|
|
created_by=p.user.name)
|
|
await qr.insert()
|
|
created.append(qr)
|
|
else:
|
|
# Option 2 — serialized unique QR per pack
|
|
for i in range(1, batch.quantity + 1):
|
|
code = qrsvc.new_code()
|
|
qr = QRCode(org_id=str(p.org.id), product_id=str(product.id), batch_id=str(batch.id),
|
|
code=code, serial=i, qr_type=QRType.per_pack, style=style, snapshot=snapshot,
|
|
created_by=p.user.name)
|
|
await qr.insert()
|
|
created.append(qr)
|
|
|
|
product.qr_count += len(created)
|
|
await product.save()
|
|
await AuditLog(
|
|
org_id=str(p.org.id),
|
|
user_id=str(p.user.id),
|
|
user_name=p.user.name,
|
|
user_role=p.client_role,
|
|
action="qr.generate",
|
|
module="qr",
|
|
details=f"Generated {len(created)} QR(s) for batch {batch.batch_number}",
|
|
).insert()
|
|
return {
|
|
"batch_id": str(batch.id), "qr_type": batch.qr_type, "generated": len(created),
|
|
"codes": [{"id": str(q.id), "code": q.code, "serial": q.serial,
|
|
"url": qrsvc.verify_url(q.code)} for q in created],
|
|
"download_zip": batch.qr_type == QRType.per_pack,
|
|
}
|
|
|
|
|
|
|
|
@router.get("/summary")
|
|
async def qr_summary(p: Principal = Depends(require_org)):
|
|
from datetime import datetime, timezone
|
|
qrs = await QRCode.find(QRCode.org_id == str(p.org.id)).to_list()
|
|
scans = await ScanLog.find(ScanLog.org_id == str(p.org.id)).to_list()
|
|
total = len(qrs)
|
|
active = sum(1 for q in qrs if q.status == QRStatus.active)
|
|
inactive = total - active
|
|
today = datetime.now(timezone.utc).date()
|
|
today_scans = sum(1 for s in scans
|
|
if (s.scanned_at if s.scanned_at.tzinfo else s.scanned_at.replace(tzinfo=timezone.utc)).date() == today)
|
|
return {
|
|
"total": total, "active": active, "inactive": inactive,
|
|
"total_scans": sum(q.scan_count for q in qrs),
|
|
"today_scans": today_scans,
|
|
"active_pct": round(active / total * 100) if total else 0,
|
|
"inactive_pct": round(inactive / total * 100) if total else 0,
|
|
}
|
|
|
|
from datetime import datetime, timedelta
|
|
@router.get("")
|
|
async def list_qr(p: Principal = Depends(require_org), product_id: Optional[str] = None,
|
|
batch_id: Optional[str] = None, status_filter: Optional[str] = None,
|
|
brand_id: Optional[str] = None, qr_type: Optional[str] = None,
|
|
search: Optional[str] = None,start_date: Optional[str] = None,
|
|
end_date: Optional[str] = None):
|
|
qrs = await QRCode.find(QRCode.org_id == str(p.org.id)).to_list()
|
|
# join product + batch info for display
|
|
products = {str(x.id): x for x in await Product.find(Product.org_id == str(p.org.id)).to_list()}
|
|
batches = {str(b.id): b for b in await Batch.find(Batch.org_id == str(p.org.id)).to_list()}
|
|
|
|
rows = []
|
|
for q in qrs:
|
|
prod = products.get(q.product_id)
|
|
batch = batches.get(q.batch_id)
|
|
brand = await _safe_get(Brand, prod.brand_id) if prod else None
|
|
if product_id and q.product_id != product_id:
|
|
continue
|
|
if batch_id and q.batch_id != batch_id:
|
|
continue
|
|
if status_filter and q.status != status_filter:
|
|
continue
|
|
if qr_type and q.qr_type != qr_type:
|
|
continue
|
|
if start_date:
|
|
start = datetime.fromisoformat(start_date)
|
|
if q.created_at.replace(tzinfo=None) < start:
|
|
continue
|
|
if end_date:
|
|
end = datetime.fromisoformat(end_date) + timedelta(days=1)
|
|
if q.created_at.replace(tzinfo=None) >= end:
|
|
continue
|
|
|
|
|
|
if brand_id and (not prod or prod.brand_id != brand_id):
|
|
continue
|
|
if search:
|
|
s = search.lower()
|
|
hay = f"{q.code} {prod.name if prod else ''} {prod.sku if prod else ''} {batch.batch_number if batch else ''}".lower()
|
|
if s not in hay:
|
|
continue
|
|
rows.append({
|
|
"id": str(q.id), "code": q.code, "serial": q.serial, "qr_type": q.qr_type,
|
|
"status": q.status, "scan_count": q.scan_count, "last_scan_at": q.last_scan_at,
|
|
"last_scan_location": q.last_scan_location,
|
|
"product_id": q.product_id, "product_name": prod.name if prod else None,
|
|
"sku": prod.sku if prod else None, "brand_id": prod.brand_id if prod else None,"brand_name": brand.name if brand else None,
|
|
"batch_id": q.batch_id, "batch_number": batch.batch_number if batch else None,
|
|
"image_url": q.image_url, "url": qrsvc.verify_url(q.code), "created_at": q.created_at,"created_by": q.created_by,
|
|
})
|
|
rows.sort(key=lambda r: r["created_at"], reverse=True)
|
|
return rows
|
|
|
|
|
|
@router.post("/bulk")
|
|
async def bulk_action(body: dict = Body(...),
|
|
p: Principal = Depends(require_client_roles(ClientRole.admin, ClientRole.user))):
|
|
"""body = {ids: [...], action: 'disable'|'enable'|'archive'|'delete'}"""
|
|
ids = body.get("ids", [])
|
|
action = body.get("action")
|
|
n = 0
|
|
for qid in ids:
|
|
qr = await QRCode.get(qid)
|
|
if not qr or qr.org_id != str(p.org.id):
|
|
continue
|
|
if action == "disable":
|
|
qr.status = QRStatus.disabled
|
|
elif action == "enable":
|
|
qr.status = QRStatus.active
|
|
elif action == "archive":
|
|
qr.status = QRStatus.inactive
|
|
elif action == "delete":
|
|
await qr.delete(); n += 1; continue
|
|
else:
|
|
continue
|
|
await qr.save(); n += 1
|
|
return {"affected": n, "action": action}
|
|
|
|
|
|
async def _get_owned(qr_id: str, p: Principal) -> QRCode:
|
|
qr = await QRCode.get(qr_id)
|
|
if not qr or qr.org_id != str(p.org.id):
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "QR not found")
|
|
return qr
|
|
|
|
|
|
@router.get("/{qr_id}")
|
|
async def get_qr(qr_id: str, p: Principal = Depends(require_org)):
|
|
qr = await _get_owned(qr_id, p)
|
|
return {"id": str(qr.id), "code": qr.code, "serial": qr.serial, "qr_type": qr.qr_type,
|
|
"status": qr.status, "scan_count": qr.scan_count, "style": qr.style.model_dump(),
|
|
"url": qrsvc.verify_url(qr.code), "snapshot": qr.snapshot}
|
|
|
|
|
|
@router.get("/{qr_id}/detail")
|
|
async def qr_detail(qr_id: str, p: Principal = Depends(require_org)):
|
|
"""Rich detail for the QR Details drawer: overview + scan stats + history."""
|
|
from datetime import datetime, timezone, timedelta
|
|
qr = await _get_owned(qr_id, p)
|
|
prod = await _safe_get(Product, qr.product_id)
|
|
batch = await _safe_get(Batch, qr.batch_id)
|
|
scans = await ScanLog.find(ScanLog.qr_id == str(qr.id)).to_list()
|
|
|
|
def aware(dt):
|
|
return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
|
|
now_ = datetime.now(timezone.utc)
|
|
today = now_.date()
|
|
week_ago = now_ - timedelta(days=7)
|
|
scans.sort(key=lambda s: s.scanned_at, reverse=True)
|
|
|
|
serial_suffix = f"-{qr.serial:04d}" if qr.serial else "-0001"
|
|
qr_id_label = f"QR-{batch.batch_number}{serial_suffix}" if batch else f"QR-{qr.code}"
|
|
return {
|
|
"id": str(qr.id), "code": qr.code, "qr_id_label": qr_id_label,
|
|
"status": qr.status, "qr_type": qr.qr_type,
|
|
"short_url": qrsvc.verify_url(qr.code),
|
|
"image_url": qr.image_url,
|
|
"overview": {
|
|
"product": prod.name if prod else None,
|
|
"sku": prod.sku if prod else None,
|
|
"batch_number": batch.batch_number if batch else None,
|
|
"created_by": qr.created_by,
|
|
"created_at": qr.created_at,
|
|
"redirect_url": qrsvc.verify_url(qr.code),
|
|
"verification_enabled": qr.status == QRStatus.active,
|
|
"expiry": "No Expiry",
|
|
"remarks": qr.remarks,
|
|
},
|
|
"stats": {
|
|
"total": qr.scan_count,
|
|
"today": sum(1 for s in scans if aware(s.scanned_at).date() == today),
|
|
"week": sum(1 for s in scans if aware(s.scanned_at) >= week_ago),
|
|
},
|
|
"history": [{"at": s.scanned_at, "city": s.city, "country": s.country,
|
|
"device": s.device} for s in scans[:25]],
|
|
}
|
|
|
|
|
|
@router.post("/{qr_id}/image")
|
|
async def upload_qr_image(qr_id: str,
|
|
payload: dict = Body(...),
|
|
p: Principal = Depends(require_org)):
|
|
"""Store the client-rendered WYSIWYG image (data URL) for a QR code.
|
|
|
|
payload = {"data_url": "data:image/png;base64,....", "format": "png"|"svg"}
|
|
"""
|
|
qr = await _get_owned(qr_id, p)
|
|
data_url = payload.get("data_url", "")
|
|
fmt = payload.get("format", "png")
|
|
if "," not in data_url:
|
|
raise HTTPException(status.HTTP_400_BAD_REQUEST, "Invalid data URL")
|
|
b64 = data_url.split(",", 1)[1]
|
|
try:
|
|
content = base64.b64decode(b64)
|
|
except (binascii.Error, ValueError):
|
|
raise HTTPException(status.HTTP_400_BAD_REQUEST, "Bad base64")
|
|
mime = "image/svg+xml" if fmt == "svg" else "image/png"
|
|
url = await upload_file(content, f"{qr.code}.{fmt}", mime)
|
|
qr.image_url = url
|
|
await qr.save()
|
|
return {"image_url": url}
|
|
|
|
|
|
@router.get("/{qr_id}/download")
|
|
async def download_qr(qr_id: str, format: str = "png"):
|
|
# Public: the rendered QR only encodes the public verify URL — no private data.
|
|
# This lets <a>/<img> links (which can't send the Bearer header) work directly.
|
|
qr = await QRCode.get(qr_id)
|
|
if not qr:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "QR not found")
|
|
data, mime = qrsvc.render(qr.code, qr.style, format)
|
|
ext = {"image/svg+xml": "svg", "application/pdf": "pdf"}.get(mime, "png")
|
|
return Response(content=data, media_type=mime,
|
|
headers={"Content-Disposition": f'attachment; filename="{qr.code}.{ext}"'})
|
|
|
|
|
|
@router.get("/batch/{batch_id}/download-zip")
|
|
async def download_zip(batch_id: str, format: str = "png"):
|
|
batch = await Batch.get(batch_id)
|
|
if not batch:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Batch not found")
|
|
qrs = await QRCode.find(QRCode.batch_id == batch_id).to_list()
|
|
if not qrs:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "No QR codes for this batch")
|
|
style = qrs[0].style
|
|
data = qrsvc.zip_codes([q.code for q in qrs], style, format)
|
|
return Response(content=data, media_type="application/zip",
|
|
headers={"Content-Disposition": f'attachment; filename="batch_{batch.batch_number}.zip"'})
|
|
|
|
|
|
@router.put("/{qr_id}/disable")
|
|
async def disable_qr(qr_id: str, p: Principal = Depends(require_client_roles(ClientRole.admin, ClientRole.user))):
|
|
qr = await _get_owned(qr_id, p)
|
|
qr.status = QRStatus.disabled
|
|
await qr.save()
|
|
await AuditLog(
|
|
org_id=str(p.org.id),
|
|
user_id=str(p.user.id),
|
|
user_name=p.user.name,
|
|
user_role=p.client_role,
|
|
action="qr.disable",
|
|
module="qr",
|
|
details=f"Disabled QR {qr.code}",
|
|
).insert()
|
|
return {"message": "disabled"}
|
|
|
|
|
|
@router.post("/{qr_id}/regenerate")
|
|
async def regenerate_qr(qr_id: str, p: Principal = Depends(require_client_roles(ClientRole.admin, ClientRole.user))):
|
|
"""Regenerate the code (free, no cost). Keeps style; old code stops resolving."""
|
|
qr = await _get_owned(qr_id, p)
|
|
qr.code = qrsvc.new_code()
|
|
qr.status = QRStatus.active
|
|
await qr.save()
|
|
await AuditLog(
|
|
org_id=str(p.org.id),
|
|
user_id=str(p.user.id),
|
|
user_name=p.user.name,
|
|
user_role=p.client_role,
|
|
action="qr.regenerate",
|
|
module="qr",
|
|
details=f"Regenerated QR {qr.code}",
|
|
).insert()
|
|
return {"id": str(qr.id), "code": qr.code, "url": qrsvc.verify_url(qr.code)}
|
|
|
|
from fastapi.responses import StreamingResponse
|
|
from io import BytesIO
|
|
from openpyxl import Workbook
|
|
from datetime import datetime, timedelta
|
|
|
|
|
|
@router.get("/export")
|
|
async def export_qr(
|
|
p: Principal = Depends(require_org),
|
|
product_id: Optional[str] = None,
|
|
batch_id: Optional[str] = None,
|
|
brand_id: Optional[str] = None,
|
|
qr_type: Optional[str] = None,
|
|
status_filter: Optional[str] = None,
|
|
search: Optional[str] = None,
|
|
start_date: Optional[str] = None,
|
|
end_date: Optional[str] = None,
|
|
):
|
|
|
|
qrs = await QRCode.find(QRCode.org_id == str(p.org.id)).to_list()
|
|
|
|
products = {
|
|
str(x.id): x
|
|
for x in await Product.find(Product.org_id == str(p.org.id)).to_list()
|
|
}
|
|
|
|
brands = {
|
|
str(x.id): x
|
|
for x in await Brand.find(Brand.org_id == str(p.org.id)).to_list()
|
|
}
|
|
|
|
batches = {
|
|
str(x.id): x
|
|
for x in await Batch.find(Batch.org_id == str(p.org.id)).to_list()
|
|
}
|
|
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "QR Codes"
|
|
|
|
ws.append([
|
|
"QR Code",
|
|
"Status",
|
|
"QR Type",
|
|
"Serial",
|
|
"Product",
|
|
"SKU",
|
|
"Brand",
|
|
"Category",
|
|
"Batch No",
|
|
"Quantity",
|
|
"Manufacture Date",
|
|
"Expiry Date",
|
|
"Scan Count",
|
|
"Last Scan",
|
|
"Last Scan Location",
|
|
"Created By",
|
|
"Created At",
|
|
"Verification URL",
|
|
])
|
|
|
|
for q in qrs:
|
|
|
|
product = products.get(q.product_id)
|
|
batch = batches.get(q.batch_id)
|
|
brand = brands.get(product.brand_id) if product else None
|
|
|
|
if product_id and q.product_id != product_id:
|
|
continue
|
|
|
|
if batch_id and q.batch_id != batch_id:
|
|
continue
|
|
|
|
if status_filter and q.status != status_filter:
|
|
continue
|
|
|
|
if qr_type and q.qr_type != qr_type:
|
|
continue
|
|
|
|
if brand_id and (not product or product.brand_id != brand_id):
|
|
continue
|
|
|
|
if search:
|
|
s = search.lower()
|
|
text = f"{q.code} {product.name if product else ''} {product.sku if product else ''} {batch.batch_number if batch else ''}".lower()
|
|
|
|
if s not in text:
|
|
continue
|
|
|
|
if start_date:
|
|
start = datetime.fromisoformat(start_date)
|
|
if q.created_at.replace(tzinfo=None) < start:
|
|
continue
|
|
|
|
if end_date:
|
|
end = datetime.fromisoformat(end_date) + timedelta(days=1)
|
|
if q.created_at.replace(tzinfo=None) >= end:
|
|
continue
|
|
|
|
ws.append([
|
|
q.code,
|
|
q.status,
|
|
q.qr_type,
|
|
q.serial,
|
|
product.name if product else "",
|
|
product.sku if product else "",
|
|
brand.name if brand else "",
|
|
product.category if product else "",
|
|
batch.batch_number if batch else "",
|
|
batch.quantity if batch else "",
|
|
str(batch.mfg_date) if batch else "",
|
|
str(batch.expiry_date) if batch else "",
|
|
q.scan_count,
|
|
str(q.last_scan_at) if q.last_scan_at else "",
|
|
q.last_scan_location,
|
|
q.created_by,
|
|
str(q.created_at),
|
|
qrsvc.verify_url(q.code),
|
|
|
|
])
|
|
|
|
stream = BytesIO()
|
|
wb.save(stream)
|
|
stream.seek(0)
|
|
|
|
return StreamingResponse(
|
|
stream,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={
|
|
"Content-Disposition": "attachment; filename=QR_Report.xlsx"
|
|
},
|
|
)
|
|
|