402 lines
11 KiB
Python
402 lines
11 KiB
Python
"""Beanie document models (MongoDB collections).
|
|
|
|
Design note: QRCode embeds a denormalized snapshot of product/brand/compliance so the
|
|
public verification page (`GET /verify/{qr_code}`) resolves in a single document read.
|
|
"""
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
from beanie import Document, Indexed
|
|
from pydantic import BaseModel, Field, EmailStr
|
|
|
|
from app.models.enums import (
|
|
ProductAdminRole, ClientRole, SubscriptionStatus, ProductStatus,
|
|
QRType, QRStatus, ComplianceStatus, PaymentStatus,
|
|
)
|
|
|
|
|
|
def now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
# ---------- embedded sub-documents ----------
|
|
|
|
class ManufacturerInfo(BaseModel):
|
|
company: Optional[str] = None
|
|
plant: Optional[str] = None
|
|
country: Optional[str] = None
|
|
website: Optional[str] = None
|
|
email: Optional[str] = None
|
|
phone: Optional[str] = None
|
|
license_number: Optional[str] = None
|
|
location: Optional[str] = None
|
|
|
|
|
|
class ProductDetails(BaseModel):
|
|
net_weight: Optional[str] = None
|
|
pack_size: Optional[str] = None
|
|
unit_type: Optional[str] = None
|
|
form_type: Optional[str] = None
|
|
storage_condition: Optional[str] = None
|
|
shelf_life: Optional[str] = None
|
|
temperature_range: Optional[str] = None
|
|
dosage: Optional[str] = None
|
|
gtin: Optional[str] = None
|
|
|
|
|
|
class ComplianceBadge(BaseModel):
|
|
type: str # FSSAI, EPR, GMP, ISO, ...
|
|
number: Optional[str] = None
|
|
status: ComplianceStatus = ComplianceStatus.complete
|
|
|
|
|
|
class QRStyle(BaseModel):
|
|
dot_shape: str = "square" # square|rounded|dots|classy|classy-rounded|extra-rounded
|
|
corner_square_shape: str = "square" # square|dot|extra-rounded
|
|
corner_dot_shape: str = "square" # square|dot
|
|
eye_shape: str = "square" # legacy alias, kept for compatibility
|
|
fg_color: str = "#000000"
|
|
bg_color: str = "#ffffff"
|
|
transparent_bg: bool = False
|
|
gradient: bool = False
|
|
gradient_color: Optional[str] = None
|
|
logo_url: Optional[str] = None
|
|
logo_size: float = 0.4
|
|
error_correction: str = "M" # L|M|Q|H (auto H when logo set)
|
|
quiet_zone: int = 4
|
|
|
|
|
|
# ---------- top-level documents ----------
|
|
|
|
class User(Document):
|
|
email: Indexed(EmailStr, unique=True)
|
|
name: str
|
|
hashed_password: Optional[str] = None # null for pure OAuth users
|
|
google_id: Optional[str] = None
|
|
avatar_url: Optional[str] = None
|
|
email_verified: bool = False
|
|
is_product_admin: bool = False
|
|
product_admin_role: Optional[ProductAdminRole] = None
|
|
disabled: bool = False
|
|
last_login: Optional[datetime] = None
|
|
verify_code: Optional[str] = None # 6-digit email verification code
|
|
verify_code_expires: Optional[datetime] = None
|
|
created_at: datetime = Field(default_factory=now)
|
|
login_otp: Optional[str] = None
|
|
login_otp_expires: Optional[datetime] = None
|
|
|
|
class Settings:
|
|
name = "users"
|
|
|
|
|
|
class Organization(Document):
|
|
name: str
|
|
slug: Indexed(str, unique=True)
|
|
logo_url: Optional[str] = None
|
|
gstin: Optional[str] = None
|
|
pan: Optional[str] = None
|
|
address: Optional[str] = None
|
|
contact_email: Optional[str] = None
|
|
phone: Optional[str] = None
|
|
website: Optional[str] = None
|
|
timezone: str = "Asia/Kolkata"
|
|
language: str = "en"
|
|
plan_id: Optional[str] = None
|
|
subscription_status: SubscriptionStatus = SubscriptionStatus.active
|
|
product_count: int = 0
|
|
user_count: int = 1
|
|
storage_bytes: int = 0
|
|
country: Optional[str] = None
|
|
created_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "organizations"
|
|
|
|
|
|
class OrgMember(Document):
|
|
org_id: Indexed(str)
|
|
user_id: Indexed(str)
|
|
role: ClientRole = ClientRole.user
|
|
department: Optional[str] = None
|
|
status: str = "active" # active|pending|disabled
|
|
invited_email: Optional[str] = None
|
|
joined_at: Optional[datetime] = None
|
|
created_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "org_members"
|
|
|
|
|
|
class Brand(Document):
|
|
org_id: Indexed(str)
|
|
name: str
|
|
logo_url: Optional[str] = None
|
|
description: Optional[str] = None
|
|
product_count: int = 0
|
|
created_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "brands"
|
|
|
|
|
|
class Product(Document):
|
|
org_id: Indexed(str)
|
|
brand_id: Indexed(str)
|
|
brand_name: Optional[str] = None
|
|
name: str
|
|
sku: Optional[str] = None
|
|
category: Optional[str] = None
|
|
description: Optional[str] = None
|
|
image_url: Optional[str] = None
|
|
manufacturer: ManufacturerInfo = Field(default_factory=ManufacturerInfo)
|
|
details: ProductDetails = Field(default_factory=ProductDetails)
|
|
compliance_badges: list[ComplianceBadge] = []
|
|
recycling_info: Optional[str] = None
|
|
internal_notes: Optional[str] = None
|
|
qr_type: str = "dynamic" # dynamic | static
|
|
display_brand_logo: bool = True
|
|
trust_score: int = 90
|
|
status: ProductStatus = ProductStatus.draft
|
|
recalled: bool = False
|
|
batch_count: int = 0
|
|
qr_count: int = 0
|
|
scan_count: int = 0
|
|
created_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "products"
|
|
|
|
|
|
class Batch(Document):
|
|
org_id: Indexed(str)
|
|
product_id: Indexed(str)
|
|
batch_number: str
|
|
mfg_date: Optional[datetime] = None
|
|
expiry_date: Optional[datetime] = None
|
|
quantity: int = 1
|
|
qr_type: QRType = QRType.batch
|
|
notes: Optional[str] = None
|
|
created_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "batches"
|
|
|
|
|
|
class QRCode(Document):
|
|
org_id: Indexed(str)
|
|
product_id: Indexed(str)
|
|
batch_id: Indexed(str)
|
|
code: Indexed(str, unique=True) # short slug used in /v/{code}
|
|
serial: Optional[int] = None # per-pack index for Option 2
|
|
qr_type: QRType = QRType.batch
|
|
status: QRStatus = QRStatus.active
|
|
style: QRStyle = Field(default_factory=QRStyle)
|
|
image_url: Optional[str] = None # WYSIWYG client-rendered image (S3/local)
|
|
created_by: Optional[str] = None # name of the user who generated it
|
|
remarks: Optional[str] = None
|
|
scan_count: int = 0
|
|
last_scan_at: Optional[datetime] = None
|
|
last_scan_location: Optional[str] = None
|
|
# denormalized snapshot for single-read verification
|
|
snapshot: dict = {}
|
|
created_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "qr_codes"
|
|
|
|
|
|
class QRTemplate(Document):
|
|
org_id: Indexed(str)
|
|
name: str
|
|
style: QRStyle = Field(default_factory=QRStyle)
|
|
created_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "qr_templates"
|
|
|
|
|
|
class ScanLog(Document):
|
|
org_id: Indexed(str)
|
|
qr_id: Indexed(str)
|
|
qr_code: str
|
|
product_id: str
|
|
city: Optional[str] = None
|
|
country: Optional[str] = None
|
|
device: Optional[str] = None # Android|iOS|Desktop|Other
|
|
browser: Optional[str] = None
|
|
os: Optional[str] = None
|
|
ip: Optional[str] = None
|
|
scanned_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "scan_logs"
|
|
|
|
|
|
class Compliance(Document):
|
|
org_id: Indexed(str)
|
|
product_id: Optional[str] = None
|
|
compliance_type: str
|
|
license_number: Optional[str] = None
|
|
issuing_authority: Optional[str] = None
|
|
issue_date: Optional[datetime] = None
|
|
expiry_date: Optional[datetime] = None
|
|
country: Optional[str] = None
|
|
state: Optional[str] = None
|
|
plant: Optional[str] = None
|
|
license_name: Optional[str] = None
|
|
category: Optional[str] = None
|
|
applicable_to: Optional[str] = None
|
|
remarks: Optional[str] = None
|
|
document_url: Optional[str] = None
|
|
documents: list = [] # [{name, url, uploaded_at}]
|
|
created_by: Optional[str] = None
|
|
status: ComplianceStatus = ComplianceStatus.pending
|
|
created_at: datetime = Field(default_factory=now)
|
|
updated_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "compliance"
|
|
|
|
|
|
class Plan(Document):
|
|
name: Indexed(str, unique=True)
|
|
price: int # INR / month
|
|
product_limit: Optional[int] = None # None = unlimited
|
|
user_limit: Optional[int] = None
|
|
storage_gb: int = 5
|
|
grace_days: int = 30
|
|
archive_years: int = 1
|
|
features: list[str] = []
|
|
enabled: bool = True
|
|
created_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "plans"
|
|
|
|
|
|
class Subscription(Document):
|
|
org_id: Indexed(str)
|
|
plan_id: str
|
|
plan_name: str
|
|
status: SubscriptionStatus = SubscriptionStatus.active
|
|
auto_renew: bool = True
|
|
start_date: datetime = Field(default_factory=now)
|
|
renewal_date: Optional[datetime] = None
|
|
grace_end: Optional[datetime] = None
|
|
archive_until: Optional[datetime] = None
|
|
payment_status: PaymentStatus = PaymentStatus.paid
|
|
created_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "subscriptions"
|
|
|
|
|
|
class PlanHistory(Document):
|
|
org_id: Indexed(str)
|
|
from_plan: Optional[str] = None
|
|
to_plan: str
|
|
changed_by: Optional[str] = None
|
|
reason: Optional[str] = None
|
|
created_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "plan_history"
|
|
|
|
|
|
class Invoice(Document):
|
|
org_id: Indexed(str)
|
|
invoice_number: Indexed(str, unique=True)
|
|
plan_name: str
|
|
period: Optional[str] = None
|
|
subtotal: int = 0
|
|
gst_amount: int = 0
|
|
total: int = 0
|
|
status: PaymentStatus = PaymentStatus.pending
|
|
payment_method: Optional[str] = None
|
|
transaction_id: Optional[str] = None
|
|
gstin: Optional[str] = None
|
|
place_of_supply: Optional[str] = None
|
|
created_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "invoices"
|
|
|
|
|
|
class Payment(Document):
|
|
org_id: Indexed(str)
|
|
invoice_id: Optional[str] = None
|
|
razorpay_order_id: Optional[str] = None
|
|
razorpay_payment_id: Optional[str] = None
|
|
amount: int = 0
|
|
status: PaymentStatus = PaymentStatus.pending
|
|
created_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "payments"
|
|
|
|
|
|
class StorageUsage(Document):
|
|
org_id: Indexed(str, unique=True)
|
|
used_bytes: int = 0
|
|
total_bytes: int = 5 * 1024 ** 3
|
|
updated_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "storage_usage"
|
|
|
|
|
|
class AuditLog(Document):
|
|
org_id: Optional[str] = None
|
|
user_id: Optional[str] = None
|
|
user_name: Optional[str] = None
|
|
user_role: Optional[str] = None
|
|
action: str
|
|
module: Optional[str] = None
|
|
details: Optional[str] = None
|
|
ip: Optional[str] = None
|
|
user_agent: Optional[str] = None
|
|
impersonator_id: Optional[str] = None
|
|
created_at: datetime = Field(default_factory=now)
|
|
|
|
|
|
class Settings:
|
|
name = "audit_logs"
|
|
|
|
|
|
class SupportNote(Document):
|
|
org_id: Indexed(str)
|
|
author_id: Optional[str] = None
|
|
author_name: Optional[str] = None
|
|
note: str
|
|
flag: Optional[str] = None
|
|
created_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "support_notes"
|
|
|
|
|
|
class Alert(Document):
|
|
org_id: Indexed(str)
|
|
type: str
|
|
message: str
|
|
severity: str = "info" # info|warning|critical
|
|
resolved: bool = False
|
|
created_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "alerts"
|
|
|
|
|
|
class EmailLog(Document):
|
|
to_email: str
|
|
subject: str
|
|
template: Optional[str] = None
|
|
status: str = "sent"
|
|
error: Optional[str] = None
|
|
created_at: datetime = Field(default_factory=now)
|
|
|
|
class Settings:
|
|
name = "email_logs"
|
|
|
|
|