feat(market): 新增 get_my_usages CRUD 方法

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/crud/material.py

@@ -9,7 +9,7 @@ from sqlalchemy import func, or_
 from sqlalchemy.orm import Session
 
 from app.core.utils import generate_uuid
-from app.models.material import BizMaterial
+from app.models.material import BizMaterial, BizMaterialStockBatch, BizMaterialUsage
 
 
 def _dec(v: Any) -> float:
@@ -362,3 +362,153 @@ def options_list(db: Session, tenant_no: str) -> List[Dict[str, Any]]:
         }
         for r in rows
     ]
+
+
+def _my_usage_row_to_dict(
+    r: BizMaterialUsage,
+    *,
+    create_by_name: str = "",
+    batch: Optional[BizMaterialStockBatch] = None,
+    receiver_name: str = "",
+) -> Dict[str, Any]:
+    """将 BizMaterialUsage 行转换为字典（用于我的领用记录）。"""
+    create_by = r.create_by or ""
+    receiver = (getattr(r, "receiver", None) or "").strip()
+    purpose = (getattr(r, "purpose", None) or "").strip()
+    create_time = r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else ""
+    stock_after = float(r.stock_after) if r.stock_after is not None else 0
+    change_qty = float(r.change_qty) if r.change_qty is not None else 0
+    stock_before = (
+        float(r.stock_before)
+        if r.stock_before is not None
+        else stock_after - change_qty
+    )
+    unit = (getattr(r, "usage_unit", None) or "").strip()
+    if not unit and batch:
+        unit = (batch.unit or "").strip()
+    return {
+        "id": r.id,
+        "materialId": r.material_id,
+        "materialCode": r.material_code or "",
+        "materialName": r.material_name or "",
+        "batchId": r.batch_id or "",
+        "stockInNo": r.stock_in_no or "",
+        "catalogNo": r.catalog_no or "",
+        "batchNo": r.batch_no or "",
+        "manufacturer": r.manufacturer or "",
+        "unitPrice": float(r.unit_price) if r.unit_price is not None else None,
+        "amount": float(r.amount) if r.amount is not None else None,
+        "changeType": r.change_type,
+        "changeQty": change_qty,
+        "usageUnit": unit,
+        "stockBefore": stock_before,
+        "stockAfter": stock_after,
+        "receiver": receiver,
+        "receiverName": receiver_name or receiver,
+        "purpose": purpose,
+        "sourceType": r.source_type or "",
+        "requisitionStatus": (getattr(r, "requisition_status", None) or "open").strip() or "open",
+        "useTime": create_time,
+        "createBy": create_by,
+        "createByName": create_by_name or create_by,
+        "createTime": create_time,
+        "updateTime": r.update_time.strftime("%Y-%m-%d %H:%M:%S") if r.update_time else "",
+    }
+
+
+def _my_usage_batch_map(db: Session, batch_ids: List[str]) -> Dict[str, BizMaterialStockBatch]:
+    ids = [x for x in batch_ids if x]
+    if not ids:
+        return {}
+    rows = (
+        db.query(BizMaterialStockBatch)
+        .filter(BizMaterialStockBatch.id.in_(ids), BizMaterialStockBatch.del_flag == 0)
+        .all()
+    )
+    return {r.id: r for r in rows}
+
+
+def _my_usage_user_nick_name_map(db: Session, user_ids: List[str]) -> Dict[str, str]:
+    ids = list({x for x in user_ids if x})
+    if not ids:
+        return {}
+    from app.models.models import SysUser
+    rows = db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(ids)).all()
+    return {u.id: (u.nick_name or "").strip() for u in rows}
+
+
+def get_my_usages(
+    db: Session,
+    current_user_id: str,
+    tenant_no: str,
+    *,
+    page: int = 1,
+    page_size: int = 10,
+    keyword: Optional[str] = None,
+    material_id: Optional[str] = None,
+    material_type: Optional[str] = None,
+    change_type: Optional[str] = None,
+    start_time: Optional[str] = None,
+    end_time: Optional[str] = None,
+) -> Tuple[List[Dict[str, Any]], int]:
+    """
+    获取当前用户的物料领用记录（手动出库记录）
+
+    筛选条件:
+    - tenant_no = 当前用户租户
+    - source_type = 'manual_out'（手动出库）
+    - receiver = 当前用户ID
+    - del_flag = 0
+
+    按 create_time DESC 排序
+    """
+    q0 = db.query(BizMaterialUsage).filter(
+        BizMaterialUsage.tenant_no == tenant_no,
+        BizMaterialUsage.del_flag == 0,
+        BizMaterialUsage.source_type == "manual_out",
+        BizMaterialUsage.receiver == current_user_id,
+    )
+
+    if material_id:
+        q0 = q0.filter(BizMaterialUsage.material_id == material_id)
+
+    kw = (keyword or "").strip()
+    if kw:
+        like = f"%{kw}%"
+        q0 = q0.filter(
+            or_(
+                BizMaterialUsage.material_code.like(like),
+                BizMaterialUsage.material_name.like(like),
+                BizMaterialUsage.stock_in_no.like(like),
+                BizMaterialUsage.batch_no.like(like),
+                BizMaterialUsage.catalog_no.like(like),
+                BizMaterialUsage.purpose.like(like),
+            )
+        )
+
+    if start_time:
+        q0 = q0.filter(BizMaterialUsage.create_time >= start_time)
+    if end_time:
+        q0 = q0.filter(BizMaterialUsage.create_time <= end_time + " 23:59:59")
+
+    total = q0.order_by(None).with_entities(func.count(BizMaterialUsage.id)).scalar() or 0
+    page = max(1, page)
+    page_size = min(max(1, page_size), 100)
+    rows = (
+        q0.order_by(BizMaterialUsage.create_time.desc())
+        .offset((page - 1) * page_size)
+        .limit(page_size)
+        .all()
+    )
+
+    bmap = _my_usage_batch_map(db, [x.batch_id for x in rows if x.batch_id])
+    name_map = _my_usage_user_nick_name_map(db, [x.create_by for x in rows if x.create_by])
+
+    return [
+        _my_usage_row_to_dict(
+            x,
+            create_by_name=name_map.get(x.create_by or "", ""),
+            batch=bmap.get(x.batch_id or ""),
+        )
+        for x in rows
+    ], int(total)