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

fix(outbound+trans): 修复POST接口错误数据清洗导致的sku/quantity字段被清除Bug，并新增出库审批工作流全链路

Browse Source
This commit is contained in:
DXC
2026-04-28 16:02:34 +08:00
parent 97e7618bf3
commit 62c0e3738e
12 changed files with 1248 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
inventory-backend/app/api/v1/auth.py
View File

@ -318,6 +318,41 @@ def get_my_permissions():
return jsonify({'msg': f'获取权限失败: {str(e)}'}), 500
# ==============================================================================
# 获取可指定审批人列表（SUPERVISOR / SUPER_ADMIN 且 status=active）
# ==============================================================================
@auth_bp.route('/users/approvers', methods=['GET'])
@jwt_required()
def get_approvers():
"""
查询角色为 SUPER_ADMIN 或 SUPERVISOR 且状态为活跃的用户列表
返回: [{id, username, email, role}]
"""
try:
from app.models.system import SysUser
users = SysUser.query.filter(
SysUser.role.in_(['SUPER_ADMIN', 'SUPERVISOR']),
SysUser.status == 'active'
).all()
return jsonify({
'msg': '获取成功',
'data': [
{
'id': u.id,
'username': u.username,
'email': u.email or '',
'role': u.role
} for u in users
]
}), 200
except Exception as e:
current_app.logger.error(f"Get Approvers Failed: {str(e)}")
return jsonify({'msg': f'获取审批人列表失败: {str(e)}'}), 500
# ==============================================================================
# 获取当前用户个人资料（自我查看）
# ==============================================================================

279
inventory-backend/app/api/v1/outbound.py
View File

