ffbb2199f0
- ScrapApproval 模型 + ScrapApprovalService：提交/列表/审批/执行(锁库存扣 available_quantity 写 TransScrap) - 新路由 POST /scrap/request、/request/check-approval、GET /request、PATCH /request/<id>/approve、POST /request/<id>/execute；权限码 scrap_apply/scrap_approval/scrap_execute（无角色硬编码） - 旧 /scrap 直接报废入口保持不变 - /inbound/stock/list 每项返回 source_table，供按单流程精准选库存
91 lines
3.4 KiB
Python
91 lines
3.4 KiB
Python
from app.extensions import db
|
||
import json
|
||
|
||
|
||
def _beijing_now():
|
||
from datetime import datetime, timezone, timedelta
|
||
return datetime.now(timezone(timedelta(hours=8)))
|
||
|
||
|
||
class ScrapApproval(db.Model):
|
||
"""
|
||
报废申请/审批单（镜像 BorrowApproval / OutboundApproval）
|
||
|
||
与出/借库的关键差异：items_json 里每条明细必须携带
|
||
source_table + stock_id（精准定位到某条实物库存记录），
|
||
因为报废从“申请那一刻”就要求指明具体要报废的东西。
|
||
"""
|
||
__tablename__ = 'scrap_approval'
|
||
|
||
id = db.Column(db.Integer, primary_key=True)
|
||
request_no = db.Column(db.String(100), unique=True, nullable=False, index=True)
|
||
|
||
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, index=True)
|
||
|
||
allowed_approvers = db.Column(db.Text) # JSON: [{"type":"user","value":123}, ...]
|
||
actual_approver_id = db.Column(db.Integer, index=True)
|
||
approved_at = db.Column(db.DateTime)
|
||
reject_reason = db.Column(db.Text)
|
||
|
||
# 明细快照: [{"source_table","stock_id","sku","name","spec_model","location","batch_number","scrap_qty"}]
|
||
items_json = db.Column(db.Text, nullable=False)
|
||
|
||
# 执行信息
|
||
executed_at = db.Column(db.DateTime)
|
||
executor_name = db.Column(db.String(100))
|
||
|
||
created_at = db.Column(db.DateTime, default=_beijing_now, nullable=False)
|
||
updated_at = db.Column(db.DateTime, default=_beijing_now, onupdate=_beijing_now, nullable=False)
|
||
|
||
def set_items(self, items):
|
||
self.items_json = json.dumps(items or [], ensure_ascii=False)
|
||
|
||
def get_items(self):
|
||
try:
|
||
return json.loads(self.items_json) if self.items_json else []
|
||
except Exception:
|
||
return []
|
||
|
||
def set_allowed_approvers(self, approvers):
|
||
self.allowed_approvers = json.dumps(approvers or [], ensure_ascii=False)
|
||
|
||
def get_allowed_approvers(self):
|
||
try:
|
||
return json.loads(self.allowed_approvers) if self.allowed_approvers else []
|
||
except Exception:
|
||
return []
|
||
|
||
@staticmethod
|
||
def _user_name(user_id):
|
||
if not user_id:
|
||
return ''
|
||
try:
|
||
from app.models.system import SysUser
|
||
u = SysUser.query.get(user_id)
|
||
return u.username if u else ''
|
||
except Exception:
|
||
return ''
|
||
|
||
def to_dict(self):
|
||
return {
|
||
'id': self.id,
|
||
'request_no': self.request_no,
|
||
'applicant_id': self.applicant_id,
|
||
'applicant_name': self._user_name(self.applicant_id),
|
||
'remark': self.remark or '',
|
||
'status': self.status,
|
||
'allowed_approvers': self.get_allowed_approvers(),
|
||
'actual_approver_id': self.actual_approver_id,
|
||
'actual_approver_name': self._user_name(self.actual_approver_id),
|
||
'approved_at': self.approved_at.strftime('%Y-%m-%d %H:%M:%S') if self.approved_at else None,
|
||
'reject_reason': self.reject_reason or '',
|
||
'items': self.get_items(),
|
||
'executed_at': self.executed_at.strftime('%Y-%m-%d %H:%M:%S') if self.executed_at else None,
|
||
'executor_name': self.executor_name or '',
|
||
'created_at': self.created_at.strftime('%Y-%m-%d %H:%M:%S') if self.created_at else None,
|
||
}
|