Python 3 编程语法规律 —— 小白逆袭手册

# 你创建了一个在线文档，写了第一项
购物清单 = ["牛奶"]         # ← 这是一个"列表对象"，就像一份在线文档

# 你把文档链接发给室友——他打开的不是复制品，就是同一个文档！
室友的清单 = 购物清单       # ← 不是新建！是同一条链接，指向同一个文档

# 室友往清单里加了"鸡蛋"
室友的清单.append("鸡蛋")   # ← 他在同一个文档里加了一行

# 你刷新自己的文档——多了一行"鸡蛋"！
print(购物清单)             # ['牛奶', '鸡蛋'] ← 变了！因为是同一个文档！

# ===== 不可变类型：像一张已经印好的纸质名片 =====
# 纸上的字改不了，想改只能重新印一张

name = "张三"            # 名片A：印着"张三"
print(id(name))          # 名片A的编号，比如 140723456789

name = name + "先生"     # 不是涂改名片的字，而是重新印了名片B："张三先生"
print(id(name))          # 编号变了！这是新名片B了！

# 原来的"张三"名片还在吗？在抽屉里，只是没人用了


# ===== 可变类型：像一份在线协作文档 =====
# 文档还是那个文档，但内容可以随时编辑

scores = [90, 80, 70]    # 在线文档A
print(id(scores))        # 文档A的编号

scores.append(85)        # 在文档A里加了一行
print(id(scores))        # 编号没变！还是文档A，只是内容多了

不可变（纸质名片）：int, float, str, tuple, bool, frozenset
                  改 = 重新印一张

可变（在线文档）：list, dict, set, 自定义类的实例
                改 = 同一个文档里编辑

import copy

# 场景：一份共享文件夹，里面有文件，还有一个子文件夹
original = ["照片.jpg", {"名字": "证件照", "大小": 100}]   # 文件夹套文件夹

# ===== 赋值：把同一个文件夹链接发给同事 =====
ref = original
# ref 和 original 是同一个文件夹的两个分享链接

# ===== 浅拷贝：复制了外层文件夹，但里面的子文件夹还是同一个 =====
shallow = copy.copy(original)
# 新文件夹 shallow，但子文件夹还是原来的——你在里面改文件，同事也会看到

# ===== 深拷贝：连外层带子文件夹全部复制了一份 =====
deep = copy.deepcopy(original)
# 全新的文件夹 + 全新的子文件夹，各改各的，互不影响

# 现在修改子文件夹里的内容
original[1]["大小"] = 200

print(ref[1])       # {'名字': '证件照', '大小': 200}  ← 同一个子文件夹，跟着变了
print(shallow[1])   # {'名字': '证件照', '大小': 200}  ← 子文件夹是同一个，也变了
print(deep[1])      # {'名字': '证件照', '大小': 100}  ← 独立的子文件夹，没变！

# 场景：顾客排队点单，先来先服务
order_queue = ["珍珠奶茶", "杨枝甘露", "柠檬茶"]

# ===== 增删改查四件套 =====
# 增：新顾客来了
order_queue.append("芝士葡萄")         # 排在最后：['珍珠奶茶', '杨枝甘露', '柠檬茶', '芝士葡萄']
order_queue.insert(0, "招牌烤奶")      # 插队到最前：['招牌烤奶', '珍珠奶茶', '杨枝甘露', '柠檬茶', '芝士葡萄']

# 删：做好了取走（两种删除方式）
done = order_queue.pop()              # pop() 默认弹出最后一个：done='芝士葡萄'
cancelled = order_queue.pop(0)        # pop(0) 弹出指定位置第0个：cancelled='招牌烤奶'
order_queue.remove("柠檬茶")          # remove() 按值删除：找到"柠檬茶"删掉

# 改：顾客要换口味
order_queue[1] = "芋泥波波"          # 把第2个（索引从0开始）换成芋泥波波

# 查：看看还有多少单
print(order_queue[0])                 # 第一个：'珍珠奶茶'
print(order_queue[-1])                # 最后一个：'芝士葡萄'
print(order_queue[0:3])               # 前三个：切片 [start:stop] 不含stop

# ===== 最 Pythonic 的写法：推导式 =====
# 把菜单里的价格都打8折
prices = [15, 18, 12, 22, 20]
discount = [p * 0.8 for p in prices]  # [12.0, 14.4, 9.6, 17.6, 16.0]

# 只要15块以上的
expensive = [p for p in prices if p > 15]  # [18, 22, 20]

# 打印带编号的菜单
menu = ["珍珠奶茶", "杨枝甘露", "柠檬茶"]
for i, item in enumerate(menu):
    print(f"{i+1}. {item}")
# 输出：
# 1. 珍珠奶茶
# 2. 杨枝甘露
# 3. 柠檬茶

# 场景：每杯饮品有名称、价格、库存、配料
menu = {
    "珍珠奶茶": {
        "price": 15,
        "stock": 50,
        "ingredients": ["茶底", "牛奶", "珍珠"]
    },
    "杨枝甘露": {
        "price": 18,
        "stock": 30,
        "ingredients": ["芒果", "西柚", "椰奶"]
    },
}

# ===== 安全访问（不会报错崩溃）=====
# 用 [] 访问不存在会报错：menu["不存在"]["price"] → KeyError，系统崩了
# 用 .get() 安全访问：
price = menu.get("咖啡", {}).get("price", "暂无此商品")  # "暂无此商品"

# ===== 增删改查 =====
# 增：上新
menu["芝士葡萄"] = {"price": 22, "stock": 40, "ingredients": ["葡萄", "芝士", "茶底"]}

# 改：涨价
menu["珍珠奶茶"]["price"] = 16

# 删：下架
# ⚠️ 注意：dict.pop("key") 是按"键名"删除，跟 list.pop() 按"位置"删除不一样！
removed = menu.pop("杨枝甘露")  # 删掉键叫"杨枝甘露"的整条记录，并返回它的值
print(f"已下架：{removed}")     # 已下架：{'price': 18, 'stock': 30, 'ingredients': ['芒果', '西柚', '椰奶']}

# 查：遍历剩余商品（杨枝甘露已被删除，不会出现）
for name, info in menu.items():
    print(f"{name}: ¥{info['price']}, 库存{info['stock']}杯")
# 珍珠奶茶: ¥16, 库存50杯
# 芝士葡萄: ¥22, 库存40杯

