shanxia2.0/backend/tests/test_pl_plan_list_export.py

"""运输计划列表导出：产品名称 / 品牌 / 别名 / 状态中文 / 列表列对齐。"""
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

from app.api.pl.plan import (
    PLAN_EXPORT_COLUMNS,
    _build_plan_list_export_rows,
    _format_plan_list_export_brand,
    _normalize_plan_export_columns,
    _plan_list_export_filename_stem,
)
from app.utils.excel_export import filter_export_columns


def test_normalize_plan_export_columns_maps_list_props():
    assert _normalize_plan_export_columns(
        [
            "productBrand",
            "productAlias",
            "productName",
            "planNo",
            "productSpec",
            "productPackageSpec",
            "productUnit",
            "planQty",
            "actualQty",
        ]
    ) == [
        "brandName",
        "aliasName",
        "productName",
        "planNo",
        "model",
        "packageSpec",
        "unit",
        "planQty",
        "actualQty",
    ]
    assert _normalize_plan_export_columns(None) is None


def test_format_plan_list_export_brand_only_required():
    required = SimpleNamespace(brand_mode="required", brand_name="君正")
    any_mode = SimpleNamespace(brand_mode="any", brand_name="君正")
    assert _format_plan_list_export_brand(required) == "君正"
    assert _format_plan_list_export_brand(any_mode) == ""


def test_filter_export_columns_keeps_list_props_after_normalize():
    requested = _normalize_plan_export_columns(
        [
            "planNo",
            "belongsCompany",
            "followStatus",
            "planType",
            "loadCustomer",
            "deliveryCustomer",
            "productName",
            "productBrand",
            "productAlias",
            "productSpec",
            "productPackageSpec",
            "productUnit",
            "planQty",
            "actualQty",
        ]
    )
    cols = filter_export_columns(PLAN_EXPORT_COLUMNS, requested)
    assert [c.field for c in cols] == [
        "planNo",
        "belongsCompany",
        "followStatus",
        "planType",
        "loadCustomer",
        "deliveryCustomer",
        "productName",
        "brandName",
        "aliasName",
        "model",
        "packageSpec",
        "unit",
        "planQty",
        "actualQty",
    ]


def test_plan_list_export_filename_stem_by_stock_role():
    assert _plan_list_export_filename_stem("inbound", "20261015") == "入库单_20261015"
    assert _plan_list_export_filename_stem("outbound", "20261015") == "出库单_20261015"
    assert _plan_list_export_filename_stem(None, "20261015") == "计划列表_20261015"


