"""查询数据源协议与通用引擎（阶段 A 抽象层）
定义 ``QueryDataSource`` 协议，使 ``BaseQueryView`` 能以统一方式驱动不同数据源
（销售计划 / 校正队列 / 基础数据 / BOM 查询），无需在视图里写 ``if`` 分支。
能力标志（capability flags）决定 UI 行为：
- ``supports_server_sort``：列头点击走「重新查询」（销售计划 / 校正队列）。
- ``supports_server_paging``：分页由服务端承担（销售计划 / 校正队列）。
- ``quick_search_or``：是否支持跨字段 OR 快查（仅销售计划）。
- ``editable`` / ``correction``：是否暴露行内编辑 / 校正动作（客户端源按需在子类开启）。
客户端数据源（基础数据 / BOM）复用本模块的 ``client_filter_records`` /
``client_sort_records`` 在内存中过滤排序，避免给每个表写一遍逻辑。
"""
from __future__ import annotations
import csv
from abc import ABC, abstractmethod
from math import ceil
from typing import Any, Protocol, runtime_checkable
def _record_get(record: Any, field: str) -> Any:
"""统一从 dict 或 ORM 对象取字段值。"""
if isinstance(record, dict):
return record.get(field)
return getattr(record, field, None)
def _to_number(value: Any) -> float | None:
"""尽力把值转成 float；失败返回 None。"""
if isinstance(value, bool):
return None
if isinstance(value, (int, float)):
return float(value)
try:
return float(str(value).strip().replace(",", ""))
except (ValueError, TypeError, AttributeError):
return None
def _match_op(rv: Any, op: str, value: Any) -> bool: # noqa: C901
"""单条记录字段值对单个 operator 的匹配（故意集中分派，复杂度豁免）。"""
if op == "is_null":
return rv is None or rv == ""
if op == "is_not_null":
return not (rv is None or rv == "")
if rv is None:
return False
if op == "contains":
return value is None or str(value) in str(rv)
if op == "eq":
return str(rv) == str(value)
if op == "ne":
return str(rv) != str(value)
if op in ("gte", "ge", "lte", "le", "gt", "lt"):
a, b = _to_number(rv), _to_number(value)
if a is None or b is None:
return False
if op in ("gte", "ge"):
return a >= b
if op in ("lte", "le"):
return a <= b
if op == "gt":
return a > b
return a < b
if op == "between" and isinstance(value, (list, tuple)) and len(value) == 2:
a = _to_number(rv)
lo, hi = _to_number(value[0]), _to_number(value[1])
return a is not None and lo is not None and hi is not None and lo <= a <= hi
# 未知算子按 contains 兜底
return value is None or str(value) in str(rv)
[文档]
def client_filter_records(records: list[Any], conditions: dict | None) -> list[Any]:
"""客户端条件过滤。
``conditions`` 形如 ``{field: {"value": v, "operator": op}}``，支持的 operator：
``contains`` / ``eq`` / ``ne`` / ``gte`` / ``lte`` / ``gt`` / ``lt`` /
``between``(value=[lo,hi]) / ``is_null`` / ``is_not_null``。
"""
if not conditions:
return list(records)
out: list[Any] = []
for rec in records:
matched = True
for field, cond in conditions.items():
cond = cond or {}
op = cond.get("operator", "contains")
value = cond.get("value")
rv = _record_get(rec, field)
if not _match_op(rv, op, value):
matched = False
break
if matched:
out.append(rec)
return out
def _sort_key(rec: Any, field: str) -> tuple:
v = _record_get(rec, field)
if v is None:
return (1, 0)
if isinstance(v, bool):
return (0, int(v))
if isinstance(v, (int, float)):
return (0, v)
return (0, str(v))
[文档]
def client_sort_records(
records: list[Any], order_by: str | None, order_desc: bool = False
) -> list[Any]:
"""客户端排序：None 恒排末尾；同列类型同质，数字按数值、其余按字符串。"""
if not order_by:
return list(records)
return sorted(records, key=lambda r: _sort_key(r, order_by), reverse=order_desc)
[文档]
def export_records_to_csv(records: list[Any], columns: list[tuple[str, str]], path: str) -> int:
"""把记录按列定义导出为 CSV（UTF-8-BOM，Excel 友好）。返回行数。"""
with open(path, "w", newline="", encoding="utf-8-sig") as f:
writer = csv.writer(f)
writer.writerow([h for h, _ in columns])
for rec in records:
writer.writerow(
[
"" if _record_get(rec, field) is None else _record_get(rec, field)
for _, field in columns
]
)
return len(records)
[文档]
@runtime_checkable
class QueryDataSource(Protocol):
"""查询数据源协议（结构化契约，供 ``BaseQueryView`` 驱动）。"""
supports_server_sort: bool
supports_server_paging: bool
quick_search_or: bool
editable: bool
correction: bool
[文档]
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]:
"""返回 ``{records, page, page_size, total, total_pages, conditions}``。"""
...
[文档]
def get_columns(self) -> list[tuple[str, str]]:
"""返回 ``[(表头, 字段), ...]``。"""
...
[文档]
def export_csv(self, path: str) -> int:
"""导出当前结果到 CSV，返回行数。"""
...
[文档]
class BaseQueryDataSource(ABC):
"""数据源基类：提供能力标志默认值与默认筛选列推导。"""
supports_server_sort: bool = False
supports_server_paging: bool = False
quick_search_or: bool = False
editable: bool = False
correction: bool = False
[文档]
@abstractmethod
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]: ...
[文档]
@abstractmethod
def get_columns(self) -> list[tuple[str, str]]: ...
[文档]
@abstractmethod
def export_csv(self, path: str) -> int: ...
[文档]
def get_filter_columns(self) -> list[dict]:
"""默认把所有显示列作为可筛选字段；子类可覆盖以精简。"""
return [{"label": h, "field": f} for h, f in self.get_columns()]
[文档]
def client_paginate(
self,
records: list[Any],
page: int,
page_size: int,
conditions: dict | None,
) -> dict[str, Any]:
"""客户端分页：对已过滤/排序的内存 ``records`` 切片，返回统一契约。
供 ``BaseDataDataSource`` / ``BomQueryDataSource`` 等客户端数据源复用，
避免每个子类重复书写 total/total_pages/page_records 切片逻辑。
"""
total = len(records)
total_pages = max(1, ceil(total / page_size)) if page_size > 0 else 1
start = (page - 1) * page_size
page_records = records[start : start + page_size]
return {
"records": page_records,
"page": page,
"page_size": page_size,
"total": total,
"total_pages": total_pages,
"conditions": conditions,
}