# ===== 合并菜单（Python 3.9+）=====
base_menu = {"珍珠奶茶": 15, "柠檬茶": 12}
new_menu = {"柠檬茶": 10, "芝士葡萄": 22}    # 柠檬茶降价了
merged = base_menu | new_menu               # {'珍珠奶茶': 15, '柠檬茶': 10, '芝士葡萄': 22}
# 注意：同名的取后面的值

# 场景：统计来过店里的顾客，不能重复记录
customers_today = {"张三", "李四", "王五", "张三"}  # 重复的"张三"自动去掉
print(customers_today)  # {'张三', '李四', '王五'}

# 顾客来了就加上
customers_today.add("赵六")

# ===== 集合运算（太常用了！）=====
vip_customers = {"张三", "李四", "钱七", "孙八"}
new_customers = {"赵六", "周九", "张三"}

# 谁既是VIP又是新客户？
both = vip_customers & new_customers        # 交集：{'张三'}

# 所有来过的人（去重）
all_people = vip_customers | new_customers  # 并集：{'张三', '李四', '钱七', '孙八', '赵六', '周九'}

# VIP但不是新客户
vip_only = vip_customers - new_customers    # 差集：{'李四', '钱七', '孙八'}

# 快速去重（最常用）
tags = ["Python", "Java", "Python", "Go", "Java"]
unique_tags = list(set(tags))              # ['Python', 'Java', 'Go']

# 场景：每杯奶茶的固定配方比例，不能乱改
formula = ("茶底", "牛奶", "珍珠")   # 小括号 = 元组
# formula[0] = "绿茶"  ← 报错！元组不可修改

# ===== 最常见的用法：函数返回多个值 =====
def get_store_info():
    """获取店铺信息"""
    return "码力奶茶", "深圳市南山区", "09:00-22:00"
    # 逗号分隔返回多个值 → 实际打包成一个元组

name, address, hours = get_store_info()  # 解包接收
print(f"{name} 在 {address}，营业时间 {hours}")

# ===== 解包技巧 =====
orders = [("珍珠奶茶", 2), ("柠檬茶", 1), ("芝士葡萄", 3)]
for drink, count in orders:              # 直接解包
    print(f"{drink} x {count}杯")

# 交换变量（不用临时变量！）
a, b = 5, 10
a, b = b, a                              # a=10, b=5

# 场景：根据会员等级算折扣
def calculate_discount(level: str) -> float:
    """
    计算折扣率
    level: 会员等级 "V0"普通 / "V1"银卡 / "V2"金卡 / "V3"钻石
    返回: 折扣率 1.0=原价 0.8=8折
    """
    if level == "V3":           # 钻石会员
        return 0.7              # 7折
    elif level == "V2":         # 金卡会员
        return 0.8              # 8折
    elif level == "V1":         # 银卡会员
        return 0.9              # 9折
    else:                       # 普通会员兜底
        return 1.0              # 原价

# 使用
print(calculate_discount("V2"))  # 0.8

# Python 3.10+ 可以用 match（更优雅）
def get_level_name(level: str) -> str:
    """把等级代码转成中文"""
    match level:
        case "V0":
            return "普通会员"
        case "V1":
            return "银卡会员"
        case "V2":
            return "金卡会员"
        case "V3":
            return "钻石会员"
        case _:                  # _ 是通配符，匹配任何没列出的情况
            return "未知等级"

# 场景一：打印今日订单汇总
orders = [
    {"drink": "珍珠奶茶", "count": 2, "price": 15},
    {"drink": "柠檬茶", "count": 1, "price": 12},
    {"drink": "芝士葡萄", "count": 3, "price": 22},
]

print("=== 今日订单汇总 ===")
total = 0
for i, order in enumerate(orders, start=1):  # start=1 让编号从1开始
    subtotal = order["count"] * order["price"]
    total += subtotal
    print(f"{i}. {order['drink']} ×{order['count']}杯 = ¥{subtotal}")
print(f"总金额：¥{total}")
# 输出：
# === 今日订单汇总 ===
# 1. 珍珠奶茶 ×2杯 = ¥30
# 2. 柠檬茶 ×1杯 = ¥12
# 3. 芝士葡萄 ×3杯 = ¥66
# 总金额：¥108

# 场景二：用 while 处理库存递减
stock = 5       # 珍珠奶茶还剩5杯
order_count = 0

while stock > 0:
    stock -= 1                     # 卖一杯减一杯
    order_count += 1
    print(f"卖出第{order_count}杯，剩余{stock}杯")

print("珍珠奶茶售罄！")

# ===== 循环的 else：Python 独有的特性 =====
# else 在循环正常结束（没有 break）时执行
def find_vip(name: str, vip_list: list) -> str:
    """查找是否VIP，找完没找到就提示"""
    for vip in vip_list:
        if vip == name:
            print(f"找到VIP：{name}")
            break
    else:                           # ← 只有没 break 时才执行
        print(f"{name} 不是VIP")
    return "查询完毕"

find_vip("张三", ["李四", "王五"])   # 张三 不是VIP
find_vip("李四", ["李四", "王五"])   # 找到VIP：李四

# ===== 场景：订单创建函数，展示所有参数类型 =====

def create_order(
    drink_name: str,               # ① 必填参数：没有默认值，必须传
    count: int = 1,                # ② 默认参数：不传就用默认值
    *addons: str,                  # ③ *args：接收任意多个额外配料
    sugar: str = "正常",           # ④ 仅关键字参数（*后面的）
    ice: str = "正常",             #    必须用 sugar=xxx 的方式传
    **notes: str,                  # ⑤ **kwargs：接收任意键值对备注
) -> dict:                         # 返回值类型注解
    """
    创建一个奶茶订单
    参数顺序记忆口诀：必填 → 默认 → *打包 → 关键字 → **打包字典
    """
    order = {
        "drink": drink_name,
        "count": count,
        "addons": list(addons),    # addons 是元组，转成列表
        "sugar": sugar,
        "ice": ice,
        "notes": notes,            # notes 本身就是字典
    }
    return order

# ===== 各种调用方式 =====
# 最简单调用
order1 = create_order("珍珠奶茶")
print(order1)
# {'drink': '珍珠奶茶', 'count': 1, 'addons': [], 'sugar': '正常', 'ice': '正常', 'notes': {}}