def test_build_plan_list_export_rows_expands_products_with_crm_names():
    plan = SimpleNamespace(
        id="p1",
        plan_no="PL001",
        follow_status="arranged",
        plan_date="2026-10-11",
        plan_type="direct",
        supplier_name="供方",
        customer_name="需方",
        license_plate="皖A12345",
        trailer_no=None,
        driver_name="司机",
        transport_status="arranged",
        archive_status="unarchived",
        erp_entered=0,
        remark="",
        products=[],
    )
    # 计划级产品只用于品牌 lookup；行展开走段产品。
    product = SimpleNamespace(
        id="pp1",
        plan_id="p1",
        product_id="prod1",
        product_name="甲醇别名",
        brand_mode="required",
        brand_name="君正",
        model="工业级",
        package_spec="吨袋",
        unit="ton",
        quantity=10,
        actual_qty=9.5,
        order_num=1,
    )
    load_product = SimpleNamespace(
        id="lp1",
        plan_product_id="pp1",
        product_id="prod1",
        product_name="甲醇别名",
        model="工业级",
        package_spec="吨袋",
        unit="ton",
        plan_qty=10,
        inbound_qty=9.5,
        outbound_qty=None,
    )
    delivery_product = SimpleNamespace(
        id="dp1",
        plan_product_id="pp1",
        product_id="prod1",
        product_name="甲醇别名",
        model="工业级",
        package_spec="吨袋",
        unit="ton",
        plan_qty=10,
        inbound_qty=None,
        outbound_qty=9.5,
    )
    db = MagicMock()
    query = MagicMock()
    db.query.return_value = query
    query.filter.return_value = query
    query.order_by.return_value = query
    query.all.return_value = [product]

    load_stop = SimpleNamespace(
        customer_name="装货客户A",
        address="装货路1号",
        load_time="2026-10-11",
        generate_dibang=1,
        dibang_direction="outbound",
        contracts=[SimpleNamespace(contract_no="HT001")],
        products=[load_product],
    )
    delivery_stop = SimpleNamespace(
        seq=1,
        customer_name="送货客户B",
        address="送货路2号",
        delivery_time="2026-10-12",
        generate_dibang=1,
        dibang_direction="inbound",
        contracts=[SimpleNamespace(contract_no="HT002")],
        products=[delivery_product],
    )

    fake_dump = {
        "planNo": "PL001",
        "belongsCompany": "shanxia",
        "followStatus": "arranged",
        "planDate": "2026-10-11",
        "planType": "inbound",
        "supplierName": "供方",
        "customerName": "需方",
        "licensePlate": "皖A12345",
        "trailerNo": None,
        "personnelSummary": "司机张三",
        "followUserName": "跟进人李四",
        "driverName": "司机",
        "transportStatus": "arranged",
        "archiveStatus": "unarchived",
        "erpEntered": 0,
        "remark": "",
        "products": [{"id": "pp1"}],
    }

    with patch(
        "app.api.pl.plan.PlanWithProducts.model_validate",
        return_value=MagicMock(model_dump=MagicMock(return_value=dict(fake_dump))),
    ), patch(
        "app.api.pl.plan.load_crm_product_alias_maps",
        return_value=({"prod1": "甲醇"}, {"prod1": ["甲醇别名"]}),
    ):
        rows = _build_plan_list_export_rows(
            db,
            tenant_no="t1",
            plans=[plan],
            stops_by_plan={"p1": (load_stop, [delivery_stop])},
        )

    assert len(rows) == 2
    assert [row["stopSegment"] for row in rows] == ["装货", "送货 #1"]
    assert rows[0]["_stopKind"] == "load"
    assert rows[1]["_stopKind"] == "delivery"
    assert rows[1]["_deliverySeq"] == 1
    load_row, delivery_row = rows
    for row in rows:
        assert row["planNo"] == "PL001"
        assert row["productName"] == "甲醇"
        assert row["brandName"] == "君正"
        assert row["aliasName"] == "甲醇别名"
        assert row["followStatus"] == "已安排"
        assert row["planType"] == "入库单"
        assert row["belongsCompany"] == "安庆山峡"
        assert row["archiveStatus"] == "未归档"
        assert row["model"] == "工业级"
        assert row["packageSpec"] == "吨袋"
        assert row["unit"] == "ton"
        assert row["planQty"] == 10
        assert row["actualQty"] == 9.5
        assert "products" not in row
    assert load_row["deliveryTimes"] == ""
    assert load_row["loadCustomer"] == "装货客户A"
    assert load_row["deliveryCustomer"] == ""
    assert load_row["loadAddressTime"] == "装货路1号 / 2026-10-11"
    assert load_row["deliveryAddressTime"] == ""
    assert load_row["productContract"] == "HT001"
    assert load_row["loadDibang"] == "出库"
    assert load_row["deliveryDibang"] == ""
    assert delivery_row["deliveryTimes"] == "2026-10-12"
    assert delivery_row["loadCustomer"] == ""
    assert delivery_row["deliveryCustomer"] == "送货客户B"
    assert delivery_row["loadAddressTime"] == ""
    assert delivery_row["deliveryAddressTime"] == "送货路2号 / 2026-10-12"
    assert delivery_row["productContract"] == "HT002"
    assert delivery_row["loadDibang"] == ""
    assert delivery_row["deliveryDibang"] == "入库"


