1214 lines
66 KiB
Python
1214 lines
66 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""AI 分析数据源层：把系统已按规则计算好的数据封装为数据源，AI 只选择数据源+参数，不直接查原始表。
|
||
|
||
每个数据源 = {key, name, desc, fields, handler(params) -> {columns, rows}}
|
||
handler 返回 {columns: [列名...], rows: [{列名: 值}...]}
|
||
"""
|
||
from decimal import Decimal, ROUND_HALF_UP
|
||
from services.db import query
|
||
from services.inventory_calc import (
|
||
compute_inventory_cost_summary,
|
||
compute_product_inventory_cached,
|
||
compute_cloud_inventory_cached,
|
||
compute_external_inventory_cached,
|
||
compute_product_in_transit_cached,
|
||
compute_product_inventory_15,
|
||
resolve_cost_price,
|
||
_get_latest_inv_date,
|
||
)
|
||
from services.sales_calc import (compute_product_sales_extended_cached,
|
||
compute_product_sales_cached)
|
||
from services.forecast import sales_predict_config_cached
|
||
from services.product_type import get_product_type_cached
|
||
|
||
|
||
# 英文字段名 → 中文字段名映射（AI 可能输出英文列名，需归一化后才能排序）
|
||
_FIELD_ALIASES = {
|
||
'available_stock': '库存',
|
||
'stock': '库存',
|
||
'inventory': '库存',
|
||
'cloud_stock': '云仓库存',
|
||
'cloud_inventory': '云仓库存',
|
||
'external_stock': '外仓库存',
|
||
'external_inventory': '外仓库存',
|
||
'factory_stock': '加工厂库存',
|
||
'factory_inventory': '加工厂库存',
|
||
'production_transit': '生产在途',
|
||
'purchase_transit': '采购在途',
|
||
'product_code': 'product_code',
|
||
'昨日销量': '昨日销量',
|
||
'yesterday_sales': '昨日销量',
|
||
'3天平均销量': '3天平均销量',
|
||
'7天平均销量': '7天平均销量',
|
||
'30天平均销量': '30天平均销量',
|
||
'60天平均销量': '60天平均销量',
|
||
}
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 工具：合并库存/在途数据为每个产品的完整视图
|
||
# ──────────────────────────────────────────────
|
||
def _products_map(codes, active_only=False):
|
||
"""批量取产品资料（编码 → 名称/类型/品牌），分片查询避免 IN 参数过长"""
|
||
codes = [c for c in dict.fromkeys(codes) if c]
|
||
out = {}
|
||
for i in range(0, len(codes), 500):
|
||
chunk = codes[i:i + 500]
|
||
ph = ','.join(['%s'] * len(chunk))
|
||
for r in query(
|
||
f"SELECT product_code, product_name, product_type, brand FROM products WHERE product_code IN ({ph})"
|
||
+ (" AND deleted_at IS NULL" if active_only else ""),
|
||
chunk
|
||
):
|
||
out[r['product_code']] = r
|
||
return out
|
||
|
||
|
||
def _build_product_inventory_view():
|
||
inv_map = compute_product_inventory_cached() or {}
|
||
cloud_map = compute_cloud_inventory_cached() or {}
|
||
ext_map = compute_external_inventory_cached() or {}
|
||
inv15_map = compute_product_inventory_15() or {}
|
||
transit_map = compute_product_in_transit_cached() or {}
|
||
|
||
codes = set(inv_map) | set(cloud_map) | set(ext_map) | set(inv15_map) | set(transit_map)
|
||
# 产品资料（判断包材/内料）
|
||
prod_info = {}
|
||
if codes:
|
||
_codes = list(codes)
|
||
_ph = ','.join(['%s'] * len(_codes))
|
||
_rows = query(f"SELECT product_code, product_name, product_type, brand FROM products WHERE deleted_at IS NULL AND product_code IN ({_ph})", _codes)
|
||
prod_info = {r['product_code']: r for r in _rows}
|
||
|
||
def _is_inner_or_packaging(code):
|
||
"""内料：名称含「内料」或编码含 -NL；包材：产品类型为包材；无产品资料的视为虚拟货品（邮费链接/服务包/晒单礼等）排除"""
|
||
code_u = (code or '').upper()
|
||
if '-NL' in code_u:
|
||
return True
|
||
info = prod_info.get(code)
|
||
if not info:
|
||
return True # 无产品资料：虚拟货品/异常编码，不参与库存排行
|
||
name = info.get('product_name') or ''
|
||
ptype = info.get('product_type') or ''
|
||
if '内料' in name:
|
||
return True
|
||
if ptype == '包材':
|
||
return True
|
||
if ptype in ('赠品', '其他', '原材料'):
|
||
return True
|
||
return False
|
||
|
||
rows = []
|
||
for code in codes:
|
||
if _is_inner_or_packaging(code):
|
||
continue
|
||
t = transit_map.get(code, {})
|
||
rows.append({
|
||
'product_code': code,
|
||
'品牌': (prod_info.get(code) or {}).get('brand') or '',
|
||
'库存': int(inv_map.get(code, 0)),
|
||
'云仓库存': int(cloud_map.get(code, 0)),
|
||
'外仓库存': int(ext_map.get(code, 0)),
|
||
'加工厂库存': int(inv15_map.get(code, 0)),
|
||
'生产在途': int(t.get('生产在途', 0)),
|
||
'采购在途': int(t.get('采购在途', 0)),
|
||
})
|
||
return rows
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 数据源 handlers
|
||
# ──────────────────────────────────────────────
|
||
def _h_inventory_cost_summary(params):
|
||
"""库存成本汇总（与数据概览一致：特殊成本价、排除内料、在途规则）"""
|
||
cost = compute_inventory_cost_summary()
|
||
row = {
|
||
'核对日期': cost['latest_date'],
|
||
'库存总成本(元)': cost['total_cost'],
|
||
'成品成本(元)': cost['finished_cost'],
|
||
'包材成本(元)': cost['packaging_cost'],
|
||
'成品库存(元)': cost['finished_inventory_cost'],
|
||
'成品生产在途(元)': cost['finished_production_transit_cost'],
|
||
'成品采购在途(元)': cost['finished_purchase_transit_cost'],
|
||
'包材库存(元)': cost['packaging_inventory_cost'],
|
||
'包材采购在途(元)': cost['packaging_purchase_transit_cost'],
|
||
}
|
||
return {'columns': list(row.keys()), 'rows': [row]}
|
||
|
||
|
||
def _apply_filters(rows, params, text_fields=()):
|
||
"""通用筛选：product_code / brand / limit / sort_by / sort_order
|
||
|
||
sort_by 支持中文字段名或英文别名（available_stock→库存 等）；
|
||
降序排行时自动剔除排序字段为 0 的行，避免 TOP 榜混入无效数据。
|
||
"""
|
||
code = (params.get('product_code') or '').strip()
|
||
if code:
|
||
code_lower = code.lower()
|
||
exact = [r for r in rows if code_lower == str(r.get('product_code', '')).strip().lower()]
|
||
# 编码存在精确结果时优先精确匹配，避免成品编码同时命中其-BT/-CH等子物料。
|
||
rows = exact or [r for r in rows if code_lower in str(r.get('product_code', '')).lower()
|
||
or code_lower in str(r.get('product_name', '')).lower()
|
||
or code_lower in str(r.get('货品名称', '')).lower()]
|
||
brand = (params.get('brand') or '').strip()
|
||
if brand and '品牌' in (rows[0] if rows else {}):
|
||
rows = [r for r in rows if brand.lower() in str(r.get('品牌', '')).lower()]
|
||
# P0 多业务域通用筛选。字段不存在时忽略，避免每个handler重复实现。
|
||
common_filters = {
|
||
'supplier': ('供应商', '物料供应商'),
|
||
'status': ('状态',),
|
||
'product_type': ('产品类型', '类型'),
|
||
'planner': ('采购计划',),
|
||
'follower': ('采购跟单',),
|
||
'operator': ('运营',),
|
||
'material_code': ('material_code',),
|
||
}
|
||
sample = rows[0] if rows else {}
|
||
for param_key, candidates in common_filters.items():
|
||
expected = str(params.get(param_key) or '').strip().lower()
|
||
if not expected:
|
||
continue
|
||
target = next((field for field in candidates if field in sample), None)
|
||
if target:
|
||
rows = [r for r in rows if expected in str(r.get(target) or '').lower()]
|
||
sort_by = (params.get('sort_by') or '').strip()
|
||
sort_order = (params.get('sort_order') or 'desc').strip().lower()
|
||
if sort_by and rows:
|
||
# 英文字段名 → 中文字段名映射
|
||
sort_field = _FIELD_ALIASES.get(sort_by, sort_by)
|
||
if sort_field in rows[0]:
|
||
rev = sort_order == 'desc'
|
||
try:
|
||
# 数值字段排序（None 排最后）
|
||
rows = sorted(rows, key=lambda r: (r.get(sort_field) is None, float(r.get(sort_field) or 0)), reverse=rev)
|
||
except (TypeError, ValueError):
|
||
# 文本字段（如是否调拨/产品分级/销售趋势）按字符串排序
|
||
rows = sorted(rows, key=lambda r: (r.get(sort_field) is None, str(r.get(sort_field) or '')), reverse=rev)
|
||
# 降序排行：剔除排序字段为 0 的行（TOP 榜不含无效数据）
|
||
if rev:
|
||
try:
|
||
float(rows[0].get(sort_field) or 0)
|
||
rows = [r for r in rows if r.get(sort_field) is not None and float(r.get(sort_field) or 0) != 0]
|
||
except (TypeError, ValueError):
|
||
pass
|
||
if params.get('_no_limit'):
|
||
return rows
|
||
limit = params.get('limit')
|
||
try:
|
||
limit = int(limit) if limit else 50
|
||
except (TypeError, ValueError):
|
||
limit = 50
|
||
if limit < 1:
|
||
limit = 50
|
||
if limit > 200:
|
||
limit = 200
|
||
return rows[:limit]
|
||
|
||
|
||
def _h_product_inventory(params):
|
||
"""每个产品的最新库存视图（可用/云仓/外仓/加工厂/在途）"""
|
||
rows = _build_product_inventory_view()
|
||
rows = _apply_filters(rows, params)
|
||
columns = ['product_code', '品牌', '库存', '云仓库存', '外仓库存', '加工厂库存', '生产在途', '采购在途']
|
||
return {'columns': columns, 'rows': rows}
|
||
|
||
|
||
def _h_product_sales(params):
|
||
"""每个产品的销量统计（总销量：昨日/3/7/14/30/60天平均、最高）"""
|
||
data = compute_product_sales_extended_cached() or {}
|
||
total = data.get('total') or {}
|
||
brand_map = {c: (p.get('brand') or '') for c, p in _products_map(list(total.keys())).items()}
|
||
rows = []
|
||
for code, s in total.items():
|
||
rows.append({
|
||
'product_code': code,
|
||
'品牌': brand_map.get(code, ''),
|
||
'昨日销量': s.get('昨日销量', 0),
|
||
'3天平均销量': s.get('3天平均销量', 0),
|
||
'7天平均销量': s.get('7天平均销量', 0),
|
||
'14天平均销量': s.get('14天平均销量', 0),
|
||
'30天平均销量': s.get('30天平均销量', 0),
|
||
'60天平均销量': s.get('60天平均销量', 0),
|
||
'3天最高销量': s.get('3天最高销量', 0),
|
||
'7天最高销量': s.get('7天最高销量', 0),
|
||
'14天最高销量': s.get('14天最高销量', 0),
|
||
'30天最高销量': s.get('30天最高销量', 0),
|
||
})
|
||
rows = _apply_filters(rows, params)
|
||
columns = ['product_code', '品牌', '昨日销量', '3天平均销量', '7天平均销量', '14天平均销量',
|
||
'30天平均销量', '60天平均销量', '3天最高销量', '7天最高销量', '14天最高销量', '30天最高销量']
|
||
return {'columns': columns, 'rows': rows}
|
||
|
||
|
||
def _h_outbound_stats(params):
|
||
"""出库统计：按日期/品牌/店铺/单品 TOP（与数据概览一致口径）
|
||
|
||
支持 days 参数限定最近 N 天（如"最近30天"传 days=30）；
|
||
自动排除无效品牌/店铺（无/-/现货/NA/合计等）和 0 出库数据。
|
||
"""
|
||
kind = (params.get('kind') or 'product').strip()
|
||
days = params.get('days') or None
|
||
try:
|
||
days = int(days)
|
||
except (TypeError, ValueError):
|
||
days = None
|
||
if days is not None:
|
||
days = max(1, min(days, 365))
|
||
limit = params.get('limit') or 10
|
||
try:
|
||
limit = int(limit)
|
||
except (TypeError, ValueError):
|
||
limit = 10
|
||
limit = max(1, min(limit, 50))
|
||
|
||
if kind == 'brand':
|
||
date_cond = ''
|
||
date_args = []
|
||
if days:
|
||
date_cond = ' AND 日期 >= DATE_SUB(CURDATE(), INTERVAL %s DAY)'
|
||
date_args.append(days)
|
||
rows = query("""
|
||
SELECT 品牌, SUM(货品数量) AS 总数量, ROUND(SUM(货品成本),2) AS 总成本,
|
||
COUNT(DISTINCT 货品编号) AS SKU数
|
||
FROM daily_outbound
|
||
WHERE 品牌 IS NOT NULL AND 品牌 != '' AND 品牌 != 'NA'
|
||
AND 品牌 NOT IN ('无', '-', '现货', '新媒体现货', '潘思宇代发')
|
||
""" + date_cond + """
|
||
GROUP BY 品牌
|
||
HAVING SUM(货品数量) > 0
|
||
ORDER BY 总数量 DESC LIMIT %s
|
||
""", tuple(date_args + [limit]))
|
||
return {'columns': ['品牌', '总数量', '总成本', 'SKU数'], 'rows': rows}
|
||
if kind == 'shop':
|
||
date_cond = ''
|
||
date_args = []
|
||
if days:
|
||
date_cond = ' AND 日期 >= DATE_SUB(CURDATE(), INTERVAL %s DAY)'
|
||
date_args.append(days)
|
||
rows = query("""
|
||
SELECT 店铺, SUM(货品数量) AS 总数量, ROUND(SUM(货品成本),2) AS 总成本
|
||
FROM daily_outbound
|
||
WHERE 店铺 IS NOT NULL AND 店铺 != '' AND 店铺 != 'NA'
|
||
AND 店铺 NOT LIKE '%%合计%%'
|
||
""" + date_cond + """
|
||
GROUP BY 店铺
|
||
HAVING SUM(货品数量) > 0
|
||
ORDER BY 总数量 DESC LIMIT %s
|
||
""", tuple(date_args + [limit]))
|
||
return {'columns': ['店铺', '总数量', '总成本'], 'rows': rows}
|
||
# 默认：单品 TOP
|
||
date_cond = ''
|
||
date_args = []
|
||
if days:
|
||
date_cond = ' AND d.日期 >= DATE_SUB(CURDATE(), INTERVAL %s DAY)'
|
||
date_args.append(days)
|
||
rows = query("""
|
||
SELECT d.货品编号, MAX(d.货品名称) AS 货品名称, SUM(d.货品数量) AS 总数量,
|
||
ROUND(SUM(d.货品成本),2) AS 总成本
|
||
FROM daily_outbound d
|
||
LEFT JOIN products p ON d.货品编号 = p.product_code
|
||
WHERE d.货品编号 IS NOT NULL AND d.货品编号 != '' AND d.货品编号 != 'NA'
|
||
AND d.货品编号 NOT LIKE '%%合计%%'
|
||
AND d.货品名称 NOT LIKE '%%赠品%%'
|
||
AND d.货品名称 NOT LIKE '%%小样%%'
|
||
AND COALESCE(p.product_type, '') != '赠品'
|
||
""" + date_cond + """
|
||
GROUP BY d.货品编号
|
||
HAVING SUM(d.货品数量) > 0
|
||
ORDER BY 总数量 DESC LIMIT %s
|
||
""", tuple(date_args + [limit]))
|
||
return {'columns': ['货品编号', '货品名称', '总数量', '总成本'], 'rows': rows}
|
||
|
||
|
||
def _h_outbound_daily(params):
|
||
"""按日期出库统计（最近 N 天）"""
|
||
days = params.get('days') or 10
|
||
try:
|
||
days = int(days)
|
||
except (TypeError, ValueError):
|
||
days = 10
|
||
days = max(1, min(days, 60))
|
||
rows = query("""
|
||
SELECT DATE_FORMAT(日期, '%%Y-%%m-%%d') AS 日期, COUNT(*) AS 记录数,
|
||
SUM(货品数量) AS 总数量, ROUND(SUM(货品成本),2) AS 总成本
|
||
FROM daily_outbound GROUP BY 日期 ORDER BY 日期 DESC LIMIT %s
|
||
""", (days,))
|
||
return {'columns': ['日期', '记录数', '总数量', '总成本'], 'rows': rows}
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 计划预测口径（复用计划预测页同一套计算，避免口径不一致）
|
||
# ──────────────────────────────────────────────
|
||
PLANNING_FIELDS = ['product_code', '品牌', '日销量预测', '总支撑天数', '云仓支撑天数', '外仓支撑天数',
|
||
'库存', '云仓库存', '外仓库存', '加工厂库存', '生产在途', '采购在途',
|
||
'云仓可分配库存', '外仓可分配库存', '是否调拨', '建议下单数量',
|
||
'产品分级', '销售趋势', '库存可支撑天数', '昨日销量', '30天平均销量']
|
||
|
||
|
||
def _h_planning_metrics(params):
|
||
"""计划预测口径：日销量预测、支撑天数、可分配库存、是否调拨、建议下单数量、分级/趋势"""
|
||
from routes.planning import _enrich_row_with_planning_data, _ensure_global_config
|
||
base_rows = query("""
|
||
SELECT p.product_code, p.product_name, p.brand, p.product_type,
|
||
p.lead_time AS 生产货期, p.logistics_time AS 物流时间, p.created_at AS 创建时间
|
||
FROM products p
|
||
WHERE p.deleted_at IS NULL
|
||
AND COALESCE(p.product_type, '') NOT IN ('包材', '其他')
|
||
AND p.product_code NOT IN (SELECT product_code FROM plan_hidden)
|
||
""")
|
||
sales = compute_product_sales_cached()
|
||
inv_map = compute_product_inventory_cached()
|
||
inv15_map = compute_product_inventory_15()
|
||
transit_map = compute_product_in_transit_cached()
|
||
cloud_inv_map = compute_cloud_inventory_cached()
|
||
ext_inv_map = compute_external_inventory_cached()
|
||
extended_sales = compute_product_sales_extended_cached()
|
||
sp_map = {r['trend_type']: r for r in sales_predict_config_cached()}
|
||
gc = _ensure_global_config()
|
||
rows = []
|
||
for row in base_rows:
|
||
row['生产货期'] = row.get('生产货期') or 0
|
||
row['物流时间'] = row.get('物流时间') or 0
|
||
if row.get('创建时间') and hasattr(row['创建时间'], 'strftime'):
|
||
row['创建时间'] = row['创建时间'].strftime('%Y-%m-%d')
|
||
row['产品种类'] = get_product_type_cached(row.get('product_name', ''), row.get('brand', ''), row.get('product_type'))
|
||
row['品牌'] = row.get('brand') or ''
|
||
row = _enrich_row_with_planning_data(
|
||
row, sales, inv_map, inv15_map, transit_map, sp_map, gc,
|
||
cloud_inv_map=cloud_inv_map, ext_inv_map=ext_inv_map, extended_sales=extended_sales
|
||
)
|
||
rows.append({k: row.get(k) for k in PLANNING_FIELDS})
|
||
rows = _apply_filters(rows, params)
|
||
return {'columns': PLANNING_FIELDS, 'rows': rows}
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 订单在途与到货
|
||
# ──────────────────────────────────────────────
|
||
def _h_orders_in_transit(params):
|
||
"""未完结订单的未到数量（国内/海外/包材），包材按所属成品归集"""
|
||
ORDER_FIELDS = ['product_code', '品牌', '货品名称', '国内未到数量', '海外未到数量', '包材未到数量',
|
||
'在途未到合计', '国内未完结订单数', '海外未完结订单数', '包材未完结订单数']
|
||
by_code = {}
|
||
|
||
def _merge(code, name, qty, orders, qty_key, order_key):
|
||
code = (code or '').strip()
|
||
if not code:
|
||
return
|
||
row = by_code.setdefault(code, {'product_code': code, '货品名称': name or '',
|
||
'国内未到数量': 0, '海外未到数量': 0, '包材未到数量': 0,
|
||
'国内未完结订单数': 0, '海外未完结订单数': 0, '包材未完结订单数': 0})
|
||
row[qty_key] += int(qty or 0)
|
||
row[order_key] += int(orders or 0)
|
||
if name and not row['货品名称']:
|
||
row['货品名称'] = name
|
||
|
||
for r in query("""
|
||
SELECT 货品编号 AS code, MAX(货品名称) AS name, COUNT(*) AS n,
|
||
SUM(COALESCE(未到货数, COALESCE(数量, 0) - COALESCE(实际到货数量, 0))) AS qty
|
||
FROM domestic_orders WHERE COALESCE(状态, '') != '已完结'
|
||
GROUP BY 货品编号
|
||
"""):
|
||
_merge(r['code'], r['name'], r['qty'], r['n'], '国内未到数量', '国内未完结订单数')
|
||
for r in query("""
|
||
SELECT 产品编码 AS code, MAX(货品名称) AS name, COUNT(*) AS n,
|
||
SUM(COALESCE(未到货数, COALESCE(数量, 0) - COALESCE(实际到货数量, 0))) AS qty
|
||
FROM overseas_orders WHERE COALESCE(状态, '') != '已完结'
|
||
GROUP BY 产品编码
|
||
"""):
|
||
_merge(r['code'], r['name'], r['qty'], r['n'], '海外未到数量', '海外未完结订单数')
|
||
for r in query("""
|
||
SELECT 来源产品编码 AS code, MAX(名称) AS name, COUNT(*) AS n,
|
||
SUM(COALESCE(未到货数, COALESCE(数量, 0) - COALESCE(实际到货数量, 0))) AS qty
|
||
FROM packaging_orders WHERE COALESCE(状态, '') != '已完结'
|
||
GROUP BY 来源产品编码
|
||
"""):
|
||
_merge(r['code'], r['name'], r['qty'], r['n'], '包材未到数量', '包材未完结订单数')
|
||
|
||
rows = list(by_code.values())
|
||
brand_map = {c: (p.get('brand') or '') for c, p in _products_map([r['product_code'] for r in rows]).items()}
|
||
for r in rows:
|
||
r['品牌'] = brand_map.get(r['product_code'], '')
|
||
r['在途未到合计'] = r['国内未到数量'] + r['海外未到数量'] + r['包材未到数量']
|
||
rows = _apply_filters(rows, params)
|
||
return {'columns': ORDER_FIELDS, 'rows': rows}
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 包材/内料库存
|
||
# ──────────────────────────────────────────────
|
||
def _h_material_inventory(params):
|
||
"""包材/内料库存：产品库存视图已排除的物料维度库存（可用/云仓/外仓/加工厂/在途）"""
|
||
MATERIAL_FIELDS = ['product_code', '名称', '类型', '品牌', '可用库存', '云仓库存', '外仓库存',
|
||
'加工厂库存', '生产在途', '采购在途']
|
||
latest = _get_latest_inv_date()
|
||
raw = query("""
|
||
SELECT product_code, MAX(product_name) AS 名称,
|
||
SUM(available_qty) AS 可用库存,
|
||
SUM(production_in_transit) AS 生产在途,
|
||
SUM(purchase_in_transit) AS 采购在途
|
||
FROM inventory_detail WHERE data_date=%s
|
||
GROUP BY product_code
|
||
""", (latest,))
|
||
prod = _products_map([r['product_code'] for r in raw])
|
||
cloud_map = compute_cloud_inventory_cached() or {}
|
||
ext_map = compute_external_inventory_cached() or {}
|
||
inv15_map = compute_product_inventory_15() or {}
|
||
rows = []
|
||
for r in raw:
|
||
code = r['product_code'] or ''
|
||
info = prod.get(code) or {}
|
||
name = info.get('product_name') or r.get('名称') or ''
|
||
is_inner = ('-NL' in code.upper()) or ('内料' in name)
|
||
is_packaging = (info.get('product_type') or '') == '包材'
|
||
if not (is_inner or is_packaging):
|
||
continue
|
||
rows.append({
|
||
'product_code': code,
|
||
'名称': name,
|
||
'类型': '内料' if is_inner else '包材',
|
||
'品牌': info.get('brand') or '',
|
||
'可用库存': int(r['可用库存'] or 0),
|
||
'云仓库存': int(cloud_map.get(code, 0) or 0),
|
||
'外仓库存': int(ext_map.get(code, 0) or 0),
|
||
'加工厂库存': int(inv15_map.get(code, 0) or 0),
|
||
'生产在途': int(r['生产在途'] or 0),
|
||
'采购在途': int(r['采购在途'] or 0),
|
||
})
|
||
rows = _apply_filters(rows, params)
|
||
return {'columns': MATERIAL_FIELDS, 'rows': rows}
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 产品库存金额
|
||
# ──────────────────────────────────────────────
|
||
VALUE_FIELDS = ['product_code', '品牌', '产品名称', '库存', '成本价', '库存金额']
|
||
|
||
|
||
def _inventory_value_rows(latest=None):
|
||
"""成品口径的逐商品库存金额（库存 × 成本价，命中特殊成本价规则）"""
|
||
latest = latest or _get_latest_inv_date()
|
||
raw = query("""
|
||
SELECT product_code, ANY_VALUE(cost_price) AS 成本价, SUM(available_qty) AS 库存
|
||
FROM inventory_detail
|
||
WHERE data_date=%s AND product_code NOT LIKE '%%-NL%%' AND product_name NOT LIKE '%%内料%%'
|
||
GROUP BY product_code
|
||
""", (latest,))
|
||
prod = _products_map([r['product_code'] for r in raw], active_only=True)
|
||
rows = []
|
||
for r in raw:
|
||
code = r['product_code'] or ''
|
||
info = prod.get(code) or {}
|
||
if (info.get('product_type') or '') == '包材' or '内料' in (info.get('product_name') or ''):
|
||
continue
|
||
if not info:
|
||
continue # 无产品资料的虚拟货品/异常编码不计入
|
||
stock = int(r['库存'] or 0)
|
||
price = resolve_cost_price(code, float(r['成本价'] or 0))
|
||
rows.append({
|
||
'product_code': code,
|
||
'品牌': (info.get('brand') or '').strip(),
|
||
'产品名称': info.get('product_name') or '',
|
||
'库存': stock,
|
||
'成本价': price,
|
||
'库存金额': float((Decimal(stock) * Decimal(str(price))).quantize(Decimal('0.01'), rounding=ROUND_HALF_UP)),
|
||
})
|
||
return rows
|
||
|
||
|
||
def _inventory_value_filtered(rows, params):
|
||
"""汇总只应用业务筛选；排序、零值排行剔除和条数限制只影响明细。"""
|
||
business = {k: v for k, v in params.items() if k not in ('sort_by', 'sort_order', 'limit')}
|
||
brand = str(params.get('brand') or '').strip()
|
||
if brand:
|
||
exact = [r for r in rows if str(r.get('品牌') or '').strip().casefold() == brand.casefold()]
|
||
if exact:
|
||
rows = exact
|
||
business.pop('brand', None)
|
||
return _apply_filters(rows, {**business, '_no_limit': True})
|
||
|
||
|
||
def _inventory_value_summary(filtered, latest, params):
|
||
total_value = sum((Decimal(str(r.get('库存金额') or 0)) for r in filtered), Decimal('0'))
|
||
top = sorted(filtered, key=lambda r: (-float(r.get('库存金额') or 0), str(r.get('product_code') or '')))[:5]
|
||
return {
|
||
'kind': 'inventory_value',
|
||
'authoritative': True,
|
||
'date': latest,
|
||
'row_count': len(filtered),
|
||
'sku_count': len(filtered),
|
||
'stock_qty': sum(int(r.get('库存') or 0) for r in filtered),
|
||
'inventory_value': float(total_value.quantize(Decimal('0.01'), rounding=ROUND_HALF_UP)),
|
||
'zero_stock_sku_count': sum(1 for r in filtered if not r.get('库存')),
|
||
'unpriced_stock_sku_count': sum(1 for r in filtered if r.get('库存') and not r.get('成本价')),
|
||
'filter': {k: str(params.get(k) or '').strip() for k in ('brand', 'product_code')
|
||
if str(params.get(k) or '').strip()},
|
||
'top_items': [
|
||
{'product_code': r.get('product_code'), 'product_name': r.get('产品名称'),
|
||
'stock_qty': int(r.get('库存') or 0), 'inventory_value': float(r.get('库存金额') or 0),
|
||
'cost_price': float(r.get('成本价') or 0)}
|
||
for r in top
|
||
],
|
||
}
|
||
|
||
|
||
def _h_inventory_value(params):
|
||
"""逐商品库存金额明细及独立于展示条数的全量汇总。"""
|
||
latest = _get_latest_inv_date()
|
||
filtered = _inventory_value_filtered(_inventory_value_rows(latest), params)
|
||
return {'columns': VALUE_FIELDS, 'rows': _apply_filters(filtered, params),
|
||
'summary': _inventory_value_summary(filtered, latest, params)}
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 品牌维度汇总
|
||
# ──────────────────────────────────────────────
|
||
BRAND_FIELDS = ['品牌', 'SKU数', '库存', '库存金额', '昨日销量', '30天平均销量',
|
||
'近30天动销SKU数', '近100天动销SKU数']
|
||
|
||
|
||
def _h_brand_overview(params):
|
||
"""按品牌汇总库存/金额/销量与动销SKU数（默认按库存金额倒序）"""
|
||
latest = _get_latest_inv_date()
|
||
value_rows = _inventory_value_filtered(_inventory_value_rows(latest), params)
|
||
sales = compute_product_sales_cached() or {}
|
||
active_30 = {r['货品编号'] for r in query(
|
||
"SELECT DISTINCT 货品编号 FROM daily_outbound WHERE 日期 >= DATE_SUB(%s, INTERVAL 29 DAY)", (latest,))}
|
||
active_100 = {r['货品编号'] for r in query(
|
||
"SELECT DISTINCT 货品编号 FROM daily_outbound WHERE 日期 >= DATE_SUB(%s, INTERVAL 99 DAY)", (latest,))}
|
||
agg = {}
|
||
for r in value_rows:
|
||
b = r['品牌'] or '（无品牌）'
|
||
row = agg.setdefault(b, {'品牌': b, 'SKU数': 0, '库存': 0, '库存金额': 0.0,
|
||
'昨日销量': 0, '30天平均销量': 0.0,
|
||
'近30天动销SKU数': 0, '近100天动销SKU数': 0})
|
||
code = r['product_code']
|
||
s = sales.get(code) or {}
|
||
row['SKU数'] += 1
|
||
row['库存'] += int(r['库存'] or 0)
|
||
row['库存金额'] = float(Decimal(str(row['库存金额'])) + Decimal(str(r['库存金额'] or 0)))
|
||
row['昨日销量'] += int(s.get('昨日销量', 0) or 0)
|
||
row['30天平均销量'] = round(row['30天平均销量'] + float(s.get('30天平均销量', 0) or 0), 2)
|
||
if code in active_30:
|
||
row['近30天动销SKU数'] += 1
|
||
if code in active_100:
|
||
row['近100天动销SKU数'] += 1
|
||
rows = sorted(agg.values(), key=lambda r: r['库存金额'], reverse=True)
|
||
summary = _inventory_value_summary(value_rows, latest, params)
|
||
summary.update(kind='brand_overview', row_count=len(rows))
|
||
return {
|
||
'columns': BRAND_FIELDS,
|
||
'rows': _apply_filters(rows, params),
|
||
'summary': summary,
|
||
}
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# P0 经营分析补充数据源：产品档案 / BOM供应链 / 订单明细 / 供应商履约
|
||
# ──────────────────────────────────────────────
|
||
def _h_product_profile(params):
|
||
"""产品主数据档案（不返回成本价，避免绕过金额字段权限）。"""
|
||
fields = ['product_code', '货品名称', '品牌', '产品类型', 'SKU类型', '生产货期', '物流时间',
|
||
'起订量', '供应商', '采购计划', '采购跟单', '运营', '开发', '创建时间']
|
||
rows = query("""
|
||
SELECT product_code, product_name AS 货品名称, COALESCE(brand, '') AS 品牌,
|
||
COALESCE(product_type, '') AS 产品类型, COALESCE(sku_type, '') AS SKU类型,
|
||
COALESCE(lead_time, 0) AS 生产货期, COALESCE(logistics_time, 0) AS 物流时间,
|
||
COALESCE(min_order_qty, 0) AS 起订量,
|
||
COALESCE(supplier_name_match, '') AS 供应商,
|
||
COALESCE(procurement_planner, '') AS 采购计划,
|
||
COALESCE(procurement_follower, '') AS 采购跟单,
|
||
COALESCE(operator, '') AS 运营, COALESCE(developer, '') AS 开发,
|
||
DATE_FORMAT(created_at, '%%Y-%%m-%%d') AS 创建时间
|
||
FROM products
|
||
WHERE deleted_at IS NULL
|
||
""")
|
||
rows = _apply_filters(rows, params)
|
||
return {'columns': fields, 'rows': rows}
|
||
|
||
|
||
def _h_bom_supply_chain(params):
|
||
"""成品-BOM-包材库存与在途联查，用于识别“成品能下单但包材不足”。"""
|
||
fields = ['product_code', '成品名称', '品牌', 'material_code', '物料名称', '单位用量',
|
||
'物料供应商', '物料库存', '加工厂库存', '生产在途', '采购在途',
|
||
'可支撑成品数量', '包材下单前数量', '包材下单后数量']
|
||
rows = query("""
|
||
SELECT b.product_code, MAX(b.product_name) AS 成品名称,
|
||
COALESCE(MAX(p.brand), '') AS 品牌,
|
||
b.material_code, MAX(b.material_name) AS 物料名称,
|
||
COALESCE(MAX(b.quantity), 0) AS 单位用量,
|
||
COALESCE(MAX(b.supplier), '') AS 物料供应商,
|
||
COALESCE(MAX(bp.pre_pack_qty), 0) AS 包材下单前数量,
|
||
COALESCE(MAX(bp.post_pack_qty), 0) AS 包材下单后数量
|
||
FROM bom b
|
||
LEFT JOIN products p ON p.product_code=b.product_code AND p.deleted_at IS NULL
|
||
LEFT JOIN bom_packaging bp ON bp.product_code=b.product_code AND bp.material_code=b.material_code
|
||
GROUP BY b.product_code, b.material_code
|
||
""")
|
||
inv = compute_product_inventory_cached() or {}
|
||
inv15 = compute_product_inventory_15() or {}
|
||
transit = compute_product_in_transit_cached() or {}
|
||
for r in rows:
|
||
code = r.get('material_code')
|
||
qty = float(r.get('单位用量') or 0)
|
||
t = transit.get(code) or {}
|
||
r['物料库存'] = int(inv.get(code, 0) or 0)
|
||
r['加工厂库存'] = int(inv15.get(code, 0) or 0)
|
||
r['生产在途'] = int(t.get('生产在途', 0) or 0)
|
||
r['采购在途'] = int(t.get('采购在途', 0) or 0)
|
||
available = r['物料库存'] + r['加工厂库存'] + r['生产在途'] + r['采购在途']
|
||
r['可支撑成品数量'] = int(available / qty) if qty > 0 else 0
|
||
rows = _apply_filters(rows, params)
|
||
return {'columns': fields, 'rows': rows}
|
||
|
||
|
||
def _h_order_detail(params):
|
||
"""国内/海外/包材订单统一明细，用于分析交付、未到货和延期。"""
|
||
fields = ['订单类型', '订单ID', 'product_code', '货品名称', '供应商', '计划下单时间',
|
||
'预计到货日期', '实际到货日期', '下单数量', '实际到货数量', '未到货数量',
|
||
'不良品数量', '状态', '进度', '采购计划', '下单人']
|
||
rows = []
|
||
for r in query("""
|
||
SELECT '国内订单' AS 订单类型, d.id AS 订单ID, d.货品编号 AS product_code,
|
||
d.货品名称 AS 货品名称, COALESCE(d.供货商, '') AS 供应商,
|
||
d.计划下单时间, d.预计到货日期, d.实际到货日期,
|
||
COALESCE(d.数量, 0) AS 下单数量, COALESCE(d.实际到货数量, 0) AS 实际到货数量,
|
||
COALESCE(d.未到货数, COALESCE(d.数量,0)-COALESCE(d.实际到货数量,0)) AS 未到货数量,
|
||
COALESCE((SELECT SUM(a.defective_qty) FROM arrival_batches a
|
||
WHERE a.order_type='domestic' AND a.order_id=d.id), 0) AS 不良品数量,
|
||
COALESCE(d.状态, '') AS 状态, COALESCE(d.进度, '') AS 进度,
|
||
COALESCE(d.procurement_planner, '') AS 采购计划, COALESCE(d.下单人, '') AS 下单人
|
||
FROM domestic_orders d
|
||
"""):
|
||
rows.append(r)
|
||
for r in query("""
|
||
SELECT '海外订单' AS 订单类型, o.id AS 订单ID, o.产品编码 AS product_code,
|
||
COALESCE(o.货品名称, o.product_name, '') AS 货品名称, COALESCE(o.供应商, '') AS 供应商,
|
||
o.计划下单时间, o.预计到货日期, o.实际到货日期,
|
||
COALESCE(o.数量, 0) AS 下单数量, COALESCE(o.实际到货数量, 0) AS 实际到货数量,
|
||
COALESCE(o.未到货数, COALESCE(o.数量,0)-COALESCE(o.实际到货数量,0)) AS 未到货数量,
|
||
COALESCE((SELECT SUM(a.defective_qty) FROM arrival_batches a
|
||
WHERE a.order_type='overseas' AND a.order_id=o.id), 0) AS 不良品数量,
|
||
COALESCE(o.状态, '') AS 状态, COALESCE(o.进度, '') AS 进度,
|
||
COALESCE(o.procurement_planner, '') AS 采购计划, COALESCE(o.下单人, '') AS 下单人
|
||
FROM overseas_orders o
|
||
"""):
|
||
rows.append(r)
|
||
for r in query("""
|
||
SELECT '包材订单' AS 订单类型, p.id AS 订单ID,
|
||
COALESCE(NULLIF(p.来源产品编码,''), p.编码) AS product_code,
|
||
p.名称 AS 货品名称, COALESCE(p.供应商, '') AS 供应商,
|
||
p.计划下单时间, '' AS 预计到货日期, p.到货日期 AS 实际到货日期,
|
||
COALESCE(p.数量, 0) AS 下单数量, COALESCE(p.实际到货数量, 0) AS 实际到货数量,
|
||
COALESCE(p.未到货数, COALESCE(p.数量,0)-COALESCE(p.实际到货数量,0)) AS 未到货数量,
|
||
COALESCE((SELECT SUM(a.defective_qty) FROM arrival_batches a
|
||
WHERE a.order_type='packaging' AND a.order_id=p.id), 0) AS 不良品数量,
|
||
COALESCE(p.状态, '') AS 状态, COALESCE(p.进度, '') AS 进度,
|
||
COALESCE(p.采购计划, '') AS 采购计划, COALESCE(p.下单人, p.采购下单人, '') AS 下单人
|
||
FROM packaging_orders p
|
||
"""):
|
||
rows.append(r)
|
||
|
||
code = (params.get('product_code') or '').strip().lower()
|
||
if code:
|
||
rows = [r for r in rows if code in str(r.get('product_code') or '').lower()
|
||
or code in str(r.get('货品名称') or '').lower()]
|
||
supplier = (params.get('supplier') or '').strip().lower()
|
||
if supplier:
|
||
rows = [r for r in rows if supplier in str(r.get('供应商') or '').lower()]
|
||
status = (params.get('status') or '').strip().lower()
|
||
if status:
|
||
rows = [r for r in rows if status in str(r.get('状态') or '').lower()]
|
||
if not params.get('_no_limit'):
|
||
rows = _apply_filters(rows, params)
|
||
return {'columns': fields, 'rows': rows}
|
||
|
||
|
||
def _h_supplier_delivery(params):
|
||
"""按供应商汇总订单交付表现，作为P0供应商风险分析的确定性数据底座。"""
|
||
fields = ['供应商', '订单数', '下单数量', '有效到货数量', '未到货数量', '不良品数量',
|
||
'未完结订单数', '到货完成率', '不良品率']
|
||
detail = _h_order_detail({'_no_limit': True}).get('rows') or []
|
||
supplier_alias = {}
|
||
for s in query("SELECT supplier_name, short_name, contract_short_name FROM suppliers"):
|
||
full = (s.get('supplier_name') or '').strip()
|
||
if not full:
|
||
continue
|
||
for alias in (full, s.get('short_name'), s.get('contract_short_name')):
|
||
alias = (alias or '').strip()
|
||
if alias:
|
||
supplier_alias[alias] = full
|
||
agg = {}
|
||
for r in detail:
|
||
raw_supplier = (r.get('供应商') or '（未填写供应商）').strip()
|
||
supplier = supplier_alias.get(raw_supplier, raw_supplier)
|
||
row = agg.setdefault(supplier, {'供应商': supplier, '订单数': 0, '下单数量': 0,
|
||
'有效到货数量': 0, '未到货数量': 0, '不良品数量': 0,
|
||
'未完结订单数': 0})
|
||
ordered = int(r.get('下单数量') or 0)
|
||
arrived = int(r.get('实际到货数量') or 0)
|
||
defective = int(r.get('不良品数量') or 0)
|
||
outstanding = max(0, int(r.get('未到货数量') or 0))
|
||
row['订单数'] += 1
|
||
row['下单数量'] += ordered
|
||
row['有效到货数量'] += max(0, arrived - defective)
|
||
row['未到货数量'] += outstanding
|
||
row['不良品数量'] += defective
|
||
if str(r.get('状态') or '') not in ('已完结', '已完成'):
|
||
row['未完结订单数'] += 1
|
||
rows = []
|
||
for row in agg.values():
|
||
ordered = row['下单数量']
|
||
arrived = row['有效到货数量']
|
||
row['到货完成率'] = round(arrived * 100.0 / ordered, 2) if ordered else 0
|
||
row['不良品率'] = round(row['不良品数量'] * 100.0 / max(arrived + row['不良品数量'], 1), 2)
|
||
rows.append(row)
|
||
supplier = (params.get('supplier') or '').strip().lower()
|
||
if supplier:
|
||
rows = [r for r in rows if supplier in str(r.get('供应商') or '').lower()]
|
||
rows = _apply_filters(rows, params)
|
||
return {'columns': fields, 'rows': rows}
|
||
|
||
|
||
def _h_operation_history(params):
|
||
"""按板块及业务对象检索结构化审计日志，旧记录允许文本兼容匹配。"""
|
||
import json
|
||
import re
|
||
code = str(params.get('product_code') or '').strip()
|
||
product_name = str(params.get('product_name') or '').strip()
|
||
identity = code or product_name
|
||
module = str(params.get('module') or '').strip()
|
||
field = str(params.get('field') or '').strip()
|
||
operator = str(params.get('operator') or '').strip()
|
||
start_date = str(params.get('start_date') or '').strip()
|
||
end_date = str(params.get('end_date') or '').strip()
|
||
try:
|
||
limit = max(1, min(int(params.get('limit') or 100), 200))
|
||
except (TypeError, ValueError):
|
||
limit = 100
|
||
|
||
where = ["l.action <> '登录'",
|
||
"COALESCE(l.event_type,'') NOT IN ('import','export','ai_analysis','ai_report','ai_simulation','login')",
|
||
"COALESCE(l.action,'') NOT LIKE '%%导入%%'"]
|
||
values = []
|
||
allowed_pages = params.get('_allowed_pages')
|
||
if allowed_pages is not None:
|
||
allowed_pages = [str(page) for page in allowed_pages if str(page).strip()]
|
||
if not allowed_pages:
|
||
return {'columns': ['操作时间','操作人','板块','操作','字段','修改内容','匹配方式','结果','请求编号','日志ID'],
|
||
'rows': [], 'summary': {'row_count': 0, 'returned_rows': 0,
|
||
'exact_object_matches': 0, 'text_fallback_matches': 0, 'limit': limit}}
|
||
where.append("COALESCE(NULLIF(l.page,''),l.action) IN (" + ','.join(['%s'] * len(allowed_pages)) + ")")
|
||
values.extend(allowed_pages)
|
||
if code:
|
||
boundary_pattern = r'(^|[^A-Za-z0-9-])' + re.escape(code) + r'($|[^A-Za-z0-9-])'
|
||
where.append("(l.object_id=%s OR l.new_values REGEXP %s OR l.old_values REGEXP %s OR l.detail REGEXP %s)")
|
||
values.extend((code, boundary_pattern, boundary_pattern, boundary_pattern))
|
||
elif product_name:
|
||
where.append("(l.new_values LIKE %s OR l.old_values LIKE %s OR l.detail LIKE %s)")
|
||
values.extend(['%' + product_name + '%'] * 3)
|
||
if module:
|
||
where.append("COALESCE(NULLIF(l.page,''),l.action) LIKE %s")
|
||
values.append('%' + module + '%')
|
||
if field:
|
||
where.append("(COALESCE(l.field,'') LIKE %s OR l.new_values LIKE %s OR l.old_values LIKE %s)")
|
||
values.extend(['%' + field + '%'] * 3)
|
||
if operator:
|
||
where.append("(u.real_name LIKE %s OR u.username LIKE %s)")
|
||
values.extend(['%' + operator + '%'] * 2)
|
||
if start_date:
|
||
where.append('l.created_at >= %s')
|
||
values.append(start_date + ' 00:00:00')
|
||
if end_date:
|
||
where.append('l.created_at < DATE_ADD(%s, INTERVAL 1 DAY)')
|
||
values.append(end_date)
|
||
|
||
where_sql = ' AND '.join(where)
|
||
total_rows = query('SELECT COUNT(*) AS total FROM dashboard_logs l LEFT JOIN dashboard_users u ON u.id=l.user_id WHERE ' + where_sql,
|
||
values)[0].get('total', 0)
|
||
rows = query("""SELECT l.id,l.created_at,l.user_id,l.page,l.action,l.field,l.object_type,l.object_id,
|
||
l.old_values,l.new_values,l.detail,l.result_status,l.request_id,
|
||
u.real_name,u.username
|
||
FROM dashboard_logs l LEFT JOIN dashboard_users u ON u.id=l.user_id
|
||
WHERE """ + where_sql + ' ORDER BY l.created_at DESC,l.id DESC LIMIT %s',
|
||
values + [limit])
|
||
|
||
def decode(value):
|
||
if not value:
|
||
return {}
|
||
try:
|
||
parsed = json.loads(value) if isinstance(value, str) else value
|
||
return scrub(parsed) if isinstance(parsed, (dict, list)) else {'内容': parsed}
|
||
except (TypeError, ValueError):
|
||
return {'历史原文': str(value)[:1000]}
|
||
|
||
def scrub(value):
|
||
sensitive = ('password', 'token', 'secret', 'authorization', 'cookie', 'api_key',
|
||
'密码', '令牌', '密钥', '授权信息')
|
||
if isinstance(value, dict):
|
||
return {key: ('[已隐藏]' if any(mark in str(key).lower() for mark in sensitive)
|
||
else scrub(item)) for key, item in value.items()}
|
||
if isinstance(value, list):
|
||
return [scrub(item) for item in value]
|
||
return value
|
||
|
||
field_labels = {
|
||
'product_code': '货品编号', 'product_name': '货品名称', 'product_type': '货品类型',
|
||
'brand': '品牌', 'supplier': '供应商', 'supplier_code': '供应商编码', 'supplier_name': '供应商名称',
|
||
'planner': '采购计划', 'purchase_planner': '采购计划', 'follower': '采购跟单',
|
||
'purchase_follower': '采购跟单', 'operator': '运营', 'developer': '开发专员',
|
||
'barcode': '条形码', 'category': '类目', 'platform': '平台', 'department': '部门',
|
||
'group_name': '组别', 'status': '状态', 'progress': '进度', 'remark': '备注',
|
||
'order_remark': '下单备注', 'purchase_plan': '采购计划', 'lead_time': '生产货期',
|
||
'logistics_time': '物流时间', 'moq': '最低起订量', 'cost_price': '成本价',
|
||
'price': '单价', 'quantity': '数量', 'order_quantity': '下单数量',
|
||
'inventory': '库存', 'stock': '库存', 'arrival_date': '到货日期',
|
||
'completion_date': '完结日期', 'new_old': '新旧品', 'is_new': '新旧品',
|
||
'id': '记录编号', 'product_codes': '货品编号列表', 'materials': '物料列表',
|
||
'bom_data': 'BOM数据', 'material_code': '物料编号', 'material_name': '物料名称',
|
||
'use_quantity': '用量', 'unit_quantity': '用量', 'old_values': '修改前内容',
|
||
'new_values': '修改后内容', 'value': '字段值', 'field': '字段', 'data': '提交数据',
|
||
'warehouse': '仓库', 'warehouse_code': '仓库编号', 'warehouse_name': '仓库名称',
|
||
'supplier_contract': '供应商合同', 'contract_code': '合同编号', 'purchase_order': '采购单',
|
||
'purchase_plan_name': '采购计划', 'purchase_group': '跟单小组',
|
||
'view': '视图', 'columns': '字段列', 'sort_by': '排序字段', 'sort_order': '排序顺序',
|
||
'width': '列宽', 'visible': '是否显示', 'default_show': '默认显示', 'order': '顺序',
|
||
'settings': '设置', 'config': '配置', 'filters': '筛选条件', 'page': '页码',
|
||
'size': '每页条数', 'limit': '数量上限', 'enabled': '是否启用', 'deleted': '是否删除',
|
||
}
|
||
page_labels = {
|
||
'planning': '计划预测', 'products': '产品信息管理', 'product': '产品信息管理',
|
||
'order_domestic': '国内订单管理', 'order_overseas': '海外订单管理',
|
||
'order_packaging': '包材订单管理', 'purchase_pending': '待下单管理',
|
||
'purchase_backup': '下单备份管理', 'bom': '物料清单管理', 'suppliers': '供应商管理',
|
||
'spot_application': '现货采样申请', 'spot_confirmation': '现货采样确认',
|
||
'spot_sales': '现货销量管理', 'success': '成功', 'failed': '失败', 'denied': '已拒绝',
|
||
}
|
||
|
||
def label_for(key):
|
||
key = str(key)
|
||
if key in field_labels:
|
||
return field_labels[key]
|
||
if key in ('货品编号', '货品名称', '产品编码', '产品名称'):
|
||
return key
|
||
# 常见 camelCase key 先拆分，再保留语义清晰的中文前缀。
|
||
readable = re.sub(r'([a-z0-9])([A-Z])', r'\1_\2', key).replace('-', '_')
|
||
parts = readable.split('_')
|
||
if len(parts) > 1 and parts[0] in ('product', 'material', 'supplier', 'order'):
|
||
roots = {'product': '产品', 'material': '物料', 'supplier': '供应商', 'order': '订单'}
|
||
tails = {'code': '编号', 'name': '名称', 'type': '类型', 'id': '编号', 'quantity': '数量',
|
||
'status': '状态', 'remark': '备注', 'price': '单价', 'date': '日期'}
|
||
return roots[parts[0]] + ''.join(tails.get(part, part) for part in parts[1:])
|
||
return key
|
||
|
||
def format_value(value, field_key=''):
|
||
if field_key in ('field', 'sort_by') and isinstance(value, str):
|
||
return label_for(value)
|
||
if field_key == 'sort_order' and value in ('asc', 'desc'):
|
||
return '升序' if value == 'asc' else '降序'
|
||
if field_key == 'view' and value == 'default':
|
||
return '默认视图'
|
||
if isinstance(value, (dict, list)):
|
||
return json.dumps(translate_data(value), ensure_ascii=False, default=str)
|
||
return str(value)
|
||
|
||
def translate_data(value):
|
||
if isinstance(value, dict):
|
||
translated = {}
|
||
for key, item in value.items():
|
||
if key in ('field', 'sort_by') and isinstance(item, str):
|
||
translated[label_for(key)] = label_for(item)
|
||
else:
|
||
translated[label_for(key)] = translate_data(item)
|
||
return translated
|
||
if isinstance(value, list):
|
||
return [translate_data(item) for item in value]
|
||
return value
|
||
|
||
def diff_values(old_value, new_value, path):
|
||
"""Return leaf-level changes so large submitted objects do not expose unchanged siblings."""
|
||
if isinstance(old_value, dict) and isinstance(new_value, dict):
|
||
changes = []
|
||
for key in dict.fromkeys(list(old_value) + list(new_value)):
|
||
child_path = path + [str(key)]
|
||
if key not in old_value:
|
||
changes.append((child_path, None, new_value[key]))
|
||
elif key not in new_value:
|
||
changes.append((child_path, old_value[key], None))
|
||
else:
|
||
changes.extend(diff_values(old_value[key], new_value[key], child_path))
|
||
return changes
|
||
if isinstance(old_value, list) and isinstance(new_value, list):
|
||
identity_keys = ('product_code', 'material_code', 'product_id', 'id', 'code',
|
||
'货品编号', '物料编号', '产品编码')
|
||
if all(isinstance(item, dict) for item in old_value + new_value):
|
||
def item_key(item, index):
|
||
for identity_key in identity_keys:
|
||
if item.get(identity_key) not in (None, ''):
|
||
return str(item[identity_key])
|
||
return '#%s' % index
|
||
old_items = {item_key(item, index): item for index, item in enumerate(old_value)}
|
||
new_items = {item_key(item, index): item for index, item in enumerate(new_value)}
|
||
changes = []
|
||
for item_id in dict.fromkeys(list(old_items) + list(new_items)):
|
||
item_path = path + [item_id]
|
||
if item_id not in old_items:
|
||
changes.append((item_path, None, new_items[item_id]))
|
||
elif item_id not in new_items:
|
||
changes.append((item_path, old_items[item_id], None))
|
||
else:
|
||
changes.extend(diff_values(old_items[item_id], new_items[item_id], item_path))
|
||
return changes
|
||
if old_value == new_value:
|
||
return []
|
||
return [(path, old_value, new_value)]
|
||
return [] if old_value == new_value else [(path, old_value, new_value)]
|
||
|
||
def display_value(value, field_key=''):
|
||
if value is None:
|
||
return '（空）'
|
||
if isinstance(value, (dict, list)):
|
||
return json.dumps(translate_data(value), ensure_ascii=False, default=str)
|
||
return format_value(value, field_key)
|
||
|
||
result = []
|
||
exact_count = 0
|
||
text_fallback_count = 0
|
||
for row in rows:
|
||
exact = bool(code and str(row.get('object_id') or '').casefold() == code.casefold())
|
||
if identity and not exact:
|
||
pattern = (re.compile(r'(?<![A-Za-z0-9-])' + re.escape(code) + r'(?![A-Za-z0-9-])', re.I)
|
||
if code else re.compile(re.escape(product_name), re.I))
|
||
if not any(pattern.search(str(row.get(key) or '')) for key in ('new_values', 'old_values', 'detail')):
|
||
continue
|
||
exact_count += int(exact)
|
||
text_fallback_count += int(bool(identity and not exact))
|
||
old_values = decode(row.get('old_values'))
|
||
new_values = decode(row.get('new_values'))
|
||
changed = []
|
||
changed_fields = []
|
||
if isinstance(new_values, dict):
|
||
old_map = old_values if isinstance(old_values, dict) else {}
|
||
for key, new_value in new_values.items():
|
||
# A new-only request payload is not evidence that every submitted field changed.
|
||
if key not in old_map:
|
||
continue
|
||
for path, old_value, current_value in diff_values(old_map[key], new_value, [str(key)]):
|
||
label = ' · '.join(label_for(part) for part in path)
|
||
changed_fields.append(label)
|
||
changed.append('%s：%s → %s' % (
|
||
label, display_value(old_value, path[-1]), display_value(current_value, path[-1])))
|
||
created = row.get('created_at')
|
||
if hasattr(created, 'strftime'):
|
||
created = created.strftime('%Y-%m-%d %H:%M:%S')
|
||
result.append({
|
||
'操作时间': str(created or ''),
|
||
'操作人': row.get('real_name') or row.get('username') or ('用户#%s' % row.get('user_id')),
|
||
'板块': page_labels.get(str(row.get('page') or ''), row.get('page') or ''),
|
||
'操作': page_labels.get(str(row.get('action') or ''), row.get('action') or ''),
|
||
'字段': '、'.join(dict.fromkeys(changed_fields)) or label_for(row.get('field') or '请求字段（旧日志未结构化）'),
|
||
'修改内容': '；'.join(changed) if changed else '该记录未保存可比较的字段前后值',
|
||
'匹配方式': '精确对象匹配' if exact else ('历史文本匹配' if identity else '条件匹配'),
|
||
'结果': page_labels.get(str(row.get('result_status') or ''), row.get('result_status') or ''),
|
||
'请求编号': row.get('request_id') or '',
|
||
'日志ID': row.get('id'),
|
||
})
|
||
return {'columns': ['操作时间','操作人','板块','操作','字段','修改内容','匹配方式','结果','请求编号','日志ID'],
|
||
'rows': result,
|
||
'summary': {'row_count': int(total_rows or 0), 'returned_rows': len(result),
|
||
'exact_object_matches': exact_count,
|
||
'text_fallback_matches': text_fallback_count, 'limit': limit}}
|
||
|
||
def _latest_outbound_date():
|
||
"""最新出库日期（销量/出库类数据源的核对日期）"""
|
||
rows = query("SELECT MAX(日期) AS d FROM daily_outbound")
|
||
d = rows[0]['d'] if rows else None
|
||
if hasattr(d, 'strftime'):
|
||
return d.strftime('%Y-%m-%d')
|
||
return str(d)[:10] if d else ''
|
||
|
||
|
||
def datasource_meta(key):
|
||
"""数据源展示信息：名称 + 数据日期（供前端标明口径）"""
|
||
ds = AI_DATASOURCES.get(key) or {}
|
||
kind = ds.get('date_kind') or ''
|
||
date = ''
|
||
if kind == 'inventory':
|
||
date = _get_latest_inv_date()
|
||
elif kind == 'outbound':
|
||
date = _latest_outbound_date()
|
||
return {'key': key, 'name': ds.get('name') or key, 'data_date': date, 'date_kind': kind}
|
||
|
||
|
||
def suggest_datasources(question, limit=3):
|
||
"""按关键词匹配推荐最接近的数据源（AI 选不到数据源时给用户兜底建议）"""
|
||
q = (question or '').lower()
|
||
scored = []
|
||
for idx, (key, ds) in enumerate(AI_DATASOURCES.items()):
|
||
score = 0
|
||
for word in ds.get('keywords') or []:
|
||
w = str(word).lower()
|
||
if w and w in q:
|
||
score += len(w)
|
||
name = str(ds.get('name') or '')
|
||
for token in [name, key]:
|
||
if token and token.lower() in q:
|
||
score += len(token)
|
||
if score > 0:
|
||
scored.append((score, -idx, key, ds.get('name') or key))
|
||
scored.sort(reverse=True)
|
||
return [{'key': k, 'name': n} for _, _, k, n in scored[:limit]]
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 数据源注册表（AI 可见）
|
||
# ──────────────────────────────────────────────
|
||
AI_DATASOURCES = {
|
||
'inventory_cost_summary': {
|
||
'name': '库存成本汇总',
|
||
'desc': '当前库存总成本（按系统数据概览口径：含特殊成本价覆盖、排除内料、成品/包材分类、在途成本）。可回答：库存总额/总成本/库存金额是多少、成品与包材成本各多少。',
|
||
'fields': ['核对日期', '库存总成本(元)', '成品成本(元)', '包材成本(元)', '成品库存(元)', '成品生产在途(元)', '成品采购在途(元)', '包材库存(元)', '包材采购在途(元)'],
|
||
'handler': _h_inventory_cost_summary,
|
||
'permission_pages': ['data_summary'],
|
||
'date_kind': 'inventory',
|
||
'keywords': ['库存总额', '总成本', '库存金额合计', '库存总值', '多少钱', '库存价值'],
|
||
'perm_key': '库存成本汇总',
|
||
},
|
||
'product_inventory': {
|
||
'name': '产品库存视图',
|
||
'desc': '每个产品的最新库存（可用库存、云仓、外仓、加工厂、生产在途、采购在途）。可回答：某产品还有多少库存、库存最多的SKU排行、云仓/外仓库存等。',
|
||
'fields': ['product_code', '库存', '云仓库存', '外仓库存', '加工厂库存', '生产在途', '采购在途'],
|
||
'handler': _h_product_inventory,
|
||
'permission_pages': ['inventory_finished'],
|
||
'date_kind': 'inventory',
|
||
'keywords': ['库存最多', '库存排行', '还有多少库存', '云仓库存', '外仓库存', '加工厂库存', '在途库存'],
|
||
},
|
||
'product_sales': {
|
||
'name': '产品销量统计',
|
||
'desc': '每个产品的销量（昨日销量、3/7/14/30/60天平均销量、各周期最高销量）。可回答：销量最大的SKU、某产品日均销量、昨日销量排行等。',
|
||
'fields': ['product_code', '昨日销量', '3天平均销量', '7天平均销量', '14天平均销量', '30天平均销量', '60天平均销量', '3天最高销量', '7天最高销量', '14天最高销量', '30天最高销量'],
|
||
'handler': _h_product_sales,
|
||
'permission_pages': ['data_summary'],
|
||
'date_kind': 'outbound',
|
||
'keywords': ['销量', '日均销量', '昨日销量', '销量排行', '平均销量'],
|
||
},
|
||
'outbound_stats': {
|
||
'name': '出库统计排行',
|
||
'desc': '出库量排行（kind=brand 按品牌 / kind=shop 按店铺 / 默认按单品TOP）。可回答：出库最多的品牌/店铺/SKU 是哪些。',
|
||
'fields': ['kind(品牌=brand/店铺=shop/单品=product)', 'limit'],
|
||
'handler': _h_outbound_stats,
|
||
'permission_pages': ['data_summary'],
|
||
'date_kind': 'outbound',
|
||
'keywords': ['出库排行', '出库量', 'TOP', '品牌出库', '店铺出库', '出库最多'],
|
||
},
|
||
'outbound_daily': {
|
||
'name': '按日出库统计',
|
||
'desc': '按日期统计出库（最近 N 天，参数 days）。可回答：最近几天每天出库多少。',
|
||
'fields': ['日期', '记录数', '总数量', '总成本'],
|
||
'handler': _h_outbound_daily,
|
||
'permission_pages': ['data_summary'],
|
||
'date_kind': 'outbound',
|
||
'keywords': ['每天出库', '按天出库', '出库趋势', '最近几天', '日出库'],
|
||
},
|
||
'planning_metrics': {
|
||
'name': '计划预测口径',
|
||
'desc': '计划预测页每个商品的计算结果（与页面同一套计算）：日销量预测、总/云仓/外仓支撑天数、库存可支撑天数、云仓与外仓可分配库存、是否调拨、建议下单数量、产品分级、销售趋势、各仓库存与在途。可回答：哪些商品需要调拨、建议下单数量最多的是谁、支撑天数最短的SKU、某商品的日销量预测等。',
|
||
'fields': PLANNING_FIELDS,
|
||
'handler': _h_planning_metrics,
|
||
'permission_pages': ['planning'],
|
||
'date_kind': 'inventory',
|
||
'keywords': ['计划预测', '支撑天数', '建议下单', '调拨', '产品分级', '销售趋势', '日销量预测', '可分配库存'],
|
||
},
|
||
'orders_in_transit': {
|
||
'name': '订单在途与到货',
|
||
'desc': '未完结订单的未到货数量（国内/海外/包材三类，按商品汇总；包材按所属成品归集）。可回答：某商品还有多少货没到、谁的在途最多、哪个商品有未完结订单。',
|
||
'fields': ['product_code', '品牌', '货品名称', '国内未到数量', '海外未到数量', '包材未到数量',
|
||
'在途未到合计', '国内未完结订单数', '海外未完结订单数', '包材未完结订单数'],
|
||
'handler': _h_orders_in_transit,
|
||
'permission_pages': ['planning'],
|
||
'date_kind': '',
|
||
'keywords': ['在途', '未到货', '到货', '未完结订单', '订单'],
|
||
},
|
||
'material_inventory': {
|
||
'name': '包材与内料库存',
|
||
'desc': '包材、内料物料的库存（可用/云仓/外仓/加工厂/生产在途/采购在途）。产品库存视图已把这些物料排除，问包材/内料库存时用本数据源。可回答：某包材（瓶子/彩盒/泵头）还有多少库存、内料库存多少等。',
|
||
'fields': ['product_code', '名称', '类型', '品牌', '可用库存', '云仓库存', '外仓库存',
|
||
'加工厂库存', '生产在途', '采购在途'],
|
||
'handler': _h_material_inventory,
|
||
'permission_pages': ['inventory_packaging'],
|
||
'date_kind': 'inventory',
|
||
'keywords': ['包材', '内料', '瓶子', '彩盒', '泵头', '标贴', '收缩膜'],
|
||
},
|
||
'inventory_value': {
|
||
'name': '商品库存金额',
|
||
'desc': '逐商品的库存金额（库存 × 成本价，含特殊成本价规则，排除已删除产品、内料与包材）。summary提供独立于limit和排序的全量SKU数、库存数量及金额，含零库存商品。可回答：某品牌库存总金额、库存金额最高的商品、某商品库存值多少钱。注意：含包材及在途的全库合计请用「库存成本汇总」数据源。',
|
||
'fields': VALUE_FIELDS,
|
||
'handler': _h_inventory_value,
|
||
'permission_pages': ['inventory_finished'],
|
||
'date_kind': 'inventory',
|
||
'keywords': ['库存金额', '金额排行', '库存值', '成本价', '库存金额最高'],
|
||
'perm_key': '商品库存金额',
|
||
},
|
||
'product_profile': {
|
||
'name': '产品基础档案',
|
||
'desc': '产品编码、名称、品牌、产品类型、货期、物流时间、供应商和负责人等基础信息，不返回成本价。可用于产品属性、供应商、货期和责任人分析。',
|
||
'fields': ['product_code', '货品名称', '品牌', '产品类型', '供应商', '生产货期', '物流时间', '采购计划', '采购跟单', '运营', '开发', '创建时间'],
|
||
'handler': _h_product_profile,
|
||
'permission_pages': ['products'],
|
||
'date_kind': '',
|
||
'keywords': ['产品信息', '产品档案', '货期', '物流时间', '供应商', '负责人', '创建时间'],
|
||
},
|
||
'bom_supply_chain': {
|
||
'name': 'BOM物料供应链',
|
||
'desc': '成品与BOM物料、包材库存、在途及可支撑成品数量的联查数据。可识别成品下单后包材是否不足。',
|
||
'fields': ['product_code', '成品名称', '品牌', 'material_code', '物料名称', '单位用量', '物料供应商', '物料库存', '加工厂库存', '生产在途', '采购在途', '可支撑成品数量'],
|
||
'handler': _h_bom_supply_chain,
|
||
'permission_pages': ['bom', 'inventory_packaging'],
|
||
'date_kind': 'inventory',
|
||
'keywords': ['BOM', '物料清单', '包材缺口', '物料不足', '包材支撑', '成品包材'],
|
||
},
|
||
'order_detail': {
|
||
'name': '订单明细与交付',
|
||
'desc': '国内、海外、包材订单统一明细，包含下单、到货、不良品、未到货、供应商和订单状态。',
|
||
'fields': ['订单类型', '订单ID', 'product_code', '货品名称', '供应商', '计划下单时间', '预计到货日期', '实际到货日期', '下单数量', '实际到货数量', '未到货数量', '不良品数量', '状态', '进度'],
|
||
'handler': _h_order_detail,
|
||
'permission_pages': ['order_domestic', 'order_overseas', 'order_packaging'],
|
||
'date_kind': '',
|
||
'keywords': ['订单明细', '订单交付', '不良品', '延期', '订单状态', '到货完成率'],
|
||
},
|
||
'supplier_delivery': {
|
||
'name': '供应商交付表现',
|
||
'desc': '按供应商汇总订单数、下单数量、有效到货、不良品、未到货和未完结订单，可用于供应商交付风险排行。',
|
||
'fields': ['供应商', '订单数', '下单数量', '有效到货数量', '未到货数量', '不良品数量', '未完结订单数', '到货完成率', '不良品率'],
|
||
'handler': _h_supplier_delivery,
|
||
'permission_pages': ['suppliers', 'order_domestic', 'order_overseas', 'order_packaging'],
|
||
'date_kind': '',
|
||
'keywords': ['供应商表现', '供应商交付', '供应商排名', '交付率', '不良品率', '供应商风险'],
|
||
},
|
||
'operation_history': {
|
||
'name': '手动修改记录',
|
||
'desc': '只读检索结构化操作日志，按板块、货品编号、字段、操作人和日期查询；精确对象匹配与旧日志文本匹配会分别标注。需要操作日志菜单查看权限。可回答某商品谁在何时修改了哪些内容。',
|
||
'fields': ['操作时间','操作人','板块','操作','字段','修改内容','匹配方式','结果','请求编号','日志ID'],
|
||
'handler': _h_operation_history,
|
||
'permission_pages': ['logs'],
|
||
'date_kind': '',
|
||
'keywords': ['操作日志','修改记录','编辑记录','手动修改','修改历史','变更历史','谁修改','修改过','修改前','修改后'],
|
||
},
|
||
'brand_overview': {
|
||
'name': '品牌汇总',
|
||
'desc': '按品牌汇总的 SKU数、库存数量、库存金额、昨日销量、30天平均销量与近30/100天动销SKU数（默认按库存金额倒序）。可回答：哪个品牌库存金额最高、某品牌有多少个动销SKU、各品牌销量对比。',
|
||
'fields': BRAND_FIELDS,
|
||
'handler': _h_brand_overview,
|
||
'permission_pages': ['data_summary'],
|
||
'date_kind': 'inventory',
|
||
'keywords': ['品牌', '动销', '品牌汇总', '品牌对比', '品牌库存'],
|
||
'perm_key': '品牌汇总',
|
||
},
|
||
}
|
||
|
||
# AI 可用的数据源清单文本（注入提示词）
|
||
AI_DATASOURCES_DESC = "\n".join(
|
||
f"- {key}（{ds['name']}）：{ds['desc']}"
|
||
for key, ds in AI_DATASOURCES.items()
|
||
)
|