161 lines
7.5 KiB
Python
161 lines
7.5 KiB
Python
"""Seed demo data: plans, a Product Admin, a demo org with brand/product/batch/QR + scans.
|
|
|
|
Run: python -m app.db.seed
|
|
With mock DB it just verifies the pipeline; with a real Mongo it persists.
|
|
"""
|
|
import asyncio
|
|
import random
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from app.db.mongo import init_db
|
|
from app.core.security import hash_password
|
|
from app.models import (
|
|
User, Organization, OrgMember, Brand, Product, Batch, QRCode,
|
|
ScanLog, Plan, Subscription, StorageUsage, AuditLog,
|
|
)
|
|
from app.models.documents import (
|
|
ManufacturerInfo, ProductDetails, ComplianceBadge, QRStyle,
|
|
)
|
|
from app.models.enums import (
|
|
ProductAdminRole, ClientRole, ProductStatus, QRType, SubscriptionStatus,
|
|
)
|
|
from app.services import qr as qrsvc
|
|
|
|
PLANS = [
|
|
dict(name="Free", price=0, product_limit=1, user_limit=1, storage_gb=1,
|
|
features=["1 product", "1 user", "Unlimited QRs"]),
|
|
dict(name="Starter", price=999, product_limit=100, user_limit=3, storage_gb=10,
|
|
features=["100 products", "3 users", "Unlimited QRs", "Analytics"]),
|
|
dict(name="Growth", price=2999, product_limit=1000, user_limit=15, storage_gb=50,
|
|
features=["1,000 products", "15 users", "Unlimited QRs", "Advanced analytics"]),
|
|
dict(name="Business", price=5999, product_limit=None, user_limit=50, storage_gb=200,
|
|
features=["Unlimited products", "50 users", "Priority support"]),
|
|
dict(name="Enterprise", price=0, product_limit=None, user_limit=None, storage_gb=1000,
|
|
features=["Unlimited everything", "Custom pricing", "Dedicated support"]),
|
|
]
|
|
|
|
CITIES = ["Bangalore", "Mumbai", "Delhi", "Chennai", "Hyderabad"]
|
|
DEVICES = ["Android", "iOS", "Desktop", "Other"]
|
|
|
|
|
|
async def seed():
|
|
await init_db()
|
|
|
|
# Plans
|
|
plan_map = {}
|
|
for pdata in PLANS:
|
|
existing = await Plan.find_one(Plan.name == pdata["name"])
|
|
if existing:
|
|
plan_map[pdata["name"]] = existing
|
|
continue
|
|
pl = Plan(**pdata)
|
|
await pl.insert()
|
|
plan_map[pdata["name"]] = pl
|
|
|
|
# Product Admin (super admin)
|
|
if not await User.find_one(User.email == "admin@verifypack.example"):
|
|
await User(name="VerifyPack Admin", email="admin@verifypack.example",
|
|
hashed_password=hash_password("Admin@123"), email_verified=True,
|
|
is_product_admin=True,
|
|
product_admin_role=ProductAdminRole.super_admin).insert()
|
|
|
|
# Demo org owner
|
|
owner = await User.find_one(User.email == "owner@abcpharma.example")
|
|
if not owner:
|
|
owner = User(name="Admin", email="owner@abcpharma.example",
|
|
hashed_password=hash_password("Owner@123"), email_verified=True,
|
|
last_login=datetime.now(timezone.utc))
|
|
await owner.insert()
|
|
|
|
org = await Organization.find_one(Organization.slug == "abc-pharma")
|
|
if not org:
|
|
starter = plan_map["Starter"]
|
|
org = Organization(name="ABC Pharma Pvt Ltd", slug="abc-pharma",
|
|
gstin="29ABCDE1234F1Z5", plan_id=str(starter.id),
|
|
subscription_status=SubscriptionStatus.active,
|
|
user_count=3)
|
|
await org.insert()
|
|
await OrgMember(org_id=str(org.id), user_id=str(owner.id),
|
|
role=ClientRole.admin, status="active").insert()
|
|
await StorageUsage(org_id=str(org.id), used_bytes=int(12.4 * 1024 ** 3),
|
|
total_bytes=int(50 * 1024 ** 3)).insert()
|
|
await Subscription(org_id=str(org.id), plan_id=str(starter.id),
|
|
plan_name="Starter",
|
|
renewal_date=datetime.now(timezone.utc) + timedelta(days=24)).insert()
|
|
|
|
brand = Brand(org_id=str(org.id), name="ABC Healthcare")
|
|
await brand.insert()
|
|
|
|
sample = [
|
|
("Pain Relief 500mg", "PR500", "Pharma", 320),
|
|
("Vitamin C 1000mg", "VC1000", "Supplement", 210),
|
|
("Paracetamol 650mg", "PARA650", "Pharma", 180),
|
|
("Calcium Tablet", "CAL100", "Supplement", 120),
|
|
("Omega 3 Capsules", "OMG3", "Supplement", 90),
|
|
]
|
|
for name, sku, cat, scans in sample:
|
|
product = Product(
|
|
org_id=str(org.id), brand_id=str(brand.id), brand_name=brand.name,
|
|
name=name, sku=sku, category=cat, status=ProductStatus.active,
|
|
trust_score=random.randint(90, 98), scan_count=scans,
|
|
manufacturer=ManufacturerInfo(company="ABC Pharma Pvt Ltd",
|
|
country="India", plant="Bangalore Unit-1"),
|
|
details=ProductDetails(net_weight="100g", shelf_life="24 months",
|
|
storage_condition="Store below 25°C"),
|
|
compliance_badges=[ComplianceBadge(type="FSSAI", number="10012345"),
|
|
ComplianceBadge(type="GMP")],
|
|
)
|
|
await product.insert()
|
|
org.product_count += 1
|
|
|
|
batch = Batch(org_id=str(org.id), product_id=str(product.id),
|
|
batch_number=f"{sku}{datetime.now():%y%m%d}",
|
|
mfg_date=datetime.now(timezone.utc) - timedelta(days=30),
|
|
expiry_date=datetime.now(timezone.utc) + timedelta(days=700),
|
|
quantity=1, qr_type=QRType.batch)
|
|
await batch.insert()
|
|
product.batch_count = 1
|
|
|
|
code = qrsvc.new_code()
|
|
snapshot = {
|
|
"product": {"name": product.name, "sku": product.sku,
|
|
"trust_score": product.trust_score, "recalled": False,
|
|
"details": product.details.model_dump(),
|
|
"manufacturer": product.manufacturer.model_dump(),
|
|
"compliance": [c.model_dump() for c in product.compliance_badges]},
|
|
"brand": {"name": brand.name, "logo_url": None},
|
|
"batch": {"number": batch.batch_number,
|
|
"mfg_date": str(batch.mfg_date),
|
|
"expiry_date": str(batch.expiry_date)},
|
|
}
|
|
qr = QRCode(org_id=str(org.id), product_id=str(product.id),
|
|
batch_id=str(batch.id), code=code, qr_type=QRType.batch,
|
|
style=QRStyle(), snapshot=snapshot, scan_count=scans)
|
|
await qr.insert()
|
|
product.qr_count = 1
|
|
await product.save()
|
|
|
|
# scan logs spread over the past week
|
|
for _ in range(min(scans, 60)):
|
|
await ScanLog(org_id=str(org.id), qr_id=str(qr.id), qr_code=code,
|
|
product_id=str(product.id),
|
|
city=random.choice(CITIES), country="India",
|
|
device=random.choice(DEVICES),
|
|
scanned_at=datetime.now(timezone.utc)
|
|
- timedelta(days=random.randint(0, 6),
|
|
hours=random.randint(0, 23))).insert()
|
|
|
|
await org.save()
|
|
await AuditLog(org_id=str(org.id), user_id=str(owner.id), user_name="Admin",
|
|
action="product.create", module="products",
|
|
details="Pain Relief 500mg").insert()
|
|
print(f"Seeded org ABC Pharma with {org.product_count} products.")
|
|
print(f"Demo verify URL: {qrsvc.verify_url(code)}")
|
|
|
|
print("Login: owner@abcpharma.example / Owner@123 (client admin)")
|
|
print("Admin: admin@verifypack.example / Admin@123 (super admin)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(seed())
|