def test_build_plan_list_export_rows_expands_stop_segments():
    """一装多送：先装货产品行，再各送货段产品行；侧栏字段按行填写。"""
    plan = SimpleNamespace(
        id="p-multi",
        plan_no="PO20261009016",
        follow_status="arranged",
        plan_date="2026-10-11",
        plan_type="inbound",
        supplier_name="供方",
        customer_name="需方",
        license_plate="皖A12345",
        trailer_no=None,
        driver_name="司机",
        transport_status="arranged",
        archive_status="unarchived",
        erp_entered=0,
        remark="",
        products=[],
    )
    db = MagicMock()
    query = MagicMock()
    db.query.return_value = query
    query.filter.return_value = query
    query.order_by.return_value = query
    # 有段结构时不应依赖计划级产品展开；返回空列表即可
    query.all.return_value = []

    load_product = SimpleNamespace(
        id="lp1",
        plan_product_id="pp1",
        product_id="prod1",
        product_name="尿素别名",
        model="工业级",
        package_spec="吨袋",
        unit="ton",
        plan_qty=30,
        inbound_qty=29,
        outbound_qty=None,
        brand_mode=None,
        brand_name=None,
    )
    del1_product = SimpleNamespace(
        id="dp1",
        plan_product_id="pp1",
        product_id="prod1",
        product_name="尿素别名",
        model="工业级",
        package_spec="吨袋",
        unit="ton",
        plan_qty=20,
        inbound_qty=None,
        outbound_qty=19,
        brand_mode=None,
        brand_name=None,
    )
    del2_product = SimpleNamespace(
        id="dp2",
        plan_product_id="pp1",
        product_id="prod1",
        product_name="尿素别名",
        model="工业级",
        package_spec="吨袋",
        unit="ton",
        plan_qty=10,
        inbound_qty=None,
        outbound_qty=9,
        brand_mode=None,
        brand_name=None,
    )
    load_stop = SimpleNamespace(
        customer_name="装货客户A",
        address="装货路1号",
        load_time="2026-10-11",
        generate_dibang=1,
        dibang_direction="outbound",
        contracts=[SimpleNamespace(contract_no="HT-L")],
        products=[load_product],
    )
    delivery1 = SimpleNamespace(
        seq=1,
        customer_name="送货客户B1",
        address="送货路B1",
        delivery_time="2026-10-12",
        generate_dibang=1,
        dibang_direction="inbound",
        contracts=[SimpleNamespace(contract_no="HT-B1")],
        products=[del1_product],
    )
    delivery2 = SimpleNamespace(
        seq=2,
        customer_name="送货客户B2",
        address="送货路B2",
        delivery_time="2026-10-13",
        generate_dibang=0,
        dibang_direction=None,
        contracts=[SimpleNamespace(contract_no="HT-B2")],
        products=[del2_product],
    )

    fake_dump = {
        "planNo": "PO20261009016",
        "belongsCompany": "shanxia",
        "followStatus": "arranged",
        "planDate": "2026-10-11",
        "planType": "inbound",
        "supplierName": "供方",
        "customerName": "需方",
        "licensePlate": "皖A12345",
        "trailerNo": None,
        "personnelSummary": "司机张三",
        "followUserName": "跟进人李四",
        "driverName": "司机",
        "transportStatus": "arranged",
        "archiveStatus": "unarchived",
        "erpEntered": 0,
        "remark": "",
        "products": [],
    }

    with patch(
        "app.api.pl.plan.PlanWithProducts.model_validate",
        return_value=MagicMock(model_dump=MagicMock(return_value=dict(fake_dump))),
    ), patch(
        "app.api.pl.plan.load_crm_product_alias_maps",
        return_value=({"prod1": "尿素"}, {"prod1": ["尿素别名"]}),
    ):
        rows = _build_plan_list_export_rows(
            db,
            tenant_no="t1",
            plans=[plan],
            stops_by_plan={"p-multi": (load_stop, [delivery1, delivery2])},
        )

    assert len(rows) == 3
    assert [r["stopSegment"] for r in rows] == ["装货", "送货 #1", "送货 #2"]
    assert [r["planQty"] for r in rows] == [30, 20, 10]
    assert [r["actualQty"] for r in rows] == [29, 19, 9]
    assert rows[0]["loadCustomer"] == "装货客户A"
    assert rows[0]["deliveryCustomer"] == ""
    assert rows[0]["productContract"] == "HT-L"
    assert rows[1]["loadCustomer"] == ""
    assert rows[1]["deliveryCustomer"] == "送货客户B1"
    assert rows[1]["deliveryTimes"] == "2026-10-12"
    assert rows[1]["productContract"] == "HT-B1"
    assert rows[2]["deliveryCustomer"] == "送货客户B2"
    assert rows[2]["deliveryDibang"] == ""
    assert rows[0]["productName"] == "尿素"