@ -148,44 +148,6 @@ def create_outbound():
if not data.get('consumer_name') or not data.get('signature_path'):
return jsonify({'code': 400, 'msg': '领用人及签名信息缺失'}), 400
# 数据清洗：移除用户没有权限的字段
user_permissions = get_current_user_permissions()
# 超级管理员不过滤
if 'outbound_list:*' not in user_permissions:
# 字段名到权限码的映射（与前端 permissionMap 保持一致）
field_to_perm = {
'outbound_no': 'outbound_list:outbound_no',
'outbound_time': 'outbound_list:outbound_time',
'outbound_type': 'outbound_list:outbound_type',
'total_amount': 'outbound_list:total_amount',
'consumer_name': 'outbound_list:consumer_name',
'operator_name': 'outbound_list:operator_name',
'remark': 'outbound_list:remark',
'signature_path': 'outbound_list:signature_path',
# 明细字段
'sku': 'outbound_list:sku',
'name': 'outbound_list:name',
'material_type': 'outbound_list:material_type',
'category': 'outbound_list:category',
'spec_model': 'outbound_list:spec_model',
'quantity': 'outbound_list:quantity',
'unit_price': 'outbound_list:unit_price',
'price': 'outbound_list:unit_price', # 兼容 price 字段
'subtotal': 'outbound_list:subtotal',
}
# 清洗顶层字段
for field in list(data.keys()):
perm_code = field_to_perm.get(field)
if perm_code and perm_code not in user_permissions:
data.pop(field, None)
# 清洗 items 中的每个商品字段
if 'items' in data and isinstance(data['items'], list):
for item in data['items']:
for field in list(item.keys()):
perm_code = field_to_perm.get(field)
if perm_code and perm_code not in user_permissions:
item.pop(field, None)
try:
# ★ [修改] 调用批量创建服务
outbound_no = OutboundService.create_outbound_batch(data, operator_name=final_operator)
@ -233,3 +195,244 @@ def get_outbound_list():
except Exception as e:
traceback.print_exc()
return jsonify({'code': 500, 'msg': str(e)}), 500
# ==============================================================================
# 出库审批相关接口
# ==============================================================================
from app.services.outbound_service import OutboundApprovalService
def get_current_user_id():
"""获取当前用户ID"""
from app.models.system import SysUser
identity = get_jwt_identity()
if not identity:
return None
# JWT identity 是数据库主键整数，直接用 .get() 查询
user = SysUser.query.get(identity)
return user.id if user else None
def get_current_user_info():
"""获取当前用户信息和角色"""
from app.models.system import SysUser
identity = get_jwt_identity()
if not identity:
return None, None
# JWT identity 是数据库主键整数，直接用 .get() 查询
user = SysUser.query.get(identity)
return user.id if user else None, user.role if user else None
# --------------------------------------------------------
# 4. 创建出库审批单
# POST /api/v1/outbound/request
# --------------------------------------------------------
@outbound_bp.route('/request', methods=['POST'])
@jwt_required()
def create_outbound_request():
"""
创建出库审批单（申请阶段，用户只需提交宏观物料信息，无需关联具体库存记录）
请求体示例:
{
"items": [
{
"name": "物料A", // 物料名称 (必填)
"spec_model": "规格1", // 规格型号 (必填)
"quantity": 10, // 计划出库数量 (必填)
"warehouse_location": "A区-01-01", // 库位 (可选)
"remark": "备注信息" // 物品备注 (可选)
}
],
"allowed_approvers": [
{"type": "role", "value": "SUPERVISOR"},
{"type": "role", "value": "SUPER_ADMIN"}
],
"remark": "紧急出库申请"
}
"""
try:
user_id, user_role = get_current_user_info()
if not user_id:
return jsonify({'code': 401, 'msg': '用户未登录'}), 401
data = request.get_json()
if not data:
return jsonify({'code': 400, 'msg': '无有效数据'}), 400
items = data.get('items', [])
if not items:
return jsonify({'code': 400, 'msg': '出库物品列表不能为空'}), 400
# ★ 申请阶段仅校验宏观字段：名称、规格、数量
required_fields = ['name', 'spec_model', 'quantity']
for idx, item in enumerate(items):
missing = [f for f in required_fields if f not in item or item.get(f) is None or str(item.get(f)).strip() == '']
if missing:
return jsonify({
'code': 400,
'msg': f'第{idx + 1}条物品缺少必填字段: {", ".join(missing)}。'
f'必须包含: name(名称), spec_model(规格), quantity(数量)'
}), 400
try:
qty = float(item.get('quantity', 0))
if qty <= 0:
return jsonify({'code': 400, 'msg': f'第{idx + 1}条物品的出库数量必须大于0'}), 400
except (TypeError, ValueError):
return jsonify({'code': 400, 'msg': f'第{idx + 1}条物品的 quantity 格式无效'}), 400
# ★ 指定审批人：前端传 approver_id 则精准通知，否则用默认角色规则
approver_id = data.get('approver_id')
_default_approvers = [
{"type": "role", "value": "SUPERVISOR"},
{"type": "role", "value": "SUPER_ADMIN"}
]
allowed_approvers = data.get('allowed_approvers') or _default_approvers
# 创建审批单（直接存储前端传来的宏观信息快照，不查询库存）
approval = OutboundApprovalService.create_request(
applicant_id=user_id,
items=items,
allowed_approvers=allowed_approvers,
remark=data.get('remark'),
approver_id=approver_id
)
return jsonify({
'code': 200,
'msg': '审批单创建成功',
'data': approval.to_dict()
}), 200
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
# --------------------------------------------------------
# 5. 审批出库申请
# PATCH /api/v1/outbound/request/<id>/approve
# --------------------------------------------------------
@outbound_bp.route('/request/<int:request_id>/approve', methods=['PATCH'])
@jwt_required()
def approve_outbound_request(request_id):
"""
审批出库申请
请求体示例:
{
"action": "approve", // "approve" 通过, "reject" 驳回
"reject_reason": "库存不足" // 仅在驳回时需要
}
"""
try:
user_id, user_role = get_current_user_info()
if not user_id:
return jsonify({'code': 401, 'msg': '用户未登录'}), 401
data = request.get_json() or {}
action = data.get('action', 'approve')
reject_reason = data.get('reject_reason')
if action not in ('approve', 'reject'):
return jsonify({'code': 400, 'msg': '无效的审批操作，仅支持 approve 或 reject'}), 400
if action == 'reject' and not reject_reason:
return jsonify({'code': 400, 'msg': '驳回时必须提供原因'}), 400
success, message, approval = OutboundApprovalService.approve(
request_id=request_id,
user_id=user_id,
user_role=user_role,
action=action,
reject_reason=reject_reason
)
if not success:
return jsonify({'code': 400, 'msg': message}), 400
return jsonify({
'code': 200,
'msg': message,
'data': approval.to_dict() if approval else None
}), 200
except Exception as e:
traceback.print_exc()
return jsonify({'code': 500, 'msg': f'服务器内部错误: {str(e)}'}), 500
# --------------------------------------------------------
# 6. 获取审批单列表
# GET /api/v1/outbound/request
# --------------------------------------------------------
@outbound_bp.route('/request', methods=['GET'])
@jwt_required()
def get_outbound_request_list():
"""
获取出库审批单列表
Query参数:
- page: 页码 (默认1)
- limit: 每页数量 (默认10)
- applicant_id: 按申请人筛选 (可选)
- status: 按状态筛选 (0待审/1通过/2驳回/3完成, 可选)
"""
try:
page = int(request.args.get('page', 1))
limit = int(request.args.get('limit', 10))
applicant_id = request.args.get('applicant_id')
if applicant_id:
applicant_id = int(applicant_id)
status = request.args.get('status')
if status is not None:
status = int(status)
result = OutboundApprovalService.get_request_list(
page=page,
per_page=limit,
applicant_id=applicant_id,
status=status
)
return jsonify({
'code': 200,
'msg': '获取成功',
'data': result
}), 200
except Exception as e:
traceback.print_exc()
return jsonify({'code': 500, 'msg': str(e)}), 500
# --------------------------------------------------------
# 7. 获取单个审批单详情
# GET /api/v1/outbound/request/<id>
# --------------------------------------------------------
@outbound_bp.route('/request/<int:request_id>', methods=['GET'])
@jwt_required()
def get_outbound_request_detail(request_id):
"""获取出库审批单详情"""
try:
approval = OutboundApprovalService.get_request_by_id(request_id)
if not approval:
return jsonify({'code': 404, 'msg': '审批单不存在'}), 404
return jsonify({
'code': 200,
'msg': '获取成功',
'data': approval.to_dict()
}), 200
except Exception as e:
traceback.print_exc()
return jsonify({'code': 500, 'msg': str(e)}), 500

