26a1857acb
修复一个权限旁路：待采购清单的「参考单价」原本后端无条件返回、前端列也没有 任何门控，而 material_list:referencePrice 只授予 SUPER_ADMIN 与 SUPERVISOR。 结果是 WAREHOUSE_MGR(库管) / INBOUND(入库员) / SALES(销售) 虽然都持有 inbound_purchase:pending_pool、能打开待采购清单，就会看到参考价格 —— 而这个 数字他们在物料列表里是被挡住的。等于本页面成了绕过该权限的后门。 参考价格来自 material_base.reference_price，与物料列表同源，因此复用同一个 权限码管控，不另造新码（同源数据用同码，口径才不会漂）。 改动： - 后端 get_pending_purchase_pool 新增 include_reference_price 参数， fail-closed 默认 False；无权限时**整个字段不返回**而不是给 None - 新增 _has_material_reference_price_perm()，判定方式与既有的 _filter_purchase_prices 保持一致（超管/主管放行 + 逐个权限码比对） - 前端「参考单价」列加 hasPermission 门控，TS 类型改为可选 ★ 只挡前端等于没挡（接口仍然裸奔），前后端必须同时改。
786 lines
34 KiB
Python
786 lines
34 KiB
Python
import json
|
||
from datetime import datetime, timezone, timedelta, date
|
||
from sqlalchemy import func
|
||
from app.extensions import db, beijing_time
|
||
from app.models.purchase import PurchaseRequest
|
||
from app.models.base import MaterialBase
|
||
|
||
|
||
class PurchaseService:
|
||
|
||
@staticmethod
|
||
def generate_request_no(batch_seq: int = None):
|
||
"""
|
||
生成采购单号: PUR-yyyyMMdd-HHmm-批次-批内序号
|
||
|
||
Args:
|
||
batch_seq: 今日第几次采购批次（前端提交一批时传入，同一批共享）
|
||
None 时回退为旧格式 PUR-日期-时间-流水
|
||
|
||
返回:
|
||
- 有 batch_seq: PUR-20260831-1021-0001-0001（批次号-批内序号）
|
||
- 无 batch_seq: PUR-20260831-1021-0001（旧格式兼容）
|
||
"""
|
||
beijing_tz = timezone(timedelta(hours=8))
|
||
now = datetime.now(beijing_tz)
|
||
date_str = now.strftime('%Y%m%d')
|
||
time_str = now.strftime('%H%M')
|
||
|
||
if batch_seq is not None:
|
||
# 批次前缀: PUR-日期-时间-批次
|
||
batch_prefix = f"PUR-{date_str}-{time_str}-{batch_seq:04d}"
|
||
# 批内序号: 该批次前缀下的记录数 + 1
|
||
item_count = db.session.query(func.count(func.distinct(PurchaseRequest.request_no))) \
|
||
.filter(PurchaseRequest.request_no.like(f"{batch_prefix}-%")).scalar()
|
||
return f"{batch_prefix}-{(item_count + 1):04d}"
|
||
|
||
# 旧格式兼容: PUR-日期-时间-流水
|
||
prefix = f"PUR-{date_str}-{time_str}-"
|
||
existing_count = db.session.query(func.count(func.distinct(PurchaseRequest.request_no))) \
|
||
.filter(PurchaseRequest.request_no.like(f"{prefix}%")).scalar()
|
||
return f"{prefix}{(existing_count + 1):04d}"
|
||
|
||
@staticmethod
|
||
def get_next_batch_seq():
|
||
"""
|
||
返回今日下一个采购批次号（今日第几次）
|
||
统计今日已有的不同批次（单号第4段）数量，+1
|
||
"""
|
||
beijing_tz = timezone(timedelta(hours=8))
|
||
now = datetime.now(beijing_tz)
|
||
date_str = now.strftime('%Y%m%d')
|
||
time_str = now.strftime('%H%M')
|
||
prefix = f"PUR-{date_str}-{time_str}-"
|
||
|
||
# 查询今日所有单号，提取第4段（批次号）去重
|
||
rows = db.session.query(PurchaseRequest.request_no) \
|
||
.filter(PurchaseRequest.request_no.like(f"{prefix}%")).all()
|
||
batch_seqs = set()
|
||
for (rn,) in rows:
|
||
parts = rn.split('-')
|
||
if len(parts) >= 4:
|
||
batch_seqs.add(parts[3])
|
||
return len(batch_seqs) + 1
|
||
|
||
@staticmethod
|
||
def auto_fill_from_material(keyword: str):
|
||
"""
|
||
根据 name 或 spec_model 自动补全另一个字段
|
||
keyword: 用户输入的名称或规格
|
||
返回: {'name': ..., 'spec_model': ...} 或 None
|
||
"""
|
||
if not keyword:
|
||
return None
|
||
material = MaterialBase.query.filter(
|
||
(MaterialBase.name.ilike(f'%{keyword}%')) |
|
||
(MaterialBase.spec_model.ilike(f'%{keyword}%'))
|
||
).first()
|
||
if material:
|
||
return {
|
||
'name': material.name,
|
||
'spec_model': material.spec_model or ''
|
||
}
|
||
return None
|
||
|
||
@staticmethod
|
||
def create_purchase_request(data: dict, requester_id: int):
|
||
"""
|
||
创建采购申请
|
||
data 包含: name, spec_model, quantity, purchase_date, supplier_link, remark, images,
|
||
unit_price, total_price, approver_id, base_id (可选)
|
||
"""
|
||
batch_seq = data.get('batch_seq')
|
||
request_no = PurchaseService.generate_request_no(batch_seq=batch_seq)
|
||
|
||
purchase_date = data.get('purchase_date')
|
||
if isinstance(purchase_date, str):
|
||
purchase_date = datetime.strptime(purchase_date, '%Y-%m-%d').date()
|
||
elif isinstance(purchase_date, datetime):
|
||
purchase_date = purchase_date.date()
|
||
|
||
# [新增] 自动匹配/关联基础物料
|
||
base_id = data.get('base_id')
|
||
if not base_id and data.get('name'):
|
||
# 尝试通过 name + spec_model 精确匹配 MaterialBase
|
||
material = MaterialBase.query.filter(
|
||
MaterialBase.name == data['name'],
|
||
MaterialBase.spec_model == data.get('spec_model', ''),
|
||
MaterialBase.is_enabled == True
|
||
).first()
|
||
if material:
|
||
base_id = material.id
|
||
|
||
purchase = PurchaseRequest(
|
||
request_no=request_no,
|
||
base_id=base_id, # [新增]
|
||
name=data['name'],
|
||
spec_model=data.get('spec_model', ''),
|
||
quantity=float(data['quantity']),
|
||
purchase_date=purchase_date,
|
||
supplier_link=data.get('supplier_link', ''),
|
||
remark=data.get('remark', ''),
|
||
images=json.dumps(data.get('images', []), ensure_ascii=False) if data.get('images') else '[]',
|
||
unit_price=float(data.get('unit_price', 0) or 0),
|
||
total_price=float(data.get('total_price', 0) or 0),
|
||
tax_rate=float(data.get('tax_rate', 0) or 0),
|
||
requester_id=requester_id,
|
||
approver_id=data.get('approver_id'),
|
||
status=0
|
||
)
|
||
db.session.add(purchase)
|
||
db.session.commit()
|
||
|
||
# 发送邮件给审批人
|
||
PurchaseService._notify_new_request(purchase)
|
||
|
||
return purchase
|
||
|
||
@staticmethod
|
||
def approve_purchase_request(purchase_id: int, user_id: int, action: str, reject_reason: str = None):
|
||
"""
|
||
审批采购申请
|
||
action: 'approve' 或 'reject'
|
||
"""
|
||
purchase = db.session.get(PurchaseRequest, purchase_id)
|
||
if not purchase:
|
||
raise ValueError("采购申请不存在")
|
||
|
||
# ★ approved_at 与 created_at(beijing_time) 同为 naive 本地时间，避免存库时被转成 UTC 早 8 小时
|
||
now = beijing_time()
|
||
|
||
# ★ 完结：库管将「已通过(1)」的申请单置为「已完结(4)」（仿出库审批）
|
||
if action == 'close':
|
||
if purchase.status != 1:
|
||
raise ValueError("仅「已通过」的采购申请可完结")
|
||
purchase.status = 4
|
||
purchase.approver_id = user_id
|
||
purchase.approved_at = now
|
||
db.session.commit()
|
||
return purchase
|
||
|
||
if purchase.status != 0:
|
||
raise ValueError("当前状态不允许审批")
|
||
|
||
if action == 'approve':
|
||
purchase.status = 1
|
||
purchase.approver_id = user_id
|
||
purchase.approved_at = now
|
||
db.session.commit()
|
||
PurchaseService._notify_approved(purchase)
|
||
elif action == 'reject':
|
||
purchase.status = 2
|
||
purchase.approver_id = user_id
|
||
purchase.approved_at = now
|
||
purchase.reject_reason = reject_reason or ''
|
||
db.session.commit()
|
||
PurchaseService._notify_rejected(purchase)
|
||
else:
|
||
raise ValueError("无效的审批操作")
|
||
|
||
return purchase
|
||
|
||
@staticmethod
|
||
def get_purchase_list(page=1, per_page=20, requester_id=None, status=None,
|
||
keyword=None, search_type='all',
|
||
start_date=None, end_date=None):
|
||
"""
|
||
获取采购申请列表，普通用户只看自己的，主管/超管看同公司全部。
|
||
|
||
搜索参数（与出库/报废记录保持同一套语义，便于用户迁移习惯）：
|
||
keyword 关键词
|
||
search_type all / no(单号) / name(物料名称) / spec_model / requester(申请人)
|
||
start_date / end_date 按采购日期过滤（含边界）
|
||
"""
|
||
from app.utils.decorators import get_current_company_filter
|
||
from app.models.system import SysUser
|
||
from sqlalchemy import or_, cast, String
|
||
|
||
query = PurchaseRequest.query
|
||
|
||
if requester_id is not None:
|
||
query = query.filter(PurchaseRequest.requester_id == requester_id)
|
||
|
||
if status is not None:
|
||
query = query.filter(PurchaseRequest.status == status)
|
||
|
||
# ---- 关键词搜索 ----
|
||
#
|
||
# 申请人姓名存在 SysUser.username，格式为「姓名/账号」（如 韩善龙/hanshanlong），
|
||
# 因此按「姓名」搜索时用 ilike 直接匹配整串即可命中。
|
||
# 该类字段只存在于 SysUser，故需要 join；单号/名称/规格则不需要 ——
|
||
# 只在必要时 join，避免无谓的联表。
|
||
needs_user = bool(keyword) and search_type in ('requester', 'all')
|
||
if needs_user:
|
||
query = query.outerjoin(SysUser, PurchaseRequest.requester_id == SysUser.id)
|
||
|
||
if keyword:
|
||
kw = f'%{keyword}%'
|
||
if search_type == 'no':
|
||
query = query.filter(PurchaseRequest.request_no.ilike(kw))
|
||
elif search_type == 'name':
|
||
query = query.filter(PurchaseRequest.name.ilike(kw))
|
||
elif search_type == 'spec_model':
|
||
query = query.filter(PurchaseRequest.spec_model.ilike(kw))
|
||
elif search_type == 'requester':
|
||
query = query.filter(SysUser.username.ilike(kw))
|
||
else: # all —— 覆盖单号 / 名称 / 规格 / 备注 / 申请人
|
||
query = query.filter(or_(
|
||
PurchaseRequest.request_no.ilike(kw),
|
||
PurchaseRequest.name.ilike(kw),
|
||
PurchaseRequest.spec_model.ilike(kw),
|
||
PurchaseRequest.remark.ilike(kw),
|
||
SysUser.username.ilike(kw),
|
||
))
|
||
|
||
# ---- 日期范围（按采购日期）----
|
||
if start_date:
|
||
query = query.filter(PurchaseRequest.purchase_date >= start_date)
|
||
if end_date:
|
||
query = query.filter(PurchaseRequest.purchase_date <= end_date)
|
||
|
||
# 【行级数据隔离】同公司可见：匹配 MaterialBase.company_name 或 SysUser.department
|
||
company_limit = get_current_company_filter()
|
||
if company_limit is not None:
|
||
query = query.outerjoin(MaterialBase, PurchaseRequest.base_id == MaterialBase.id)
|
||
# SysUser 可能已被上面的关键词分支 join 过 —— 重复 join 会产生
|
||
# 笛卡尔积导致结果翻倍，此处按需补 join。
|
||
if not needs_user:
|
||
query = query.outerjoin(SysUser, PurchaseRequest.requester_id == SysUser.id)
|
||
query = query.filter(or_(
|
||
MaterialBase.company_name == company_limit,
|
||
SysUser.department == company_limit
|
||
))
|
||
|
||
# ★ 兜底去重：上述 join 在特定组合下（如 base_id 为空 + requester 匹配）
|
||
# 仍可能产生重复行，用 distinct 保证每个单据只出现一次。
|
||
query = query.distinct()
|
||
|
||
query = query.order_by(PurchaseRequest.created_at.desc())
|
||
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
|
||
|
||
return {
|
||
'items': [p.to_dict() for p in pagination.items],
|
||
'total': pagination.total,
|
||
'pages': pagination.pages,
|
||
'current_page': page
|
||
}
|
||
|
||
@staticmethod
|
||
def get_purchase_by_id(purchase_id: int):
|
||
purchase = db.session.get(PurchaseRequest, purchase_id)
|
||
return purchase.to_dict() if purchase else None
|
||
|
||
@staticmethod
|
||
def get_approved_requests(page=1, per_page=20, keyword=None):
|
||
"""
|
||
获取已审批通过且未入库的采购申请列表（专供库管按单入库使用）
|
||
|
||
筛选条件:
|
||
- status == 1（已审批通过）
|
||
- 尚未被任何 StockBuy 关联（request_id 未被引用）
|
||
|
||
返回字段包含: 采购申请信息 + MaterialBase 基础物料信息
|
||
"""
|
||
from app.models.inbound.buy import StockBuy
|
||
|
||
# 子查询：所有已被入库引用的 request_id（去重）
|
||
stocked_ids = db.session.query(StockBuy.request_id).filter(
|
||
StockBuy.request_id.isnot(None)
|
||
).distinct().subquery()
|
||
|
||
# 主查询：已通过 且 不在已入库集合中
|
||
query = db.session.query(PurchaseRequest).filter(
|
||
PurchaseRequest.status == 1
|
||
).filter(
|
||
~PurchaseRequest.id.in_(stocked_ids)
|
||
)
|
||
|
||
# 可选关键词搜索：采购单号 / 名称 / 规格
|
||
if keyword:
|
||
k = f'%{keyword.strip()}%'
|
||
query = query.filter(
|
||
PurchaseRequest.request_no.ilike(k) |
|
||
PurchaseRequest.name.ilike(k) |
|
||
PurchaseRequest.spec_model.ilike(k)
|
||
)
|
||
|
||
query = query.order_by(PurchaseRequest.approved_at.desc().nullslast(),
|
||
PurchaseRequest.created_at.desc())
|
||
|
||
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
|
||
|
||
# ★ 批量预取 MaterialBase：三段式查询（base_id 精确 → name 精确 → name 模糊）
|
||
purchase_items = list(pagination.items)
|
||
base_ids = {p.base_id for p in purchase_items if p.base_id}
|
||
names_no_base = [p.name for p in purchase_items if p.name and not p.base_id]
|
||
|
||
material_map = {} # base_id → MaterialBase
|
||
name_map = {} # (name, spec) or (name, '__fallback__') → MaterialBase
|
||
|
||
# 阶段1: base_id 精确预取
|
||
if base_ids:
|
||
materials = MaterialBase.query.filter(MaterialBase.id.in_(base_ids)).all()
|
||
material_map = {m.id: m for m in materials}
|
||
|
||
# 阶段2: name 精确预取（name.in_）
|
||
if names_no_base:
|
||
exact_matches = MaterialBase.query.filter(
|
||
MaterialBase.name.in_(names_no_base),
|
||
MaterialBase.is_enabled == True
|
||
).all()
|
||
for m in exact_matches:
|
||
name_map[(m.name, m.spec_model or '')] = m
|
||
|
||
# 阶段3: name 模糊预取（仅>=2字符 + LIMIT 500 防爆炸）
|
||
unmatched_names = []
|
||
for p in purchase_items:
|
||
if p.base_id and p.base_id in material_map:
|
||
continue
|
||
if p.name and (p.name, p.spec_model or '') in name_map:
|
||
continue
|
||
n = (p.name or '').strip()
|
||
if len(n) >= 2: # ★ 安全阈值：至少2字符才做模糊匹配
|
||
unmatched_names.append(n)
|
||
|
||
if unmatched_names:
|
||
fuzzy_conditions = [
|
||
MaterialBase.name.ilike(f'%{n}%') for n in unmatched_names
|
||
]
|
||
fuzzy_matches = MaterialBase.query.filter(
|
||
db.or_(*fuzzy_conditions),
|
||
MaterialBase.is_enabled == True
|
||
).order_by(MaterialBase.id.desc()).limit(500).all() # ★ 硬上限：最多500条
|
||
for m in fuzzy_matches:
|
||
key = (m.name, '__fuzzy__')
|
||
if key not in name_map:
|
||
name_map[key] = m
|
||
|
||
# 内存匹配（O(1) 字典查找）
|
||
items = []
|
||
for p in purchase_items:
|
||
item = p.to_dict()
|
||
material = None
|
||
|
||
# 1. base_id 精确
|
||
if p.base_id:
|
||
material = material_map.get(p.base_id)
|
||
|
||
# 2. name + spec 精确
|
||
if not material and p.name:
|
||
material = name_map.get((p.name, p.spec_model or ''))
|
||
|
||
# 3. name 模糊回退
|
||
if not material and p.name:
|
||
material = name_map.get((p.name, '__fuzzy__'))
|
||
|
||
if material:
|
||
item['material'] = {
|
||
'id': material.id,
|
||
'company_name': material.company_name or '',
|
||
'name': material.name,
|
||
'spec_model': material.spec_model or '',
|
||
'category': material.category or '',
|
||
'unit': material.unit or '',
|
||
'type': material.material_type or '',
|
||
'is_inspection_required': bool(material.is_inspection_required),
|
||
}
|
||
else:
|
||
item['material'] = None
|
||
items.append(item)
|
||
|
||
return {
|
||
'items': items,
|
||
'total': pagination.total,
|
||
'pages': pagination.pages,
|
||
'current_page': page
|
||
}
|
||
|
||
@staticmethod
|
||
def search_base_material(keyword: str, page: int = 1, limit: int = 20):
|
||
"""
|
||
物料基础信息搜索，支持 name/spec_model/company_name 模糊匹配，返回分页结果
|
||
用于采购申请弹窗的物料远程搜索
|
||
"""
|
||
from sqlalchemy import and_, or_
|
||
|
||
query = MaterialBase.query.filter(MaterialBase.is_enabled == True)
|
||
|
||
if keyword:
|
||
k = keyword.strip()
|
||
k_str = f'%{k}%'
|
||
query = query.filter(or_(
|
||
MaterialBase.name.ilike(k_str),
|
||
MaterialBase.spec_model.ilike(k_str),
|
||
MaterialBase.company_name.ilike(k_str)
|
||
))
|
||
|
||
query = query.order_by(MaterialBase.id.desc())
|
||
pagination = query.paginate(page=page, per_page=limit, error_out=False)
|
||
|
||
items = []
|
||
for item in pagination.items:
|
||
items.append({
|
||
'id': item.id,
|
||
'company_name': item.company_name,
|
||
'name': item.name,
|
||
'spec_model': item.spec_model,
|
||
'category': item.category,
|
||
'unit': item.unit,
|
||
'type': item.material_type,
|
||
'pinyin': getattr(item, 'pinyin', ''),
|
||
'status': '启用'
|
||
})
|
||
|
||
return {
|
||
'items': items,
|
||
'total': pagination.total,
|
||
'page': page,
|
||
'has_next': pagination.has_next
|
||
}
|
||
|
||
@staticmethod
|
||
def _notify_new_request(purchase):
|
||
"""发送新申请邮件给审批人"""
|
||
try:
|
||
from app.utils.email_service import send_email_async
|
||
from app.models.system import SysUser
|
||
|
||
if not purchase.approver_id:
|
||
return
|
||
|
||
approver = db.session.get(SysUser, purchase.approver_id)
|
||
if not approver or not approver.email:
|
||
return
|
||
|
||
subject = f"【待审批】采购申请单 {purchase.request_no}"
|
||
content = f"""您好，
|
||
|
||
您有一笔新的采购申请待审批：
|
||
|
||
申请单号：{purchase.request_no}
|
||
采购物品：{purchase.name}
|
||
规格型号：{purchase.spec_model or '-'}
|
||
采购数量：{float(purchase.quantity)}
|
||
申请时间：{purchase.created_at.strftime('%Y-%m-%d %H:%M') if purchase.created_at else '-'}
|
||
备注说明：{purchase.remark or '无'}
|
||
|
||
请登录仓库管理系统进行审批。
|
||
|
||
此邮件由系统自动发送，请勿回复。
|
||
"""
|
||
send_email_async(approver.email, subject, content)
|
||
except Exception as e:
|
||
try:
|
||
from flask import current_app
|
||
current_app.logger.error(f"[Email] 采购申请通知审批人失败: {e}")
|
||
except Exception:
|
||
print(f"[Email] 采购申请通知审批人失败: {e}")
|
||
|
||
@staticmethod
|
||
def _notify_approved(purchase):
|
||
"""审批通过后通知申请人"""
|
||
try:
|
||
from app.utils.email_service import send_email_async
|
||
from app.models.system import SysUser
|
||
|
||
requester = db.session.get(SysUser, purchase.requester_id)
|
||
if not requester or not requester.email:
|
||
return
|
||
|
||
subject = f"【已通过】采购申请单 {purchase.request_no}"
|
||
content = f"""{"尊敬的 " + requester.username + "，您好" if requester.username else "您好"}，
|
||
|
||
您的采购申请单 {purchase.request_no}（{purchase.name}）已审批通过，现已交给库管。
|
||
|
||
待库管完成入库后，您可在系统中查询采购记录。
|
||
|
||
此邮件由系统自动发送，请勿回复。
|
||
"""
|
||
send_email_async(requester.email, subject, content)
|
||
except Exception as e:
|
||
try:
|
||
from flask import current_app
|
||
current_app.logger.error(f"[Email] 采购申请通过通知申请人失败: {e}")
|
||
except Exception:
|
||
print(f"[Email] 采购申请通过通知申请人失败: {e}")
|
||
|
||
@staticmethod
|
||
def _notify_rejected(purchase):
|
||
"""审批驳回后通知申请人"""
|
||
try:
|
||
from app.utils.email_service import send_email_async
|
||
from app.models.system import SysUser
|
||
|
||
requester = db.session.get(SysUser, purchase.requester_id)
|
||
if not requester or not requester.email:
|
||
return
|
||
|
||
subject = f"【已驳回】采购申请单 {purchase.request_no}"
|
||
content = f"""{"尊敬的 " + requester.username + "，您好" if requester.username else "您好"}，
|
||
|
||
您的采购申请单 {purchase.request_no}（{purchase.name}）已被驳回。
|
||
|
||
驳回原因：{purchase.reject_reason or '未说明'}
|
||
|
||
请登录仓库管理系统查看详情。
|
||
|
||
此邮件由系统自动发送，请勿回复。
|
||
"""
|
||
send_email_async(requester.email, subject, content)
|
||
except Exception as e:
|
||
print(f"[Email] 采购申请驳回通知失败: {e}")
|
||
|
||
# ============================================================
|
||
# 待采购池（防重复采购）
|
||
# ============================================================
|
||
@staticmethod
|
||
def get_pending_purchase_pool(page=1, per_page=20, keyword=None,
|
||
category=None, material_type=None,
|
||
warning_status=None,
|
||
include_reference_price=False):
|
||
"""
|
||
待采购池：找出「**有效供给**仍不足、需要再买」的物料。
|
||
|
||
业务目的：杜绝多名采购员对同一短缺物料重复购买，同时**不能漏掉缺口**。
|
||
|
||
核心是有效供给：
|
||
|
||
有效供给 = 当前物理总库存 + 在途量
|
||
在途量 = SUM(该物料所有活跃采购单的剩余待入库量)
|
||
= SUM(GREATEST(采购量 - 累计入库量, 0))
|
||
|
||
过滤规则：有效供给 <= 红/黄阈值 才留在池中。
|
||
|
||
★ 为什么不是「有采购单就排除」（本接口的第一版）：
|
||
红线 10、库存 0、某采购员只建了一张数量 5 的单 —— 若按 EXISTS 一刀切，
|
||
该物料会**立刻从池中消失**，剩下 5 个缺口永远无人认领。这是掩耳盗铃。
|
||
按量计算后它继续留在池中，suggested_qty 自动降到 5，提醒下一位只补 5 个。
|
||
|
||
★ 判定口径必须与物料列表预警（inbound/base_service.get_list）严格一致，
|
||
否则会出现「列表亮红灯、池子却说不用买」的撕裂：
|
||
- 用**物理库存总量**（三个 stock_* 表的 stock_quantity 之和），不是可用量
|
||
- 阈值可能为 None，红黄两个阈值**各自独立**判断，不能 coalesce 成一个
|
||
- 用 `<=` 而非 `<` —— 恰好等于阈值即算不足
|
||
|
||
★ 参考价格（reference_price）默认**不返回**（fail-closed）。
|
||
它是 material_base.reference_price，属于受 material_list:referencePrice
|
||
管控的价格数据 —— 与物料列表同源，就必须同码管控，否则这里会变成
|
||
绕过该权限的后门（物料列表看不到、待采购清单却看得到）。
|
||
调用方（API 层）按权限显式传 include_reference_price=True。
|
||
|
||
★ 关于 `<=`（2026-09-18 与业务方确认，**刻意维持，勿擅自改成 `<`**）：
|
||
阈值是「警戒下限」，到线即需关注；且物料列表预警、预警邮件用的是同一套
|
||
`<=` 口径（那是系统原有约定，不是本接口另立的）。若只把本接口改成 `<`，
|
||
会出现「列表亮黄灯、池子却把它排除」的撕裂。
|
||
代价是有效供给恰好等于阈值时 suggested_qty 会落到最小值 1（看似
|
||
「不用买却建议买 1 个」）—— 这是有意的缓冲，前端 tooltip 已单独说明。
|
||
真要改成 `<`，必须**同时**改 base_service 的判级与 inventory_task 的
|
||
触发条件，三处一起动。
|
||
"""
|
||
import math
|
||
|
||
from sqlalchemy import and_, case, cast, desc, or_
|
||
from sqlalchemy.types import Numeric as SqlNumeric
|
||
|
||
from app.models.base import MaterialWarningSetting
|
||
from app.models.inbound.buy import StockBuy
|
||
from app.models.inbound.product import StockProduct
|
||
from app.models.inbound.semi import StockSemi
|
||
from app.utils.decorators import get_current_company_filter
|
||
from app.utils.purchase_activity import in_transit_subquery
|
||
|
||
# ---- 三表库存聚合（口径同 base_service.get_list:146-170）----
|
||
buy_sub = db.session.query(
|
||
StockBuy.base_id,
|
||
func.sum(StockBuy.stock_quantity).label('buy_inv'),
|
||
func.sum(StockBuy.available_quantity).label('buy_avail')
|
||
).group_by(StockBuy.base_id).subquery()
|
||
|
||
semi_sub = db.session.query(
|
||
StockSemi.base_id,
|
||
func.sum(StockSemi.stock_quantity).label('semi_inv'),
|
||
func.sum(StockSemi.available_quantity).label('semi_avail')
|
||
).group_by(StockSemi.base_id).subquery()
|
||
|
||
prod_sub = db.session.query(
|
||
StockProduct.base_id,
|
||
func.sum(StockProduct.stock_quantity).label('prod_inv'),
|
||
func.sum(StockProduct.available_quantity).label('prod_avail')
|
||
).group_by(StockProduct.base_id).subquery()
|
||
|
||
total_inv = (func.coalesce(buy_sub.c.buy_inv, 0)
|
||
+ func.coalesce(semi_sub.c.semi_inv, 0)
|
||
+ func.coalesce(prod_sub.c.prod_inv, 0))
|
||
total_avail = (func.coalesce(buy_sub.c.buy_avail, 0)
|
||
+ func.coalesce(semi_sub.c.semi_avail, 0)
|
||
+ func.coalesce(prod_sub.c.prod_avail, 0))
|
||
|
||
# ★ 必须先物化再过滤：total_inv 是聚合表达式，直接在带 GROUP BY 的查询里
|
||
# 参与比较会落到 HAVING 语义；物化成子查询后，它在外层就是普通列，
|
||
# 可以安全地进 WHERE（与 base_service.get_list:175-191 同一手法）。
|
||
inner_sub = (
|
||
db.session.query(
|
||
MaterialBase.id.label('base_id'),
|
||
total_inv.label('total_inv'),
|
||
total_avail.label('total_avail'),
|
||
)
|
||
.outerjoin(buy_sub, MaterialBase.id == buy_sub.c.base_id)
|
||
.outerjoin(semi_sub, MaterialBase.id == semi_sub.c.base_id)
|
||
.outerjoin(prod_sub, MaterialBase.id == prod_sub.c.base_id)
|
||
.subquery()
|
||
)
|
||
|
||
# ---- 在途量聚合（每个物料一张单一张单地累加剩余待入库量）----
|
||
intransit_sub = in_transit_subquery()
|
||
|
||
inv_col = inner_sub.c.total_inv
|
||
avail_col = inner_sub.c.total_avail
|
||
# 没有活跃采购单的物料不在 intransit_sub 里，outerjoin 后为 NULL → 兜成 0
|
||
intransit_col = func.coalesce(intransit_sub.c.intransit, 0)
|
||
# ★ 有效供给：判缺货、算建议量，全部基于它，而不是光看库存
|
||
effective_col = inv_col + intransit_col
|
||
red_col = cast(MaterialWarningSetting.red_threshold, SqlNumeric)
|
||
yellow_col = cast(MaterialWarningSetting.yellow_threshold, SqlNumeric)
|
||
|
||
# inner join 预警配置：缺货的前提就是预警已启用，outer join 无意义。
|
||
# （material_warning_settings 按约定与物料 1:1 且已核实无重复行，
|
||
# 故不会因 join 放大行数。若将来出现重复行，此处需要改为每物料取一条。）
|
||
query = (
|
||
db.session.query(MaterialBase, MaterialWarningSetting,
|
||
inv_col, avail_col, intransit_col)
|
||
.join(inner_sub, MaterialBase.id == inner_sub.c.base_id)
|
||
.outerjoin(intransit_sub, MaterialBase.id == intransit_sub.c.base_id)
|
||
.join(MaterialWarningSetting, MaterialBase.id == MaterialWarningSetting.base_id)
|
||
.filter(MaterialWarningSetting.is_enabled.is_(True))
|
||
)
|
||
|
||
# ---- ① 有效供给不足 ----
|
||
# 对应 Python 侧的 `if 供给<=红 ... elif 供给<=黄`：命中任一即不足。
|
||
# ★ 绝不写成 `supply <= coalesce(red, yellow)` —— 那会把「红阈值未配置」
|
||
# 偷换成「用黄阈值」，且在两者都为 NULL 时静默退化成 NULL 比较。
|
||
red_hit = and_(MaterialWarningSetting.red_threshold.isnot(None), effective_col <= red_col)
|
||
yellow_hit = and_(MaterialWarningSetting.yellow_threshold.isnot(None), effective_col <= yellow_col)
|
||
query = query.filter(or_(red_hit, yellow_hit))
|
||
|
||
# ---- 行级数据隔离（多租户）----
|
||
company_limit = get_current_company_filter()
|
||
if company_limit is not None:
|
||
query = query.filter(MaterialBase.company_name == company_limit)
|
||
|
||
# ---- 可选筛选 ----
|
||
if keyword:
|
||
kw = f'%{keyword.strip()}%'
|
||
query = query.filter(or_(
|
||
MaterialBase.name.ilike(kw),
|
||
MaterialBase.spec_model.ilike(kw),
|
||
MaterialBase.company_name.ilike(kw),
|
||
))
|
||
|
||
if category:
|
||
query = query.filter(MaterialBase.category.ilike(f"{category.strip()}%"))
|
||
|
||
if material_type:
|
||
query = query.filter(MaterialBase.material_type.ilike(material_type.strip()))
|
||
|
||
if warning_status == 2:
|
||
query = query.filter(red_hit)
|
||
elif warning_status == 1:
|
||
# 黄色要在红之外，否则 1/2 语义重叠
|
||
query = query.filter(yellow_hit, ~red_hit)
|
||
|
||
# ---- 排序：最缺的排最上面（复刻 base_service.get_list:372-391 的口径）----
|
||
warning_level = case(
|
||
(red_hit, 2),
|
||
(yellow_hit, 1),
|
||
else_=0,
|
||
)
|
||
# 缺口按**有效供给**算，在途已经补上的部分不再计入缺口
|
||
gap = case(
|
||
(red_hit, red_col - effective_col),
|
||
(yellow_hit, yellow_col - effective_col),
|
||
else_=0,
|
||
)
|
||
query = query.order_by(desc(warning_level), desc(gap), MaterialBase.id.asc())
|
||
|
||
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
|
||
|
||
items = []
|
||
for row in pagination.items:
|
||
material = row[0]
|
||
setting = row[1]
|
||
inv = float(row[2]) if row[2] is not None else 0.0
|
||
avail = float(row[3]) if row[3] is not None else 0.0
|
||
intransit = float(row[4]) if row[4] is not None else 0.0
|
||
effective = inv + intransit
|
||
|
||
red = float(setting.red_threshold) if setting.red_threshold is not None else None
|
||
yellow = float(setting.yellow_threshold) if setting.yellow_threshold is not None else None
|
||
|
||
# 判级用有效供给 —— 与上面的 SQL 过滤条件保持同一口径
|
||
status = 0
|
||
if red is not None and effective <= red:
|
||
status = 2
|
||
elif yellow is not None and effective <= yellow:
|
||
status = 1
|
||
|
||
# 建议采购量 = 阈值 - 有效供给，一次性把供给抬出**整个**预警带，
|
||
# 否则刚补完货下一条预警马上又来。
|
||
# 取 max(红,黄) 而非「被触发的那个」——正常配置下红<黄，补到黄线即可
|
||
# 完全退出；红黄配反的脏数据下也安全。
|
||
# 双阈值都为 None 时降级为 None，不会出现 None - effective 的 TypeError。
|
||
#
|
||
# ★ 注意在途已计入 effective，所以「已买 5」时这里只会建议剩下的 5，
|
||
# 而不是按库存从零算。这正是本次改造要修的那个漏洞。
|
||
targets = [t for t in (red, yellow) if t is not None]
|
||
target = max(targets) if targets else None
|
||
suggested = max(1, math.ceil(target - effective)) if target is not None else None
|
||
|
||
items.append({
|
||
'material_id': material.id,
|
||
'name': material.name,
|
||
'spec_model': material.spec_model or '',
|
||
'unit': material.unit or '',
|
||
'category': material.category or '',
|
||
'material_type': material.material_type or '',
|
||
'company_name': material.company_name or '',
|
||
'image': PurchaseService._first_image(material.product_image),
|
||
'purchase_link': material.purchase_link or '',
|
||
# fail-closed：无权限时整个字段不出现，而不是给 None ——
|
||
# 前端据字段是否存在决定列是否渲染，语义更干净
|
||
**({'reference_price': float(material.reference_price)
|
||
if material.reference_price is not None else None}
|
||
if include_reference_price else {}),
|
||
'inventory_count': inv,
|
||
'available_count': avail,
|
||
# 在途量与有效供给一并返回，前端才能向采购员解释「为什么只建议买这么多」
|
||
'in_transit_qty': intransit,
|
||
'effective_supply': effective,
|
||
'warning_status': status,
|
||
'warning_red': red,
|
||
'warning_yellow': yellow,
|
||
'target_threshold': target,
|
||
'suggested_qty': suggested,
|
||
'is_ordered': bool(setting.is_ordered),
|
||
})
|
||
|
||
return {
|
||
'items': items,
|
||
'total': pagination.total,
|
||
'pages': pagination.pages,
|
||
'current_page': page,
|
||
}
|
||
|
||
@staticmethod
|
||
def _first_image(product_image):
|
||
"""从 material_base.product_image 的 JSON 字符串里取第一张图，取不到返回空串。"""
|
||
if not product_image:
|
||
return ''
|
||
try:
|
||
if isinstance(product_image, str) and not product_image.startswith('['):
|
||
return product_image # 兼容旧数据：单条 URL 直接存字符串
|
||
parsed = json.loads(product_image)
|
||
if isinstance(parsed, list) and parsed:
|
||
return parsed[0] or ''
|
||
except Exception:
|
||
return ''
|
||
return '' |