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

全局审计日fix: 使用鸭子类型强制安全解包 SQLAlchemy Row 对象，彻底解决 to_dict 报错志

Browse Source
This commit is contained in:
dxc
2026-03-11 13:11:16 +08:00
parent ac97c6066b
commit d2d9abe201
8 changed files with 434 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
inventory-backend/app/__init__.py
View File

@ -105,7 +105,17 @@ def create_app():
print(f"❌ 错误: Outbound 模块导入失败: {e}")
# -----------------------------------------------------
# 2.6 注册 BOM 模块
# 2.6 注册报废模块
# -----------------------------------------------------
try:
from app.api.v1.scrap import scrap_bp
app.register_blueprint(scrap_bp, url_prefix='/api/v1/scrap')
print("✅ Scrap 模块注册成功")
except ImportError as e:
print(f"❌ 错误: Scrap 模块导入失败: {e}")
# -----------------------------------------------------
# 2.7 注册 BOM 模块
# -----------------------------------------------------
try:
from app.api.v1.bom import bom_bp

76
inventory-backend/app/api/v1/inbound/base.py
View File

@ -122,7 +122,8 @@ def get_list():
'isEnabled': request.args.get('isEnabled', None),
'orderByColumn': request.args.get('orderByColumn', ''),
'isAsc': request.args.get('isAsc', None),
'advancedFilters': advanced_filters_list
'advancedFilters': advanced_filters_list,
'enableWarningSort': request.args.get('enableWarningSort', 'false').lower() == 'true'
}
user_permissions = get_current_user_permissions()
@ -325,3 +326,76 @@ def delete(id):
except Exception as e:
traceback.print_exc()
return jsonify({"code": 500, "msg": str(e)}), 500
# ==============================================================================
# 2.5 批量设置预警 API (POST /api/v1/inbound/base/warning/batch-set)
# ==============================================================================
@inbound_base_bp.route('/warning/batch-set', methods=['POST'])
@permission_required('material_list:edit_warning')
def batch_set_warning():
"""
批量设置物料预警配置
请求体格式: [
{"baseId": 1, "isEnabled": true, "yellowThreshold": 10, "redThreshold": 5},
{"baseId": 2, "isEnabled": false}
]
"""
try:
data = request.get_json()
if not isinstance(data, list):
return jsonify({"code": 400, "msg": "请求体必须为数组"})
from app.models.base import MaterialWarningSetting
updated_count = 0
created_count = 0
for item in data:
base_id = item.get('baseId')
if not base_id:
continue
# 查找物料是否存在
material = MaterialBase.query.get(base_id)
if not material:
current_app.logger.warning(f"物料ID {base_id} 不存在，跳过")
continue
# 查找现有预警设置
warning = MaterialWarningSetting.query.filter_by(base_id=base_id).first()
if warning:
# 更新现有记录
if 'isEnabled' in item:
warning.is_enabled = bool(item['isEnabled'])
if 'yellowThreshold' in item:
warning.yellow_threshold = item['yellowThreshold']
if 'redThreshold' in item:
warning.red_threshold = item['redThreshold']
updated_count += 1
else:
# 创建新记录
warning = MaterialWarningSetting(
base_id=base_id,
is_enabled=item.get('isEnabled', False),
yellow_threshold=item.get('yellowThreshold'),
red_threshold=item.get('redThreshold')
)
db.session.add(warning)
created_count += 1
db.session.commit()
return jsonify({
"code": 200,
"msg": "批量设置成功",
"data": {
"created": created_count,
"updated": updated_count
}
})
except Exception as e:
db.session.rollback()
current_app.logger.error(f"批量设置预警失败: {str(e)}")
return jsonify({"code": 500, "msg": f"批量设置预警失败: {str(e)}"}), 500

2
inventory-backend/app/models/__init__.py
View File