40
inventory-backend/app/api/v1/transactions.py
View File

@ -66,26 +66,6 @@ def filter_item_by_permissions(item_dict, user_permissions, prefix='op_records')
)
def create_borrow():
data = request.get_json()
# 数据清洗：移除用户没有权限的字段
user_permissions = get_current_user_permissions()
# 超级管理员不过滤
if '*' not in user_permissions:
field_to_perm = {
'borrow_no': 'op_records:borrow_no',
'borrower_name': 'op_records:borrower_name',
'sku': 'op_records:sku',
'borrow_time': 'op_records:borrow_time',
'return_time': 'op_records:return_time',
'status': 'op_records:status',
'expected_return_time': 'op_records:expected_return_time',
'return_location': 'op_records:return_location',
'borrow_signature': 'op_records:borrow_signature',
'return_signature': 'op_records:return_signature',
}
for field in list(data.keys()):
perm_code = field_to_perm.get(field)
if perm_code and perm_code not in user_permissions:
data.pop(field, None)
try:
no = TransService.create_borrow(data)
return jsonify({'code': 200, 'msg': '借用成功', 'data': {'borrow_no': no}})
@ -120,26 +100,6 @@ def scan_borrowed_item():
)
def submit_return():
data = request.get_json()
# 数据清洗：移除用户没有权限的字段
user_permissions = get_current_user_permissions()
# 超级管理员不过滤
if '*' not in user_permissions:
field_to_perm = {
'borrow_no': 'op_records:borrow_no',
'borrower_name': 'op_records:borrower_name',
'sku': 'op_records:sku',
'borrow_time': 'op_records:borrow_time',
'return_time': 'op_records:return_time',
'status': 'op_records:status',
'expected_return_time': 'op_records:expected_return_time',
'return_location': 'op_records:return_location',
'borrow_signature': 'op_records:borrow_signature',
'return_signature': 'op_records:return_signature',
}
for field in list(data.keys()):
perm_code = field_to_perm.get(field)
if perm_code and perm_code not in user_permissions:
data.pop(field, None)
user = get_jwt_identity() # 库管
try:
TransService.process_return(data, operator_name=user)

