"""设备出库明细 — 业务逻辑层（合并后的唯一入口） 「这台设备对应 MOM 的哪些出库单、领了哪些料」在本模块只有一个概念、 一张表（`product_outbound_materials`）、一组函数。原先那两张表 （`product_outbounds` 单据级 / `task_outbound_materials` 明细级）已停止写入， 只作回滚备份保留 —— 详见该模型的模块注释。 ═══ 两条写入路径 ═══ · `link_outbound_lines` —— 人在界面上挂（网页端/移动端选 MOM 出库单） → source='manual'，可删 · `archive_from_webhook` —— MOM 出库回调自动存档（按 SN 匹配到设备） → source='webhook'，系统事实，不可删（要撤得去 MOM 撤回） 两条路径共用同一张表、同一套幂等约束，但**来源不同、删除规则不同** —— 这是它们唯一的差别，`source` 列把它记下来。 """ import logging import uuid from fastapi.concurrency import run_in_threadpool from fastapi import HTTPException, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.product import Product from app.models.product_outbound_material import ProductOutboundMaterial from app.schemas.product import ProductOutboundMaterialResponse from app.services import mom_outbound_service logger = logging.getLogger(__name__) async def list_product_materials( db: AsyncSession, product_id: uuid.UUID, ) -> list[ProductOutboundMaterialResponse]: """列出某设备挂载的全部出库明细。 排序：先按 MOM 出库时间倒序，没有时间的沉底，再按写入时间兜底 —— 避免 outbound_time 为空的行插在最前面。 """ rows = ( await db.execute( select(ProductOutboundMaterial) .where(ProductOutboundMaterial.product_id == product_id) .order_by( ProductOutboundMaterial.outbound_time.desc().nullslast(), ProductOutboundMaterial.created_at.desc(), ProductOutboundMaterial.id.desc(), ) ) ).scalars().all() items = [ProductOutboundMaterialResponse.model_validate(r) for r in rows] fill_added_by_names(items) return items def fill_added_by_names( items: list[ProductOutboundMaterialResponse], ) -> list[ProductOutboundMaterialResponse]: """就地补上 `added_by_name`（谁挂上去的，中文姓名），并**返回同一个列表**。 返回列表是为了能写成 `fill(x)` 直接当值用 —— 就地修改却返回 None 的话， `fill(x) or []` 会静默变成空数组（写这行时就差点踩到）。 界面要显示「谁挂的」——`added_by` 存的是 Track 用户名（如 `zhangsan01`）， 直接摆出来现场看不懂。解析交给服务端：移动端与网页端各抄一份 username→姓名 的映射迟早会漂移，而且 MOM 的 username 是「姓名/账号」格式，规则不止一条。 ⚠️ 解析失败**不能影响列表**：MOM 连不上时姓名降级为空串，前端回落到显示 用户名。查一次是批量 SQL（mom_cache 还带 2h TTL），不给每行单独打库。 """ from app.services.mom_cache import get_display_names names = [i.added_by for i in items if i.added_by] if not names: return items try: mapping = get_display_names(list(dict.fromkeys(names))) except Exception as e: logger.warning(f"[OutboundMaterial] 解析挂载人姓名失败，降级显示用户名: {e}") return items for i in items: if i.added_by: i.added_by_name = mapping.get(i.added_by, "") return items async def link_outbound_lines( db: AsyncSession, product: Product, mom_line_ids: list[int], *, task_id: uuid.UUID | None = None, added_by: str | None = None, ) -> int: """把 MOM 出库**明细行**挂到设备上（人工路径）。返回实际新增行数。 用户勾的是**整张出库单**，提交时把该单全部明细行 id 一起带过来 —— 本表按明细行成行，所以一张单会展开成 N 行。 ⚠️ 只接受 `mom_line_ids`，物料快照一律由后端拿 id 去 MOM 现查， 否则前端可以伪造「挂的是什么」。 ⚠️ MOM 查询是同步 psycopg2，用 run_in_threadpool 扔出去，别阻塞事件循环。 ⚠️ 调用方负责 commit —— 本函数只 flush，好让挂载与产品创建同事务。 幂等：已挂过的明细跳过（部分唯一索引兜底）。 """ ids = list(dict.fromkeys(int(i) for i in mom_line_ids)) # 去重且保持顺序 if not ids: return 0 lines = await run_in_threadpool(mom_outbound_service.get_lines_by_ids, ids) if not lines: # MOM 侧查不到（数据被清理 / ID 传错）—— 静默返回 0，由调用方比对数量 return 0 existing = set( ( await db.execute( select(ProductOutboundMaterial.mom_line_id).where( ProductOutboundMaterial.product_id == product.id, ProductOutboundMaterial.mom_line_id.in_( [ln["line_id"] for ln in lines]), ) ) ).scalars().all() ) added = 0 for ln in lines: if ln["line_id"] in existing: continue db.add(ProductOutboundMaterial( product_id=product.id, serial_number=product.serial_number, task_id=task_id, mom_line_id=ln["line_id"], outbound_no=ln["outbound_no"], request_no=ln.get("request_no") or None, applicant_name=None, # 人工挂载拿不到申请人：MOM 的 request_id 存量全为 NULL remark=None, sku=ln.get("sku") or None, material_name=ln.get("material_name") or None, spec_model=ln.get("spec_model") or None, quantity=ln.get("quantity"), unit_price=ln.get("unit_price"), outbound_type=ln.get("outbound_type") or None, consumer_name=ln.get("consumer_name") or None, operator_name=ln.get("operator_name") or None, warehouse_location=ln.get("warehouse_location") or None, outbound_time=ln.get("outbound_time"), source="manual", added_by=added_by, )) added += 1 await db.flush() return added async def archive_from_webhook( db: AsyncSession, product: Product, payload, *, company_name: str | None = None, ) -> bool: """MOM 出库回调 → 存档到本表（webhook 路径）。返回是否有新写入。 按 `payload.outbound_no` 去 MOM **现查明细**，逐行落 —— 本表是明细级， 只写一条单据级信息的话，这台设备上「领了什么料」就永远是空的。 ⚠️ 查不到明细（MOM 数据被清理、或该单确实没有可解析的明细）时， 退化成写**一行 `mom_line_id=NULL` 的存档**：宁可显示「有这张单但看不到明细」， 也不要静默丢掉这张单 —— 用户会以为出库记录丢了。 ⚠️ 幂等：已有存档就跳过（webhook 可能因运维重放而重入）。 """ outbound_no = (getattr(payload, "outbound_no", "") or "").strip() if not outbound_no: return False # 已有该单的存档 → 不重复写。**按 outbound_no 判**（不是按明细行）， # 因为一张单的所有明细必然一起写入，任一行存在即整单已存过。 already = ( await db.execute( select(ProductOutboundMaterial.id).where( ProductOutboundMaterial.product_id == product.id, ProductOutboundMaterial.outbound_no == outbound_no, ).limit(1) ) ).scalar_one_or_none() if already is not None: return False lines = await run_in_threadpool( mom_outbound_service.get_lines_by_outbound_no, outbound_no) if not lines: # 退化成单据级存档：能看、能标撤回，但没有明细行 id，**不能报废** db.add(ProductOutboundMaterial( product_id=product.id, serial_number=product.serial_number, task_id=None, mom_line_id=None, outbound_no=outbound_no, request_no=getattr(payload, "request_no", None) or None, applicant_name=getattr(payload, "applicant_name", None) or None, remark=getattr(payload, "remark", None) or None, outbound_type=getattr(payload, "outbound_type", None) or None, consumer_name=getattr(payload, "consumer_name", None) or None, operator_name=(getattr(payload, "operator", None) or "")[:100] or None, outbound_time=getattr(payload, "outbound_time", None), source="webhook", )) logger.info(f"[OutboundMaterial] {outbound_no} MOM 查不到明细，退化为单据级存档") return True for ln in lines: db.add(ProductOutboundMaterial( product_id=product.id, serial_number=product.serial_number, task_id=None, mom_line_id=ln["line_id"], outbound_no=outbound_no, request_no=getattr(payload, "request_no", None) or None, applicant_name=getattr(payload, "applicant_name", None) or None, remark=getattr(payload, "remark", None) or None, sku=ln.get("sku") or None, material_name=ln.get("material_name") or None, spec_model=ln.get("spec_model") or None, quantity=ln.get("quantity"), unit_price=ln.get("unit_price"), outbound_type=(getattr(payload, "outbound_type", None) or ln.get("outbound_type") or None), consumer_name=(getattr(payload, "consumer_name", None) or ln.get("consumer_name") or None), operator_name=(getattr(payload, "operator", None) or ln.get("operator_name") or None), warehouse_location=ln.get("warehouse_location") or None, outbound_time=(getattr(payload, "outbound_time", None) or ln.get("outbound_time")), source="webhook", )) return True async def mark_revoked( db: AsyncSession, product_id: uuid.UUID, outbound_no: str | None = None, ) -> bool: """MOM 撤回回调 → 把该单在本设备上的存档标为已撤回。返回是否有改动。 **只置位不删行** ——「出过又撤了」本身就是要看得见的历史。 :param outbound_no: 指定单号则只标那一单；为空则标**最近一条未撤回的**。 MOM 的撤回载荷不保证带单号，而「最近一条未撤回的」是这批里最可能被撤的那张 —— 一批里同一台设备理论上不该出现两条未撤回的出库单（出库后设备已不在 仓库池，再出库匹配不到）。真出现时标错一条，也好过把历史全标脏。 """ from app.core.time_utils import get_beijing_time q = select(ProductOutboundMaterial).where( ProductOutboundMaterial.product_id == product_id, ProductOutboundMaterial.is_revoked.is_(False), ) if outbound_no: q = q.where(ProductOutboundMaterial.outbound_no == outbound_no) rows = (await db.execute(q)).scalars().all() else: # 先定位到「哪一张单」，再把那张单的**全部明细行**一起标 —— # 只标一行的会让同一张单呈现「一半撤回一半没撤」的鬼状态 latest = ( await db.execute( q.order_by( ProductOutboundMaterial.outbound_time.desc().nullslast(), ProductOutboundMaterial.created_at.desc(), ).limit(1) ) ).scalars().first() if latest is None: return False rows = ( await db.execute( select(ProductOutboundMaterial).where( ProductOutboundMaterial.product_id == product_id, ProductOutboundMaterial.outbound_no == latest.outbound_no, ) ) ).scalars().all() changed = False for r in rows: if not r.is_revoked: r.is_revoked = True r.revoked_at = get_beijing_time() changed = True return changed async def remove_product_order( db: AsyncSession, product_id: uuid.UUID, outbound_no: str, ) -> list[ProductOutboundMaterialResponse]: """整张出库单一起摘掉（挂错了要能撤）。 界面上更容易碰到的是「这一单整个挂错了」，逐条删要删好几下，所以补这个。 规则与逐条删**完全一致**：只要这张单在本设备上有一行是 `webhook` （MOM 回调自动存档的系统事实），整单就不给删 —— 要撤得去 MOM 撤回。 这样「整单删」不会成为绕过单行规则的后门。 """ no = (outbound_no or "").strip() if not no: raise HTTPException(status_code=400, detail="出库单号不能为空") rows = ( await db.execute( select(ProductOutboundMaterial).where( ProductOutboundMaterial.product_id == product_id, ProductOutboundMaterial.outbound_no == no, ) ) ).scalars().all() if not rows: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="这张出库单没有挂在这台设备上") if any(r.source != "manual" for r in rows): raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="该出库单含 MOM 回调自动存档的记录，不能在 Track 里删除；如需撤销请在 MOM 中撤回", ) for r in rows: await db.delete(r) await db.commit() return await list_product_materials(db, product_id) async def remove_product_material( db: AsyncSession, product_id: uuid.UUID, material_id: int, ) -> list[ProductOutboundMaterialResponse]: """摘掉一条**人工挂载**的出库明细（挂错了要能撤）。 只允许删 `source='manual'`：webhook 存档是 MOM 出库回调留下的系统事实， 删了 Track 与 MOM 就对不上（MOM 那边单还在）。要撤该去 MOM 撤回， 由回调置 `is_revoked` 留痕。 """ row = ( await db.execute( select(ProductOutboundMaterial).where( ProductOutboundMaterial.id == material_id, ProductOutboundMaterial.product_id == product_id, ) ) ).scalars().first() if row is None: # 不属于这台设备的一律按「查不到」处理，不泄漏别的设备挂了什么 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="这条出库明细没有挂在这台设备上") if row.source != "manual": raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="该出库单由 MOM 回调自动存档，不能在 Track 里删除；如需撤销请在 MOM 中撤回", ) await db.delete(row) await db.commit() return await list_product_materials(db, product_id)