405 lines
14 KiB
Python
405 lines
14 KiB
Python
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.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),
|
|
status: int | None = Query(default=None),
|
|
page_no: int = Query(default=1),
|
|
page_size: int = Query(default=20),
|
|
session: Session = Depends(get_db_session),
|
|
_user: dict = Depends(require_roles("admin", "manager")),
|
|
_perm: dict = Depends(require_permissions("master-data:list")),
|
|
) -> dict:
|
|
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,
|
|
session: Session = Depends(get_db_session),
|
|
_user: dict = Depends(require_roles("admin", "manager")),
|
|
) -> dict:
|
|
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)
|
|
return success_payload(_rule_to_dict(rule))
|
|
|
|
|
|
@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:
|
|
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()
|
|
session.refresh(rule)
|
|
return success_payload(_rule_to_dict(rule))
|
|
|
|
|
|
@router.put("/pricing-rules/{rule_id}")
|
|
def update_pricing_rule(
|
|
rule_id: int,
|
|
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 = 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()
|
|
session.refresh(rule)
|
|
return success_payload(_rule_to_dict(rule))
|
|
|
|
|
|
@router.delete("/pricing-rules/{rule_id}")
|
|
def delete_pricing_rule(
|
|
rule_id: int,
|
|
session: Session = Depends(get_db_session),
|
|
_user: dict = Depends(require_roles("admin", "manager")),
|
|
_perm: dict = Depends(require_permissions("master-data:update")),
|
|
) -> dict:
|
|
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()
|
|
return success_payload({"deleted": True})
|
|
|
|
|
|
# ======================================================================
|
|
# 供应商成本 CRUD
|
|
# ======================================================================
|
|
|
|
@router.get("/supplier-costs")
|
|
def list_supplier_costs(
|
|
product_id: int = Query(...),
|
|
session: Session = Depends(get_db_session),
|
|
_user: dict = Depends(require_roles("admin", "manager")),
|
|
) -> dict:
|
|
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:
|
|
cost = SupplierProductCost(**payload.model_dump())
|
|
session.add(cost)
|
|
session.commit()
|
|
session.refresh(cost)
|
|
return success_payload(_cost_to_dict(cost))
|
|
|
|
|
|
@router.put("/supplier-costs/{cost_id}")
|
|
def update_supplier_cost(
|
|
cost_id: int,
|
|
payload: UpdateSupplierCostRequest,
|
|
session: Session = Depends(get_db_session),
|
|
_user: dict = Depends(require_roles("admin", "manager")),
|
|
) -> dict:
|
|
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()
|
|
session.refresh(cost)
|
|
return success_payload(_cost_to_dict(cost))
|
|
|
|
|
|
@router.delete("/supplier-costs/{cost_id}")
|
|
def delete_supplier_cost(
|
|
cost_id: int,
|
|
session: Session = Depends(get_db_session),
|
|
_user: dict = Depends(require_roles("admin", "manager")),
|
|
) -> dict:
|
|
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()
|
|
return success_payload({"deleted": True})
|
|
|
|
|
|
# ======================================================================
|
|
# 价格层级 CRUD
|
|
# ======================================================================
|
|
|
|
@router.get("/price-tiers")
|
|
def list_price_tiers(
|
|
product_id: int = Query(...),
|
|
session: Session = Depends(get_db_session),
|
|
_user: dict = Depends(require_roles("admin", "manager", "salesman")),
|
|
) -> dict:
|
|
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:
|
|
tier = ProductPriceTier(**payload.model_dump())
|
|
session.add(tier)
|
|
session.commit()
|
|
session.refresh(tier)
|
|
return success_payload(_tier_to_dict(tier))
|
|
|
|
|
|
@router.put("/price-tiers/{tier_id}")
|
|
def update_price_tier(
|
|
tier_id: int,
|
|
payload: UpdatePriceTierRequest,
|
|
session: Session = Depends(get_db_session),
|
|
_user: dict = Depends(require_roles("admin", "manager")),
|
|
) -> dict:
|
|
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()
|
|
session.refresh(tier)
|
|
return success_payload(_tier_to_dict(tier))
|
|
|
|
|
|
@router.delete("/price-tiers/{tier_id}")
|
|
def delete_price_tier(
|
|
tier_id: int,
|
|
session: Session = Depends(get_db_session),
|
|
_user: dict = Depends(require_roles("admin", "manager")),
|
|
) -> dict:
|
|
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()
|
|
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:
|
|
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,
|
|
})
|
|
|
|
|
|
# ======================================================================
|
|
# 序列化工具
|
|
# ======================================================================
|
|
|
|
def _rule_to_dict(r: ProductPricingRule) -> dict:
|
|
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,
|
|
"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:
|
|
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:
|
|
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,
|
|
}
|