refactor: 简化StatusBadge组件并优化N+1查询

- 重构StatusBadge组件，统一状态颜色和标签映射
- 优化库存列表N+1查询，使用批量查询替代循环查询
- 修复删除会话时保留当前会话
- 更新Redis缓存失效策略文档
- 优化库存分页逻辑

Author: hzb666

app/api/inventory.py

@@ -235,30 +235,30 @@ def _add_specification(item_dict: dict) -> dict:
     return item_dict
 
 
-def _add_user_names(db: Session, item_dict: dict) -> dict:
-    """Add user names to inventory response dict"""
-    # Get borrower name
-    if item_dict.get("borrower_id"):
-        borrower = db.get(User, item_dict["borrower_id"])
-        item_dict["borrower_name"] = borrower.full_name or borrower.username if borrower else None
-    else:
-        item_dict["borrower_name"] = None
+# def _add_user_names(db: Session, item_dict: dict) -> dict:
+#     """Add user names to inventory response dict"""
+#     # Get borrower name
+#     if item_dict.get("borrower_id"):
+#         borrower = db.get(User, item_dict["borrower_id"])
+#         item_dict["borrower_name"] = borrower.full_name or borrower.username if borrower else None
+#     else:
+#         item_dict["borrower_name"] = None
 
-    # Get last borrower name
-    if item_dict.get("last_borrower_id"):
-        last_borrower = db.get(User, item_dict["last_borrower_id"])
-        item_dict["last_borrower_name"] = last_borrower.full_name or last_borrower.username if last_borrower else None
-    else:
-        item_dict["last_borrower_name"] = None
+#     # Get last borrower name
+#     if item_dict.get("last_borrower_id"):
+#         last_borrower = db.get(User, item_dict["last_borrower_id"])
+#         item_dict["last_borrower_name"] = last_borrower.full_name or last_borrower.username if last_borrower else None
+#     else:
+#         item_dict["last_borrower_name"] = None
 
-    # Get created by name
-    if item_dict.get("created_by_id"):
-        created_by = db.get(User, item_dict["created_by_id"])
-        item_dict["created_by_name"] = created_by.full_name or created_by.username if created_by else None
-    else:
-        item_dict["created_by_name"] = None
+#     # Get created by name
+#     if item_dict.get("created_by_id"):
+#         created_by = db.get(User, item_dict["created_by_id"])
+#         item_dict["created_by_name"] = created_by.full_name or created_by.username if created_by else None
+#     else:
+#         item_dict["created_by_name"] = None
 
-    return item_dict
+#     return item_dict
 
 
 # ==================== Named Routes (BEFORE /{id}) ====================
@@ -757,11 +757,41 @@ def norm_field(field):
     else:
         items = db.exec(base.order_by(order_expr)).all()
 
