393 lines
20 KiB
Python
393 lines
20 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""AI分析P1：确定性经营异常扫描与角色化洞察。
|
||
|
||
异常由后端规则计算，LLM只负责后续报告归纳，避免把关键数字交给模型臆测。
|
||
"""
|
||
from collections import defaultdict
|
||
from datetime import date, datetime
|
||
import hashlib
|
||
import re
|
||
|
||
from services.ai_datasources import AI_DATASOURCES, datasource_meta
|
||
from services.simple_cache import get as cache_get, set as cache_set, delete as cache_delete
|
||
|
||
CACHE_KEY = 'ai_p1_operational_insights_v1'
|
||
CACHE_TTL = 300
|
||
|
||
|
||
def _num(value, default=0.0):
|
||
try:
|
||
return float(value or 0)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def _int(value):
|
||
return int(round(_num(value)))
|
||
|
||
|
||
def _parse_date(value):
|
||
if isinstance(value, (datetime, date)):
|
||
return value.date() if isinstance(value, datetime) else value
|
||
text = str(value or '').strip()
|
||
if not text:
|
||
return None
|
||
# 订单历史数据包含 YYYY-MM-DD、YYYY-MM-DD HH:mm、YYYY/M/D 等形式。
|
||
match = re.search(r'(20\d{2})[-/年](\d{1,2})[-/月](\d{1,2})', text)
|
||
if not match:
|
||
return None
|
||
try:
|
||
return date(int(match.group(1)), int(match.group(2)), int(match.group(3)))
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _severity_rank(level):
|
||
return {'高': 3, '中': 2, '低': 1, '信息': 0}.get(level, 0)
|
||
|
||
|
||
def _stable_id(kind, *parts):
|
||
raw = '|'.join([kind] + [str(x or '') for x in parts])
|
||
return kind + '-' + hashlib.md5(raw.encode('utf-8')).hexdigest()[:12]
|
||
|
||
|
||
def _insight(kind, category, severity, title, detail, question, source_keys,
|
||
product_code='', supplier='', metrics=None, source_page=''):
|
||
return {
|
||
'id': _stable_id(kind, product_code, supplier, title),
|
||
'type': kind,
|
||
'category': category,
|
||
'severity': severity,
|
||
'title': title,
|
||
'detail': detail,
|
||
'question': question,
|
||
'source_keys': source_keys,
|
||
'product_code': product_code,
|
||
'supplier': supplier,
|
||
'metrics': metrics or {},
|
||
'source_page': source_page,
|
||
}
|
||
|
||
|
||
def _fetch(key):
|
||
ds = AI_DATASOURCES[key]
|
||
result = ds['handler']({'_no_limit': True}) or {}
|
||
return result.get('rows') or []
|
||
|
||
|
||
def _build_raw_insights():
|
||
planning = _fetch('planning_metrics')
|
||
profiles = _fetch('product_profile')
|
||
sales = _fetch('product_sales')
|
||
orders = _fetch('order_detail')
|
||
suppliers = _fetch('supplier_delivery')
|
||
bom = _fetch('bom_supply_chain')
|
||
values = _fetch('inventory_value')
|
||
|
||
profile_map = {str(r.get('product_code') or ''): r for r in profiles}
|
||
sales_map = {str(r.get('product_code') or ''): r for r in sales}
|
||
planning_map = {str(r.get('product_code') or ''): r for r in planning}
|
||
value_map = {str(r.get('product_code') or ''): r for r in values}
|
||
today = date.today()
|
||
items = []
|
||
|
||
# 1. 缺货/补货风险、积压与销量异常。
|
||
for row in planning:
|
||
code = str(row.get('product_code') or '')
|
||
if not code:
|
||
continue
|
||
profile = profile_map.get(code) or {}
|
||
sale = sales_map.get(code) or {}
|
||
daily = _num(row.get('日销量预测'))
|
||
support = _num(row.get('总支撑天数'), 999)
|
||
inventory_support = _num(row.get('库存可支撑天数'), 999)
|
||
suggested = _int(row.get('建议下单数量'))
|
||
inventory = _int(row.get('库存')) + _int(row.get('加工厂库存'))
|
||
lead = _int(profile.get('生产货期'))
|
||
logistics = _int(profile.get('物流时间'))
|
||
product_type = str(profile.get('产品类型') or '')
|
||
product_name = str(profile.get('货品名称') or '')
|
||
# 物料/内料/标签等非销售SKU不进入成品销量与积压预警，避免历史分类不规范造成噪音。
|
||
material_terms = ('【内料】', '【瓶子】', '【瓶盖】', '【泵头】', '【彩盒】', '【标签】', '防伪码标签')
|
||
if product_type in ('包材', '原材料', '其他') or any(term in product_name for term in material_terms):
|
||
continue
|
||
threshold = lead + logistics + (0 if product_type == '现货' else 7)
|
||
avg7 = _num(sale.get('7天平均销量'))
|
||
avg30 = _num(sale.get('30天平均销量'))
|
||
|
||
if daily > 0 and (support < threshold or suggested > 0):
|
||
severity = '高' if support < max(lead + logistics, 1) else '中'
|
||
items.append(_insight(
|
||
'stockout', '缺货与补货', severity,
|
||
f'{code} 存在补货风险',
|
||
f'总支撑天数{support:.1f}天，补货阈值{threshold}天，建议下单{suggested}件，日销量预测{daily:g}件。',
|
||
f'分析{code}为什么需要下单，并检查库存、在途和BOM包材是否足够',
|
||
['planning_metrics', 'product_profile'],
|
||
product_code=code,
|
||
metrics={'总支撑天数': support, '补货阈值': threshold, '建议下单数量': suggested,
|
||
'日销量预测': daily}, source_page='/planning'))
|
||
|
||
if daily <= 0 and inventory >= 1000:
|
||
severity = '高' if inventory >= 10000 else '中'
|
||
items.append(_insight(
|
||
'overstock', '库存健康', severity,
|
||
f'{code} 无销量但仍有库存',
|
||
f'日销量预测为0，当前库存及加工厂库存合计{inventory}件，存在滞销或库存积压风险。',
|
||
f'分析{code}为什么近期无销量，库存应该如何处理',
|
||
['planning_metrics'],
|
||
product_code=code, metrics={'库存合计': inventory, '日销量预测': daily},
|
||
source_page='/planning'))
|
||
elif daily > 0 and support >= 120 and inventory >= daily * 60:
|
||
value = value_map.get(code) or {}
|
||
severity = '高' if support >= 180 else '中'
|
||
items.append(_insight(
|
||
'overstock', '库存健康', severity,
|
||
f'{code} 支撑天数偏高',
|
||
f'总支撑天数{support:.1f}天，库存及加工厂库存合计{inventory}件，明显高于近期销售需要。',
|
||
f'分析{code}库存积压原因和去库存建议',
|
||
['planning_metrics'],
|
||
product_code=code,
|
||
metrics={'总支撑天数': support, '库存合计': inventory,
|
||
}, source_page='/planning'))
|
||
|
||
if avg30 >= 3 and avg7 <= avg30 * 0.6:
|
||
decline = round((1 - avg7 / avg30) * 100, 1) if avg30 else 0
|
||
items.append(_insight(
|
||
'sales_decline', '销量异常', '高' if decline >= 60 else '中',
|
||
f'{code} 近7天销量明显下降',
|
||
f'7天平均销量{avg7:g}件，30天平均销量{avg30:g}件，下降约{decline}%。',
|
||
f'分析{code}近7天销量下降的原因及库存风险',
|
||
['product_sales'],
|
||
product_code=code, metrics={'7天平均销量': avg7, '30天平均销量': avg30, '下降幅度': decline},
|
||
source_page='/outbound'))
|
||
elif avg30 >= 1 and avg7 >= max(avg30 * 1.5, avg30 + 3):
|
||
growth = round((avg7 / avg30 - 1) * 100, 1) if avg30 else 0
|
||
severity = '高' if inventory_support < max(threshold, 15) else '中'
|
||
items.append(_insight(
|
||
'sales_growth', '销量异常', severity,
|
||
f'{code} 近7天销量快速增长',
|
||
f'7天平均销量{avg7:g}件，30天平均销量{avg30:g}件，增长约{growth}%；库存可支撑{inventory_support:.1f}天。',
|
||
f'分析{code}销量增长后是否需要提前补货',
|
||
['product_sales', 'planning_metrics'],
|
||
product_code=code, metrics={'7天平均销量': avg7, '30天平均销量': avg30,
|
||
'增长幅度': growth, '库存支撑天数': inventory_support},
|
||
source_page='/planning'))
|
||
|
||
# 2. 已逾期但仍未到货的订单。
|
||
for row in orders:
|
||
status = str(row.get('状态') or '')
|
||
outstanding = _int(row.get('未到货数量'))
|
||
expected = _parse_date(row.get('预计到货日期'))
|
||
if status in ('已完结', '已完成') or outstanding <= 0 or not expected or expected >= today:
|
||
continue
|
||
overdue_days = (today - expected).days
|
||
severity = '高' if overdue_days >= 7 or outstanding >= 5000 else '中'
|
||
code = str(row.get('product_code') or '')
|
||
supplier = str(row.get('供应商') or '')
|
||
items.append(_insight(
|
||
'overdue_order', '订单交付', severity,
|
||
f'{row.get("订单类型") or "订单"} {row.get("订单ID")} 已逾期{overdue_days}天',
|
||
f'{code or row.get("货品名称") or "该商品"}未到货{outstanding}件，预计到货日期{expected.isoformat()}，供应商{supplier or "未填写"}。',
|
||
f'分析订单{row.get("订单ID")}延期风险及对{code or "相关商品"}库存的影响',
|
||
['order_detail'], product_code=code,
|
||
supplier=supplier, metrics={'逾期天数': overdue_days, '未到货数量': outstanding,
|
||
'预计到货日期': expected.isoformat()},
|
||
source_page={'国内订单': '/orders/domestic', '海外订单': '/orders/overseas',
|
||
'包材订单': '/orders/packaging'}.get(row.get('订单类型'), '/orders/domestic')))
|
||
|
||
# 3. 供应商履约风险。
|
||
for row in suppliers:
|
||
supplier = str(row.get('供应商') or '')
|
||
open_orders = _int(row.get('未完结订单数'))
|
||
outstanding = _int(row.get('未到货数量'))
|
||
completion = _num(row.get('到货完成率'))
|
||
defect = _num(row.get('不良品率'))
|
||
if open_orders < 2 and outstanding < 3000 and defect < 3:
|
||
continue
|
||
if completion < 50 or outstanding >= 20000 or defect >= 5:
|
||
severity = '高'
|
||
elif completion < 75 or outstanding >= 5000 or defect >= 2:
|
||
severity = '中'
|
||
else:
|
||
continue
|
||
items.append(_insight(
|
||
'supplier_risk', '供应商风险', severity,
|
||
f'{supplier or "未填写供应商"} 交付风险偏高',
|
||
f'未完结订单{open_orders}单，未到货{outstanding}件，到货完成率{completion:.2f}%，不良品率{defect:.2f}%。',
|
||
f'分析{supplier}供应商的未完结订单和交付风险',
|
||
['supplier_delivery'], supplier=supplier,
|
||
metrics={'未完结订单数': open_orders, '未到货数量': outstanding,
|
||
'到货完成率': completion, '不良品率': defect}, source_page='/suppliers'))
|
||
|
||
# 4. BOM瓶颈：按成品取最小可支撑数量，与建议下单/30天需求对比。
|
||
bom_by_product = defaultdict(list)
|
||
for row in bom:
|
||
code = str(row.get('product_code') or '')
|
||
if code:
|
||
bom_by_product[code].append(row)
|
||
for code, materials in bom_by_product.items():
|
||
plan = planning_map.get(code) or {}
|
||
daily = _num(plan.get('日销量预测'))
|
||
suggested = _int(plan.get('建议下单数量'))
|
||
# P1只把系统已经建议下单的成品纳入BOM预警，避免将低销量/历史BOM误报为即时缺口。
|
||
target = suggested
|
||
if target <= 0:
|
||
continue
|
||
usable = [m for m in materials if _num(m.get('单位用量')) > 0]
|
||
if not usable:
|
||
continue
|
||
bottleneck = min(usable, key=lambda m: _num(m.get('可支撑成品数量')))
|
||
support_qty = _int(bottleneck.get('可支撑成品数量'))
|
||
if support_qty >= target:
|
||
continue
|
||
gap = target - support_qty
|
||
severity = '高' if support_qty < target * 0.7 else '中'
|
||
material = str(bottleneck.get('物料名称') or bottleneck.get('material_code') or 'BOM物料')
|
||
items.append(_insight(
|
||
'bom_shortage', 'BOM物料', severity,
|
||
f'{code} 的{material}可能不足',
|
||
f'目标需求{target}件成品，当前该物料仅可支撑{support_qty}件，预计缺口{gap}件。',
|
||
f'分析{code}的BOM包材缺口并给出补货建议',
|
||
['bom_supply_chain', 'planning_metrics'], product_code=code,
|
||
supplier=str(bottleneck.get('物料供应商') or ''),
|
||
metrics={'目标需求': target, '可支撑成品数量': support_qty, '预计缺口': gap,
|
||
'瓶颈物料': material}, source_page='/bom'))
|
||
|
||
# 同一类/对象保留最高严重度，避免相同规则重复刷屏。
|
||
unique = {}
|
||
for item in items:
|
||
key = (item['type'], item.get('product_code'), item.get('supplier'), item['title'])
|
||
current = unique.get(key)
|
||
if not current or _severity_rank(item['severity']) > _severity_rank(current['severity']):
|
||
unique[key] = item
|
||
items = list(unique.values())
|
||
items.sort(key=lambda x: (-_severity_rank(x['severity']), x['category'], x['title']))
|
||
return {
|
||
'generated_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
||
'items': items,
|
||
'data_dates': {
|
||
key: datasource_meta(key).get('data_date') or ''
|
||
for key in ('planning_metrics', 'product_sales', 'product_inventory', 'inventory_value')
|
||
},
|
||
}
|
||
|
||
|
||
def get_operational_insights(force=False):
|
||
if force:
|
||
cache_delete(CACHE_KEY)
|
||
data = None if force else cache_get(CACHE_KEY)
|
||
if data is None:
|
||
data = _build_raw_insights()
|
||
cache_set(CACHE_KEY, data, ttl=CACHE_TTL)
|
||
return data
|
||
|
||
|
||
def position_profile(position, role=''):
|
||
position = str(position or '')
|
||
profiles = {
|
||
'procurement_planner': {
|
||
'name': '采购计划', 'focus_types': ['stockout', 'bom_shortage', 'overstock', 'sales_growth'],
|
||
'questions': ['哪些商品最需要优先补货？', '哪些建议下单商品存在BOM包材缺口？',
|
||
'哪些商品库存支撑天数过高？', '哪些销量增长商品需要提前备货？']},
|
||
'procurement_follower': {
|
||
'name': '采购跟单', 'focus_types': ['overdue_order', 'supplier_risk', 'bom_shortage'],
|
||
'questions': ['当前有哪些逾期未到货订单？', '哪些供应商未完结订单最多？',
|
||
'哪些延期订单会造成商品缺货？', '供应商到货完成率和不良品情况如何？']},
|
||
'operator': {
|
||
'name': '运营专员', 'focus_types': ['sales_decline', 'sales_growth', 'overstock', 'stockout'],
|
||
'questions': ['哪些商品近7天销量明显下降？', '哪些商品销量增长但库存不足？',
|
||
'哪些商品无销量但仍有大量库存？', '分析各品牌最近30天出库表现']},
|
||
'product_developer': {
|
||
'name': '产品开发', 'focus_types': ['bom_shortage', 'supplier_risk', 'overstock'],
|
||
'questions': ['哪些成品存在BOM物料瓶颈？', '哪些包材供应商交付风险较高？',
|
||
'新品相关产品是否存在库存积压？', '查询某个产品的BOM和供应商信息']},
|
||
'finance_specialist': {
|
||
'name': '财务专员', 'focus_types': ['overstock', 'supplier_risk'],
|
||
'questions': ['当前库存金额及主要构成是什么？', '哪些高库存金额商品动销较差？',
|
||
'供应商未到货数量和资金占用情况如何？', '各品牌库存金额和动销SKU数如何？']},
|
||
'bp': {
|
||
'name': 'BP', 'focus_types': ['stockout', 'overstock', 'sales_decline', 'sales_growth',
|
||
'overdue_order', 'supplier_risk', 'bom_shortage'],
|
||
'questions': ['生成今日经营风险分析报告', '当前最高风险的商品和订单有哪些？',
|
||
'库存、销量、订单和供应商有哪些异常？', '哪些问题需要管理层优先推动？']},
|
||
'procurement_trainee': {
|
||
'name': '采购管培生', 'focus_types': ['stockout', 'overdue_order', 'supplier_risk', 'bom_shortage'],
|
||
'questions': ['当前有哪些需要关注的采购风险？', '哪些订单已经逾期？',
|
||
'哪些商品需要补货？', '哪些BOM物料可能不足？']},
|
||
}
|
||
if position in profiles:
|
||
return profiles[position]
|
||
if role in ('admin', 'super_admin'):
|
||
return profiles['bp'] | {'name': '管理人员'}
|
||
return {
|
||
'name': '业务人员',
|
||
'focus_types': ['stockout', 'overstock', 'sales_decline', 'overdue_order', 'bom_shortage'],
|
||
'questions': ['当前有哪些高风险业务异常？', '哪些商品需要补货？',
|
||
'哪些订单已经延期？', '哪些商品库存积压？']
|
||
}
|
||
|
||
|
||
def _impact_score(item):
|
||
"""在同一严重度内按业务影响排序，防止大量低影响异常淹没真正风险。"""
|
||
metrics = item.get('metrics') or {}
|
||
qty = max(_num(metrics.get(k)) for k in (
|
||
'预计缺口', '未到货数量', '建议下单数量', '库存合计', '下单数量'
|
||
))
|
||
days = max(_num(metrics.get(k)) for k in ('逾期天数', '补货阈值'))
|
||
ratio = max(_num(metrics.get(k)) for k in ('下降幅度', '增长幅度', '不良品率'))
|
||
amount = _num(metrics.get('库存金额')) / 100.0
|
||
return qty + days * 100 + ratio * 10 + amount
|
||
|
||
def filter_insights(raw, allowed_sources, profile, limit=80):
|
||
allowed = set(allowed_sources or [])
|
||
focus = profile.get('focus_types') or []
|
||
items = []
|
||
for item in raw.get('items') or []:
|
||
source_keys = item.get('source_keys') or []
|
||
# 生成该异常实际用到的数据源必须全部有权限，避免跨菜单侧漏。
|
||
if not source_keys or any(k not in allowed for k in source_keys):
|
||
continue
|
||
copy = dict(item)
|
||
copy['source_keys'] = list(source_keys)
|
||
copy['metrics'] = dict(copy.get('metrics') or {})
|
||
if 'inventory_value' not in allowed:
|
||
copy['metrics'].pop('库存金额', None)
|
||
copy['focused'] = copy.get('type') in focus
|
||
items.append(copy)
|
||
items.sort(key=lambda x: (0 if x.get('focused') else 1, -_severity_rank(x['severity']),
|
||
-_impact_score(x), x['title']))
|
||
# 按异常类型和风险等级均衡取样，避免高风险数量太多导致中风险完全看不到。
|
||
max_items = max(1, min(int(limit or 80), 200))
|
||
grouped = defaultdict(list)
|
||
for item in items:
|
||
grouped[item['type']].append(item)
|
||
quota = max(4, (max_items + max(len(grouped), 1) - 1) // max(len(grouped), 1))
|
||
balanced = []
|
||
for group_items in grouped.values():
|
||
by_severity = defaultdict(list)
|
||
for item in group_items:
|
||
by_severity[item['severity']].append(item)
|
||
chosen = by_severity['高'][:max(1, int(quota * 0.65))]
|
||
chosen += by_severity['中'][:max(1, int(quota * 0.3))]
|
||
chosen += by_severity['低'][:max(1, quota - len(chosen))]
|
||
if len(chosen) < quota:
|
||
seen_ids = {x['id'] for x in chosen}
|
||
chosen += [x for x in group_items if x['id'] not in seen_ids][:quota-len(chosen)]
|
||
balanced.extend(chosen[:quota])
|
||
balanced.sort(key=lambda x: (0 if x.get('focused') else 1, -_severity_rank(x['severity']),
|
||
-_impact_score(x), x['title']))
|
||
items = balanced[:max_items]
|
||
counts = defaultdict(int)
|
||
categories = defaultdict(int)
|
||
for item in items:
|
||
counts[item['severity']] += 1
|
||
categories[item['category']] += 1
|
||
return {
|
||
'generated_at': raw.get('generated_at'),
|
||
'data_dates': raw.get('data_dates') or {},
|
||
'items': items,
|
||
'counts': {'high': counts['高'], 'medium': counts['中'], 'low': counts['低'], 'total': len(items)},
|
||
'categories': dict(categories),
|
||
'profile': profile,
|
||
}
|