b4445f07d9
base_service.get_list： 新增 isPurchasing 字段，与待采购池共用同一份活跃单判定。放在预警权限 判断之外 —— 能看到物料列表的人都该知道这个物料已经在采购了。 inventory_task： 静音条件从人工标记 is_ordered 改为「该物料无活跃采购单」。 ★ 为什么必须同时改邮件：预警邮件由出库执行流程触发（outbound_service 出库 完成后的调用），功能是活的；而它的静音入口（前端人工「标记已采购」）随本次 改造一并移除。若不改这里，用户会彻底失去静音能力，每个缺货物料在每次出库后 都会发信。自动判定在建单时静默、单据驳回或强制结案后自动恢复，比人工标记准。 两张表的判定口径都由 utils/purchase_activity.py 提供，不各写一遍。
292 lines
12 KiB
Python
292 lines
12 KiB
Python
"""
|
||
库存预警扫描与邮件通知服务
|
||
|
||
定时（或手动触发）扫描所有 is_enabled=True 且 is_ordered=False 的预警配置，
|
||
按物料配置的邮箱独立发送，不依赖 SysUser 角色。
|
||
|
||
- 库存 <= red_threshold → 红色预警邮件（发 setting.red_emails）
|
||
- red_threshold < 库存 <= yellow_threshold → 黄色预警邮件（发 setting.yellow_emails）
|
||
- 同一收件人在多条记录中出现 → 聚合为一封邮件
|
||
- 发送成功后更新 last_notified_at
|
||
"""
|
||
from datetime import datetime, timezone, timedelta
|
||
from collections import defaultdict
|
||
from sqlalchemy import func
|
||
|
||
from app.extensions import db
|
||
from app.models.base import MaterialBase, MaterialWarningSetting
|
||
from app.models.inbound.buy import StockBuy
|
||
from app.models.inbound.semi import StockSemi
|
||
from app.models.inbound.product import StockProduct
|
||
|
||
|
||
class InventoryWarningService:
|
||
|
||
@staticmethod
|
||
def _prefetch_inventory_map(settings):
|
||
"""
|
||
批量预取所有预警物料的库存总计（单条 SQL，跨三表聚合）
|
||
|
||
性能优化（v2）：
|
||
- 消除 N+1：用 UNION ALL + GROUP BY 替代循环内逐条 scalar 查询
|
||
- 消除额外的 MaterialBase.get()：批量返回 name / spec_model
|
||
|
||
Returns:
|
||
dict[int, dict]: {base_id: {'inv': float, 'avail': float, 'name': str, 'spec': str}, ...}
|
||
"""
|
||
from sqlalchemy import text
|
||
|
||
base_ids = list({s.base_id for s in settings})
|
||
|
||
if not base_ids:
|
||
return {}
|
||
|
||
# ── 单条 SQL：三表 UNION ALL → 外层 GROUP BY ──
|
||
sql = text("""
|
||
SELECT base_id,
|
||
SUM(stock_qty) AS total_stock,
|
||
SUM(avail_qty) AS total_avail
|
||
FROM (
|
||
SELECT base_id, stock_quantity AS stock_qty, available_quantity AS avail_qty
|
||
FROM stock_buy
|
||
WHERE base_id = ANY(:ids)
|
||
|
||
UNION ALL
|
||
|
||
SELECT base_id, stock_quantity, available_quantity
|
||
FROM stock_semi
|
||
WHERE base_id = ANY(:ids)
|
||
|
||
UNION ALL
|
||
|
||
SELECT base_id, stock_quantity, available_quantity
|
||
FROM stock_product
|
||
WHERE base_id = ANY(:ids)
|
||
) AS combined
|
||
GROUP BY base_id
|
||
""")
|
||
rows = db.session.execute(sql, {'ids': base_ids}).fetchall()
|
||
|
||
# 批量查询 MaterialBase
|
||
materials = MaterialBase.query.filter(MaterialBase.id.in_(base_ids)).all()
|
||
mat_map = {m.id: {'name': m.name, 'spec': m.spec_model or ''} for m in materials}
|
||
|
||
result = {}
|
||
for row in rows:
|
||
bid = row.base_id
|
||
mat = mat_map.get(bid, {'name': '', 'spec': ''})
|
||
result[bid] = {
|
||
'inv': float(row.total_stock or 0),
|
||
'avail': float(row.total_avail or 0),
|
||
'name': mat['name'],
|
||
'spec': mat['spec']
|
||
}
|
||
return result
|
||
|
||
@staticmethod
|
||
def _parse_emails(email_str: str) -> list:
|
||
"""从逗号分隔字符串中提取并清洗有效邮箱列表"""
|
||
if not email_str or not email_str.strip():
|
||
return []
|
||
return [e.strip() for e in email_str.split(',') if e.strip() and '@' in e.strip()]
|
||
|
||
@staticmethod
|
||
def _build_text_table(rows: list, level: str) -> str:
|
||
"""
|
||
构建纯文本物料清单表格
|
||
|
||
Args:
|
||
rows: [{"name": ..., "spec": ..., "qty": ..., "threshold": ..., "shortfall": ...}, ...]
|
||
level: "red" 或 "yellow"，决定阈值列标题
|
||
"""
|
||
lines = [
|
||
"名称 | 规格 | 当前库存 | 缺少数量",
|
||
"-" * 55,
|
||
]
|
||
for r in rows:
|
||
name = r.get('name', '-') or '-'
|
||
spec = r.get('spec', '-') or '-'
|
||
qty = r.get('qty', '-')
|
||
shortfall = r.get('shortfall', '-')
|
||
lines.append(f"{name} | {spec} | {qty} | 差{shortfall}个")
|
||
return '\n'.join(lines)
|
||
|
||
@staticmethod
|
||
def check_and_send_warning_emails() -> dict:
|
||
"""
|
||
执行库存预警扫描与邮件发送
|
||
|
||
1. 查询所有 is_enabled=True 且**没有活跃采购单**的预警配置
|
||
2. 按 level 归类物料，按邮箱聚合（同一邮箱 → 一封邮件）
|
||
3. 调用 send_email 发送，更新 last_notified_at
|
||
|
||
★ 静音条件从「人工标记 is_ordered」改成了「是否有活跃采购单」：
|
||
人工标记的前端入口已随「采购在途」自动化改造一并移除，若不改这里，
|
||
用户将彻底失去静音能力（本函数由出库执行流程触发，见
|
||
outbound_service.py 出库完成后的调用）。自动化判定与待采购池、
|
||
物料列表 isPurchasing 共用同一份定义，口径一致。
|
||
|
||
Returns:
|
||
{
|
||
"red_count": N, # 触发红色预警的物料数
|
||
"yellow_count": N, # 触发黄色预警的物料数
|
||
"red_sent": True/False,
|
||
"yellow_sent": True/False,
|
||
"timestamp": "..."
|
||
}
|
||
"""
|
||
from app.utils.email_service import send_email_async
|
||
|
||
beijing_tz = timezone(timedelta(hours=8))
|
||
now = datetime.now(beijing_tz)
|
||
|
||
# 查询启用了预警、且当前没有活跃采购单的配置。
|
||
# 有在途单 → 不必再催采购，静音；单据被驳回或强制结案后自动恢复告警。
|
||
from app.utils.purchase_activity import active_purchase_exists
|
||
|
||
settings = MaterialWarningSetting.query.filter(
|
||
MaterialWarningSetting.is_enabled == True,
|
||
~active_purchase_exists(MaterialWarningSetting.base_id),
|
||
).all()
|
||
|
||
red_rows_by_email = defaultdict(list) # email -> [物料row, ...]
|
||
yellow_rows_by_email = defaultdict(list)
|
||
|
||
total_red = 0
|
||
total_yellow = 0
|
||
total_red_cascaded = 0 # 红色顺延到黄色
|
||
total_yellow_cascaded = 0 # 黄色顺延到红色
|
||
sent_red = False
|
||
sent_yellow = False
|
||
processed_settings = []
|
||
|
||
# ★ 性能优化：批量预取所有 setting 对应的库存 + 物料信息（1 条 SQL 替代 N*4 条）
|
||
inv_map = InventoryWarningService._prefetch_inventory_map(settings)
|
||
|
||
for setting in settings:
|
||
base_id = setting.base_id
|
||
mat_data = inv_map.get(base_id)
|
||
if not mat_data:
|
||
continue
|
||
|
||
name = mat_data['name']
|
||
spec = mat_data['spec']
|
||
inv = mat_data['inv']
|
||
red_th = float(setting.red_threshold) if setting.red_threshold is not None else None
|
||
yellow_th = float(setting.yellow_threshold) if setting.yellow_threshold is not None else None
|
||
|
||
# ★ 红色预警：库存 <= red_threshold，走 setting.red_emails ★
|
||
if red_th is not None and inv <= red_th:
|
||
total_red += 1
|
||
red_emails = InventoryWarningService._parse_emails(setting.red_emails)
|
||
emails_to_use = red_emails
|
||
use_yellow_channel = False # 是否走黄色通道（顺延时为 True）
|
||
if not emails_to_use:
|
||
# ★ 红色预警但无 red_emails，顺延使用 yellow_emails ★
|
||
emails_to_use = InventoryWarningService._parse_emails(setting.yellow_emails)
|
||
if emails_to_use:
|
||
total_yellow += 1
|
||
total_red_cascaded += 1
|
||
use_yellow_channel = True
|
||
print(f"[InventoryWarning] 物料「{name}」红色预警触发，但 red_emails 为空，顺延使用 yellow_emails 发黄色预警")
|
||
|
||
if emails_to_use:
|
||
processed_settings.append(setting)
|
||
row = {
|
||
'name': name,
|
||
'spec': spec,
|
||
'qty': round(inv, 2),
|
||
'threshold': round(red_th, 2),
|
||
'shortfall': round(red_th - inv, 2),
|
||
}
|
||
if use_yellow_channel:
|
||
for email in emails_to_use:
|
||
yellow_rows_by_email[email].append(row)
|
||
else:
|
||
for email in emails_to_use:
|
||
red_rows_by_email[email].append(row)
|
||
else:
|
||
print(f"[InventoryWarning] 物料「{name}」红单跳过：无 red_emails 且 yellow_emails 也为空")
|
||
|
||
# ★ 黄色预警：red_threshold < 库存 <= yellow_threshold，走 setting.yellow_emails ★
|
||
elif (
|
||
(red_th is not None and yellow_th is not None and red_th < inv <= yellow_th)
|
||
or (red_th is None and yellow_th is not None and inv <= yellow_th)
|
||
):
|
||
total_yellow += 1
|
||
yellow_emails = InventoryWarningService._parse_emails(setting.yellow_emails)
|
||
emails_to_use = yellow_emails
|
||
use_red_channel = False # 是否走红色通道（顺延时为 True）
|
||
if not emails_to_use:
|
||
# ★ 黄色预警但无 yellow_emails，顺延使用 red_emails ★
|
||
emails_to_use = InventoryWarningService._parse_emails(setting.red_emails)
|
||
if emails_to_use:
|
||
total_red += 1
|
||
total_yellow_cascaded += 1
|
||
use_red_channel = True
|
||
print(f"[InventoryWarning] 物料「{name}」黄色预警触发，但 yellow_emails 为空，顺延使用 red_emails 发红色预警")
|
||
|
||
if emails_to_use:
|
||
processed_settings.append(setting)
|
||
row = {
|
||
'name': name,
|
||
'spec': spec,
|
||
'qty': round(inv, 2),
|
||
'threshold': round(yellow_th, 2),
|
||
'shortfall': round(yellow_th - inv, 2),
|
||
}
|
||
if use_red_channel:
|
||
for email in emails_to_use:
|
||
red_rows_by_email[email].append(row)
|
||
else:
|
||
for email in emails_to_use:
|
||
yellow_rows_by_email[email].append(row)
|
||
else:
|
||
print(f"[InventoryWarning] 物料「{name}」黄单跳过：无 yellow_emails 且 red_emails 也为空")
|
||
else:
|
||
continue
|
||
|
||
# ★ 按邮箱聚合，批量发送红色预警邮件 ★
|
||
for email, rows in red_rows_by_email.items():
|
||
table = InventoryWarningService._build_text_table(rows, 'red')
|
||
subject = f"【红色预警】库存告急（共 {len(rows)} 条）"
|
||
content = (
|
||
f"您好，\n\n"
|
||
f"以下物料当前库存已达到红色预警阈值，请立即处理采购：\n\n"
|
||
f"{table}\n\n"
|
||
"详情请登录仓库管理系统查看。\n\n"
|
||
"此邮件由系统自动发送，请勿回复。"
|
||
)
|
||
send_email_async(email, subject, content)
|
||
sent_red = True
|
||
|
||
# ★ 按邮箱聚合，批量发送黄色预警邮件 ★
|
||
for email, rows in yellow_rows_by_email.items():
|
||
table = InventoryWarningService._build_text_table(rows, 'yellow')
|
||
subject = f"【黄色预警】库存偏低（共 {len(rows)} 条）"
|
||
content = (
|
||
f"您好，\n\n"
|
||
f"以下物料当前库存已达到黄色预警阈值，请关注采购进度：\n\n"
|
||
f"{table}\n\n"
|
||
"详情请登录仓库管理系统查看。\n\n"
|
||
"此邮件由系统自动发送，请勿回复。"
|
||
)
|
||
send_email_async(email, subject, content)
|
||
sent_yellow = True
|
||
|
||
# ★ 批量更新 last_notified_at ★
|
||
if processed_settings:
|
||
for s in processed_settings:
|
||
s.last_notified_at = now
|
||
db.session.commit()
|
||
|
||
return {
|
||
'red_count': total_red,
|
||
'yellow_count': total_yellow,
|
||
'red_cascaded_count': total_red_cascaded,
|
||
'yellow_cascaded_count': total_yellow_cascaded,
|
||
'red_sent': sent_red,
|
||
'yellow_sent': sent_yellow,
|
||
'timestamp': now.strftime('%Y-%m-%d %H:%M:%S')
|
||
}
|