100 lines
5.1 KiB
Python
100 lines
5.1 KiB
Python
"""报价路由：从 drawing_data 算料计价（对齐目标站 offer-price 语义，自研实现）。"""
|
|
from fastapi import APIRouter, Depends
|
|
from pydantic import BaseModel, Field
|
|
|
|
from huamenchuang.core.deps import CurrentUser, DbSession, require_module
|
|
from huamenchuang.core.response import fail, ok
|
|
from huamenchuang.services.quote_service import quote as calc_quote
|
|
|
|
router = APIRouter(prefix="/offer-price", tags=["报价"],
|
|
dependencies=[Depends(require_module("quote"))])
|
|
|
|
|
|
class QuoteRequest(BaseModel):
|
|
drawing_data: str | dict | None = Field(default=None, description="画布 drawing_data(与 template_id/order_id 三选一)")
|
|
template_id: int | None = Field(default=None, description="或直接指定样板")
|
|
order_id: int | None = Field(default=None, description="或对已有订单报价")
|
|
series_id: int | None = Field(default=None, description="材料系列(成本系列模板): 附加该系列的成本估算供商家选材")
|
|
prices: dict | None = Field(default=None, description="价格本覆盖 {frame,mullion,sash,glass,hardware}")
|
|
use_price_book: bool = Field(default=False, description="从产品管理价格本取价(按材料类别均价)")
|
|
use_scripts: bool = Field(default=False, description="启用价格脚本计价(公司启用脚本覆盖默认单价/总价)")
|
|
scripts: dict | None = Field(default=None, description="临时脚本 {target: formula}, 优先于脚本库")
|
|
|
|
|
|
async def _resolve_drawing(db, body: QuoteRequest) -> str | dict | None:
|
|
if body.drawing_data:
|
|
return body.drawing_data
|
|
if body.template_id:
|
|
from sqlalchemy import select
|
|
from huamenchuang.models.windoor_template import WindoorTemplate
|
|
t = (await db.execute(select(WindoorTemplate).where(WindoorTemplate.id == body.template_id))).scalar_one_or_none()
|
|
return t.drawing_data if t else None
|
|
if body.order_id:
|
|
from huamenchuang.models.order import Order
|
|
o = await db.get(Order, body.order_id)
|
|
return o.drawing_data if o else None
|
|
return None
|
|
|
|
|
|
async def _price_book(db, company_id: int) -> dict | None:
|
|
"""从产品价格本推导报价参数：各类材料均价。"""
|
|
from sqlalchemy import func, select
|
|
from huamenchuang.models.business import Product
|
|
rows = (await db.execute(
|
|
select(Product.material, func.avg(Product.price))
|
|
.where(Product.company_id == company_id, Product.enabled == 1)
|
|
.group_by(Product.material))).all()
|
|
if not rows:
|
|
return None
|
|
avg = {m: float(p or 0) for m, p in rows}
|
|
prices: dict = {}
|
|
if avg.get("profile"):
|
|
prices["frame"] = prices["mullion"] = prices["sash"] = round(avg["profile"], 2)
|
|
if avg.get("glass"):
|
|
prices["glass"] = round(avg["glass"], 2)
|
|
if avg.get("hardware"):
|
|
prices["hardware"] = round(avg["hardware"], 2)
|
|
return prices or None
|
|
|
|
|
|
@router.post("")
|
|
async def offer_price(body: QuoteRequest, user: CurrentUser, db: DbSession):
|
|
"""算料报价：型材清单 + 玻璃面积 + 五金 + 总价（可选价格脚本计价）。"""
|
|
drawing = await _resolve_drawing(db, body)
|
|
if drawing is None:
|
|
return fail("未找到图纸数据")
|
|
prices = body.prices
|
|
if prices is None and body.use_price_book:
|
|
prices = await _price_book(db, user.company_id)
|
|
# 脚本解析：显式传入的临时脚本优先，其次取脚本库中启用脚本
|
|
scripts = body.scripts
|
|
if scripts is None and body.use_scripts:
|
|
from sqlalchemy import select as _select
|
|
from huamenchuang.api.script import load_script_map
|
|
from huamenchuang.models.business import PriceScript
|
|
rows = list((await db.execute(
|
|
_select(PriceScript).where(PriceScript.company_id == user.company_id))).scalars())
|
|
scripts = load_script_map(rows) or None
|
|
try:
|
|
result = calc_quote(drawing, prices, scripts)
|
|
except Exception as e: # 公式/图纸解析错误 → 用户可读失败
|
|
msg = str(e) if "公式" in str(e) or "未知变量" in str(e) else "图纸数据解析失败"
|
|
return fail(msg)
|
|
# 材料系列成本参考：商家选材依据(系列模板八分区估算, 按图纸实际尺寸)
|
|
if body.series_id:
|
|
from huamenchuang.models.business import CostSeriesTemplate
|
|
from huamenchuang.services import costbook_service as _svc
|
|
tpl = await db.get(CostSeriesTemplate, body.series_id)
|
|
if tpl is not None and tpl.company_id == user.company_id:
|
|
w = result.get("width_mm") or 1600
|
|
h = result.get("height_mm") or 1600
|
|
est = _svc.estimate(tpl, w, h, 1)
|
|
result["series"] = {"id": tpl.id, "name": tpl.name, "series": tpl.series, "wall_thick": tpl.wall_thick}
|
|
result["series_estimate"] = {
|
|
"width": w, "height": h, "total": est["total"], "per_sqm": est["per_sqm"],
|
|
"profile_cost": est["profile_cost"], "glass_cost": est["glass_cost"],
|
|
"hardware_cost": est["hardware_cost"], "labor_cost": est["labor_cost"],
|
|
"margin_ladder": est["margin_ladder"],
|
|
}
|
|
return ok(data=result, message="报价成功")
|