27f66a1878
All Coffer txs are public (agent/company ids only, no human PII). Snapshot to public-ledger.json; publish-news updates leora/news coffer-public-ledger article (branch → PR → merge → surge). Design note locks the rule. Quarry 3005d129.
120 lines
4.4 KiB
Python
120 lines
4.4 KiB
Python
"""CLI for Coffer: wallets, grants, subscription, shirt, contract, history."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from coffer.product import CofferProduct
|
|
|
|
DEFAULT_DB = Path(os.environ.get("COFFER_DB", "data/coffer.db"))
|
|
|
|
|
|
def _print(obj: object) -> None:
|
|
print(json.dumps(obj, indent=2, sort_keys=True))
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(prog="coffer", description="Coffer prepaid store and ledger")
|
|
p.add_argument("--db", type=Path, default=DEFAULT_DB, help="SQLite path (default data/coffer.db)")
|
|
sub = p.add_subparsers(dest="cmd", required=True)
|
|
|
|
sub.add_parser("bootstrap", help="Seed opening Coffer/Leora books (no business txs)")
|
|
sub.add_parser("record-contract", help="Record 200c Coffer→Leora build contract (first tx)")
|
|
|
|
w = sub.add_parser("create-wallet", help="Ensure an agent or company wallet exists")
|
|
w.add_argument("account_id")
|
|
|
|
b = sub.add_parser("balance", help="Show wallet balance")
|
|
b.add_argument("account_id")
|
|
|
|
g = sub.add_parser("grant-signup", help="Grant signup promotional credits (default 40)")
|
|
g.add_argument("agent_id")
|
|
g.add_argument("--amount", type=int, default=40)
|
|
|
|
s = sub.add_parser("charge-subscription", help="Charge seat via Coffer (default 20c, 1% cut)")
|
|
s.add_argument("agent_id")
|
|
s.add_argument("--amount", type=int, default=20)
|
|
|
|
sf = sub.add_parser("signup", help="Wallet + 40c grant + 20c subscription flow")
|
|
sf.add_argument("agent_id")
|
|
|
|
sh = sub.add_parser("buy-shirt", help="Buy Coffer shirt (20c)")
|
|
sh.add_argument("agent_id")
|
|
sh.add_argument("--qty", type=int, default=1)
|
|
sh.add_argument("--idempotency-key")
|
|
|
|
t = sub.add_parser("top-up", help="Handler funded top-up")
|
|
t.add_argument("agent_id")
|
|
t.add_argument("amount", type=int)
|
|
|
|
sub.add_parser("history", help="Print full ledger transaction history")
|
|
sub.add_parser("catalog", help="Show catalog goods")
|
|
sub.add_parser("public-ledger", help="Print public ledger JSON (no human PII)")
|
|
sub.add_parser("publish-news", help="Publish public ledger article to leora/news")
|
|
|
|
return p
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
product = CofferProduct(args.db)
|
|
try:
|
|
if args.cmd == "bootstrap":
|
|
_print(product.bootstrap_opening_books())
|
|
elif args.cmd == "record-contract":
|
|
_print(product.record_build_contract())
|
|
elif args.cmd == "create-wallet":
|
|
_print(product.create_wallet(args.account_id))
|
|
elif args.cmd == "balance":
|
|
_print(product.balance(args.account_id))
|
|
elif args.cmd == "grant-signup":
|
|
_print(product.grant_signup(args.agent_id, args.amount))
|
|
elif args.cmd == "charge-subscription":
|
|
_print(product.charge_subscription(args.agent_id, args.amount))
|
|
elif args.cmd == "signup":
|
|
_print(product.signup_flow(args.agent_id))
|
|
elif args.cmd == "buy-shirt":
|
|
_print(
|
|
product.buy_shirt(
|
|
args.agent_id, quantity=args.qty, idempotency_key=args.idempotency_key
|
|
)
|
|
)
|
|
elif args.cmd == "top-up":
|
|
_print(product.top_up(args.agent_id, args.amount))
|
|
elif args.cmd == "history":
|
|
_print(product.history())
|
|
elif args.cmd == "public-ledger":
|
|
_print(product.public_snapshot())
|
|
elif args.cmd == "publish-news":
|
|
# Explicit publish always attempts News PR+merge
|
|
import os
|
|
os.environ["COFFER_AUTO_PUBLISH_NEWS"] = "1"
|
|
_print(product.publish_public_ledger())
|
|
elif args.cmd == "catalog":
|
|
_print(
|
|
{
|
|
sku: {
|
|
"name": g.name,
|
|
"credit_price": g.credit_price,
|
|
"stock": g.stock,
|
|
"supplier_id": g.supplier_id,
|
|
"supplier_unit_cost": g.supplier_unit_cost,
|
|
}
|
|
for sku, g in product.store.goods.items()
|
|
}
|
|
)
|
|
else:
|
|
print(f"unknown command {args.cmd}", file=sys.stderr)
|
|
return 2
|
|
return 0
|
|
finally:
|
|
product.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|