105
inventory-backend/app/models/outbound.py
View File

@ -1,5 +1,110 @@
from app.extensions import db, beijing_time
from app.models.system import SysUser
from datetime import datetime
import json
class OutboundApproval(db.Model):
"""
出库审批单模型
用于管理出库申请的多级审批流程
"""
__tablename__ = 'outbound_approval'
id = db.Column(db.Integer, primary_key=True)
# 审批单号
request_no = db.Column(db.String(100), unique=True, nullable=False, index=True)
# 申请人ID
applicant_id = db.Column(db.Integer, nullable=False, index=True)
# 申请说明
remark = db.Column(db.Text)
# 状态: 0-待审批, 1-已通过, 2-已驳回, 3-已完成(已出库)
status = db.Column(db.Integer, default=0, nullable=False)
# 允许审批的人员列表 (JSON格式: [{"type": "role", "value": "admin"}, {"type": "user", "value": "123"}])
allowed_approvers = db.Column(db.Text)
# 实际审批人ID (多人审批时记录第一个通过的)
actual_approver_id = db.Column(db.Integer, index=True)
# 审批时间
approved_at = db.Column(db.DateTime)
# 驳回原因
reject_reason = db.Column(db.Text)
# 明细快照 (存储出库物品的名称、规格、库位、数量等信息，无SKU字段)
items_json = db.Column(db.Text)
# 创建时间和更新时间
created_at = db.Column(db.DateTime, default=beijing_time, nullable=False)
updated_at = db.Column(db.DateTime, default=beijing_time, onupdate=beijing_time, nullable=False)
def _safe_parse_json(self, value):
"""
安全解析 JSON 字段：
- 如果 value 已是 list/dict，直接返回
- 如果是 str，尝试 json.loads()
- 解析失败或为 None/空，均返回 []
"""
if value is None:
return []
if isinstance(value, (list, dict)):
return value
if isinstance(value, str):
val = value.strip()
if not val:
return []
try:
parsed = json.loads(val)
return parsed if isinstance(parsed, list) else []
except (json.JSONDecodeError, TypeError, ValueError):
return []
return []
def get_items(self):
"""解析 items_json，返回物品列表"""
return self._safe_parse_json(self.items_json)
def set_items(self, items):
"""设置 items_json"""
self.items_json = json.dumps(items, ensure_ascii=False) if items else '[]'
def get_allowed_approvers(self):
"""解析 allowed_approvers，返回审批人列表"""
return self._safe_parse_json(self.allowed_approvers)
def set_allowed_approvers(self, approvers):
"""设置 allowed_approvers"""
self.allowed_approvers = json.dumps(approvers, ensure_ascii=False) if approvers else '[]'
def to_dict(self):
return {
'id': self.id,
'request_no': self.request_no,
'applicant_id': self.applicant_id,
'applicant_name': self._get_user_name(self.applicant_id),
'remark': self.remark,
'status': self.status,
'status_text': ['待审批', '已通过', '已驳回', '已完成'][self.status] if self.status in [0, 1, 2, 3] else '未知',
'allowed_approvers': self.get_allowed_approvers(),
'actual_approver_id': self.actual_approver_id,
'approver_name': self._get_user_name(self.actual_approver_id) if self.actual_approver_id else None,
'approved_at': self.approved_at.strftime('%Y-%m-%d %H:%M:%S') if self.approved_at else None,
'reject_reason': self.reject_reason,
'items': self.get_items(),
'created_at': self.created_at.strftime('%Y-%m-%d %H:%M:%S') if self.created_at else None,
'updated_at': self.updated_at.strftime('%Y-%m-%d %H:%M:%S') if self.updated_at else None,
}
def _get_user_name(self, user_id):
"""根据用户ID获取用户名"""
if not user_id:
return ""
from app.models.system import SysUser
try:
# ★ 必须用 .get() 按主键 ID 查询，千万不能用 username=user_id 去查
user = SysUser.query.get(user_id)
return user.username if user else f"未知用户({user_id})"
except Exception as e:
return f"用户({user_id})"
class TransOutbound(db.Model):

