| |
| """Issue a batch of one-time D2 customer invitations. Prints codes exactly once. |
| |
| Usage: |
| python scripts/d2_issue_customer_invites.py --db-path /tmp/amanpay-d2/amanpay.db \ |
| --tenant event-alpha --count 25 [--ttl 86400] [--csv] |
| |
| --csv emits id,code rows (still safe: codes are one-time and never re-derivable from the DB). |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import csv |
| import sys |
|
|
| from amanpay.identity.cli import emit, open_storage, resolve_tenant_id, snapshot_best_effort |
| from amanpay.identity.service import IdentityService |
|
|
|
|
| def main(argv=None) -> int: |
| ap = argparse.ArgumentParser(description="Issue D2 customer invitations") |
| ap.add_argument("--db-path", required=True) |
| ap.add_argument("--tenant", required=True, help="tenant id or slug") |
| ap.add_argument("--count", type=int, default=1) |
| ap.add_argument("--ttl", type=int, default=86400) |
| ap.add_argument("--csv", action="store_true") |
| args = ap.parse_args(argv) |
|
|
| count = max(1, min(int(args.count), 500)) |
| storage = open_storage(args.db_path) |
| try: |
| ident = IdentityService.from_storage(storage) |
| tenant_id = resolve_tenant_id(ident.repo, args.tenant) |
| pairs = [] |
| for _ in range(count): |
| inv, code = ident.invites.create(tenant_id=tenant_id, role="customer", |
| ttl_seconds=args.ttl) |
| pairs.append((inv.id, code)) |
| gen = snapshot_best_effort(storage) |
| if args.csv: |
| w = csv.writer(sys.stdout) |
| w.writerow(["invitation_id", "code"]) |
| for iid, code in pairs: |
| w.writerow([iid, code]) |
| else: |
| emit({"tenant_id": tenant_id, "count": len(pairs), "snapshot_generation": gen, |
| "invitations": [{"id": iid, "code": code} for iid, code in pairs], |
| "note": "Codes shown once — only hashes are stored."}) |
| finally: |
| storage.close() |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