+    # ================= 性能优化核心：批量查询用户（消除 N+1） =================
+    # 遍历当前页的数据，收集所有需要查询的用户 ID
+    user_ids = set()
+    for item in items:
+        if item.borrower_id:
+            user_ids.add(item.borrower_id)
+        if item.last_borrower_id:
+            user_ids.add(item.last_borrower_id)
+        if item.created_by_id:
+            user_ids.add(item.created_by_id)
+
+    # 用一条 SQL IN 语句，一次性查出这 50 条数据对应的所有用户
+    users_map = {}
+    if user_ids:
+        users = db.exec(select(User).where(User.id.in_(user_ids))).all()
+        # 构建内存字典 { user_id: "姓名" }，查找速度是 O(1)
+        users_map = {u.id: (u.full_name or u.username) for u in users}
+
+    # 在内存中完成数据组装，绝不再向数据库发请求
+    result_data = []
+    for item in items:
+        # 序列化单条数据
+        item_dict = InventoryResponse.model_validate(item).model_dump()
+        item_dict = _add_specification(item_dict)
+
+        # 直接从内存字典中塞入用户名字
+        item_dict["borrower_name"] = users_map.get(item.borrower_id)
+        item_dict["last_borrower_name"] = users_map.get(item.last_borrower_id)
+        item_dict["created_by_name"] = users_map.get(item.created_by_id)
+
+        result_data.append(item_dict)
+    # =========================================================================
+
     result = {
-        "data": [
-            _add_user_names(db, _add_specification(InventoryResponse.model_validate(i).model_dump()))
-            for i in items
-        ],
+        "data": result_data,
         "total": total,
         "skip": skip,
         "limit": limit,

app/api/user_sessions.py

@@ -17,6 +17,9 @@
 
 router = APIRouter(prefix="/sessions", tags=["Sessions"])
 
+# 导入 get_current_session 用于获取当前会话
+from app.api.deps import get_current_session
+
 
 class SessionResponse(BaseModel):
     """Session response model"""
@@ -87,13 +90,16 @@ def delete_session(
 
 @router.delete("/")
 def delete_all_sessions(
+    request: Request,
     db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user)
+    current_user: User = Depends(get_current_user),
+    current_session: UserSession = Depends(get_current_session)
 ):
-    """Delete all sessions for current user"""
+    """Delete all sessions for current user except the current session"""
     sessions = db.exec(
         select(UserSession)
         .where(UserSession.user_id == current_user.id)
+        .where(UserSession.token_hash != current_session.token_hash)
     ).all()
 
     deleted_count = 0

app/models/user.py

@@ -25,6 +25,8 @@ class UserBase(SQLModel):
 
 class User(UserBase, table=True):
     """User database model"""
+    __tablename__ = "users"
+    
     id: Optional[int] = Field(default=None, primary_key=True)
     password_hash: str
     created_at: datetime = Field(default_factory=datetime.utcnow)

docs/plans/gemini.md

@@ -0,0 +1,118 @@
+### 1. 高危安全漏洞修复 (Security)
+
+根据深度安全审计，当前库存模块存在多个越权（IDOR）和并发业务逻辑漏洞。
+
+#### 1.1 修复库存详情接口的未授权访问 (IDOR)
+
+在 `inventory.py` 中，`get_inventory` 接口完全没有使用认证依赖，任何未登录的外部访问者都可以读取库存数据。
+
+**🔧 修复代码 (`app/api/inventory.py` 约777行):**
+
+**Python**
+
+```
+@router.get("/{inventory_id}", response_model=InventoryResponse)
+def get_inventory(
+    inventory_id: int, 
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user)  # 修复：添加当前用户认证依赖
+):
+    """Get inventory item by ID"""
+    item = _get_by_id(db, inventory_id)
+    # ... 后续逻辑保持不变
+```
+
+#### 1.2 修复借用操作的竞态条件 (TOCTOU)
+
+`borrow_item` 函数在检查库存状态和更新库存状态之间存在时间差，如果存在并发请求，会导致同一件物品被多个人同时借出。
+
+**🔧 修复代码 (`app/api/inventory.py` 约833行):**
+需引入悲观锁（Pessimistic Locking），在查询时锁定该行数据：
+
+**Python**
+
+```
+from sqlmodel import select
+
+@router.post("/{inventory_id}/borrow", response_model=InventoryResponse)
+def borrow_item(
+    inventory_id: int,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    # 修复：使用 with_for_update() 锁定该行，防止并发修改
+    statement = select(Inventory).where(Inventory.id == inventory_id).with_for_update()
+    item = db.exec(statement).first()
+  
+    if not item:
+        raise HTTPException(status_code=404, detail="Inventory item not found")
+
+    if item.status != InventoryStatus.IN_STOCK:
+        raise HTTPException(status_code=400, detail=f"Cannot borrow item with status: {item.status}")
+
+    # ... 执行借出逻辑与 db.commit()
+```
+
+---
+
+### 2. 严重性能瓶颈优化 (Performance)
+
+#### 2.1 消除 N+1 查询风暴
+
+在原代码的 `_add_user_names` 辅助函数中，针对列表中的每一项数据，都会发起最多 3 次 `db.get(User, id)` 查询。如果列表包含 100 条数据，将额外触发 300 次数据库查询。
+
+**🔧 优化方案：改用批量映射获取用户信息**
+废弃原本针对单条记录查询的 `_add_user_names`，在 `list_inventory` 接口中统一进行批量抓取：
+
+**Python**
+
+```
+# 替换 list_inventory 中原有的列表推导式
+items = db.exec(base.order_by(order_expr).offset(skip).limit(limit)).all()
+
+# 1. 收集当前页面所有涉及到的 user_id
+user_ids = set()
+for item in items:
+    if item.borrower_id: user_ids.add(item.borrower_id)
+    if item.last_borrower_id: user_ids.add(item.last_borrower_id)
+    if item.created_by_id: user_ids.add(item.created_by_id)
+
+# 2. 一次性查出所有 User
+users_map = {}
+if user_ids:
+    users = db.exec(select(User).where(User.id.in_(user_ids))).all()
+    users_map = {u.id: (u.full_name or u.username) for u in users}
+
+# 3. 组装数据，不再进行额外的 DB 查询
+result_data = []
+for item in items:
+    item_dict = _add_specification(InventoryResponse.model_validate(item).model_dump())
+    item_dict["borrower_name"] = users_map.get(item.borrower_id)
+    item_dict["last_borrower_name"] = users_map.get(item.last_borrower_id)
+    item_dict["created_by_name"] = users_map.get(item.created_by_id)
+    result_data.append(item_dict)
+```
+
+#### 2.2 修复拼音排序导致的全表加载
+
+当用户使用拼音排序时，原始代码逻辑是 `items = db.exec(base.order_by(order_expr)).all()`，即使设定了 `limit` 也会拉取全部数据到内存中，有极大的 OOM 内存溢出风险。
+由于代码中已经添加了 `pinyin_sort_field_map`，需确保 `offset` 和 `limit` 在所有排序场景下都生效：
+
+**Python**
+
+```
+# 修复 inventory.py 约 740 行左右的逻辑：
+if limit > 0:
+    items = db.exec(base.order_by(order_expr).offset(skip).limit(limit)).all()
+else:
+    items = db.exec(base.order_by(order_expr)).all()
+```
+
+---
+
+### 3. 前端质量与架构同步优化 (Frontend Architecture)
+
+根据《代码质量.md》，由于后端 API 的调整（特别是去除了冗余查询和提升了响应速度），前端也应同步进行瘦身以避免页面卡顿：
+
+1. **组件职责拆分** ：当前 `Inventory.tsx` 长达近 900 行。应当将“新增入库”和“编辑库存”的弹窗（Modal）完全抽离为独立的组件，比如 `<AddInventoryModal />`。这能阻止弹窗内用户的输入导致整个大型数据表格重渲染。
+2. **引入现代数据获取库** ：目前的 React 页面依然在使用 `useEffect` 和手动状态管理（`loading`, `data`, `error`）来拉取数据。建议将后端刚优化好的分页与缓存逻辑搭配 `@tanstack/react-query` 使用，以原生支持请求防抖与数据缓存。

docs/plans/redis_cache_optimization.md

@@ -177,6 +177,63 @@ def invalidate_cas_info(cas_number: str) -> None:
 - 新建订单时，自动填充 CAS 对应的名称、类别、品牌
 - 库存列表中相同 CAS 号只查一次数据库
 
+### 4.3 缓存失效策略
+
+缓存的最大挑战是**缓存失效（Cache Invalidation）**。以下是针对不同字段的处理策略：
+
+#### 缓存字段与失效规则
+
+| 缓存字段 | 说明 | 更新时处理 |
+|----------|------|------------|
+| `name` | 中文名称 | 删除缓存 |
+| `english_name` | 英文名称 | 删除缓存 |
+| `brand` | 品牌 | 删除缓存 |
+| `category` | 分类 | 删除缓存 |
+| `alias` | 别名 | 删除缓存 |
+| `is_hazardous` | 是否危化品 | 删除缓存 |
+| `unit` | 单位 | 删除缓存 |
+| `specification` | 规格（初始值） | 删除缓存 |
+| `initial_quantity` | 初始数量（默认值参考） | 删除缓存 |
+| `price` | 价格（参考价） | 删除缓存 |
+
+#### 失效策略：写入时删除（Write Invalidate）
+
+```python
+def update_cas_info(cas_number: str, info: dict):
+    """更新 CAS 信息"""
+    # 1. 更新数据库
+    db.update(...)
+
+    # 2. 删除缓存（强制下次查询回源数据库）
+    invalidate_cas_info(cas_number)
+
+def create_order(cas_number: str, order_data: dict):
+    """创建订单时填充 CAS 信息"""
+    # 1. 查询 CAS 信息（优先缓存）
+    cas_info = get_cached_cas_info(cas_number)
+    if not cas_info:
+        cas_info = db.query(CASInfo).filter_by(cas_number=cas_number).first()
+        if cas_info:
+            # 2. 写入缓存
+            cache_cas_info(cas_number, cas_info.to_dict())
+
+    # 3. 填充订单
+    order_data['name'] = cas_info['name']
+    order_data['brand'] = cas_info['brand']
+```
+
+#### TTL 作为最终保障
+
+```python
+CAS_INFO_TTL = 86400  # 24小时
+```
+
+#### 需要失效缓存的场景
+
+1. **库存编辑** - 修改了 CAS 对应的名称/品牌/分类
+2. **手动入库** - 新增 CAS 信息
+3. **批量导入** - 导入新的 CAS 数据
+
 ### 5.2 方案二：分类/品牌列表缓存
 
 **目标**：缓存分类和品牌列表，快速加载下拉选项
@@ -321,12 +378,20 @@ def invalidate_user_cache(user_id: int) -> None:
 - [X] Session 缓存功能
 - [X] 设备管理后端 API
 - [X] 禁用用户时清理 Session 缓存
+- [X] 缓存失效策略文档（写入时删除 + TTL）
+
+### 缓存失效策略
+
+- [X] `invalidate_cas_info` 函数定义
+- [X] 缓存字段列表（name, english_name, brand, category, alias, is_hazardous, unit）
+- [X] 失效场景说明（库存编辑、手动入库、批量导入）
 
 ### 待实现
 
 - [ ] CAS 基础信息缓存
 - [ ] 分类/品牌列表缓存
 - [ ] 用户权限缓存
+- [ ] 实现缓存失效调用（在库存编辑、手动入库、批量导入时）
 
 ---