362
inventory-backend/app/services/outbound_service.py
View File

@ -2,7 +2,7 @@ import uuid # .material -> .base refactor checked
from datetime import datetime, timezone, timedelta
from sqlalchemy import or_, func, desc, and_
from app.extensions import db
from app.models.outbound import TransOutbound
from app.models.outbound import TransOutbound, OutboundApproval
# 引入所有库存模型以进行查询
from app.models.inbound.buy import StockBuy
@ -12,6 +12,8 @@ from app.models.inbound.product import StockProduct
from app.models.base import MaterialBase
# 引入维修单表
from app.models.transaction import TransRepair
# 引入系统用户表
from app.models.system import SysUser
class OutboundService:
@ -169,6 +171,22 @@ class OutboundService:
beijing_tz = timezone(timedelta(hours=8))
current_time = datetime.now(beijing_tz).replace(tzinfo=None)
# ★ 审批单相关逻辑
request_id = data.get('request_id')
approval = None
if request_id:
# 根据 request_id 查询审批单
approval = OutboundApproval.query.get(request_id)
if not approval:
raise ValueError(f"关联的审批单不存在 (ID: {request_id})")
if approval.status != 1:
status_map = {0: '待审批', 1: '已通过', 2: '已驳回', 3: '已完成'}
current_status = status_map.get(approval.status, str(approval.status))
raise ValueError(
f"关联的审批单状态不允许出库 (当前状态: {current_status})，"
f"仅已通过的审批单方可执行出库"
)
model_map = {
'stock_buy': StockBuy,
'stock_semi': StockSemi,
@ -190,11 +208,11 @@ class OutboundService:
repair = TransRepair.query.with_for_update().get(stock_id)
if not repair:
raise ValueError(f"维修单不存在 (ID: {stock_id})")
# 更新维修单状态为已出库
repair.repair_status = '已出库'
repair.shipping_date = current_time
# 创建出库记录
new_record = TransOutbound(
sku=item.get('sku'),
@ -235,6 +253,11 @@ class OutboundService:
)
db.session.add(new_record)
# ★ 如果关联了审批单，出库成功后更新审批单状态为"已完成"
if approval:
approval.status = 3 # 3-已完成
# updated_at 会在 commit 时由 SQLAlchemy 自动更新
db.session.commit()
return outbound_no
@ -525,3 +548,336 @@ class OutboundService:
'pages': pagination.pages,
'current_page': page
}
class OutboundApprovalService:
"""出库审批服务"""
@staticmethod
def generate_request_no():
"""
生成审批单号: APR-OUT-yyyyMMdd-HHmm-当日流水(4位)
"""
beijing_tz = timezone(timedelta(hours=8))
now = datetime.now(beijing_tz)
date_str = now.strftime('%Y%m%d')
time_str = now.strftime('%H%M')
prefix = f"APR-OUT-{date_str}-"
from app.models.outbound import OutboundApproval
latest = db.session.query(OutboundApproval.request_no).filter(
OutboundApproval.request_no.like(f"{prefix}%")
).order_by(OutboundApproval.id.desc()).first()
if latest:
last_seq = int(latest[0].split('-')[-1])
sequence = last_seq + 1
else:
sequence = 1
return f"APR-OUT-{date_str}-{time_str}-{sequence:04d}"
@staticmethod
def create_request(applicant_id, items, allowed_approvers, remark=None, approver_id=None):
"""
创建出库审批单（申请阶段，直接存储前端传来的物料信息快照，不关联具体库存记录）
Args:
applicant_id: 申请人ID
items: 出库物品明细列表，每个物品应包含:
- name: 物料名称 (必填)
- spec_model: 规格型号 (必填)
- quantity: 计划出库数量 (必填)
- warehouse_location: 库位 (可选)
- remark: 物品备注 (可选)
allowed_approvers: 允许审批的人员/角色列表
approver_id: 指定审批人ID（可选，传则覆盖 allowed_approvers）
remark: 申请说明
Returns:
OutboundApproval 实例
Raises:
ValueError: 当 items 为空或缺少必填字段时抛出
"""
from app.models.outbound import OutboundApproval
# 校验 items 非空
if not items:
raise ValueError("出库物品列表不能为空")
# 校验每个物品的宏观字段 (name, spec_model, quantity)
required_fields = ['name', 'spec_model', 'quantity']
for idx, item in enumerate(items):
missing_fields = [f for f in required_fields if f not in item or str(item.get(f) or '').strip() == '']
if missing_fields:
raise ValueError(
f"第 {idx + 1} 条物品缺少必填字段: {', '.join(missing_fields)}。"
f"必须包含: name, spec_model, quantity"
)
try:
qty = float(item.get('quantity', 0))
if qty <= 0:
raise ValueError(f"第 {idx + 1} 条物品的出库数量必须大于0")
except (TypeError, ValueError) as e:
raise ValueError(f"第 {idx + 1} 条物品的 quantity 格式无效: {str(e)}")
# ★ 校验 allowed_approvers 非空
if not allowed_approvers:
raise ValueError("必须指定至少一位审批人")
# ★ 指定审批人模式：approver_id 覆盖 allowed_approvers
if approver_id:
allowed_approvers = [{"type": "user", "value": int(approver_id)}]
request_no = OutboundApprovalService.generate_request_no()
approval = OutboundApproval(
request_no=request_no,
applicant_id=applicant_id,
remark=remark,
status=0, # 待审批
)
# 直接存储前端传来的物料信息快照，不查询/不关联具体库存记录
approval.set_items(items)
approval.set_allowed_approvers(allowed_approvers)
db.session.add(approval)
db.session.commit()
# ★ 创建成功后，发送邮件通知审批人（精确通知 approver_id 对应的邮箱）
OutboundApprovalService._notify_new_request(approval, applicant_id, approver_id=approver_id)
return approval
@staticmethod
def _get_emails_by_identifiers(applicant_id=None, role_codes=None):
"""
根据用户ID或角色列表查询邮箱地址
Args:
applicant_id: 用户ID (按 SysUser.id 查找)
role_codes: 角色代码列表，如 ['ADMIN', 'WAREHOUSE_ADMIN']
Returns:
去重后的邮箱地址列表
"""
emails = []
if applicant_id:
user = SysUser.query.get(int(applicant_id))
if user and user.email:
emails.append(user.email)
if role_codes:
for code in role_codes:
users = SysUser.query.filter_by(role=code).all()
for u in users:
if u.email:
emails.append(u.email)
return list(set(emails))
@staticmethod
def _notify_new_request(approval, applicant_id, approver_id=None):
"""发送新申请通知邮件给审批人（静默处理，不阻断主流程）"""
try:
from flask import current_app
from app.utils.email_service import send_new_request_notify
emails = []
if approver_id:
# ★ 精准通知模式：直接查询指定审批人
user = SysUser.query.get(int(approver_id))
if user and user.email:
emails.append(user.email)
else:
# 兜底：按角色查询
approvers = approval.get_allowed_approvers()
role_codes = []
for a in approvers:
if a.get('type') == 'role':
role_codes.append(a.get('value', ''))
emails = OutboundApprovalService._get_emails_by_identifiers(role_codes=role_codes)
if not emails:
current_app.logger.info(f"[Email] 审批单 {approval.request_no} 无审批人邮箱，跳过通知")
return
# 获取申请人姓名
applicant_name = ''
if applicant_id:
u = SysUser.query.get(applicant_id)
if u:
# username 格式为 "姓名/账号"，取姓名部分
applicant_name = str(u.username).split('/')[0] if '/' in u.username else (u.username or str(applicant_id))
send_new_request_notify(
to_emails=emails,
request_no=approval.request_no,
applicant_name=applicant_name,
remark=approval.remark or ''
)
except Exception as e:
# ★ 捕获所有异常，确保邮件发送失败不阻断主流程
try:
from flask import current_app
current_app.logger.error(f"[Email] 发送新申请通知邮件失败: {e}")
except RuntimeError:
# 如果不在 Flask 应用上下文内，降级为标准日志
logging.getLogger(__name__).error(f"[Email] 发送新申请通知邮件失败: {e}")
@staticmethod
def can_approve(approval, user_id, user_role):
"""
检查用户是否有权限审批
Args:
approval: OutboundApproval 实例
user_id: 用户ID
user_role: 用户角色
Returns:
bool, 是否有权限
"""
approvers = approval.get_allowed_approvers()
# 超级管理员可以直接审批
if user_role and user_role.upper() == 'SUPER_ADMIN':
return True
for approver in approvers:
approver_type = approver.get('type', '')
approver_value = approver.get('value', '')
if approver_type == 'user' and str(approver_value) == str(user_id):
return True
if approver_type == 'role' and approver_value == user_role:
return True
return False
@staticmethod
def approve(request_id, user_id, user_role, action='approve', reject_reason=None):
"""
执行审批操作
Args:
request_id: 审批单ID
user_id: 审批人ID
user_role: 审批人角色
action: 'approve' 通过, 'reject' 驳回
reject_reason: 驳回原因
Returns:
(success: bool, message: str, approval: OutboundApproval or None)
"""
from app.models.outbound import OutboundApproval
beijing_tz = timezone(timedelta(hours=8))
current_time = datetime.now(beijing_tz).replace(tzinfo=None)
approval = OutboundApproval.query.get(request_id)
if not approval:
return False, "审批单不存在", None
if approval.status != 0:
return False, f"审批单状态已更新，无法重复审批 (当前状态: {approval.status})", None
if not OutboundApprovalService.can_approve(approval, user_id, user_role):
return False, "您没有审批此单的权限", None
try:
if action == 'approve':
approval.status = 1 # 已通过
approval.actual_approver_id = user_id
approval.approved_at = current_time
elif action == 'reject':
approval.status = 2 # 已驳回
approval.reject_reason = reject_reason
else:
return False, "无效的审批操作", None
db.session.commit()
# ★ 审批成功后，发送邮件通知仓库管理员
OutboundApprovalService._notify_approval_result(approval, user_id, action)
return True, "审批成功", approval
except Exception as e:
db.session.rollback()
return False, f"审批失败: {str(e)}", None
@staticmethod
def _notify_approval_result(approval, approver_id, action):
"""发送审批结果通知邮件（静默处理，不阻断主流程）"""
try:
from app.utils.email_service import send_approval_result_notify
# 仓库管理员角色代码
warehouse_role_codes = ['WAREHOUSE_MGR', 'OUTBOUND']
# 查询库管邮箱 + 申请人本人邮箱
emails = OutboundApprovalService._get_emails_by_identifiers(
applicant_id=approval.applicant_id,
role_codes=warehouse_role_codes
)
if not emails:
return
send_approval_result_notify(
to_emails=emails,
request_no=approval.request_no,
is_passed=(action == 'approve'),
reject_reason=approval.reject_reason or ''
)
except Exception as e:
logging.getLogger(__name__).warning(f"[Email] 发送审批结果通知邮件失败: {e}")
@staticmethod
def get_request_list(page=1, per_page=10, applicant_id=None, status=None):
"""
获取审批单列表
Args:
page: 页码
per_page: 每页数量
applicant_id: 按申请人筛选 (可选)
status: 按状态筛选 (可选)
Returns:
分页结果
"""
from app.models.outbound import OutboundApproval
from sqlalchemy import desc
query = OutboundApproval.query
if applicant_id:
query = query.filter(OutboundApproval.applicant_id == applicant_id)
if status is not None:
query = query.filter(OutboundApproval.status == status)
query = query.order_by(desc(OutboundApproval.created_at))
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
return {
'items': [item.to_dict() for item in pagination.items],
'total': pagination.total,
'pages': pagination.pages,
'current_page': page
}
@staticmethod
def get_request_by_id(request_id):
"""根据ID获取审批单"""
from app.models.outbound import OutboundApproval
return OutboundApproval.query.get(request_id)