# 加配料、改甜度冰量
order2 = create_order(
    "芝士葡萄",
    2,                             # 第二个参数，覆盖默认count
    "椰果", "波霸",                # *addons 接收为元组 ('椰果', '波霸')
    sugar="少糖",                  # 关键字参数
    ice="去冰",                    # 关键字参数
    remark="尽快送达",              # **notes 接收
    table="A3",                    # **notes 接收
)
print(order2)
# {'drink': '芝士葡萄', 'count': 2, 'addons': ['椰果', '波霸'],
#  'sugar': '少糖', 'ice': '去冰', 'notes': {'remark': '尽快送达', 'table': 'A3'}}

# ===== Lambda：一行写完的小函数 =====
# 场景：按价格给菜单排序
menu = [("珍珠奶茶", 15), ("柠檬茶", 12), ("芝士葡萄", 22)]
sorted_menu = sorted(menu, key=lambda item: item[1])  # 按价格（第2个元素）排序
print(sorted_menu)  # [('柠檬茶', 12), ('珍珠奶茶', 15), ('芝士葡萄', 22)]

# 不用 lambda 的写法（对比）
def get_price(item):
    return item[1]
sorted_menu2 = sorted(menu, key=get_price)  # 效果一样，但多写了3行

# ===== 场景：设计一个完整的奶茶订单系统 =====

class Drink:
    """饮品类 —— 所有饮品的模板"""

    shop_name = "码力奶茶"    # 类变量：所有饮品共享同一个店名

    def __init__(self, name: str, price: int, stock: int):
        """初始化：每创建一个饮品实例就自动调用"""
        self.name = name      # 实例变量：每个饮品独有的名称
        self.price = price    # 实例变量：每个饮品独有的价格
        self.stock = stock    # 实例变量：每个饮品独有的库存

    def sell(self, count: int = 1) -> bool:
        """卖出饮品（实例方法），返回 True/False 表示成功/失败"""
        if self.stock >= count:
            self.stock -= count
            print(f"✅ 卖出 {self.name} ×{count}杯，剩余{self.stock}杯")  # 打印只是给人看的，不是返回值！
            return True                                                    # ← 这里才是真正的返回值（bool）
        else:
            print(f"❌ {self.name} 库存不足！需要{count}杯，只有{self.stock}杯")
            return False                                                   # ← 真正的返回值（bool）

    def __str__(self) -> str:
        """print() 时显示什么"""
        return f"《{self.name}》¥{self.price} 库存:{self.stock}"

    def __repr__(self) -> str:
        """调试时显示什么"""
        return f"Drink(name='{self.name}', price={self.price}, stock={self.stock})"

    # ========== ⚠️ 三种方法对比图解 ==========
    # 拿奶茶店来类比，一个「Drink 类」里有三种方法：

    # 【普通方法】def sell(self, ...):    → 需要先做一杯饮品出来，才知道卖的是哪杯
    #   调用：milk_tea.sell(3)             → 先创建 milk_tea 实例，再卖
    #   第一个参数 self = milk_tea 实例     → 自动传入，所以能访问 milk_tea.name/stock

    # 【类方法】  @classmethod            → 改的是「整个店」的事，跟单杯饮品无关
    #   调用：Drink.change_shop_name(...)  → 直接用类名调用，不用创建实例
    #   第一个参数 cls = Drink 类本身       → 自动传入，所以能访问 Drink.shop_name

    # 【静态方法】@staticmethod           → 纯粹的工具函数，跟店、跟饮品都没关系
    #   调用：Drink.is_affordable(15, 20)  → 直接用类名调用，不用创建实例
    #   没有自动传入的参数！                → 就跟普通函数一样，只是放在类里归类

    # 一句话口诀：
    # 普通方法 → 操作「某一个实例」的数据    （self.xxx）
    # 类方法   → 操作「整个类共享」的数据    （cls.xxx）
    # 静态方法 → 不操作任何数据，就是个工具  （纯逻辑）

    @classmethod
    def change_shop_name(cls, new_name: str):
        """类方法：改店名，影响所有饮品"""
        cls.shop_name = new_name
        print(f"🏪 店名已改为：{new_name}")

    @staticmethod
    def is_affordable(price: int, budget: int) -> bool:
        """静态方法：工具函数，判断买不买得起"""
        return price <= budget


# ===== 继承：特制饮品 =====
class SeasonalDrink(Drink):
    """季节限定饮品 —— 继承自 Drink，多了季节属性"""

    def __init__(self, name: str, price: int, stock: int, season: str):
        super().__init__(name, price, stock)  # 调用父类的 __init__
        self.season = season                  # 新增：季节属性

    def sell(self, count: int = 1) -> bool:
        """重写父类方法：季节饮品打折，返回 True/False"""
        print(f"🌸 {self.season}限定优惠中！")
        return super().sell(count)            # 调用父类的 sell，父类返回 True/False

    def __str__(self) -> str:
        return f"《{self.name}》¥{self.price} [{self.season}限定] 库存:{self.stock}"


# ===== 实际使用 =====
# 创建饮品
milk_tea = Drink("珍珠奶茶", 15, 50)
lemon_tea = Drink("柠檬茶", 12, 30)

# 创建季节限定
cherry_drink = SeasonalDrink("樱花拿铁", 28, 20, "春季")

# 使用
print(milk_tea)               # 《珍珠奶茶》¥15 库存:50
milk_tea.sell(3)              # ✅ 卖出 珍珠奶茶 ×3杯，剩余47杯
cherry_drink.sell(2)          # 🌸 春季限定优惠中！ ✅ 卖出 樱花拿铁 ×2杯，剩余18杯

# 静态方法：不依赖 self，不依赖 cls，直接用类名调用
print(Drink.is_affordable(15, 20))  # True，20块预算买得起15的奶茶
# 上面等价于：拿「Drink 类」里的 is_affordable 工具，传入 (15, 20)，返回 True
# 不需要创建 Drink 实例，因为 is_affordable 里没有 self，纯粹是个工具函数

# 类方法：同样用类名调用，但第一个参数 cls 自动传入 = Drink 这个类本身
Drink.change_shop_name("码力茶饮旗舰店")  # 🏪 店名已改为：码力茶饮旗舰店
# 上面等价于：change_shop_name(cls=Drink, new_name="码力茶饮旗舰店")
# cls.shop_name = new_name  →  Drink.shop_name = "码力茶饮旗舰店"
print(milk_tea.shop_name)                 # 码力茶饮旗舰店 ← 类变量变了，所有实例都跟着变

