@ -11,7 +11,7 @@ from __future__ import annotations
from datetime import datetime
from fastapi import APIRouter , Depends , Header , HTTPException
from fastapi import APIRouter , Depends , Header , HTTPException , Request
from pydantic import BaseModel
from sqlalchemy import or_ , select
from sqlalchemy . ext . asyncio import AsyncSession
@ -36,6 +36,7 @@ class MomInboundPayload(BaseModel):
event : str | None = None # 事件名,如 inbound.created / outbound.revoked
action : str | None = None # 显式动作指令,如 revoke_outbound
source_table : str | None = None # stock_product / stock_semi
company_name : str | None = None # 目标公司( IRIS / LICA) , MOM 据此分流到不同 Track 实例
# 「撤回出库」信号词 —— 只在 action / event 里做子串匹配。
@ -44,6 +45,56 @@ class MomInboundPayload(BaseModel):
_OUTBOUND_REVOKE_TOKENS = ( " revoke " , " rollback " , " revert " , " cancel " )
# ── 公司归属分流 ──────────────────────────────────────────────────────────
# MOM 现在会在载荷里带 company_name, 同一套物理库可能同时向多个 Track 实例
# ( IRIS / LICA) 回调。本实例服务的是 IRIS, 故只放行 IRIS 与空白值。
#
# ⚠️ 判定刻意做成「只排除已知的外来公司」,而非「白名单只认 IRIS」:
# MOM 在无法确定公司归属时会回落到扁平配置,该配置指向本实例 —— 这类
# 消息的 company_name 会是空 / 缺失。若此处按白名单把空白也拒掉,它们
# 就彻底丢了: MOM 那边已收到 200、认为投递成功, 不会再重推。
# 同理,未见过的新值(不是 IRIS 也不是 LICA) 也一律照常处理。
_FOREIGN_COMPANIES = { " LICA " }
def _is_foreign_company ( company_name : str | None ) - > bool :
""" 载荷是否属于本实例不该处理的其它公司。
返回 True 表示应原样忽略(仍回 200, 避免 MOM 反复重推)。
"""
# 大小写 / 首尾空白都容忍: MOM 侧常量书写方式未必冻结,误判的代价是
# 一条消息被错误地当成本公司处理(有唯一匹配约束,最坏是 matched=False) 。
return ( company_name or " " ) . strip ( ) . upper ( ) in _FOREIGN_COMPANIES
def _attribute_audit_to_mom_operator ( request : Request , operator : str | None ) - > None :
""" 把外部回调归因到 MOM 侧的实际操作人。
外部回调走 X-API-Key 鉴权、没有 JWT, 所以 JWT 依赖不执行,
审计中间件读到的 request.state.audit_user 永远是空 ——
操作审计里就出现一堆没有归属的「外部系统对接」记录。
但 MOM 载荷里本来就带着实际操作人( operator, 即 MOM 侧扫码的那位),
写进 request.state 即可让审计归因到人。
⚠️ 必须在 X-API-Key 校验【之后】调用:密钥不对说明载荷本身就不可信,
此时把 operator 写进审计等于允许伪造人。
"""
who = ( operator or " " ) . strip ( )
if not who :
# 取不到操作人时留一个明确的系统标记,而不是继续显示「未认证」——
# 「MOM系统」至少说明这是一次机器回调, 不是"一个匿名的人"。
request . state . audit_user = " MOM系统 "
return
request . state . audit_user = who
try :
# 尽力而为:查不到中文名也不影响审计(前端会回退显示账号)
from app . services . mom_cache import get_display_names
request . state . audit_display_name = get_display_names ( [ who ] ) . get ( who ) or " "
except Exception : # noqa: BLE001 —— 姓名解析失败绝不能影响回调处理
pass
def _is_outbound_revoke ( payload : MomInboundPayload ) - > bool :
""" payload 是否携带**显式**的撤回出库信号。
@ -127,6 +178,7 @@ async def _match_inbound_product(
@router.post ( " /mom-inbound " )
async def mom_inbound_webhook (
payload : MomInboundPayload ,
request : Request ,
x_api_key : str | None = Header ( default = None , alias = " X-API-Key " ) ,
db : AsyncSession = Depends ( get_db ) ,
) - > dict :
@ -136,12 +188,23 @@ async def mom_inbound_webhook(
- 常规入库:用 serial_number( 优先) 或 sku 匹配「当前位于 virtual_warehouse」
的产品,命中则标记 " 已实收 " ( overall_status=已入库 + 记录 task_logs) 。
- 撤回出库: MOM 把误出库的设备物理回滚到仓库 → 本接口强制执行特权回滚。
- 公司归属: company_name 明确写着其它公司( LICA) 时原样忽略; 空白 / 缺失
一律照常处理(见 _is_foreign_company 的说明)。
- 未命中返回 200( MOM 可能操作了非 Track 生产的物料,直接忽略)。
"""
# ── 鉴权 ──
if not settings . TRACK_WEBHOOK_KEY or x_api_key != settings . TRACK_WEBHOOK_KEY :
raise HTTPException ( status_code = 401 , detail = " Unauthorized: invalid X-API-Key " )
# ── 公司归属:不是本实例的消息原样忽略(仍回 200, 避免 MOM 当作失败而重推) ──
# ⚠️ 键名 reason / 取值 "ignored_company" 与 LICA 实例(~/track-lica) 保持一致:
# MOM 侧不解析它,但排查时两边日志对着看,字段名不一致会白白浪费时间。
if _is_foreign_company ( payload . company_name ) :
return { " ok " : True , " matched " : False , " reason " : " ignored_company " }
# 归因到 MOM 侧实际扫码的人(必须在鉴权通过之后,见函数注释)
_attribute_audit_to_mom_operator ( request , payload . operator )
explicit_revoke = _is_outbound_revoke ( payload )
# ── 匹配产品 ──
@ -329,11 +392,17 @@ class MomOutboundPayload(BaseModel):
sku : str | None = None # 规格型号 spec_model( serial 缺失时的兜底匹配)
operator : str | None = None # 出库操作人(写入 task_logs.operator_id)
outbound_time : datetime | None = None # 出库时间
company_name : str | None = None # 目标公司( IRIS / LICA) , MOM 据此分流到不同 Track 实例
# ↓ 2026-09 新增: MOM 一直在发、此前被 Pydantic 静默丢弃。当前无人读取,
# 先接住是为了与 LICA 实例(~/track-lica) 对同一载荷的解析结果保持一致 ——
# 否则将来谁写了读这个字段的代码,会在 LICA 拿到值、在本实例拿到 None。
outbound_type : str | None = None # SALES / USE / PRODUCTION
@router.post ( " /mom-outbound " )
async def mom_outbound_webhook (
payload : MomOutboundPayload ,
request : Request ,
x_api_key : str | None = Header ( default = None , alias = " X-API-Key " ) ,
db : AsyncSession = Depends ( get_db ) ,
) - > dict :
@ -342,12 +411,23 @@ async def mom_outbound_webhook(
- 鉴权: Header X-API-Key 必须等于环境变量 TRACK_WEBHOOK_KEY。
- 用 serial_number( 优先) 或 sku 匹配 " 在仓库/已入库 " 的产品;
命中则标记 " 已出库 " ( overall_status=已出库 + status=OUTBOUND + 记录 task_logs) 。
- 公司归属: company_name 明确写着其它公司( LICA) 时原样忽略; 空白 / 缺失
一律照常处理(见 _is_foreign_company 的说明)。
- 未命中返回 200( MOM 出库的可能是非 Track 生产的物料,直接忽略)。
"""
# ── 鉴权 ──
if not settings . TRACK_WEBHOOK_KEY or x_api_key != settings . TRACK_WEBHOOK_KEY :
raise HTTPException ( status_code = 401 , detail = " Unauthorized: invalid X-API-Key " )
# ── 公司归属:不是本实例的消息原样忽略(仍回 200, 避免 MOM 当作失败而重推) ──
# ⚠️ 键名 reason / 取值 "ignored_company" 与 LICA 实例(~/track-lica) 保持一致:
# MOM 侧不解析它,但排查时两边日志对着看,字段名不一致会白白浪费时间。
if _is_foreign_company ( payload . company_name ) :
return { " ok " : True , " matched " : False , " reason " : " ignored_company " }
# 归因到 MOM 侧实际出库的人(必须在鉴权通过之后,见函数注释)
_attribute_audit_to_mom_operator ( request , payload . operator )
# ── 按 serial_number( 优先) 或 sku 匹配"在仓库/已入库"的产品 ──
product = None
where_cond = or_ (