"""重点SKU预测路由：SKU基础信息列表 + 历史销量参考列表""" from datetime import datetime, timedelta import json import time from flask import Blueprint, request, jsonify, session, send_file from services.db import query, add_operation_log, execute from services.forecast import get_product_grade, is_new_product from services.inventory_calc import resolve_cost_price, EXCLUDE_WAREHOUSE_20_PRODUCTS from routes.auth import login_required, role_required, operation_required sku_forecast_bp = Blueprint('sku_forecast', __name__) SALES_MANAGEMENT_COLUMNS = [ '批次', '货品编码', '货品名称', '确权人', '确权平台', '库存区间', '建档时间', '盘盈盘亏', '差异合计', '生产日期', '过期时间', '成本价', '总库存', '可发库存', '可支撑天数', '30天销量', '仓库存合计', '29青鸟海外仓', '27启程仓', '26海外抖音仓', '25海外拼多多保税仓', '1蝶雅仓', '2青鸟仓', '4百橙仓', '5东升仓', '6CK仓', '9天猫菜鸟仓', '10唯品会海外仓', '京东海外11仓', '京东海外12仓', '京东13仓', '京东14仓', '16蝶雅南阳仓', '分类', '戴总确认意见', '运营核实更改', '运营核对备注' ] SALES_MANAGEMENT_DB_COLUMNS = [f'c{i:02d}' for i in range(1, len(SALES_MANAGEMENT_COLUMNS) + 1)] SALES_MANAGEMENT_PAGE = 'collaboration_sales_management' def _latest_sales_date(): """获取出库数据最新日期""" rows = query("SELECT MAX(日期) AS d FROM daily_outbound") if not rows or rows[0]['d'] is None: return datetime.now().date() d = rows[0]['d'] if hasattr(d, 'strftime'): return d.date() if hasattr(d, 'date') else d return datetime.strptime(str(d)[:10], '%Y-%m-%d').date() def _month_range(ref_date, months_offset=0): """计算某个月份的起止日期；months_offset=0 表示 ref_date 所在月，-1 表示上个月""" first = ref_date.replace(day=1) target_first = first for _ in range(-months_offset): target_first = (target_first.replace(day=1) - timedelta(days=1)).replace(day=1) if months_offset > 0: for _ in range(months_offset): target_first = (target_first.replace(day=28) + timedelta(days=4)).replace(day=1) next_month = (target_first.replace(day=28) + timedelta(days=4)).replace(day=1) target_last = next_month - timedelta(days=1) return target_first, target_last def _compute_sales_map(): """计算每个货品编号的：昨天销量、上月销量、本月销量、7日均销、15天均销""" latest = _latest_sales_date() yesterday = latest - timedelta(days=1) # 上月/本月范围 last_first, last_last = _month_range(latest, -1) cur_first, cur_last = _month_range(latest, 0) # 近7天/15天范围（含昨天，不含今天？出库数据最新日期即“昨天”口径，直接取最新日期往前推） start_7 = latest - timedelta(days=6) start_15 = latest - timedelta(days=14) rows = query(""" SELECT 货品编号, SUM(CASE WHEN 日期 = %s THEN 货品数量 ELSE 0 END) AS 昨天销量, SUM(CASE WHEN 日期 BETWEEN %s AND %s THEN 货品数量 ELSE 0 END) AS 上月销量, SUM(CASE WHEN 日期 BETWEEN %s AND %s THEN 货品数量 ELSE 0 END) AS 本月销量, SUM(CASE WHEN 日期 BETWEEN %s AND %s THEN 货品数量 ELSE 0 END) AS 近7天销量, SUM(CASE WHEN 日期 BETWEEN %s AND %s THEN 货品数量 ELSE 0 END) AS 近15天销量 FROM daily_outbound WHERE 日期 >= %s GROUP BY 货品编号 """, (yesterday, last_first, last_last, cur_first, cur_last, start_7, latest, start_15, latest, start_15)) result = {} for r in rows: pc = r['货品编号'] result[pc] = { '昨天销量': int(r['昨天销量'] or 0), '上月销量': int(r['上月销量'] or 0), '本月销量': int(r['本月销量'] or 0), '7日均销': float(r['近7天销量'] or 0) / 7, '15天均销': float(r['近15天销量'] or 0) / 15, } return result _HEALTH_INDEX_CACHE = {'at': 0.0, 'value': None} def _health_index_rows(): """Return a short-lived snapshot so header filtering does not recompute all metrics.""" now = time.monotonic() cached = _HEALTH_INDEX_CACHE.get('value') if cached is not None and now - _HEALTH_INDEX_CACHE.get('at', 0.0) < 30: return cached value = _compute_health_index_rows() _HEALTH_INDEX_CACHE.update(at=now, value=value) return value def _compute_health_index_rows(): """生成 SKU 库存健康指数列表，统一使用最新库存日和最新出库日。""" inv_date = query("SELECT MAX(data_date) AS d FROM inventory_detail")[0]['d'] sales_date = _latest_sales_date() if not inv_date: return [], None, sales_date start_30 = sales_date - timedelta(days=29) sales_rows = query(""" SELECT 货品编号, SUM(CASE WHEN 日期 = %s THEN 货品数量 ELSE 0 END) AS yesterday_sales, SUM(CASE WHEN 日期 BETWEEN %s AND %s THEN 货品数量 ELSE 0 END) AS sales_7, SUM(货品数量) AS sales_30 FROM daily_outbound WHERE 日期 BETWEEN %s AND %s AND 货品编号 NOT LIKE %s AND 货品名称 NOT LIKE %s GROUP BY 货品编号 """, (sales_date, sales_date - timedelta(days=6), sales_date, start_30, sales_date, '%-NL', '%内料%')) sales_map = {r['货品编号']: { '昨日销量': int(r['yesterday_sales'] or 0), '7天销量': int(r['sales_7'] or 0), '30天销量': int(r['sales_30'] or 0), } for r in sales_rows} inv_rows = query(""" SELECT product_code, MAX(product_name) AS product_name, SUM(available_qty) AS available_qty, SUM(purchase_in_transit) AS purchase_in_transit FROM inventory_detail WHERE data_date = %s AND product_code NOT LIKE %s AND product_name NOT LIKE %s GROUP BY product_code """, (inv_date, '%-NL', '%内料%')) inv_map = {r['product_code']: r for r in inv_rows if r.get('product_code')} products = query(""" SELECT p.product_code AS 货品编号, p.product_name AS 货品名称, p.brand AS 品牌, p.product_type AS SKU分类, p.lead_time, p.logistics_time, p.created_at AS 单品创建时间 FROM products p WHERE p.deleted_at IS NULL AND p.product_type NOT IN ('包材','其他','物料') AND p.product_code NOT LIKE %s AND p.product_name NOT LIKE %s ORDER BY p.product_code """, ('%-NL', '%内料%')) rows = [] days_in_month = ((sales_date.replace(day=28) + timedelta(days=4)).replace(day=1) - sales_date).days for p in products: code = p['货品编号']; inv = inv_map.get(code, {}); sales = sales_map.get(code, {}) yesterday = sales.get('昨日销量', 0); sales7 = sales.get('7天销量', 0); sales30 = sales.get('30天销量', 0) avg7 = sales7 / 7.0; avg30 = sales30 / 30.0 forecast = round(avg7 * 0.6 + avg30 * 0.4, 2) available = int(inv.get('available_qty') or 0); transit = int(inv.get('purchase_in_transit') or 0) arrival_days = int(p.get('lead_time') or 0) + int(p.get('logistics_time') or 0) stock_days = round(available / forecast, 1) if forecast > 0 else 999 health = round(stock_days / arrival_days, 2) if forecast > 0 and arrival_days > 0 else 999 month_forecast = round(forecast * days_in_month, 1) sufficient = (available + transit) >= month_forecast if forecast > 0 else available > 0 created = p.get('单品创建时间') if created and hasattr(created, 'strftime'): created = created.strftime('%Y-%m-%d') age = '' if created: try: age = max(0, (sales_date - datetime.strptime(str(created)[:10], '%Y-%m-%d').date()).days) except (TypeError, ValueError): pass if health < 1: plan = '库存不足，建议尽快采购' elif health < 1.5: plan = '关注库存，建议安排采购' elif stock_days > 90 and sales30 == 0: plan = '暂无动销，关注库存消化' else: plan = '正常' rows.append({ 'SKU分类': p.get('SKU分类') or '', '品牌': p.get('品牌') or '', '货品编号': code, '货品名称': p.get('货品名称') or inv.get('product_name') or '', '采购到货天数': arrival_days, '可用库存': available, '昨日销量': yesterday, '7天销量': sales7, '30天销量': sales30, '采购在途': transit, '预测日销量': forecast, '当月需求预测汇总': month_forecast, '昨日可销天数': round(available / yesterday, 1) if yesterday else 999, '7日可销天数': round(available / avg7, 1) if avg7 else 999, '30日可销天数': round(available / avg30, 1) if avg30 else 999, '库存可支撑天数': stock_days, '库存健康指数': health, '库存是否满足': '是' if sufficient else '否', '处理计划': plan, '新老品': is_new_product(created or ''), '单品创建时间': created or '', '库龄': age }) return rows, inv_date, sales_date @sku_forecast_bp.route('/api/sku_health_index/list') @login_required def sku_health_index_list(): page = max(1, int(request.args.get('page', 1))) size = min(200, max(1, int(request.args.get('size', 50)))) search = request.args.get('search', '').strip().lower() category = request.args.get('category', '').strip(); brand = request.args.get('brand', '').strip() status = request.args.get('status', '').strip() try: column_filters = json.loads(request.args.get('column_filters', '{}') or '{}') except (TypeError, ValueError): column_filters = {} sort_field = request.args.get('sort_field', '').strip() sort_order = request.args.get('sort_order', '').strip().lower() rows, inv_date, sales_date = _health_index_rows() if search: rows = [r for r in rows if search in str(r['货品编号']).lower() or search in str(r['货品名称']).lower()] if category: rows = [r for r in rows if r['SKU分类'] == category] if brand: rows = [r for r in rows if r['品牌'] == brand] if status: rows = [r for r in rows if r['库存是否满足'] == status] allowed_fields = ['SKU分类','品牌','货品编号','货品名称','采购到货天数','可用库存','昨日销量','7天销量','30天销量','采购在途','预测日销量','当月需求预测汇总','昨日可销天数','7日可销天数','30日可销天数','库存可支撑天数','库存健康指数','库存是否满足','处理计划','新老品','单品创建时间','库龄'] for field, values in column_filters.items(): if field not in allowed_fields or not isinstance(values, list) or not values: continue rows = [r for r in rows if str(r.get(field) if r.get(field) not in (None, '') else '__EMPTY__') in {str(v) for v in values}] if sort_field in allowed_fields and sort_order in ('asc', 'desc'): def sort_key(row): value = row.get(sort_field) if value in (None, ''): return (0, '') try: return (1, float(value)) except (TypeError, ValueError): return (1, str(value).lower()) rows.sort(key=sort_key, reverse=sort_order == 'desc') else: rows.sort(key=lambda r: (r['库存健康指数'] >= 1, r['库存健康指数'], r['货品编号'])) filter_options = {field: sorted({str(r.get(field) if r.get(field) not in (None, '') else '__EMPTY__') for r in rows}, key=str.lower) for field in allowed_fields} start = (page - 1) * size return jsonify({'code': 0, 'data': rows[start:start + size], 'total': len(rows), 'page': page, 'size': size, 'latest_inventory_date': str(inv_date or '')[:10], 'latest_sales_date': str(sales_date or '')[:10], 'categories': sorted({r['SKU分类'] for r in rows if r['SKU分类']}), 'brands': sorted({r['品牌'] for r in rows if r['品牌']}), 'filter_options': filter_options}) def _slow_moving_rows(): """按单品创建满90天且近30天可销天数不少于90天生成呆滞SKU列表。""" inv_date = query("SELECT MAX(data_date) AS d FROM inventory_detail")[0]['d'] sales_date = _latest_sales_date() if not inv_date: return [], None, sales_date start_60 = sales_date - timedelta(days=59) sales_rows = query(""" SELECT 货品编号, SUM(CASE WHEN 日期 = %s THEN 货品数量 ELSE 0 END) AS yesterday_sales, SUM(CASE WHEN 日期 BETWEEN %s AND %s THEN 货品数量 ELSE 0 END) AS sales_7, SUM(CASE WHEN 日期 BETWEEN %s AND %s THEN 货品数量 ELSE 0 END) AS sales_30 FROM daily_outbound WHERE 日期 BETWEEN %s AND %s AND 货品编号 NOT LIKE %s AND 货品名称 NOT LIKE %s GROUP BY 货品编号 """, (sales_date, sales_date - timedelta(days=6), sales_date, sales_date - timedelta(days=29), sales_date, start_60, sales_date, '%-NL', '%内料%')) total_sales_rows = query(""" SELECT 货品编号, SUM(货品数量) AS total_sales FROM daily_outbound WHERE 货品编号 NOT LIKE %s AND 货品名称 NOT LIKE %s GROUP BY 货品编号 """, ('%-NL', '%内料%')) total_sales_map = {r['货品编号']: int(r['total_sales'] or 0) for r in total_sales_rows} sales_map = {r['货品编号']: { '昨日销量': int(r['yesterday_sales'] or 0), '7天销量': int(r['sales_7'] or 0), '30天销量': int(r['sales_30'] or 0), '总销量': total_sales_map.get(r['货品编号'], 0) } for r in sales_rows} inv_rows = query(""" SELECT id.product_code, MAX(id.product_name) AS product_name, MAX(id.brand) AS inventory_brand, SUM(id.available_qty) AS inventory, SUM(id.production_in_transit) AS production_in_transit, SUM(id.purchase_in_transit) AS purchase_in_transit, ANY_VALUE(id.cost_price) AS inventory_cost, MIN(id.data_date) AS first_inventory_date FROM inventory_detail id WHERE id.data_date = %s AND id.product_code NOT LIKE %s AND id.product_name NOT LIKE %s GROUP BY id.product_code """, (inv_date, '%-NL', '%内料%')) inv_map = {r['product_code']: r for r in inv_rows if r.get('product_code')} products = query(""" SELECT p.product_code AS 货品编号, p.product_name AS 货品名称, p.brand AS 品牌, p.product_type, p.platform AS 确权平台, p.created_at AS 单品创建时间, p.cost_price AS product_cost, COALESCE(orders.total_order_qty, 0) AS total_order_qty FROM products p LEFT JOIN ( SELECT i.product_code, SUM(i.order_qty) AS total_order_qty FROM purchase_order_items i GROUP BY i.product_code ) orders ON orders.product_code = p.product_code WHERE p.deleted_at IS NULL AND p.product_type IN ('成品','清库存','现货','赠品') AND p.product_code NOT LIKE %s AND p.product_name NOT LIKE %s ORDER BY p.product_code """, ('%-NL', '%内料%')) rows = [] for p in products: code = p['货品编号']; inv = inv_map.get(code) if not inv or int(inv.get('inventory') or 0) <= 0: continue sales = sales_map.get(code, {}) sales30 = sales.get('30天销量', 0); avg30 = sales30 / 30.0 saleable_days = round(int(inv['inventory']) / avg30, 1) if avg30 else 999 created_raw = p.get('单品创建时间') try: created_date = created_raw.date() if hasattr(created_raw, 'date') else datetime.strptime(str(created_raw)[:10], '%Y-%m-%d').date() except (TypeError, ValueError): continue if (inv_date - created_date).days < 90 or saleable_days < 90: continue forecast = round(avg30, 2) stock_days = saleable_days cost_raw = inv.get('inventory_cost') or p.get('product_cost') or 0 cost = resolve_cost_price(code, float(cost_raw)) created = p.get('单品创建时间') if created and hasattr(created, 'strftime'): created = created.strftime('%Y-%m-%d') rows.append({ '品牌': p.get('品牌') or inv.get('inventory_brand') or '', '货品编号': code, '货品名称': p.get('货品名称') or inv.get('product_name') or '', '库存': int(inv['inventory']), '昨日销量': sales.get('昨日销量', 0), '7天销量': sales.get('7天销量', 0), '30天销量': sales30, '总销量': sales.get('总销量', 0), '可销天数': saleable_days, '预测日销量': forecast, '库存支撑天数': stock_days, '总订单数量': int(p.get('total_order_qty') or 0), '采购在途': int(inv.get('purchase_in_transit') or 0), '生产在途': int(inv.get('production_in_transit') or 0), '成本价': round(cost, 4), '成品库存金额': round(int(inv['inventory']) * cost, 2), '单品创建时间': created or '', '新老品': is_new_product(created or ''), '确权平台': p.get('确权平台') or '' }) return rows, inv_date, sales_date @sku_forecast_bp.route('/api/slow_moving_sku/list') @login_required def slow_moving_sku_list(): page = max(1, int(request.args.get('page', 1))) size = min(200, max(1, int(request.args.get('size', 50)))) search = request.args.get('search', '').strip().lower() brand = request.args.get('brand', '').strip() platform = request.args.get('platform', '').strip() rows, inv_date, sales_date = _slow_moving_rows() if search: rows = [r for r in rows if search in str(r['货品编号']).lower() or search in str(r['货品名称']).lower()] if brand: rows = [r for r in rows if r['品牌'] == brand] if platform: rows = [r for r in rows if r['确权平台'] == platform] rows.sort(key=lambda r: (-r['成品库存金额'], -r['库存'], r['货品编号'])) start = (page - 1) * size return jsonify({'code': 0, 'data': rows[start:start + size], 'total': len(rows), 'page': page, 'size': size, 'latest_inventory_date': str(inv_date or '')[:10], 'latest_sales_date': str(sales_date or '')[:10], 'brands': sorted({r['品牌'] for r in rows if r['品牌']}), 'platforms': sorted({r['确权平台'] for r in rows if r['确权平台']})}) @sku_forecast_bp.route('/api/sales_management/list', methods=['GET', 'POST']) @login_required def sales_management_list(): """查询动销管理导入数据；本页面只读取导入表。""" if not operation_required(SALES_MANAGEMENT_PAGE, 'view'): return jsonify({'code': 403, 'msg': '无访问权限'}), 403 try: page = max(1, int(request.args.get('page', 1))) except (TypeError, ValueError): page = 1 try: size = min(100, max(1, int(request.args.get('size', 50)))) except (TypeError, ValueError): size = 50 search = (request.args.get('search') or '').strip().lower() try: column_filters = request.get_json(silent=True) if request.method == 'POST' else None except Exception: column_filters = None if not isinstance(column_filters, dict): column_filters = {} column_filters = column_filters.get('filters', column_filters) if not isinstance(column_filters, dict): column_filters = {} allowed_columns = set(SALES_MANAGEMENT_COLUMNS) normalized_filters = {} for field, value in column_filters.items(): if str(field) not in allowed_columns: continue if isinstance(value, list): choices = [str(item).strip().lower() for item in value if str(item).strip()] if choices: normalized_filters[str(field)] = choices elif str(value).strip(): normalized_filters[str(field)] = str(value).strip().lower() column_filters = normalized_filters stored_rows = query( 'SELECT id, ' + ', '.join('`%s`' % c for c in SALES_MANAGEMENT_DB_COLUMNS) + ' FROM sales_management_records ORDER BY id DESC' ) result = [] for stored in stored_rows: row = {label: stored.get(db_col) or '' for label, db_col in zip(SALES_MANAGEMENT_COLUMNS, SALES_MANAGEMENT_DB_COLUMNS)} row['_id'] = stored.get('id') if search and search not in str(row['货品编码']).lower() and search not in str(row['货品名称']).lower(): continue matched = True for field, value in column_filters.items(): actual = str(row.get(field) or '').lower() if isinstance(value, list): choices = [str(item).strip().lower() for item in value] if '__empty__' in choices and not actual: continue if actual not in choices: matched = False break elif str(value) not in actual: matched = False break if not matched: continue result.append(row) offset = (page - 1) * size return jsonify({'code': 0, 'columns': SALES_MANAGEMENT_COLUMNS, 'data': result[offset:offset + size], 'total': len(result), 'page': page, 'size': size}) @sku_forecast_bp.route('/api/sales_management/column-values') @login_required def sales_management_column_values(): if not operation_required(SALES_MANAGEMENT_PAGE, 'view'): return jsonify({'code': 403, 'msg': '无访问权限'}), 403 field = (request.args.get('field') or '').strip() if field not in SALES_MANAGEMENT_COLUMNS: return jsonify({'code': 400, 'msg': '字段参数错误'}) search = (request.args.get('search') or '').strip().lower() stored_rows = query('SELECT ' + ', '.join('`%s`' % c for c in SALES_MANAGEMENT_DB_COLUMNS) + ' FROM sales_management_records') values = set() field_index = SALES_MANAGEMENT_COLUMNS.index(field) db_col = SALES_MANAGEMENT_DB_COLUMNS[field_index] for row in stored_rows: value = str(row.get(db_col) or '').strip() if search: combined = ' '.join(str(row.get(c) or '') for c in SALES_MANAGEMENT_DB_COLUMNS).lower() if search not in combined: continue values.add(value if value else '__EMPTY__') return jsonify({'code': 0, 'data': sorted(values, key=lambda value: (value == '__EMPTY__', value.casefold()))}) @sku_forecast_bp.route('/api/sales_management/detail/') @login_required def sales_management_detail(record_id): if not operation_required(SALES_MANAGEMENT_PAGE, 'view'): return jsonify({'code': 403, 'msg': '无访问权限'}), 403 rows = query( 'SELECT id, ' + ', '.join('`%s`' % c for c in SALES_MANAGEMENT_DB_COLUMNS) + ' FROM sales_management_records WHERE id=%s', (record_id,) ) if not rows: return jsonify({'code': 404, 'msg': '记录不存在'}), 404 return jsonify({'code': 0, 'data': {label: rows[0].get(db_col) or '' for label, db_col in zip(SALES_MANAGEMENT_COLUMNS, SALES_MANAGEMENT_DB_COLUMNS)}}) @sku_forecast_bp.route('/api/sales_management/delete/', methods=['POST', 'DELETE']) @login_required def sales_management_delete(record_id): if not operation_required(SALES_MANAGEMENT_PAGE, 'delete'): return jsonify({'code': 403, 'msg': '无删除权限'}), 403 affected = execute('DELETE FROM sales_management_records WHERE id=%s', (record_id,)) if not affected: return jsonify({'code': 404, 'msg': '记录不存在'}), 404 add_operation_log('动销管理删除', f'删除动销管理记录 ID={record_id}', page='动销管理', field='删除') return jsonify({'code': 0, 'msg': '删除成功'}) @sku_forecast_bp.route('/api/sales_management/template') @login_required def sales_management_template(): """下载包含全部字段且顺序固定的动销管理导入模板。""" if not operation_required(SALES_MANAGEMENT_PAGE, 'import'): return jsonify({'code': 403, 'msg': '无导入权限'}), 403 import io import openpyxl wb = openpyxl.Workbook() ws = wb.active ws.title = '动销管理导入模板' ws.append(SALES_MANAGEMENT_COLUMNS) for cell in ws[1]: cell.font = openpyxl.styles.Font(name='微软雅黑', bold=True, color='FFFFFF') cell.fill = openpyxl.styles.PatternFill('solid', fgColor='1677FF') cell.alignment = openpyxl.styles.Alignment(horizontal='center', vertical='center') for index, label in enumerate(SALES_MANAGEMENT_COLUMNS, 1): ws.column_dimensions[openpyxl.utils.get_column_letter(index)].width = max(12, min(24, len(label) * 2 + 4)) ws.freeze_panes = 'A2' output = io.BytesIO() wb.save(output) output.seek(0) return send_file(output, as_attachment=True, download_name='动销管理导入模板.xlsx', mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet') @sku_forecast_bp.route('/api/sales_management/import', methods=['POST']) @login_required def sales_management_import(): """导入动销管理 Excel；导入记录追加保存，不读取库存/出库表。""" if not operation_required(SALES_MANAGEMENT_PAGE, 'import'): return jsonify({'code': 403, 'msg': '无导入权限'}), 403 upload = request.files.get('file') if not upload or not upload.filename: return jsonify({'code': 400, 'msg': '请选择要导入的Excel文件'}), 400 if not upload.filename.lower().endswith(('.xlsx', '.xlsm')): return jsonify({'code': 400, 'msg': '仅支持 .xlsx 或 .xlsm 文件'}), 400 try: import openpyxl workbook = openpyxl.load_workbook(upload, read_only=True, data_only=True) sheet = workbook.active rows = list(sheet.iter_rows(values_only=True)) if not rows: return jsonify({'code': 400, 'msg': 'Excel没有数据'}), 400 headers = [str(value or '').strip() for value in rows[0]] if headers != SALES_MANAGEMENT_COLUMNS: return jsonify({'code': 400, 'msg': '表头不匹配，请使用“动销管理导入模板”并保持37个字段及顺序不变'}), 400 values = [] for raw in rows[1:]: cells = list(raw[:len(SALES_MANAGEMENT_COLUMNS)]) if len(cells) < len(SALES_MANAGEMENT_COLUMNS): cells.extend([''] * (len(SALES_MANAGEMENT_COLUMNS) - len(cells))) if not any(value is not None and str(value).strip() for value in cells): continue normalized = [] for value in cells: if value is None: normalized.append('') elif hasattr(value, 'strftime'): normalized.append(value.strftime('%Y-%m-%d %H:%M:%S')) else: normalized.append(str(value).strip()) values.append(normalized) if not values: return jsonify({'code': 400, 'msg': '没有可导入的数据行'}), 400 placeholders = ','.join(['%s'] * len(SALES_MANAGEMENT_DB_COLUMNS)) sql = 'INSERT INTO sales_management_records (' + ','.join('`%s`' % c for c in SALES_MANAGEMENT_DB_COLUMNS) + ') VALUES (' + placeholders + ')' for values_row in values: query(sql, tuple(values_row)) add_operation_log('动销管理导入', f'导入动销管理数据，共{len(values)}条', page='动销管理', field='导入') return jsonify({'code': 0, 'msg': f'导入成功，共{len(values)}条'}) except Exception as exc: return jsonify({'code': 500, 'msg': f'导入失败：{exc}'}), 500 @sku_forecast_bp.route('/api/sales_management/export') @login_required def sales_management_export(): """导出动销管理导入数据；fields按表头字段名逗号分隔。""" if not operation_required(SALES_MANAGEMENT_PAGE, 'export'): return jsonify({'code': 403, 'msg': '无导出权限'}), 403 import io import openpyxl from openpyxl.styles import Font, PatternFill, Alignment selected = [value.strip() for value in (request.args.get('fields') or '').split(',') if value.strip()] if not selected: selected = SALES_MANAGEMENT_COLUMNS[:] selected = [value for value in selected if value in SALES_MANAGEMENT_COLUMNS] if not selected: return jsonify({'code': 400, 'msg': '请至少选择一个导出字段'}), 400 field_pairs = [(label, db_col) for label, db_col in zip(SALES_MANAGEMENT_COLUMNS, SALES_MANAGEMENT_DB_COLUMNS) if label in selected] order = {label: index for index, label in enumerate(selected)} field_pairs.sort(key=lambda pair: order[pair[0]]) stored_rows = query('SELECT ' + ', '.join('`%s`' % db_col for _, db_col in field_pairs) + ' FROM sales_management_records ORDER BY id DESC') wb = openpyxl.Workbook() ws = wb.active ws.title = '动销管理' ws.append([label for label, _ in field_pairs]) for cell in ws[1]: cell.font = Font(name='微软雅黑', bold=True, color='FFFFFF') cell.fill = PatternFill('solid', fgColor='1677FF') cell.alignment = Alignment(horizontal='center', vertical='center') for row in stored_rows: ws.append([row.get(db_col) or '' for _, db_col in field_pairs]) for index, (label, _) in enumerate(field_pairs, 1): ws.column_dimensions[openpyxl.utils.get_column_letter(index)].width = max(12, min(24, len(label) * 2 + 4)) ws.freeze_panes = 'A2' output = io.BytesIO() wb.save(output) output.seek(0) return send_file(output, as_attachment=True, download_name='动销管理导出.xlsx', mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet') @sku_forecast_bp.route('/api/sku_forecast/basic_list') @login_required def sku_forecast_basic_list(): """SKU基础信息列表：产品分级、链接、产品ID、货品编号、货品名称、运营负责人、产品类型(爆品/日销/新品)""" page = int(request.args.get('page', 1)) size = int(request.args.get('size', 50)) offset = (page - 1) * size search = request.args.get('search', '').strip() grade_filter = request.args.get('grade', '').strip() sku_type_filter = request.args.get('sku_type', '').strip() # 与“计划预测”主列表保持同一 SKU 范围：排除包材/其他，以及已从计划预测隐藏的 SKU。 where = [ "p.product_type NOT IN ('包材', '其他')", "p.product_code NOT IN (SELECT product_code FROM plan_hidden)", "p.deleted_at IS NULL", ] params = [] if search: where.append("(p.product_code LIKE %s OR p.product_name LIKE %s)") params.extend([f"%{search}%", f"%{search}%"]) if sku_type_filter: where.append("p.sku_type = %s") params.append(sku_type_filter) where_clause = " AND ".join(where) # 销量数据 sales_map = _compute_sales_map() # 全量捞取（分级按销量计算，需在 Python 端处理） all_rows = query(f""" SELECT p.id AS 产品ID, p.product_code AS 货品编号, p.product_name AS 货品名称, p.operator AS 运营负责人, p.sku_type AS 产品类型, p.link AS 链接 FROM products p WHERE {where_clause} ORDER BY p.product_code ASC """, params) for row in all_rows: pc = row['货品编号'] sales = sales_map.get(pc, {}) monthly_sales = sales.get('本月销量', 0) or 0 if monthly_sales <= 0: monthly_sales = sales.get('上月销量', 0) or 0 row['产品分级'] = get_product_grade(monthly_sales, row.get('货品名称') or '', '') # 分级筛选（Python端） if grade_filter: all_rows = [r for r in all_rows if r.get('产品分级') == grade_filter] total = len(all_rows) page_rows = all_rows[offset:offset + size] return jsonify({"code": 0, "data": page_rows, "total": total, "page": page, "size": size}) @sku_forecast_bp.route('/api/sku_forecast/update_field', methods=['POST']) @login_required def sku_forecast_update_field(): """更新SKU基础信息的可编辑字段（运营负责人/产品类型/链接）""" data = request.get_json(silent=True) or {} product_code = str(data.get('product_code', '')).strip() field = data.get('field', '') value = data.get('value', '') if not product_code: return jsonify({"code": 400, "msg": "货品编号不能为空"}) allowed = { '运营负责人': 'operator', '产品类型': 'sku_type', '链接': 'link', } if field not in allowed: return jsonify({"code": 400, "msg": "不允许的字段"}) if field == '产品类型' and value not in ('', '爆品', '日销', '新品'): return jsonify({"code": 400, "msg": "产品类型只能为：爆品/日销/新品"}) value = str(value).strip() db_col = allowed[field] query(f"UPDATE products SET `{db_col}`=%s WHERE product_code=%s", (value, product_code)) add_operation_log('重点SKU预测', f'修改 {product_code} 的 {field} 为 {value}') return jsonify({"code": 0, "msg": "更新成功"}) @sku_forecast_bp.route('/api/sku_forecast/sales_list') @login_required def sku_forecast_sales_list(): """历史销量参考列表：货品编号、货品名称、昨天销量、上月销量、本月销量、7日均销、15天均销""" page = int(request.args.get('page', 1)) size = int(request.args.get('size', 50)) offset = (page - 1) * size search = request.args.get('search', '').strip() sales_map = _compute_sales_map() where = ["p.deleted_at IS NULL"] params = [] if search: where.append("(p.product_code LIKE %s OR p.product_name LIKE %s)") params.extend([f"%{search}%", f"%{search}%"]) where_clause = " AND ".join(where) all_rows = query(f""" SELECT p.product_code AS 货品编号, p.product_name AS 货品名称 FROM products p WHERE {where_clause} ORDER BY p.product_code ASC """, params) for row in all_rows: sales = sales_map.get(row['货品编号'], {}) row['昨天销量'] = sales.get('昨天销量', 0) row['上月销量'] = sales.get('上月销量', 0) row['本月销量'] = sales.get('本月销量', 0) row['7日均销'] = sales.get('7日均销', 0) row['15天均销'] = sales.get('15天均销', 0) # 无销量数据的排后 all_rows.sort(key=lambda r: (r['昨天销量'] == 0 and r['本月销量'] == 0, -r['本月销量'], r['货品编号'])) total = len(all_rows) page_rows = all_rows[offset:offset + size] return jsonify({"code": 0, "data": page_rows, "total": total, "page": page, "size": size})