# isinstance 判断继承关系
print(isinstance(cherry_drink, Drink))           # True
print(isinstance(cherry_drink, SeasonalDrink))   # True

import time
from functools import wraps

# ===== 场景：记录每个操作的耗时 =====

def timer(func):
    """装饰器：记录函数执行时间"""
    @wraps(func)                     # 保留原函数的名字和文档
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)  # ← 执行原函数
        end = time.time()
        print(f"⏱️  [{func.__name__}] 耗时: {end - start:.3f}秒")
        return result                 # ← 返回原函数的结果
    return wrapper                     # ← 返回包装后的函数

# 使用：在函数上面加 @装饰器名
@timer
def make_drink(drink_name: str) -> str:
    """制作饮品"""
    time.sleep(1.5)                   # 模拟制作过程
    return f"🧋 {drink_name} 做好了！"

@timer
def calculate_total(orders: list) -> int:
    """计算订单总额"""
    total = sum(item["price"] * item["count"] for item in orders)
    time.sleep(0.3)
    return total

# 调用时自动打印耗时
result = make_drink("珍珠奶茶")
print(result)
# 输出：
# ⏱️  [make_drink] 耗时: 1.501秒
# 🧋 珍珠奶茶 做好了！


# ===== 进阶：带参数的装饰器（三层嵌套）=====
def retry(max_attempts: int = 3):
    """装饰器工厂：创建一个可以重试的装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts:
                        print(f"❌ 重试{max_attempts}次都失败了: {e}")
                        raise                    # 最后一次还失败就抛出
                    print(f"⚠️  第{attempt}次失败，{1}秒后重试...")
                    time.sleep(1)
            return None
        return wrapper
    return decorator

# 使用：模拟一个不稳定的支付接口
@retry(max_attempts=3)
def pay(amount: int) -> str:
    """支付（模拟不稳定网络）"""
    import random
    if random.random() < 0.7:         # 70%概率失败
        raise ConnectionError("网络超时")
    return f"✅ 支付成功: ¥{amount}"

print(pay(100))
# 可能输出：⚠️  第1次失败，1秒后重试... → ✅ 支付成功: ¥100

# ===== 场景：奶茶店叫号系统 =====

def call_number(total: int = 100):
    """
    叫号机生成器
    每次只产出下一个号码，不会一次性生成100个号
    """
    for num in range(1, total + 1):
        print(f"📢 请 {num} 号取餐！")
        yield num                     # ← 暂停，返回当前号码
        # 下次调用 next() 时从这里继续


# 创建叫号机（此时还没有开始叫号！）
ticket_machine = call_number(5)

# 一个一个取号
print(next(ticket_machine))   # 📢 请 1 号取餐！ → 1
print(next(ticket_machine))   # 📢 请 2 号取餐！ → 2
# 可以随时停止，后面的号根本不会生成
print("休息10分钟...")
print(next(ticket_machine))   # 📢 请 3 号取餐！ → 3


# ===== 对比：不用生成器 vs 用生成器 =====
# 不用生成器：一次性创建100万个号码（内存爆炸）
# ⚠️ 1_000_000 就是 1000000，下划线只是方便人读，Python 会自动忽略它
all_numbers = [i for i in range(1_000_000)]  # 内存占用约8MB

# 用生成器：只在需要时才产生（内存几乎不占）
def number_generator():
    for i in range(1_000_000):  # 一百万个号码，但不一次性生成，叫到谁才产生谁
        yield i

gen = number_generator()         # 内存占用几乎为0
print(next(gen))                 # 1
print(next(gen))                 # 2
# 只取了2个，后面999998个根本没生成！


# ===== 生成器表达式：一行版生成器 =====
# 跟列表推导式一模一样，只是把 [] 换成 ()
squares = (x**2 for x in range(10))   # 生成器表达式
print(next(squares))                   # 0
print(next(squares))                   # 1
print(list(squares))                   # [4, 9, 16, 25, 36, 49, 64, 81]  取剩下的


# ===== yield 在 FastAPI 中的实际应用 =====
# FastAPI 的数据库依赖注入就靠 yield！
# 原理：yield 前面的代码在请求处理前执行，后面的代码在请求处理后执行

# 简化版示意：
# async def get_db():
#     db = await create_connection()   # ← 请求前：打开连接
#     try:
#         yield db                     # ← 请求中：给路由函数用
#     finally:
#         await db.close()             # ← 请求后：自动关闭连接

import asyncio

# ===== 场景：奶茶店接单流程 =====

# 同步版本：一个订单一个订单处理
def make_drink_sync(name: str, time_cost: float):
    """同步制作饮品（模拟耗时操作）"""
    import time
    print(f"🔨 开始制作 {name}...")
    time.sleep(time_cost)              # 傻等，什么也不做
    print(f"✅ {name} 完成！")
    return f"{name}"

def process_orders_sync():
    """同步处理3个订单"""
    start = time.time()
    make_drink_sync("珍珠奶茶", 2)
    make_drink_sync("柠檬茶", 1)
    make_drink_sync("芝士葡萄", 3)
    end = time.time()
    print(f"⏱️  同步总耗时: {end - start:.1f}秒")
    # 总耗时: 6秒（2+1+3）


# 异步版本：同时处理
async def make_drink_async(name: str, time_cost: float):
    """异步制作饮品（不傻等）"""
    print(f"🔨 开始制作 {name}...")
    await asyncio.sleep(time_cost)     # 不傻等，让出CPU去干别的
    print(f"✅ {name} 完成！")
    return f"{name}"

async def process_orders_async():
    """异步同时处理3个订单"""
    import time
    start = time.time()
    # 同时发起3个任务，一起等
    results = await asyncio.gather(
        make_drink_async("珍珠奶茶", 2),
        make_drink_async("柠檬茶", 1),
        make_drink_async("芝士葡萄", 3),
    )
    end = time.time()
    print(f"⏱️  异步总耗时: {end - start:.1f}秒")
    print(f"📦 全部完成: {results}")
    # 总耗时: 3秒（取最长的那个）而不是6秒！

# 运行异步函数
# asyncio.run(process_orders_async())


# ===== async/await 三步曲 =====
# 第一步：用 async def 定义协程函数
async def fetch_data():
    return "数据"

# 第二步：用 await 等待协程结果
async def main():
    data = await fetch_data()    # ✅ 在 async 函数里可以用 await
    print(data)

# 第三步：只能在 async 函数里用 await
# await fetch_data()            # ❌ 在普通函数里用 await 会报错！

# 运行入口
# asyncio.run(main())

码力奶茶API/
├── main.py              # ① 入口：创建app，注册路由
├── database.py          # ② 数据库：连接配置
├── models.py            # ③ 数据层：数据库表结构
├── schemas.py           # ④ 校验层：请求/响应格式
├── router.py            # ⑤ 接口层：定义 API 端点
└── requirements.txt     # ⑥ 依赖：需要安装的包

fastapi==0.115.0
uvicorn[standard]==0.30.0
sqlalchemy==2.0.35
pydantic==2.9.0

# ① 先确认 Python 已安装（3.8 以上版本）
python --version       # 应该看到 Python 3.x.x

# ② 创建项目文件夹并进入
mkdir 码力奶茶API
cd 码力奶茶API

# ③ 把上面的 requirements.txt 内容保存为 requirements.txt 文件
#    （Windows 用记事本，Mac 用文本编辑器）

# ④ 安装依赖（这一步会联网下载包，需要几分钟）
pip install -r requirements.txt

# 如果 pip 报错，试试：
# python -m pip install -r requirements.txt

"""
database.py —— 数据库连接配置

