| |
| """Initialize a D2 event tenant (and optionally the first operator invitation). |
| |
| Trusted operator CLI. Requires an EXPLICIT confirmed-local D2 database path. Prints any |
| generated invitation code exactly once; never prints secret configuration. |
| |
| Usage: |
| python scripts/d2_initialize_tenant.py --db-path /tmp/amanpay-d2/amanpay.db \ |
| --slug event-alpha --name-en "Alpha Event" --name-ar "فعالية ألفا" [--operator-invite] |
| |
| The database is created fresh only with --initialize (single-use bootstrap). |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
|
|
| from amanpay.identity.cli import emit, open_storage, snapshot_best_effort |
| from amanpay.identity.repository import D2Repository |
| from amanpay.identity.service import IdentityService |
|
|
|
|
| def main(argv=None) -> int: |
| ap = argparse.ArgumentParser(description="Initialize a D2 event tenant") |
| ap.add_argument("--db-path", required=True) |
| ap.add_argument("--slug", required=True) |
| ap.add_argument("--name-en", default="") |
| ap.add_argument("--name-ar", default="") |
| ap.add_argument("--initialize", action="store_true", |
| help="create a fresh empty DB if none exists (single-use bootstrap)") |
| ap.add_argument("--operator-invite", action="store_true", |
| help="also issue a one-time operator invitation and print its code once") |
| ap.add_argument("--invite-ttl", type=int, default=86400) |
| args = ap.parse_args(argv) |
|
|
| storage = open_storage(args.db_path, init=args.initialize) |
| try: |
| repo = D2Repository(storage.db) |
| existing = repo.get_tenant_by_slug(args.slug) |
| tenant = existing or repo.create_tenant( |
| slug=args.slug, name_en=args.name_en, name_ar=args.name_ar) |
| out = {"tenant_id": tenant.id, "slug": tenant.slug, "status": tenant.status, |
| "created": existing is None} |
| if args.operator_invite: |
| ident = IdentityService.from_storage(storage) |
| inv, code = ident.invites.create( |
| tenant_id=tenant.id, role="operator", ttl_seconds=args.invite_ttl) |
| out["operator_invitation"] = {"id": inv.id, "code": code, |
| "note": "Shown once — not stored in plaintext."} |
| gen = snapshot_best_effort(storage) |
| out["snapshot_generation"] = gen |
| emit(out) |
| finally: |
| storage.close() |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
