"""计划预测路由（最肥模块）""" from datetime import datetime from flask import Blueprint, request, jsonify, session, redirect, url_for, render_template from services.db import query, add_operation_log from services.product_type import get_product_type_cached from services.sales_calc import compute_product_sales_cached, compute_product_sales_extended_cached from services.inventory_calc import (compute_product_inventory_cached, compute_product_inventory_15, compute_product_in_transit_cached, compute_cloud_inventory_cached, compute_external_inventory_cached, _determine_trend) from services.forecast import (is_new_product, get_product_grade, compute_daily_forecast, sales_predict_config_cached, get_global_config) from routes.auth import login_required, role_required, operation_required, _ensure_personnel_table from services.field_perms import filter_rows as _filter_field_rows import json planning_bp = Blueprint('planning', __name__) @planning_bp.route('/api/planning/new-old/save', methods=['POST']) @login_required def planning_new_old_save(): from services.field_perms import field_perms_for_page allowed = operation_required('planning', 'edit') perms = field_perms_for_page(session.get('user_id'), session.get('role', ''), 'planning', page_edit=allowed) if not allowed or not perms.get('新旧品', {}).get('edit'): return jsonify({'code': 403, 'msg': '没有编辑新旧品字段的权限'}), 403 data = request.get_json(silent=True) or {} code, value = data.get('product_code'), data.get('value') if not code or value not in ('是', '否'): return jsonify({'code': 400, 'msg': '新旧品只能选择是或否'}), 400 rows = query('SELECT planning_new_old FROM products WHERE product_code=%s AND deleted_at IS NULL', (code,)) if not rows: return jsonify({'code': 404, 'msg': '未找到该产品'}), 404 query('UPDATE products SET planning_new_old=%s WHERE product_code=%s', (value, code)) add_operation_log('计划预测编辑', f'{code} 新旧品：{rows[0].get("planning_new_old") or "否"} → {value}', page='planning', field='新旧品') return jsonify({'code': 0, 'msg': '已保存'}) def _personnel_group_for_value(raw_value, groups): """Infer a group's label from a stored person list without cross-group leakage.""" names = [name.strip().casefold() for name in str(raw_value or '').split(',') if name.strip()] if not names: return '' normalized_groups = { str(group_name).casefold(): {str(member).strip().casefold() for member in members} for group_name, members in groups.items() } if len(names) == 1 and names[0] in normalized_groups: return names[0] counts = {group_name: sum(1 for name in names if name in members) for group_name, members in normalized_groups.items()} counts = {group_name: count for group_name, count in counts.items() if count} if not counts: return '' best = max(counts.values()) winners = [name for name, count in counts.items() if count == best] return winners[0] if len(winners) == 1 and best * 2 >= len(names) else '' def _personnel_group_members(): """Read the editable personnel groups used by Personnel Management.""" _ensure_personnel_table() rows = query("SELECT group_name, members_json FROM personnel_groups ORDER BY group_order, group_name") groups = {} for row in rows: try: members = json.loads(row.get('members_json') or '[]') except (TypeError, ValueError): members = [] groups[row['group_name']] = [str(member).strip() for member in members if str(member).strip()] return groups def _personnel_value_matches_group(raw_value, selected_value, groups): """Match a product's stored follower value against a group or person.""" raw_names = {name.strip().casefold() for name in str(raw_value or '').split(',') if name.strip()} selected = str(selected_value or '').strip().casefold() if not selected: return True for group_name, members in groups.items(): if group_name.casefold() == selected: member_names = {str(member).strip().casefold() for member in members} return bool(raw_names & member_names) or selected in raw_names return selected in raw_names def _ensure_global_config(): """确保全局配置已加载""" gc = get_global_config() return gc def _enrich_row_with_planning_data(row, sales, inv_map, inv15_map, transit_map, sp_map, gc, cloud_inv_map=None, ext_inv_map=None, cloud_sales_ext=None, ext_sales_ext=None, extended_sales=None): """给一行产品数据补充计划预测相关的所有计算字段""" pc = row.get('product_code', '') row['新旧品'] = row.get('planning_new_old') or '否' inv_val = inv_map.get(pc, 0) inv15_val = inv15_map.get(pc, 0) t = transit_map.get(pc, {}) prod_transit = t.get('生产在途', 0) purch_transit = t.get('采购在途', 0) row['总库存'] = inv_val + inv15_val + prod_transit + purch_transit row['库存'] = inv_val row['总在途库存'] = inv15_val + prod_transit + purch_transit row['云仓库存'] = cloud_inv_map.get(pc, 0) if cloud_inv_map else 0 row['外仓库存'] = ext_inv_map.get(pc, 0) if ext_inv_map else 0 # 云仓/外仓销量平均值 def _safe_get(d, pc, key): return (d.get(pc) or {}).get(key, 0) if d else 0 ext_data = extended_sales or {} cs = ext_data.get('cloud', {}) es = ext_data.get('external', {}) row['云仓昨日销量'] = _safe_get(cs, pc, '昨日销量') row['外仓昨日销量'] = _safe_get(es, pc, '昨日销量') row['云仓3天平均销量'] = _safe_get(cs, pc, '3天平均销量') row['外仓3天平均销量'] = _safe_get(es, pc, '3天平均销量') row['云仓7天平均销量'] = _safe_get(cs, pc, '7天平均销量') row['外仓7天平均销量'] = _safe_get(es, pc, '7天平均销量') row['云仓14天平均销量'] = _safe_get(cs, pc, '14天平均销量') row['外仓14天平均销量'] = _safe_get(es, pc, '14天平均销量') 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) # 云仓/外仓可分配库存：按「云仓外仓权重参数」计算 # X = 云仓销量/昨日销量，Y = 外仓销量/昨日销量（昨日销量为0时权重按0计） # 云仓可分配库存 = 总库存 * X - 云仓库存；外仓可分配库存 = 总库存 * Y - 外仓库存 total_inv_val = float(row.get('总库存', 0) or 0) yesterday_val = float(row.get('昨日销量', 0) or 0) cloud_yesterday_val = float(row.get('云仓昨日销量', 0) or 0) ext_yesterday_val = float(row.get('外仓昨日销量', 0) or 0) weight_x = (cloud_yesterday_val / yesterday_val) if yesterday_val > 0 else 0 weight_y = (ext_yesterday_val / yesterday_val) if yesterday_val > 0 else 0 row['云仓可分配库存'] = round(total_inv_val * weight_x - float(row.get('云仓库存', 0) or 0), 1) row['外仓可分配库存'] = round(total_inv_val * weight_y - float(row.get('外仓库存', 0) or 0), 1) # 是否调拨：按可分配库存自动判定（两边都不大于0时留空） cloud_alloc_ok = row['云仓可分配库存'] > 0 ext_alloc_ok = row['外仓可分配库存'] > 0 if cloud_alloc_ok and ext_alloc_ok: row['是否调拨'] = '云仓和外仓' elif cloud_alloc_ok: row['是否调拨'] = '云仓' elif ext_alloc_ok: row['是否调拨'] = '外仓' else: row['是否调拨'] = '' row['产品分级'] = get_product_grade(row.get('30天平均销量', 0) * 30, row.get('product_name', ''), row.get('产品种类', row.get('product_type', ''))) # 是否新品 row['是否新品'] = is_new_product(row.get('创建时间', '')) # 日销量预测 + 各种支撑天数计算 row['日销量预测'] = compute_daily_forecast(row, sp_map) # 云仓/外仓日销量预测（构建临时row对象传入） def _cloud_forecast(val_3, val_7, val_14, val_30, trend_val): tmp = { '3天平均销量': val_3, '7天平均销量': val_7, '14天平均销量': val_14, '30天最高销量': val_30, '销售趋势': trend_val } f = compute_daily_forecast(tmp, sp_map) return f row['云仓日销量预测'] = _cloud_forecast( row.get('云仓3天平均销量', 0), row.get('云仓7天平均销量', 0), row.get('云仓14天平均销量', 0), _safe_get(cs, pc, '30天最高销量'), _determine_trend(row.get('云仓3天平均销量', 0), row.get('云仓7天平均销量', 0), row.get('云仓14天平均销量', 0)) ) row['外仓日销量预测'] = _cloud_forecast( row.get('外仓3天平均销量', 0), row.get('外仓7天平均销量', 0), row.get('外仓14天平均销量', 0), _safe_get(es, pc, '30天最高销量'), _determine_trend(row.get('外仓3天平均销量', 0), row.get('外仓7天平均销量', 0), row.get('外仓14天平均销量', 0)) ) _calc_support_days(row, gc) # 云仓/外仓支撑天数 =（本仓库存 + 本仓可分配库存）/ 本仓日销量预测 def _num(v): try: return float(v) except (TypeError, ValueError): return 0.0 cloud_forecast = _num(row.get('云仓日销量预测', 0)) ext_forecast = _num(row.get('外仓日销量预测', 0)) cloud_days_base = _num(row.get('云仓库存', 0)) + _num(row.get('云仓可分配库存', 0)) ext_days_base = _num(row.get('外仓库存', 0)) + _num(row.get('外仓可分配库存', 0)) row['云仓支撑天数'] = round(cloud_days_base / cloud_forecast, 1) if cloud_forecast > 0 else 999 row['外仓支撑天数'] = round(ext_days_base / ext_forecast, 1) if ext_forecast > 0 else 999 _calc_suggested_qty(row, gc) return row def _calc_support_days(row, gc): """计算库存支撑天数、健康指数、建议下单等""" 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) 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 = inv_val + inv15_val + prod_transit + purch_transit row['总支撑天数'] = round(total_inv / fcast, 1) if fcast > 0 else 999 # 库存健康指数 sup_days = row.get('库存可支撑天数', 999) s1 = sup_days if isinstance(sup_days, (int, float)) and sup_days != 999 else None lead_h = float(row.get('生产货期', 0) or 0) logis_h = float(row.get('物流时间', 0) or 0) if s1 is not None and (lead_h + logis_h) > 0: health = s1 / (lead_h + logis_h) row['库存健康指数'] = round(min(health, 999), 1) else: row['库存健康指数'] = 999 # 风险把控 created = row.get('创建时间', '') sup_days_val = row.get('库存可支撑天数', 999) try: sup_days_val = float(sup_days_val) except: sup_days_val = 999 days_since_created = 999 try: if created: if isinstance(created, str): created_dt = datetime.strptime(created[:10], '%Y-%m-%d') else: created_dt = created days_since_created = (datetime.now() - created_dt).days except: pass if days_since_created < 90: row['风险把控'] = '新品' elif days_since_created > 180 and sup_days_val > 240: row['风险把控'] = '需启动清库存' elif days_since_created > 90 and sup_days_val > 90: row['风险把控'] = '呆滞品' else: row['风险把控'] = '正常' # 建议下单数量 / 实际下单数量：需等云仓、外仓支撑天数算完后由 _calc_suggested_qty 计算 # 建议回货数量 total_transit = prod_transit + purch_transit if fcast > 0 and total_transit > 0: calc_val = fcast * return_days - inv_val round_up = int(calc_val * 10 + 0.99999) / 10 if round_up < 0: round_up = 0 row['建议回货数量'] = round_up if round_up <= total_transit else total_transit if row['建议回货数量'] < 0: row['建议回货数量'] = 0 else: row['建议回货数量'] = 0 def _calc_suggested_qty(row, gc): """按日销量预测、总支撑天数和产品类型计算建议下单数量。""" safety_days = int(gc.get('safety_stock_days', '7')) def _num(v): try: return float(v) except (TypeError, ValueError): return 0.0 lead_days = _num(row.get('生产货期', 0)) logistics_days = _num(row.get('物流时间', 0)) threshold = lead_days + logistics_days + safety_days forecast_val = row.get('日销量预测', 0) daily_forecast = 0.0 if forecast_val in ('近期无销售', 0, '0', None) else _num(forecast_val) total_support_days = _num(row.get('总支撑天数', 999)) product_type = str(row.get('product_type') or '').strip() if daily_forecast == 0 or total_support_days >= threshold: qty = 0 elif product_type == '现货': qty = int(daily_forecast * (lead_days + logistics_days)) else: qty = int(daily_forecast * threshold) row['建议下单数量'] = qty # 等同 Excel MROUND(建议下单数量, 500)：四舍五入到最接近的 500 倍数 row['实际下单数量'] = ((qty + 250) // 500) * 500 if qty > 0 else 0 return row def _merge_pending_formal_qty(rows): """从 pending_orders 表合并正式下单数量到规划数据行""" if not rows: return codes = [r.get('product_code', '') for r in rows if r.get('product_code')] if not codes: return placeholders = ','.join(['%s'] * len(codes)) pending = query(f"SELECT product_code, formal_order_qty FROM pending_orders WHERE status IN ('pending','draft') AND product_code IN ({placeholders})", codes) qty_map = {r['product_code']: int(r['formal_order_qty'] or 0) for r in pending} for row in rows: pc = row.get('product_code', '') if pc in qty_map and qty_map[pc] > 0: row['正式下单数量'] = qty_map[pc] # ─── 列宽与列方案管理 ─── ORDER_WIDTH_PAGES = {'planning', 'purchase', 'purchase_ready', 'orders_finished', 'orders_domestic', 'orders_overseas', 'orders_packaging', 'new_products'} @planning_bp.route('/api/planning/column-widths') @login_required def planning_column_widths(): page = request.args.get('page', '') if page not in ORDER_WIDTH_PAGES: return jsonify({"code": 400, "msg": "页面参数错误"}) rows = query( "SELECT field_name, width FROM user_column_widths WHERE user_id=%s AND page=%s", (session['user_id'], page) ) widths = {row['field_name']: int(row['width']) for row in rows} return jsonify({"code": 0, "data": widths}) @planning_bp.route('/api/planning/column-widths/save', methods=['POST']) @login_required def planning_column_widths_save(): data = request.get_json(silent=True) or {} page = data.get('page', '') widths = data.get('widths') if page not in ORDER_WIDTH_PAGES: return jsonify({"code": 400, "msg": "页面参数错误"}) if not isinstance(widths, dict) or not widths: return jsonify({"code": 400, "msg": "列宽数据不能为空"}) if len(widths) > 100: return jsonify({"code": 400, "msg": "列宽字段数量超出限制"}) cleaned = [] for field_name, width in widths.items(): field_name = str(field_name).strip() try: width = int(round(float(width))) except (TypeError, ValueError): return jsonify({"code": 400, "msg": "列宽格式错误"}) if not field_name or len(field_name) > 100 or width < 60 or width > 2000: return jsonify({"code": 400, "msg": "列宽数据超出允许范围"}) cleaned.append((session['user_id'], page, field_name, width)) placeholders = ','.join(['(%s,%s,%s,%s)'] * len(cleaned)) params = tuple(value for row in cleaned for value in row) query( "INSERT INTO user_column_widths (user_id, page, field_name, width) VALUES " + placeholders + " ON DUPLICATE KEY UPDATE width=VALUES(width), updated_at=CURRENT_TIMESTAMP", params ) return jsonify({"code": 0, "msg": "保存成功"}) @planning_bp.route('/api/planning/schemes') @login_required def planning_schemes(): user_id = session['user_id'] page = request.args.get('page') if page: schemes = query( "SELECT id, user_id, page, scheme_name, columns_json, is_default, created_at, updated_at " "FROM user_column_schemes WHERE user_id=%s AND page=%s ORDER BY is_default DESC, id ASC", (user_id, page) ) else: schemes = query( "SELECT id, user_id, page, scheme_name, columns_json, is_default, created_at, updated_at " "FROM user_column_schemes WHERE user_id=%s ORDER BY is_default DESC, id ASC", (user_id,) ) return jsonify({"code": 0, "data": schemes}) @planning_bp.route('/api/planning/scheme/save', methods=['POST']) @login_required def planning_scheme_save(): user_id = session['user_id'] data = request.json scheme_name = data.get('scheme_name', '').strip() columns_json = data.get('columns_json', '') page = data.get('page', 'planning') is_default = int(data.get('is_default', 0)) if not scheme_name: return jsonify({"code": 400, "msg": "方案名称不能为空"}) if not columns_json: return jsonify({"code": 400, "msg": "字段配置不能为空"}) try: existing = query( "SELECT id FROM user_column_schemes WHERE user_id=%s AND page=%s AND scheme_name=%s", (user_id, page, scheme_name) ) if existing: query( "UPDATE user_column_schemes SET columns_json=%s, is_default=%s WHERE id=%s", (columns_json, is_default, existing[0]['id']) ) scheme_id = existing[0]['id'] else: scheme_id = query( "INSERT INTO user_column_schemes (user_id, page, scheme_name, columns_json, is_default) " "VALUES (%s, %s, %s, %s, %s)", (user_id, page, scheme_name, columns_json, is_default) ) return jsonify({"code": 0, "msg": "保存成功", "data": {"id": scheme_id}}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @planning_bp.route('/api/planning/scheme/delete', methods=['POST']) @login_required def planning_scheme_delete(): user_id = session['user_id'] data = request.json scheme_id = data.get('id') if not scheme_id: return jsonify({"code": 400, "msg": "参数错误"}) query("DELETE FROM user_column_schemes WHERE id=%s AND user_id=%s", (scheme_id, user_id)) return jsonify({"code": 0, "msg": "删除成功"}) @planning_bp.route('/api/planning/scheme/set_default', methods=['POST']) @login_required def planning_scheme_set_default(): user_id = session['user_id'] data = request.json scheme_id = data.get('id') page = data.get('page', 'planning') if not scheme_id: return jsonify({"code": 400, "msg": "参数错误"}) try: query("UPDATE user_column_schemes SET is_default=0 WHERE user_id=%s AND page=%s", (user_id, page)) query("UPDATE user_column_schemes SET is_default=1 WHERE id=%s AND user_id=%s", (scheme_id, user_id)) return jsonify({"code": 0, "msg": "设置成功"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) # ─── API: 计划预测主列表（492行核心） ─── @planning_bp.route('/api/planning/list') @login_required def planning_list(): # 加载全局配置 gc = _ensure_global_config() page = int(request.args.get('page', 1)) size = int(request.args.get('size', 50)) offset = (page - 1) * size search = request.args.get('search', '') prod_code = request.args.get('prod_code', '') prod_name = request.args.get('prod_name', '') brand_filter = request.args.get('brand', '') type_filter = request.args.get('product_type', '') category_filter = request.args.get('category', '') trend_filter = request.args.get('trend', '') supplier_filter = request.args.get('supplier', '') operator_filter = request.args.get('operator', '') platform_filter = request.args.get('platform', '') planner_filter = request.args.get('procurement_planner', '') follower_filter = request.args.get('procurement_follower', '') # The planning page uses personnel groups as the display/filter vocabulary. GROUP_TO_MEMBERS = _personnel_group_members() view_filter = request.args.get('view', 'default') grade_filter = request.args.get('product_grade', '') custom_field = request.args.get('custom_field', '') custom_value = request.args.get('custom_value', '') date_cond = request.args.get('date_cond', '') date_val = request.args.get('date_val', '') sort_by = request.args.get('sort_by', '') sort_order = request.args.get('sort_order', 'asc') where = [] params = [] if search: where.append("(p.product_name LIKE %s OR p.product_code LIKE %s)") params.extend([f"%{search}%", f"%{search}%"]) if prod_code: where.append("p.product_code LIKE %s") params.append(f"%{prod_code}%") if prod_name: where.append("p.product_name LIKE %s") params.append(f"%{prod_name}%") if brand_filter: where.append("p.brand=%s") params.append(brand_filter) # 解析多选筛选条件，将数据库字段的筛选转成 SQL WHERE 条件 if custom_field == '__multi__' and custom_value and isinstance(custom_value, str): json_str = custom_value if json_str.startswith('__multi__:'): json_str = json_str[len('__multi__:'):] try: import json multi_filters = json.loads(json_str) for mf in multi_filters: ff = mf.get('field', '') ft = mf.get('type', '') values = mf.get('values', []) if not ff: continue if ft != 'num' and not values: continue # 人员分组从人员信息管理表实时读取，避免前端/后端维护多套名单 GROUP_TO_MEMBERS = _personnel_group_members() PERSONNEL_GROUPS = list(GROUP_TO_MEMBERS.items()) db_col_map = { 'product_type': 'p.product_type', 'product_code': 'p.product_code', 'product_name': 'p.product_name', 'brand': 'p.brand', 'procurement_follower': 'p.procurement_follower', '采购跟单': 'p.procurement_follower', 'procurement_planner': 'p.procurement_planner', '采购计划': 'p.procurement_planner', '成本价': 'p.cost_price', '生产货期': 'p.lead_time', '物流时间': 'p.logistics_time', '最低起订量': 'p.min_order_qty', '新旧品': 'p.planning_new_old', } db_col = db_col_map.get(ff) if ft == 'search': val = mf.get('value', '') if val and db_col: like_conds = [] for v in [val]: like_conds.append(f"{db_col} LIKE %s") params.append(f"%{v}%") if like_conds: where.append("(" + " OR ".join(like_conds) + ")") elif ft == 'num': # 数字列筛选：生成 SQL 条件（全量过滤） _cond_op = mf.get('cond', '') _cond_val = mf.get('val', '') _op_map = {'eq':'=','neq':'!=','gt':'>','gte':'>=','lt':'<','lte':'<='} if _cond_op == 'range' and _cond_val: _parts = [x.strip() for x in _cond_val.split(',') if x.strip()] if len(_parts) == 2 and db_col: where.append(f"{db_col} >= %s AND {db_col} <= %s") params.extend([float(_parts[0]), float(_parts[1])]) elif _cond_op in _op_map and db_col: try: _nv = float(_cond_val) where.append(f"{db_col} {_op_map[_cond_op]} %s") params.append(_nv) except ValueError: pass elif db_col and ft in ('list', 'text'): if ff == 'product_name': like_conds = [] for v in values: like_conds.append(f"{db_col} LIKE %s") params.append(f"%{v}%") if like_conds: where.append("(" + " OR ".join(like_conds) + ")") elif ff in ('procurement_follower', '采购跟单'): # 人员字段：将组名展开为具体人名做 LIKE 匹配 expanded = [] for v in values: # Products may store either the current group label or legacy member names. # Match both so the SQL pre-filter does not discard valid group rows. expanded.append(v) if v in GROUP_TO_MEMBERS: expanded.extend(GROUP_TO_MEMBERS[v]) like_conds = [] for name in expanded: like_conds.append(f"{db_col} LIKE %s") params.append('%' + name + '%') if like_conds: where.append("(" + " OR ".join(like_conds) + ")") else: placeholders = ', '.join(['%s'] * len(values)) where.append(f"{db_col} IN ({placeholders})") params.extend(values) except (json.JSONDecodeError, TypeError): pass # 与成品基础信息表保持一致：排除包材和其他类 where.append("p.product_type NOT IN ('包材', '其他')") where.append("p.product_code NOT IN (SELECT product_code FROM plan_hidden)") where.append("p.deleted_at IS NULL") # 国内品/海外品视图过滤（按品牌区分） OVERSEAS_BRANDS = ['Qsimelr', 'PUNOMO', 'Aromat Herbs', 'MUTO', 'sedate', 'SEDATE-WOPUKA', 'WOPUKA', 'UOUONE', 'AVAA', 'MOONOTE', 'CIKITI'] if view_filter == 'cloud': # 国内品：品牌不在海外品列表中的 placeholders = ','.join(['%s'] * len(OVERSEAS_BRANDS)) where.append(f"(p.brand NOT IN ({placeholders}) OR p.brand IS NULL OR p.brand = '')") params.extend(OVERSEAS_BRANDS) elif view_filter == 'external': # 海外品：品牌在海外品列表中的 placeholders = ','.join(['%s'] * len(OVERSEAS_BRANDS)) where.append(f"p.brand IN ({placeholders})") params.extend(OVERSEAS_BRANDS) where_clause = " AND ".join(where) if where else "1=1" total = query(f"SELECT COUNT(*) AS cnt FROM products p WHERE {where_clause}", params)[0]['cnt'] # Debug: check params count if total == 2299 and view_filter == 'cloud': # The brand filter might not be applied - check view_filter value pass need_full_fetch = (type_filter or category_filter or trend_filter or supplier_filter or operator_filter or platform_filter or planner_filter or follower_filter or grade_filter or custom_field or date_cond or bool(sort_by) or view_filter != 'default') if need_full_fetch: # 全量捞取 + Python端筛选排序 all_rows = query(f""" SELECT p.product_code, p.product_name, p.brand, p.product_type, p.specification, p.cost_price, p.planning_new_old, p.lead_time AS 生产货期, p.logistics_time AS 物流时间, p.min_order_qty AS 最低起订量, p.created_at AS 创建时间, COALESCE(s.supplier_name, p.supplier_name_match) AS 供应商名称, p.procurement_follower AS 采购跟单, p.procurement_planner AS 采购计划, p.operator AS 运营专员, p.developer AS 开发专员, p.platform AS 归属平台, p.category1 AS 一级类目, p.category2 AS 二级类目, p.category3 AS 三级类目, p.custom_field1, p.custom_field2, p.custom_field3, p.custom_field4, p.custom_field5, p.remark AS 批注 FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE {where_clause} ORDER BY p.product_code ASC """, params) # 全量计算辅助数据 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()} filtered = [] for row in all_rows: if '创建时间' in row and row['创建时间']: row['创建时间'] = row['创建时间'].strftime('%Y-%m-%d') if hasattr(row['创建时间'], 'strftime') else str(row['创建时间'])[:10] if row.get('生产货期') is None: row['生产货期'] = 0 if row.get('物流时间') is None: row['物流时间'] = 0 row['产品种类'] = get_product_type_cached(row.get('product_name', ''), row.get('brand', ''), row.get('product_type')) 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) # 筛选 match = True if match and type_filter and type_filter.lower() not in (row.get('product_type') or '').lower(): match = False if match and category_filter and category_filter.lower() not in (row.get('产品种类') or '').lower(): match = False if match and trend_filter and trend_filter.lower() not in (row.get('销售趋势') or '').lower(): match = False if match and supplier_filter and supplier_filter.lower() not in (row.get('供应商名称') or '').lower(): match = False if match and operator_filter and operator_filter.lower() not in (row.get('运营专员') or '').lower(): match = False if match and platform_filter and platform_filter.lower() not in (row.get('归属平台') or '').lower(): match = False if match and planner_filter and planner_filter.lower() not in (row.get('采购计划') or '').lower(): match = False if match and follower_filter and not _personnel_value_matches_group( row.get('采购跟单'), follower_filter, GROUP_TO_MEMBERS): match = False if match and grade_filter and grade_filter.lower() not in (row.get('产品分级') or '').lower(): match = False if match and custom_field and custom_value: multi_filters = [] if custom_field == '__multi__' and isinstance(custom_value, str): # 支持 __multi__:JSON 格式 json_str = custom_value if json_str.startswith('__multi__:'): json_str = json_str[len('__multi__:'):] try: import json multi_filters = json.loads(json_str) except (json.JSONDecodeError, TypeError): pass else: multi_filters = [{'field': custom_field, 'type': 'list' if ',' in custom_value else 'text', 'values': [custom_value]}] if isinstance(custom_value, str) and custom_value.startswith('__num_filter__'): expr = custom_value[len('__num_filter__'):] parts = expr.split(':', 1) multi_filters = [{'field': custom_field, 'type': 'num', 'cond': parts[0], 'val': parts[1] if len(parts) > 1 else ''}] for f in multi_filters: ft = f.get('type', '') ff = f.get('field', custom_field) if ft == 'num': cond = f.get('cond', '') val_str = f.get('val', '') row_val = row.get(ff) try: row_num = float(row_val) if row_val not in (None, '', 0) else 0 if cond == 'eq': match = row_num == float(val_str) elif cond == 'neq': match = row_num != float(val_str) elif cond == 'gt': match = row_num > float(val_str) elif cond == 'gte': match = row_num >= float(val_str) elif cond == 'lt': match = row_num < float(val_str) elif cond == 'lte': match = row_num <= float(val_str) elif cond == 'range': vals = [float(v.strip()) for v in val_str.split(',') if v.strip()] if len(vals) >= 2: match = vals[0] <= row_num <= vals[1] except (ValueError, TypeError): match = False elif ft == 'search': val = f.get('value', '') if val: row_val = str(row.get(ff) or '').lower() if val.lower() not in row_val: match = False elif ft == 'list': values_list = [v.lower() for v in f.get('values', []) if v] if values_list: raw_row_val = row.get(ff) row_val = '' if raw_row_val is None else str(raw_row_val).lower() if '__empty__' in values_list and row_val == '': continue # 人员字段：将组名展开为具体人名做匹配 if ff in ('procurement_follower', '采购跟单'): selected_groups = [ (group_name.casefold(), members) for group_name, members in PERSONNEL_GROUPS if group_name.casefold() in values_list ] if selected_groups: row_group = _personnel_group_for_value(raw_row_val, GROUP_TO_MEMBERS) if not any(row_group == group_name for group_name, _ in selected_groups): match = False else: names = [name.strip().casefold() for name in str(raw_row_val or '').split(',') if name.strip()] if not any(v in names for v in values_list): match = False elif ff in ( '昨日销量','云仓昨日销量','外仓昨日销量','库存','云仓库存','外仓库存','加工厂库存', '总库存','总在途库存','生产在途','采购在途','云仓可分配库存','外仓可分配库存', '云仓支撑天数','外仓支撑天数', '日销量预测','云仓日销量预测','外仓日销量预测', '建议下单数量','实际下单数量','正式下单数量','建议回货数量','库存健康指数','成本价', '生产货期','物流时间','最低起订量','3天平均销量','云仓3天平均销量','外仓3天平均销量', '7天平均销量','云仓7天平均销量','外仓7天平均销量','14天平均销量','云仓14天平均销量', '外仓14天平均销量','30天平均销量','60天平均销量','3天最高销量','7天最高销量', '14天最高销量','30天最高销量','14天最低销量','14天标准差','库存可支撑天数', '库存工厂可支撑天数','加工厂库存支撑天数','库存工厂待生产可支撑天数','总支撑天数', '最低销量偏差率','最高销量偏差率','14天波动率' ): # 数字列内容筛选必须按数值精确匹配，不能用字符串包含（0 不应命中 30） try: row_num = float(raw_row_val) if raw_row_val not in (None, '') else 0.0 selected_nums = [float(v) for v in values_list] if row_num not in selected_nums: match = False except (ValueError, TypeError): match = False else: if row_val == '': # 空值不得命中任何具体选项（勾选「（空）」的情况已在上面 continue 处理） match = False elif not any(v == row_val or v in row_val for v in values_list): # 只允许「完全相等」或「单元格含该选项」，不再反向用单元格去包含选项 match = False if not match: break if match and date_cond and date_val: create_date = row.get('创建时间', '') if not create_date: match = False else: if isinstance(create_date, str): create_date = create_date[:10] else: create_date = str(create_date)[:10] if date_cond == 'lt': if create_date >= date_val: match = False elif date_cond == 'gt': if create_date <= date_val: match = False elif date_cond == 'eq': if create_date != date_val: match = False if match: filtered.append(row) if sort_by: rev = sort_order.lower() != 'asc' grade_order = {'S级': 0, 'A级': 1, 'B级': 2, 'C级': 3, '不参与评级': 4} def sort_key(row): val = row.get(sort_by) if sort_by == '产品分级': return grade_order.get(val, 99) if val is None or val == '': return float('-inf') if rev else float('inf') try: return float(val) except (ValueError, TypeError): return str(val).lower() filtered.sort(key=sort_key, reverse=rev) filtered_total = len(filtered) page_rows = filtered[offset:offset + size] print(f'[PLAN_DEBUG] view={view_filter} filtered_len={filtered_total} page_rows={len(page_rows)} all_rows={len(all_rows)}', flush=True) all_codes = [r['product_code'] for r in filtered] _merge_pending_formal_qty(page_rows) _filter_field_rows('planning', page_rows, session.get('user_id'), session.get('role'), page_view=bool(role_required('planning')), page_edit=operation_required('planning', 'edit')) return jsonify({"code": 0, "data": {"total": filtered_total, "page": page, "size": size, "rows": page_rows, "all_codes": all_codes}}) else: # SQL分页 + 少量Python计算 rows = query(f""" SELECT p.product_code, p.product_name, p.brand, p.product_type, p.specification, p.cost_price, p.planning_new_old, p.lead_time AS 生产货期, p.logistics_time AS 物流时间, p.min_order_qty AS 最低起订量, p.created_at AS 创建时间, COALESCE(s.supplier_name, p.supplier_name_match) AS 供应商名称, p.procurement_follower AS 采购跟单, p.procurement_planner AS 采购计划, p.operator AS 运营专员, p.developer AS 开发专员, p.platform AS 归属平台, p.category1 AS 一级类目, p.category2 AS 二级类目, p.category3 AS 三级类目, p.custom_field1, p.custom_field2, p.custom_field3, p.custom_field4, p.custom_field5, p.remark AS 批注 FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE {where_clause} ORDER BY p.id DESC LIMIT {size} OFFSET {offset} """, params) for row in rows: if '创建时间' in row and row['创建时间']: row['创建时间'] = row['创建时间'].strftime('%Y-%m-%d') if hasattr(row['创建时间'], 'strftime') else str(row['创建时间'])[:10] if row.get('生产货期') is None: row['生产货期'] = 0 if row.get('物流时间') is None: row['物流时间'] = 0 row['产品种类'] = get_product_type_cached(row.get('product_name', ''), row.get('brand', ''), row.get('product_type')) # 批量计算辅助数据 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()} for row in rows: 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) # 获取所有匹配的产品编号（用于全选） all_codes_rows = query(f"SELECT product_code FROM products p WHERE {where_clause}", params) all_codes = [r['product_code'] for r in all_codes_rows] _merge_pending_formal_qty(rows) _filter_field_rows('planning', rows, session.get('user_id'), session.get('role'), page_view=bool(role_required('planning')), page_edit=operation_required('planning', 'edit')) return jsonify({"code": 0, "data": {"total": total, "page": page, "size": size, "rows": rows, "all_codes": all_codes}}) @planning_bp.route('/api/planning/export', methods=['POST']) @login_required def planning_export(): """导出计划预测全部数据（Excel） 请求体: {view, sort_by, sort_order, columns: [{field,label}]} """ import io import openpyxl from openpyxl.styles import Font, Alignment, Border, Side, PatternFill from openpyxl.utils import get_column_letter body = request.get_json(silent=True) or {} gc = _ensure_global_config() view_filter = body.get('view', 'default') sort_by = body.get('sort_by', '') sort_order = body.get('sort_order', 'asc') columns = body.get('columns') or [] # 防止旧客户端/脚本未传字段方案时回退为全字段导出；空列请求直接拒绝。 if not columns: return jsonify({"code": 400, "msg": "请先在字段设置中选择要导出的字段"}), 400 # 只允许计划预测字段，排除操作列及未知字段。 allowed_fields = { 'product_type','产品种类','product_code','product_name','brand','批注','产品分级', '昨日销量','云仓昨日销量','外仓昨日销量','销售趋势','总库存','库存','总在途库存', '云仓库存','外仓库存','加工厂库存','云仓可分配库存','外仓可分配库存','是否调拨', '生产在途','采购在途','日销量预测','云仓日销量预测','外仓日销量预测','建议下单数量', '实际下单数量','正式下单数量','库存健康指数','风险把控','供应商名称','采购跟单', '采购计划','开发专员','归属平台','运营专员','cost_price','3天平均销量','云仓3天平均销量', '外仓3天平均销量','7天平均销量','云仓7天平均销量','外仓7天平均销量','14天平均销量', '云仓14天平均销量','外仓14天平均销量','库存可支撑天数','库存工厂可支撑天数', '加工厂库存支撑天数','库存工厂待生产可支撑天数','总支撑天数','云仓支撑天数','外仓支撑天数', '建议回货数量','specification', '30天平均销量','60天平均销量','3天最高销量','7天最高销量','14天最高销量','30天最高销量', '14天最低销量','14天标准差','最低销量偏差率','最高销量偏差率','极值判断','14天波动率', '波动风险','生产货期','物流时间','最低起订量','创建时间','是否新品','新旧品','一级类目','二级类目','三级类目' } columns = [c for c in columns if isinstance(c, dict) and c.get('field') in allowed_fields] if not columns: return jsonify({"code": 400, "msg": "没有可导出的有效字段"}), 400 # ── 基础 where（与列表一致）── where = [] params = [] where.append("p.product_type NOT IN ('包材', '其他')") where.append("p.product_code NOT IN (SELECT product_code FROM plan_hidden)") where.append("p.deleted_at IS NULL") OVERSEAS_BRANDS = ['Qsimelr', 'PUNOMO', 'Aromat Herbs', 'MUTO', 'sedate', 'SEDATE-WOPUKA', 'WOPUKA', 'UOUONE', 'AVAA', 'MOONOTE', 'CIKITI'] if view_filter == 'cloud': placeholders = ','.join(['%s'] * len(OVERSEAS_BRANDS)) where.append(f"(p.brand NOT IN ({placeholders}) OR p.brand IS NULL OR p.brand = '')") params.extend(OVERSEAS_BRANDS) elif view_filter == 'external': placeholders = ','.join(['%s'] * len(OVERSEAS_BRANDS)) where.append(f"p.brand IN ({placeholders})") params.extend(OVERSEAS_BRANDS) where_clause = " AND ".join(where) if where else "1=1" # ── 全量拉取（与列表全量分支一致）── all_rows = query(f""" SELECT p.product_code, p.product_name, p.brand, p.product_type, p.specification, p.cost_price, p.planning_new_old, p.lead_time AS 生产货期, p.logistics_time AS 物流时间, p.min_order_qty AS 最低起订量, p.created_at AS 创建时间, COALESCE(s.supplier_name, p.supplier_name_match) AS 供应商名称, p.procurement_follower AS 采购跟单, p.procurement_planner AS 采购计划, p.operator AS 运营专员, p.developer AS 开发专员, p.platform AS 归属平台, p.category1 AS 一级类目, p.category2 AS 二级类目, p.category3 AS 三级类目, p.custom_field1, p.custom_field2, p.custom_field3, p.custom_field4, p.custom_field5, p.remark AS 批注 FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE {where_clause} ORDER BY p.product_code ASC """, params) 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()} rows = [] for row in all_rows: if '创建时间' in row and row['创建时间']: row['创建时间'] = row['创建时间'].strftime('%Y-%m-%d') if hasattr(row['创建时间'], 'strftime') else str(row['创建时间'])[:10] if row.get('生产货期') is None: row['生产货期'] = 0 if row.get('物流时间') is None: row['物流时间'] = 0 row['产品种类'] = get_product_type_cached(row.get('product_name', ''), row.get('brand', ''), row.get('product_type')) 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(row) if sort_by: rev = sort_order.lower() != 'asc' grade_order = {'S级': 0, 'A级': 1, 'B级': 2, 'C级': 3, '不参与评级': 4} def sort_key(row): val = row.get(sort_by) if sort_by == '产品分级': return grade_order.get(val, 99) if val is None or val == '': return float('-inf') if rev else float('inf') try: return float(val) except (ValueError, TypeError): return str(val).lower() rows.sort(key=sort_key, reverse=rev) _merge_pending_formal_qty(rows) # ── 生成 Excel ── wb = openpyxl.Workbook() ws = wb.active ws.title = '计划预测' header_font = Font(bold=True, color='FFFFFF') header_fill = PatternFill('solid', fgColor='1890FF') thin = Side(style='thin', color='D9D9D9') border = Border(left=thin, right=thin, top=thin, bottom=thin) ws.append([c.get('label', c.get('field', '')) for c in columns]) for cell in ws[1]: cell.font = header_font cell.fill = header_fill cell.alignment = Alignment(horizontal='center', vertical='center') cell.border = border for row in rows: vals = [] for c in columns: v = row.get(c.get('field', '')) if v is None: v = '' vals.append(v) ws.append(vals) for i, c in enumerate(columns, 1): max_len = len(str(c.get('label', ''))) for row in ws.iter_rows(min_row=2, min_col=i, max_col=i): for cell in row: if cell.value is not None: max_len = max(max_len, len(str(cell.value))) ws.column_dimensions[get_column_letter(i)].width = min(max(max_len * 1.6, 10), 40) for row in ws.iter_rows(min_row=2): for cell in row: cell.border = border output = io.BytesIO() wb.save(output) output.seek(0) from flask import send_file filename = f'计划预测_{datetime.now().strftime("%Y%m%d_%H%M%S")}.xlsx' return send_file( output, as_attachment=True, download_name=filename, mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' ) # ─── 页面路由 ─── @planning_bp.route('/planning') @login_required def planning_page(): if not role_required('planning'): return redirect(url_for('auth.index')) return render_template('base.html', page='planning') @planning_bp.route('/api/planning/hide', methods=['POST']) @login_required def planning_hide_product(): """将产品从计划预测中隐藏""" data = request.json product_code = data.get('product_code', '') if not product_code: return jsonify({"code": 400, "msg": "缺少货品编号"}) query("INSERT IGNORE INTO plan_hidden (product_code) VALUES (%s)", (product_code,)) return jsonify({"code": 0, "msg": "已从计划预测中移除"}) @planning_bp.route('/api/planning/unhide', methods=['POST']) @login_required def planning_unhide_product(): """将产品恢复显示到计划预测""" data = request.json product_code = data.get('product_code', '') if not product_code: return jsonify({"code": 400, "msg": "缺少货品编号"}) query("DELETE FROM plan_hidden WHERE product_code=%s", (product_code,)) return jsonify({"code": 0, "msg": "已恢复显示"}) @planning_bp.route('/api/product-types') def product_type_list(): """返回所有产品类型列表""" from services.db import query rows = query("SELECT DISTINCT product_type FROM products WHERE deleted_at IS NULL AND product_type IS NOT NULL AND product_type != '' ORDER BY product_type") types = [r['product_type'] for r in rows] return jsonify({"code": 0, "data": types}) @planning_bp.route('/api/planning/brands') @login_required def planning_brands(): """获取计划预测页面的品牌列表（按当前视图过滤，支持联动筛选）""" view_filter = request.args.get('view', 'default') plan_filter_str = request.args.get('plan_filter', '') or '' where = ["product_type NOT IN ('包材', '其他')"] where.append("product_code NOT IN (SELECT product_code FROM plan_hidden)") where.append("p.deleted_at IS NULL") OVERSEAS_BRANDS = ['Qsimelr', 'PUNOMO', 'Aromat Herbs', 'MUTO', 'sedate', 'SEDATE-WOPUKA', 'WOPUKA', 'UOUONE', 'AVAA', 'MOONOTE', 'CIKITI'] if view_filter == 'cloud': placeholders = ','.join(['%s'] * len(OVERSEAS_BRANDS)) where.append(f"(brand NOT IN ({placeholders}) OR brand IS NULL OR brand = '')") params = list(OVERSEAS_BRANDS) elif view_filter == 'external': placeholders = ','.join(['%s'] * len(OVERSEAS_BRANDS)) where.append(f"brand IN ({placeholders})") params = list(OVERSEAS_BRANDS) else: params = [] # 联动筛选 if plan_filter_str: try: import json _pfs = json.loads(plan_filter_str) _pms = { 'product_code': 'p.product_code', 'product_name': 'p.product_name', 'product_type': 'p.product_type', 'specification': 'p.specification', '供应商名称': "COALESCE(s.supplier_name, p.supplier_name_match)", '采购跟单': 'p.procurement_follower', '采购计划': 'p.procurement_planner', '运营专员': 'p.operator', '开发专员': 'p.developer', '归属平台': 'p.platform', '一级类目': 'p.category1', '二级类目': 'p.category2', '三级类目': 'p.category3', '成本价': 'p.cost_price', '生产货期': 'p.lead_time', '物流时间': 'p.logistics_time', '最低起订量': 'p.min_order_qty', '新旧品': 'p.planning_new_old', } for _k, _v in _pfs.items(): if _k == 'brand': continue if not isinstance(_v, dict): continue _dbc = _pms.get(_k) if not _dbc: continue _t = _v.get('type', '') if _t == 'search': _val = _v.get('value', '') if _val: where.append(_dbc + ' LIKE %s') params.append('%' + _val + '%') elif _t == 'list': _vals = _v.get('values', []) if _vals: _non_empty = [x for x in _vals if x != '__EMPTY__'] if _non_empty: _phs = ', '.join(['%s'] * len(_non_empty)) where.append(_dbc + ' IN (' + _phs + ')') params.extend(_non_empty) if '__EMPTY__' in _vals: where.append('(' + _dbc + ' IS NULL OR ' + _dbc + " = '')") except: pass where_clause = " AND ".join(where) from services.db import query if params: rows = query(f"SELECT DISTINCT brand FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE {where_clause} AND brand IS NOT NULL AND brand != '' ORDER BY brand", params) else: rows = query(f"SELECT DISTINCT brand FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE {where_clause} AND brand IS NOT NULL AND brand != '' ORDER BY brand") if params: rows = query(f"SELECT DISTINCT brand FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE {where_clause} AND brand IS NOT NULL AND brand != '' ORDER BY brand", params) else: rows = query(f"SELECT DISTINCT brand FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE {where_clause} AND brand IS NOT NULL AND brand != '' ORDER BY brand") vals = [r['brand'] for r in rows] emptySql = "SELECT COUNT(*) AS cnt FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE " + where_clause + " AND (brand IS NULL OR brand = '')" emptyRows = query(emptySql, params) if params else query(emptySql) if emptyRows and emptyRows[0]['cnt'] > 0: vals.append('__EMPTY__') return jsonify({"code": 0, "data": vals}) @planning_bp.route('/api/planning/column-values') @login_required def planning_column_values(): """获取计划预测页面指定列的所有唯一值（全量，支持联动筛选）""" field = request.args.get('field', '') view_filter = request.args.get('view', 'default') plan_filter_str = request.args.get('plan_filter', '') or '' if not field: return jsonify({"code": 400, "msg": ":缺少field参数"}) OVERSEAS_BRANDS = ['Qsimelr', 'PUNOMO', 'Aromat Herbs', 'MUTO', 'sedate', 'SEDATE-WOPUKA', 'WOPUKA', 'UOUONE', 'AVAA', 'MOONOTE', 'CIKITI'] where = ["p.product_type NOT IN ('包材', '其他')"] where.append("p.product_code NOT IN (SELECT product_code FROM plan_hidden)") where.append("p.deleted_at IS NULL") if view_filter == 'cloud': placeholders = ','.join(['%s'] * len(OVERSEAS_BRANDS)) where.append(f"(p.brand NOT IN ({placeholders}) OR p.brand IS NULL OR p.brand = '')") params = list(OVERSEAS_BRANDS) elif view_filter == 'external': placeholders = ','.join(['%s'] * len(OVERSEAS_BRANDS)) where.append(f"p.brand IN ({placeholders})") params = list(OVERSEAS_BRANDS) else: params = [] # 联动筛选：解析 plan_filter 并追加条件 if plan_filter_str: try: import json _pfs = json.loads(plan_filter_str) _pms = db_col_map = { 'product_code': 'p.product_code', 'product_name': 'p.product_name', 'product_type': 'p.product_type', 'specification': 'p.specification', '供应商名称': "COALESCE(s.supplier_name, p.supplier_name_match)", 'procurement_follower': 'p.procurement_follower', '采购跟单': 'p.procurement_follower', 'procurement_planner': 'p.procurement_planner', '采购计划': 'p.procurement_planner', '运营专员': 'p.operator', '开发专员': 'p.developer', '归属平台': 'p.platform', '一级类目': 'p.category1', '二级类目': 'p.category2', '三级类目': 'p.category3', '成本价': 'p.cost_price', '生产货期': 'p.lead_time', '物流时间': 'p.logistics_time', '最低起订量': 'p.min_order_qty', '新旧品': 'p.planning_new_old', } for _k, _v in _pfs.items(): if _k == field: continue if not isinstance(_v, dict): continue _dbc = _pms.get(_k) if not _dbc: continue _t = _v.get('type', '') if _t == 'search': _val = _v.get('value', '') if _val: where.append(_dbc + ' LIKE %s') params.append('%' + _val + '%') elif _t == 'list': _vals = _v.get('values', []) if _vals: _non_empty = [x for x in _vals if x != '__EMPTY__'] _parts = [] if _non_empty: if _k in ('procurement_follower', '采购跟单'): _groups = _personnel_group_members() _names = [] for _selected in _non_empty: _names.extend(_groups.get(_selected, [_selected])) if _selected in _groups: _names.append(_selected) for _name in _names: _parts.append(_dbc + ' LIKE %s') params.append('%' + _name + '%') else: _phs = ', '.join(['%s'] * len(_non_empty)) _parts.append(_dbc + ' IN (' + _phs + ')') params.extend(_non_empty) if '__EMPTY__' in _vals: _parts.append('(' + _dbc + ' IS NULL OR ' + _dbc + " = '')") if _parts: where.append('(' + ' OR '.join(_parts) + ')') except: pass where_clause = " AND ".join(where) db_col_map = { 'product_code': 'p.product_code', 'product_name': 'p.product_name', 'product_type': 'p.product_type', 'specification': 'p.specification', '供应商名称': "COALESCE(s.supplier_name, p.supplier_name_match)", '采购跟单': 'p.procurement_follower', '采购计划': 'p.procurement_planner', '运营专员': 'p.operator', '开发专员': 'p.developer', '归属平台': 'p.platform', '一级类目': 'p.category1', '二级类目': 'p.category2', '三级类目': 'p.category3', '成本价': 'p.cost_price', '生产货期': 'p.lead_time', '物流时间': 'p.logistics_time', '最低起订量': 'p.min_order_qty', '新旧品': 'p.planning_new_old', } db_col = db_col_map.get(field) # 如果是计算字段（不在数据库表中），走完整查询+计算后提取 COMPUTED_NUMERIC_FIELDS = { '昨日销量','云仓昨日销量','外仓昨日销量', '库存','云仓库存','外仓库存','加工厂库存','总库存','总在途库存', '云仓可分配库存','外仓可分配库存', '云仓支撑天数','外仓支撑天数', '生产在途','采购在途', '日销量预测','云仓日销量预测','外仓日销量预测', '建议下单数量','实际下单数量','正式下单数量','建议回货数量', '库存健康指数', '3天平均销量','云仓3天平均销量','外仓3天平均销量', '7天平均销量','云仓7天平均销量','外仓7天平均销量', '14天平均销量','云仓14天平均销量','外仓14天平均销量', '30天平均销量','60天平均销量', '3天最高销量','7天最高销量','14天最高销量','30天最高销量', '14天最低销量','14天标准差', '库存可支撑天数','库存工厂可支撑天数','加工厂库存支撑天数', '库存工厂待生产可支撑天数','总支撑天数', '最低销量偏差率','最高销量偏差率','14天波动率', } if field in COMPUTED_NUMERIC_FIELDS: # 计算字段：走全量查询+enrichment 后提取唯一值 need_full_fetch = True if not db_col and field not in COMPUTED_NUMERIC_FIELDS: return jsonify({"code": 0, "data": []}) from services.db import query if field in COMPUTED_NUMERIC_FIELDS: # 计算字段：获取全量产品，走 enrichment 提取唯一值 try: all_rows = query(f""" SELECT p.product_code, p.product_name, p.brand, p.product_type, p.specification, p.cost_price, p.planning_new_old, p.lead_time AS 生产货期, p.logistics_time AS 物流时间, p.min_order_qty AS 最低起订量, p.created_at AS 创建时间, COALESCE(s.supplier_name, p.supplier_name_match) AS 供应商名称, p.procurement_follower AS 采购跟单, p.procurement_planner AS 采购计划, p.operator AS 运营专员, p.developer AS 开发专员, p.platform AS 归属平台, p.category1 AS 一级类目, p.category2 AS 二级类目, p.category3 AS 三级类目 FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE {where_clause} """, params) except Exception as _e: import logging logging.error(f"column-values computed field query failed: {_e}") return jsonify({"code": 0, "data": []}) # 使用已导入的 cached 函数 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() seen = {} values = [] for row in all_rows: if row.get('创建时间'): created = row['创建时间'] row['创建时间'] = created.strftime('%Y-%m-%d') if hasattr(created, 'strftime') else str(created)[:10] row['生产货期'] = row.get('生产货期') or 0 row['物流时间'] = row.get('物流时间') or 0 row['产品种类'] = get_product_type_cached( row.get('product_name', ''), row.get('brand', ''), row.get('product_type') ) enriched = _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 ) # 联动筛选：有 plan_filter 时过滤其他字段的条件 if plan_filter_str: try: _pf = json.loads(plan_filter_str) _skip = False for _k, _v in _pf.items(): if _k == field: continue if not isinstance(_v, dict): continue # 数据库字段已经在上方 SQL WHERE 中完成联动过滤， # 此处只重复校验计算字段，避免中文显示字段取不到值而被误判为空。 if _k in _pms: continue _vt = _v.get('type', '') _raw = enriched.get(_k) _rv = '' if _raw is None else str(_raw) if _vt == 'list': _vls = [str(x) for x in _v.get('values', [])] if '__EMPTY__' in _vls and _rv == '': pass elif _vls and _rv not in _vls: _skip = True; break elif _vt == 'search': _sv = _v.get('value', '') if _sv and _sv not in _rv: _skip = True; break elif _vt == 'num': try: _rn = float(_rv) if _rv not in ('', 'None') else 0 _cn = float(_v.get('val', 0)) _co = _v.get('cond', '') if _co == 'gt' and _rn <= _cn: _skip = True; break if _co == 'gte' and _rn < _cn: _skip = True; break if _co == 'lt' and _rn >= _cn: _skip = True; break if _co == 'lte' and _rn > _cn: _skip = True; break if _co == 'eq' and _rn != _cn: _skip = True; break except: pass if _skip: continue except: pass val = enriched.get(field) if val is not None and val != '' and str(val) != '-': txt = str(val).strip() if txt and txt not in seen: seen[txt] = 1 values.append(txt) # 数值排序 try: values.sort(key=lambda x: float(x)) except: values.sort() return jsonify({"code": 0, "data": values}) sql = f"SELECT DISTINCT {db_col} AS val FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE {where_clause} AND {db_col} IS NOT NULL AND {db_col} != '' ORDER BY val" rows = query(sql, params) if params else query(sql) values = [r['val'] for r in rows] # 空值检测 emptySql = f"SELECT COUNT(*) AS cnt FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE {where_clause} AND ({db_col} IS NULL OR {db_col} = '')" emptyRows = query(emptySql, params) if params else query(emptySql) if emptyRows and emptyRows[0]['cnt'] > 0: values.append('__EMPTY__') return jsonify({"code": 0, "data": values})