为什么需要这个文件？
- 所有数据库操作都需要一个"连接"，这里统一创建和管理
- SQLAlchemy 是 Python 最流行的 ORM（对象关系映射），让你用 Python 代码操作数据库，不用写 SQL
- SQLite 是零配置数据库：不需要安装任何数据库软件，一个文件就是整个数据库！
"""

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

# ① 数据库地址：sqlite:/// 表示用 SQLite 文件数据库
#    ./mali_tea.db 意思是「当前目录下的 mali_tea.db 文件」
#    第一次运行时文件不存在？没关系，SQLAlchemy 会自动创建！
#
#    三种常见数据库的写法对比：
#    - SQLite：   sqlite:///./mali_tea.db         （一个文件，零配置）
#    - MySQL：    mysql://root:123456@localhost:3306/mali_tea
#    - PostgreSQL：postgresql://user:pass@localhost:5432/mali_tea
DATABASE_URL = "sqlite:///./mali_tea.db"

# ② 创建引擎：引擎就是"数据库连接工厂"
#    echo=True 会打印所有 SQL 语句到控制台，方便看到每一步发生了什么
#    （生产环境要改成 echo=False，否则日志会刷屏）
#
#    ⚠️ connect_args 说明：
#    SQLite 默认只允许同一个线程访问，但 FastAPI 是多线程处理请求的
#    所以必须加 check_same_thread=False，让多个线程能同时操作 SQLite
engine = create_engine(
    DATABASE_URL,
    echo=True,                              # 打印 SQL 语句，学习时打开
    connect_args={"check_same_thread": False},  # 允许 FastAPI 多线程访问 SQLite
)

# ③ 创建会话工厂：每次请求创建一个新的数据库会话
#    autocommit=False: 不自动提交，手动控制事务（保证数据一致性）
#    autoflush=False:  不自动刷新，手动控制何时写入
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# ④ 基类：所有数据库表模型都继承它
#    SQLAlchemy 2.0 风格：用 DeclarativeBase
class Base(DeclarativeBase):
    pass


# ⑤ 依赖注入函数：FastAPI 用 Depends() 调用它来获取数据库会话
def get_db():
    """
    每次 API 请求时，FastAPI 会自动调用这个函数

    谁在调用它？
    在 router.py 里，每个接口函数都写：db: Session = Depends(get_db)
    FastAPI 看到 Depends(get_db) 就会自动执行 get_db()，把返回值传给 db 参数。
    举例：
        def list_drinks(db: Session = Depends(get_db)):
            # ↑ FastAPI 帮你调了 get_db()，把 SessionLocal() 实例给了 db
            drinks = db.query(Drink).all()
            return drinks
        # ↑ 函数结束，回到 get_db() 的 finally，自动 db.close()

    执行顺序：
    ① db = SessionLocal()     → 打开一个数据库连接
    ② yield db                → 把连接交给路由函数用
    ③ 路由函数执行完毕         → 回到这里，执行 finally
    ④ db.close()              → 关闭连接，防止连接泄漏

    为什么要用 yield？
    yield 让函数「暂停」在这里，把 db 先给别人用，
    等别人用完了，再回到 yield 的下一行继续执行。
    就像奶茶店店员：把奶茶递给顾客（yield db），
    等顾客走了再清理台面（db.close()）。
    """
    db = SessionLocal()
    try:
        yield db        # ← 把 db 给路由函数用
    finally:
        db.close()      # ← 请求结束后自动关闭，防止连接泄漏

"""
models.py —— 数据库表结构定义