def test_build_plan_list_export_rows_empty_delivery_products_placeholder():
    """送货段无产品仍占一行；产品/量为空。"""
    plan = SimpleNamespace(
        id="p-empty",
        plan_no="PL-EMPTY",
        follow_status="arranged",
        plan_date="2026-10-11",
        plan_type="inbound",
        supplier_name="供方",
        customer_name="需方",
        license_plate="皖A12345",
        trailer_no=None,
        driver_name="司机",
        transport_status="arranged",
        archive_status="unarchived",
        erp_entered=0,
        remark="",
        products=[],
    )
    db = MagicMock()
    query = MagicMock()
    db.query.return_value = query
    query.filter.return_value = query
    query.order_by.return_value = query
    query.all.return_value = []

    load_product = SimpleNamespace(
        id="lp1",
        plan_product_id="pp1",
        product_id="prod1",
        product_name="甲醇别名",
        model="工业级",
        package_spec="吨袋",
        unit="ton",
        plan_qty=5,
        inbound_qty=5,
        outbound_qty=None,
        brand_mode=None,
        brand_name=None,
    )
    load_stop = SimpleNamespace(
        customer_name="装货客户A",
        address="装货路1号",
        load_time="2026-10-11",
        generate_dibang=1,
        dibang_direction="outbound",
        contracts=[SimpleNamespace(contract_no="HT-L")],
        products=[load_product],
    )
    delivery = SimpleNamespace(
        seq=1,
        customer_name="送B",
        address="送货路B",
        delivery_time="2026-10-12",
        generate_dibang=1,
        dibang_direction="inbound",
        contracts=[SimpleNamespace(contract_no="HT-B")],
        products=[],
    )

    fake_dump = {
        "planNo": "PL-EMPTY",
        "belongsCompany": "shanxia",
        "followStatus": "arranged",
        "planDate": "2026-10-11",
        "planType": "inbound",
        "supplierName": "供方",
        "customerName": "需方",
        "licensePlate": "皖A12345",
        "trailerNo": None,
        "personnelSummary": "司机张三",
        "followUserName": "跟进人李四",
        "driverName": "司机",
        "transportStatus": "arranged",
        "archiveStatus": "unarchived",
        "erpEntered": 0,
        "remark": "",
        "products": [],
    }

    with patch(
        "app.api.pl.plan.PlanWithProducts.model_validate",
        return_value=MagicMock(model_dump=MagicMock(return_value=dict(fake_dump))),
    ), patch(
        "app.api.pl.plan.load_crm_product_alias_maps",
        return_value=({"prod1": "甲醇"}, {"prod1": ["甲醇别名"]}),
    ):
        rows = _build_plan_list_export_rows(
            db,
            tenant_no="t1",
            plans=[plan],
            stops_by_plan={"p-empty": (load_stop, [delivery])},
        )

    assert len(rows) == 2
    assert rows[1]["stopSegment"] == "送货 #1"
    assert rows[1]["productName"] == ""
    assert rows[1]["planQty"] is None
    assert rows[1]["deliveryCustomer"] == "送B"


