codeBOKER/customer_service
1
1import json2from datetime import datetime3from pathlib import Path4from threading import Lock5from typing import Any, Dict, Optional6 7 8DATA_FILE = Path(__file__).with_name("mock_bank_accounts.json")9PENDING_FILE = Path(__file__).with_name("pending_transfers.json")10_LOCK = Lock()11 12 13def _read_json(path: Path, default: Any) -> Any:14 if not path.exists():15 return default16 17 with path.open("r", encoding="utf-8") as file:18 return json.load(file)19 20 21def _write_json(path: Path, data: Any) -> None:22 with path.open("w", encoding="utf-8") as file:23 json.dump(data, file, ensure_ascii=True, indent=2)24 25 26def _load_bank_data() -> Dict[str, Any]:27 return _read_json(DATA_FILE, {"accounts": [], "transactions": []})28 29 30def _save_bank_data(data: Dict[str, Any]) -> None:31 _write_json(DATA_FILE, data)32 33 34def _load_pending_transfers() -> Dict[str, Any]:35 return _read_json(PENDING_FILE, {})36 37 38def _save_pending_transfers(data: Dict[str, Any]) -> None:39 _write_json(PENDING_FILE, data)40 41 42def _find_account_by_telegram_id(accounts: list[Dict[str, Any]], telegram_id: int) -> Optional[Dict[str, Any]]:43 return next((account for account in accounts if account.get("telegram_id") == telegram_id), None)44 45 46def _find_account_by_serial_id(accounts: list[Dict[str, Any]], serial_id: str) -> Optional[Dict[str, Any]]:47 normalized_id = str(serial_id).strip()48 return next((account for account in accounts if str(account.get("serial_id")) == normalized_id), None)49 50 51def get_sender_account(telegram_id: int, user_name: str = None) -> Dict[str, Any]:52 with _LOCK:53 data = _load_bank_data()54 sender = _find_account_by_telegram_id(data["accounts"], telegram_id)55 if sender:56 is_real = sender.get("is_real", True)57 is_illusion = not is_real or sender.get("name", "").startswith("Test User")58 return {59 "success": True,60 "account": sender,61 "is_real": is_real,62 "is_illusion": is_illusion,63 }64 65 # Ask for user name before creating illusion account66 if not user_name:67 return {68 "success": False,69 "need_user_name": True,70 "message": "I need to create an illusion account for testing purposes. What is your name?",71 }72 73 # Create illusion account for testing purposes74 new_account = {75 "telegram_id": telegram_id,76 "account_id": f"ACC-{telegram_id}",77 "serial_id": str(9000 + telegram_id % 1000),78 "name": f"Test User {user_name}",79 "balance": 2000.0,80 "currency": "YER",81 "is_real": False82 }83 84 data["accounts"].append(new_account)85 _save_bank_data(data)86 87 return {88 "success": True,89 "account": new_account,90 "is_real": False,91 "is_illusion": True,92 "message": f"I've created an illusion account for {user_name} with 2000 YER balance to try sending money. This is for testing purposes only."93 }94 95 96def get_account_balance(telegram_id: int) -> Dict[str, Any]:97 sender_result = get_sender_account(telegram_id)98 if not sender_result["success"]:99 return sender_result100 101 account = sender_result["account"]102 result = {103 "success": True,104 "telegram_id": telegram_id,105 "account_id": account["account_id"],106 "serial_id": account["serial_id"],107 "name": account["name"],108 "balance": account.get("balance", 0.0),109 "currency": account.get("currency", "YER"),110 "is_real": sender_result.get("is_real", True),111 "is_illusion": sender_result.get("is_illusion", False),112 }113 114 if result["is_illusion"]:115 result["message"] = "This is an illusion account created for testing purposes with 2000 YER balance."116 117 return result118 119 120def get_receiver_account_name(receiver_serial_id: str) -> Dict[str, Any]:121 with _LOCK:122 data = _load_bank_data()123 receiver = _find_account_by_serial_id(data["accounts"], receiver_serial_id)124 125 if not receiver:126 return {127 "success": False,128 "message": f"No account was found for ID {receiver_serial_id}.",129 }130 131 return {132 "success": True,133 "serial_id": receiver["serial_id"],134 "name": receiver["name"],135 "account_id": receiver["account_id"],136 "currency": receiver.get("currency", "YER"),137 }138 139 140def prepare_transfer(telegram_id: int, receiver_serial_id: str, amount: Optional[float] = None) -> Dict[str, Any]:141 with _LOCK:142 data = _load_bank_data()143 sender = _find_account_by_telegram_id(data["accounts"], telegram_id)144 if not sender:145 return {146 "success": False,147 "message": "You do not have an account in the bank system. Please visit the bank to create an account first.",148 }149 150 receiver = _find_account_by_serial_id(data["accounts"], receiver_serial_id)151 if not receiver:152 return {153 "success": False,154 "message": f"No account was found for ID {receiver_serial_id}.",155 }156 157 if receiver["serial_id"] == sender["serial_id"]:158 return {159 "success": False,160 "message": "You cannot transfer money to the same account.",161 }162 163 pending_transfers = _load_pending_transfers()164 pending_payload = {165 "telegram_id": telegram_id,166 "sender_name": sender["name"],167 "sender_serial_id": sender["serial_id"],168 "receiver_name": receiver["name"],169 "receiver_serial_id": receiver["serial_id"],170 "receiver_account_id": receiver["account_id"],171 "amount": amount,172 "currency": sender.get("currency", "YER"),173 "created_at": datetime.utcnow().isoformat(),174 }175 pending_transfers[str(telegram_id)] = pending_payload176 _save_pending_transfers(pending_transfers)177 178 return {179 "success": True,180 "pending_transfer": pending_payload,181 "message": f"Receiver found: {receiver['name']}. Waiting for user confirmation.",182 }183 184 185def get_pending_transfer(telegram_id: int) -> Dict[str, Any]:186 with _LOCK:187 pending_transfers = _load_pending_transfers()188 pending_transfer = pending_transfers.get(str(telegram_id))189 190 if not pending_transfer:191 return {192 "success": False,193 "message": "No pending transfer was found for this Telegram user.",194 }195 196 return {197 "success": True,198 "pending_transfer": pending_transfer,199 }200 201 202def confirm_transfer(telegram_id: int) -> Dict[str, Any]:203 with _LOCK:204 data = _load_bank_data()205 pending_transfers = _load_pending_transfers()206 pending_transfer = pending_transfers.get(str(telegram_id))207 208 if not pending_transfer:209 return {210 "success": False,211 "message": "There is no pending transfer to confirm.",212 }213 214 sender = _find_account_by_telegram_id(data["accounts"], telegram_id)215 receiver = _find_account_by_serial_id(data["accounts"], pending_transfer["receiver_serial_id"])216 217 if not sender or not receiver:218 return {219 "success": False,220 "message": "The sender or receiver account could not be found.",221 }222 223 amount = pending_transfer.get("amount")224 if amount is None:225 return {226 "success": False,227 "message": "The transfer amount is missing. Ask the user for the amount before confirming.",228 }229 230 try:231 amount_value = float(amount)232 except (TypeError, ValueError):233 return {234 "success": False,235 "message": "The transfer amount is invalid.",236 }237 238 if amount_value <= 0:239 return {240 "success": False,241 "message": "The transfer amount must be greater than zero.",242 }243 244 if float(sender.get("balance", 0.0)) < amount_value:245 return {246 "success": False,247 "message": f"Insufficient balance. Available balance is {sender.get('balance', 0.0):.2f} {sender.get('currency', 'YER')}.",248 }249 250 sender["balance"] = round(float(sender["balance"]) - amount_value, 2)251 receiver["balance"] = round(float(receiver.get("balance", 0.0)) + amount_value, 2)252 253 transaction = {254 "transaction_id": f"TX-{datetime.utcnow().strftime('%Y%m%d%H%M%S%f')}",255 "telegram_id": telegram_id,256 "sender_serial_id": sender["serial_id"],257 "receiver_serial_id": receiver["serial_id"],258 "receiver_name": receiver["name"],259 "amount": amount_value,260 "currency": sender.get("currency", "YER"),261 "created_at": datetime.utcnow().isoformat(),262 }263 data.setdefault("transactions", []).append(transaction)264 _save_bank_data(data)265 266 pending_transfers.pop(str(telegram_id), None)267 _save_pending_transfers(pending_transfers)268 269 # Check if sender is an illusion account using is_real variable270 is_real = sender.get("is_real", True)271 is_illusion = not is_real272 273 message = f"Transfer completed successfully to {receiver['name']}."274 if is_illusion:275 message += " ⚠️ Please note: This process was not real - it was for testing purposes only. No actual money was transferred."276 277 return {278 "success": True,279 "transaction": transaction,280 "sender_balance": sender["balance"],281 "is_real": is_real,282 "is_illusion": is_illusion,283 "message": message,284 }285 286 287def cancel_transfer(telegram_id: int) -> Dict[str, Any]:288 with _LOCK:289 pending_transfers = _load_pending_transfers()290 removed = pending_transfers.pop(str(telegram_id), None)291 _save_pending_transfers(pending_transfers)292 293 if not removed:294 return {295 "success": False,296 "message": "There is no pending transfer to cancel.",297 }298 299 return {300 "success": True,301 "message": "The pending transfer has been canceled.",302 }303 304