为什么需要这个文件？
- 把 Python 类和数据库表一一对应
- 定义了 Drink 类 → 对应数据库 drinks 表
- 定义了 Order 类 → 对应数据库 orders 表
- SQLAlchemy 会自动把 Python 代码翻译成 SQL 语句
"""

from sqlalchemy import Column, Integer, String, Float, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from datetime import datetime, timezone
from database import Base


class Drink(Base):
    """饮品表 —— 存菜单信息"""
    __tablename__ = "drinks"          # 表名（复数形式是惯例）

    # 主键：每条记录的唯一ID，自动递增
    id = Column(Integer, primary_key=True, index=True)

    # 饮品名称，不能为空，最大50字符
    name = Column(String(50), nullable=False, comment="饮品名称")

    # 价格，不能为空
    price = Column(Float, nullable=False, comment="价格(元)")

    # 库存数量，默认0
    stock = Column(Integer, default=0, comment="库存数量")

    # 分类：茶饮/果茶/咖啡等
    category = Column(String(20), default="茶饮", comment="分类")

    # 创建时间和更新时间（自动记录）
    created_at = Column(
        DateTime,
        default=lambda: datetime.now(timezone.utc),
        comment="创建时间"
    )
    updated_at = Column(
        DateTime,
        default=lambda: datetime.now(timezone.utc),
        onupdate=lambda: datetime.now(timezone.utc),
        comment="更新时间"
    )

    # 关联：一个饮品可以出现在多个订单中
    orders = relationship("Order", back_populates="drink")

    def __repr__(self):
        return f"<Drink(id={self.id}, name='{self.name}', price={self.price})>"


class Order(Base):
    """订单表 —— 存每一笔订单"""
    __tablename__ = "orders"

    id = Column(Integer, primary_key=True, index=True)

    # 外键：指向 drinks 表的 id
    # 为什么用外键？确保订单里的饮品一定存在于菜单中
    drink_id = Column(Integer, ForeignKey("drinks.id"), nullable=False, comment="饮品ID")

    # 购买杯数
    count = Column(Integer, nullable=False, default=1, comment="杯数")

    # 甜度选择
    sugar = Column(String(10), default="正常", comment="甜度: 无糖/少糖/正常/多糖")

    # 冰量选择
    ice = Column(String(10), default="正常", comment="冰量: 去冰/少冰/正常/多冰")

    # 总金额（单价×杯数，下单时计算好存起来，避免后续价格变动影响历史订单）
    total_price = Column(Float, nullable=False, comment="总金额(元)")

    # 订单状态
    status = Column(
        String(20),
        default="pending",
        comment="状态: pending=待制作, making=制作中, done=已完成, cancelled=已取消"
    )

    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))

    # 关联：反向查饮品信息
    drink = relationship("Drink", back_populates="orders")

    def __repr__(self):
        return f"<Order(id={self.id}, drink_id={self.drink_id}, count={self.count})>"

"""
schemas.py —— 请求和响应的格式定义

为什么需要这个文件？
- 前端发来的数据格式可能不对（比如传了负数价格），需要校验
- 返回给前端的数据需要控制哪些字段可见（比如不返回数据库内部ID）
- Pydantic 会自动校验数据类型、范围，不合规就返回友好的错误信息
"""

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional


# ===== 饮品相关 Schema =====

class DrinkCreate(BaseModel):
    """创建饮品时前端传来的数据"""
    name: str = Field(
        ...,                              # ... 表示必填！
        min_length=1,                     # 至少1个字符
        max_length=50,                    # 最多50个字符
        description="饮品名称"
    )
    price: float = Field(
        ...,
        gt=0,                             # gt=大于，价格必须大于0
        le=999,                           # le=小于等于，最高999元
        description="价格(元)"
    )
    stock: int = Field(
        default=0,
        ge=0,                             # ge=大于等于，库存不能为负
        description="库存数量"
    )
    category: str = Field(
        default="茶饮",
        description="分类"
    )


class DrinkUpdate(BaseModel):
    """更新饮品时传来的数据（所有字段可选，只传要改的）"""
    name: Optional[str] = Field(None, min_length=1, max_length=50)
    price: Optional[float] = Field(None, gt=0, le=999)
    stock: Optional[int] = Field(None, ge=0)
    category: Optional[str] = None


class DrinkResponse(BaseModel):
    """返回给前端的饮品数据"""
    id: int
    name: str
    price: float
    stock: int
    category: str
    created_at: datetime

    # model_config 告诉 Pydantic：这个对象可以从 ORM 模型直接转换
    # 之前叫 class Config: orm_mode = True，Pydantic v2 改了写法
    model_config = {"from_attributes": True}


# ===== 订单相关 Schema =====

class OrderCreate(BaseModel):
    """创建订单时前端传来的数据"""
    drink_id: int = Field(..., gt=0, description="饮品ID")
    count: int = Field(default=1, ge=1, le=99, description="杯数")
    sugar: str = Field(default="正常", description="甜度")
    ice: str = Field(default="正常", description="冰量")


class OrderResponse(BaseModel):
    """返回给前端的订单数据"""
    id: int
    drink_id: int
    drink_name: str = ""       # 冗余饮品名称，方便前端直接显示
    count: int
    sugar: str
    ice: str
    total_price: float
    status: str
    created_at: datetime

    model_config = {"from_attributes": True}


# ===== 统一响应格式 =====

class ApiResponse(BaseModel):
    """
    所有接口统一返回这个格式
    前端只要看到 code=0 就知道成功了
    """
    code: int = 0              # 0=成功, 非0=错误码
    message: str = "success"   # 提示信息
    data: Optional[dict | list] = None  # 具体数据


class ErrorResponse(BaseModel):
    """错误时的响应格式"""
    code: int
    message: str
    detail: Optional[str] = None

"""
router.py —— API 接口定义

