"""BOM 零件牌号查询数据源（客户端过滤 + 排序）
按生产令号查询 ``bom_materials``（模糊），结果在内存中进一步过滤/排序/分页。
复用于「基础数据」tab 下的 BOM 零件牌号查询子 tab。
"""
from __future__ import annotations
from typing import Any
from certflow.views.bases.query_data_source import (
BaseQueryDataSource,
client_filter_records,
client_sort_records,
export_records_to_csv,
)
# 显示列：(表头, 字段) —— 与 BomMaterialQueryView 对齐
# 源文件名(_source_name) 为派生列，由 source_file 取 basename，在视图层渲染/导出
_BOM_COLUMNS = [
("生产令号", "production_order_no"),
("序号", "seq"),
("代号", "part_no"),
("零件名称", "part_name"),
("材料牌号", "material_grade"),
("数量", "quantity"),
("材料型态", "material_form"),
("来源", "material_source"),
("源文件名", "_source_name"),
]
def _source_basename(source_file: str | None) -> str:
"""从 source_file 路径取文件名（basename），跨平台分隔符安全。"""
if not source_file:
return ""
return source_file.rsplit("\\", 1)[-1].rsplit("/", 1)[-1]
[文档]
class BomQueryDataSource(BaseQueryDataSource):
"""BOM 零件牌号查询数据源（客户端过滤/排序）。"""
supports_server_sort = False
supports_server_paging = False
quick_search_or = False
editable = True # 牌号可手填订正（见 BomMaterialQueryView 工具栏/右键）
correction = False
def __init__(self, controller: Any) -> None:
self._controller = controller
[文档]
def get_columns(self) -> list[tuple[str, str]]:
return list(_BOM_COLUMNS)
[文档]
def get_filter_columns(self) -> list[dict]:
# 生产令号作为首列筛选项，驱动「按令号检索」（query 内走模糊/精确搜索）
cols = [{"label": "生产令号", "field": "production_order_no"}]
cols += [
{"label": h, "field": f}
for h, f in _BOM_COLUMNS
if f not in ("production_order_no", "_source_name")
]
return cols
[文档]
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]:
pno_cond = conditions.get("production_order_no") if conditions else None
pno = pno_cond.get("value") if isinstance(pno_cond, dict) else None
records = self._controller.search(pno, fuzzy=True) if pno else self._controller.list_all()
others = {k: v for k, v in (conditions or {}).items() if k != "production_order_no"}
records = client_filter_records(records, others)
records = client_sort_records(records, order_by, order_desc)
return self.client_paginate(records, page, page_size, conditions)
[文档]
def export_csv(self, path: str) -> int:
records = self._controller.list_all()
# 构造 dict 行，便于派生列 源文件名 正确导出（ORM 无该属性）
rows = []
for rec in records:
d = {f: getattr(rec, f, None) for _, f in self.get_columns() if f != "_source_name"}
d["_source_name"] = _source_basename(getattr(rec, "source_file", None))
rows.append(d)
return export_records_to_csv(rows, self.get_columns(), path)