shanxia2.0/backend/tests/test_pl_plan_archive_export.py

from decimal import Decimal
from unittest.mock import MagicMock, patch

from app.schemas.pl.plan import PlanDeliveryRead, PlanLoadRead, PlanStopProductRead
from app.services.pl_internal_customer import INTERNAL_CUSTOMER_NAMES
from app.services.pl_plan_archive_export import (
    ARCHIVE_MERGED_SHEET_TITLE,
    _plan_common_export_fields,
    append_archive_list_merge_sheet,
    build_plan_archive_export_rows,
    prepare_archive_list_display_rows,
    resolve_archive_list_io_column,
)
from app.utils.excel_export import ExportColumn


def _stop_product(**kwargs) -> PlanStopProductRead:
    defaults = {
        "id": "sp1",
        "planId": "p1",
        "tenantNo": "t1",
        "productId": "prod1",
        "productNo": "P001",
        "productName": "甲醇",
        "model": "99%",
        "unit": "吨",
        "packageSpec": "散装",
        "planQty": Decimal("10"),
        "createTime": "2026-08-27T00:00:00",
        "updateTime": "2026-08-27T00:00:00",
    }
    defaults.update(kwargs)
    return PlanStopProductRead(**defaults)


def test_resolve_archive_list_io_column():
    internal = next(iter(INTERNAL_CUSTOMER_NAMES))
    assert resolve_archive_list_io_column("load", internal) == "outbound"
    assert resolve_archive_list_io_column("load", "外部客户") == "inbound"
    assert resolve_archive_list_io_column("delivery", internal) == "inbound"
    assert resolve_archive_list_io_column("delivery", "外部客户") == "outbound"


# 配对宽表 helper（_append_paired_product_rows / _append_stop_product_rows）单测已删除：
# 归档导出改为段次×产品多行，不再断言同排 loadPlanQty+deliveryPlanQty / deliveryStopKind。
# 品牌别名与内部客户合同占位改由 build_plan_archive_export_rows 覆盖。


def _bare_plan(**overrides):
    plan = MagicMock()
    plan.id = "p1"
    plan.plan_no = "PL001"
    plan.contract_no = "C1"
    plan.plan_date = "2026-08-27"
    plan.archive_status = "archived"
    plan.follow_status = "arranged"
    plan.plan_type = "direct"
    plan.license_plate = "皖A12345"
    plan.trailer_no = None
    plan.erp_entered = 0
    plan.freight_amount = None
    plan.remark = ""
    plan.transport_id = None
    plan.customer_name = "外部客户"
    plan.supplier_name = None
    plan.destination_name = None
    plan.delivery_time = None
    for key, value in overrides.items():
        setattr(plan, key, value)
    return plan


def _archive_rows(
    plan,
    load,
    deliveries,
    *,
    stop_kind=None,
    legacy_products=None,
    alias_maps=None,
):
    if legacy_products is None:
        legacy_products = {}
    if alias_maps is None:
        alias_maps = ({}, {})
    with patch(
        "app.services.pl_plan_archive_export.load_plan_stops_grouped",
        return_value={plan.id: (load, deliveries)},
    ), patch(
        "app.services.pl_plan_archive_export._load_legacy_products_by_plan",
        return_value=legacy_products,
    ), patch(
        "app.services.pl_plan_archive_export._load_contract_nos_by_plan",
        return_value={plan.id: []},
    ), patch(
        "app.services.pl_plan_archive_export.load_crm_product_alias_maps",
        return_value=alias_maps,
    ):
        return build_plan_archive_export_rows(
            MagicMock(), "t1", [plan], {}, stop_kind=stop_kind
        )