为什么这个文件是核心？
- 定义了"前端怎么调用后端"
- 每个函数 = 一个 API 端点
- GET /drinks → 获取菜单列表
- POST /drinks → 新增饮品
- POST /orders → 下单
- PATCH /orders/{id}/status → 更新订单状态
"""

# ===== 导入说明（小白必看！）=====
# Q: 为什么是 from database 而不是 from ./database？
# A: Python import 不支持 ./ 路径，直接写模块名就行，Python 会自动找
#
# Q: 跨目录导入怎么写？（如 utils/helpers.py）
# A: from utils.helpers import xxx（用 . 表示目录层级，不是 /）
#
# Q: 为什么 schemas 导入用 () 括起来？
# A: 一行写不下时，用 () 包起来就能换行
#
# Q: 为什么没有 export？class/def 怎么暴露？
# A: Python 默认全部公开！没有 export 关键字
#    下划线 _ 开头 = 约定私有（如 _secret_func），但只是约定，仍能导入
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from sqlalchemy import select
from database import get_db
from models import Drink, Order
from schemas import (
    DrinkCreate, DrinkUpdate, DrinkResponse,
    OrderCreate, OrderResponse, ApiResponse, ErrorResponse,
)

router = APIRouter(prefix="/api/v1", tags=["码力奶茶"])  # prefix 给所有路由加前缀


# ==========================================
# 饮品管理接口
# ==========================================

@router.get(
    "/drinks",
    response_model=ApiResponse,
    summary="获取菜单列表",
    description="返回所有在售饮品，支持按分类筛选"
)
def list_drinks(
    category: str | None = None,   # 可选查询参数：?category=果茶
    db: Session = Depends(get_db),  # ← 依赖注入：FastAPI 自动调用 get_db() 获取数据库会话
):
    """
    获取饮品列表

    业务流程：
    1. 如果传了 category 参数，就筛选该分类
    2. 如果没传，返回全部
    3. 按创建时间倒序排列（最新的在前）
    """
    # 构建查询
    stmt = select(Drink)
    if category:
        stmt = stmt.where(Drink.category == category)

    stmt = stmt.order_by(Drink.created_at.desc())
    drinks = db.execute(stmt).scalars().all()

    # 把 ORM 对象转成 Pydantic Schema（自动过滤掉不该返回的字段）
    drink_list = [DrinkResponse.model_validate(d).model_dump() for d in drinks]

    return ApiResponse(data=drink_list)


@router.post(
    "/drinks",
    response_model=ApiResponse,
    status_code=status.HTTP_201_CREATED,  # 201 = 创建成功
    summary="新增饮品",
)
def create_drink(
    data: DrinkCreate,              # ← Pydantic 自动校验：名称不能空、价格>0
    db: Session = Depends(get_db),
):
    """
    新增饮品到菜单

    业务流程：
    1. Pydantic 自动校验传入数据（名称长度、价格范围等）
    2. 创建 Drink 模型实例
    3. 保存到数据库
    4. 返回新创建的饮品信息
    """
    # 创建数据库记录
    drink = Drink(
        name=data.name,
        price=data.price,
        stock=data.stock,
        category=data.category,
    )
    db.add(drink)          # 加入待保存队列
    db.commit()            # 真正写入数据库
    db.refresh(drink)      # 刷新：获取数据库自动生成的 id 和 created_at

    return ApiResponse(
        message="饮品添加成功",
        data=DrinkResponse.model_validate(drink).model_dump(),
    )


@router.put(
    "/drinks/{drink_id}",
    response_model=ApiResponse,
    summary="更新饮品信息",
)
def update_drink(
    drink_id: int,                   # ← 路径参数：/drinks/3
    data: DrinkUpdate,               # ← 请求体
    db: Session = Depends(get_db),
):
    """
    更新饮品信息（支持部分更新）

    业务流程：
    1. 根据 ID 查找饮品
    2. 只更新传了的字段（没传的保持原值）
    3. 保存到数据库
    """
    # 查找饮品
    drink = db.get(Drink, drink_id)
    if not drink:
        raise HTTPException(
            status_code=404,
            detail=f"饮品 ID={drink_id} 不存在"
        )

    # model_dump(exclude_unset=True) 只包含用户明确传了的字段
    # 比如只传了 {"price": 20}，就只更新价格，不改其他
    update_data = data.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(drink, key, value)   # 等价于 drink.price = 20

    db.commit()
    db.refresh(drink)

    return ApiResponse(
        message="饮品信息已更新",
        data=DrinkResponse.model_validate(drink).model_dump(),
    )


@router.delete(
    "/drinks/{drink_id}",
    response_model=ApiResponse,
    summary="删除饮品",
)
def delete_drink(
    drink_id: int,
    db: Session = Depends(get_db),
):
    """删除饮品"""
    drink = db.get(Drink, drink_id)
    if not drink:
        raise HTTPException(status_code=404, detail=f"饮品 ID={drink_id} 不存在")

    db.delete(drink)
    db.commit()

    return ApiResponse(message=f"已删除：{drink.name}")


# ==========================================
# 订单管理接口
# ==========================================

@router.post(
    "/orders",
    response_model=ApiResponse,
    status_code=status.HTTP_201_CREATED,
    summary="下单",
)
def create_order(
    data: OrderCreate,
    db: Session = Depends(get_db),
):
    """
    下单买饮品

    业务流程（这就是真实的电商逻辑！）：
    1. 查饮品是否存在
    2. 查库存是否足够
    3. 计算总价 = 单价 × 杯数
    4. 扣减库存
    5. 创建订单
    6. 返回订单信息

    为什么在同一个事务里做这些？
    如果扣了库存但创建订单失败，库存就白白扣了
    SQLAlchemy 的事务保证要么全成功，要么全回滚
    """
    # ① 查饮品
    drink = db.get(Drink, data.drink_id)
    if not drink:
        raise HTTPException(status_code=404, detail="饮品不存在")

    # ② 查库存
    if drink.stock < data.count:
        raise HTTPException(
            status_code=400,
            detail=f"库存不足！需要{data.count}杯，只剩{drink.stock}杯"
        )

    # ③ 计算总价（为什么下单时算好存起来？防止后续涨价影响历史订单）
    total_price = round(drink.price * data.count, 2)

    # ④ 扣减库存
    drink.stock -= data.count

    # ⑤ 创建订单
    order = Order(
        drink_id=data.drink_id,
        count=data.count,
        sugar=data.sugar,
        ice=data.ice,
        total_price=total_price,
        status="pending",
    )
    db.add(order)
    db.commit()
    db.refresh(order)

    # ⑥ 组装响应（把饮品名称也带上，前端不用再查一次）
    response_data = OrderResponse.model_validate(order).model_dump()
    response_data["drink_name"] = drink.name  # 补充饮品名称

    return ApiResponse(
        message=f"下单成功！{drink.name} ×{data.count}杯，共 ¥{total_price}",
        data=response_data,
    )


@router.get(
    "/orders",
    response_model=ApiResponse,
    summary="获取订单列表",
)
def list_orders(
    status_filter: str | None = None,  # 可选筛选：?status_filter=done
    db: Session = Depends(get_db),
):
    """获取订单列表，支持按状态筛选"""
    stmt = select(Order)
    if status_filter:
        stmt = stmt.where(Order.status == status_filter)

    stmt = stmt.order_by(Order.created_at.desc())
    orders = db.execute(stmt).scalars().all()

    order_list = []
    for order in orders:
        data = OrderResponse.model_validate(order).model_dump()
        data["drink_name"] = order.drink.name if order.drink else "未知"
        order_list.append(data)

    return ApiResponse(data=order_list)


@router.patch(
    "/orders/{order_id}/status",
    response_model=ApiResponse,
    summary="更新订单状态",
)
def update_order_status(
    order_id: int,
    new_status: str,                # 查询参数：?new_status=done
    db: Session = Depends(get_db),
):
    """
    更新订单状态

    状态流转：pending(待制作) → making(制作中) → done(已完成)
    也可以：pending → cancelled(已取消)

    为什么用 PATCH 而不是 PUT？
    PATCH = 部分更新，PUT = 完整替换
    这里只改状态字段，所以用 PATCH 更语义化
    """
    # 允许的状态值
    valid_statuses = ["pending", "making", "done", "cancelled"]
    if new_status not in valid_statuses:
        raise HTTPException(
            status_code=400,
            detail=f"无效状态！可选: {', '.join(valid_statuses)}"
        )

    order = db.get(Order, order_id)
    if not order:
        raise HTTPException(status_code=404, detail=f"订单 ID={order_id} 不存在")

    old_status = order.status
    order.status = new_status

    # 如果取消订单，退还库存
    if new_status == "cancelled" and old_status != "cancelled":
        drink = db.get(Drink, order.drink_id)
        if drink:
            drink.stock += order.count
            print(f"🔄 退还库存：{drink.name} +{order.count}杯")

    db.commit()

    return ApiResponse(
        message=f"订单状态：{old_status} → {new_status}",
        data={"order_id": order_id, "status": new_status},
    )


# ==========================================
# 统计接口
# ==========================================

@router.get(
    "/stats",
    response_model=ApiResponse,
    summary="店铺统计",
)
def get_stats(db: Session = Depends(get_db)):
    """
    获取店铺统计数据

    这里展示纯 Python 逻辑：拿到数据后，用 Python 代码做计算
    这些计算跟框架无关，换 Django 也能直接用
    """
    # 查所有订单
    orders = db.execute(select(Order)).scalars().all()

    if not orders:
        return ApiResponse(data={"total_orders": 0, "total_revenue": 0, "avg_price": 0})

    # ===== 以下是纯 Python 计算，换任何框架都一样 =====
    total_orders = len(orders)
    total_revenue = sum(o.total_price for o in orders)      # 生成器表达式求和
    avg_price = round(total_revenue / total_orders, 2)

    # 按状态分组统计（用字典推导式）
    status_count = {}
    for o in orders:
        status_count[o.status] = status_count.get(o.status, 0) + 1

    # 最受欢迎的饮品（按订单数排序）
    from collections import Counter
    drink_counter = Counter(o.drink_id for o in orders)
    # Counter 自动统计出现次数

    return ApiResponse(data={
        "total_orders": total_orders,
        "total_revenue": total_revenue,
        "avg_price": avg_price,
        "status_breakdown": status_count,
        "top_drink_ids": drink_counter.most_common(3),  # 前3名
    })

"""
main.py —— 应用入口

