基于Telegram WebApp的Aiogram 3购物车功能实现求助

核心思路：内存式用户购物车存储 #

用全局字典user_carts存储每个用户的购物车数据，键为用户ID（int类型），值为购物车商品列表。每个商品用字典存储{"product_id": xxx, "name": "xxx", "quantity": x, "price": x}这类字段，确保能唯一标识商品、记录数量与价格。

具体功能实现代码示例 #

1. 初始化全局购物车存储 #

# 全局变量，存储用户购物车：key=user_id(int), value=list[dict]
user_carts = {}

2. 添加/更新商品到购物车 #

当WebApp回传选中的商品数据时，判断用户是否已有购物车，无则初始化；有则检查商品是否已存在，存在则更新数量，不存在则新增：

from aiogram.types import WebAppData, types
import json

async def handle_webapp_data(message: types.Message, web_app_data: WebAppData):
    product_data = json.loads(web_app_data.data)
    user_id = message.from_user.id
    
    # 初始化用户购物车（如果不存在）
    if user_id not in user_carts:
        user_carts[user_id] = []
    
    cart = user_carts[user_id]
    # 检查商品是否已在购物车中
    existing_product = next((item for item in cart if item["product_id"] == product_data["product_id"]), None)
    if existing_product:
        # 更新数量
        existing_product["quantity"] += product_data["quantity"]
    else:
        # 添加新商品
        cart.append(product_data)
    
    await message.answer(f"已处理：{product_data['name']} x {product_data['quantity']}")

3. 清空购物车 #

直接重置用户对应的购物车列表：

async def clear_cart_handler(message: types.Message):
    user_id = message.from_user.id
    if user_id in user_carts:
        user_carts[user_id] = []
        await message.answer("购物车已清空")
    else:
        await message.answer("购物车为空")

4. 单独删除商品 #

根据商品ID从用户购物车中过滤掉对应条目：

async def delete_product_handler(message: types.Message):
    # 假设通过命令参数传入商品ID，比如 /delete 1001
    try:
        product_id = int(message.text.split()[1])
    except (IndexError, ValueError):
        await message.answer("请传入正确的商品ID，格式：/delete 商品ID")
        return
    
    user_id = message.from_user.id
    if user_id not in user_carts or not user_carts[user_id]:
        await message.answer("购物车为空")
        return
    
    cart = user_carts[user_id]
    new_cart = [item for item in cart if item["product_id"] != product_id]
    if len(new_cart) == len(cart):
        await message.answer("购物车中无此商品")
    else:
        user_carts[user_id] = new_cart
        await message.answer("商品已从购物车移除")

5. 查看购物车 #

返回用户当前购物车的商品明细与总价：

async def show_cart_handler(message: types.Message):
    user_id = message.from_user.id
    if user_id not in user_carts or not user_carts[user_id]:
        await message.answer("购物车为空")
        return
    
    cart = user_carts[user_id]
    cart_text = "你的购物车：\n"
    total_price = 0
    for item in cart:
        item_total = item["quantity"] * item["price"]
        total_price += item_total
        cart_text += f"- {item['name']} x {item['quantity']} | 小计：{item_total}\n"
    cart_text += f"总计：{total_price}"
    await message.answer(cart_text)

补充优化：临时持久化（避免重启丢失数据） #

内存存储的缺点是机器人重启后数据丢失，可通过pickle将购物车数据定期保存到本地文件，重启时加载：

import pickle

# 保存购物车到本地文件
def save_carts():
    with open("carts.pkl", "wb") as f:
        pickle.dump(user_carts, f)

# 从本地文件加载购物车
def load_carts():
    global user_carts
    try:
        with open("carts.pkl", "rb") as f:
            user_carts = pickle.load(f)
    except FileNotFoundError:
        user_carts = {}

# 机器人启动时加载数据
load_carts()

# 可在机器人关闭前触发保存，比如通过aiogram的shutdown事件
async def on_shutdown(dispatcher):
    save_carts()

内容的提问来源于stack exchange，提问作者KitPolosa