3e131fa584
此前入库只校验库位「非空字符串」，任意值都能写进 stock_buy.warehouse_location — 存量数据里那 5 条裸数字（'1' '5' '7' '14' '22'）就是这么来的。 新增 app/utils/warehouse_location.py： - location_error(location) —— 库位必须存在于 sys_warehouse_location - usable_locations(company, xs) —— 批量筛出「存在且本公司可见」的库位 - material_company(base_id) —— 取物料所属公司 ★ 只校验存在，**不校验归属公司** —— 跨公司使用库位不做限制（业务上可能共用）。 若日后要收紧，location_error 里传一次 material_company(base_id) 即可。 空值仍由各接口原有的「必填」校验负责，两件事语义分开、互不干扰。 三个接口（buy / product / semi 的 /submit）各加一行调用。
66 lines
2.6 KiB
Python
66 lines
2.6 KiB
Python
# app/utils/warehouse_location.py
|
||
"""
|
||
库位可用性校验 —— 回填历史库位前的把关。
|
||
|
||
库位带上公司归属之后，库存行里的历史库位字符串可能有两种失效：
|
||
· 指向一个**已被删除**的库位（改名/删除后残留的旧路径）
|
||
· 属于**另一家公司**
|
||
|
||
两种都不该被静默回填进入库表单 —— 操作员会以为那是本公司的库位，
|
||
把货记到一个系统里根本不存在（或不该用）的地址上，而界面上看不出任何异常。
|
||
|
||
口径与 api/v1/warehouse.py 的 _visible_to 一致：本公司 或 全局共享(NULL)。
|
||
"""
|
||
from sqlalchemy import or_
|
||
|
||
from app.extensions import db
|
||
|
||
|
||
def location_error(location):
|
||
"""
|
||
入库写台账前的「库位必须存在」校验。合法返回 None，否则返回给用户的错误消息。
|
||
|
||
★ 只管**存在**，不管归属公司 —— 跨公司使用库位不做限制（业务上可能共用）。
|
||
这一层挡的是历史那种裸数字脏值（'1' / '5' / '7' / '14' / '22'）。
|
||
它们当年能写进 stock_buy，只是因为旧校验仅拦了空串；本次摸底时实测到 5 条。
|
||
|
||
★ 空值不归这里管：各接口自己的「库位必填」校验先跑，语义更清楚。
|
||
"""
|
||
loc = (location or '').strip()
|
||
if not loc:
|
||
return None
|
||
if usable_locations(None, [loc]):
|
||
return None
|
||
return f'库位「{loc}」不存在，请从库位列表中选择'
|
||
|
||
|
||
def material_company(base_id):
|
||
"""取物料所属公司。查不到返回 None —— 此时只做「库位是否存在」的校验。"""
|
||
from app.models.base import MaterialBase
|
||
row = db.session.query(MaterialBase.company_name).filter(
|
||
MaterialBase.id == base_id).first()
|
||
return (row[0] or None) if row else None
|
||
|
||
|
||
def usable_locations(company_name, locations):
|
||
"""
|
||
从 locations 里筛出「仍存在、且该公司可见」的那些，保持传入顺序并去重。
|
||
|
||
company_name 为空时不做公司过滤，但**仍要求库位存在** ——
|
||
「已被删除」这件事跟公司归属无关，任何情况下都不该回填一个不存在的库位。
|
||
"""
|
||
wanted = [x for x in dict.fromkeys(locations) if x]
|
||
if not wanted:
|
||
return []
|
||
|
||
from app.models.system import SysWarehouseLocation
|
||
q = db.session.query(SysWarehouseLocation.full_path).filter(
|
||
SysWarehouseLocation.full_path.in_(wanted))
|
||
if company_name:
|
||
q = q.filter(or_(
|
||
SysWarehouseLocation.company_name == company_name,
|
||
SysWarehouseLocation.company_name.is_(None),
|
||
))
|
||
ok = {r[0] for r in q.all()}
|
||
return [x for x in wanted if x in ok]
|