def test_build_plan_archive_export_rows_segment_rows_and_single_side():
    """默认按段次×产品展开：1 个装货产品 + 1 个送货产品 = 2 行；单侧筛选只留对应段。"""
    internal = next(iter(INTERNAL_CUSTOMER_NAMES))
    load_product = _stop_product(
        planQty=Decimal("10"),
        inboundQty=Decimal("10"),
        inboundPrice=Decimal("100"),
    )
    delivery_product = _stop_product(
        id="dp1",
        planQty=Decimal("12"),
        outboundQty=Decimal("9"),
        outboundPrice=Decimal("110"),
        outboundAmount=Decimal("990"),
    )
    load = PlanLoadRead(
        id="load1",
        planId="p1",
        tenantNo="t1",
        customerName=internal,
        address="装货地址",
        loadTime="2026-08-27T08:00:00",
        products=[load_product],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    delivery = PlanDeliveryRead(
        id="del1",
        planId="p1",
        tenantNo="t1",
        seq=2,
        customerName="外部客户",
        address="送货地址",
        deliveryTime="2026-08-28T10:00:00",
        products=[delivery_product],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    plan = _bare_plan()

    rows = _archive_rows(plan, load, [delivery], stop_kind=None)
    load_only = _archive_rows(plan, load, [delivery], stop_kind="load")
    delivery_only = _archive_rows(plan, load, [delivery], stop_kind="delivery")

    assert len(rows) == 2
    assert [row["stopSegment"] for row in rows] == ["装货", "送货 #2"]
    assert rows[0]["productName"] == "甲醇"
    assert rows[1]["productName"] == "甲醇"
    assert rows[0]["loadCustomer"] == internal
    assert rows[0]["deliveryCustomer"] == ""
    assert rows[0]["loadAddress"] == "装货地址"
    assert rows[0]["loadTime"] == "2026-08-27"
    assert rows[0]["deliveryAddress"] == ""
    assert rows[0]["deliveryTime"] == ""
    assert rows[0]["planQty"] == Decimal("10")
    assert rows[0]["actualQty"] == Decimal("10")
    assert rows[0]["unitPrice"] == Decimal("100")
    assert rows[0]["amount"] == Decimal("1000")
    assert rows[1]["loadCustomer"] == ""
    assert rows[1]["deliveryCustomer"] == "外部客户"
    assert rows[1]["loadAddress"] == ""
    assert rows[1]["loadTime"] == ""
    assert rows[1]["deliveryAddress"] == "送货地址"
    assert rows[1]["deliveryTime"] == "2026-08-28"
    assert rows[1]["planQty"] == Decimal("12")
    assert rows[1]["actualQty"] == Decimal("9")
    assert rows[1]["unitPrice"] == Decimal("110")
    assert rows[1]["amount"] == Decimal("990")
    assert rows[0].get("loadPlanQty") is None
    assert rows[0].get("deliveryPlanQty") is None
    assert rows[1].get("loadPlanQty") is None
    assert rows[1].get("deliveryPlanQty") is None
    assert not rows[0].get("deliveryStopKind")
    assert not rows[1].get("deliveryStopKind")

    assert len(load_only) == 1
    assert load_only[0]["stopSegment"] == "装货"
    assert load_only[0]["loadCustomer"] == internal
    assert load_only[0]["deliveryCustomer"] == ""
    assert load_only[0]["planQty"] == Decimal("10")
    assert load_only[0]["actualQty"] == Decimal("10")
    assert load_only[0]["unitPrice"] == Decimal("100")
    assert load_only[0]["amount"] == Decimal("1000")
    assert load_only[0].get("loadPlanQty") is None

    assert len(delivery_only) == 1
    assert delivery_only[0]["stopSegment"] == "送货 #2"
    assert delivery_only[0]["deliveryCustomer"] == "外部客户"
    assert delivery_only[0]["loadCustomer"] == ""
    assert delivery_only[0]["planQty"] == Decimal("12")
    assert delivery_only[0]["actualQty"] == Decimal("9")
    assert delivery_only[0]["unitPrice"] == Decimal("110")
    assert delivery_only[0]["amount"] == Decimal("990")
    assert delivery_only[0].get("deliveryPlanQty") is None


class _FakePlan:
    id = "p1"
    plan_no = "PL001"
    contract_no = "SX/X-JT20260827001"
    plan_date = "2026-08-27"
    archive_status = "archived"
    follow_status = "arranged"
    plan_type = "direct"
    license_plate = "皖A12345"
    trailer_no = "挂001"
    erp_entered = 1
    freight_amount = Decimal("500")
    remark = "测试"


def test_plan_common_export_fields():
    fields = _plan_common_export_fields(_FakePlan(), None)
    assert fields["planId"] == "p1"
    assert fields["planNo"] == "PL001"
    assert fields["contractNo"] == "SX/X-JT20260827001"
    assert fields["archiveStatus"] == "已归档"
    assert fields["followStatus"] == "已安排"
    assert fields["planType"] == "直销单"
    assert fields["licensePlate"] == "皖A12345挂001"
    assert fields["erpEntered"] == "是"


def test_build_plan_archive_export_rows_includes_brand_and_alias():
    plan_product = MagicMock()
    plan_product.id = "pp1"
    plan_product.product_id = "prod1"
    plan_product.product_name = "甲醇别名"
    plan_product.brand_mode = "required"
    plan_product.brand_name = "君正"

    load_product = _stop_product(planProductId="pp1", productName="甲醇别名")
    delivery_product = _stop_product(
        id="dp1",
        planProductId="pp1",
        productName="甲醇别名",
        outboundQty=Decimal("9"),
        outboundPrice=Decimal("110"),
    )
    load = PlanLoadRead(
        id="load1",
        planId="p1",
        tenantNo="t1",
        customerName="装货客户",
        products=[load_product],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    delivery = PlanDeliveryRead(
        id="del1",
        planId="p1",
        tenantNo="t1",
        seq=1,
        customerName="送货客户",
        products=[delivery_product],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    rows = _archive_rows(
        _bare_plan(),
        load,
        [delivery],
        legacy_products={"p1": [plan_product]},
        alias_maps=({"prod1": "甲醇"}, {"prod1": ["甲醇别名"]}),
    )
    assert len(rows) == 2
    assert [row["stopSegment"] for row in rows] == ["装货", "送货 #1"]
    assert rows[0]["loadCustomer"] == "装货客户"
    assert rows[0]["deliveryCustomer"] == ""
    assert rows[1]["loadCustomer"] == ""
    assert rows[1]["deliveryCustomer"] == "送货客户"
    for row in rows:
        assert row["productName"] == "甲醇"
        assert row["brandName"] == "君正"
        assert row["aliasName"] == "甲醇别名"


def test_plan_common_export_fields_merges_link_contract_nos():
    fields = _plan_common_export_fields(
        _FakePlan(),
        None,
        contract_nos=["SX/X-JT20260827001", "SX/X-JT20260827002"],
    )
    assert fields["contractNo"] == "SX/X-JT20260827001、SX/X-JT20260827002"


def test_build_plan_archive_export_rows_internal_customer_contract_per_segment():
    """内部客户缺合同：装货行合同为出库，送货行为入库，不再拼在同一行。"""
    internal = next(iter(INTERNAL_CUSTOMER_NAMES))
    load_product = _stop_product(id="lp1", inboundQty=Decimal("10"))
    delivery_product = _stop_product(id="dp1", outboundQty=Decimal("10"))
    load = PlanLoadRead(
        id="load1",
        planId="p1",
        tenantNo="t1",
        customerName=internal,
        products=[load_product],
        contracts=[],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    delivery = PlanDeliveryRead(
        id="del1",
        planId="p1",
        tenantNo="t1",
        seq=1,
        customerName=internal,
        products=[delivery_product],
        contracts=[],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    rows = _archive_rows(_bare_plan(contract_no=""), load, [delivery])
    assert len(rows) == 2
    assert [row["stopSegment"] for row in rows] == ["装货", "送货 #1"]
    assert rows[0]["contractNo"] == "出库"
    assert rows[0]["loadCustomer"] == internal
    assert rows[0]["deliveryCustomer"] == ""
    assert rows[1]["contractNo"] == "入库"
    assert rows[1]["deliveryCustomer"] == internal
    assert rows[1]["loadCustomer"] == ""


class _CellResolver:
    def row_to_cells(self, row, columns):
        return ["" if row.get(column.field) is None else row.get(column.field) for column in columns]


_MERGE_COLUMNS = [
    ExportColumn(title="序号", field="rowNo"),
    ExportColumn(title="计划编号", field="planNo"),
    ExportColumn(title="装货客户", field="loadCustomer"),
    ExportColumn(title="送货客户", field="deliveryCustomer"),
    ExportColumn(title="合同编号", field="contractNo"),
    ExportColumn(title="产品名称", field="productName"),
    ExportColumn(title="计划量", field="planQty"),
]


def _merged_ranges(sheet):
    return {str(item) for item in sheet.merged_cells.ranges}


def test_prepare_single_delivery_puts_delivery_customer_on_first_row():
    internal = next(iter(INTERNAL_CUSTOMER_NAMES))
    load = PlanLoadRead(
        id="load1",
        planId="p1",
        tenantNo="t1",
        customerName="合肥玉坤",
        address="装货地址",
        loadTime="2026-10-09T00:00:00",
        products=[
            _stop_product(id="lp1", productName="片碱", planQty=Decimal("0.9")),
            _stop_product(id="lp2", productName="熟石灰", planQty=Decimal("1.75")),
        ],
        contracts=[],
        createTime="2026-10-09T00:00:00",
        updateTime="2026-10-09T00:00:00",
    )
    delivery = PlanDeliveryRead(
        id="del1",
        planId="p1",
        tenantNo="t1",
        seq=1,
        customerName="凤阳京环",
        address="送货地址",
        deliveryTime="2026-10-09T00:00:00",
        products=[
            _stop_product(id="dp1", productName="片碱", planQty=Decimal("0.9")),
            _stop_product(id="dp2", productName="熟石灰", planQty=Decimal("1.75")),
        ],
        contracts=[],
        createTime="2026-10-09T00:00:00",
        updateTime="2026-10-09T00:00:00",
    )
    rows = _archive_rows(_bare_plan(plan_type="direct", customer_name=internal), load, [delivery])
    display = prepare_archive_list_display_rows(rows)
    assert [row["rowNo"] for row in display] == [1, 1, 1, 1]
    assert display[0]["deliveryCustomer"] == "凤阳京环"
    assert display[0]["loadCustomer"] == "合肥玉坤"
    assert [row["productName"] for row in display] == ["片碱", "熟石灰", "片碱", "熟石灰"]
    assert display[0]["_contractMergeKey"] != display[2]["_contractMergeKey"]


def test_merge_sheet_matches_list_rowspan():
    from openpyxl import Workbook

    internal = next(iter(INTERNAL_CUSTOMER_NAMES))
    load = PlanLoadRead(
        id="load1",
        planId="p1",
        tenantNo="t1",
        customerName=internal,
        products=[
            _stop_product(id="lp1", productName="氯化钠", planQty=Decimal("0.45")),
            _stop_product(id="lp2", productName="次氯酸钠", planQty=Decimal("1.17")),
        ],
        contracts=[],
        createTime="2026-10-07T00:00:00",
        updateTime="2026-10-07T00:00:00",
    )
    delivery_a = PlanDeliveryRead(
        id="del1",
        planId="p1",
        tenantNo="t1",
        seq=1,
        customerName="安庆奇创",
        deliveryTime="2026-10-07T00:00:00",
        products=[_stop_product(id="dp1", productName="次氯酸钠", planQty=Decimal("1.17"))],
        contracts=[],
        createTime="2026-10-07T00:00:00",
        updateTime="2026-10-07T00:00:00",
    )
    delivery_b = PlanDeliveryRead(
        id="del2",
        planId="p1",
        tenantNo="t1",
        seq=2,
        customerName="安庆奇创",
        deliveryTime="2026-10-07T00:00:00",
        products=[_stop_product(id="dp2", productName="氯化钠", planQty=Decimal("0.45"))],
        contracts=[],
        createTime="2026-10-07T00:00:00",
        updateTime="2026-10-07T00:00:00",
    )
    rows = _archive_rows(_bare_plan(plan_type="outbound"), load, [delivery_a, delivery_b])
    workbook = Workbook()
    workbook.active.title = "计划归档"
    append_archive_list_merge_sheet(workbook, _MERGE_COLUMNS, rows, _CellResolver())
    sheet = workbook[ARCHIVE_MERGED_SHEET_TITLE]
    assert [cell.value for cell in sheet[1]] == [column.title for column in _MERGE_COLUMNS]
    assert sheet["F2"].value == "氯化钠"
    assert sheet["F5"].value == "氯化钠"
    ranges = _merged_ranges(sheet)
    assert "A2:A5" in ranges
    assert "B2:B5" in ranges
    assert "C2:C5" in ranges
    assert "D2:D3" in ranges
    assert "D4:D5" not in ranges
    assert sheet["D2"].value == "—"
    assert sheet["D4"].value == "安庆奇创"
    assert sheet["D5"].value == "安庆奇创"
    assert "F2:F5" not in ranges