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 / 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" }, )