taiyi
0adccb60d2
新建 cache.py 缓存工具模块，封装 Redis 读写和容错降级。 订单详情/列表、客户列表、定价规则读取走缓存，写操作后精确清除。 Redis 不可用时自动降级查 DB，不影响正常业务。 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
543 lines
23 KiB
Python
543 lines
23 KiB
Python
"""
|
||
定价管理路由模块
|
||
|
||
职责：
|
||
处理定价规则、供应商成本、价格层级的增删改查以及报价计算接口，URL 前缀为 /api。
|
||
包括：
|
||
- 定价规则 CRUD（/pricing-rules）
|
||
- 供应商成本 CRUD（/supplier-costs）
|
||
- 价格层级 CRUD（/price-tiers）
|
||
- 报价计算（/quotation/calculate）
|
||
"""
|
||
import json
|
||
|
||
from fastapi import APIRouter, Depends, Query
|
||
from sqlalchemy.orm import Session
|
||
|
||
from backend.app.api.deps import get_current_user, require_permissions, require_roles
|
||
from backend.app.core.cache import cache_delete_pattern, cache_get, cache_set
|
||
from backend.app.core.exceptions import AppException
|
||
from backend.app.core.error_codes import ErrorCode
|
||
from backend.app.db import get_db_session
|
||
from backend.app.models.business import (
|
||
ProductPriceTier,
|
||
ProductPricingRule,
|
||
SupplierProductCost,
|
||
)
|
||
from backend.app.schemas.common import success_payload
|
||
from backend.app.schemas.pricing import (
|
||
CreatePriceTierRequest,
|
||
CreatePricingRuleRequest,
|
||
CreateSupplierCostRequest,
|
||
QuotationCalculateRequest,
|
||
UpdatePriceTierRequest,
|
||
UpdatePricingRuleRequest,
|
||
UpdateSupplierCostRequest,
|
||
)
|
||
from backend.app.services.pricing_engine import pricing_engine
|
||
|
||
router = APIRouter(prefix="/api", tags=["pricing"])
|
||
|
||
|
||
# ======================================================================
|
||
# 定价规则 CRUD
|
||
# ======================================================================
|
||
|
||
@router.get("/pricing-rules")
|
||
def list_pricing_rules(
|
||
product_id: int | None = Query(default=None), # 产品 ID 筛选
|
||
status: int | None = Query(default=None), # 规则状态筛选（启用/停用）
|
||
page_no: int = Query(default=1), # 页码，默认第 1 页
|
||
page_size: int = Query(default=20), # 每页条数，默认 20 条
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
_user: dict = Depends(require_roles("admin", "manager")), # 角色鉴权：管理员/经理
|
||
_perm: dict = Depends(require_permissions("master-data:list")), # 权限鉴权：基础数据查看
|
||
) -> dict:
|
||
"""分页查询定价规则列表
|
||
|
||
用途：获取所有产品的定价规则，支持按产品 ID 和状态筛选。
|
||
请求参数：Query 参数筛选 + 分页参数。
|
||
返回值：分页定价规则列表，包含 total、page_no、page_size、list。
|
||
权限要求：管理员（admin）或经理（manager），且需 master-data:list 权限。
|
||
"""
|
||
q = session.query(ProductPricingRule).filter(ProductPricingRule.deleted == 0)
|
||
if product_id is not None:
|
||
q = q.filter(ProductPricingRule.product_id == product_id)
|
||
if status is not None:
|
||
q = q.filter(ProductPricingRule.status == status)
|
||
q = q.order_by(ProductPricingRule.id.desc())
|
||
total = q.count()
|
||
items = q.offset((page_no - 1) * page_size).limit(page_size).all()
|
||
return success_payload({
|
||
"total": total,
|
||
"page_no": page_no,
|
||
"page_size": page_size,
|
||
"list": [_rule_to_dict(r) for r in items],
|
||
})
|
||
|
||
|
||
@router.get("/pricing-rules/{product_id}")
|
||
def get_pricing_rule(
|
||
product_id: int, # 产品 ID（路径参数）
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
_user: dict = Depends(require_roles("admin", "manager")), # 角色鉴权：管理员/经理
|
||
) -> dict:
|
||
"""查询指定产品的定价规则
|
||
|
||
用途：根据产品 ID 获取该产品对应的定价规则详情。
|
||
请求参数：product_id - 产品 ID（路径参数）。
|
||
返回值：定价规则详情信息。
|
||
权限要求：管理员（admin）或经理（manager）。
|
||
"""
|
||
cache_key = f"pricing:rule:{product_id}"
|
||
cached = cache_get(cache_key)
|
||
if cached is not None:
|
||
return success_payload(cached)
|
||
rule = session.query(ProductPricingRule).filter(
|
||
ProductPricingRule.product_id == product_id,
|
||
ProductPricingRule.deleted == 0,
|
||
).first()
|
||
if not rule:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="定价规则不存在", status_code=404)
|
||
rule_data = _rule_to_dict(rule)
|
||
cache_set(cache_key, rule_data, ttl=300)
|
||
return success_payload(rule_data)
|
||
|
||
|
||
@router.post("/pricing-rules")
|
||
def create_pricing_rule(
|
||
payload: CreatePricingRuleRequest, # 创建定价规则的请求体
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
_user: dict = Depends(require_roles("admin", "manager")), # 角色鉴权：管理员/经理
|
||
_perm: dict = Depends(require_permissions("master-data:update")), # 权限鉴权：基础数据更新
|
||
) -> dict:
|
||
"""创建定价规则
|
||
|
||
用途：为指定产品创建一条定价规则，每个产品只能有一条生效规则。
|
||
请求参数：CreatePricingRuleRequest（产品 ID、定价类型、基准单价、公式等）。
|
||
返回值：创建成功后的定价规则信息。
|
||
权限要求：管理员（admin）或经理（manager），且需 master-data:update 权限。
|
||
"""
|
||
existing = session.query(ProductPricingRule).filter(
|
||
ProductPricingRule.product_id == payload.product_id,
|
||
ProductPricingRule.deleted == 0,
|
||
).first()
|
||
if existing:
|
||
raise AppException(code=ErrorCode.PARAM_ERROR, message="该产品已存在定价规则", status_code=400)
|
||
rule = ProductPricingRule(
|
||
product_id=payload.product_id,
|
||
product_name=payload.product_name,
|
||
pricing_type=payload.pricing_type,
|
||
base_unit_price=payload.base_unit_price,
|
||
pricing_unit=payload.pricing_unit,
|
||
pricing_inputs=payload.pricing_inputs,
|
||
formula_expr=payload.formula_expr,
|
||
formula_constants=payload.formula_constants,
|
||
surcharge_json=payload.surcharge_json,
|
||
formula_note=payload.formula_note,
|
||
status=payload.status,
|
||
)
|
||
session.add(rule)
|
||
session.commit()
|
||
cache_delete_pattern("pricing:*")
|
||
session.refresh(rule)
|
||
return success_payload(_rule_to_dict(rule))
|
||
|
||
|
||
@router.put("/pricing-rules/{rule_id}")
|
||
def update_pricing_rule(
|
||
rule_id: int, # 定价规则 ID（路径参数）
|
||
payload: UpdatePricingRuleRequest, # 更新定价规则的请求体
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
_user: dict = Depends(require_roles("admin", "manager")), # 角色鉴权：管理员/经理
|
||
_perm: dict = Depends(require_permissions("master-data:update")), # 权限鉴权：基础数据更新
|
||
) -> dict:
|
||
"""更新定价规则
|
||
|
||
用途：修改已有定价规则的公式、单价、附加费等参数。
|
||
请求参数：rule_id（路径参数）+ UpdatePricingRuleRequest（更新字段）。
|
||
返回值：更新后的定价规则信息。
|
||
权限要求：管理员（admin）或经理（manager），且需 master-data:update 权限。
|
||
"""
|
||
rule = session.query(ProductPricingRule).filter(
|
||
ProductPricingRule.id == rule_id,
|
||
ProductPricingRule.deleted == 0,
|
||
).first()
|
||
if not rule:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="定价规则不存在", status_code=404)
|
||
data = payload.model_dump(exclude_unset=True)
|
||
for k, v in data.items():
|
||
setattr(rule, k, v)
|
||
session.commit()
|
||
cache_delete_pattern("pricing:*")
|
||
session.refresh(rule)
|
||
return success_payload(_rule_to_dict(rule))
|
||
|
||
|
||
@router.delete("/pricing-rules/{rule_id}")
|
||
def delete_pricing_rule(
|
||
rule_id: int, # 定价规则 ID（路径参数）
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
_user: dict = Depends(require_roles("admin", "manager")), # 角色鉴权：管理员/经理
|
||
_perm: dict = Depends(require_permissions("master-data:update")), # 权限鉴权：基础数据更新
|
||
) -> dict:
|
||
"""删除定价规则（软删除）
|
||
|
||
用途：将指定定价规则标记为已删除，不从数据库物理删除。
|
||
请求参数：rule_id - 定价规则 ID（路径参数）。
|
||
返回值：{"deleted": true}
|
||
权限要求：管理员（admin）或经理（manager），且需 master-data:update 权限。
|
||
"""
|
||
rule = session.query(ProductPricingRule).filter(
|
||
ProductPricingRule.id == rule_id,
|
||
ProductPricingRule.deleted == 0,
|
||
).first()
|
||
if not rule:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="定价规则不存在", status_code=404)
|
||
rule.deleted = 1
|
||
session.commit()
|
||
cache_delete_pattern("pricing:*")
|
||
return success_payload({"deleted": True})
|
||
|
||
|
||
# ======================================================================
|
||
# 供应商成本 CRUD
|
||
# ======================================================================
|
||
|
||
@router.get("/supplier-costs")
|
||
def list_supplier_costs(
|
||
product_id: int = Query(...), # 产品 ID（必填）
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
_user: dict = Depends(require_roles("admin", "manager")), # 角色鉴权：管理员/经理
|
||
) -> dict:
|
||
"""查询指定产品的供应商成本列表
|
||
|
||
用途：获取某个产品在各供应商处的成本信息，按主供应商优先排序。
|
||
请求参数：product_id - 产品 ID（必填，Query 参数）。
|
||
返回值：供应商成本列表。
|
||
权限要求：管理员（admin）或经理（manager）。
|
||
"""
|
||
items = session.query(SupplierProductCost).filter(
|
||
SupplierProductCost.product_id == product_id,
|
||
SupplierProductCost.deleted == 0,
|
||
).order_by(SupplierProductCost.is_primary.desc(), SupplierProductCost.id.asc()).all()
|
||
return success_payload({"list": [_cost_to_dict(c) for c in items]})
|
||
|
||
|
||
@router.post("/supplier-costs")
|
||
def create_supplier_cost(
|
||
payload: CreateSupplierCostRequest, # 创建供应商成本的请求体
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
_user: dict = Depends(require_roles("admin", "manager")), # 角色鉴权：管理员/经理
|
||
) -> dict:
|
||
"""创建供应商成本记录
|
||
|
||
用途：新增某产品在某供应商处的成本价格。
|
||
请求参数：CreateSupplierCostRequest（产品 ID、供应商 ID、成本价、单位等）。
|
||
返回值：创建成功后的供应商成本信息。
|
||
权限要求：管理员（admin）或经理（manager）。
|
||
"""
|
||
cost = SupplierProductCost(**payload.model_dump())
|
||
session.add(cost)
|
||
session.commit()
|
||
cache_delete_pattern("pricing:*")
|
||
session.refresh(cost)
|
||
return success_payload(_cost_to_dict(cost))
|
||
|
||
|
||
@router.put("/supplier-costs/{cost_id}")
|
||
def update_supplier_cost(
|
||
cost_id: int, # 供应商成本 ID（路径参数）
|
||
payload: UpdateSupplierCostRequest, # 更新供应商成本的请求体
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
_user: dict = Depends(require_roles("admin", "manager")), # 角色鉴权：管理员/经理
|
||
) -> dict:
|
||
"""更新供应商成本记录
|
||
|
||
用途：修改已有供应商成本的价格、备注等信息。
|
||
请求参数：cost_id（路径参数）+ UpdateSupplierCostRequest（更新字段）。
|
||
返回值：更新后的供应商成本信息。
|
||
权限要求：管理员（admin）或经理（manager）。
|
||
"""
|
||
cost = session.query(SupplierProductCost).filter(
|
||
SupplierProductCost.id == cost_id,
|
||
SupplierProductCost.deleted == 0,
|
||
).first()
|
||
if not cost:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="供应商成本不存在", status_code=404)
|
||
data = payload.model_dump(exclude_unset=True)
|
||
for k, v in data.items():
|
||
setattr(cost, k, v)
|
||
session.commit()
|
||
cache_delete_pattern("pricing:*")
|
||
session.refresh(cost)
|
||
return success_payload(_cost_to_dict(cost))
|
||
|
||
|
||
@router.delete("/supplier-costs/{cost_id}")
|
||
def delete_supplier_cost(
|
||
cost_id: int, # 供应商成本 ID（路径参数）
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
_user: dict = Depends(require_roles("admin", "manager")), # 角色鉴权：管理员/经理
|
||
) -> dict:
|
||
"""删除供应商成本记录（软删除）
|
||
|
||
用途：将指定供应商成本记录标记为已删除。
|
||
请求参数：cost_id - 供应商成本 ID（路径参数）。
|
||
返回值：{"deleted": true}
|
||
权限要求：管理员（admin）或经理（manager）。
|
||
"""
|
||
cost = session.query(SupplierProductCost).filter(
|
||
SupplierProductCost.id == cost_id,
|
||
SupplierProductCost.deleted == 0,
|
||
).first()
|
||
if not cost:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="供应商成本不存在", status_code=404)
|
||
cost.deleted = 1
|
||
session.commit()
|
||
cache_delete_pattern("pricing:*")
|
||
return success_payload({"deleted": True})
|
||
|
||
|
||
# ======================================================================
|
||
# 价格层级 CRUD
|
||
# ======================================================================
|
||
|
||
@router.get("/price-tiers")
|
||
def list_price_tiers(
|
||
product_id: int = Query(...), # 产品 ID（必填）
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
_user: dict = Depends(require_roles("admin", "manager", "salesman")), # 角色鉴权：所有角色可查看
|
||
) -> dict:
|
||
"""查询指定产品的价格层级列表
|
||
|
||
用途：获取某个产品的各层级售价信息（如普通价、会员价等）。
|
||
请求参数：product_id - 产品 ID（必填，Query 参数）。
|
||
返回值：价格层级列表。
|
||
权限要求：管理员（admin）、经理（manager）、业务员（salesman）。
|
||
"""
|
||
items = session.query(ProductPriceTier).filter(
|
||
ProductPriceTier.product_id == product_id,
|
||
ProductPriceTier.deleted == 0,
|
||
).order_by(ProductPriceTier.id.asc()).all()
|
||
return success_payload({"list": [_tier_to_dict(t) for t in items]})
|
||
|
||
|
||
@router.post("/price-tiers")
|
||
def create_price_tier(
|
||
payload: CreatePriceTierRequest, # 创建价格层级的请求体
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
_user: dict = Depends(require_roles("admin", "manager")), # 角色鉴权：管理员/经理
|
||
) -> dict:
|
||
"""创建价格层级
|
||
|
||
用途：为指定产品新增一个价格层级（如普通客户价、VIP 客户价）。
|
||
请求参数：CreatePriceTierRequest（产品 ID、层级编码、层级名称、价格、单位等）。
|
||
返回值：创建成功后的价格层级信息。
|
||
权限要求：管理员（admin）或经理（manager）。
|
||
"""
|
||
tier = ProductPriceTier(**payload.model_dump())
|
||
session.add(tier)
|
||
session.commit()
|
||
cache_delete_pattern("pricing:*")
|
||
session.refresh(tier)
|
||
return success_payload(_tier_to_dict(tier))
|
||
|
||
|
||
@router.put("/price-tiers/{tier_id}")
|
||
def update_price_tier(
|
||
tier_id: int, # 价格层级 ID（路径参数）
|
||
payload: UpdatePriceTierRequest, # 更新价格层级的请求体
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
_user: dict = Depends(require_roles("admin", "manager")), # 角色鉴权：管理员/经理
|
||
) -> dict:
|
||
"""更新价格层级
|
||
|
||
用途：修改已有价格层级的名称、价格等信息。
|
||
请求参数：tier_id（路径参数）+ UpdatePriceTierRequest（更新字段）。
|
||
返回值：更新后的价格层级信息。
|
||
权限要求：管理员（admin）或经理（manager）。
|
||
"""
|
||
tier = session.query(ProductPriceTier).filter(
|
||
ProductPriceTier.id == tier_id,
|
||
ProductPriceTier.deleted == 0,
|
||
).first()
|
||
if not tier:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="价格层级不存在", status_code=404)
|
||
data = payload.model_dump(exclude_unset=True)
|
||
for k, v in data.items():
|
||
setattr(tier, k, v)
|
||
session.commit()
|
||
cache_delete_pattern("pricing:*")
|
||
session.refresh(tier)
|
||
return success_payload(_tier_to_dict(tier))
|
||
|
||
|
||
@router.delete("/price-tiers/{tier_id}")
|
||
def delete_price_tier(
|
||
tier_id: int, # 价格层级 ID（路径参数）
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
_user: dict = Depends(require_roles("admin", "manager")), # 角色鉴权：管理员/经理
|
||
) -> dict:
|
||
"""删除价格层级（软删除）
|
||
|
||
用途：将指定价格层级标记为已删除。
|
||
请求参数：tier_id - 价格层级 ID（路径参数）。
|
||
返回值：{"deleted": true}
|
||
权限要求：管理员（admin）或经理（manager）。
|
||
"""
|
||
tier = session.query(ProductPriceTier).filter(
|
||
ProductPriceTier.id == tier_id,
|
||
ProductPriceTier.deleted == 0,
|
||
).first()
|
||
if not tier:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="价格层级不存在", status_code=404)
|
||
tier.deleted = 1
|
||
session.commit()
|
||
cache_delete_pattern("pricing:*")
|
||
return success_payload({"deleted": True})
|
||
|
||
|
||
# ======================================================================
|
||
# 报价计算
|
||
# ======================================================================
|
||
|
||
@router.post("/quotation/calculate")
|
||
def calculate_quotation(
|
||
payload: QuotationCalculateRequest, # 报价计算请求体
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
_user: dict = Depends(require_roles("admin", "manager", "salesman")), # 角色鉴权：所有角色可计算
|
||
) -> dict:
|
||
"""计算产品报价
|
||
|
||
用途：根据定价规则、产品属性、客户层级和附加费选项，计算最终报价。
|
||
请求参数：QuotationCalculateRequest（产品 ID、客户 ID、用户输入参数、附加费选择等）。
|
||
返回值：报价结果，包含面积、基础成本、附加费明细、成本价、各层级售价、公式详情等。
|
||
权限要求：管理员（admin）、经理（manager）、业务员（salesman）。
|
||
"""
|
||
from backend.app.models.business import Product, Customer
|
||
|
||
# 查询产品定价规则
|
||
rule = session.query(ProductPricingRule).filter(
|
||
ProductPricingRule.product_id == payload.product_id,
|
||
ProductPricingRule.deleted == 0,
|
||
).first()
|
||
if not rule:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="该产品未配置定价规则", status_code=404)
|
||
|
||
# 获取产品属性（厚度、克重等），用于公式计算
|
||
product = session.query(Product).filter(Product.id == payload.product_id).first()
|
||
product_attrs = {}
|
||
if product:
|
||
if product.thickness:
|
||
try:
|
||
product_attrs["thickness"] = float(product.thickness)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
if product.weight_gsm:
|
||
product_attrs["weight_gsm"] = product.weight_gsm
|
||
|
||
# 合并前端传入的 surcharge 选项和自定义输入
|
||
full_inputs = {**payload.user_inputs, **payload.surcharge_selections, **payload.surcharge_inputs}
|
||
|
||
# 调用定价引擎计算
|
||
result = pricing_engine.calculate(rule, full_inputs, product_attrs)
|
||
|
||
# 查询客户所属价格层级，确定对应售价
|
||
tier_prices = {}
|
||
if payload.customer_id:
|
||
customer = session.query(Customer).filter(Customer.id == payload.customer_id).first()
|
||
tier_code = getattr(customer, "price_tier", None) or "default"
|
||
else:
|
||
tier_code = "default"
|
||
tiers = session.query(ProductPriceTier).filter(
|
||
ProductPriceTier.product_id == payload.product_id,
|
||
ProductPriceTier.deleted == 0,
|
||
).all()
|
||
for t in tiers:
|
||
tier_prices[t.tier_code] = float(t.price)
|
||
|
||
# 获取可用附加费选项列表
|
||
available_options = pricing_engine.get_available_surcharge_options(rule)
|
||
|
||
# 根据基础成本和单价反算面积
|
||
area_sqm = round(
|
||
(result.get("base_cost", 0) / float(rule.base_unit_price)) if rule.base_unit_price else 0, 4
|
||
)
|
||
|
||
return success_payload({
|
||
"product_name": rule.product_name, # 产品名称
|
||
"area_sqm": area_sqm, # 计算面积（平方米）
|
||
"base_cost": result["base_cost"], # 基础成本
|
||
"surcharge_items": result["surcharge_items"], # 附加费明细列表
|
||
"total_surcharge": result["total_surcharge"], # 附加费合计
|
||
"cost_price": result["cost_price"], # 总成本价
|
||
"sale_price_tier": tier_prices, # 各层级售价映射
|
||
"recommended_sale_price": tier_prices.get(tier_code, 0), # 推荐售价（按客户层级）
|
||
"formula_detail": result["formula_detail"], # 公式计算详情
|
||
"formula_note": rule.formula_note or "", # 公式说明
|
||
"available_surcharge_options": available_options, # 可用附加费选项
|
||
"tax_rate": result["tax_rate"], # 税率
|
||
"price_ex_tax": result["price_ex_tax"], # 不含税价
|
||
"tax_amount": result["tax_amount"], # 税额
|
||
"price_in_tax": result["price_in_tax"], # 含税价
|
||
})
|
||
|
||
|
||
# ======================================================================
|
||
# 序列化工具（将 ORM 模型转换为字典，便于返回 JSON 响应）
|
||
# ======================================================================
|
||
|
||
def _rule_to_dict(r: ProductPricingRule) -> dict:
|
||
"""将定价规则 ORM 对象序列化为字典"""
|
||
return {
|
||
"id": r.id,
|
||
"product_id": r.product_id,
|
||
"product_name": r.product_name,
|
||
"pricing_type": r.pricing_type,
|
||
"base_unit_price": float(r.base_unit_price or 0),
|
||
"pricing_unit": r.pricing_unit,
|
||
"pricing_inputs": r.pricing_inputs,
|
||
"formula_expr": r.formula_expr,
|
||
"formula_constants": r.formula_constants,
|
||
"surcharge_json": r.surcharge_json,
|
||
"formula_note": r.formula_note,
|
||
"tax_rate": float(r.tax_rate or 0),
|
||
"tax_inclusive": int(r.tax_inclusive or 0),
|
||
"status": r.status,
|
||
"created_at": r.created_at.strftime("%Y-%m-%d %H:%M:%S") if r.created_at else "",
|
||
}
|
||
|
||
|
||
def _cost_to_dict(c: SupplierProductCost) -> dict:
|
||
"""将供应商成本 ORM 对象序列化为字典"""
|
||
return {
|
||
"id": c.id,
|
||
"product_id": c.product_id,
|
||
"supplier_id": c.supplier_id,
|
||
"supplier_model": c.supplier_model,
|
||
"our_model": c.our_model,
|
||
"thickness": c.thickness,
|
||
"weight_gsm": c.weight_gsm,
|
||
"base_fabric_weight": c.base_fabric_weight,
|
||
"cost_price": float(c.cost_price or 0),
|
||
"cost_unit": c.cost_unit,
|
||
"is_primary": c.is_primary,
|
||
"remark": c.remark,
|
||
"status": c.status,
|
||
}
|
||
|
||
|
||
def _tier_to_dict(t: ProductPriceTier) -> dict:
|
||
"""将价格层级 ORM 对象序列化为字典"""
|
||
return {
|
||
"id": t.id,
|
||
"product_id": t.product_id,
|
||
"tier_code": t.tier_code,
|
||
"tier_name": t.tier_name,
|
||
"price": float(t.price or 0),
|
||
"price_unit": t.price_unit,
|
||
"remark": t.remark,
|
||
"status": t.status,
|
||
}
|