1
inventory-backend/app/services/permission_service.py
View File

@ -435,6 +435,7 @@ class PermissionService:
('outbound_selection', '出库选单', '/outbound/selection', 'outbound_mgmt', 1),
('outbound_create', '扫码出库', '/outbound/create', 'outbound_mgmt', 2),
('outbound_list', '出库记录', '/outbound/index', 'outbound_mgmt', 3),
('outbound_approval', '出库审批', '/outbound/approval', 'outbound_mgmt', 4),
# BOM管理子菜单
('bom_manage', 'BOM配方管理', '/bom/manage', 'bom_mgmt', 1),

22
inventory-backend/config.py
View File

@ -48,4 +48,24 @@ class Config:
# =========================================================
# 5. Redis 配置 (用于单设备登录互踢)
# =========================================================
REDIS_URL = os.getenv('REDIS_URL', 'redis://localhost:6379/0')
REDIS_URL = os.getenv('REDIS_URL', 'redis://localhost:6379/0')
# =========================================================
# 6. 邮件配置
# =========================================================
# 发件人邮箱（阿里企业邮箱）
MAIL_USERNAME = os.getenv('MAIL_USERNAME', 'wms@iris-rs.cn')
# 发件人邮箱密码 / 授权码
MAIL_PASSWORD = os.getenv('MAIL_PASSWORD', 'Q7nYyyESWlaThKjx')
# SMTP 服务器地址（阿里企业邮发信服务器）
MAIL_SERVER = os.getenv('MAIL_SERVER', 'smtp.mxhichina.com')
# SMTP 端口（阿里邮箱使用 SSL 465）
MAIL_PORT = int(os.getenv('MAIL_PORT', 465))
# 是否启用 TLS (587 端口通常需要)
MAIL_USE_TLS = os.getenv('MAIL_USE_TLS', 'false').lower() in ('true', '1', 'yes')
# 是否启用 SSL (465 端口通常需要，阿里邮箱必须启用 SSL)
MAIL_USE_SSL = os.getenv('MAIL_USE_SSL', 'true').lower() in ('true', '1', 'yes')
# 默认发件人（显示名称 <邮箱地址>，阿里要求 MAIL_USERNAME 与此处邮箱地址完全一致）
MAIL_DEFAULT_SENDER = os.getenv('MAIL_DEFAULT_SENDER', 'WMS系统 <wms@iris-rs.cn>')
# 是否启用邮件发送功能（开发环境可设为 false 禁用）
MAIL_ENABLED = os.getenv('MAIL_ENABLED', 'true').lower() in ('true', '1', 'yes')