85 lines
2.6 KiB
Python
85 lines
2.6 KiB
Python
"""计划归档按品牌数量校验与持久化"""
|
|
from decimal import Decimal
|
|
from typing import Any, Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.utils import generate_uuid
|
|
from app.models.pl_models import PlPlanProductBrandQty
|
|
|
|
|
|
def validate_brand_qty_rows(
|
|
*,
|
|
brand_mode: Optional[str],
|
|
required_brand_id: Optional[str],
|
|
actual_qty: Decimal,
|
|
rows: list[dict],
|
|
) -> None:
|
|
actual_qty = Decimal(str(actual_qty))
|
|
if not brand_mode:
|
|
return
|
|
positive = [r for r in rows if Decimal(str(r.get("qty") or 0)) > 0]
|
|
if brand_mode == "required":
|
|
if len(positive) != 1 or positive[0].get("brandId") != required_brand_id:
|
|
raise ValueError("指定品牌时只能填写该品牌数量")
|
|
if Decimal(str(positive[0]["qty"])) != actual_qty:
|
|
raise ValueError("指定品牌数量必须等于归档实际数量")
|
|
return
|
|
if brand_mode == "any":
|
|
total = sum((Decimal(str(r.get("qty") or 0)) for r in rows), Decimal("0"))
|
|
if total != actual_qty:
|
|
raise ValueError("各品牌数量合计必须等于归档实际数量")
|
|
return
|
|
raise ValueError("无效的 brand_mode")
|
|
|
|
|
|
def persist_plan_product_brand_qtys(
|
|
db: Session,
|
|
*,
|
|
tenant_no: str,
|
|
plan_id: str,
|
|
plan_product_id: str,
|
|
brand_mode: Optional[str],
|
|
rows: list[dict[str, Any]],
|
|
user_id: str,
|
|
dept_id: str = "",
|
|
) -> None:
|
|
"""软删旧品牌量行后按归档结果插入；无品牌约定则只软删不插入。"""
|
|
existing = (
|
|
db.query(PlPlanProductBrandQty)
|
|
.filter(
|
|
PlPlanProductBrandQty.plan_product_id == plan_product_id,
|
|
PlPlanProductBrandQty.tenant_no == tenant_no,
|
|
PlPlanProductBrandQty.del_flag == 0,
|
|
)
|
|
.all()
|
|
)
|
|
for row in existing:
|
|
row.del_flag = 1
|
|
row.update_by = user_id
|
|
|
|
if not brand_mode:
|
|
return
|
|
|
|
for idx, item in enumerate(rows):
|
|
qty = Decimal(str(item.get("qty") or 0))
|
|
if qty <= 0:
|
|
continue
|
|
brand_id = item.get("brandId") or item.get("brand_id") or ""
|
|
brand_name = item.get("brandName") or item.get("brand_name")
|
|
db.add(
|
|
PlPlanProductBrandQty(
|
|
id=generate_uuid(),
|
|
tenant_no=tenant_no,
|
|
plan_id=plan_id,
|
|
plan_product_id=plan_product_id,
|
|
brand_id=brand_id,
|
|
brand_name=brand_name,
|
|
qty=qty,
|
|
order_num=idx,
|
|
create_by=user_id,
|
|
create_dept_id=dept_id or "",
|
|
update_by=user_id,
|
|
)
|
|
)
|