from app.extensions import db, beijing_time import json # ============================================================================= # 报废原因分类 —— 用来回答「这笔损失出在哪个环节」 # # 业务口径（2026-09 与业务确认）： # PRODUCTION 生产环节 —— **好料领到产线之后**，在生产中变坏了。 # STOCK 库存/采购 —— 东西还在仓库里就有问题、买来就坏、或售后从客户 # 退回后发现的不良。凡「不是生产过程中弄坏的」都归这里。 # OTHER 其他 —— 兜底，给以后说不清的情况留个位置。 # # 大前提：这几类都发生在**发货之前**，都属于「生产未发货前的损失」， # 与「已发货后客户退货产生的损失」不是一回事。 # # ★★★ 为什么分类**必须显式记录**，绝不能从 source_table 推导 ★★★ # Track 提交的生产报废走的是「出库领用 → 退回(不良品) → 在管不良品 → 报废」， # 而 MOM 界面手工报的不良品退回**走的是同一条路、落进同一张 # trans_defective_goods 表**。两条路在 source_table 上长得一模一样： # · Track 提交的 → 料是好的，生产时弄坏的 → PRODUCTION # · MOM 手工报的 → 料本来就有问题 → STOCK # 一旦写成「trans_defective_goods 就推成 STOCK」，Track 报的生产损失会被 # 静默算成库存损失，「统计生产报废金额」这个目标当场失效。 # 所以：**提交时谁传什么就是什么，推导是错的。** # # ★ 单一事实来源：模型定义码表，service / API / 前端文案都从这里取， # 不要再抄一份 —— 抄出去就开始漂移。 # ★ 加新分类只需扩这个 dict，**不需要 DDL**（列是 varchar(50)）。 # 刻意不预置「质量/运输/存储」等更多分类：那是给 UI 的承诺，没有业务确认不要先造。 # ============================================================================= SCRAP_CATEGORY_PRODUCTION = 'PRODUCTION' # 生产报废：**所有走 Track 的** SCRAP_CATEGORY_STOCK = 'STOCK' # 库存报废：MOM 自身流程走下来的 SCRAP_CATEGORY_OTHER = 'OTHER' # 其他（兜底，暂未使用） SCRAP_CATEGORY_LABELS = { SCRAP_CATEGORY_PRODUCTION: '生产报废', SCRAP_CATEGORY_STOCK: '库存报废', SCRAP_CATEGORY_OTHER: '其他', } # 中文别名 → 码。外部系统/前端传中文时用得上（码本身大小写不敏感，此处只列中文） # 老叫法（生产损耗 / 库存采购）保留兼容 —— Track 侧可能已经在传旧名字 _SCRAP_CATEGORY_ALIASES = { '生产报废': SCRAP_CATEGORY_PRODUCTION, '生产损耗': SCRAP_CATEGORY_PRODUCTION, # 旧标签 '生产导致': SCRAP_CATEGORY_PRODUCTION, # 更早的叫法 '生产': SCRAP_CATEGORY_PRODUCTION, '库存报废': SCRAP_CATEGORY_STOCK, '库存/采购': SCRAP_CATEGORY_STOCK, # 旧标签 '库存': SCRAP_CATEGORY_STOCK, '采购': SCRAP_CATEGORY_STOCK, '其他': SCRAP_CATEGORY_OTHER, } def normalize_scrap_category(raw, default=None): """报废原因分类入参归一化 → 分类码。 接受分类码（大小写不敏感）或中文别名。空值返回 `default`。 ⚠️ 未知值**抛 ValueError** 而不是静默回落 —— 白名单外的东西一律挡在入口， 否则会写进库里一个谁也看不懂的码，统计时才发现分类是脏的。 """ text = str(raw or '').strip() if not text: return default upper = text.upper() if upper in SCRAP_CATEGORY_LABELS: return upper if text in _SCRAP_CATEGORY_ALIASES: return _SCRAP_CATEGORY_ALIASES[text] raise ValueError( '报废原因分类不支持：{}，仅支持 {}'.format( text, '、'.join(sorted(SCRAP_CATEGORY_LABELS)), ) ) def scrap_category_label(code): """分类码 → 中文名。未知/空值返回空串（前端自行兜底 '-'）。""" return SCRAP_CATEGORY_LABELS.get(str(code or '').strip().upper(), '') def _beijing_now(): """ 统一时间口径：naive 北京时间，与 created_at / OutboundApproval / BorrowApproval 一致。 ★ 原先返回 tz-aware 时间，配合 timestamptz 列才正确；现已将该表列类型统一为 timestamp without time zone（见 db_migrations/unify_approval_timezone.sql）， 故此处必须返回 naive 值，否则存库会被当 UTC 处理而早 8 小时。 """ return beijing_time() 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) # 报废原因 # 申请所属公司（= 申请人 sys_user.department）。 # ★ 这是**角色级审批**能做公司隔离的前提：主管共 6 人（IRIS 5 / LICA 1）， # 只按角色放行而不比公司，LICA 主管就能审 IRIS 的报废单。 # 存快照而不联表查 sys_user：申请人被删后仍要能定性这张单属于谁。 company_name = db.Column(db.String(255), index=True) # 报废原因分类码（SCRAP_CATEGORY_*）。执行时经 _ledger_kwargs 带到 trans_scrap。 reason_category = db.Column(db.String(50)) # 外部系统唯一引用，格式 :<外部单号>。 # 幂等与追溯的锚点（Redis 未部署，prevent_double_submit 全程 fail-open， # 唯一索引是唯一防线；DDL 见 db_migrations/phase12_production_scrap.sql）。 source_ref = db.Column(db.String(100), index=True) # 状态: 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 '', 'company_name': self.company_name or '', 'reason_category': self.reason_category or '', 'reason_category_label': scrap_category_label(self.reason_category), 'source_ref': self.source_ref 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, }