| |
| """Seed controlled synthetic Saudi-oriented demo finance data — explicit, never at import time. |
| |
| Creates clearly-fictional (simulation_only) institutions and merchants, and — in synthetic mode — |
| optional synthetic customer/operator users with opening balances posted through balanced journals. |
| Requires --confirm-d3. No real participant data. |
| |
| Example: |
| python scripts/d3_seed_demo_finance.py --db-path /tmp/amanpay-d2/amanpay.db --tenant event \ |
| --confirm-d3 --customers 2 --allow-unknown-fs --json |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import sys |
|
|
| from amanpay.finance.cli import (add_common_args, build_runtime_for_cli, emit, resolve_tenant_id) |
| from amanpay.finance.seed import seed_customer, seed_institutions, seed_merchants |
| from amanpay.identity.handles import normalize_handle |
|
|
|
|
| def _ensure_tenant(rt, tenant: str): |
| tid = resolve_tenant_id(rt, tenant) |
| if tid is not None: |
| return tid |
| t = rt.identity.repo.create_tenant(slug=tenant, name_en="Demo Event", name_ar="فعالية تجريبية") |
| return t.id |
|
|
|
|
| def _synthetic_user(rt, tid, handle, role="customer"): |
| existing = rt.identity.repo.find_user_by_handle(tid, normalize_handle(handle)) |
| if existing is not None: |
| return existing |
| u = rt.identity.repo.create_user( |
| tenant_id=tid, webauthn_user_handle=f"syn-{handle}", public_handle=handle, |
| public_handle_norm=normalize_handle(handle), display_alias=handle.title(), |
| preferred_language="en", role=role, status="enrollment_pending") |
| rt.identity.repo.set_user_status(u.id, "active") |
| return rt.identity.repo.get_user(u.id) |
|
|
|
|
| def main(argv=None) -> int: |
| p = argparse.ArgumentParser(description="Seed synthetic D3 demo finance data.") |
| add_common_args(p, tenant_required=True) |
| p.add_argument("--confirm-d3", action="store_true", required=True) |
| p.add_argument("--customers", type=int, default=0, |
| help="create N synthetic customer users with opening balances (synthetic mode)") |
| p.add_argument("--opening-minor", type=int, default=100000) |
| args = p.parse_args(argv) |
|
|
| rt = build_runtime_for_cli(args.db_path, allow_unknown_fs=args.allow_unknown_fs) |
| tid = _ensure_tenant(rt, args.tenant) |
| operator = _synthetic_user(rt, tid, "demo_operator", role="operator") if args.customers else None |
| institutions = seed_institutions(rt, tid) |
| merchants = seed_merchants(rt, tid, operator_user_id=(operator.id if operator else None)) |
| customers = [] |
| for i in range(max(0, args.customers)): |
| u = _synthetic_user(rt, tid, f"demo_customer_{i+1}") |
| accts = seed_customer(rt, tid, u.id, opening_minor=args.opening_minor) |
| customers.append({"user_id": u.id, "handle": u.public_handle, **accts}) |
| emit({"ok": True, "tenant_id": tid, "institutions": len(institutions), |
| "merchants": len(merchants), "customers": len(customers)}, as_json=args.json) |
| rt.close() |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main()) |
|
|
