duxingchen/KCGL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 基础信息排序问题

Browse Source
This commit is contained in:
dxc
2026-03-12 09:29:50 +08:00
parent 3ecdb121d2
commit d8e86959b8
5 changed files with 234 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

269
inventory-backend/app/services/inbound/base_service.py
View File

@ -114,6 +114,10 @@ class MaterialBaseService:
"""
获取基础信息列表 (带分页、高级筛选和全字段排序)
支持库存预警功能（如果用户有 view_warning 权限）
修复说明：
1. 使用子查询方案确保聚合列(total_inv)能正确参与预警排序计算
2. 高级筛选中针对聚合字段使用 HAVING 而非 WHERE
"""
try:
# 检查用户是否有查看预警的权限
@ -144,7 +148,7 @@ class MaterialBaseService:
func.sum(StockProduct.available_quantity).label('prod_avail')
).group_by(StockProduct.base_id).subquery()
# 总库存和可用数的 SQL 表达式
# 总库存和可用数的 SQL 表达式（用于后续计算）
total_inv = func.coalesce(buy_sub.c.buy_inv, 0) + \
func.coalesce(semi_sub.c.semi_inv, 0) + \
func.coalesce(prod_sub.c.prod_inv, 0)
@ -155,155 +159,180 @@ class MaterialBaseService:
# 导入预警设置模型
from app.models.base import MaterialWarningSetting
# 主查询，关联聚合子查询和预警设置
query = db.session.query(
# ============================================================
# 【核心修复】使用子查询方案：将聚合结果作为子查询
# 这样 total_inv 在外层查询中是一个普通列，可安全参与计算
# ============================================================
# 内层子查询：基础数据 + 聚合库存
inner_sub = db.session.query(
MaterialBase.id.label('base_id'),
MaterialBase,
total_inv.label('total_inv'),
total_avail.label('total_avail'),
MaterialWarningSetting.is_enabled.label('warning_enabled'),
MaterialWarningSetting.yellow_threshold.label('warning_yellow'),
MaterialWarningSetting.red_threshold.label('warning_red')
total_avail.label('total_avail')
).outerjoin(buy_sub, MaterialBase.id == buy_sub.c.base_id) \
.outerjoin(semi_sub, MaterialBase.id == semi_sub.c.base_id) \
.outerjoin(prod_sub, MaterialBase.id == prod_sub.c.base_id) \
.outerjoin(MaterialWarningSetting, MaterialBase.id == MaterialWarningSetting.base_id)
.outerjoin(prod_sub, MaterialBase.id == prod_sub.c.base_id)
# 关键词精确搜索（需要在子查询层完成）
if filters:
# 1. 关键词精准搜索（支持指定字段）
search_field = filters.get('searchField', 'all')
keyword = filters.get('keyword')
if keyword:
kw = f"%{keyword}%"
if search_field == 'name':
query = query.filter(MaterialBase.name.ilike(kw))
inner_sub = inner_sub.filter(MaterialBase.name.ilike(kw))
elif search_field == 'common_name':
query = query.filter(MaterialBase.common_name.ilike(kw))
inner_sub = inner_sub.filter(MaterialBase.common_name.ilike(kw))
elif search_field == 'spec':
query = query.filter(MaterialBase.spec_model.ilike(kw))
else: # 'all' 默认全局模糊匹配
query = query.filter(or_(
inner_sub = inner_sub.filter(MaterialBase.spec_model.ilike(kw))
else:
inner_sub = inner_sub.filter(or_(
MaterialBase.name.ilike(kw),
MaterialBase.common_name.ilike(kw),
MaterialBase.spec_model.ilike(kw)
))
# 2. 精确筛选
company = filters.get('company')
if company is not None and company != '':
query = query.filter(MaterialBase.company_name.ilike(company.strip()))
inner_sub = inner_sub.filter(MaterialBase.company_name.ilike(company.strip()))
category = filters.get('category')
if category is not None and category != '':
query = query.filter(MaterialBase.category.ilike(category.strip()))
inner_sub = inner_sub.filter(MaterialBase.category.ilike(category.strip()))
type_val = filters.get('type')
if type_val is not None and type_val != '':
query = query.filter(MaterialBase.material_type.ilike(type_val.strip()))
inner_sub = inner_sub.filter(MaterialBase.material_type.ilike(type_val.strip()))
# 【核心修改】：增强布尔值解析
if filters.get('isEnabled') is not None:
val_str = str(filters['isEnabled']).lower()
is_active = val_str in ['1', 'true', 'yes', 't']
# 必须使用 filter() 而非 filter_by()，因为 query 是 join 后的复杂查询
query = query.filter(MaterialBase.is_enabled == is_active)
inner_sub = inner_sub.filter(MaterialBase.is_enabled == is_active)
# 【新增】：库存状态筛选 (has_stock)
# 库存状态筛选 (has_stock) - 使用子查询列
has_stock = filters.get('has_stock')
if has_stock and str(has_stock).lower() in ['true', '1', 'yes']:
query = query.filter(total_inv > 0)
inner_sub = inner_sub.filter(total_inv > 0)
# 3. 高级动态筛选
advanced_filters = filters.get('advancedFilters', [])
if advanced_filters:
allowed_fields = {
'companyName': 'company_name',
'name': 'name',
'commonName': 'common_name',
'category': 'category',
'type': 'material_type',
'spec': 'spec_model',
'unit': 'unit',
'inventoryCount': total_inv,
'availableCount': total_avail
}
# 字段到权限码的映射
field_permission_map = {
'companyName': 'material_list:companyName',
'name': 'material_list:name',
'commonName': 'material_list:commonName',
'category': 'material_list:category',
'type': 'material_list:type',
'spec': 'material_list:spec',
'unit': 'material_list:unit',
'inventoryCount': 'material_list:inventoryCount',
'availableCount': 'material_list:availableCount'
}
filter_conditions = []
for condition in advanced_filters:
field = condition.get('field')
operator = condition.get('operator')
value = condition.get('value')
if not field or not operator or value is None:
# 将内层子查询具体化
inner_sub = inner_sub.subquery()
# 外层查询：关联预警设置，并在外层处理排序和高级筛选
query = db.session.query(
MaterialBase,
inner_sub.c.total_inv,
inner_sub.c.total_avail,
MaterialWarningSetting.is_enabled.label('warning_enabled'),
MaterialWarningSetting.yellow_threshold.label('warning_yellow'),
MaterialWarningSetting.red_threshold.label('warning_red')
).outerjoin(inner_sub, MaterialBase.id == inner_sub.c.base_id) \
.outerjoin(MaterialWarningSetting, MaterialBase.id == MaterialWarningSetting.base_id)
# ============================================================
# 【修复3】高级筛选：聚合字段使用 HAVING，非聚合字段使用 WHERE
# ============================================================
advanced_filters = filters.get('advancedFilters', []) if filters else []
having_conditions = []
filter_conditions = []
if advanced_filters:
allowed_fields = {
'companyName': 'company_name',
'name': 'name',
'commonName': 'common_name',
'category': 'category',
'type': 'material_type',
'spec': 'spec_model',
'unit': 'unit',
'inventoryCount': 'total_inv',
'availableCount': 'total_avail'
}
# 聚合字段列表（这些需要用 HAVING）
aggregate_fields = {'inventoryCount', 'availableCount'}
field_permission_map = {
'companyName': 'material_list:companyName',
'name': 'material_list:name',
'commonName': 'material_list:commonName',
'category': 'material_list:category',
'type': 'material_list:type',
'spec': 'material_list:spec',
'unit': 'material_list:unit',
'inventoryCount': 'material_list:inventoryCount',
'availableCount': 'material_list:availableCount'
}
for condition in advanced_filters:
field = condition.get('field')
operator = condition.get('operator')
value = condition.get('value')
if not field or not operator or value is None:
continue
db_field = allowed_fields.get(field)
if not db_field:
continue
# 权限校验
if user_permissions is not None:
perm_code = field_permission_map.get(field)
if 'material_list:*' in user_permissions:
pass
elif perm_code and perm_code not in user_permissions:
continue
db_field = allowed_fields.get(field)
if not db_field:
# 判断是否为聚合字段
is_aggregate = field in aggregate_fields
if is_aggregate:
# 聚合字段：使用 HAVING - 通过子查询列进行比较
col = inner_sub.c.total_inv if field == 'inventoryCount' else inner_sub.c.total_avail
try:
num_val = float(value)
except ValueError:
continue
# 权限校验
if user_permissions is not None:
perm_code = field_permission_map.get(field)
if 'material_list:*' in user_permissions:
# 超级管理员拥有全部权限
pass
elif perm_code and perm_code not in user_permissions:
# 无权限，跳过该条件
continue
# 对于聚合字段 (inventoryCount, availableCount)，需要使用子查询别名
if isinstance(db_field, type(total_inv)):
column = db_field
else:
column = getattr(MaterialBase, db_field, None)
if operator == 'eq':
having_conditions.append(col == num_val)
elif operator == 'ne':
having_conditions.append(col != num_val)
elif operator == 'ge':
having_conditions.append(col >= num_val)
elif operator == 'le':
having_conditions.append(col <= num_val)
else:
# 非聚合字段：使用 WHERE
column = getattr(MaterialBase, db_field, None)
if column is None:
continue
# 处理操作符
# 对于数值型列（聚合字段）只支持 eq, ne, ge, le
if isinstance(column, type(total_inv)):
# 数值型列
if operator == 'eq':
filter_conditions.append(column == value)
elif operator == 'ne':
filter_conditions.append(column != value)
elif operator == 'contains':
filter_conditions.append(column.ilike(f'%{value}%'))
elif operator == 'ge':
try:
num_val = float(value)
except ValueError:
# 转换失败则跳过该条件
continue
if operator == 'eq':
filter_conditions.append(column == num_val)
elif operator == 'ne':
filter_conditions.append(column != num_val)
elif operator == 'ge':
filter_conditions.append(column >= num_val)
elif operator == 'le':
except ValueError:
continue
elif operator == 'le':
try:
num_val = float(value)
filter_conditions.append(column <= num_val)
# 对于 contains 操作符，数值型列不支持，忽略
else:
# 字符串型列
if operator == 'eq':
filter_conditions.append(column == value)
elif operator == 'ne':
filter_conditions.append(column != value)
elif operator == 'contains':
filter_conditions.append(column.ilike(f'%{value}%'))
elif operator == 'ge':
try:
num_val = float(value)
filter_conditions.append(column >= num_val)
except ValueError:
continue
elif operator == 'le':
try:
num_val = float(value)
filter_conditions.append(column <= num_val)
except ValueError:
continue
if filter_conditions:
query = query.filter(and_(*filter_conditions))
except ValueError:
continue
# 应用 WHERE 条件
if filter_conditions:
query = query.filter(and_(*filter_conditions))
# 应用 HAVING 条件（聚合字段筛选）
if having_conditions:
query = query.having(and_(*having_conditions))
# 排序处理（支持全字段）
order_by_column = filters.get('orderByColumn', '')
@ -314,25 +343,31 @@ class MaterialBaseService:
if enable_warning_sort:
print("====== [DEBUG] 成功进入预警强排逻辑 (SQLA 2.0) ======")
# 强制统一数据类型
# 【核心修复】使用子查询列进行预警排序计算
# total_inv 现在是 inner_sub.c.total_inv，可以安全参与 case 计算
inv_val = inner_sub.c.total_inv
red_val = cast(MaterialWarningSetting.red_threshold, Numeric)
yellow_val = cast(MaterialWarningSetting.yellow_threshold, Numeric)
inv_val = cast(total_inv, Numeric)
# 注意：移除了 case 内部的 []
# 预警等级计算：红=2, 黄=1, 正常=0
warning_level = case(
(and_(MaterialWarningSetting.is_enabled.is_(True), inv_val <= red_val), 2),
(and_(MaterialWarningSetting.is_enabled.is_(True), inv_val <= yellow_val), 1),
else_=0
).label('sort_level')
# 红色预警时的缺口（库存与红色阈值的差距）
red_shortage = case(
(and_(MaterialWarningSetting.is_enabled.is_(True), inv_val <= red_val), red_val - inv_val),
else_=0
).label('sort_red')
# 黄色预警时的缺口
yellow_distance = case(
(and_(MaterialWarningSetting.is_enabled.is_(True), inv_val > red_val, inv_val <= yellow_val), inv_val - red_val),
else_=999999
).label('sort_yellow')
query = query.add_columns(warning_level, red_shortage, yellow_distance)
query = query.order_by(None).order_by(
text("sort_level DESC"),
@ -341,7 +376,7 @@ class MaterialBaseService:
MaterialBase.id.desc()
)
elif order_by_column:
# 字段映射
# 字段映射 - 使用子查询列进行排序
sort_field_map = {
'companyName': MaterialBase.company_name,
'name': MaterialBase.name,
@ -350,8 +385,8 @@ class MaterialBaseService:
'type': MaterialBase.material_type,
'spec': MaterialBase.spec_model,
'unit': MaterialBase.unit,
'inventoryCount': total_inv,
'availableCount': total_avail
'inventoryCount': inner_sub.c.total_inv,
'availableCount': inner_sub.c.total_avail
}
sort_column = sort_field_map.get(order_by_column)
if sort_column is not None:
@ -361,7 +396,7 @@ class MaterialBaseService:
query = query.order_by(sort_column.desc())
else:
# 默认排序：优先按总库存数降序，当库存相同时，再按规格型号升序
query = query.order_by(total_inv.desc(), MaterialBase.spec_model.asc())
query = query.order_by(inner_sub.c.total_inv.desc(), MaterialBase.spec_model.asc())
# 分页
pagination = query.paginate(page=page, per_page=limit, error_out=False)