"""销售计划数据源（服务端排序 + 分页 + OR 快查）
薄封装 ``QueryController.query``，输出与 ``QueryDataSource`` 一致契约。
列定义将在阶段 B 接入 query_view 时与既有显示列对齐；此处先给代表列。
"""
from __future__ import annotations
from typing import Any
from certflow.config.settings import get_views_config
from certflow.views.bases.query_data_source import (
BaseQueryDataSource,
export_records_to_csv,
)
# 代表显示列（阶段 B 接入 query_view 后替换为完整列集）
SALE_PLAN_COLUMNS = [
("ID", "id"),
("计划日期", "plan_date"),
("订货单位", "customer"),
("项目名称", "project_name"),
("要货计划单号", "plan_no"),
("产品型号", "product_model"),
("产品规格", "product_spec"),
("材质牌号", "material_grade"),
("状态", "status"),
]
[文档]
class SalePlanDataSource(BaseQueryDataSource):
"""销售计划数据源：委托 QueryController 走服务端查询。
``_fixed_conditions`` / ``_default_order_by`` 供子类（如 ``CorrectionDataSource``）
参数化覆盖，注入固定查询条件与默认排序列，免去重写 ``query``。
列定义：默认使用 ``SALE_PLAN_COLUMNS`` 代表列；``from_views_config`` 可读取
ui.yaml 的默认销售计划视图列（与 ``query_view`` 显示列对齐，落实阶段 B 接入）。
"""
supports_server_sort = True
supports_server_paging = True
quick_search_or = True
editable = False
correction = False
_fixed_conditions: dict = {}
_default_order_by: str = "sort_group"
def __init__(self, controller: Any, columns: list[tuple[str, str]] | None = None) -> None:
self._controller = controller
self._columns = list(columns) if columns else None
[文档]
def get_columns(self) -> list[tuple[str, str]]:
if self._columns is not None:
return list(self._columns)
return list(SALE_PLAN_COLUMNS)
[文档]
@classmethod
def from_views_config(cls, controller: Any, view_key: str | None = None) -> SalePlanDataSource:
"""从 ui.yaml 的 ``query_page.views`` 读取视图列，构造与 ``query_view`` 显示列一致的数据源。
Args:
controller: QueryController 实例。
view_key: 视图标识；缺省取首个视图（即 ``QueryView`` 默认视图）。
"""
views = get_views_config()
if not views:
return cls(controller)
key = view_key or next(iter(views))
template = views.get(key, {})
cols = template.get("columns", []) or []
columns = [
(c.get("label", c.get("field", "")), c.get("field", "")) for c in cols if c.get("field")
]
return cls(controller, columns=columns or None)
[文档]
def query(
self,
conditions: dict,
page: int,
page_size: int,
order_by: str | None = None,
order_desc: bool = False,
or_filters: Any = None,
) -> dict[str, Any]:
merged = {**self._fixed_conditions, **(conditions or {})}
return self._controller.query(
conditions=merged,
page=page,
page_size=page_size,
order_by=order_by or self._default_order_by,
order_desc=order_desc,
or_filters=or_filters,
)
[文档]
def export_csv(self, path: str) -> int:
# 拉全量（大页）再导出，避免依赖 controller 私有导出实现
result = self._controller.query({}, 1, 1_000_000, order_by="id", order_desc=False)
return export_records_to_csv(result["records"], self.get_columns(), path)