"""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())