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
| import redis
import json
from datetime import datetime
class ShoppingCart:
def __init__(self):
self.r = redis.Redis(host='localhost', port=6379, decode_responses=True)
def get_cart_key(self, user_id):
return f"cart:{user_id}"
def initialize_cart(self, user_id):
"""初始化購物車"""
cart = {
"user_id": user_id,
"items": [],
"created_at": datetime.now().isoformat(),
"updated_at": datetime.now().isoformat()
}
self.r.execute_command(
'JSON.SET',
self.get_cart_key(user_id),
'$',
json.dumps(cart)
)
return cart
def add_item(self, user_id, product_id, name, price, quantity=1):
"""添加商品到購物車"""
key = self.get_cart_key(user_id)
# 檢查購物車是否存在
if not self.r.exists(key):
self.initialize_cart(user_id)
# 檢查商品是否已在購物車中
items = json.loads(
self.r.execute_command('JSON.GET', key, '$.items')
)[0]
for i, item in enumerate(items):
if item['product_id'] == product_id:
# 更新數量
self.r.execute_command(
'JSON.NUMINCRBY',
key,
f'$.items[{i}].quantity',
quantity
)
self._update_timestamp(key)
return self.get_cart(user_id)
# 添加新商品
new_item = {
"product_id": product_id,
"name": name,
"price": price,
"quantity": quantity
}
self.r.execute_command(
'JSON.ARRAPPEND',
key,
'$.items',
json.dumps(new_item)
)
self._update_timestamp(key)
return self.get_cart(user_id)
def remove_item(self, user_id, product_id):
"""從購物車移除商品"""
key = self.get_cart_key(user_id)
items = json.loads(
self.r.execute_command('JSON.GET', key, '$.items')
)[0]
for i, item in enumerate(items):
if item['product_id'] == product_id:
self.r.execute_command('JSON.DEL', key, f'$.items[{i}]')
self._update_timestamp(key)
break
return self.get_cart(user_id)
def update_quantity(self, user_id, product_id, quantity):
"""更新商品數量"""
key = self.get_cart_key(user_id)
items = json.loads(
self.r.execute_command('JSON.GET', key, '$.items')
)[0]
for i, item in enumerate(items):
if item['product_id'] == product_id:
self.r.execute_command(
'JSON.SET',
key,
f'$.items[{i}].quantity',
quantity
)
self._update_timestamp(key)
break
return self.get_cart(user_id)
def get_cart(self, user_id):
"""取得購物車內容"""
key = self.get_cart_key(user_id)
result = self.r.execute_command('JSON.GET', key)
return json.loads(result) if result else None
def get_total(self, user_id):
"""計算購物車總金額"""
cart = self.get_cart(user_id)
if not cart:
return 0
return sum(item['price'] * item['quantity'] for item in cart['items'])
def clear_cart(self, user_id):
"""清空購物車"""
self.r.execute_command('JSON.DEL', self.get_cart_key(user_id))
def _update_timestamp(self, key):
"""更新時間戳記"""
self.r.execute_command(
'JSON.SET',
key,
'$.updated_at',
f'"{datetime.now().isoformat()}"'
)
# 使用範例
cart = ShoppingCart()
# 添加商品
cart.add_item("user123", "P001", "機械鍵盤", 2500, 1)
cart.add_item("user123", "P002", "電競滑鼠", 1200, 2)
cart.add_item("user123", "P003", "滑鼠墊", 300, 1)
# 取得購物車
print(json.dumps(cart.get_cart("user123"), indent=2, ensure_ascii=False))
# 計算總金額
print(f"總金額: NT${cart.get_total('user123')}")
# 更新數量
cart.update_quantity("user123", "P002", 3)
# 移除商品
cart.remove_item("user123", "P003")
|