@ -1,7 +1,7 @@
# app/models/__init__.py
# 1. 基础物料 (必须先加载，因为 buy 依赖它)
from app.models.base import MaterialBase
from app.models.base import MaterialBase, MaterialWarningSetting
# 2. 采购入库 (现在的类名是 StockBuy)
from app.models.inbound.buy import StockBuy

29
inventory-backend/app/models/base.py
View File

@ -47,6 +47,9 @@ class MaterialBase(db.Model):
# 4. 关联服务库存 (StockService)
stock_services = db.relationship('StockService', back_populates='base', lazy='dynamic')
# 5. 关联预警设置 (MaterialWarningSetting)
warning_settings = db.relationship('MaterialWarningSetting', back_populates='material', lazy='dynamic', cascade='all, delete-orphan')
def to_dict(self):
"""
序列化方法
@ -78,4 +81,30 @@ class MaterialBase(db.Model):
'generalImage': parse_list(self.product_image),
# 【核心修改】：直接返回布尔值，不再转成 1 或 0
'isEnabled': bool(self.is_enabled),
}
class MaterialWarningSetting(db.Model):
"""
物料预警设置表模型
对应数据库表: material_warning_settings
"""
__tablename__ = 'material_warning_settings'
id = db.Column(db.Integer, primary_key=True)
base_id = db.Column(db.Integer, db.ForeignKey('material_base.id'), nullable=False, comment='物料基础信息ID')
is_enabled = db.Column(db.Boolean, default=False, comment='是否启用预警')
yellow_threshold = db.Column(db.Numeric(10, 2), nullable=True, comment='黄色预警阈值')
red_threshold = db.Column(db.Numeric(10, 2), nullable=True, comment='红色预警阈值')
# 关联关系
material = db.relationship('MaterialBase', back_populates='warning_settings')
def to_dict(self):
return {
'id': self.id,
'baseId': self.base_id,
'isEnabled': bool(self.is_enabled),
'yellowThreshold': float(self.yellow_threshold) if self.yellow_threshold is not None else None,
'redThreshold': float(self.red_threshold) if self.red_threshold is not None else None
}

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

@ -113,8 +113,18 @@ class MaterialBaseService:
def get_list(page, limit, filters=None, user_permissions=None):
"""
获取基础信息列表 (带分页、高级筛选和全字段排序)
支持库存预警功能（如果用户有 view_warning 权限）
"""
try:
# 检查用户是否有查看预警的权限
has_warning_permission = False
if user_permissions:
# 超级管理员有所有权限
if '*' in user_permissions or 'material_list:*' in user_permissions:
has_warning_permission = True
elif 'material_list:view_warning' in user_permissions:
has_warning_permission = True
# 构建聚合子查询
buy_sub = db.session.query(
StockBuy.base_id,
@ -142,14 +152,21 @@ class MaterialBaseService:
func.coalesce(semi_sub.c.semi_avail, 0) + \
func.coalesce(prod_sub.c.prod_avail, 0)
# 主查询，关联聚合子查询
# 导入预警设置模型
from app.models.base import MaterialWarningSetting
# 主查询，关联聚合子查询和预警设置
query = db.session.query(
MaterialBase,
total_inv.label('total_inv'),
total_avail.label('total_avail')
total_avail.label('total_avail'),
MaterialWarningSetting.is_enabled.label('warning_enabled'),
MaterialWarningSetting.yellow_threshold.label('warning_yellow'),
MaterialWarningSetting.red_threshold.label('warning_red')
).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(prod_sub, MaterialBase.id == prod_sub.c.base_id) \
.outerjoin(MaterialWarningSetting, MaterialBase.id == MaterialWarningSetting.base_id)
if filters:
# 1. 关键词模糊搜索
@ -277,7 +294,64 @@ class MaterialBaseService:
# 排序处理（支持全字段）
order_by_column = filters.get('orderByColumn', '')
is_asc = filters.get('isAsc', None)
if order_by_column:
# 检查是否启用了预警智能排序
enable_warning_sort = has_warning_permission and filters.get('enableWarningSort', False)
if enable_warning_sort:
# 预警智能排序：先按预警状态排序，再按缺口/余量排序
# 使用 CASE 表达式计算预警状态
# 状态: 2=红(库存<=red), 1=黄(red<库存<=yellow), 0=正常/未开启
from sqlalchemy import case
# 计算预警状态
warning_status = case(
(
(MaterialWarningSetting.is_enabled == True) &
(total_inv <= MaterialWarningSetting.red_threshold),
2 # 红色预警
),
(
(MaterialWarningSetting.is_enabled == True) &
(total_inv > MaterialWarningSetting.red_threshold) &
(total_inv <= MaterialWarningSetting.yellow_threshold),
1 # 黄色预警
),
else_=0 # 正常或未开启
).label('warning_status')
# 红色组内部：按缺口降序 (red_threshold - inventory)
red_gap = case(
(
(MaterialWarningSetting.is_enabled == True) &
(total_inv <= MaterialWarningSetting.red_threshold),
MaterialWarningSetting.red_threshold - total_inv
),
else_=0
).label('red_gap')
# 黄色组内部：按余量升序 (inventory - red_threshold)，离红线越近越靠前
yellow_gap = case(
(
(MaterialWarningSetting.is_enabled == True) &
(total_inv > MaterialWarningSetting.red_threshold) &
(total_inv <= MaterialWarningSetting.yellow_threshold),
total_inv - MaterialWarningSetting.red_threshold
),
else_=999999 # 正常物料放在最后
).label('yellow_gap')
# 添加排序字段到查询
query = query.add_columns(warning_status, red_gap, yellow_gap)
# 强制排序规则：预警状态降序 -> 红色缺口降序 -> 黄色余量升序 -> 规格型号升序
query = query.order_by(
warning_status.desc(),
red_gap.desc(),
yellow_gap.asc(),
MaterialBase.spec_model.asc()
)
elif order_by_column:
# 字段映射
sort_field_map = {
'companyName': MaterialBase.company_name,
@ -304,10 +378,47 @@ class MaterialBaseService:
pagination = query.paginate(page=page, per_page=limit, error_out=False)
items_list = []
for item, inv, avail in pagination.items:
for row in pagination.items:
# 防弹解包逻辑：直接判断自身是否有 to_dict 方法
if hasattr(row, 'to_dict'):
# 说明查询只返回了 MaterialBase 单一对象
item = row
inv = avail = warning_enabled = warning_yellow = warning_red = None
else:
# 说明返回了 Row 对象 (包含多个字段)
item = row[0]
inv = row[1] if len(row) > 1 else None
avail = row[2] if len(row) > 2 else None
warning_enabled = row[3] if len(row) > 3 else False
warning_yellow = row[4] if len(row) > 4 else 0
warning_red = row[5] if len(row) > 5 else 0
# 安全兜底
if not hasattr(item, 'to_dict'):
continue
item_dict = item.to_dict()
item_dict['inventoryCount'] = float(inv) if inv is not None else 0.0
item_dict['availableCount'] = float(avail) if avail is not None else 0.0
item_dict['inventoryCount'] = float(inv) if inv is not None else 0
item_dict['availableCount'] = float(avail) if avail is not None else 0
# 处理预警信息（仅当用户有权限时）
if has_warning_permission:
item_dict['warningEnabled'] = bool(warning_enabled) if warning_enabled is not None else False
item_dict['warningYellow'] = float(warning_yellow) if warning_yellow is not None else None
item_dict['warningRed'] = float(warning_red) if warning_red is not None else None
# 计算预警状态
if warning_enabled and warning_red is not None:
invQty = item_dict['inventoryCount']
if invQty <= warning_red:
item_dict['warningStatus'] = 2 # 红色
elif warning_yellow is not None and invQty <= warning_yellow:
item_dict['warningStatus'] = 1 # 黄色
else:
item_dict['warningStatus'] = 0 # 正常
else:
item_dict['warningStatus'] = 0 # 未开启或未设置
items_list.append(item_dict)
return {"total": pagination.total, "items": items_list}