diff --git a/backend/app/services/pl_feedback_brand_qty.py b/backend/app/services/pl_feedback_brand_qty.py new file mode 100644 index 000000000..f8696e0d3 --- /dev/null +++ b/backend/app/services/pl_feedback_brand_qty.py @@ -0,0 +1,217 @@ +"""现场入出库明细与运输回传的品牌数量校验与持久化""" +from decimal import Decimal +from typing import Any, Optional, Type + +from sqlalchemy.orm import Session + +from app.core.utils import generate_uuid +from app.services.pl_brand_archive import validate_brand_qty_rows + + +def format_brand_qty_summary(rows: list[dict]) -> Optional[str]: + parts = [] + for r in rows: + qty = Decimal(str(r.get("qty") or 0)) + if qty <= 0: + continue + name = (r.get("brandName") or r.get("brand_name") or "").strip() or "未命名品牌" + qty_text = format(qty.normalize(), "f") + parts.append(f"{name} {qty_text}") + return " / ".join(parts) if parts else None + + +def require_brand_qtys_if_needed( + *, + brand_mode: Optional[str], + required_brand_id: Optional[str], + actual_qty: Decimal, + rows: list[dict], + enforce: bool, +) -> None: + if not brand_mode: + return + if not enforce: + return + positive = [r for r in rows if Decimal(str(r.get("qty") or 0)) > 0] + if not positive: + raise ValueError("有品牌产品须填写至少一个品牌数量") + validate_brand_qty_rows( + brand_mode=brand_mode, + required_brand_id=required_brand_id, + actual_qty=Decimal(str(actual_qty)), + rows=rows, + ) + + +def persist_io_item_brand_qtys( + db: Session, + *, + model: Type[Any], + tenant_no: str, + order_id: str, + order_item_id: str, + plan_product_id: str, + brand_mode: Optional[str], + rows: list[dict], + user_id: Optional[str] = None, +) -> None: + existing = ( + db.query(model) + .filter( + model.order_item_id == order_item_id, + model.tenant_no == tenant_no, + model.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 + db.add( + model( + id=generate_uuid(), + tenant_no=tenant_no, + order_id=order_id, + order_item_id=order_item_id, + plan_product_id=plan_product_id, + brand_id=item.get("brandId") or item.get("brand_id") or "", + brand_name=item.get("brandName") or item.get("brand_name"), + qty=qty, + order_num=idx, + create_by=user_id, + create_dept_id="", + update_by=user_id, + ) + ) + + +def _dict_field(item: dict, *keys: str): + for key in keys: + if key in item and item[key] is not None: + return item[key] + return None + + +def persist_transport_product_feedback( + db: Session, + *, + tenant_no: str, + transport, + plan_id: str, + products: list[dict], + user_id: Optional[str] = None, + enforce_brand: bool = True, +) -> None: + from app.models.pl_models import ( + PlPlanProduct, + PlTransportProductBrandQty, + PlTransportProductFeedback, + ) + + transport_id = transport.id + + for row in ( + db.query(PlTransportProductFeedback) + .filter( + PlTransportProductFeedback.transport_id == transport_id, + PlTransportProductFeedback.tenant_no == tenant_no, + PlTransportProductFeedback.del_flag == 0, + ) + .all() + ): + row.del_flag = 1 + row.update_by = user_id + + for row in ( + db.query(PlTransportProductBrandQty) + .filter( + PlTransportProductBrandQty.transport_id == transport_id, + PlTransportProductBrandQty.tenant_no == tenant_no, + PlTransportProductBrandQty.del_flag == 0, + ) + .all() + ): + row.del_flag = 1 + row.update_by = user_id + + total_qty = Decimal("0") + + for idx, item in enumerate(products): + plan_product_id = _dict_field(item, "planProductId", "plan_product_id") + actual_qty = Decimal(str(_dict_field(item, "actualQty", "actual_qty") or 0)) + brand_qtys = _dict_field(item, "brandQtys", "brand_qtys") or [] + total_qty += actual_qty + + plan_product = ( + db.query(PlPlanProduct) + .filter( + PlPlanProduct.id == plan_product_id, + PlPlanProduct.tenant_no == tenant_no, + PlPlanProduct.del_flag == 0, + ) + .first() + ) + + brand_mode = _dict_field(item, "brandMode", "brand_mode") + required_brand_id = _dict_field(item, "brandId", "brand_id") + if plan_product: + brand_mode = brand_mode or plan_product.brand_mode + required_brand_id = required_brand_id if required_brand_id is not None else plan_product.brand_id + + require_brand_qtys_if_needed( + brand_mode=brand_mode, + required_brand_id=required_brand_id, + actual_qty=actual_qty, + rows=brand_qtys, + enforce=enforce_brand, + ) + + feedback_id = generate_uuid() + db.add( + PlTransportProductFeedback( + id=feedback_id, + tenant_no=tenant_no, + transport_id=transport_id, + plan_id=plan_id, + plan_product_id=plan_product_id, + actual_qty=actual_qty, + order_num=idx, + create_by=user_id, + create_dept_id="", + update_by=user_id, + ) + ) + + if not brand_mode: + continue + + for bidx, bitem in enumerate(brand_qtys): + qty = Decimal(str(bitem.get("qty") or 0)) + if qty <= 0: + continue + db.add( + PlTransportProductBrandQty( + id=generate_uuid(), + tenant_no=tenant_no, + transport_id=transport_id, + plan_product_id=plan_product_id, + feedback_id=feedback_id, + brand_id=bitem.get("brandId") or bitem.get("brand_id") or "", + brand_name=bitem.get("brandName") or bitem.get("brand_name"), + qty=qty, + order_num=bidx, + create_by=user_id, + create_dept_id="", + update_by=user_id, + ) + ) + + transport.feedback_qty = total_qty diff --git a/backend/tests/test_pl_feedback_brand_qty.py b/backend/tests/test_pl_feedback_brand_qty.py new file mode 100644 index 000000000..30716bf53 --- /dev/null +++ b/backend/tests/test_pl_feedback_brand_qty.py @@ -0,0 +1,75 @@ +from decimal import Decimal + +import pytest + +from app.services.pl_feedback_brand_qty import format_brand_qty_summary, require_brand_qtys_if_needed + + +def test_format_brand_qty_summary(): + s = format_brand_qty_summary([ + {"brandName": "君正", "qty": Decimal("16")}, + {"brandName": "红四方", "qty": Decimal("12.000")}, + ]) + assert s == "君正 16 / 红四方 12" + + +def test_format_brand_qty_summary_empty(): + assert format_brand_qty_summary([]) is None + + +def test_format_brand_qty_summary_zero_only(): + assert format_brand_qty_summary([{"brandName": "君正", "qty": Decimal("0")}]) is None + + +def test_require_skips_when_no_brand_mode(): + require_brand_qtys_if_needed( + brand_mode=None, + required_brand_id=None, + actual_qty=Decimal("10"), + rows=[], + enforce=True, + ) + + +def test_require_skips_when_not_enforced(): + require_brand_qtys_if_needed( + brand_mode="any", + required_brand_id=None, + actual_qty=Decimal("10"), + rows=[], + enforce=False, + ) + + +def test_require_enforces_when_completed(): + with pytest.raises(ValueError, match="至少一个品牌数量"): + require_brand_qtys_if_needed( + brand_mode="any", + required_brand_id=None, + actual_qty=Decimal("10"), + rows=[], + enforce=True, + ) + + +def test_require_calls_validate_when_positive_rows(monkeypatch): + called = {} + + def mock_validate(**kwargs): + called.update(kwargs) + + monkeypatch.setattr( + "app.services.pl_feedback_brand_qty.validate_brand_qty_rows", + mock_validate, + ) + rows = [{"brandId": "b1", "brandName": "君正", "qty": Decimal("10")}] + require_brand_qtys_if_needed( + brand_mode="any", + required_brand_id=None, + actual_qty=Decimal("10"), + rows=rows, + enforce=True, + ) + assert called["brand_mode"] == "any" + assert called["actual_qty"] == Decimal("10") + assert called["rows"] == rows