kx_guo/dy-dashboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: unify planning support days across detail and export

Browse Source
This commit is contained in:
Codex 2026-10-12 15:29:10 +08:00
parent 8c229bc162
commit ae0cae8cbf
3 changed files with 1524 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1461
routes/planning.py Normal file
View File

File diff suppressed because it is too large Load Diff

138
routes/products.py
View File

@ -851,126 +851,39 @@ def product_planning_detail():
product['产品种类'] = get_product_type_cached(product.get('product_name', ''), product.get('brand', ''), product.get('product_type'))
product['批注'] = product.get('remark') or ''
# 计算计划预测字段
# 复用计划预测列表/导出的唯一计算逻辑，避免详情弹窗另维护一套公式。
from routes.planning import _enrich_row_with_planning_data
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()}
from services.forecast import get_global_config
gc = get_global_config()
row = product
pc = row.get('product_code', '')
row['库存'] = inv_map.get(pc, 0)
row['加工厂库存'] = inv15_map.get(pc, 0)
t = transit_map.get(pc, {})
row['生产在途'] = t.get('生产在途', 0)
row['采购在途'] = t.get('采购在途', 0)
if pc in sales:
s = sales[pc]
for k in ['昨日销量', '3天平均销量', '7天平均销量', '14天平均销量',
'30天平均销量', '60天平均销量', '3天最高销量', '7天最高销量',
'14天最高销量', '30天最高销量', '14天最低销量', '14天标准差',
'最低销量偏差率', '最高销量偏差率', '极值判断', '14天波动率', '波动风险']:
row[k] = s.get(k, 0)
row['销售趋势'] = _determine_trend(s['3天平均销量'], s['7天平均销量'], s['14天平均销量'])
else:
for k in ['昨日销量', '3天平均销量', '7天平均销量', '14天平均销量',
'30天平均销量', '60天平均销量', '3天最高销量', '7天最高销量',
'14天最高销量', '30天最高销量', '14天最低销量', '14天标准差',
'最低销量偏差率', '最高销量偏差率', '极值判断', '14天波动率', '波动风险']:
row[k] = 0
row['销售趋势'] = _determine_trend(0, 0, 0)
row['产品分级'] = get_product_grade(row.get('30天平均销量', 0) * 30,
row.get('product_name', ''),
row.get('产品种类', ''))
row['是否新品'] = is_new_product(row.get('created_at', ''))
row['日销量预测'] = compute_daily_forecast(row, sp_map)
# 支撑天数
from services.forecast import get_global_config
gc = get_global_config()
safety_days = int(gc.get('safety_stock_days', '7'))
return_days = int(gc.get('suggested_return_days', '15'))
forecast_val = row.get('日销量预测', 0)
fcast = float(forecast_val) if forecast_val not in ('近期无销售', 0, '0') else 0
inv_val = float(row.get('库存', 0) or 0)
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,
)
inv15_val = float(row.get('加工厂库存', 0) or 0)
prod_transit = float(row.get('生产在途', 0) or 0)
purch_transit = float(row.get('采购在途', 0) or 0)
if fcast == 0:
row['库存可支撑天数'] = 999
row['库存工厂可支撑天数'] = 999
row['库存工厂待生产可支撑天数'] = 999
row['加工厂库存支撑天数'] = 999
row['总支撑天数'] = 999
else:
row['库存可支撑天数'] = round(inv_val / fcast, 1) if fcast > 0 else 999
row['库存工厂可支撑天数'] = round(purch_transit / fcast, 1) if fcast > 0 else 999
row['库存工厂待生产可支撑天数'] = round(prod_transit / fcast, 1) if fcast > 0 else 999
row['加工厂库存支撑天数'] = round(inv15_val / fcast, 1) if fcast > 0 else 999
total_inv = prod_transit + inv15_val
row['总支撑天数'] = round(total_inv / fcast, 1) if fcast > 0 else 999
lead_h = float(row.get('lead_time', 0) or 0)
logis_h = float(row.get('logistics_time', 0) or 0)
sup_days = row.get('库存可支撑天数', 999)
sup_days_val = float(sup_days) if isinstance(sup_days, (int, float)) and sup_days != 999 else None
if sup_days_val is not None and (lead_h + logis_h) > 0:
row['库存健康指数'] = round(min(sup_days_val / (lead_h + logis_h), 999), 1)
else:
row['库存健康指数'] = 999
created = row.get('created_at', '')
days_since_created = 999
try:
if created:
created_dt = datetime.strptime(created[:10], '%Y-%m-%d') if isinstance(created, str) else created
days_since_created = (datetime.now() - created_dt).days
except:
pass
sup_days_val_f = float(sup_days) if isinstance(sup_days, (int, float)) and sup_days != 999 else 999
if days_since_created < 90:
row['风险把控'] = '新品'
elif days_since_created > 180 and sup_days_val_f > 240:
row['风险把控'] = '需启动清库存'
elif days_since_created > 90 and sup_days_val_f > 90:
row['风险把控'] = '呆滞品'
else:
row['风险把控'] = '正常'
if fcast == 0 and inv_val > 0:
row['建议下单数量'] = 0
else:
row['建议下单数量'] = int(fcast * (safety_days + lead_h + logis_h))
row['实际下单数量'] = ((row['建议下单数量'] + 999) // 1000) * 1000 if row['建议下单数量'] > 0 else 0
total_transit = prod_transit + purch_transit
if fcast > 0 and total_transit > 0:
calc_val = fcast * return_days - inv_val
round_up = max(0, int(calc_val * 10 + 0.99999) / 10)
row['建议回货数量'] = round_up if round_up <= total_transit else total_transit
else:
row['建议回货数量'] = 0
# 云仓页签沿用默认数据的字段顺序，数值改用云仓库存和云仓销量口径。
cloud_sales = compute_product_sales_extended_cached().get('cloud', {}).get(pc, {})
cloud_inventory = compute_cloud_inventory_cached().get(pc, 0)
cloud_sales = extended_sales.get('cloud', {}).get(pc, {})
cloud_inventory = row.get('云仓库存', 0)
cloud_3 = cloud_sales.get('3天平均销量', 0)
cloud_7 = cloud_sales.get('7天平均销量', 0)
cloud_14 = cloud_sales.get('14天平均销量', 0)
cloud_30 = cloud_sales.get('30天平均销量', 0)
cloud_trend = _determine_trend(cloud_3, cloud_7, cloud_14)
cloud_forecast_row = {
'3天平均销量': cloud_3,
'7天平均销量': cloud_7,
'14天平均销量': cloud_14,
'30天最高销量': cloud_sales.get('30天最高销量', 0),
'销售趋势': cloud_trend,
}
cloud_forecast = compute_daily_forecast(cloud_forecast_row, sp_map)
cloud_forecast = row.get('云仓日销量预测', 0)
cloud_fcast = float(cloud_forecast) if cloud_forecast not in ('近期无销售', 0, '0') else 0
row['云仓数据'] = {
'产品分级': row.get('产品分级', ''),
@ -986,26 +899,19 @@ def product_planning_detail():
'14天平均销量': cloud_14,
'库存可支撑天数': round(float(cloud_inventory or 0) / cloud_fcast, 1) if cloud_fcast else 999,
'加工厂库存支撑天数': round(inv15_val / cloud_fcast, 1) if cloud_fcast else 999,
'总支撑天数': round((prod_transit + inv15_val) / cloud_fcast, 1) if cloud_fcast else 999,
'总支撑天数': row.get('云仓支撑天数', 999),
'30天平均销量': cloud_30,
'极值判断': '-',
}
external_sales = compute_product_sales_extended_cached().get('external', {}).get(pc, {})
external_inventory = compute_external_inventory_cached().get(pc, 0)
external_sales = extended_sales.get('external', {}).get(pc, {})
external_inventory = row.get('外仓库存', 0)
external_3 = external_sales.get('3天平均销量', 0)
external_7 = external_sales.get('7天平均销量', 0)
external_14 = external_sales.get('14天平均销量', 0)
external_30 = external_sales.get('30天平均销量', 0)
external_trend = _determine_trend(external_3, external_7, external_14)
external_forecast_row = {
'3天平均销量': external_3,
'7天平均销量': external_7,
'14天平均销量': external_14,
'30天最高销量': external_sales.get('30天最高销量', 0),
'销售趋势': external_trend,
}
external_forecast = compute_daily_forecast(external_forecast_row, sp_map)
external_forecast = row.get('外仓日销量预测', 0)
external_fcast = float(external_forecast) if external_forecast not in ('近期无销售', 0, '0') else 0
row['外仓数据'] = {
'产品分级': row.get('产品分级', ''),
@ -1021,7 +927,7 @@ def product_planning_detail():
'14天平均销量': external_14,
'库存可支撑天数': round(float(external_inventory or 0) / external_fcast, 1) if external_fcast else 999,
'加工厂库存支撑天数': round(inv15_val / external_fcast, 1) if external_fcast else 999,
'总支撑天数': round((prod_transit + inv15_val) / external_fcast, 1) if external_fcast else 999,
'总支撑天数': row.get('外仓支撑天数', 999),
'30天平均销量': external_30,
'极值判断': '-',
}

41
tests/test_planning_support_days_consistency.py Normal file
View File

@ -0,0 +1,41 @@
"""计划预测支撑天数统一口径回归测试。"""
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
PRODUCTS_PY = (ROOT / "routes" / "products.py").read_text(encoding="utf-8")
PLANNING_PY = (ROOT / "routes" / "planning.py").read_text(encoding="utf-8")
class PlanningSupportDaysConsistencyTest(unittest.TestCase):
def test_detail_reuses_the_list_and_export_enrichment(self):
detail_source = PRODUCTS_PY.split("def product_planning_detail():", 1)[1].split(
"@products_bp.route('/api/bom/list')", 1
)[0]
self.assertIn("from routes.planning import _enrich_row_with_planning_data", detail_source)
self.assertIn("row = _enrich_row_with_planning_data(", detail_source)
self.assertNotIn("total_inv = prod_transit + inv15_val", detail_source)
def test_cloud_and_external_detail_use_shared_support_days(self):
detail_source = PRODUCTS_PY.split("def product_planning_detail():", 1)[1].split(
"@products_bp.route('/api/bom/list')", 1
)[0]
self.assertIn("'总支撑天数': row.get('云仓支撑天数', 999)", detail_source)
self.assertIn("'总支撑天数': row.get('外仓支撑天数', 999)", detail_source)
def test_export_accepts_real_support_day_field_names(self):
export_source = PLANNING_PY.split("def planning_export():", 1)[1].split(
"def planning_page():", 1
)[0]
for field in (
"库存工厂可支撑天数",
"加工厂库存支撑天数",
"库存工厂待生产可支撑天数",
):
self.assertIn(field, export_source)
if __name__ == "__main__":
unittest.main()