"""Seed the procurement DB with realistic demo data (v1.3). Wipes the app tables and inserts a clean, relational dataset: 8 topic categories (already bootstrapped at API startup), base units, realistic brands/materials/suppliers, 3 projects with 2 warehouses each, 4 quotes covering every lifecycle state (offered / accepted-in-transit / partially-received / arrived-completed) with termsheets, decisions, partial receipts + warehouse allocations, and matching inventory rows. Run AFTER the stack is up (api image has this script + the ORM): docker compose run --rm api python tools/seed.py User roles are linked to the Casdoor users by username (razaei = admin, mousavi = مدیر تدارکات, mohaghegh/tester = مدیر پروژه). The Casdoor org must be seeded first (python backend/tools/seed_casdoor.py); if a user is missing there, the row is created with a pending casdoor_id and the role can be assigned later from the Organization panel. The existing junk/test rows (random names) are replaced — the dumps in backups/ should be refreshed after running (see README). """ import asyncio import sys from datetime import date, datetime, timedelta, timezone from decimal import Decimal from sqlalchemy import select, text from app.db.engine import AsyncSessionLocal from app.db.init_db import BASE_UNITS, TOPIC_CATEGORIES from app.models.catalog import ( Brand, BrandMaterial, Material, MaterialCategory, Unit, ) from app.models.inventory import InventoryItem, InventoryTransaction, Warehouse from app.models.procurement import ( ItemDecision, ItemReceipt, ItemReceiptAllocation, ItemTermsheet, LineItem, LineItemProjects, Project, Quote, QuoteEvent, QuoteProjects, Supplier, SupplierMaterial, Termsheet, TermsheetScheduleStep, User, ) NOW = datetime.now(timezone.utc) def d(days_ago: int) -> datetime: return NOW - timedelta(days=days_ago) # --------------------------------------------------------------------------- # Casdoor user ids (username -> id) — used to bind roles to real identities # --------------------------------------------------------------------------- async def casdoor_user_ids() -> dict[str, str]: import httpx from app.config import settings try: async with httpx.AsyncClient(timeout=15) as client: login = await client.post( f"{settings.CASDOOR_ENDPOINT}/api/login", json={ "type": "login", "organization": "built-in", "username": settings.CASDOOR_ADMIN_USER, "password": settings.CASDOOR_ADMIN_PASSWORD, "application": "app-built-in", }, ) if login.status_code != 200 or (login.json() or {}).get("status") != "ok": print("!! Casdoor admin login failed — roles will use pending ids") return {} resp = await client.get( f"{settings.CASDOOR_ENDPOINT}/api/get-users?owner={settings.CASDOOR_ORGANIZATION}" ) body = resp.json() if body.get("status") != "ok": return {} return {str(u.get("name")): str(u.get("id")) for u in (body.get("data") or []) if u.get("id")} except Exception as exc: # noqa: BLE001 — best effort print(f"!! Casdoor unreachable ({exc}) — roles will use pending ids") return {} # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- async def get_or_create_unit(db, code: str, name_fa: str, dimension: str, si_factor=1, is_base=False) -> int: unit = await db.scalar(select(Unit).where(Unit.code == code)) if unit: return unit.id unit = Unit(code=code, name_fa=name_fa, dimension=dimension, si_factor=si_factor, is_base=is_base) db.add(unit) await db.flush() return unit.id async def get_category(db, slug: str) -> MaterialCategory: cat = await db.scalar(select(MaterialCategory).where(MaterialCategory.slug == slug)) if cat: return cat meta = next(c for c in TOPIC_CATEGORIES if c["slug"] == slug) cat = MaterialCategory(slug=slug, name=meta["name"]) db.add(cat) await db.flush() return cat # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- USER_PLAN = [ # (casdoor username, first, last, role) ("razaei", "علی", "رضایی", "admin"), ("mousavi", "حسین", "موسوی", "procurement_manager"), ("mohaghegh", "محمد", "محقق", "pm"), ("tester", "کاربر", "تست", "pm"), ] MATERIALS = [ # (name, category_slug, unit_code, brand_name) ("کابل مسی افشان ۲×۲٫۵", "electrical-equipment-supplies", "m", "کابل البرز"), ("کلید و پریز آرین مدل سفید", "electrical-equipment-supplies", "piece", None), ("میلگرد آجدار A3 سایز ۱۴", "metals-steel-sections", "ton", "ذوب آهن اصفهان"), ("تیرآهن IPE ۱۴", "metals-steel-sections", "ton", "ذوب آهن اصفهان"), ("سیمان تیپ ۲", "construction-materials", "bag", "سیمان تهران"), ("بلوک سیمانی ۲۰×۲۰×۴۰", "construction-materials", "piece", None), ("لوله پلی‌اتیلن PE100 سایز ۱۶", "mechanical-installations-pipes", "m", "پارس لوله"), ("لوله پلی‌اتیلن PE100 سایز ۲۵", "mechanical-installations-pipes", "m", "پارس لوله"), ("چسب بتن", "construction-chemicals", "kg", "شیمی ساختمان"), ("پنجره UPVC دو جداره", "doors-windows-glass", "piece", None), ("سرامیک پرسلان ۶۰×۶۰", "finishing-interior-decoration", "piece", None), ("ویبراتور بتن", "construction-tools-equipment", "piece", None), ] SUPPLIERS = [ # (name, category_slug, location, phone, contact_person, tax_id) ("شرکت صنایع کابل البرز", "electrical-equipment-supplies", "تهران — شهرک صنعتی", "۰۲۱-۳۳۹۹۱۲۳۴", "مهندس کریمی", "10102345678"), ("فولاد ناب آریا", "metals-steel-sections", "اصفهان — ذوب‌آهن", "۰۳۱-۳۴۵۶۷۸۹۰", "آقای صادقی", "10203456789"), ("پخش لوله و اتصالات پارس", "mechanical-installations-pipes", "تهران — بازار آهن", "۰۲۱-۵۵۹۹۸۷۶۵", "آقای نادری", "10304567890"), ("بازرگانی مصالح تهران", "construction-materials", "تهران — خیابان شوش", "۰۲۱-۶۶۷۷۸۸۹۹", "خانم احمدی", "10405678901"), ] PROJECTS = [ # (name, description, warehouse names) ("مجتمع مسکونی آرام", "ساخت ۲۴۰ واحد مسکونی — منطقه ۲۲ تهران", ["انبار مرکزی آرام", "انبار مصالح آرام"]), ("برج اداری سپهر", "برج اداری ۱۸ طبقه — سعادت‌آباد", ["انبار سپهر یک", "انبار سپهر دو"]), ("پروژه زیرساخت شرق تهران", "شبکه معابر و زیرساخت — منطقه ۱۳", ["انبار شرق اصلی", "انبار سیار شرق"]), ] # quote index -> (supplier idx, project idx, status, days since quote_date, # days since accepted, [(material idx, brand?, qty, unit_price), ...]) QUOTES = [ { "supplier": 0, "project": 0, "status": "offered", "quote_days": 12, "accepted_days": None, "items": [(0, 500, "120000"), (1, 300, "65000")], }, { "supplier": 1, "project": 0, "status": "accepted", "quote_days": 20, "accepted_days": 13, "items": [(2, 20, "38000000"), (3, 15, "52000000")], }, { "supplier": 2, "project": 1, "status": "accepted", "quote_days": 15, "accepted_days": 9, "items": [(6, 1000, "45000"), (7, 600, "145000")], # partial receipts: item0 1000/1000 (400→wh0, 600→wh1), item1 200/600 (→wh0) "receipts": [ {"item": 0, "qty": 400, "wh": 0, "days": 7}, {"item": 0, "qty": 600, "wh": 1, "days": 4}, {"item": 1, "qty": 200, "wh": 0, "days": 2}, ], }, { "supplier": 3, "project": 2, "status": "arrived", "quote_days": 30, "accepted_days": 24, "items": [(4, 400, "220000"), (5, 2000, "28000")], "receipts": [ {"item": 0, "qty": 400, "wh": 0, "days": 3}, {"item": 1, "qty": 2000, "wh": 1, "days": 3}, ], }, ] UNITS_BY_CODE = {u["code"]: u for u in BASE_UNITS} async def run() -> None: casdoor_ids = await casdoor_user_ids() async with AsyncSessionLocal() as db: # ---- wipe previous (junk) data, keep categories/units bootstrap ---- await db.execute(text( "TRUNCATE TABLE quote_attachments, quote_events, quote_projects, line_item_projects, " "item_receipt_allocations, item_receipts, inventory_transactions, inventory, " "item_decisions, item_termsheets, termsheet_schedule_steps, termsheets, line_item_specs, " "line_items, quotes, supplier_materials, suppliers, brand_materials, " "material_attribute_values, materials, brands, warehouses, projects, users, " "category_reference_units, unit_conversions, standard_conversions " "RESTART IDENTITY CASCADE" )) # ---- users (roles bound to Casdoor identities by username) ---- users: dict[str, User] = {} for name, first, last, role in USER_PLAN: cid = casdoor_ids.get(name) if not cid: print(f"!! Casdoor user '{name}' not found — created with pending id (run seed_casdoor.py)") user = User( casdoor_id=cid or f"pending-{name}", first_name=first, last_name=last, role=role, ) db.add(user) users[name] = user await db.flush() # ---- units (base set; fetch-or-create) ---- unit_ids: dict[str, int] = {} for meta in BASE_UNITS: unit_ids[meta["code"]] = await get_or_create_unit( db, meta["code"], meta["name_fa"], meta["dimension"], meta["si_factor"], meta["is_base"] ) # ---- brands ---- brand_ids: dict[str, object] = {} brand_names = sorted({b for _, _, _, b in MATERIALS if b}) for bname in brand_names: brand = Brand(name=bname, country="ایران") db.add(brand) await db.flush() brand_ids[bname] = brand.id # ---- materials (one per category at least) + brand links ---- material_by_name: dict[str, Material] = {} brand_material_by_material: dict[str, object] = {} for name, cat_slug, unit_code, brand_name in MATERIALS: cat = await get_category(db, cat_slug) material = Material( category_id=cat.id, name=name, canonical_unit_id=unit_ids[unit_code], status="active", ) db.add(material) await db.flush() material_by_name[name] = material if brand_name: bm = BrandMaterial(brand_id=brand_ids[brand_name], material_id=material.id, is_active=True) db.add(bm) await db.flush() brand_material_by_material[name] = bm.id # ---- suppliers + supplier-material links ---- suppliers: list[Supplier] = [] for name, cat_slug, location, phone, contact, tax in SUPPLIERS: cat = await get_category(db, cat_slug) sup = Supplier( name=name, category_id=cat.id, location=location, phone=phone, contact_person=contact, tax_id=tax, ) db.add(sup) await db.flush() suppliers.append(sup) # last-quoted prices: link each supplier to 2 materials of its category for sup in suppliers: cat_materials = [m for m in material_by_name.values() if m.category_id == sup.category_id] for m in cat_materials[:2]: db.add(SupplierMaterial( supplier_id=sup.id, material_id=m.id, is_active=True, last_quoted_price=Decimal("1000000"), last_quoted_at=d(6), )) await db.flush() # ---- projects + warehouses ---- project_rows: list[Project] = [] warehouse_rows: list[list[Warehouse]] = [] pm_names = {"مجتمع مسکونی آرام": "tester", "برج اداری سپهر": "mohaghegh", "پروژه زیرساخت شرق تهران": None} for name, desc, wh_names in PROJECTS: proj = Project(name=name, description=desc, start_date=date.today().replace(year=date.today().year)) pm = pm_names.get(name) if pm: proj.pm_user_id = users[pm].id db.add(proj) await db.flush() project_rows.append(proj) whs: list[Warehouse] = [] for wn in wh_names: wh = Warehouse(project_id=proj.id, name=wn) db.add(wh) await db.flush() whs.append(wh) warehouse_rows.append(whs) await db.flush() # ---- quotes ---- for qidx, qdef in enumerate(QUOTES): sup = suppliers[qdef["supplier"]] proj = project_rows[qdef["project"]] accepted_at = d(qdef["accepted_days"]) if qdef["accepted_days"] is not None else None quote = Quote( supplier_id=sup.id, project_id=proj.id, quote_date=d(qdef["quote_days"]), expiry_date=(d(qdef["quote_days"]) + timedelta(days=30)).date(), accepted_date=accepted_at, arrived_date=d(1) if qdef["status"] == "arrived" else None, status=qdef["status"], is_official=True, notes="پیشنهاد نمونه — داده دمو", ) db.add(quote) await db.flush() db.add(QuoteProjects(quote_id=quote.id, project_id=proj.id)) db.add(QuoteEvent(quote_id=quote.id, event_type="offered", occurred_at=d(qdef["quote_days"]), actor=sup.name)) if accepted_at: db.add(QuoteEvent(quote_id=quote.id, event_type="accepted", occurred_at=accepted_at, actor=users["razaei"].first_name)) if qdef["status"] == "arrived": db.add(QuoteEvent(quote_id=quote.id, event_type="arrived", occurred_at=d(1), actor=users["razaei"].first_name)) ts = Termsheet(quote_id=quote.id, label="گزینه نقدی — دمو", status="accepted" if accepted_at else "offered") db.add(ts) await db.flush() total_all = Decimal("0") line_items: list[LineItem] = [] for item_idx, (mat_idx, qty, price) in enumerate(qdef["items"]): mat_name = MATERIALS[mat_idx][0] mat = material_by_name[mat_name] li = LineItem( quote_id=quote.id, material_id=mat.id, brand_material_id=brand_material_by_material.get(mat_name), unit_id=mat.canonical_unit_id, quantity=Decimal(str(qty)), ) db.add(li) await db.flush() line_items.append(li) db.add(LineItemProjects(line_item_id=li.id, project_id=proj.id)) its = ItemTermsheet(line_item_id=li.id, termsheet_id=ts.id, unit_price=Decimal(price)) db.add(its) await db.flush() total_all += Decimal(str(qty)) * Decimal(price) if accepted_at: decision = ItemDecision( line_item_id=li.id, status="chosen", project_id=proj.id, chosen_item_termsheet_id=its.id, adjusted_unit_price=Decimal(price), adjusted_notes="تأیید قیمت — دمو", decided_at=accepted_at, ) db.add(decision) # schedule steps: ۳۰٪ پیش، ۴۰٪ حین تحویل، ۳۰٪ پس از تحویل p1 = (total_all * Decimal("30") / 100).quantize(Decimal("1")) p2 = (total_all * Decimal("40") / 100).quantize(Decimal("1")) p3 = total_all - p1 - p2 for step_no, amount in enumerate([p1, p2, p3], start=1): db.add(TermsheetScheduleStep( termsheet_id=ts.id, step_no=step_no, due_date=(d(qdef["quote_days"]) + timedelta(days=step_no * 20)).date(), amount=amount, )) # ---- receipts + allocations + inventory (only for quotes with receipts) ---- for rdef in qdef.get("receipts", []): li = line_items[rdef["item"]] mat = material_by_name[MATERIALS[qdef["items"][rdef["item"]][0]][0]] wh = warehouse_rows[qdef["project"]][rdef["wh"]] decision = await db.scalar(select(ItemDecision).where(ItemDecision.line_item_id == li.id)) confirmer = users["mohaghegh"] if qdef["project"] == 1 else users["tester"] if qdef["project"] == 0 else users["razaei"] receipt = ItemReceipt( item_decision_id=decision.id, confirmed_by=confirmer.id, received_at=d(rdef["days"]), received_qty=Decimal(str(rdef["qty"])), on_time=True, ) db.add(receipt) await db.flush() alloc = ItemReceiptAllocation(receipt_id=receipt.id, warehouse_id=wh.id, quantity=Decimal(str(rdef["qty"]))) db.add(alloc) await db.flush() inv = await db.scalar(select(InventoryItem).where( InventoryItem.warehouse_id == wh.id, InventoryItem.material_id == mat.id, InventoryItem.specs_hash == "default", )) if inv is None: inv = InventoryItem( warehouse_id=wh.id, material_id=mat.id, brand_id=None, quantity=Decimal("0"), unit_id=mat.canonical_unit_id, specs={}, specs_hash="default", ) db.add(inv) await db.flush() inv.quantity += Decimal(str(rdef["qty"])) db.add(InventoryTransaction( inventory_id=inv.id, warehouse_id=wh.id, receipt_allocation_id=alloc.id, transaction_type="PURCHASE", quantity=Decimal(str(rdef["qty"])), performed_by=f"{confirmer.first_name} {confirmer.last_name}", note=f"رسید {str(receipt.id)[:8]} — {qdef['status']}", transaction_date=d(rdef["days"]), )) await db.flush() await db.commit() # summary print("\nSeed complete ✓") print("Projects : 3 (2 warehouses each)") print("Materials :", len(MATERIALS), "across all 8 categories") print("Suppliers :", len(SUPPLIERS)) print("Quotes :", len(QUOTES), "(offered / accepted-in-transit / partial / arrived)") print("Roles : razaei=admin · mousavi=مدیر تدارکات · mohaghegh/tester=مدیر پروژه") print("\nLogin: http://localhost — dev-mode admin: razaei / 123456") if __name__ == "__main__": try: asyncio.run(run()) except KeyboardInterrupt: sys.exit(130)