Spaces:
Running
Running
File size: 7,061 Bytes
a4e6593 500061c a4e6593 500061c a4e6593 aea9d7d a4e6593 aea9d7d a4e6593 500061c a4e6593 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 | """Billing system simulator for SentinelOps Arena."""
import uuid
from typing import Dict, List
from sentinelops_arena.models import Invoice, InvoiceStatus, RefundPolicy
class BillingSystem:
def __init__(self):
self.invoices: Dict[str, Dict] = {}
self.refund_policy: RefundPolicy = RefundPolicy()
self._rate_limit: int = 0 # 0 means no limit
self._call_count: int = 0
self._field_map: Dict[str, str] = {} # old_name -> new_name for drift
def initialize(self, invoices: List[Invoice]):
"""Populate billing from Invoice models."""
self.invoices = {inv.invoice_id: inv.model_dump() for inv in invoices}
self.refund_policy = RefundPolicy()
self._rate_limit = 0
self._call_count = 0
self._field_map = {}
def check_balance(self, customer_id: str) -> Dict:
"""Return all invoices for a customer and total balance."""
if self._rate_limit_check():
return {"error": "Rate limit exceeded. Try again next tick."}
customer_invoices = [
inv for inv in self.invoices.values()
if inv["customer_id"] == customer_id
]
if not customer_invoices:
return {"error": f"No invoices found for customer {customer_id}"}
total = sum(
inv["amount"] for inv in customer_invoices
if inv["status"] in (InvoiceStatus.PENDING.value, InvoiceStatus.OVERDUE.value)
)
return {
"success": True,
"customer_id": customer_id,
"invoices": customer_invoices,
"outstanding_balance": total,
"invoice_count": len(customer_invoices),
}
def issue_refund(self, invoice_id: str, amount: float, reason: str, current_tick: int = 0) -> Dict:
"""Validate refund against current policy and process it."""
if self._rate_limit_check():
return {"error": "Rate limit exceeded. Try again next tick."}
if invoice_id not in self.invoices:
return {"error": f"Invoice {invoice_id} not found"}
invoice = self.invoices[invoice_id]
# Check refund policy
if amount > self.refund_policy.max_amount:
return {
"error": f"Refund amount ${amount:.2f} exceeds max allowed ${self.refund_policy.max_amount:.2f}",
"policy_violation": True,
}
# Check refund window
invoice_tick = invoice.get("date_tick", 0)
if current_tick - invoice_tick > self.refund_policy.window_ticks:
return {
"error": (
f"Invoice {invoice_id} is outside the refund window "
f"({self.refund_policy.window_ticks} ticks). "
f"Invoice date tick: {invoice_tick}, current tick: {current_tick}"
),
"policy_violation": True,
}
if invoice["status"] == InvoiceStatus.REFUNDED.value:
return {"error": f"Invoice {invoice_id} has already been refunded"}
if amount > invoice["amount"]:
return {
"error": f"Refund amount ${amount:.2f} exceeds invoice amount ${invoice['amount']:.2f}"
}
if self.refund_policy.requires_approval:
return {
"success": True,
"status": "pending_approval",
"invoice_id": invoice_id,
"amount": amount,
"reason": reason,
"message": "Refund requires manager approval under current policy",
}
# Process the refund
invoice["status"] = InvoiceStatus.REFUNDED.value
return {
"success": True,
"status": "refunded",
"invoice_id": invoice_id,
"amount": amount,
"reason": reason,
}
def apply_credit(self, customer_id: str, amount: float) -> Dict:
"""Apply a credit to a customer's account by creating a credit invoice."""
if self._rate_limit_check():
return {"error": "Rate limit exceeded. Try again next tick."}
credit_id = f"CREDIT-{uuid.uuid4().hex[:8].upper()}"
credit_invoice = {
"invoice_id": credit_id,
"customer_id": customer_id,
"amount": -amount,
"status": InvoiceStatus.PAID.value,
"date_tick": 0,
"items": [f"Account credit: ${amount:.2f}"],
}
self.invoices[credit_id] = credit_invoice
return {
"success": True,
"customer_id": customer_id,
"credit_id": credit_id,
"amount": amount,
}
def generate_invoice(self, customer_id: str, items: List[str], amount: float) -> Dict:
"""Create a new invoice."""
if self._rate_limit_check():
return {"error": "Rate limit exceeded. Try again next tick."}
invoice_id = f"INV-{uuid.uuid4().hex[:8].upper()}"
new_invoice = {
"invoice_id": invoice_id,
"customer_id": customer_id,
"amount": amount,
"status": InvoiceStatus.PENDING.value,
"date_tick": 0,
"items": items,
}
self.invoices[invoice_id] = new_invoice
return {
"success": True,
"invoice_id": invoice_id,
"customer_id": customer_id,
"amount": amount,
"items": items,
}
def get_current_policy(self) -> Dict:
"""Return current refund policy."""
return {
"success": True,
"policy": self.refund_policy.model_dump(),
}
def get_schema(self) -> Dict:
"""Return current field names after any drift."""
fields = list(Invoice.model_fields.keys())
for old, new in self._field_map.items():
fields = [new if f == old else f for f in fields]
return {"system": "billing", "fields": fields}
def apply_schema_drift(self, old_field: str, new_field: str):
"""Rename a field across all invoice records."""
self._field_map[old_field] = new_field
for inv_id in self.invoices:
if old_field in self.invoices[inv_id]:
self.invoices[inv_id][new_field] = self.invoices[inv_id].pop(old_field)
def apply_policy_drift(self, changes: Dict):
"""Modify refund policy fields."""
data = self.refund_policy.model_dump()
data.update(changes)
self.refund_policy = RefundPolicy(**data)
def set_rate_limit(self, max_calls_per_tick: int):
"""Set rate limit for API calls per tick."""
self._rate_limit = max_calls_per_tick
def reset_rate_limit_counter(self):
"""Reset call counter. Called each tick."""
self._call_count = 0
def _rate_limit_check(self) -> bool:
"""Return True if over limit."""
self._call_count += 1
if self._rate_limit > 0 and self._call_count > self._rate_limit:
return True
return False
|