"""查询服务模块
提供销售计划的多条件组合查询功能，支持参数化查询和分页。
支持关联 Certificate 表进行联合查询。
"""
from __future__ import annotations
from typing import Any
from sqlalchemy import and_, func, or_
from sqlalchemy.orm import Session, selectinload
from certflow.models import SalePlan
from certflow.models.certificate import Certificate # 新增导入
from certflow.utils.logger import logger
# 快查候选项字段组合分隔符（与 query_view._on_quick_search 回填切分共用，必须一致）。
# 用「空格+竖线+空格」可读且低冲突；保留空段以保证段数恒等于字段数，
# 避免中间字段为空时按位置回填整体错位。
QUICK_SEARCH_SEP = " | "
[文档]
class QueryService:
"""销售计划查询服务
提供灵活的多条件组合查询功能:
- 支持模糊匹配、精确匹配、范围查询
- 支持多字段组合查询
- 支持分页和排序
- 支持关联 Certificate 表查询
Examples:
>>> service = QueryService(session)
>>> results = service.query(
... conditions={
... "product_model": {"value": "阀门", "operator": "contains"},
... "customer": {"value": "东方电气", "operator": "contains"},
... "quantity": {"value": 10, "operator": "gte"}
... }
... )
"""
OPERATORS = {
"eq": lambda field, value: field == value,
"ne": lambda field, value: field != value,
"contains": lambda field, value: field.contains(value),
"not_contains": lambda field, value: ~field.contains(value),
"startswith": lambda field, value: field.startswith(value),
"endswith": lambda field, value: field.endswith(value),
"gt": lambda field, value: field > value,
"gte": lambda field, value: field >= value,
"lt": lambda field, value: field < value,
"lte": lambda field, value: field <= value,
"in": lambda field, value: field.in_(value),
"is_null": lambda field, _: field.is_(None),
"is_not_null": lambda field, _: field.isnot(None),
}
# SalePlan 表的字段列表（用于判断字段归属）
_salepian_fields: set[str] | None = None
# Certificate 表的字段列表
_certificate_fields: set[str] | None = None
@classmethod
def _get_saleplan_fields(cls) -> set[str]:
if cls._salepian_fields is None:
cls._salepian_fields = {col.key for col in SalePlan.__table__.columns}
return cls._salepian_fields
@classmethod
def _get_certificate_fields(cls) -> set[str]:
if cls._certificate_fields is None:
cls._certificate_fields = {col.key for col in Certificate.__table__.columns}
return cls._certificate_fields
@property
def saleplan_fields(self) -> set[str]:
"""SalePlan 表全部列名集合。
惰性计算并缓存（``_salepian_fields``），用于判定查询字段归属 SalePlan 表。
Returns:
set[str]: SalePlan 表列名集合。
"""
return self._get_saleplan_fields()
@property
def certificate_fields(self) -> set[str]:
"""Certificate 表全部列名集合。
惰性计算并缓存（``_certificate_fields``），用于判定查询字段归属 Certificate 表。
Returns:
set[str]: Certificate 表列名集合。
"""
return self._get_certificate_fields()
# 混合属性字段映射（@hybrid_property，不是列但可用于查询过滤）
HYBRID_FIELDS: dict[str, Any] = {}
# 字段别名映射（YAML 用名 → 数据库实际列名）
FIELD_ALIASES = {
"cert_status": "print_status",
"print_status": "print_status",
}
def __init__(self, session: Session):
"""初始化查询服务
Args:
session: SQLAlchemy数据库会话对象
"""
self.session = session
if not self.HYBRID_FIELDS:
self.HYBRID_FIELDS["is_ready_to_ship"] = SalePlan.is_ready_to_ship
[文档]
def build_status_filter(self, status: str) -> Any:
"""构建状态筛选条件（同时匹配 production_status 和 execution_status）
当用户筛选标准化状态时，同时模糊匹配 execution_status 原始文本，
避免因原始文本包含状态关键词而遗漏记录。
关键词从 StatusInference 配置中获取（config.yaml production_status.query_keywords）。
Args:
status: 状态值，如 "已发货"、"已完成"、"生产中"
Returns:
SQLAlchemy 过滤表达式
"""
from certflow.utils.status_inference import StatusInference
# 确保从配置加载
StatusInference.init_from_config()
keyword = StatusInference.get_query_keyword(status)
return or_(
SalePlan.production_status == status,
SalePlan.execution_status.contains(keyword),
)
def _resolve_field(self, field_name: str) -> Any:
"""解析字段名，返回对应表的 SQLAlchemy 列属性
先通过 FIELD_ALIASES 映射，再判断属于 Certificate 还是 SalePlan 表。
最后检查混合属性字段。
Args:
field_name: 字段名（支持别名）
Returns:
SQLAlchemy 列属性对象或混合属性，未知字段返回 None
"""
actual_name = self.FIELD_ALIASES.get(field_name, field_name)
# 检查混合属性
if actual_name in self.HYBRID_FIELDS:
return self.HYBRID_FIELDS[actual_name]
if actual_name in self.certificate_fields and actual_name not in self.saleplan_fields:
return getattr(Certificate, actual_name)
if actual_name in self.saleplan_fields:
return getattr(SalePlan, actual_name)
logger.warning(f"未知字段: {field_name}")
return None
def _is_cert_field(self, field_name: str) -> bool:
"""判断字段是否属于 Certificate 表（且不是 SalePlan 共有字段）
先经过 FIELD_ALIASES 映射，再判断归属。
Args:
field_name: 字段名（支持别名）
Returns:
bool: True 表示该字段仅属于 Certificate 表
"""
actual_name = self.FIELD_ALIASES.get(field_name, field_name)
return actual_name in self.certificate_fields and actual_name not in self.saleplan_fields
def _build_boolean_filter(
self,
field: Any,
operator: str,
value: bool,
*,
is_hybrid: bool = False,
) -> Any:
"""为布尔字段/hybrid_property 构建过滤条件
Args:
field: SQLAlchemy 列属性或 hybrid_property
operator: 操作符名称
value: 布尔值
is_hybrid: 是否为 hybrid_property 类型
"""
if is_hybrid:
# hybrid_property 布尔字段：直接使用表达式本身作为条件，
# 避免 field == True 生成 (expression) = 1 这类无效 SQL
if operator in ("contains", "eq"):
return field if value else ~field
if operator == "ne":
return ~field if value else field
logger.debug(f"混合属性不支持操作符: {operator}，已回退为 eq")
return field if value else ~field
# 普通 Column 布尔列使用精确匹配（eq/ne）
if operator in ("contains", "eq"):
return field == value
if operator == "ne":
return field != value
logger.debug(f"布尔字段不支持操作符: {operator}，已回退为 eq")
return field == value
def _build_filter(self, field: Any, operator: str, value: Any) -> Any:
"""根据操作符构建单个 SQLAlchemy 过滤条件
Args:
field: SQLAlchemy 列属性
operator: 操作符名称 (eq/ne/contains/gt/gte/lt/lte/in/between/is_null 等)
value: 过滤值
Returns:
SQLAlchemy 过滤表达式，未知操作符返回 None
"""
from sqlalchemy import Boolean as BooleanType
from sqlalchemy.ext.hybrid import hybrid_property as hybrid_property_type
# 检测布尔字段：Column(Boolean) 有 .type 属性，hybrid_property 无
field_is_bool = hasattr(field, "type") and isinstance(field.type, BooleanType)
field_is_hybrid = isinstance(field, hybrid_property_type)
# 自动转换布尔值（字符串 "True"/"true"/"False"/"false" → bool）
if isinstance(value, str) and value.lower() in ("true", "false"):
value = value.lower() == "true"
# 布尔/hybrid 字段处理
if isinstance(value, bool) and (field_is_bool or field_is_hybrid):
return self._build_boolean_filter(field, operator, value, is_hybrid=field_is_hybrid)
if operator == "contains":
return field.contains(str(value))
if operator == "in":
return field.in_(value)
if operator == "between":
if isinstance(value, list) and len(value) == 2:
return and_(field >= value[0], field <= value[1])
logger.warning(f"between 操作符需要 [min, max] 列表: {value}")
return None
op_func = self.OPERATORS.get(operator)
if not op_func:
logger.warning(f"未知操作符: {operator}")
return None
return op_func(field, value)
def _build_filters(self, conditions: dict[str, Any]) -> list[Any]:
"""从查询条件字典构建 SQLAlchemy 过滤表达式列表
Args:
conditions: 查询条件字典
Returns:
过滤表达式列表
"""
filters = []
for field_name, condition in conditions.items():
# 支持两种条件格式：
# - 富格式: {"operator": "contains", "value": "V100"}
# - 简写格式: "V100" / 123（标量即按 `eq` 精确匹配）
if not isinstance(condition, dict):
condition = {"operator": "eq", "value": condition}
operator = condition.get("operator") or "contains"
value = condition.get("value")
# is_null / is_not_null 的 value 可以是 None，不能跳过
if operator not in ("is_null", "is_not_null") and (
value is None or (isinstance(value, str) and not value.strip())
):
continue
field = self._resolve_field(field_name)
if field is None:
continue
# 生产状态使用专门的过滤方法（同时匹配 execution_status）
if field_name == "production_status" and operator in ("eq", "contains"):
filter_expr = self.build_status_filter(value)
if filter_expr is not None:
filters.append(filter_expr)
continue
filter_expr = self._build_filter(field, operator, value)
if filter_expr is not None:
filters.append(filter_expr)
return filters
def _build_or_filters(self, or_filters: Any) -> list[Any]:
"""将 ``or_filters``（``[{field, operator, value}, ...]``）转为 SQLAlchemy 表达式列表。
用于快查「无分隔符」场景：在 preset 全部字段上做 ``contains`` OR 匹配，
任一字段命中即返回（避免「只匹配首字段 → 0 条」）。
"""
exprs: list[Any] = []
for cond in or_filters or []:
if not isinstance(cond, dict):
continue
field_name = cond.get("field")
if not field_name:
continue
operator = cond.get("operator") or "contains"
value = cond.get("value")
if value is None or (isinstance(value, str) and not value.strip()):
continue
field = self._resolve_field(field_name)
if field is None:
continue
if field_name == "production_status" and operator in ("eq", "contains"):
fe = self.build_status_filter(value)
else:
fe = self._build_filter(field, operator, value)
if fe is not None:
exprs.append(fe)
return exprs
def _needs_join(
self,
conditions: dict[str, Any],
or_filters: Any,
order_by: str | None = None,
) -> bool:
"""判断查询是否需要在 SalePlan 与 Certificate 间 JOIN。"""
return (
any(self._is_cert_field(fn) for fn in conditions)
or any(self._is_cert_field(c.get("field")) for c in (or_filters or []))
or (order_by is not None and self._is_cert_field(order_by))
)
def _apply_conditions(
self, query: Any, conditions: dict[str, Any], or_filters: Any = None
) -> Any:
"""在主 conditions（AND）基础上，将 or_filters 以 OR 组合并应用过滤。"""
filters = self._build_filters(conditions)
or_exprs = self._build_or_filters(or_filters)
if or_exprs:
filters.append(or_(*or_exprs))
if filters:
query = query.filter(and_(*filters))
return query
[文档]
def build_query(self, conditions: dict[str, Any], or_filters: Any = None) -> Any:
"""根据条件构建查询对象
自动识别字段所属表：
- SalePlan 字段直接过滤
- Certificate 字段通过 LEFT JOIN 过滤
Args:
conditions: 查询条件字典，格式:
{
"字段名": {
"value": 查询值,
"operator": "操作符" # 默认 "contains"
}
}
or_filters: 可选 OR 条件列表 ``[{field, operator, value}, ...]``，
与主 conditions（AND）以 OR 组合，用于跨字段「任一命中」检索。
Returns:
SQLAlchemy查询对象（可能包含 JOIN）
"""
needs_join = self._needs_join(conditions, or_filters)
query = self.session.query(SalePlan)
if needs_join:
query = query.outerjoin(
Certificate,
SalePlan.unique_key == Certificate.unique_key,
).options(selectinload(SalePlan.certificates))
return self._apply_conditions(query, conditions, or_filters)
[文档]
def query(
self,
conditions: dict[str, Any] = None,
page: int = 1,
page_size: int = 50,
order_by: str = "sort_group",
order_desc: bool = False,
or_filters: Any = None,
) -> dict[str, Any]:
"""执行多条件组合查询
Args:
conditions: 查询条件字典
page: 页码（从1开始）
page_size: 每页记录数
order_by: 排序字段
order_desc: 是否降序
Returns:
Dict: 包含 total, page, page_size, total_pages, records, conditions
"""
conditions = conditions or {}
# 判断是否需要 JOIN（筛选条件 + OR 跨字段条件 + 排序字段，一次性判断）
needs_join = self._needs_join(conditions, or_filters, order_by)
query = self.session.query(SalePlan)
if needs_join:
query = query.outerjoin(
Certificate, SalePlan.unique_key == Certificate.unique_key
).options(selectinload(SalePlan.certificates))
query = self._apply_conditions(query, conditions, or_filters)
# 排序
actual_order_by = self.FIELD_ALIASES.get(order_by, order_by)
python_sort_field: str | None = None
if self._is_cert_field(actual_order_by):
order_field = getattr(Certificate, actual_order_by)
elif hasattr(SalePlan, actual_order_by):
attr = getattr(SalePlan, actual_order_by)
# hybrid_property 只有定义了 SQL 表达式（expr is not None）才能用于排序
from sqlalchemy.ext.hybrid import hybrid_property as hp_type
descriptor = getattr(attr, "descriptor", None)
if isinstance(descriptor, hp_type) and descriptor.expr is None:
# 仅有 Python getter 的 hybrid_property，改为 Python 内存排序
python_sort_field = actual_order_by
order_field = None
else:
order_field = attr
else:
order_field = SalePlan.sort_group
if python_sort_field:
# Python 排序：只查主键 + 排序字段，排序后按主键查当前页完整记录
pk_col = SalePlan.id
all_ids = query.with_entities(pk_col, getattr(SalePlan, python_sort_field)).all()
total = len(all_ids)
def _sort_key(row):
val = row[1]
if isinstance(val, str) and val.endswith("%"):
return int(val.rstrip("%"))
return val or ""
all_ids.sort(key=_sort_key, reverse=order_desc)
offset = (page - 1) * page_size
page_ids = [row[0] for row in all_ids[offset : offset + page_size]]
# 按主键批量查当前页完整记录，保持排序顺序
if page_ids:
records = self.session.query(SalePlan).filter(SalePlan.id.in_(page_ids)).all()
id_order = {pk: i for i, pk in enumerate(page_ids)}
records.sort(key=lambda r: id_order[r.id])
else:
records = []
else:
query = query.order_by(order_field.desc() if order_desc else order_field)
# 统计总数
total = query.count()
# 分页
offset = (page - 1) * page_size
records = query.offset(offset).limit(page_size).all()
# 计算总页数
total_pages = (total + page_size - 1) // page_size if total > 0 else 1
logger.info(f"查询完成: 总记录数={total}, 当前页={page}, 每页={page_size}")
return {
"total": total,
"page": page,
"page_size": page_size,
"total_pages": total_pages,
"records": records,
"conditions": conditions,
}
[文档]
def get_field_values(
self, field_name: str, search: str = None, order_desc: bool = True
) -> list[str]:
"""获取某字段的所有唯一值（用于下拉框）
Args:
field_name: 字段名
search: 搜索关键字（可选）
order_desc: 是否降序排序（最新在前），默认True
Returns:
list[str]: 所有唯一值列表
"""
if not hasattr(SalePlan, field_name):
logger.warning(f"未知字段: {field_name}")
return []
field = getattr(SalePlan, field_name)
query = self.session.query(field).filter(field.isnot(None), field != "")
if search:
query = query.filter(field.contains(search))
# 排序
query = query.order_by(field.desc() if order_desc else field.asc())
values = query.distinct().all()
return [v[0] for v in values if v[0]]
[文档]
def get_distinct_field_combos(
self, fields: list[str], conditions: dict[str, Any] | None = None
) -> list[str]:
"""返回指定字段组合的去重值列表（全量，不受分页影响），用于快查框候选项。
跨整个结果集（尊重 conditions，但忽略分页）计算去重组合，
避免候选项仅来自当前页导致「越查越窄 / 只能选已显示项」。
Args:
fields: 字段名列表（如 ["customer", "product_model"]）
conditions: 当前筛选条件（可选），把候选项限定在已筛选范围内
Returns:
list[str]: 去重后的组合字符串（字段值以 QUICK_SEARCH_SEP 连接，保留空段），按出现顺序
"""
cols = []
for f in fields:
if hasattr(SalePlan, f):
cols.append(getattr(SalePlan, f))
elif hasattr(Certificate, f):
cols.append(getattr(Certificate, f))
else:
logger.warning(f"未知字段(快查候选项): {f}")
if not cols:
return []
needs_join = any(self._is_cert_field(f) for f in fields)
query = self.session.query(*cols)
if needs_join:
query = query.join(Certificate, SalePlan.unique_key == Certificate.unique_key)
filters = self._build_filters(conditions or {})
if filters:
query = query.filter(and_(*filters))
rows = query.distinct().all()
seen: set[str] = set()
combos: list[str] = []
for row in rows:
# 保留空段（不跳过），确保段数恒等于字段数，回填时按位置一一对应不错位
vals = [str(v or "") for v in row]
key = QUICK_SEARCH_SEP.join(vals)
if key not in seen:
seen.add(key)
combos.append(key)
return combos
[文档]
def search(
self,
conditions: dict[str, Any] = None,
page: int = 1,
page_size: int = 50,
order_by: str = "sort_group",
order_desc: bool = False,
) -> dict[str, Any]:
"""执行查询并返回与 ``query`` 一致的结果字典
兼容旧测试/调用方：在 ``query`` 基础上将记录字段命名为 ``results``
（``query`` 使用 ``records``），其余字段（total/page/page_size/
total_pages/conditions）保持一致。
Args:
conditions: 查询条件字典
page: 页码（从1开始）
page_size: 每页记录数
order_by: 排序字段
order_desc: 是否降序
Returns:
Dict: 包含 total, page, page_size, total_pages, results, conditions
"""
result = self.query(
conditions=conditions,
page=page,
page_size=page_size,
order_by=order_by,
order_desc=order_desc,
)
result["results"] = result.pop("records", [])
return result
[文档]
def get_statistics(self, conditions: dict[str, Any] = None) -> dict[str, Any]:
"""获取查询结果统计信息
Args:
conditions: 查询条件
Returns:
Dict: 统计信息
"""
query = self.build_query(conditions or {})
# 数量统计
total = query.count()
# 总数量（按 quantity 求和，缺失按 0 计）
total_quantity = (
query.with_entities(func.coalesce(func.sum(SalePlan.quantity), 0)).scalar() or 0
)
# 分组统计（按生产状态）
status_stats = {}
status_results = (
query.with_entities(SalePlan.production_status, func.count(SalePlan.id))
.group_by(SalePlan.production_status)
.all()
)
for status, count in status_results:
status_stats[status or "未设置"] = count
return {
"total": total,
"total_count": total,
"total_quantity": total_quantity,
"status_stats": status_stats,
"by_status": status_stats,
}
# ============================================================
# CRUD 操作（从 QueryController 迁移）
# ============================================================
[文档]
def get_records_by_ids(self, ids: list[int]) -> list[Any]:
"""根据 ID 列表获取 SalePlan 记录
Args:
ids: 记录 ID 列表
Returns:
记录列表
"""
return self.session.query(SalePlan).filter(SalePlan.id.in_(ids)).all()
[文档]
def get_certificate_count(self, plan_ids: list[int]) -> int:
"""获取关联的合格证数量
Args:
plan_ids: 销售计划 ID 列表
Returns:
关联的合格证数量
"""
return (
self.session.query(Certificate).filter(Certificate.sale_plan_id.in_(plan_ids)).count()
)
[文档]
def delete_records(self, ids: list[int]) -> tuple[bool, int, str]:
"""删除 SalePlan 记录（事务回滚）
检查是否有关联合格证，如有则拒绝删除。
Args:
ids: 记录 ID 列表
Returns:
tuple[bool, int, str]: (是否成功, 删除数量, 消息)
"""
from sqlalchemy.exc import SQLAlchemyError
plans = self.session.query(SalePlan).filter(SalePlan.id.in_(ids)).all()
if not plans:
return False, 0, "未找到要删除的记录"
cert_count = self.get_certificate_count([p.id for p in plans])
if cert_count > 0:
return False, 0, f"存在 {cert_count} 条关联合格证，请使用级联删除"
deleted_count = 0
try:
for plan in plans:
self.session.delete(plan)
deleted_count += 1
self.session.commit()
logger.info(f"成功删除 {deleted_count} 条记录")
return True, deleted_count, ""
except SQLAlchemyError as e:
self.session.rollback()
error_msg = f"数据库错误: {e}"
logger.error(f"删除失败，已回滚: {e}")
return False, 0, error_msg
except Exception as e:
self.session.rollback()
error_msg = f"未知错误: {e}"
logger.error(f"删除失败，已回滚: {e}")
return False, 0, error_msg
[文档]
def delete_with_certificates(self, ids: list[int]) -> tuple[bool, int, int, str]:
"""级联删除 SalePlan 记录及关联的 Certificate
Args:
ids: 记录 ID 列表
Returns:
tuple[bool, int, int, str]: (是否成功, 删除计划数, 删除合格证数, 消息)
"""
from sqlalchemy.exc import SQLAlchemyError
plans = self.session.query(SalePlan).filter(SalePlan.id.in_(ids)).all()
if not plans:
return False, 0, 0, "未找到要删除的记录"
plan_ids = [p.id for p in plans]
deleted_count = 0
cert_deleted = 0
try:
cert_deleted = (
self.session.query(Certificate)
.filter(Certificate.sale_plan_id.in_(plan_ids))
.delete(synchronize_session=False)
)
for plan in plans:
self.session.delete(plan)
deleted_count += 1
self.session.commit()
msg = f"已删除 {deleted_count} 条记录及 {cert_deleted} 条合格证"
logger.info(msg)
return True, deleted_count, cert_deleted, msg
except SQLAlchemyError as e:
self.session.rollback()
error_msg = f"数据库错误: {e}"
logger.error(f"级联删除失败，已回滚: {e}")
return False, 0, 0, error_msg
except Exception as e:
self.session.rollback()
error_msg = f"未知错误: {e}"
logger.error(f"级联删除失败，已回滚: {e}")
return False, 0, 0, error_msg
[文档]
def batch_update_field(self, ids: list[int], field_name: str, value: Any) -> int:
"""批量更新 SalePlan 字段（事务回滚）
Args:
ids: 记录 ID 列表
field_name: 字段名
value: 要设置的值
Returns:
更新记录数
"""
from sqlalchemy.exc import SQLAlchemyError
plans = self.session.query(SalePlan).filter(SalePlan.id.in_(ids)).all()
updated = 0
try:
for plan in plans:
old_value = getattr(plan, field_name)
setattr(plan, field_name, value)
updated += 1
logger.debug(f"更新 {field_name}: {old_value!r} → {value!r} (ID={plan.id})")
# #30 P0 A 方案：手动改发货状态时写入溯源，使其在下月重导入时不受字体色覆盖
if field_name == "shipping_status":
self._stamp_shipping_source(plans, source="manual")
self.session.commit()
logger.info(f"批量更新 {field_name}={value} 完成，共 {updated} 条")
return updated
except SQLAlchemyError as e:
self.session.rollback()
logger.error(f"批量更新失败，已回滚: {e}")
raise
except Exception as e:
self.session.rollback()
logger.error(f"批量更新失败，已回滚: {e}")
raise
[文档]
def batch_update_production_status(
self, ids: list[int], status: str, update_relations: list[dict] | None = None
) -> int:
"""批量更新生产状态（支持关联字段更新）
Args:
ids: 记录 ID 列表
status: 目标状态
update_relations: 关联更新规则列表
Returns:
更新记录数
"""
from datetime import datetime
from sqlalchemy.exc import SQLAlchemyError
plans = self.session.query(SalePlan).filter(SalePlan.id.in_(ids)).all()
updated = 0
update_relations = update_relations or []
try:
for plan in plans:
old_status = plan.production_status
plan.production_status = status
for rel in update_relations:
if rel.get("when_status", "") == status:
for f_name, f_val in rel.get("set_fields", {}).items():
actual_val = datetime.now() if f_val == "datetime('now')" else f_val
setattr(plan, f_name, actual_val)
updated += 1
logger.debug(f"更新生产状态: {old_status!r} → {status!r} (ID={plan.id})")
self.session.commit()
logger.info(f"批量更新生产状态完成: {status}，共 {updated} 条")
return updated
except SQLAlchemyError as e:
self.session.rollback()
logger.error(f"批量更新生产状态失败，已回滚: {e}")
raise
def _stamp_shipping_source(self, plans: list[SalePlan], source: str = "manual") -> None:
"""为发货状态写入溯源信息（#30 P0 A 方案）
使 UI 手动设置的发货状态在重导入时不被字体色派生值还原。
Args:
plans: 已更新 shipping_status 的 SalePlan 列表
source: 来源标记，'manual'=UI 手动设置
"""
from datetime import datetime
from getpass import getuser
now = datetime.now()
operator = getuser() or "unknown"
for plan in plans:
plan.shipping_status_source = source
plan.shipping_status_set_at = now
plan.shipping_status_set_by = operator
# 手动标记已发货时记录发货日期
if (plan.shipping_status or "") == "已发货" and not (plan.shipped_date or "").strip():
plan.shipped_date = now.strftime("%Y-%m-%d")
[文档]
def copy_cert_info_to_plan(self, ids: list[int]) -> int:
"""复制合同信息到合格证字段
Args:
ids: 记录 ID 列表
Returns:
更新记录数
"""
from sqlalchemy.exc import SQLAlchemyError
plans = self.session.query(SalePlan).filter(SalePlan.id.in_(ids)).all()
updated = 0
try:
for plan in plans:
if not plan.cert_product_name:
plan.cert_product_name = plan.product_name
if not plan.cert_product_model:
plan.cert_product_model = plan.product_model
if not plan.cert_product_spec:
plan.cert_product_spec = plan.product_spec
updated += 1
self.session.commit()
logger.info(f"复制合同信息到合格证字段，共 {updated} 条")
return updated
except SQLAlchemyError as e:
self.session.rollback()
logger.error(f"复制失败，已回滚: {e}")
raise
[文档]
def fill_plan_no(self, ids: list[int], sequence: str) -> int:
"""填充计划单号（批量更新版本）
使用每条记录的 plan_date 生成对应的年月格式 plan_no。
如果 plan_date 为空，则回退到当前年月。
Args:
ids: 记录 ID 列表
sequence: 序号
Returns:
更新记录数
"""
from collections import defaultdict
from datetime import datetime
from sqlalchemy.exc import SQLAlchemyError
plans = (
self.session.query(SalePlan)
.filter(SalePlan.id.in_(ids), SalePlan.plan_no.is_(None))
.all()
)
if not plans:
logger.info("没有需要填充的计划单号")
return 0
stats: dict[str, int] = defaultdict(int)
update_mappings: list[dict] = []
for plan in plans:
if plan.plan_date:
ym = plan.plan_date.strftime("%Y%m")
else:
ym = datetime.now().strftime("%Y%m")
logger.warning(f"计划 {plan.id} plan_date 为空，使用当前年月 {ym}")
update_mappings.append({"id": plan.id, "plan_no": f"{ym}_{sequence}"})
stats[ym] += 1
try:
self.session.bulk_update_mappings(SalePlan, update_mappings)
self.session.commit()
stats_msg = ", ".join(f"{ym}: {cnt}条" for ym, cnt in sorted(stats.items()))
logger.info(
f"填充计划单号 {len(update_mappings)} 条，序号: {sequence}，分布: {stats_msg}"
)
return len(update_mappings)
except SQLAlchemyError as e:
self.session.rollback()
logger.error(f"填充计划单号失败，已回滚: {e}")
raise
[文档]
def get_certificates_by_plan_ids(self, plan_ids: list[int]) -> list[Any]:
"""根据销售计划 ID 查询关联的合格证记录
Args:
plan_ids: 销售计划 ID 列表
Returns:
合格证记录列表
"""
return self.session.query(Certificate).filter(Certificate.sale_plan_id.in_(plan_ids)).all()
[文档]
def get_latest_certificate(self, plan_ids: list[int]) -> Any:
"""获取最近创建的一条合格证记录（用于参数补充）
Args:
plan_ids: 销售计划 ID 列表
Returns:
最近创建的合格证记录，或 None
"""
return (
self.session.query(Certificate)
.filter(
Certificate.sale_plan_id.in_(plan_ids),
Certificate.print_status == "已打印",
)
.order_by(Certificate.id.desc())
.first()
)