404 lines
16 KiB
Python
404 lines
16 KiB
Python
# 文件路径: app/api/v1/inbound/base.py
|
||
|
||
from flask import Blueprint, request, jsonify, send_file, g
|
||
from app.services.inbound.base_service import MaterialBaseService
|
||
from app.utils.decorators import login_required, permission_required, audit_log
|
||
import traceback
|
||
import datetime
|
||
import json
|
||
|
||
inbound_base_bp = Blueprint('stock_base', __name__)
|
||
|
||
|
||
# ==============================================================================
|
||
# 辅助函数：获取当前用户的完整权限列表（基于角色查询）
|
||
# ==============================================================================
|
||
def get_current_user_permissions():
|
||
"""
|
||
返回当前用户拥有的所有权限码列表（包括菜单和元素）
|
||
此函数根据角色查询数据库得到权限。
|
||
"""
|
||
from flask_jwt_extended import get_jwt
|
||
from app.services.auth_service import AuthService
|
||
claims = get_jwt()
|
||
user_role = claims.get('role')
|
||
if not user_role:
|
||
return []
|
||
# 超级管理员返回所有字段权限 (忽略大小写)
|
||
if user_role.upper() == 'SUPER_ADMIN':
|
||
# 返回通配符权限（供列表脱敏使用）以及所有具体权限（供导出脱敏使用）
|
||
return [
|
||
'material_list:*',
|
||
'material_list:id',
|
||
'material_list:companyName',
|
||
'material_list:name',
|
||
'material_list:commonName',
|
||
'material_list:category',
|
||
'material_list:type',
|
||
'material_list:spec',
|
||
'material_list:unit',
|
||
'material_list:inventoryCount',
|
||
'material_list:availableCount',
|
||
'material_list:files',
|
||
'material_list:isEnabled',
|
||
'material_list:operation'
|
||
]
|
||
perm_dict = AuthService.get_user_permissions(user_role)
|
||
# 合并菜单和元素权限
|
||
perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
|
||
return perms
|
||
|
||
|
||
def filter_item_by_permissions(item_dict, user_permissions):
|
||
"""
|
||
根据用户权限过滤 item 字典，无权限的字段值置为 None
|
||
"""
|
||
# 如果用户拥有通配符权限，则不过滤
|
||
if 'material_list:*' in user_permissions:
|
||
return item_dict
|
||
# 字段名到权限码的映射（与前端 permissionMap 保持一致）
|
||
field_to_perm = {
|
||
'id': 'material_list:id',
|
||
'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',
|
||
'generalManual': 'material_list:files',
|
||
'generalImage': 'material_list:files',
|
||
'isEnabled': 'material_list:isEnabled'
|
||
}
|
||
for field, perm_code in field_to_perm.items():
|
||
if field in item_dict and perm_code not in user_permissions:
|
||
item_dict[field] = None
|
||
return item_dict
|
||
|
||
|
||
# ==============================================================================
|
||
# 1. 搜索接口 (GET /api/v1/inbound/base/search)
|
||
# ==============================================================================
|
||
@inbound_base_bp.route('/search', methods=['GET'])
|
||
@permission_required('material_list')
|
||
def search_base():
|
||
try:
|
||
keyword = request.args.get('keyword', '')
|
||
data = MaterialBaseService.search_material(keyword)
|
||
# 字段级脱敏
|
||
user_permissions = get_current_user_permissions()
|
||
filtered_data = [filter_item_by_permissions(item, user_permissions) for item in data]
|
||
return jsonify({"code": 200, "msg": "success", "data": filtered_data})
|
||
except Exception as e:
|
||
traceback.print_exc()
|
||
return jsonify({"code": 500, "msg": str(e)}), 500
|
||
|
||
|
||
# ==============================================================================
|
||
# 2. 列表接口 (GET /api/v1/inbound/base/list)
|
||
# ==============================================================================
|
||
@inbound_base_bp.route('/list', methods=['GET'])
|
||
@permission_required('material_list')
|
||
def get_list():
|
||
try:
|
||
page = request.args.get('pageNum', 1, type=int)
|
||
limit = request.args.get('pageSize', 10, type=int)
|
||
|
||
# 解析高级筛选条件
|
||
advanced_filters_raw = request.args.get('advancedFilters', '[]')
|
||
try:
|
||
advanced_filters_list = json.loads(advanced_filters_raw)
|
||
except:
|
||
advanced_filters_list = []
|
||
|
||
# 构造筛选条件
|
||
filters = {
|
||
'keyword': request.args.get('keyword', ''),
|
||
'company': request.args.get('company', ''),
|
||
'category': request.args.get('category', ''),
|
||
'type': request.args.get('type', ''),
|
||
'isEnabled': request.args.get('isEnabled', None),
|
||
'orderByColumn': request.args.get('orderByColumn', ''),
|
||
'isAsc': request.args.get('isAsc', None),
|
||
'advancedFilters': advanced_filters_list,
|
||
'enableWarningSort': request.args.get('enableWarningSort', 'false').lower() == 'true',
|
||
'has_stock': request.args.get('has_stock', ''),
|
||
'searchField': request.args.get('searchField', 'all')
|
||
}
|
||
|
||
user_permissions = get_current_user_permissions()
|
||
result = MaterialBaseService.get_list(page, limit, filters, user_permissions)
|
||
# 字段级脱敏
|
||
user_permissions = get_current_user_permissions()
|
||
if result.get('items'):
|
||
result['items'] = [filter_item_by_permissions(item, user_permissions) for item in result['items']]
|
||
return jsonify({"code": 200, "msg": "success", "data": result})
|
||
except Exception as e:
|
||
traceback.print_exc()
|
||
return jsonify({"code": 500, "msg": str(e)}), 500
|
||
|
||
|
||
# ==============================================================================
|
||
# 2.1 选项接口 (GET /api/v1/inbound/base/options)
|
||
# ==============================================================================
|
||
@inbound_base_bp.route('/options', methods=['GET'])
|
||
@permission_required('material_list')
|
||
def get_options():
|
||
try:
|
||
data = MaterialBaseService.get_distinct_options()
|
||
return jsonify({"code": 200, "msg": "success", "data": data})
|
||
except Exception as e:
|
||
traceback.print_exc()
|
||
return jsonify({"code": 500, "msg": str(e)}), 500
|
||
|
||
|
||
# ==============================================================================
|
||
# 2.2 导出接口 (GET /api/v1/inbound/base/export)
|
||
# ==============================================================================
|
||
@inbound_base_bp.route('/export', methods=['GET'])
|
||
@permission_required('material_list')
|
||
def export_data():
|
||
try:
|
||
# 获取筛选条件
|
||
filters = {
|
||
'keyword': request.args.get('keyword', ''),
|
||
'company': request.args.get('company', ''),
|
||
'category': request.args.get('category', ''),
|
||
'type': request.args.get('type', ''),
|
||
'isEnabled': request.args.get('isEnabled', None)
|
||
}
|
||
|
||
# 获取当前用户权限
|
||
user_permissions = get_current_user_permissions()
|
||
|
||
# 生成 Excel 文件流（传入用户权限进行脱敏）
|
||
file_stream = MaterialBaseService.export_excel(filters, user_permissions)
|
||
|
||
# 生成文件名：库存统计+年月日+时分秒 (北京时间 UTC+8)
|
||
# 简单处理：UTC时间 + 8小时
|
||
beijing_time = datetime.datetime.utcnow() + datetime.timedelta(hours=8)
|
||
filename = f"库存统计_{beijing_time.strftime('%Y%m%d_%H%M%S')}.xlsx"
|
||
|
||
# 发送文件
|
||
# 注意：download_name 仅在较新 Flask 版本有效，旧版本可能需要手动 header，
|
||
# 但通常浏览器下载名由前端 Blob 处理或 Content-Disposition 决定。
|
||
return send_file(
|
||
file_stream,
|
||
mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
||
as_attachment=True,
|
||
download_name=filename
|
||
)
|
||
except Exception as e:
|
||
traceback.print_exc()
|
||
return jsonify({"code": 500, "msg": f"导出失败: {str(e)}"}), 500
|
||
|
||
|
||
# ==============================================================================
|
||
# 3. 新增接口 (POST /api/v1/inbound/base/)
|
||
# ==============================================================================
|
||
@inbound_base_bp.route('/', methods=['POST'])
|
||
@permission_required('material_list:operation')
|
||
@audit_log(
|
||
module='基础信息管理',
|
||
action='新增',
|
||
get_target_name_fn=lambda: request.get_json().get('name') if request.get_json() else None
|
||
)
|
||
def create():
|
||
try:
|
||
data = request.get_json()
|
||
if not data:
|
||
return jsonify({"code": 400, "msg": "No data provided"}), 400
|
||
|
||
# 获取当前用户权限
|
||
user_permissions = get_current_user_permissions()
|
||
# 字段到权限码的映射（与 filter_item_by_permissions 一致）
|
||
field_to_perm = {
|
||
'id': 'material_list:id',
|
||
'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',
|
||
'generalManual': 'material_list:files',
|
||
'generalImage': 'material_list:files',
|
||
'isEnabled': 'material_list:isEnabled'
|
||
}
|
||
# 过滤用户没有权限的字段
|
||
filtered_data = {}
|
||
# 如果拥有通配符权限，则不过滤
|
||
if 'material_list:*' in user_permissions:
|
||
filtered_data = data
|
||
else:
|
||
for key, value in data.items():
|
||
if key in field_to_perm:
|
||
perm_code = field_to_perm[key]
|
||
if perm_code in user_permissions:
|
||
filtered_data[key] = value
|
||
# 没有权限则跳过，不包含在 filtered_data 中
|
||
else:
|
||
# 不在映射中的字段，默认允许（例如 visibilityLevel）
|
||
filtered_data[key] = value
|
||
|
||
MaterialBaseService.create_material(filtered_data)
|
||
return jsonify({"code": 200, "msg": "新增成功"})
|
||
except ValueError as e:
|
||
# 捕获业务逻辑验证错误 (如名称为空)
|
||
return jsonify({"code": 400, "msg": str(e)}), 400
|
||
except Exception as e:
|
||
# 捕获系统错误
|
||
traceback.print_exc()
|
||
return jsonify({"code": 500, "msg": f"系统错误: {str(e)}"}), 500
|
||
|
||
|
||
# ==============================================================================
|
||
# 4. 修改接口 (PUT /api/v1/inbound/base/<id>)
|
||
# ==============================================================================
|
||
@inbound_base_bp.route('/<int:id>', methods=['PUT'])
|
||
@permission_required('material_list:operation')
|
||
@audit_log(
|
||
module='基础信息管理',
|
||
action='修改',
|
||
get_target_id_fn=lambda: request.view_args.get('id'),
|
||
get_target_name_fn=lambda: request.get_json().get('name') if request.get_json() else None
|
||
)
|
||
def update(id):
|
||
try:
|
||
data = request.get_json()
|
||
# 获取当前用户权限
|
||
user_permissions = get_current_user_permissions()
|
||
# 字段到权限码的映射（与 filter_item_by_permissions 一致）
|
||
field_to_perm = {
|
||
'id': 'material_list:id',
|
||
'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',
|
||
'generalManual': 'material_list:files',
|
||
'generalImage': 'material_list:files',
|
||
'isEnabled': 'material_list:isEnabled'
|
||
}
|
||
# 过滤用户没有权限的字段
|
||
filtered_data = {}
|
||
# 如果拥有通配符权限，则不过滤
|
||
if 'material_list:*' in user_permissions:
|
||
filtered_data = data
|
||
else:
|
||
for key, value in data.items():
|
||
if key in field_to_perm:
|
||
perm_code = field_to_perm[key]
|
||
if perm_code in user_permissions:
|
||
filtered_data[key] = value
|
||
# 没有权限则跳过，不包含在 filtered_data 中
|
||
else:
|
||
# 不在映射中的字段，默认允许（例如 visibilityLevel）
|
||
filtered_data[key] = value
|
||
# 使用过滤后的数据调用服务
|
||
MaterialBaseService.update_material(id, filtered_data)
|
||
return jsonify({"code": 200, "msg": "修改成功"})
|
||
except Exception as e:
|
||
traceback.print_exc()
|
||
return jsonify({"code": 500, "msg": str(e)}), 500
|
||
|
||
|
||
# ==============================================================================
|
||
# 5. 删除接口 (DELETE /api/v1/inbound/base/<id>)
|
||
# ==============================================================================
|
||
@inbound_base_bp.route('/<int:id>', methods=['DELETE'])
|
||
@permission_required('material_list:operation')
|
||
@audit_log(
|
||
module='基础信息管理',
|
||
action='删除',
|
||
get_target_id_fn=lambda: request.view_args.get('id')
|
||
)
|
||
def delete(id):
|
||
try:
|
||
MaterialBaseService.delete_material(id)
|
||
return jsonify({"code": 200, "msg": "删除成功"})
|
||
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
|