def test_build_plan_list_export_rows_legacy_without_stops():
    """无段结构：回退计划级产品；stopSegment 为空。"""
    plan = SimpleNamespace(
        id="p1",
        plan_no="PL001",
        follow_status="arranged",
        plan_date="2026-10-11",
        plan_type="direct",
        supplier_name="供方",
        customer_name="需方",
        license_plate="皖A12345",
        trailer_no=None,
        driver_name="司机",
        transport_status="arranged",
        archive_status="unarchived",
        erp_entered=0,
        remark="",
        products=[],
    )
    product = SimpleNamespace(
        id="pp1",
        plan_id="p1",
        product_id="prod1",
        product_name="甲醇别名",
        brand_mode="required",
        brand_name="君正",
        model="工业级",
        package_spec="吨袋",
        unit="ton",
        quantity=10,
        actual_qty=9.5,
        order_num=1,
    )
    db = MagicMock()
    query = MagicMock()
    db.query.return_value = query
    query.filter.return_value = query
    query.order_by.return_value = query
    query.all.return_value = [product]

    fake_dump = {
        "planNo": "PL001",
        "belongsCompany": "shanxia",
        "followStatus": "arranged",
        "planDate": "2026-10-11",
        "planType": "inbound",
        "supplierName": "供方",
        "customerName": "需方",
        "licensePlate": "皖A12345",
        "trailerNo": None,
        "personnelSummary": "司机张三",
        "followUserName": "跟进人李四",
        "driverName": "司机",
        "transportStatus": "arranged",
        "archiveStatus": "unarchived",
        "erpEntered": 0,
        "remark": "",
        "products": [{"id": "pp1"}],
    }

    with patch(
        "app.api.pl.plan.PlanWithProducts.model_validate",
        return_value=MagicMock(model_dump=MagicMock(return_value=dict(fake_dump))),
    ), patch(
        "app.api.pl.plan.load_crm_product_alias_maps",
        return_value=({"prod1": "甲醇"}, {"prod1": ["甲醇别名"]}),
    ):
        rows = _build_plan_list_export_rows(
            db,
            tenant_no="t1",
            plans=[plan],
            stops_by_plan={"p1": (None, [])},
        )

    assert len(rows) == 1
    assert rows[0]["stopSegment"] == ""
    assert rows[0]["planQty"] == 10
    assert rows[0]["productName"] == "甲醇"


def test_build_plan_list_export_rows_internal_customer_contract_placeholder():
    """山峡/云栋缺合同：装货行合同为出库，送货行为入库（与列表 UI 一致）。"""
    from app.services.pl_internal_customer import INTERNAL_CUSTOMER_NAMES

    internal = next(iter(INTERNAL_CUSTOMER_NAMES))
    plan = SimpleNamespace(
        id="p-io",
        plan_no="PL-IO",
        follow_status="arranged",
        plan_date="2026-10-11",
        plan_type="direct",
        supplier_name=internal,
        customer_name=internal,
        license_plate="皖A1",
        trailer_no=None,
        driver_name="",
        transport_status="arranged",
        archive_status="unarchived",
        erp_entered=0,
        remark="",
        products=[],
    )
    db = MagicMock()
    query = MagicMock()
    db.query.return_value = query
    query.filter.return_value = query
    query.order_by.return_value = query
    query.all.return_value = []

    stop_product = SimpleNamespace(
        id="sp1",
        plan_product_id=None,
        product_id="prod1",
        product_name="尿素",
        model="",
        package_spec="",
        unit="ton",
        plan_qty=5,
        inbound_qty=None,
        outbound_qty=None,
    )
    load_stop = SimpleNamespace(
        customer_name=internal,
        address="装货地址",
        load_time="2026-10-11",
        generate_dibang=0,
        dibang_direction=None,
        contracts=[],
        products=[stop_product],
    )
    delivery_stop = SimpleNamespace(
        seq=1,
        customer_name=internal,
        address="送货地址",
        delivery_time="2026-10-12",
        generate_dibang=0,
        dibang_direction=None,
        contracts=[],
        products=[
            SimpleNamespace(
                id="sp2",
                plan_product_id=None,
                product_id="prod1",
                product_name="尿素",
                model="",
                package_spec="",
                unit="ton",
                plan_qty=5,
                inbound_qty=None,
                outbound_qty=5,
            )
        ],
    )
    fake_dump = {
        "planNo": "PL-IO",
        "belongsCompany": "shanxia",
        "followStatus": "arranged",
        "planDate": "2026-10-11",
        "planType": "direct",
        "supplierName": internal,
        "customerName": internal,
        "licensePlate": "皖A1",
        "trailerNo": None,
        "personnelSummary": "",
        "followUserName": "",
        "driverName": "",
        "transportStatus": "arranged",
        "archiveStatus": "unarchived",
        "erpEntered": 0,
        "remark": "",
        "products": [],
    }

    with patch(
        "app.api.pl.plan.PlanWithProducts.model_validate",
        return_value=MagicMock(model_dump=MagicMock(return_value=dict(fake_dump))),
    ), patch(
        "app.api.pl.plan.load_crm_product_alias_maps",
        return_value=({"prod1": "尿素"}, {"prod1": []}),
    ):
        rows = _build_plan_list_export_rows(
            db,
            tenant_no="t1",
            plans=[plan],
            stops_by_plan={"p-io": (load_stop, [delivery_stop])},
        )

    assert len(rows) == 2
    assert rows[0]["stopSegment"] == "装货"
    assert rows[0]["productContract"] == "出库"
    assert rows[1]["stopSegment"] == "送货 #1"
    assert rows[1]["productContract"] == "入库"