为什么叫 main.py？
- 这是整个项目的启动文件
- 创建 FastAPI app → 创建数据库表 → 注册路由 → 启动服务器
- 运行命令：uvicorn main:app --reload
"""

from fastapi import FastAPI
from contextlib import asynccontextmanager
from database import engine, Base
from router import router


# ① 应用生命周期管理
#    lifespan 让你在应用启动/关闭时执行代码
#    启动时：创建数据库表 + 初始化菜单数据
#    关闭时：清理资源
@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用启动和关闭时的钩子"""
    # ===== 启动时执行 =====
    print("🚀 码力奶茶 API 启动中...")

    # 创建所有数据库表（如果表已存在则跳过）
    # Base.metadata 里记录了所有继承 Base 的模型（Drink、Order）
    # create_all 会根据模型定义自动生成 CREATE TABLE 语句
    Base.metadata.create_all(bind=engine)
    print("✅ 数据库表已就绪")

    # 初始化菜单数据（只在表为空时才插入，避免重复）
    from database import SessionLocal
    from models import Drink
    db = SessionLocal()
    try:
        # 检查 drinks 表里有没有数据
        if db.query(Drink).count() == 0:
            default_menu = [
                Drink(name="珍珠奶茶", price=15, stock=50, category="茶饮"),
                Drink(name="杨枝甘露", price=18, stock=30, category="果茶"),
                Drink(name="柠檬茶", price=12, stock=40, category="茶饮"),
                Drink(name="芝士葡萄", price=22, stock=25, category="果茶"),
                Drink(name="生椰拿铁", price=20, stock=35, category="咖啡"),
            ]
            db.add_all(default_menu)
            db.commit()
            print("📋 初始菜单已导入（5款饮品）")
        else:
            drink_count = db.query(Drink).count()
            print(f"📋 菜单已有 {drink_count} 款饮品，跳过初始化")
    finally:
        db.close()

    # 打印数据库文件位置
    import os
    db_path = os.path.abspath("mali_tea.db")
    print(f"🗄️  SQLite 数据库文件：{db_path}")
    print(f"💡 要重置数据？删掉这个文件再重启就行！")
    print(f"🌐 API 文档：http://127.0.0.1:8000/docs")

    yield  # ← 应用运行中，代码暂停在这里

    # ===== 关闭时执行 =====
    print("👋 码力奶茶 API 已关闭")


# ② 创建 FastAPI 应用实例
app = FastAPI(
    title="码力奶茶 API",
    description="一个教学用的奶茶店后端 API，展示 FastAPI + SQLAlchemy + SQLite 的完整用法",
    version="1.0.0",
    lifespan=lifespan,
)

# ③ 注册路由
app.include_router(router)


# ④ 健康检查接口（直接写在 main.py，不放 router.py，因为它不属于业务）
@app.get("/", summary="根路径", include_in_schema=False)
def root():
    """根路径返回欢迎信息"""
    return {
        "message": "欢迎来到码力奶茶！",
        "docs": "/docs",        # FastAPI 自动生成的 Swagger 文档（可以在这里直接测试接口！）
        "redoc": "/redoc",      # FastAPI 自动生成的 ReDoc 文档（更美观的文档风格）
    }


# ===== 启动方式 =====
# 在终端运行（先 cd 到项目文件夹）：
# uvicorn main:app --reload
#
# 参数说明：
# main:app → 文件 main.py 里的 app 变量
# --reload → 代码修改后自动重启（开发用，生产环境去掉）
# --host 0.0.0.0 → 允许局域网访问（可选）
# --port 8000 → 指定端口（可选，默认 8000）
#
# 启动后访问：
# http://127.0.0.1:8000      → 欢迎页
# http://127.0.0.1:8000/docs → 交互式 API 文档，可以直接在网页上测试接口！
#
# 数据库文件：
# 项目文件夹下会自动生成 mali_tea.db，这就是你的 SQLite 数据库
# 可以用任何 SQLite 工具打开查看，比如：
# - VS Code 插件：SQLite Viewer
# - 命令行：sqlite3 mali_tea.db