duxingchen
e45c97bd1f
本轮之前累积的未提交工作，一并固化：
- 组织隔离：同一份代码部署给不同部门只需改 config 的 ORG_DEPARTMENT 与
MATERIAL_CATEGORY_PREFIX。过滤点在登录/人员列表/物料/MOM 出库单四处，
全部服务端钉死，客户端传什么都放不大。
★ 物料必须用**前缀** LIKE，不能反推成 ILIKE '%IRIS%'：MOM 里 LICA 的物料是
`LICA/<中文>`，而本部门分类树里另有 `IRIS/成品/LICA/…`（本就属于本部门），
前缀匹配天然区分得开。
- MOM 出库单只读查询（直连 MOM 库）：不走 MOM 现成的 /outbound 接口 ——
那个要 JWT + permission_required，且对非特权账号按 consumer_name 做行级
隔离，服务账号只能拿到自己名下的单。分页必须两段式（先按单号 GROUP BY
分页，再 IN 捞明细），对宽表直接分页会得到明细行数而不是单据数。
- 出料功能：产品 ↔ 出库单存档（product_outbounds）与任务 ↔ 出库明细
（task_outbound_materials），供「这台设备对应 MOM 哪张单」的展示。
⚠️ 快照一律由后端拿 ID 去 MOM 现查，不接受前端传入，否则前端可伪造单据。
399 lines
16 KiB
Python
399 lines
16 KiB
Python
"""MOM 出库单只读查询 — 直连 MOM 库（跨库，无 ORM）
|
||
|
||
为什么直连库，而不是调 MOM 现成的 `GET /api/v1/outbound`：
|
||
那个接口要 JWT + `permission_required`，且对非特权账号按 `consumer_name`
|
||
做**行级隔离**（非特权用户只能看到自己名下的单）。Track 用服务账号去调只会
|
||
拿到该账号名下的数据、不是全量。Track 已有只读 MOM 连接
|
||
（`app/core/mom_database.py`，mom_cache 也在用），直连才是对的。
|
||
|
||
分页必须两段式（照抄 MOM `outbound_service.get_grouped_list` 的做法）：
|
||
1) 先 `GROUP BY outbound_no` 分页，拿到本页的**单据号**
|
||
2) 再 `WHERE outbound_no IN (...)` 捞这些单的明细
|
||
⚠️ 绝不能对 join 后的宽表直接分页 —— 那是**明细行数**不是单据数。
|
||
`outbound_no` 不唯一（批量出库多商品共用），一张单最多 55 条明细，
|
||
实测 517 张单平均 2.64 条、65% 是单条。
|
||
|
||
物料名解析（MOM 自己没有视图，是代码里硬拼的）：
|
||
`trans_outbound.(source_table, stock_id)` 多态指向三张库存表之一，再经它们的
|
||
`base_id` 回到 `material_base`。用 COALESCE 三路 LEFT JOIN 一次拉全。
|
||
⚠️ **不要用 `trans_outbound.sku` 做 JOIN** —— 它只是库存表 sku 的冗余快照，
|
||
不唯一，历史还可能漂移。
|
||
"""
|
||
import logging
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from sqlalchemy import text
|
||
|
||
from app.core.config import settings
|
||
from app.core.mom_database import MomSessionLocal
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# MOM 的 outbound_time 是 `timestamp without time zone`，存的是**北京墙上时间**。
|
||
BEIJING_TZ = timezone(timedelta(hours=8))
|
||
|
||
# 跨部门例外领用人 —— 这几个领用人的出库单，即使物料分类不属于本部门，本实例
|
||
# 也要能看见。定义与理由见 config.EXTRA_VISIBLE_CONSUMERS。
|
||
# 本模块只负责**执行**这条例外、不负责判定；改范围请改配置，不要改这里。
|
||
_EXTRA_CONSUMERS: list[str] = settings.EXTRA_VISIBLE_CONSUMERS_LIST
|
||
|
||
# 三张库存表 → material_base 的 JOIN 片段。
|
||
# 抽成常量是因为「搜索命中」与「拉明细」两处都要用，写歪一处两边就不一致了。
|
||
_STOCK_JOIN = """
|
||
LEFT JOIN stock_buy sb ON o.source_table = 'stock_buy' AND sb.id = o.stock_id
|
||
LEFT JOIN stock_semi ss ON o.source_table = 'stock_semi' AND ss.id = o.stock_id
|
||
LEFT JOIN stock_product sp ON o.source_table = 'stock_product' AND sp.id = o.stock_id
|
||
LEFT JOIN material_base b ON b.id = COALESCE(sb.base_id, ss.base_id, sp.base_id)
|
||
"""
|
||
|
||
# 明细行的公共 SELECT 列表（两处查询共用，保证返回字段一致）
|
||
_LINE_COLUMNS = """
|
||
o.id AS line_id, o.outbound_no, o.sku,
|
||
b.name AS material_name, b.spec_model,
|
||
o.quantity, o.unit_price, o.returned_quantity,
|
||
o.outbound_type, o.consumer_name, o.operator_name,
|
||
o.warehouse_location, o.outbound_time,
|
||
a.request_no
|
||
"""
|
||
|
||
# `trans_outbound.outbound_type` 的中文名 —— 码表源头是 MOM 前端的
|
||
# `inventory-web/src/views/outbound/index.vue::formatType`，这里照抄一份**统一
|
||
# 下发**（响应里的 outbound_type_label），前端不再自建映射，否则两边会开始漂移。
|
||
# ⚠️ 只用于展示，**不做任何业务判断** —— MOM 码表未冻结（`types/api.ts` 同注）。
|
||
# 库里存的只有 SALES/USE/PRODUCTION 三种，其余是按 MOM 下拉预留的。
|
||
_OUTBOUND_TYPE_LABELS = {
|
||
"SALES": "销售出库",
|
||
"USE": "内部领用",
|
||
"PRODUCTION": "生产出库",
|
||
"SCRAP": "报废",
|
||
"LOSS": "盘亏出库",
|
||
"REPAIR": "维修出库",
|
||
}
|
||
|
||
|
||
def describe_outbound_type(code: str | None) -> str:
|
||
"""出库类型码 → 中文名。
|
||
|
||
码表里没有的码**原样返回**（不吞成空串、也不显示「未知」）：宁可把英文码
|
||
露在界面上让人一眼看出是漏配的码表，也不要静默成一句看不出问题的中文。
|
||
"""
|
||
raw = (code or "").strip()
|
||
return _OUTBOUND_TYPE_LABELS.get(raw.upper(), raw)
|
||
|
||
|
||
def _as_beijing(dt: datetime | None) -> datetime | None:
|
||
"""把 MOM 的 naive 北京时间补上 +08:00 偏移。
|
||
|
||
⚠️ 少了这一步，上层（尤其写进 Track 的 timestamptz 列时）会把它当 **UTC**
|
||
处理，前端在北京显示会整整差 8 小时。这个坑在 MOM 仓的 822f897 刚踩过
|
||
一次，这里统一在数据出口处理，不留给调用方去记得。
|
||
"""
|
||
if dt is None:
|
||
return None
|
||
if dt.tzinfo is None:
|
||
return dt.replace(tzinfo=BEIJING_TZ)
|
||
return dt.astimezone(BEIJING_TZ)
|
||
|
||
|
||
def _line_dict(row) -> dict:
|
||
"""把一行查询结果转成明细 dict（字段名与 Track 侧快照列对齐）。"""
|
||
return {
|
||
"line_id": row.line_id,
|
||
"sku": row.sku or "",
|
||
"material_name": row.material_name or "",
|
||
"spec_model": row.spec_model or "",
|
||
"quantity": row.quantity,
|
||
"unit_price": row.unit_price,
|
||
"returned_quantity": row.returned_quantity,
|
||
"outbound_type": row.outbound_type or "",
|
||
"outbound_type_label": describe_outbound_type(row.outbound_type),
|
||
"consumer_name": row.consumer_name or "",
|
||
"operator_name": row.operator_name or "",
|
||
"warehouse_location": row.warehouse_location or "",
|
||
"outbound_time": _as_beijing(row.outbound_time),
|
||
# MOM 的 request_id 是最近才加的列，存量 1364 条**全为 NULL**（无从回填，
|
||
# 见该仓 models/outbound.py 的注释）。UI 要对空值显示「无关联申请单」。
|
||
"request_no": row.request_no or "",
|
||
}
|
||
|
||
|
||
def _fetch_lines(db, outbound_nos: list[str]) -> dict[str, list[dict]]:
|
||
"""捞出指定单据的明细，按出库单号分组。
|
||
|
||
申请单必须 LEFT JOIN —— trans_outbound.request_id 存量全为 NULL，用 INNER
|
||
会把这批历史单的明细整批吞掉。
|
||
"""
|
||
if not outbound_nos:
|
||
return {}
|
||
rows = db.execute(
|
||
text(f"""
|
||
SELECT {_LINE_COLUMNS}
|
||
FROM trans_outbound o
|
||
{_STOCK_JOIN}
|
||
LEFT JOIN outbound_approval a ON a.id = o.request_id
|
||
WHERE o.outbound_no = ANY(:nos)
|
||
ORDER BY o.outbound_time DESC, o.id
|
||
"""),
|
||
{"nos": list(outbound_nos)},
|
||
).fetchall()
|
||
|
||
grouped: dict[str, list[dict]] = {}
|
||
for r in rows:
|
||
grouped.setdefault(r.outbound_no, []).append(_line_dict(r))
|
||
return grouped
|
||
|
||
|
||
def search_outbound_orders(
|
||
keyword: str = "",
|
||
start_date: str = "",
|
||
end_date: str = "",
|
||
skip: int = 0,
|
||
limit: int = 20,
|
||
consumers: set[str] | None = None,
|
||
) -> tuple[list[dict], int]:
|
||
"""按**单据**分页搜索 MOM 出库单，返回 `(单据列表, 单据总数)`。
|
||
|
||
keyword 命中范围：出库单号 / 物料SKU / 物料名称 / **规格型号** / 领用人。
|
||
start_date、end_date 为 `YYYY-MM-DD`，**含端点的整日**。
|
||
|
||
consumers：界面筛选的**领用人姓名**集合（AND 语义，只收窄）。
|
||
· `None` → 不限
|
||
· 非空集合 → 只返回这些领用人的单
|
||
· **空集合** → 一条都不返回（绝不退化成「不过滤」）
|
||
|
||
⚠️ 本函数只负责**执行**范围，不负责**判定**范围 —— 可见范围（公司前缀
|
||
+ 跨部门例外）在本模块内以常量化形式钉死，避免出现第二份口径。
|
||
"""
|
||
kw = keyword.strip()
|
||
# ⚠️ 日期参数用 None 而不是空串：写成 `:start = ''` 时 PostgreSQL **不保证
|
||
# 短路求值**，规划器仍会去算 `CAST('' AS timestamp)` 并直接报
|
||
# InvalidDatetimeFormat（实测踩到）。传 NULL + 显式 text 转换才安全。
|
||
params = {
|
||
"kw": kw,
|
||
"kw_like": f"%{kw}%" if kw else "",
|
||
"start": start_date.strip() or None,
|
||
"end": end_date.strip() or None,
|
||
# 公司隔离：与物料选择器同一套口径（material_base.category 前缀）。
|
||
# ⚠️ 必须是**前缀** LIKE，不能写成 ILIKE '%IRIS%'：MOM 里 LICA 的物料是
|
||
# `LICA/<中文>`，而 IRIS 分类树里另有 `IRIS/成品/LICA/…`（171 条）——
|
||
# 后者本就属于本部门，前缀匹配天然区分得开。
|
||
"cat_prefix": f"{settings.MATERIAL_CATEGORY_PREFIX}%",
|
||
}
|
||
|
||
# 可见范围 = 公司前缀（范围）∪ 跨部门例外（放行）。
|
||
# · 前缀是**范围**：本部门物料开出去的单；
|
||
# · 例外是 OR：那几个领用人跨部门领料，只按前缀过滤会把他们的单整批漏掉。
|
||
# ⚠️ 例外为空时**整段不拼**：`= ANY(ARRAY[])` 虽然返回 false（不放大范围，
|
||
# 语义安全），但留一个恒假子句只会让这条 SQL 更难排查。
|
||
if _EXTRA_CONSUMERS:
|
||
company_clause = (
|
||
"(b.category LIKE :cat_prefix OR o.consumer_name = ANY(:extra_consumers))"
|
||
)
|
||
params["extra_consumers"] = _EXTRA_CONSUMERS
|
||
else:
|
||
company_clause = "b.category LIKE :cat_prefix"
|
||
|
||
where = f"""
|
||
WHERE (:kw = '' OR o.outbound_no ILIKE :kw_like
|
||
OR o.sku ILIKE :kw_like
|
||
OR b.name ILIKE :kw_like
|
||
OR b.spec_model ILIKE :kw_like
|
||
OR o.consumer_name ILIKE :kw_like)
|
||
AND {company_clause}
|
||
AND (CAST(:start AS text) IS NULL
|
||
OR o.outbound_time >= CAST(:start AS timestamp))
|
||
AND (CAST(:end AS text) IS NULL
|
||
OR o.outbound_time < CAST(:end AS timestamp) + interval '1 day')
|
||
"""
|
||
|
||
if consumers is not None:
|
||
if not consumers:
|
||
# ⚠️ 空集合必须**显式** false。绝不能拼成 `= ANY(empty)` 或让条件消失 ——
|
||
# 空范围退化成不过滤就是全量泄漏，这是本模块的底线。
|
||
where += "\n AND false"
|
||
else:
|
||
where += "\n AND o.consumer_name = ANY(:consumers)"
|
||
params["consumers"] = sorted(consumers)
|
||
|
||
db = MomSessionLocal()
|
||
try:
|
||
total = db.execute(
|
||
text(f"""
|
||
SELECT count(DISTINCT o.outbound_no)
|
||
FROM trans_outbound o
|
||
{_STOCK_JOIN}
|
||
{where}
|
||
"""),
|
||
params,
|
||
).scalar() or 0
|
||
|
||
if total == 0:
|
||
return [], 0
|
||
|
||
order_rows = db.execute(
|
||
text(f"""
|
||
SELECT o.outbound_no,
|
||
max(o.outbound_time) AS outbound_time,
|
||
max(o.outbound_type) AS outbound_type,
|
||
max(o.consumer_name) AS consumer_name,
|
||
max(o.operator_name) AS operator_name,
|
||
count(*) AS line_count,
|
||
sum(o.quantity) AS total_quantity
|
||
FROM trans_outbound o
|
||
{_STOCK_JOIN}
|
||
{where}
|
||
GROUP BY o.outbound_no
|
||
ORDER BY outbound_time DESC, o.outbound_no DESC
|
||
LIMIT :lim OFFSET :off
|
||
"""),
|
||
{**params, "lim": limit, "off": skip},
|
||
).fetchall()
|
||
|
||
lines_by_no = _fetch_lines(db, [r.outbound_no for r in order_rows])
|
||
|
||
orders = [
|
||
{
|
||
"outbound_no": r.outbound_no,
|
||
"outbound_time": _as_beijing(r.outbound_time),
|
||
"outbound_type": r.outbound_type or "",
|
||
"outbound_type_label": describe_outbound_type(r.outbound_type),
|
||
"consumer_name": r.consumer_name or "",
|
||
"operator_name": r.operator_name or "",
|
||
"line_count": r.line_count,
|
||
"total_quantity": r.total_quantity,
|
||
"lines": lines_by_no.get(r.outbound_no, []),
|
||
}
|
||
for r in order_rows
|
||
]
|
||
return orders, total
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def get_lines_by_ids(mom_line_ids: list[int]) -> list[dict]:
|
||
"""按 MOM 明细行 ID 取快照，供挂载到任务时生成 Track 侧快照。
|
||
|
||
查不到的 ID 会被**静默跳过** —— 由调用方比对数量后决定要不要提示用户
|
||
（MOM 库清理过数据时会命中这种情况）。
|
||
"""
|
||
if not mom_line_ids:
|
||
return []
|
||
db = MomSessionLocal()
|
||
try:
|
||
rows = db.execute(
|
||
text(f"""
|
||
SELECT {_LINE_COLUMNS}
|
||
FROM trans_outbound o
|
||
{_STOCK_JOIN}
|
||
LEFT JOIN outbound_approval a ON a.id = o.request_id
|
||
WHERE o.id = ANY(:ids)
|
||
"""),
|
||
{"ids": [int(i) for i in mom_line_ids]},
|
||
).fetchall()
|
||
out = []
|
||
for r in rows:
|
||
d = _line_dict(r)
|
||
d["outbound_no"] = r.outbound_no
|
||
out.append(d)
|
||
return out
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def get_lines_by_outbound_no(outbound_no: str) -> list[dict]:
|
||
"""按**出库单号**取该单的全部明细快照（供 MOM 出库回调存档用）。
|
||
|
||
为什么需要它：统一后的 `product_outbound_materials` 是**明细级**，而
|
||
MOM 的出库回调只带单号、不带明细 —— 不现查的话，这台设备上「领了什么料」
|
||
就永远是空的。
|
||
|
||
查不到返回空列表（调用方据此退化成单据级存档，而不是丢掉这张单）。
|
||
⚠️ 批量出库多个商品共用一个单号，所以这里可能返回多行，也可能一行都没有。
|
||
"""
|
||
no = (outbound_no or "").strip()
|
||
if not no:
|
||
return []
|
||
db = MomSessionLocal()
|
||
try:
|
||
rows = db.execute(
|
||
text(f"""
|
||
SELECT {_LINE_COLUMNS}
|
||
FROM trans_outbound o
|
||
{_STOCK_JOIN}
|
||
LEFT JOIN outbound_approval a ON a.id = o.request_id
|
||
WHERE o.outbound_no = :no
|
||
ORDER BY o.outbound_time DESC, o.id
|
||
"""),
|
||
{"no": no},
|
||
).fetchall()
|
||
out = []
|
||
for r in rows:
|
||
d = _line_dict(r)
|
||
d["outbound_no"] = r.outbound_no
|
||
out.append(d)
|
||
return out
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def list_consumer_names(consumers: set[str] | None = None) -> list[str]:
|
||
"""本部门出库单里出现过的**领用人姓名**（去重、按出现次数降序）。
|
||
|
||
给前端下拉用。`consumers` 的含义与 search_outbound_orders 完全一致
|
||
（None=不限 / 空集=空结果），**同样受可见范围约束** —— 下拉里不能出现
|
||
用户本来就看不到的人名，否则等于把范围外的人员信息漏出去。
|
||
可见范围的构造与 search_outbound_orders 保持一致（前缀 ∪ 跨部门例外）。
|
||
"""
|
||
if consumers is not None and not consumers:
|
||
return []
|
||
|
||
params: dict = {"cat_prefix": f"{settings.MATERIAL_CATEGORY_PREFIX}%"}
|
||
if _EXTRA_CONSUMERS:
|
||
company_clause = (
|
||
"(b.category LIKE :cat_prefix OR o.consumer_name = ANY(:extra_consumers))"
|
||
)
|
||
params["extra_consumers"] = _EXTRA_CONSUMERS
|
||
else:
|
||
company_clause = "b.category LIKE :cat_prefix"
|
||
|
||
sql = """
|
||
SELECT o.consumer_name, count(*) AS cnt
|
||
FROM trans_outbound o
|
||
{stock_join}
|
||
WHERE {company_clause}
|
||
AND o.consumer_name IS NOT NULL
|
||
AND o.consumer_name <> ''
|
||
""".format(stock_join=_STOCK_JOIN, company_clause=company_clause)
|
||
if consumers is not None:
|
||
sql += " AND o.consumer_name = ANY(:consumers)"
|
||
params["consumers"] = sorted(consumers)
|
||
sql += " GROUP BY o.consumer_name ORDER BY cnt DESC, o.consumer_name"
|
||
|
||
db = MomSessionLocal()
|
||
try:
|
||
return [r[0] for r in db.execute(text(sql), params).fetchall()]
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def get_orders_by_line_ids(mom_line_ids: list[int]) -> list[dict]:
|
||
"""按明细行 ID 归并出**单据级**信息（同一张单只返回一条）。
|
||
|
||
用户是按整张出库单勾选的，而 product_outbounds 是单据级（一行 = 一张单），
|
||
所以提交上来的明细 ID 要先归并回单据再落库。
|
||
|
||
单据头的字段（出库时间/类型/领用人/操作员/申请单号）同单内必然一致，
|
||
取任意一条即可。
|
||
"""
|
||
merged: dict[str, dict] = {}
|
||
for ln in get_lines_by_ids(mom_line_ids):
|
||
merged.setdefault(ln["outbound_no"], {
|
||
"outbound_no": ln["outbound_no"],
|
||
"outbound_time": ln["outbound_time"],
|
||
"outbound_type": ln["outbound_type"],
|
||
"consumer_name": ln["consumer_name"],
|
||
"operator_name": ln["operator_name"],
|
||
"request_no": ln["request_no"],
|
||
})
|
||
return list(merged.values())
|