def test_plan_list_merge_sheet_follows_list_rowspan():
    from openpyxl import Workbook

    from app.services.pl_plan_archive_export import (
        _PLAN_LIST_DELIVERY_MERGE_FIELDS,
        _PLAN_LIST_LOAD_COPY_FIELDS,
        _PLAN_LIST_PLAN_MERGE_FIELDS,
        append_archive_list_merge_sheet,
    )
    from app.utils.excel_export import ExportColumn

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

    columns = [
        ExportColumn(title="计划编号", field="planNo"),
        ExportColumn(title="送货时间", field="deliveryTimes"),
        ExportColumn(title="装货客户", field="loadCustomer"),
        ExportColumn(title="送货客户", field="deliveryCustomer"),
        ExportColumn(title="合同", field="productContract"),
        ExportColumn(title="产品名称", field="productName"),
    ]
    rows = [
        {
            "id": "p1",
            "planNo": "PO1",
            "_stopKind": "load",
            "_deliverySeq": None,
            "deliveryTimes": "",
            "loadCustomer": "山峡",
            "deliveryCustomer": "",
            "productContract": "出库",
            "productName": "氯化钠",
        },
        {
            "id": "p1",
            "planNo": "PO1",
            "_stopKind": "delivery",
            "_deliverySeq": 1,
            "deliveryTimes": "2026-10-07",
            "loadCustomer": "",
            "deliveryCustomer": "奇创",
            "productContract": "HT-1",
            "productName": "次氯酸钠",
        },
        {
            "id": "p1",
            "planNo": "PO1",
            "_stopKind": "delivery",
            "_deliverySeq": 2,
            "deliveryTimes": "2026-10-08",
            "loadCustomer": "",
            "deliveryCustomer": "奇创",
            "productContract": "HT-2",
            "productName": "氯化钠",
        },
    ]
    workbook = Workbook()
    workbook.active.title = "计划列表"
    append_archive_list_merge_sheet(
        workbook,
        columns,
        rows,
        Resolver(),
        sheet_title="计划列表（合并）",
        plan_fields=_PLAN_LIST_PLAN_MERGE_FIELDS,
        delivery_fields=_PLAN_LIST_DELIVERY_MERGE_FIELDS,
        contract_field="productContract",
        load_fields=_PLAN_LIST_LOAD_COPY_FIELDS,
        number_plans=False,
        join_plan_fields=("deliveryTimes",),
    )
    sheet = workbook["计划列表（合并）"]
    ranges = {str(item) for item in sheet.merged_cells.ranges}
    assert "A2:A4" in ranges
    assert "B2:B4" in ranges
    assert "C2:C4" in ranges
    assert sheet["B2"].value == "2026-10-07\n2026-10-08"
    assert sheet["C2"].value == "山峡"
    assert sheet["D2"].value == "—"
    assert sheet["D3"].value == "奇创"
    assert "F2:F4" not in ranges
    assert sheet["F2"].value == "氯化钠"
    assert sheet["F4"].value == "氯化钠"