"""产品管理路由""" import os import uuid from datetime import datetime from flask import Blueprint, request, jsonify, session, redirect, url_for, render_template from services.db import query, add_operation_log, get_db 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_cloud_inventory_cached, compute_external_inventory_cached, compute_product_in_transit_cached, _determine_trend) from services.forecast import (is_new_product, get_product_grade, compute_daily_forecast, sales_predict_config_cached) from routes.auth import login_required, role_required, operation_required from services.field_perms import filter_rows as _filter_field_rows products_bp = Blueprint('products', __name__) # 计划预测下单信息弹窗显示内料的货品名称关键字：海外品牌还要命中国家名才显示内料 INNER_MATERIAL_NAME_KEYS = ('新西兰', '新加坡', '韩国') # BOM 的 supplier 字段历史上混入过产品类型（例如“原料”），不能作为供应商名称展示。 # 这些值只在 BOM 供应商字段中视为无效，随后回退到供应商主表或物料基础资料。 BOM_INVALID_SUPPLIER_VALUES = frozenset(('原料', '包材', '成品', '其他')) def _normalize_bom_supplier(value): """清理 BOM 供应商输入，避免把产品类型写入供应商字段。""" value = str(value or '').strip() return '' if value in BOM_INVALID_SUPPLIER_VALUES else value def _bom_material_supplier_sql(b_alias='b', material_alias='m', supplier_alias='ms'): """返回 BOM 物料供应商的统一 SQL 表达式。 优先使用 BOM 中明确填写的真实供应商；历史无效类型值（如“原料”） 视为空，再回退到供应商主表，最后回退到物料基础资料中的匹配名称。 """ invalid = ', '.join("'%s'" % value for value in sorted(BOM_INVALID_SUPPLIER_VALUES)) return ( f"COALESCE(NULLIF(CASE WHEN TRIM({b_alias}.supplier) IN ({invalid}) " f"THEN '' ELSE TRIM({b_alias}.supplier) END, ''), " f"NULLIF(TRIM({supplier_alias}.supplier_name), ''), " f"NULLIF(TRIM({material_alias}.supplier_name_match), ''), '')" ) def _check_product_edit_perm(): """统一检查产品编辑权限，角色默认权限与用户覆盖规则一致。""" return operation_required('products', 'edit') @products_bp.route('/products') @login_required def products_page(): if not role_required('products'): return redirect(url_for('auth.index')) return render_template('base.html', page='products') @products_bp.route('/brand-management') @login_required def brand_management_page(): if not role_required('products'): return redirect(url_for('auth.index')) return render_template('base.html', page='brand_management') @products_bp.route('/api/brand-management/list') @login_required def brand_management_list(): if not role_required('products'): return jsonify({"code": 403, "msg": "权限不足"}) search = (request.args.get('search') or '').strip() category = (request.args.get('category') or '').strip() page = max(int(request.args.get('page', 1)), 1) size = 10 # 产品表中的品牌 + 手动维护的品牌合并展示；产品数量不再作为页面字段。 product_rows = query("""SELECT TRIM(brand) AS brand FROM products WHERE deleted_at IS NULL AND brand IS NOT NULL AND TRIM(brand) <> '' GROUP BY TRIM(brand) ORDER BY TRIM(brand)""") saved_rows = query("SELECT brand, category, updated_at FROM brand_information ORDER BY brand") saved = {r['brand']: r for r in saved_rows} names = {r['brand'] for r in product_rows} names.update(saved.keys()) rows = [] for brand in sorted(names, key=lambda x: x.casefold()): if search and search.casefold() not in brand.casefold(): continue info = saved.get(brand) or {} cat = info.get('category') or '' if category == '未分类' and cat: continue if category in ('国内', '海外') and cat != category: continue updated = info.get('updated_at') rows.append({'brand': brand, 'category': cat, 'updated_at': updated.strftime('%Y-%m-%d %H:%M:%S') if updated and hasattr(updated, 'strftime') else (str(updated) if updated else '')}) total = len(rows) return jsonify({"code": 0, "data": rows[(page - 1) * size:page * size], "total": total, "page": page, "size": size}) @products_bp.route('/api/brand-management/create', methods=['POST']) @login_required def brand_management_create(): if not _check_product_edit_perm(): return jsonify({"code": 403, "msg": "当前账号没有编辑权限"}) data = request.get_json(silent=True) or {} brand = (data.get('brand') or '').strip() category = (data.get('category') or '').strip() if not brand: return jsonify({"code": 400, "msg": "品牌名称不能为空"}) if category not in ('', '国内', '海外'): return jsonify({"code": 400, "msg": "品牌分类只能是国内或海外"}) exists = query("SELECT 1 FROM brand_information WHERE brand=%s LIMIT 1", (brand,)) product_exists = query("SELECT 1 FROM products WHERE deleted_at IS NULL AND TRIM(brand)=%s LIMIT 1", (brand,)) if exists or product_exists: return jsonify({"code": 400, "msg": "该品牌已存在"}) query("INSERT INTO brand_information (brand, category) VALUES (%s, %s)", (brand, category)) add_operation_log('新增品牌', f"新增品牌「{brand}」", page='品牌信息管理', field='品牌名称') return jsonify({"code": 0, "msg": "新增成功"}) @products_bp.route('/api/brand-management/update', methods=['POST']) @login_required def brand_management_update(): if not _check_product_edit_perm(): return jsonify({"code": 403, "msg": "当前账号没有编辑权限"}) data = request.get_json(silent=True) or {} old_brand = (data.get('old_brand') or data.get('brand') or '').strip() new_brand = (data.get('new_brand') or old_brand).strip() category = data.get('category') category = (category if category is not None else '').strip() if not old_brand or not new_brand: return jsonify({"code": 400, "msg": "品牌名称不能为空"}) if category not in ('', '国内', '海外'): return jsonify({"code": 400, "msg": "品牌分类只能是国内或海外"}) if old_brand != new_brand: conflict = query("SELECT 1 FROM brand_information WHERE brand=%s LIMIT 1", (new_brand,)) product_conflict = query("SELECT 1 FROM products WHERE deleted_at IS NULL AND TRIM(brand)=%s LIMIT 1", (new_brand,)) if conflict or product_conflict: return jsonify({"code": 400, "msg": "新的品牌名称已存在"}) has_product = query("SELECT 1 FROM products WHERE deleted_at IS NULL AND TRIM(brand)=%s LIMIT 1", (old_brand,)) has_saved = query("SELECT 1 FROM brand_information WHERE brand=%s LIMIT 1", (old_brand,)) if not has_product and not has_saved: return jsonify({"code": 404, "msg": "未找到该品牌"}) if has_product and old_brand != new_brand: query("UPDATE products SET brand=%s WHERE deleted_at IS NULL AND TRIM(brand)=%s", (new_brand, old_brand)) if has_saved: query("UPDATE brand_information SET brand=%s, category=%s, updated_at=CURRENT_TIMESTAMP WHERE brand=%s", (new_brand, category, old_brand)) else: query("INSERT INTO brand_information (brand, category) VALUES (%s, %s)", (new_brand, category)) add_operation_log('编辑品牌', f"将品牌「{old_brand}」修改为「{new_brand}」", page='品牌信息管理', field='品牌名称') return jsonify({"code": 0, "msg": "保存成功"}) @products_bp.route('/api/brand-management/delete', methods=['POST']) @login_required def brand_management_delete(): """删除误添加且未被有效产品使用的品牌。""" if not operation_required('brand_management', 'delete'): return jsonify({"code": 403, "msg": "当前账号没有删除品牌权限"}) data = request.get_json(silent=True) or {} brand = (data.get('brand') or '').strip() if not brand: return jsonify({"code": 400, "msg": "品牌名称不能为空"}) product_rows = query( "SELECT COALESCE(NULLIF(TRIM(product_type), ''), '未分类') AS product_type, " "COUNT(*) AS cnt FROM products " "WHERE deleted_at IS NULL AND TRIM(brand)=%s " "GROUP BY COALESCE(NULLIF(TRIM(product_type), ''), '未分类') " "ORDER BY cnt DESC", (brand,), ) product_count = sum(int(row.get('cnt') or 0) for row in product_rows) if product_count: usage_counts = { (row.get('product_type') or '未分类'): int(row.get('cnt') or 0) for row in product_rows } usage_summary = '、'.join( f"{product_type}{count}条" for product_type, count in usage_counts.items() ) return jsonify({ "code": 409, "msg": f"品牌「{brand}」正在被产品信息管理基础表中的{product_count}条未回收记录使用" f"（{usage_summary}），请先修改这些产品的品牌后再删除", "product_count": product_count, "usage_counts": usage_counts, }) saved = query("SELECT brand, category FROM brand_information WHERE brand=%s LIMIT 1", (brand,)) if not saved: return jsonify({"code": 404, "msg": "未找到可删除的品牌记录"}) query("DELETE FROM brand_information WHERE brand=%s", (brand,)) add_operation_log( '删除品牌', f"删除品牌「{brand}」", page='品牌信息管理', field='品牌名称', object_type='brand', object_id=brand, old_values=saved[0], ) return jsonify({"code": 0, "msg": "删除成功"}) @products_bp.route('/bom') @login_required def bom_page(): if not role_required('bom'): return redirect(url_for('auth.index')) return render_template('base.html', page='bom') @products_bp.route('/suppliers') @login_required def suppliers_page(): if not role_required('suppliers'): return redirect(url_for('auth.index')) return render_template('base.html', page='suppliers') # ─── API: 产品基础表 ─── @products_bp.route('/api/products/list') @login_required def products_list(): page = int(request.args.get('page', 1)) size = int(request.args.get('size', 50)) offset = (page - 1) * size search = request.args.get('search', '') unified_search = request.args.get('unified_search', '') prod_code = request.args.get('prod_code', '') prod_name = request.args.get('prod_name', '') supplier_filter = request.args.get('supplier', '') brand_filter = request.args.get('brand', '') category1_filter = request.args.get('category1', '') product_type_filter = request.args.get('product_type', '') sort_by = request.args.get('sort_by', 'id') sort_order = request.args.get('sort_order', 'desc') # Whitelist sort columns to prevent SQL injection allowed_sort = ['id', 'goods_code', 'product_code', 'product_name', 'brand', 'cost_price', 'category1', 'category2', 'category3', 'created_at', 'import_date'] if sort_by not in allowed_sort: sort_by = 'id' if sort_order.lower() not in ('asc', 'desc'): sort_order = 'desc' where = [] params = [] if search or unified_search: search_value = search or unified_search where.append("(p.product_name LIKE %s OR p.product_code LIKE %s)") params.extend([f"%{search_value}%", f"%{search_value}%"]) 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 supplier_filter: where.append("(s.supplier_name LIKE %s OR p.supplier_name_match LIKE %s)") params.extend([f"%{supplier_filter}%", f"%{supplier_filter}%"]) if brand_filter: where.append("TRIM(p.brand)=%s") params.append(brand_filter) if category1_filter: where.append("p.category1=%s") params.append(category1_filter) where_clause = " AND ".join(where) if where else "1=1" # 加载供应商列表用于自动匹配编码 _suppliers = query("SELECT supplier_code, supplier_name FROM suppliers") def _format_product_row(row): if 'created_at' in row and row['created_at']: row['created_at'] = row['created_at'].strftime('%Y-%m-%d %H:%M:%S') if hasattr(row['created_at'], 'strftime') else str(row['created_at'])[:19] if 'import_date' in row and row['import_date']: row['import_date'] = row['import_date'].strftime('%Y-%m-%d') if hasattr(row['import_date'], 'strftime') else str(row['import_date'])[:10] if 'cost_price' in row and row['cost_price'] is not None: row['cost_price'] = float(row['cost_price']) if row.get('logistics_time') is None: row['logistics_time'] = 0 if row.get('lead_time') is None: row['lead_time'] = 0 row['是否新品'] = is_new_product(row.get('created_at', '')) if not row.get('product_type'): row['product_type'] = get_product_type_cached(row.get('product_name', ''), row.get('brand'), row.get('product_type')) # 自动匹配供应商编码（如果supplier_code为空但有supplier_name_match） pname = (row.get('supplier_name_match') or '').strip() pcode = (row.get('supplier_code') or '').strip() if pname and (not pcode or pcode == ''): for s in _suppliers: sname = (s['supplier_name'] or '').strip() if pname in sname or sname in pname: row['supplier_code'] = s['supplier_code'] row['supplier_name_match'] = s['supplier_name'] try: query("UPDATE products SET supplier_code=%s, supplier_name_match=%s WHERE product_code=%s", (s['supplier_code'], s['supplier_name'], row['product_code'])) except: pass break return row # 如果有类型筛选，需要全量查出来再在 Python 中过滤 if product_type_filter: all_rows = query(f""" SELECT p.id, p.goods_code, p.product_code, p.product_name, p.brand, p.cost_price, p.product_type, p.specification, p.lead_time, p.logistics_time, p.min_order_qty, p.supplier_code, p.supplier_name_match, p.procurement_follower, p.procurement_planner, p.platform, p.operator, p.developer, p.category1, p.category2, p.category3, p.created_at, p.import_date, p.custom_field1, p.custom_field2, p.custom_field3, p.custom_field4, p.custom_field5, COALESCE(s.supplier_name, p.supplier_name_match) AS supplier_name FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE {where_clause} AND p.deleted_at IS NULL ORDER BY p.id DESC """, params) filtered = [] for row in all_rows: row = _format_product_row(row) if product_type_filter == '非成品': # 非成品：排除成品 if row['product_type'] != '成品': filtered.append(row) elif product_type_filter == '非包材': # 非包材：排除包材和其他类 if row['product_type'] != '包材' and row['product_type'] != '其他': filtered.append(row) elif product_type_filter == '非成品专用': # 非成品专用：仅包材和其他 if row['product_type'] == '包材' or row['product_type'] == '其他': filtered.append(row) else: if row['product_type'] == product_type_filter: filtered.append(row) reverse = sort_order.lower() == 'desc' if sort_by in ['cost_price', 'id']: filtered.sort(key=lambda x: x.get(sort_by, 0) if x.get(sort_by) else 0, reverse=reverse) else: filtered.sort(key=lambda x: str(x.get(sort_by, '') or '').lower(), reverse=reverse) total = len(filtered) rows = filtered[offset:offset + size] else: total = query(f"""SELECT COUNT(*) AS cnt FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE {where_clause} AND p.deleted_at IS NULL""", params)[0]['cnt'] rows = query(f""" SELECT p.id, p.goods_code, p.product_code, p.product_name, p.brand, p.cost_price, p.product_type, p.specification, p.lead_time, p.logistics_time, p.min_order_qty, p.supplier_code, p.supplier_name_match, p.procurement_follower, p.procurement_planner, p.platform, p.operator, p.developer, p.category1, p.category2, p.category3, p.created_at, p.import_date, p.custom_field1, p.custom_field2, p.custom_field3, p.custom_field4, p.custom_field5, COALESCE(s.supplier_name, p.supplier_name_match) AS supplier_name FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE {where_clause} AND p.deleted_at IS NULL ORDER BY p.{sort_by} {sort_order} LIMIT {size} OFFSET {offset} """, params) for row in rows: _format_product_row(row) _filter_field_rows('products', rows, session.get('user_id'), session.get('role'), page_view=bool(role_required('products')), page_edit=operation_required('products', 'edit')) return jsonify({ "code": 0, "data": { "total": total, "page": page, "size": size, "rows": rows } }) @products_bp.route('/api/products/filters') @login_required def products_filters(): scope = request.args.get('scope', 'all') if scope == 'active': scope_clause = 'deleted_at IS NULL AND ' elif scope == 'recycle': scope_clause = 'deleted_at IS NOT NULL AND ' else: scope_clause = '' brands = query( f"SELECT DISTINCT TRIM(brand) AS brand FROM products WHERE {scope_clause}" "brand IS NOT NULL AND TRIM(brand) != '' ORDER BY TRIM(brand)" ) categories = query( f"SELECT DISTINCT category1 FROM products WHERE {scope_clause}" "category1 IS NOT NULL AND category1 != '' ORDER BY category1" ) return jsonify({ "code": 0, "data": { "brands": [b['brand'] for b in brands], "categories": [c['category1'] for c in categories] } }) @products_bp.route('/api/products/detail') @login_required def products_detail(): pid = request.args.get('id', 0) row = query(""" SELECT p.*, COALESCE(s.supplier_name, p.supplier_name_match) AS supplier_name FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE p.id=%s """, (pid,)) if not row: return jsonify({"code": 404, "msg": "未找到该产品"}) row = row[0] if 'created_at' in row and row['created_at']: row['created_at'] = row['created_at'].strftime('%Y-%m-%d %H:%M:%S') if hasattr(row['created_at'], 'strftime') else str(row['created_at'])[:19] if 'import_date' in row and row['import_date']: row['import_date'] = row['import_date'].strftime('%Y-%m-%d') if hasattr(row['import_date'], 'strftime') else str(row['import_date'])[:10] if 'cost_price' in row and row['cost_price'] is not None: row['cost_price'] = float(row['cost_price']) row['product_type'] = get_product_type_cached(row.get('product_name', ''), row.get('brand', ''), row.get('product_type')) return jsonify({"code": 0, "data": row}) @products_bp.route('/api/product/update_fields', methods=['POST']) @login_required def product_update_fields(): """更新产品字段（仅管理员）""" if not _check_product_edit_perm(): return jsonify({"code": 403, "msg": "仅管理员可操作"}) data = request.json product_code = data.get('product_code', '') field = data.get('field', '') value = data.get('value', '') allowed_fields = ['lead_time', 'logistics_time', 'min_order_qty', 'developer'] if field not in allowed_fields: return jsonify({"code": 400, "msg": "不允许的字段"}) try: query(f"UPDATE products SET {field}=%s WHERE product_code=%s", (value, product_code)) add_operation_log('产品编辑', f'修改 {product_code} 的 {field} 为 {value}') return jsonify({"code": 0, "msg": "更新成功"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/product/batch_update', methods=['POST']) @login_required def product_batch_update(): """批量更新产品字段（编辑弹窗用）""" if not _check_product_edit_perm(): return jsonify({"code": 403, "msg": "仅管理员可操作"}) data = request.json product_code = data.get('product_code', '') fields = data.get('fields', {}) if not product_code or not fields: return jsonify({"code": 400, "msg": "参数不完整"}) allowed_fields = [ 'lead_time', 'logistics_time', 'min_order_qty', 'developer', 'cost_price', 'brand', 'specification', 'product_type', 'category1', 'category2', 'category3', 'supplier_code', 'supplier_name_match', 'procurement_follower', 'procurement_planner', 'operator', 'platform', 'product_name', 'created_at', 'custom_field1', 'custom_field2', 'custom_field3', 'custom_field4', 'custom_field5', 'remark', 'product_code' ] old_code = product_code new_code = fields.get('product_code', old_code) set_parts = [] params = [] for field, value in fields.items(): if field not in allowed_fields: continue if value == '' or value is None: value = None # 数值字段范围校验 if field in ('lead_time', 'logistics_time') and value is not None: try: if int(value) > 999: return jsonify({"code": 400, "msg": f"{field} 数值不能超过 999"}) except (ValueError, TypeError): pass set_parts.append(f"{field}=%s") params.append(value) if not set_parts: return jsonify({"code": 400, "msg": "没有可更新的字段"}) set_clause = ", ".join(set_parts) params.append(old_code) try: # 如果货品编号变更 if new_code != old_code and new_code: # 检查目标编号是否已被其他产品使用 existing_owner = query("SELECT id FROM products WHERE product_code=%s AND product_code!=%s", (new_code, old_code)) if existing_owner: # 目标编号已存在 → 合并：更新关联表引用，删除旧产品 for table_name, col_name in [ ('bom', 'product_code'), ('pending_orders', 'product_code'), ('purchase_order_items', 'product_code'), ('plan_hidden', 'product_code'), ]: query(f"UPDATE IGNORE {table_name} SET {col_name}=%s WHERE {col_name}=%s", (new_code, old_code)) # 删除旧产品记录 query("DELETE FROM products WHERE product_code=%s", (old_code,)) # 更新已存在的目标产品（合并表单数据） target_params = params[:-1] # 参数（不含old_code） target_params.append(new_code) query(f"UPDATE products SET {set_clause} WHERE product_code=%s", target_params) change_info = f'{old_code}→{new_code}(已覆盖)' else: # 目标编号不存在，直接更新关联表和产品 for table_name, col_name in [ ('bom', 'product_code'), ('pending_orders', 'product_code'), ('purchase_order_items', 'product_code'), ('plan_hidden', 'product_code'), ]: query(f"UPDATE IGNORE {table_name} SET {col_name}=%s WHERE {col_name}=%s", (new_code, old_code)) query(f"UPDATE products SET {set_clause} WHERE product_code=%s", params) change_info = f'{old_code}→{new_code}' else: query(f"UPDATE products SET {set_clause} WHERE product_code=%s", params) change_info = old_code # 产品名称变更时同步更新BOM的名称 if 'product_name' in fields and fields['product_name']: new_name = fields['product_name'] final_code = new_code if new_code != old_code and new_code else old_code query("UPDATE bom SET product_name=%s WHERE product_code=%s", (new_name, final_code)) query("UPDATE bom SET material_name=%s WHERE material_code=%s", (new_name, final_code)) changed = ', '.join([f'{k}={v}' for k, v in fields.items() if v is not None]) add_operation_log('产品批量编辑', f'{change_info}: {changed}') return jsonify({"code": 0, "msg": "更新成功"}) except Exception as e: return jsonify({"code": 500, "msg": f"操作失败: {str(e)}"}) @products_bp.route('/api/product/full_detail') @login_required def product_full_detail(): code = request.args.get('code', '') if not code: return jsonify({"code": 400, "msg": "缺少货品编号"}) row = query(""" SELECT p.*, COALESCE(s.supplier_name, p.supplier_name_match) AS supplier_name FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE p.product_code=%s """, (code,)) if not row: return jsonify({"code": 404, "msg": "未找到该产品"}) product = row[0] if product.get('created_at'): product['created_at'] = product['created_at'].strftime('%Y-%m-%d %H:%M:%S') if hasattr(product['created_at'], 'strftime') else str(product['created_at'])[:19] if product.get('import_date'): product['import_date'] = product['import_date'].strftime('%Y-%m-%d') if hasattr(product['import_date'], 'strftime') else str(product['import_date'])[:10] if product.get('cost_price') is not None: product['cost_price'] = float(product['cost_price']) product['产品种类'] = get_product_type_cached(product.get('product_name', ''), product.get('brand', ''), product.get('product_type')) sales = query(""" SELECT 日期, SUM(货品数量) AS 日销量 FROM daily_outbound WHERE 货品编号=%s GROUP BY 日期 ORDER BY 日期 ASC """, (code,)) sales_data = [] for s in sales: d = s['日期'].strftime('%Y-%m-%d') if hasattr(s['日期'], 'strftime') else str(s['日期'])[:10] sales_data.append({ 'date': d, 'qty': int(s['日销量']) if s['日销量'] else 0 }) return jsonify({ "code": 0, "data": { "product": product, "sales": sales_data } }) @products_bp.route('/api/product/detail_with_bom') @login_required def product_detail_with_bom(): """产品详情 + BOM物料清单 + 订单数量""" code = request.args.get('code', '') include_inner = request.args.get('include_inner', '').lower() in ('1', 'true', 'yes') if not code: return jsonify({"code": 400, "msg": "缺少货品编号"}) # 1. 产品基础信息 row = query(""" SELECT p.*, COALESCE(s.supplier_name, p.supplier_name_match) AS supplier_name FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE p.product_code=%s """, (code,)) if not row: # BOM 管理列表可能包含仅存在于 bom 表、但产品基础资料尚未建立的历史产品。 # 编辑弹窗需要先读出这类产品的 BOM；保存接口会负责补建产品记录。 if include_inner: bom_product = query(""" SELECT product_code, MAX(product_name) AS product_name FROM bom WHERE product_code=%s GROUP BY product_code """, (code,)) if bom_product: product = { 'product_code': bom_product[0]['product_code'], 'product_name': bom_product[0].get('product_name') or code, 'brand': '', 'product_type': '成品', 'supplier_name_match': '', 'supplier_name': '', 'cost_price': 0, 'remark': '', } else: return jsonify({"code": 404, "msg": "未找到该产品"}) else: return jsonify({"code": 404, "msg": "未找到该产品"}) else: product = row[0] if product.get('created_at'): product['created_at'] = product['created_at'].strftime('%Y-%m-%d %H:%M:%S') if hasattr(product['created_at'], 'strftime') else str(product['created_at'])[:19] if product.get('import_date'): product['import_date'] = product['import_date'].strftime('%Y-%m-%d') if hasattr(product['import_date'], 'strftime') else str(product['import_date'])[:10] if product.get('cost_price') is not None: product['cost_price'] = float(product['cost_price']) product['产品种类'] = get_product_type_cached(product.get('product_name', ''), product.get('brand', ''), product.get('product_type')) product['批注'] = product.get('remark') or '' # 计划预测下单信息弹窗：include_inner=auto 时按品牌分类 + 货品名称自动决定是否包含内料。 # 海外品牌商品还要看货品名称是否含新西兰/新加坡/韩国，命中才显示内料（并同步到包材订单管理）； # 其余海外品与国内品一致，保持原逻辑不含内料。 if request.args.get('include_inner', '').lower() == 'auto': _brand = (product.get('brand') or '').strip() _brand_rows = query(""" SELECT category FROM brand_information WHERE CONVERT(brand USING utf8mb4) COLLATE utf8mb4_unicode_ci = CONVERT(%s USING utf8mb4) COLLATE utf8mb4_unicode_ci LIMIT 1 """, (_brand,)) if _brand else [] _is_overseas_brand = bool(_brand_rows and (_brand_rows[0].get('category') or '').strip() == '海外') _product_name = product.get('product_name') or '' include_inner = _is_overseas_brand and any( _key in _product_name for _key in INNER_MATERIAL_NAME_KEYS ) # 2. 成品订单数量 order = query(""" SELECT COALESCE(SUM(order_qty), 0) AS total_order_qty, COALESCE(SUM(suggested_qty), 0) AS total_suggested_qty FROM purchase_order_items WHERE product_code=%s """, (code,)) order_data = order[0] if order else {'total_order_qty': 0, 'total_suggested_qty': 0} # 3. BOM物料清单 latest_inv = query("SELECT MAX(data_date) AS md FROM inventory_detail")[0]['md'] bom_supplier_expr = _bom_material_supplier_sql() bom_sql = f""" SELECT b.material_code, b.material_name, b.quantity AS bom_qty, m.cost_price AS material_cost, {bom_supplier_expr} AS material_supplier, COALESCE(inv.available_qty, 0) AS available_qty, COALESCE(inv.production_in_transit, 0) AS production_in_transit, COALESCE(inv.purchase_in_transit, 0) AS purchase_in_transit FROM bom b LEFT JOIN products m ON b.material_code = m.product_code LEFT JOIN suppliers ms ON m.supplier_code = ms.supplier_code LEFT JOIN ( SELECT product_code, SUM(available_qty) AS available_qty, SUM(production_in_transit) AS production_in_transit, SUM(purchase_in_transit) AS purchase_in_transit FROM inventory_detail WHERE data_date=%s GROUP BY product_code ) inv ON b.material_code = inv.product_code WHERE b.product_code=%s """ if not include_inner: bom_sql += """ AND b.material_name NOT LIKE '%%内料%%' AND b.material_code NOT LIKE '%%-NL%%' """ bom_sql += " ORDER BY b.id" bom_materials = query(bom_sql, (latest_inv, code)) # 补充 BOM 用量计算的订单数量 order_qty = int(order_data['total_order_qty'] or 0) # 从 bom_packaging 表读取下单前/后包材数据（独立于 pending_orders） pkg_rows = query(""" SELECT material_code, pre_pack_qty, post_pack_qty, pkg_remark FROM bom_packaging WHERE product_code=%s """, (code,)) pkg_map = {} for pr in pkg_rows: mc = str(pr['material_code'] or '').strip() pkg_map[mc] = { 'pre_pack_qty': int(pr['pre_pack_qty'] or 0), 'post_pack_qty': int(pr['post_pack_qty'] or 0), 'pkg_remark': pr.get('pkg_remark') or '', } for mat in bom_materials: mat['bom_qty'] = float(mat['bom_qty']) if mat['bom_qty'] else 0 mat['material_cost'] = float(mat['material_cost']) if mat['material_cost'] else 0 mat['required_qty'] = order_qty * mat['bom_qty'] mat['total_cost'] = round(mat['required_qty'] * mat['material_cost'], 2) mat['available_qty'] = int(mat['available_qty']) mat['production_in_transit'] = int(mat['production_in_transit']) mat['purchase_in_transit'] = int(mat['purchase_in_transit']) saved = pkg_map.get(str(mat['material_code'] or '').strip(), {}) mat['pre_pack_qty'] = int(saved.get('pre_pack_qty', 0) or 0) mat['post_pack_qty'] = int(saved.get('post_pack_qty', 0) or 0) mat['pkg_remark'] = saved.get('pkg_remark', '') or '' # 4. 销量数据 sales = query(""" SELECT 日期, SUM(货品数量) AS 日销量 FROM daily_outbound WHERE 货品编号=%s GROUP BY 日期 ORDER BY 日期 ASC """, (code,)) sales_data = [] for s in sales: d = s['日期'].strftime('%Y-%m-%d') if hasattr(s['日期'], 'strftime') else str(s['日期'])[:10] sales_data.append({ 'date': d, 'qty': int(s['日销量']) if s['日销量'] else 0 }) # 5. 在途订单（从各订单表查询未完结的订单，增加字段） orders_in_transit = query(""" SELECT '国内' AS 订单类型, 日期 AS 下单时间, 货品编号 AS 编码, 货品名称, 数量 AS 下单数量, COALESCE(未到货数, 数量 - COALESCE(实际到货数量, 0)) AS 未到数量, COALESCE(进度, '') AS 下单进度, COALESCE(预计到货日期, '') AS 预计到货时间, 金蝶入库 AS 入库单号 FROM domestic_orders WHERE 货品编号=%s AND COALESCE(状态, '') != '已完结' UNION ALL SELECT '海外' AS 订单类型, 计划下单时间 AS 下单时间, 产品编码 AS 编码, 货品名称, 数量 AS 下单数量, COALESCE(未到货数, 数量 - COALESCE(实际到货数量, 0)) AS 未到数量, COALESCE(进度, '') AS 下单进度, COALESCE(预计到货日期, '') AS 预计到货时间, '' AS 入库单号 FROM overseas_orders WHERE 产品编码=%s AND COALESCE(状态, '') != '已完结' UNION ALL SELECT '包材' AS 订单类型, 下单时间, 编码, 名称 AS 货品名称, 数量 AS 下单数量, 0 AS 未到数量, COALESCE(进度, '') AS 下单进度, COALESCE(到货日期, '') AS 预计到货时间, '' AS 入库单号 FROM packaging_orders WHERE 编码=%s AND COALESCE(状态, '') != '已完结' ORDER BY 下单时间 DESC """, (code, code, code)) for o in orders_in_transit: if o.get('下单时间') and hasattr(o.get('下单时间'), 'strftime'): o['下单时间'] = o['下单时间'].strftime('%Y-%m-%d') if o.get('预计到货日期') and hasattr(o.get('预计到货日期'), 'strftime'): o['预计到货日期'] = o['预计到货日期'].strftime('%Y-%m-%d') if o.get('未到货数') is None: o['未到货数'] = 0 return jsonify({ "code": 0, "data": { "product": product, "order": { "total_order_qty": order_qty, "total_suggested_qty": int(order_data['total_suggested_qty'] or 0) }, "bom_materials": bom_materials, "sales": sales_data, "orders_in_transit": orders_in_transit } }) @products_bp.route('/api/product/planning_detail') @login_required def product_planning_detail(): """产品详情 + 计划预测全量计算字段（用于弹窗展示）""" code = request.args.get('code', '') if not code: return jsonify({"code": 400, "msg": "缺少货品编号"}) # 产品基础信息 row = query(""" SELECT p.*, COALESCE(s.supplier_name, p.supplier_name_match) AS supplier_name FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE p.product_code=%s """, (code,)) if not row: return jsonify({"code": 404, "msg": "未找到该产品"}) product = row[0] if product.get('created_at'): product['created_at'] = product['created_at'].strftime('%Y-%m-%d') if hasattr(product['created_at'], 'strftime') else str(product['created_at'])[:10] if product.get('cost_price') is not None: product['cost_price'] = float(product['cost_price']) if product.get('lead_time') is None: product['lead_time'] = 0 if product.get('logistics_time') is None: product['logistics_time'] = 0 # 字段映射：后端英文名 → 前端中文名 product['供应商名称'] = product.get('supplier_name', '') product['生产货期'] = product.get('lead_time', 0) product['物流时间'] = product.get('logistics_time', 0) product['创建时间'] = product.get('created_at', '') product['产品种类'] = get_product_type_cached(product.get('product_name', ''), product.get('brand', ''), product.get('product_type')) product['批注'] = product.get('remark') or '' # 计算计划预测字段 sales = compute_product_sales_cached() inv_map = compute_product_inventory_cached() inv15_map = compute_product_inventory_15() transit_map = compute_product_in_transit_cached() sp_map = {r['trend_type']: r for r in sales_predict_config_cached()} 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) 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_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_fcast = float(cloud_forecast) if cloud_forecast not in ('近期无销售', 0, '0') else 0 row['云仓数据'] = { '产品分级': row.get('产品分级', ''), '昨日销量': cloud_sales.get('昨日销量', 0), '销售趋势': cloud_trend, '库存': cloud_inventory, '加工厂库存': row.get('加工厂库存', 0), '生产在途': row.get('生产在途', 0), '采购在途': row.get('采购在途', 0), '日销量预测': cloud_forecast, '3天平均销量': cloud_3, '7天平均销量': cloud_7, '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, '30天平均销量': cloud_30, '极值判断': '-', } external_sales = compute_product_sales_extended_cached().get('external', {}).get(pc, {}) external_inventory = compute_external_inventory_cached().get(pc, 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_fcast = float(external_forecast) if external_forecast not in ('近期无销售', 0, '0') else 0 row['外仓数据'] = { '产品分级': row.get('产品分级', ''), '昨日销量': external_sales.get('昨日销量', 0), '销售趋势': external_trend, '库存': external_inventory, '加工厂库存': row.get('加工厂库存', 0), '生产在途': row.get('生产在途', 0), '采购在途': row.get('采购在途', 0), '日销量预测': external_forecast, '3天平均销量': external_3, '7天平均销量': external_7, '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, '30天平均销量': external_30, '极值判断': '-', } return jsonify({"code": 0, "data": row}) @products_bp.route('/api/bom/list') @login_required def bom_list(): """获取BOM列表""" page = int(request.args.get('page', 1)) size = int(request.args.get('size', 20)) offset = (page - 1) * size search = request.args.get('search', '') maint_filter = request.args.get('maint_filter', '') supplier_filter = request.args.get('supplier', '') where_parts = [] params = [] if search: where_parts.append('(b.product_code LIKE %s OR b.product_name LIKE %s OR b.material_code LIKE %s OR b.material_name LIKE %s)') params.extend([f'%{search}%', f'%{search}%', f'%{search}%', f'%{search}%']) having_clause = '' if maint_filter == '有BOM': having_clause = 'HAVING SUM(CASE WHEN b.material_code IS NOT NULL AND b.material_code != \'\' THEN 1 ELSE 0 END) > 0' elif maint_filter == '无BOM': having_clause = 'HAVING SUM(CASE WHEN b.material_code IS NOT NULL AND b.material_code != \'\' THEN 1 ELSE 0 END) = 0' # 供应商筛选：优先使用BOM导入的供应商，未填写时回退到物料基础资料 supplier_join = '' supplier_where = '' if supplier_filter: supplier_join = ''' LEFT JOIN products m ON (b.material_code = m.product_code) LEFT JOIN suppliers ms ON (m.supplier_code = ms.supplier_code) ''' supplier_where = f"AND {_bom_material_supplier_sql()} LIKE %s" params.append(f'%{supplier_filter}%') where_sql = ('WHERE ' + ' AND '.join(where_parts) + supplier_where) if where_parts else ('WHERE 1=1 ' + supplier_where) if supplier_filter else '' # 总数查询 total = query(f""" SELECT COUNT(*) AS cnt FROM ( SELECT b.product_code FROM bom b {supplier_join} {where_sql} GROUP BY b.product_code {having_clause} ) AS t """, params)[0]['cnt'] product_codes = query(f""" SELECT b.product_code FROM bom b {supplier_join} {where_sql} GROUP BY b.product_code {having_clause} ORDER BY MAX(b.id) DESC LIMIT %s OFFSET %s """, params + [size, offset]) if not product_codes: return jsonify({"code": 0, "data": {"total": 0, "rows": [], "product_count": 0}}) codes = [p['product_code'] for p in product_codes] placeholders = ','.join(['%s'] * len(codes)) latest_inv = query("SELECT MAX(data_date) AS md FROM inventory_detail")[0]['md'] # 成品库存映射（按 product_code） prod_inv = {} prod_inv_rows = query(f""" SELECT product_code, SUM(available_qty) AS available_qty, SUM(production_in_transit) AS production_in_transit, SUM(purchase_in_transit) AS purchase_in_transit FROM inventory_detail WHERE data_date='{latest_inv}' AND product_code IN ({placeholders}) GROUP BY product_code """, codes) for r in prod_inv_rows: prod_inv[r['product_code']] = r bom_supplier_expr = _bom_material_supplier_sql() sql = f""" SELECT b.*, p.cost_price AS product_cost_price, p.supplier_name_match AS product_supplier, {bom_supplier_expr} AS material_supplier, COALESCE(m.cost_price, 0) AS material_cost_price, inv.available_qty AS mat_available_qty, inv.production_in_transit AS mat_prod_transit, inv.purchase_in_transit AS mat_purch_transit FROM bom b LEFT JOIN products p ON b.product_code = p.product_code LEFT JOIN products m ON b.material_code = m.product_code LEFT JOIN suppliers ms ON m.supplier_code = ms.supplier_code LEFT JOIN ( SELECT product_code, SUM(available_qty) AS available_qty, SUM(production_in_transit) AS production_in_transit, SUM(purchase_in_transit) AS purchase_in_transit FROM inventory_detail WHERE data_date='{latest_inv}' GROUP BY product_code ) inv ON b.material_code = inv.product_code WHERE b.product_code IN ({placeholders}) ORDER BY b.id DESC """ rows = query(sql, codes) # 为每行补充成品库存 for r in rows: pinv = prod_inv.get(r['product_code'], {}) r['product_available_qty'] = int(pinv.get('available_qty') or 0) r['product_production_in_transit'] = int(pinv.get('production_in_transit') or 0) r['product_purchase_in_transit'] = int(pinv.get('purchase_in_transit') or 0) return jsonify({"code": 0, "data": {"total": total, "rows": rows, "product_count": total}}) # ─── API: BOM供应商列表 ─── @products_bp.route('/api/bom/suppliers') @login_required def bom_suppliers(): """获取BOM中用到的所有物料供应商名称""" bom_supplier_expr = _bom_material_supplier_sql() rows = query(f""" SELECT DISTINCT {bom_supplier_expr} AS supplier_name FROM bom b LEFT JOIN products m ON b.material_code = m.product_code LEFT JOIN suppliers ms ON m.supplier_code = ms.supplier_code WHERE {bom_supplier_expr} <> '' ORDER BY supplier_name """) suppliers = [r['supplier_name'] for r in rows] return jsonify({"code": 0, "data": suppliers}) # ─── API: 产品新增 ─── @products_bp.route('/api/product/create', methods=['POST']) @login_required def product_create(): """新增产品""" if not operation_required('products', 'create') and not operation_required('products', 'add'): return jsonify({"code": 403, "msg": "没有新增产品权限"}) data = request.json or {} product_code = data.get('product_code', '').strip() product_name = data.get('product_name', '').strip() if not product_code or not product_name: return jsonify({"code": 400, "msg": "货品编号和货品名称不能为空"}) existing = query("SELECT id FROM products WHERE product_code=%s", (product_code,)) if existing: return jsonify({"code": 400, "msg": f"货品编号 {product_code} 已存在"}) try: query(""" INSERT INTO products (product_code, product_name, brand, product_type, lead_time, logistics_time, supplier_code, supplier_name_match, procurement_follower, procurement_planner, operator, platform, developer, created_at, import_date, custom_field1, custom_field2, custom_field3, custom_field4, custom_field5) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) """, ( product_code, product_name, data.get('brand', ''), data.get('product_type', '成品'), int(data.get('lead_time', 0)), min(int(data.get('logistics_time', 0)), 999), data.get('supplier_code', ''), data.get('supplier_name', ''), data.get('procurement_follower', ''), data.get('procurement_planner', ''), data.get('operator', ''), data.get('platform', ''), data.get('developer', ''), data.get('created_at', '').replace('T', ' ') if data.get('created_at') else None, datetime.now(), data.get('custom_field1', ''), data.get('custom_field2', ''), data.get('custom_field3', ''), data.get('custom_field4', ''), data.get('custom_field5', '') )) add_operation_log('产品创建', f'新建产品 {product_code} - {product_name}') return jsonify({"code": 0, "msg": "创建成功"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) # ─── API: 产品导出（CSV）─── @products_bp.route('/api/products/export') @login_required def products_export(): """导出产品数据为 CSV（支持选字段）""" product_type = request.args.get('product_type', '') fields_param = request.args.get('fields', '') from flask import Response from urllib.parse import quote import csv, io where = [] params = [] if product_type: if product_type == '非成品': where.append("p.product_type != '成品'") elif product_type == '非包材': where.append("p.product_type NOT IN ('包材','其他')") elif product_type == '非成品专用': where.append("p.product_type IN ('包材','其他')") else: where.append("p.product_type=%s") params.append(product_type) where_clause = " AND ".join(where) if where else "1=1" rows = query(f""" SELECT p.product_code, p.product_name, p.brand, p.product_type, p.specification, p.cost_price, p.lead_time, p.logistics_time, p.min_order_qty, p.supplier_code, COALESCE(s.supplier_name, p.supplier_name_match) AS supplier_name, p.procurement_follower, p.procurement_planner, p.operator, p.developer, p.platform, p.category1, p.category2, p.category3, p.created_at, p.import_date FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE {where_clause} ORDER BY p.id DESC """, params) # 字段映射：前端字段名 → 显示名 FIELD_MAP = [ ('product_code', '货品编号'), ('product_name', '货品名称'), ('product_type', '货品类型'), ('brand', '品牌'), ('supplier_code', '供应商编码'), ('supplier_name', '供应商名称'), ('lead_time', '货期(天)'), ('logistics_time', '物流时间(天)'), ('procurement_follower', '采购跟单'), ('procurement_planner', '采购计划'), ('platform', '平台'), ('operator', '运营'), ('developer', '开发专员'), ('created_at', '创建时间'), ('import_date', '导入日期'), ] # 解析选中的字段 if fields_param: selected_fields = [f.strip() for f in fields_param.split(',') if f.strip()] else: selected_fields = [f[0] for f in FIELD_MAP] # 按选中顺序过滤 headers = [] getters = [] for f, label in FIELD_MAP: if f in selected_fields: headers.append(label) getters.append(f) if not headers: return jsonify({"code": 400, "msg": "未选择导出字段"}) output = io.StringIO() writer = csv.writer(output) writer.writerow(headers) for r in rows: row_vals = [] for f in getters: val = r.get(f, '') if f == 'created_at' and val: val = val.strftime('%Y-%m-%d %H:%M:%S') if hasattr(val, 'strftime') else str(val)[:19] elif f == 'import_date' and val: val = val.strftime('%Y-%m-%d') if hasattr(val, 'strftime') else str(val)[:10] elif val is None: val = '' row_vals.append(val) writer.writerow(row_vals) name_part = '全部' if product_type == '成品': name_part = '成品' elif product_type in ('非包材',): name_part = '成品' elif product_type: name_part = '非成品' filename = f"{name_part}基础表_{datetime.now().strftime('%Y%m%d')}.csv" encoded = quote(filename) return Response( output.getvalue(), mimetype='text/csv; charset=utf-8-sig', headers={'Content-Disposition': f"attachment; filename*=UTF-8''{encoded}"} ) # ─── API: 产品导入模板下载 ─── @products_bp.route('/api/products/template') @login_required def products_template(): """下载导入模板 CSV（与当前字段一致） ?type=包材 时生成包材导入模板（含绑定成品编码、用量） """ from flask import Response, request from urllib.parse import quote import csv, io output = io.StringIO() writer = csv.writer(output) is_packaging = request.args.get('type', '') == '包材' if is_packaging: writer.writerow(['绑定成品编码', '货品编号', '货品名称', '用量', '产品类型', '成本价', '生产货期', '物流时间', '最低起订量', '供应商编号', '供应商名称']) writer.writerow(['DEMO-FG-001', 'BZ-DEMO001', '示例包材名称', '2', '包材', '5.00', '5', '2', '500', 'SUP001', '示例供应商']) filename = '包材导入模板.csv' else: writer.writerow(['货品编号', '货品名称', '品牌', '产品类型', '成本价', '生产货期', '物流时间', '最低起订量', '供应商编号', '供应商名称', '采购跟单', '采购计划', '运营专员', '开发专员', '归属平台', '创建时间']) writer.writerow(['DEMO001', '示例产品名称', '品牌A', '成品', '15.50', '7', '3', '100', 'SUP001', '示例供应商', '张三', '李四', '王五', '赵六', '天猫', '2026-07-15']) filename = '产品导入模板.csv' encoded = quote(filename) return Response( output.getvalue(), mimetype='text/csv; charset=utf-8-sig', headers={'Content-Disposition': f"attachment; filename*=UTF-8''{encoded}"} ) # ─── API: 产品导入（CSV上传）─── @products_bp.route('/api/products/import', methods=['POST']) @login_required def products_import(): """通过 CSV 导入产品""" if not operation_required('products', 'import'): return jsonify({"code": 403, "msg": "无操作权限"}) file = request.files.get('file') if not file: return jsonify({"code": 400, "msg": "请选择文件"}) import csv, io raw = file.read() # 尝试多种编码，兼容 Excel 导出的 GBK 等格式 for enc in ('utf-8-sig', 'gbk', 'gb2312', 'utf-8'): try: content = raw.decode(enc) break except (UnicodeDecodeError, LookupError): continue else: return jsonify({"code": 400, "msg": "无法识别文件编码，请确认 CSV 文件为 UTF-8 或 GBK 格式"}) reader = csv.DictReader(io.StringIO(content)) # 判断是否为包材导入（含 BOM 绑定列） has_bom_bind = '绑定成品编码' in (reader.fieldnames or []) field_map = { '货品编号': 'product_code', '货品名称': 'product_name', '品牌': 'brand', '产品类型': 'product_type', '成本价': 'cost_price', '生产货期': 'lead_time', '物流时间': 'logistics_time', '最低起订量': 'min_order_qty', '供应商编号': 'supplier_code', '供应商名称': 'supplier_name', '采购跟单': 'procurement_follower', '采购计划': 'procurement_planner', '运营专员': 'operator', '开发专员': 'developer', '归属平台': 'platform' } imported = 0 updated = 0 bom_count = 0 errors = [] # 已存在的货品编号按文件内容覆盖更新；空白单元格不改动原值（与供应商导入口径一致） def _as_text(v): return v def _as_float(v): try: return float(v) except (TypeError, ValueError): return 0 def _as_int(v): try: return int(float(v)) except (TypeError, ValueError): return 0 update_field_map = { '货品名称': ('product_name', _as_text), '品牌': ('brand', _as_text), '产品类型': ('product_type', _as_text), '成本价': ('cost_price', _as_float), '生产货期': ('lead_time', _as_int), '物流时间': ('logistics_time', _as_int), '最低起订量': ('min_order_qty', _as_int), '供应商编号': ('supplier_code', _as_text), '供应商名称': ('supplier_name_match', _as_text), '采购跟单': ('procurement_follower', _as_text), '采购计划': ('procurement_planner', _as_text), '运营专员': ('operator', _as_text), '开发专员': ('developer', _as_text), '归属平台': ('platform', _as_text), } try: for i, row in enumerate(reader, 2): pc = row.get('货品编号', '').strip() pn = row.get('货品名称', '').strip() if not pc: errors.append(f"第{i}行：货品编号为空，已跳过") continue existing = query("SELECT id FROM products WHERE product_code=%s", (pc,)) if existing: set_parts = [] set_params = [] for header, (col, conv) in update_field_map.items(): if header not in (reader.fieldnames or []): continue raw_val = (row.get(header) or '').strip() if raw_val == '': continue set_parts.append(f"{col}=%s") set_params.append(conv(raw_val)) if not set_parts: errors.append(f"第{i}行：货品编号 {pc} 已存在，但文件内无任何可更新内容，已跳过") continue set_params.append(pc) query(f"UPDATE products SET {', '.join(set_parts)} WHERE product_code=%s", set_params) updated += 1 else: if not pn: errors.append(f"第{i}行：货品编号 {pc} 不存在且未填写货品名称，无法新增，已跳过") continue query(""" INSERT INTO products (product_code, product_name, brand, product_type, cost_price, lead_time, logistics_time, min_order_qty, supplier_code, supplier_name_match, procurement_follower, procurement_planner, operator, platform, developer, created_at, import_date) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,CURDATE()) """, ( pc, pn, row.get('品牌', '').strip(), row.get('产品类型', '包材' if has_bom_bind else '成品').strip(), (lambda v: float(v) if v else 0)(row.get('成本价', '') or 0), (lambda v: int(float(v)) if v else 0)(row.get('生产货期', '') or 0), (lambda v: int(float(v)) if v else 0)(row.get('物流时间', '') or 0), (lambda v: int(float(v)) if v else 0)(row.get('最低起订量', '') or 0), row.get('供应商编号', '').strip(), row.get('供应商名称', '').strip(), row.get('采购跟单', '').strip(), row.get('采购计划', '').strip(), row.get('运营专员', '').strip(), row.get('归属平台', '').strip(), row.get('开发专员', '').strip(), row.get('创建时间', '').strip() or None )) imported += 1 # BOM 绑定：如果导入的是包材且有绑定成品编码 if has_bom_bind: bind_code = row.get('绑定成品编码', '').strip() qty = row.get('用量', '').strip() if bind_code: # 查成品是否存在（不存在也跳过，仅提示） prod_row = query("SELECT id, product_name FROM products WHERE product_code=%s", (bind_code,)) if not prod_row: errors.append(f"第{i}行：绑定的成品编码 {bind_code} 不存在，跳过 BOM 绑定") else: qty_val = float(qty) if qty else 1 # 查是否已有该 BOM 记录，有则更新，无则插入 existing_bom = query( "SELECT id FROM bom WHERE product_code=%s AND material_code=%s", (bind_code, pc)) # 获取包材成本价 mat_cost = float(row.get('成本价', 0)) if row.get('成本价') else 0 if existing_bom: query("UPDATE bom SET quantity=%s, material_cost_price=%s WHERE id=%s", (qty_val, mat_cost, existing_bom[0]['id'])) else: # 更新场景下文件可能未提供货品名称，回退到库中已有名称 mat_name = pn or (query("SELECT product_name FROM products WHERE product_code=%s", (pc,)) or [{}])[0].get('product_name', '') or '' query( "INSERT INTO bom (product_code, product_name, material_code, material_name, quantity, material_cost_price) VALUES (%s,%s,%s,%s,%s,%s)", (bind_code, prod_row[0]['product_name'], pc, mat_name, qty_val, mat_cost)) bom_count += 1 log_detail = f'CSV批量导入产品：新增{imported}条，更新{updated}条' if bom_count: log_detail += f'，自动创建/更新BOM绑定{bom_count}条' add_operation_log('产品导入', log_detail) msg = f"导入完成：新增 {imported} 条，更新 {updated} 条" if bom_count: msg += f"，自动绑定BOM {bom_count}条" if errors: msg += "。以下行已跳过：" + "; ".join(errors[:10]) if len(errors) > 10: msg += f"...（共{len(errors)}个错误）" return jsonify({"code": 0, "msg": msg}) except Exception as e: import traceback traceback.print_exc() return jsonify({"code": 500, "msg": f"导入失败：{str(e)}"}) # ─── API: BOM导入模板下载 ─── @products_bp.route('/api/bom/import_template') @login_required def bom_import_template(): """下载BOM导入模板CSV""" from flask import Response from urllib.parse import quote template_csv = '货品编号,货品名称,物料编号,物料名称,用量,供应商,成本价\n' encoded_filename = quote('物料清单导入模板.csv') return Response( template_csv, mimetype='text/csv', headers={'Content-Disposition': f"attachment; filename*=UTF-8''{encoded_filename}"} ) @products_bp.route('/api/bom/import', methods=['POST']) @login_required def bom_import(): """导入BOM数据（按产品覆盖，非全量替换）""" data = request.json items = data.get('items', []) if not items: return jsonify({"code": 400, "msg": "数据为空"}) try: # 收集要更新的产品编码，只删除这些产品的旧BOM数据 product_codes = set(item.get('product_code', '') for item in items if item.get('product_code')) if product_codes: placeholders = ','.join(['%s'] * len(product_codes)) query(f"DELETE FROM bom WHERE product_code IN ({placeholders})", tuple(product_codes)) for item in items: query( "INSERT INTO bom (product_code, product_name, material_code, material_name, quantity, supplier, material_cost_price) VALUES (%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE product_name=VALUES(product_name), material_name=VALUES(material_name), quantity=VALUES(quantity), supplier=VALUES(supplier), material_cost_price=VALUES(material_cost_price)", (item.get('product_code', ''), item.get('product_name', ''), item.get('material_code', ''), item.get('material_name', ''), float(item.get('quantity', 1)), _normalize_bom_supplier(item.get('supplier', '')), float(item.get('cost_price', 0))) ) add_operation_log('BOM导入', f'导入BOM，共{len(items)}条') return jsonify({"code": 0, "msg": f"导入成功，共{len(items)}条"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) # ─── API: 供应商管理 ─── @products_bp.route('/api/suppliers/list') @login_required def suppliers_list(): page = int(request.args.get('page', 1)) size = int(request.args.get('size', 50)) offset = (page - 1) * size search = request.args.get('search', '') code = request.args.get('code', '') name = request.args.get('name', '') where = [] params = [] if search: where.append("(supplier_name LIKE %s OR short_name LIKE %s OR contract_short_name LIKE %s OR supplier_code LIKE %s OR production_type LIKE %s)") params.extend([f"%{search}%", f"%{search}%", f"%{search}%", f"%{search}%", f"%{search}%"]) if code: where.append("supplier_code LIKE %s") params.append(f"%{code}%") if name: where.append("supplier_name LIKE %s") params.append(f"%{name}%") where_clause = " AND ".join(where) if where else "1=1" total = query(f"SELECT COUNT(*) AS cnt FROM suppliers WHERE {where_clause}", params)[0]['cnt'] rows = query(f"SELECT * FROM suppliers WHERE {where_clause} ORDER BY supplier_code ASC LIMIT {size} OFFSET {offset}", params) # 日期转字符串，避免Flask序列化为 "Wed, 18 Mar 2026 00:00:00 GMT" 格式 for r in rows: if r.get('review_date') and hasattr(r['review_date'], 'isoformat'): r['review_date'] = r['review_date'].isoformat() return jsonify({"code": 0, "data": {"total": total, "page": page, "size": size, "rows": rows}}) @products_bp.route('/api/supplier/update', methods=['POST']) @login_required def supplier_update(): if not operation_required('suppliers', 'edit'): return jsonify({"code": 403, "msg": "无操作权限"}) data = request.json code = data.get('supplier_code', '') fields = data.get('fields', {}) if not code or not fields: return jsonify({"code": 400, "msg": "参数不完整"}) allowed = ['supplier_name', 'production_type', 'short_name', 'contract_short_name', 'supplier_group', 'procurement_follower', 'procurement_planner', 'reviewer', 'review_date', 'data_status', 'disabled_status'] set_parts = [] params = [] for field, value in fields.items(): if field in allowed: set_parts.append(f"{field}=%s") params.append(value if value != '' else None) if not set_parts: return jsonify({"code": 400, "msg": "没有可更新的字段"}) set_clause = ", ".join(set_parts) params.append(code) try: query(f"UPDATE suppliers SET {set_clause} WHERE supplier_code=%s", params) return jsonify({"code": 0, "msg": "更新成功"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/brands/all') @login_required def brands_all(): """品牌下拉/模糊查询用：品牌信息管理维护的品牌 + 产品表中已存在的品牌""" rows = query("""SELECT TRIM(brand) AS brand FROM products WHERE deleted_at IS NULL AND brand IS NOT NULL AND TRIM(brand) <> '' GROUP BY TRIM(brand)""") saved = query("SELECT brand FROM brand_information") names = {r['brand'] for r in rows} | {(r.get('brand') or '').strip() for r in saved} names.discard('') return jsonify({"code": 0, "data": sorted(names, key=lambda x: x.casefold())}) @products_bp.route('/api/brands/overseas') @login_required def brands_overseas(): """新增产品弹窗用：海外品品牌清单（来源：品牌管理的品牌分类）""" rows = query("SELECT brand FROM brand_information WHERE category='海外' ORDER BY brand") return jsonify({"code": 0, "data": [r['brand'] for r in rows]}) @products_bp.route('/api/supplier/bound_fields') @login_required def supplier_bound_fields(): """新增产品弹窗用：按供应商解析跟单小组与采购计划。 优先取供应商管理里该供应商绑定的值；管理里未维护时回退到该供应商已有产品的对应值。 """ code = (request.args.get('supplier_code') or '').strip() name = (request.args.get('supplier_name') or '').strip() follower = '' planner = '' follower_source = '' planner_source = '' supplier = None if code: rows = query("SELECT supplier_name, procurement_follower, procurement_planner FROM suppliers WHERE supplier_code=%s", (code,)) supplier = rows[0] if rows else None if not supplier and name: rows = query("SELECT supplier_name, procurement_follower, procurement_planner FROM suppliers WHERE supplier_name=%s", (name,)) supplier = rows[0] if rows else None if not supplier and name: rows = query("SELECT supplier_name, procurement_follower, procurement_planner FROM suppliers WHERE supplier_name LIKE %s ORDER BY id LIMIT 1", (f"%{name}%",)) supplier = rows[0] if rows else None if supplier: bound_follower = (supplier.get('procurement_follower') or '').strip() bound_planner = (supplier.get('procurement_planner') or '').strip() if bound_follower: follower = bound_follower follower_source = 'supplier' if bound_planner: planner = bound_planner planner_source = 'supplier' # 回退查询要用供应商全称，避免传入片段名时匹配不到产品 name = (supplier.get('supplier_name') or '').strip() or name if name and (not follower or not planner): rows = query(""" SELECT procurement_follower, procurement_planner, COUNT(*) AS cnt FROM products WHERE supplier_name_match=%s AND ((procurement_follower IS NOT NULL AND procurement_follower<>'') OR (procurement_planner IS NOT NULL AND procurement_planner<>'')) GROUP BY procurement_follower, procurement_planner ORDER BY cnt DESC LIMIT 50 """, (name,)) # 分别取出现次数最多的非空值 follower_counts = {} planner_counts = {} for r in rows: f = (r.get('procurement_follower') or '').strip() p = (r.get('procurement_planner') or '').strip() if f: follower_counts[f] = follower_counts.get(f, 0) + int(r.get('cnt') or 0) if p: planner_counts[p] = planner_counts.get(p, 0) + int(r.get('cnt') or 0) if not follower and follower_counts: follower = max(follower_counts, key=follower_counts.get) follower_source = 'product_fallback' if not planner and planner_counts: planner = max(planner_counts, key=planner_counts.get) planner_source = 'product_fallback' return jsonify({"code": 0, "data": { "follower": follower, "follower_source": follower_source, "planner": planner, "planner_source": planner_source, "supplier_name": name, }}) @products_bp.route('/api/supplier/follower_lookup') @login_required def supplier_follower_lookup(): """兼容旧调用：仅返回跟单小组""" payload = supplier_bound_fields().get_json()['data'] return jsonify({"code": 0, "data": { "follower": payload['follower'], "source": payload['follower_source'], "supplier_name": payload['supplier_name'], }}) @products_bp.route('/api/supplier/delete', methods=['POST']) @login_required def supplier_delete(): """删除供应商""" if not operation_required('suppliers', 'delete'): return jsonify({"code": 403, "msg": "无操作权限"}) data = request.json or {} code = data.get('supplier_code', '').strip() if not code: return jsonify({"code": 400, "msg": "缺少供应商编码"}) try: query("DELETE FROM suppliers WHERE supplier_code=%s", (code,)) return jsonify({"code": 0, "msg": f"供应商 [{code}] 已删除"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/suppliers/export') @login_required def suppliers_export(): """导出供应商数据为xlsx""" from flask import send_file import openpyxl from openpyxl.styles import Font, Alignment, Border, Side, PatternFill import tempfile search = request.args.get('search', '') code = request.args.get('code', '') name = request.args.get('name', '') where = [] params = [] if search: where.append("(supplier_name LIKE %s OR supplier_code LIKE %s OR production_type LIKE %s)") params.extend([f"%{search}%", f"%{search}%", f"%{search}%"]) if code: where.append("supplier_code LIKE %s") params.append(f"%{code}%") if name: where.append("supplier_name LIKE %s") params.append(f"%{name}%") where_clause = " AND ".join(where) if where else "1=1" rows = query(f"SELECT * FROM suppliers WHERE {where_clause} ORDER BY supplier_code ASC", params) wb = openpyxl.Workbook() ws = wb.active ws.title = "供应商列表" # 样式 header_font = Font(name='微软雅黑', bold=True, size=11, color='FFFFFF') header_fill = PatternFill(start_color='1890FF', end_color='1890FF', fill_type='solid') header_align = Alignment(horizontal='center', vertical='center') thin_border = Border( left=Side(style='thin'), right=Side(style='thin'), top=Side(style='thin'), bottom=Side(style='thin') ) cell_align = Alignment(horizontal='center', vertical='center') headers = ['序号', '供应商编码', '供应商名称', '做货类型', '简称', '合同简称', '供应商分组', '审核人', '审核日期', '数据状态', '禁用状态'] for col_idx, h in enumerate(headers, 1): cell = ws.cell(row=1, column=col_idx, value=h) cell.font = header_font cell.fill = header_fill cell.alignment = header_align cell.border = thin_border for i, r in enumerate(rows, 1): row_data = [ i, r.get('supplier_code', ''), r.get('supplier_name', ''), r.get('production_type', '') or '-', r.get('short_name', '') or '-', r.get('contract_short_name', '') or '-', r.get('supplier_group', '') or '-', r.get('reviewer', '') or '-', (r.get('review_date') or '-') if r.get('review_date') else '-', r.get('data_status', '') or '-', r.get('disabled_status', '') or '-', ] for col_idx, val in enumerate(row_data, 1): cell = ws.cell(row=i + 1, column=col_idx, value=str(val)) cell.alignment = cell_align cell.border = thin_border # 调整列宽 col_widths = [6, 16, 36, 20, 12, 12, 16, 10, 14, 12, 12] for col_idx, w in enumerate(col_widths, 1): ws.column_dimensions[openpyxl.utils.get_column_letter(col_idx)].width = w # 冻结首行 ws.freeze_panes = 'A2' tmp = tempfile.NamedTemporaryFile(suffix='.xlsx', delete=False) wb.save(tmp.name) tmp.close() return send_file( tmp.name, as_attachment=True, download_name='供应商列表.xlsx', mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' ) @products_bp.route('/api/suppliers/import_template') @login_required def suppliers_import_template(): """下载供应商导入模板""" from flask import send_file import openpyxl import tempfile wb = openpyxl.Workbook() ws = wb.active ws.title = "供应商导入模板" headers = ['供应商名称', '简称', '合同简称', '跟单小组', '采购计划'] for col_idx, h in enumerate(headers, 1): cell = ws.cell(row=1, column=col_idx, value=h) cell.font = openpyxl.styles.Font(name='微软雅黑', bold=True, size=11, color='FFFFFF') cell.fill = openpyxl.styles.PatternFill(start_color='1890FF', end_color='1890FF', fill_type='solid') cell.alignment = openpyxl.styles.Alignment(horizontal='center', vertical='center') cell.border = openpyxl.styles.Border( left=openpyxl.styles.Side(style='thin'), right=openpyxl.styles.Side(style='thin'), top=openpyxl.styles.Side(style='thin'), bottom=openpyxl.styles.Side(style='thin')) # 添加示例行 examples = [ ['广州市靓鑫精细化工有限公司', '靓鑫', 'LX', '跟单一组', '陈瑜'], ['广州市巧能塑料制品有限公司', '巧能', 'QN', '跟单二组', '李艳艳'], ] for i, ex in enumerate(examples, 2): for col_idx, val in enumerate(ex, 1): cell = ws.cell(row=i, column=col_idx, value=val) cell.alignment = openpyxl.styles.Alignment(horizontal='center', vertical='center') cell.border = openpyxl.styles.Border( left=openpyxl.styles.Side(style='thin'), right=openpyxl.styles.Side(style='thin'), top=openpyxl.styles.Side(style='thin'), bottom=openpyxl.styles.Side(style='thin')) ws.column_dimensions['A'].width = 36 ws.column_dimensions['B'].width = 14 ws.column_dimensions['C'].width = 14 ws.column_dimensions['D'].width = 16 ws.column_dimensions['E'].width = 16 ws.freeze_panes = 'A2' tmp = tempfile.NamedTemporaryFile(suffix='.xlsx', delete=False) wb.save(tmp.name) tmp.close() return send_file( tmp.name, as_attachment=True, download_name='供应商导入模板.xlsx', mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' ) @products_bp.route('/api/suppliers/import', methods=['POST']) @login_required def suppliers_import(): """导入xlsx更新供应商简称和合同简称""" import openpyxl if 'file' not in request.files: return jsonify({"code": 400, "msg": "未上传文件"}) f = request.files['file'] if not f.filename: return jsonify({"code": 400, "msg": "文件名为空"}) try: wb = openpyxl.load_workbook(f) ws = wb.active except Exception as e: return jsonify({"code": 400, "msg": f"无法读取Excel文件：{str(e)}"}) if ws.max_row < 2: return jsonify({"code": 400, "msg": "Excel文件中没有数据"}) # 读取表头确定列位置 headers = {} for col_idx in range(1, ws.max_column + 1): val = str(ws.cell(row=1, column=col_idx).value or '').strip() headers[val] = col_idx name_col = headers.get('供应商名称') or headers.get('供应商') short_col = headers.get('简称') or headers.get('供应商简称') contract_col = headers.get('合同简称') follower_col = headers.get('跟单小组') or headers.get('采购跟单') planner_col = headers.get('采购计划') if not name_col: return jsonify({"code": 400, "msg": "未找到供应商名称列，表头需包含'供应商名称'"}) if not short_col and not contract_col and not follower_col and not planner_col: return jsonify({"code": 400, "msg": "未找到简称、合同简称、跟单小组或采购计划列"}) # 全量加载供应商 all_suppliers = query("SELECT id, supplier_code, supplier_name, short_name, contract_short_name FROM suppliers") name_map = {} for s in all_suppliers: key = (s['supplier_name'] or '').strip() name_map[key] = s # 预校验：合同简称必须全大写 if contract_col: for row_idx in range(2, ws.max_row + 1): v = ws.cell(row=row_idx, column=contract_col).value if v is not None: val = str(v).strip() if val and val != val.upper(): supplier_name = str(ws.cell(row=row_idx, column=name_col).value or '').strip() return jsonify({"code": 400, "msg": f"合同简称必须全大写（行{row_idx}，供应商「{supplier_name}」的合同简称为「{val}」）"}) updated = [] not_found = [] skipped = [] for row_idx in range(2, ws.max_row + 1): supplier_name = str(ws.cell(row=row_idx, column=name_col).value or '').strip() if not supplier_name: continue short_val = None contract_val = None follower_val = None planner_val = None if short_col: v = ws.cell(row=row_idx, column=short_col).value short_val = str(v).strip() if v is not None else None if contract_col: v = ws.cell(row=row_idx, column=contract_col).value contract_val = str(v).strip() if v is not None else None if follower_col: v = ws.cell(row=row_idx, column=follower_col).value follower_val = str(v).strip() if v is not None else None if planner_col: v = ws.cell(row=row_idx, column=planner_col).value planner_val = str(v).strip() if v is not None else None if not short_val and not contract_val and not follower_val and not planner_val: skipped.append(supplier_name) continue match = name_map.get(supplier_name) if not match: not_found.append(supplier_name) continue # 更新 update_fields = [] update_params = [] if short_val: update_fields.append("short_name=%s") update_params.append(short_val) if contract_val: update_fields.append("contract_short_name=%s") update_params.append(contract_val) if follower_val: update_fields.append("procurement_follower=%s") update_params.append(follower_val) if planner_val: update_fields.append("procurement_planner=%s") update_params.append(planner_val) if update_fields: update_params.append(match['id']) query(f"UPDATE suppliers SET {', '.join(update_fields)} WHERE id=%s", update_params) updated.append(supplier_name) return jsonify({"code": 0, "data": { "updated": len(updated), "not_found": len(not_found), "skipped": len(skipped), "not_found_list": not_found[:20], "updated_list": updated, "message": f"成功更新 {len(updated)} 条" + (f"，{len(not_found)} 条未匹配" if not_found else "") + (f"，{len(skipped)} 条跳过（无数据）" if skipped else "") }}) @products_bp.route('/api/supplier/create', methods=['POST']) @login_required def supplier_create(): """新增供应商""" if not operation_required('suppliers', 'create'): return jsonify({"code": 403, "msg": "无操作权限"}) data = request.json or {} code = (data.get('supplier_code') or '').strip() name = (data.get('supplier_name') or '').strip() if not code or not name: return jsonify({"code": 400, "msg": "供应商编码和名称不能为空"}) exist = query("SELECT COUNT(*) AS c FROM suppliers WHERE supplier_code=%s", (code,)) if exist and exist[0]['c'] > 0: return jsonify({"code": 400, "msg": f"供应商编码 [{code}] 已存在"}) try: query("""INSERT INTO suppliers (supplier_code, supplier_name, production_type, short_name, contract_short_name, supplier_group, procurement_follower, procurement_planner, reviewer, review_date, data_status, disabled_status) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""", (code, name, data.get('production_type', '') or None, data.get('short_name', '') or None, data.get('contract_short_name', '') or None, data.get('supplier_group', '') or None, data.get('procurement_follower', '') or None, data.get('procurement_planner', '') or None, data.get('reviewer', '') or None, data.get('review_date', '') or None, data.get('data_status', '') or None, data.get('disabled_status', '') or None)) try: add_operation_log('新增供应商', f'新增供应商 {code} - {name}') except: pass return jsonify({"code": 0, "msg": "新增成功"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/product/delete', methods=['POST']) @login_required def product_delete(): """软删除产品（移入回收站）""" if not operation_required('products', 'delete'): return jsonify({"code": 403, "msg": "无操作权限"}) data = request.json code = data.get('product_code', '') if not code: return jsonify({"code": 400, "msg": "参数错误"}) try: query("UPDATE products SET deleted_at=NOW() WHERE product_code=%s", (code,)) # 级联删除：物料清单 # 先查出成品绑定的包材（需在 BOM DELETE 之前查询） materials = query("SELECT material_code FROM bom WHERE product_code=%s", (code,)) query("DELETE FROM bom WHERE product_code=%s", (code,)) query("DELETE FROM bom WHERE material_code=%s", (code,)) # 级联删除：成品绑定的包材产品（软删除） for m in materials: query("UPDATE products SET deleted_at=NOW() WHERE product_code=%s", (m['material_code'],)) # 级联删除：计划预测隐藏记录 query("DELETE FROM plan_hidden WHERE product_code=%s", (code,)) return jsonify({"code": 0, "msg": "已移入回收站"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/product/recycle', methods=['POST']) @login_required def product_recycle(): """将成品及其BOM包材移入回收站，并删除该成品的BOM。""" if not operation_required('products', 'delete'): return jsonify({"code": 403, "msg": "无操作权限"}) data = request.json or {} code = (data.get('product_code') or '').strip() if not code: return jsonify({"code": 400, "msg": "缺少货品编号"}) conn = get_db() try: conn.begin() with conn.cursor() as cursor: cursor.execute( "SELECT product_name FROM products WHERE product_code=%s AND deleted_at IS NULL", (code,) ) product = cursor.fetchone() if not product: conn.rollback() return jsonify({"code": 404, "msg": "未找到可回收的产品"}) cursor.execute( "SELECT DISTINCT b.material_code FROM bom b " "JOIN products m ON m.product_code=b.material_code " "WHERE b.product_code=%s AND b.material_code IS NOT NULL " "AND b.material_code != '' AND m.product_type IN ('包材', '其他')", (code,) ) material_codes = [row['material_code'] for row in cursor.fetchall()] cursor.execute("UPDATE products SET deleted_at=NOW() WHERE product_code=%s", (code,)) if material_codes: placeholders = ','.join(['%s'] * len(material_codes)) cursor.execute( f"UPDATE products SET deleted_at=NOW() WHERE product_code IN ({placeholders})", material_codes ) cursor.execute("DELETE FROM bom WHERE product_code=%s", (code,)) conn.commit() # 计划预测通过 deleted_at 过滤；保留用户原有的手工隐藏状态。 try: add_operation_log( '产品回收', f"回收产品 {code} - {product.get('product_name', '')}，" f"同步回收包材 {len(material_codes)} 个并删除BOM" ) except Exception: pass return jsonify({ "code": 0, "msg": f"已回收成品及{len(material_codes)}个关联包材，并删除该成品BOM", "data": {"recycled_materials": len(material_codes)} }) except Exception as e: conn.rollback() return jsonify({"code": 500, "msg": str(e)}) finally: conn.close() @products_bp.route('/api/products/recycle_bin', methods=['GET']) @login_required def products_recycle_bin(): """获取回收站产品列表""" 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', '') unified_search = request.args.get('search', '') supplier_filter = request.args.get('supplier', '') brand_filter = request.args.get('brand', '') product_type = request.args.get('product_type', '') where = ["p.deleted_at IS NOT NULL"] 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 supplier_filter: where.append("(p.supplier_name_match LIKE %s OR s.supplier_name LIKE %s)") params.extend([f"%{supplier_filter}%", f"%{supplier_filter}%"]) if brand_filter: where.append("TRIM(p.brand)=%s") params.append(brand_filter) if product_type: where.append("p.product_type=%s") params.append(product_type) where_clause = " AND ".join(where) total = query(f"""SELECT COUNT(*) AS cnt FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE {where_clause}""", params)[0]['cnt'] rows = query(f""" SELECT p.id, p.goods_code, p.product_code, p.product_name, p.brand, p.cost_price, p.product_type, p.specification, p.lead_time, p.logistics_time, p.supplier_code, p.supplier_name_match, p.procurement_follower, p.procurement_planner, p.platform, p.operator, p.developer, p.category1, p.category2, p.category3, p.created_at, p.import_date, p.deleted_at, p.custom_field1, p.custom_field2, p.custom_field3, p.custom_field4, p.custom_field5 FROM products p LEFT JOIN suppliers s ON p.supplier_code = s.supplier_code WHERE {where_clause} ORDER BY p.deleted_at DESC LIMIT {size} OFFSET {offset} """, params) for row in rows: if row.get('logistics_time') is None: row['logistics_time'] = 0 if row.get('lead_time') is None: row['lead_time'] = 0 if 'created_at' in row and row['created_at']: row['created_at'] = row['created_at'].strftime('%Y-%m-%d %H:%M:%S') if hasattr(row['created_at'], 'strftime') else str(row['created_at'])[:19] if 'import_date' in row and row['import_date']: row['import_date'] = row['import_date'].strftime('%Y-%m-%d') if hasattr(row['import_date'], 'strftime') else str(row['import_date'])[:10] if row.get('deleted_at'): row['deleted_at'] = row['deleted_at'].strftime('%Y-%m-%d %H:%M:%S') if hasattr(row['deleted_at'], 'strftime') else str(row['deleted_at'])[:19] if not row.get('product_type'): row['product_type'] = get_product_type_cached(row.get('product_name', ''), row.get('brand', ''), row.get('product_type')) return jsonify({ "code": 0, "data": { "total": total, "page": page, "size": size, "rows": rows } }) @products_bp.route('/api/product/restore', methods=['POST']) @login_required def product_restore(): """从回收站恢复产品""" if not operation_required('products', 'edit'): return jsonify({"code": 403, "msg": "无操作权限"}) data = request.json code = data.get('product_code', '') if not code: return jsonify({"code": 400, "msg": "参数错误"}) try: query("UPDATE products SET deleted_at=NULL WHERE product_code=%s", (code,)) add_operation_log('产品恢复', f'恢复产品 {code}') return jsonify({"code": 0, "msg": "已恢复到成品基础信息表"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/product/batch_restore', methods=['POST']) @login_required def product_batch_restore(): """批量恢复回收站产品。""" if not operation_required('products', 'edit'): return jsonify({"code": 403, "msg": "无操作权限"}) data = request.json or {} codes = list(dict.fromkeys(str(code).strip() for code in data.get('product_codes', []) if str(code).strip())) if not codes: return jsonify({"code": 400, "msg": "请选择需要恢复的产品"}) if len(codes) > 500: return jsonify({"code": 400, "msg": "单次最多恢复500个产品"}) try: placeholders = ','.join(['%s'] * len(codes)) rows = query( f"SELECT product_code FROM products WHERE deleted_at IS NOT NULL AND product_code IN ({placeholders})", codes ) valid_codes = [row['product_code'] for row in rows] if not valid_codes: return jsonify({"code": 404, "msg": "所选产品均不在回收站中"}) valid_placeholders = ','.join(['%s'] * len(valid_codes)) query( f"UPDATE products SET deleted_at=NULL WHERE deleted_at IS NOT NULL AND product_code IN ({valid_placeholders})", valid_codes ) add_operation_log('产品批量恢复', f"批量恢复 {len(valid_codes)} 个产品：{','.join(valid_codes)}") return jsonify({"code": 0, "msg": f"已恢复{len(valid_codes)}个产品", "data": {"restored": len(valid_codes)}}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/bom/update_materials', methods=['POST']) @login_required def bom_update_materials(): if not _check_product_edit_perm(): return jsonify({"code": 403, "msg": "仅管理员可操作"}) data = request.json materials = data.get('materials', []) if not materials: return jsonify({"code": 400, "msg": "参数不完整"}) allowed = ['material_name'] try: for m in materials: code = m.get('material_code', '') fields = m.get('fields', {}) if not code: continue set_parts = [] params = [] for field, value in fields.items(): if field in allowed: set_parts.append(f"{field}=%s") params.append(value if value != '' else None) if set_parts: set_clause = ", ".join(set_parts) params.append(code) query(f"UPDATE bom SET {set_clause} WHERE material_code=%s", params) # Also update the main product info in products table main_product_code = data.get('product_code', '') main_fields = data.get('main_fields', {}) if main_product_code and main_fields: product_allowed = ['product_name', 'cost_price', 'supplier_name'] set_parts = [] params = [] for field, value in main_fields.items(): if field in product_allowed: set_parts.append(f"{field}=%s") params.append(value if value != '' else None) if set_parts: set_clause = ", ".join(set_parts) params.append(main_product_code) query(f"UPDATE products SET {set_clause} WHERE product_code=%s", params) return jsonify({"code": 0, "msg": "更新成功"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/bom/create', methods=['POST']) @login_required def bom_create(): """新增BOM条目""" if not _check_product_edit_perm(): return jsonify({"code": 403, "msg": "仅管理员可操作"}) data = request.json or {} product_code = data.get('product_code', '').strip() material_code = data.get('material_code', '').strip() material_name = data.get('material_name', '').strip() quantity = float(data.get('quantity', 1)) if not product_code or not material_code: return jsonify({"code": 400, "msg": "货品编号和物料编号不能为空"}) existing = query("SELECT id FROM bom WHERE product_code=%s AND material_code=%s", (product_code, material_code)) if existing: return jsonify({"code": 400, "msg": "该BOM条目已存在"}) try: prod = query("SELECT product_name FROM products WHERE product_code=%s", (product_code,)) product_name = prod[0]['product_name'] if prod else product_code mat = query("SELECT product_name FROM products WHERE product_code=%s", (material_code,)) if not material_name and mat: material_name = mat[0]['product_name'] query("INSERT INTO bom (product_code, product_name, material_code, material_name, quantity) VALUES (%s,%s,%s,%s,%s)", (product_code, product_name, material_code, material_name, quantity)) return jsonify({"code": 0, "msg": "创建成功"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/bom/export') @login_required def bom_export(): """导出产品BOM状态（含有无BOM标记）""" rows = query(""" SELECT t.product_code, t.product_name, t.has_bom, p.product_type, p.supplier_name_match FROM ( SELECT b.product_code, MAX(b.product_name) as product_name, CASE WHEN SUM(CASE WHEN b.material_code IS NOT NULL AND b.material_code != '' THEN 1 ELSE 0 END) > 0 THEN '有' ELSE '无' END AS has_bom FROM bom b GROUP BY b.product_code ) t LEFT JOIN products p ON t.product_code = p.product_code ORDER BY t.product_code """) import csv, io output = io.StringIO() writer = csv.writer(output) writer.writerow(['货品编号', '货品名称', '有无BOM', '类型', '供应商']) for r in rows: writer.writerow([ r['product_code'], r['product_name'], r['has_bom'], r['product_type'] or '', r['supplier_name_match'] or '' ]) csv_data = output.getvalue() output.close() from flask import Response return Response( csv_data, mimetype='text/csv', headers={'Content-Disposition': 'attachment; filename=bom_export.csv'} ) @products_bp.route('/api/bom/delete', methods=['POST']) @login_required def bom_delete(): """删除成品BOM数据（含所有物料行）""" if not _check_product_edit_perm(): return jsonify({"code": 403, "msg": "仅管理员可操作"}) data = request.json or {} product_code = data.get('product_code', '').strip() if not product_code: return jsonify({"code": 400, "msg": "缺少成品编号"}) try: query("DELETE FROM bom WHERE product_code=%s", (product_code,)) return jsonify({"code": 0, "msg": f"已删除 [{product_code}] 的物料清单"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/bom/create_with_materials', methods=['POST']) @login_required def bom_create_with_materials(): """新建成品BOM并自动创建包材产品""" if not _check_product_edit_perm(): return jsonify({"code": 403, "msg": "仅管理员可操作"}) data = request.json or {} product_code = data.get('product_code', '').strip() materials = data.get('materials', []) if not product_code: return jsonify({"code": 400, "msg": "缺少成品编号"}) created = 0 total_cost = 0 # 自动创建成品产品（如果不存在） existing_product = query("SELECT id, product_name, product_type FROM products WHERE product_code=%s", (product_code,)) if not existing_product: prod_supplier = data.get('product_supplier', '').strip() query("""INSERT INTO products (product_code, product_name, brand, product_type, supplier_name_match, import_date) VALUES (%s,%s,'','成品',%s,NOW())""", (product_code, data.get('product_name', product_code), prod_supplier)) product_name = data.get('product_name', product_code) else: product_name = existing_product[0]['product_name'] or product_code for m in materials: mat_code = m.get('material_code', '').strip() mat_name = m.get('material_name', '').strip() qty = float(m.get('quantity', 1)) cost_price = float(m.get('cost_price', 0)) if m.get('cost_price') else 0 sup_code = m.get('supplier_code', '').strip() sup_name = _normalize_bom_supplier(m.get('supplier_name', '')) mat_type = m.get('product_type', '包材').strip() or '包材' if not mat_code: continue # 累加成本（成本价 × 用量） total_cost += qty * cost_price # 自动创建/更新包材产品 existing = query("SELECT id FROM products WHERE product_code=%s", (mat_code,)) if not existing: query("""INSERT INTO products (product_code, product_name, brand, product_type, supplier_code, supplier_name_match, import_date) VALUES (%s,%s,'',%s,%s,%s,NOW())""", (mat_code, mat_name or mat_code, mat_type, sup_code, sup_name)) else: # 已有物料更新供应商和名称 update_parts = [] update_params = [] if sup_name: update_parts.append("supplier_name_match=%s") update_params.append(sup_name) if sup_code: update_parts.append("supplier_code=%s") update_params.append(sup_code) if mat_name: update_parts.append("product_name=%s") update_params.append(mat_name) if mat_type: update_parts.append("product_type=%s") update_params.append(mat_type) if update_parts: update_params.append(mat_code) query(f"UPDATE products SET {', '.join(update_parts)} WHERE product_code=%s", update_params) # 新增/更新 BOM（含成本价） bom_exists = query("SELECT id FROM bom WHERE product_code=%s AND material_code=%s", (product_code, mat_code)) if bom_exists: query("UPDATE bom SET quantity=%s, material_name=%s, supplier=%s, material_cost_price=%s WHERE product_code=%s AND material_code=%s", (qty, mat_name, sup_name, cost_price, product_code, mat_code)) else: query("""INSERT INTO bom (product_code, product_name, material_code, material_name, quantity, supplier, material_cost_price) VALUES (%s,%s,%s,%s,%s,%s,%s)""", (product_code, product_name, mat_code, mat_name, qty, sup_name, cost_price)) created += 1 # 更新成品成本价 = 所有物料成本之和 if created > 0: query("UPDATE products SET cost_price=%s WHERE product_code=%s", (total_cost, product_code)) return jsonify({"code": 0, "msg": f"绑定完成，共处理{created}条BOM，成品成本价：¥{total_cost:.2f}"}) @products_bp.route('/api/product/sync_bom', methods=['POST']) @login_required def product_sync_bom(): """将成品BOM包材数据同步到物料清单管理（bom表）""" data = request.json or {} product_code = data.get('product_code', '').strip() if not product_code: return jsonify({"code": 400, "msg": "缺少成品编号"}) # 获取当前产品已有的BOM数据 bom_materials = query(""" SELECT material_code, material_name, quantity AS bom_qty FROM bom WHERE product_code=%s """, (product_code,)) if not bom_materials: return jsonify({"code": 0, "msg": "暂无包材数据可同步", "synced": 0}) prod = query("SELECT product_name FROM products WHERE product_code=%s", (product_code,)) product_name = prod[0]['product_name'] if prod else product_code synced = 0 for m in bom_materials: mat_code = m['material_code'] mat_name = m['material_name'] or mat_code qty = float(m['bom_qty']) if m['bom_qty'] else 1 # 确保物料已作为包材产品存在 existing = query("SELECT id FROM products WHERE product_code=%s", (mat_code,)) if not existing: query("""INSERT INTO products (product_code, product_name, brand, product_type, import_date) VALUES (%s,%s,'','包材',NOW())""", (mat_code, mat_name)) # 确保BOM条目存在 bom_exists = query("SELECT id FROM bom WHERE product_code=%s AND material_code=%s", (product_code, mat_code)) if bom_exists: query("UPDATE bom SET quantity=%s, material_name=%s WHERE product_code=%s AND material_code=%s", (qty, mat_name, product_code, mat_code)) else: query("""INSERT INTO bom (product_code, product_name, material_code, material_name, quantity) VALUES (%s,%s,%s,%s,%s)""", (product_code, product_name, mat_code, mat_name, qty)) synced += 1 return jsonify({"code": 0, "msg": f"同步完成，更新{len(bom_materials)}条，新增{synced}条", "synced": synced}) @products_bp.route('/api/products/match_supplier', methods=['POST']) @login_required def products_match_supplier(): """批量匹配包材产品的供应商名称，补全全称和编码""" if session.get("role") not in ("admin", "super_admin"): return jsonify({"code": 403, "msg": "仅管理员可操作"}) product_type = request.json.get('product_type', '') if request.json else '' if not product_type: return jsonify({"code": 400, "msg": "缺少产品类型"}) # 获取所有供应商 suppliers = query("SELECT supplier_code, supplier_name FROM suppliers") if not suppliers: return jsonify({"code": 0, "msg": "供应商列表为空", "matched": 0}) # 获取指定类型的产品（如果supplier_name_match与supplier对不上则更新） products = query(""" SELECT product_code, supplier_code, supplier_name_match FROM products WHERE product_type=%s AND deleted_at IS NULL """, (product_type,)) matched = 0 for p in products: pname = (p.get('supplier_name_match') or '').strip() if not pname: continue best = None # 模糊匹配供应商名称 for s in suppliers: sname = (s['supplier_name'] or '').strip() if not sname: continue # 如果产品供应商名包含在供应商全称中，或供应商全称包含产品供应商名 if pname in sname or sname in pname: best = s break if best: # 如果已是最佳匹配则跳过 if p.get('supplier_code') == best['supplier_code'] and p.get('supplier_name_match') == best['supplier_name']: continue query("UPDATE products SET supplier_code=%s, supplier_name_match=%s WHERE product_code=%s", (best['supplier_code'], best['supplier_name'], p['product_code'])) matched += 1 # 也匹配 supplier_code 为空但有 supplier_name_match 的包材产品 unmatched = query(""" SELECT product_code, supplier_name_match FROM products WHERE product_type=%s AND deleted_at IS NULL AND (supplier_code IS NULL OR supplier_code='') """, (product_type,)) for p in unmatched: pname = (p.get('supplier_name_match') or '').strip() if not pname: continue for s in suppliers: sname = (s['supplier_name'] or '').strip() if pname in sname or sname in pname: query("UPDATE products SET supplier_code=%s, supplier_name_match=%s WHERE product_code=%s", (s['supplier_code'], s['supplier_name'], p['product_code'])) matched += 1 break return jsonify({"code": 0, "msg": f"匹配完成，共更新{matched}条"}) @products_bp.route('/api/product/permanent_delete', methods=['POST']) @login_required def product_permanent_delete(): """从回收站彻底删除产品（仅管理员）""" if session.get("role") not in ("admin", "super_admin"): return jsonify({"code": 403, "msg": "仅管理员可操作"}) data = request.json code = data.get('product_code', '') if not code: return jsonify({"code": 400, "msg": "参数错误"}) try: product = query( "SELECT product_name FROM products WHERE product_code=%s AND deleted_at IS NOT NULL", (code,) ) if not product: return jsonify({"code": 404, "msg": "该产品不在回收站中，无法彻底删除"}) # 只删除当前产品及其关联记录，不连带删除其他产品。 query("DELETE FROM bom WHERE product_code=%s OR material_code=%s", (code, code)) query("DELETE FROM plan_hidden WHERE product_code=%s", (code,)) query("DELETE FROM products WHERE product_code=%s AND deleted_at IS NOT NULL", (code,)) add_operation_log('产品彻底删除', f"彻底删除产品 {code} - {product[0].get('product_name', '')}") return jsonify({"code": 0, "msg": "已彻底删除，无法恢复"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/product/batch_permanent_delete', methods=['POST']) @login_required def product_batch_permanent_delete(): """批量彻底删除回收站产品（仅管理员）。""" if session.get("role") not in ("admin", "super_admin"): return jsonify({"code": 403, "msg": "仅管理员可操作"}) data = request.json or {} codes = list(dict.fromkeys(str(code).strip() for code in data.get('product_codes', []) if str(code).strip())) if not codes: return jsonify({"code": 400, "msg": "请选择需要删除的产品"}) if len(codes) > 500: return jsonify({"code": 400, "msg": "单次最多删除500个产品"}) try: placeholders = ','.join(['%s'] * len(codes)) rows = query( f"SELECT product_code FROM products WHERE deleted_at IS NOT NULL AND product_code IN ({placeholders})", codes ) valid_codes = [row['product_code'] for row in rows] if not valid_codes: return jsonify({"code": 404, "msg": "所选产品均不在回收站中"}) valid_placeholders = ','.join(['%s'] * len(valid_codes)) query( f"DELETE FROM bom WHERE product_code IN ({valid_placeholders}) OR material_code IN ({valid_placeholders})", valid_codes + valid_codes ) query(f"DELETE FROM plan_hidden WHERE product_code IN ({valid_placeholders})", valid_codes) query( f"DELETE FROM products WHERE deleted_at IS NOT NULL AND product_code IN ({valid_placeholders})", valid_codes ) add_operation_log('产品批量彻底删除', f"批量彻底删除 {len(valid_codes)} 个产品：{','.join(valid_codes)}") return jsonify({"code": 0, "msg": f"已彻底删除{len(valid_codes)}个产品，无法恢复", "data": {"deleted": len(valid_codes)}}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/product_categories/list') @login_required def product_categories_list(): if not operation_required('product_category', 'view'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) """获取产品类目列表（按一级/二级/三级分组，最新在前）""" try: rows = query( "SELECT id, category1, category2, category3, category4 FROM product_categories ORDER BY id DESC" ) # 按一级类目分组（保持插入顺序，最新一组在前） grouped = {} for r in rows: c1 = r['category1'] or '' grouped.setdefault(c1, []).append({ 'id': r['id'], 'category1': c1, 'category2': r['category2'] or '', 'category3': r['category3'] or '', 'category4': r['category4'] or '' }) result = [] for c1 in grouped.keys(): items = grouped[c1] # 二级类目分组（保持插入顺序，最新在前） subs = {} for it in items: c2 = it['category2'] or '' subs.setdefault(c2, []).append(it) result.append({ 'category1': c1, 'count': len(items), 'children': [ {'category2': c2, 'count': len(sub), 'items': sub} for c2, sub in subs.items() ] }) return jsonify({"code": 0, "data": result, "total": len(rows)}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/product_categories/update', methods=['POST']) @login_required def product_categories_update(): if not operation_required('product_category', 'edit'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) """编辑保存类目（四列可改）""" data = request.get_json(silent=True) or {} cid = data.get('id') c1 = (data.get('category1') or '').strip() c2 = (data.get('category2') or '').strip() c3 = (data.get('category3') or '').strip() c4 = (data.get('category4') or '').strip() if not cid: return jsonify({"code": 400, "msg": "缺少类目ID"}) if not c1: return jsonify({"code": 400, "msg": "一级类目不能为空"}) try: cur = query( "UPDATE product_categories SET category1=%s, category2=%s, category3=%s, category4=%s WHERE id=%s", (c1, c2, c3, c4, cid) ) return jsonify({"code": 0, "msg": "保存成功"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/product_categories/create', methods=['POST']) @login_required def product_categories_create(): if not operation_required('product_category', 'create'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) """新增类目（四列）""" data = request.get_json(silent=True) or {} c1 = (data.get('category1') or '').strip() c2 = (data.get('category2') or '').strip() c3 = (data.get('category3') or '').strip() c4 = (data.get('category4') or '').strip() if not c1: return jsonify({"code": 400, "msg": "一级类目不能为空"}) try: # 新建前判断四级类目是否已存在（按完整名称：一级/二级/三级/四级组合） exists = query( "SELECT id FROM product_categories WHERE category1=%s AND category2=%s AND category3=%s AND category4=%s LIMIT 1", (c1, c2, c3, c4) ) if exists: full_name = '/'.join([x for x in (c1, c2, c3, c4) if x]) return jsonify({"code": 400, "msg": f"该四级类目已存在（完整名称：{full_name}），请勿重复新增"}) cur = query( "INSERT INTO product_categories (category1, category2, category3, category4) VALUES (%s, %s, %s, %s)", (c1, c2, c3, c4) ) return jsonify({"code": 0, "msg": "新增成功", "data": {"id": cur.lastrowid if hasattr(cur, 'lastrowid') else None}}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/product_categories/delete', methods=['POST']) @login_required def product_categories_delete(): if not operation_required('product_category', 'delete'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) """删除类目""" data = request.get_json(silent=True) or {} cid = data.get('id') if not cid: return jsonify({"code": 400, "msg": "缺少类目ID"}) try: query("DELETE FROM product_categories WHERE id=%s", (cid,)) return jsonify({"code": 0, "msg": "删除成功"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) # ─── 新品管理 API ─── NEW_PRODUCT_FIELDS = ['category1', 'category2', 'category3', 'product_type', 'brand', 'product_code', 'product_name', 'barcode', 'capacity', 'effect_img', 'remark', 'total_cost', 'total_payment', 'procurement_planner', 'product_dev', 'package_dev', 'designer', 'operator', 'upgrade_img', 'confirm_img', 'quote_inner_img', 'quote_pack_img', 'freight_included'] _NEW_PRODUCT_FREIGHT_READY = False def _ensure_new_product_freight_column(): """兼容旧环境：幂等补充新品是否含运字段。""" global _NEW_PRODUCT_FREIGHT_READY if _NEW_PRODUCT_FREIGHT_READY: return rows = query("SELECT COUNT(*) AS cnt FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='new_products' AND column_name='freight_included'") if not rows or int(rows[0].get('cnt') or 0) == 0: query("ALTER TABLE new_products ADD COLUMN freight_included VARCHAR(10) NOT NULL DEFAULT '' COMMENT '是否含运' AFTER quote_pack_img") _NEW_PRODUCT_FREIGHT_READY = True def _is_inner_material(material_name, material_code): """内料判定：名称含「内料」或编码含 -NL（与系统其他位置口径一致）。""" return ('内料' in (material_name or '')) or ('-NL' in (material_code or '')) def resolve_new_product_planner(product_id): """按品牌与 BOM 内料供应商匹配新品采购计划。 第一层：品牌为海外品时不动（留空或保留人工维护值）； 第二层：取该新品 BOM 中内料物料的供应商，匹配供应商管理里绑定的采购计划。 匹配到时更新，匹配不到不覆盖原值。 """ rows = query("SELECT brand FROM new_products WHERE id=%s", (product_id,)) if not rows: return None brand = (rows[0].get('brand') or '').strip() if brand: cat = query("SELECT category FROM brand_information WHERE brand=%s LIMIT 1", (brand,)) if cat and (cat[0].get('category') or '').strip() == '海外': return None materials = query(""" SELECT material_code, material_name, supplier_code, supplier FROM new_product_bom WHERE product_id=%s ORDER BY sort_order, id """, (product_id,)) for m in materials: if not _is_inner_material(m.get('material_name'), m.get('material_code')): continue supplier_code = (m.get('supplier_code') or '').strip() supplier_name = (m.get('supplier') or '').strip() hit = None if supplier_code: r = query("SELECT procurement_planner FROM suppliers WHERE supplier_code=%s", (supplier_code,)) hit = r[0] if r else None if not hit and supplier_name: r = query("SELECT procurement_planner FROM suppliers WHERE supplier_name=%s", (supplier_name,)) hit = r[0] if r else None if not hit and supplier_name: r = query("SELECT procurement_planner FROM suppliers WHERE supplier_name LIKE %s ORDER BY id LIMIT 1", (f"%{supplier_name}%",)) hit = r[0] if r else None planner = (hit.get('procurement_planner') or '').strip() if hit else '' if planner: query("UPDATE new_products SET procurement_planner=%s WHERE id=%s", (planner, product_id)) return planner return None def _sync_new_product_to_product_info(conn, product_id): """在同一事务中同步新品成品、包材和 BOM，不删除目标表已有记录。""" with conn.cursor() as cursor: cursor.execute("SELECT * FROM new_products WHERE id=%s FOR UPDATE", (product_id,)) product = cursor.fetchone() if not product: raise ValueError("新品数据不存在，请刷新后重试") product_code = (product.get('product_code') or '').strip() product_name = (product.get('product_name') or '').strip() if not product_code or not product_name: raise ValueError("产品编码和产品名称不能为空") cursor.execute(""" SELECT material_code, material_name, supplier_code, supplier, lead_time, unit_price, quantity, usage_quantity, common_code, common_name FROM new_product_bom WHERE product_id=%s ORDER BY sort_order, id """, (product_id,)) source_materials = cursor.fetchall() materials_by_code = {} for material in source_materials: # 物料编号/物料名称对应新品成品管理BOM里的物料编码/物料名称， # 通用物料编码/名称不参与物料清单的编号名称映射。 material_code = (material.get('material_code') or '').strip() if not material_code: continue material_name = (material.get('material_name') or '').strip() materials_by_code[material_code] = { 'material_code': material_code, 'material_name': material_name, 'supplier_code': (material.get('supplier_code') or '').strip(), 'supplier_name': (material.get('supplier') or '').strip(), 'lead_time': int(material.get('lead_time') or 0), 'unit_price': float(material.get('unit_price') or 0), # 新品BOM“数量”是新品采购数量；同步到现有BOM的用量使用独立字段。 'quantity': float(material.get('usage_quantity')) if material.get('usage_quantity') is not None else 1.0, } materials = list(materials_by_code.values()) # ─── 计算成品同步属性（供应商取内料、货期=内料货期+其他物料货期最大值、成本价=物料成本价之和）─── inner_material = None other_leads = [] total_material_cost = 0.0 for m in materials: total_material_cost += m['unit_price'] if '内料' in (m['material_name'] or ''): inner_material = m else: other_leads.append(m['lead_time']) if inner_material: prod_supplier_code = inner_material['supplier_code'] prod_supplier_name = inner_material['supplier_name'] prod_lead_time = inner_material['lead_time'] + (max(other_leads) if other_leads else 0) else: prod_supplier_code = next((m['supplier_code'] for m in materials if m['supplier_code']), '') prod_supplier_name = next((m['supplier_name'] for m in materials if m['supplier_name']), '') prod_lead_time = max((m['lead_time'] for m in materials), default=0) prod_cost_price = round(total_material_cost, 4) all_codes = [product_code] + [material['material_code'] for material in materials] placeholders = ','.join(['%s'] * len(all_codes)) cursor.execute( f"SELECT product_code, product_name FROM products WHERE product_code IN ({placeholders})", tuple(all_codes) ) existing_product_rows = cursor.fetchall() existing_product_names = { row['product_code']: row.get('product_name') or '' for row in existing_product_rows } existing_product_codes = set(existing_product_names) cursor.execute( "SELECT material_code FROM bom WHERE product_code=%s", (product_code,) ) existing_bom_codes = {row['material_code'] for row in cursor.fetchall()} cursor.execute(""" INSERT INTO products (product_code, product_name, specification, brand, product_type, category1, category2, category3, operator, developer, cost_price, lead_time, supplier_code, supplier_name_match, created_at, import_date, deleted_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,CURDATE(),NULL) ON DUPLICATE KEY UPDATE product_name=VALUES(product_name), specification=IF(VALUES(specification)='', specification, VALUES(specification)), brand=IF(VALUES(brand)='', brand, VALUES(brand)), product_type=VALUES(product_type), category1=IF(VALUES(category1)='', category1, VALUES(category1)), category2=IF(VALUES(category2)='', category2, VALUES(category2)), category3=IF(VALUES(category3)='', category3, VALUES(category3)), operator=IF(VALUES(operator)='', operator, VALUES(operator)), developer=IF(VALUES(developer)='', developer, VALUES(developer)), cost_price=IF(VALUES(cost_price)>0, VALUES(cost_price), cost_price), lead_time=IF(VALUES(lead_time)>0, VALUES(lead_time), lead_time), supplier_code=IF(VALUES(supplier_code)='', supplier_code, VALUES(supplier_code)), supplier_name_match=IF(VALUES(supplier_name_match)='', supplier_name_match, VALUES(supplier_name_match)), deleted_at=NULL """, ( product_code, product_name, (product.get('capacity') or '').strip(), (product.get('brand') or '').strip(), (product.get('product_type') or '成品').strip(), (product.get('category1') or '').strip(), (product.get('category2') or '').strip(), (product.get('category3') or '').strip(), (product.get('operator') or '').strip(), (product.get('product_dev') or '').strip(), prod_cost_price, prod_lead_time, prod_supplier_code, prod_supplier_name, product.get('created_at'), )) for material in materials: material_code = material['material_code'] material_name = material['material_name'] or existing_product_names.get(material_code) or material_code cursor.execute(""" INSERT INTO products (product_code, product_name, product_type, cost_price, lead_time, supplier_code, supplier_name_match, developer, import_date, deleted_at) VALUES (%s,%s,'包材',%s,%s,%s,%s,%s,CURDATE(),NULL) ON DUPLICATE KEY UPDATE product_name=VALUES(product_name), product_type='包材', cost_price=IF(VALUES(cost_price)>0, VALUES(cost_price), cost_price), lead_time=IF(VALUES(lead_time)>0, VALUES(lead_time), lead_time), supplier_code=IF(VALUES(supplier_code)='', supplier_code, VALUES(supplier_code)), supplier_name_match=IF(VALUES(supplier_name_match)='', supplier_name_match, VALUES(supplier_name_match)), developer=IF(VALUES(developer)='', developer, VALUES(developer)), import_date=CURDATE(), deleted_at=NULL """, ( material_code, material_name, material['unit_price'], material['lead_time'], material['supplier_code'], material['supplier_name'], (product.get('package_dev') or '').strip(), )) cursor.execute(""" INSERT INTO bom (product_code, product_name, material_code, material_name, supplier, quantity, material_cost_price) VALUES (%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE product_name=VALUES(product_name), material_name=VALUES(material_name), supplier=IF(VALUES(supplier)='', supplier, VALUES(supplier)), quantity=VALUES(quantity), material_cost_price=IF(VALUES(material_cost_price)>0, VALUES(material_cost_price), material_cost_price), updated_at=CURRENT_TIMESTAMP """, ( product_code, product_name, material_code, material_name, material['supplier_name'], material['quantity'], material['unit_price'], )) material_codes = {material['material_code'] for material in materials} return { 'product_created': 0 if product_code in existing_product_codes else 1, 'product_updated': 1 if product_code in existing_product_codes else 0, 'materials_total': len(materials), 'materials_created': len(material_codes - existing_product_codes), 'materials_updated': len(material_codes & existing_product_codes), 'bom_total': len(materials), 'bom_created': len(material_codes - existing_bom_codes), 'bom_updated': len(material_codes & existing_bom_codes), } @products_bp.route('/api/new_products/list') @login_required def new_products_list(): _ensure_new_product_freight_column() if not operation_required('product_new', 'view'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) """获取新品列表（最新在前），支持编码、名称和BOM供应商筛选""" product_code = (request.args.get('product_code') or '').strip() product_name = (request.args.get('product_name') or '').strip() unified_search = (request.args.get('search') or '').strip() supplier = (request.args.get('supplier') or '').strip() where = [] params = [] if unified_search: where.append("(product_code LIKE %s OR product_name LIKE %s)") params.extend([f"%{unified_search}%", f"%{unified_search}%"]) if product_code: where.append("product_code LIKE %s") params.append(f"%{product_code}%") if product_name: where.append("product_name LIKE %s") params.append(f"%{product_name}%") if supplier: where.append("EXISTS (SELECT 1 FROM new_product_bom npb WHERE npb.product_id = new_products.id AND (npb.supplier LIKE %s OR npb.supplier_code LIKE %s))") params.extend([f"%{supplier}%", f"%{supplier}%"]) where_clause = " AND ".join(where) if where else "1=1" try: rows = query("SELECT id, " + ",".join(NEW_PRODUCT_FIELDS) + " FROM new_products WHERE " + where_clause + " ORDER BY id DESC", tuple(params)) return jsonify({"code": 0, "data": rows, "total": len(rows)}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) def _new_product_image_file(image_value, project_root): """把新品图片字段中的静态资源地址解析为本地文件。""" raw_value = str(image_value or '').strip() if not raw_value: return None # 图片字段通常保存 /static/...，同时兼容历史记录中的相对路径、绝对路径和带域名地址。 if raw_value.startswith(('http://', 'https://')): return None if raw_value.startswith('/static/'): local_path = os.path.join(project_root, raw_value.lstrip('/')) elif raw_value.startswith('static/'): local_path = os.path.join(project_root, raw_value) elif os.path.isabs(raw_value): local_path = raw_value else: local_path = os.path.join(project_root, raw_value) return local_path if os.path.isfile(local_path) else None @products_bp.route('/api/new_products/export', methods=['POST']) @login_required def new_products_export(): _ensure_new_product_freight_column() """导出新品成品管理数据，字段由前端选择。""" if not operation_required('product_new', 'view'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) data = request.get_json(silent=True) or {} fields = data.get('fields') or [] code_filter = (data.get('product_code') or '').strip() name_filter = (data.get('product_name') or '').strip() unified_search = (data.get('search') or '').strip() supplier_filter = (data.get('supplier') or '').strip() field_map = { 'category1': ('category1', '一级类目'), 'category2': ('category2', '二级类目'), 'category3': ('category3', '三级类目'), 'product_type': ('product_type', '类别'), 'brand': ('brand', '品牌'), 'product_code': ('product_code', '产品编码'), 'product_name': ('product_name', '产品名称'), 'barcode': ('barcode', '条形码'), 'capacity': ('capacity', '容量'), 'remark': ('remark', '成品备注'), 'total_cost': ('total_cost', '产品总成本'), 'total_payment': ('total_payment', '产品总货款'), 'effect_img': ('effect_img', '产品效果图（内外包）'), 'upgrade_img': ('upgrade_img', '升级前的'), 'confirm_img': ('confirm_img', '确版内料图凭证'), 'quote_inner_img': ('quote_inner_img', '内料报价单'), 'quote_pack_img': ('quote_pack_img', '包材报价单'), 'freight_included': ('freight_included', '是否含运'), 'product_dev': ('product_dev', '产品开发'), 'package_dev': ('package_dev', '包材开发'), 'designer': ('designer', '设计师'), 'operator': ('operator', '负责运营') } selected = [f for f in fields if f in field_map] if not selected: return jsonify({"code": 400, "msg": "请至少选择一个导出字段"}) where = [] params = [] if unified_search: where.append('(product_code LIKE %s OR product_name LIKE %s)'); params.extend([f'%{unified_search}%', f'%{unified_search}%']) if code_filter: where.append('product_code LIKE %s'); params.append(f'%{code_filter}%') if name_filter: where.append('product_name LIKE %s'); params.append(f'%{name_filter}%') if supplier_filter: where.append("EXISTS (SELECT 1 FROM new_product_bom npb WHERE npb.product_id=new_products.id AND (npb.supplier LIKE %s OR npb.supplier_code LIKE %s))") params.extend([f'%{supplier_filter}%', f'%{supplier_filter}%']) where_clause = ' AND '.join(where) if where else '1=1' try: rows = query( 'SELECT ' + ','.join(f'{field_map[f][0]} AS `{f}`' for f in selected) + f' FROM new_products WHERE {where_clause} ORDER BY id DESC', params) import io import openpyxl from openpyxl.drawing.image import Image as ExcelImage from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter wb = openpyxl.Workbook(); ws = wb.active; ws.title = '新品成品管理' headers = [field_map[f][1] for f in selected] ws.append(headers) thin = Side(style='thin', color='D9D9D9') for cell in ws[1]: cell.font = Font(bold=True, color='FFFFFF'); cell.fill = PatternFill('solid', fgColor='1890FF') cell.alignment = Alignment(horizontal='center'); cell.border = Border(left=thin, right=thin, top=thin, bottom=thin) image_fields = {'effect_img', 'upgrade_img', 'confirm_img', 'quote_inner_img', 'quote_pack_img'} project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) for row_index, row in enumerate(rows, 2): row_has_embedded_image = False vals = [] for f in selected: value = row.get(f, '') if value is None: value = '' if f in ('total_cost', 'total_payment') and value != '': try: value = f'{float(value):.3f}' except (TypeError, ValueError): pass # 图片字段改为 Excel 内嵌图片，不再把 /static/... 路径作为普通文本写入。 vals.append('' if f in image_fields else value) ws.append(vals) for col_index, f in enumerate(selected, 1): if f not in image_fields or not row.get(f): continue image_paths = [str(p).strip() for p in str(row[f]).split(',') if str(p).strip()] for image_path in image_paths: local_path = _new_product_image_file(image_path, project_root) if not local_path: # 原图不在测试环境时不输出静态路径，避免把不可用地址误当成图片内容。 cell = ws.cell(row=row_index, column=col_index) cell.value = '图片缺失' continue try: image = ExcelImage(local_path) max_width, max_height = 100, 72 ratio = min(max_width / image.width, max_height / image.height, 1) image.width = max(1, int(image.width * ratio)) image.height = max(1, int(image.height * ratio)) image.anchor = f'{get_column_letter(col_index)}{row_index}' ws.add_image(image) row_has_embedded_image = True break except Exception: continue if row_has_embedded_image: ws.row_dimensions[row_index].height = 60 for row in ws.iter_rows(): for cell in row: cell.border = Border(left=thin, right=thin, top=thin, bottom=thin) for i, header in enumerate(headers, 1): ws.column_dimensions[get_column_letter(i)].width = 15 if selected[i - 1] in image_fields else min(max(len(str(header)) * 1.8, 12), 32) output = io.BytesIO(); wb.save(output); output.seek(0) from flask import send_file return send_file(output, as_attachment=True, download_name='新品成品管理.xlsx', mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet') except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/new_products/create', methods=['POST']) @login_required def new_products_create(): _ensure_new_product_freight_column() if not operation_required('product_new', 'create'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) """新增新品（全部字段）""" data = request.get_json(silent=True) or {} vals = {f: (data.get(f) or '').strip() for f in NEW_PRODUCT_FIELDS} if not vals['product_name'] and not vals['product_code']: return jsonify({"code": 400, "msg": "产品名称或产品编码至少填写一项"}) try: placeholders = ",".join(["%s"] * len(NEW_PRODUCT_FIELDS)) query( "INSERT INTO new_products (" + ",".join(NEW_PRODUCT_FIELDS) + ") VALUES (" + placeholders + ")", tuple(vals[f] for f in NEW_PRODUCT_FIELDS) ) return jsonify({"code": 0, "msg": "新增成功"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/new_products/update', methods=['POST']) @login_required def new_products_update(): _ensure_new_product_freight_column() if not operation_required('product_new', 'edit'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) """编辑保存新品""" data = request.get_json(silent=True) or {} pid = data.get('id') if not pid: return jsonify({"code": 400, "msg": "缺少ID"}) vals = {f: (data.get(f) or '').strip() for f in NEW_PRODUCT_FIELDS} try: sets = ",".join([f + "=%s" for f in NEW_PRODUCT_FIELDS]) query( "UPDATE new_products SET " + sets + " WHERE id=%s", tuple(vals[f] for f in NEW_PRODUCT_FIELDS) + (pid,) ) return jsonify({"code": 0, "msg": "保存成功"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/new_products/delete', methods=['POST']) @login_required def new_products_delete(): if not operation_required('product_new', 'delete'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) """删除新品""" data = request.get_json(silent=True) or {} pid = data.get('id') if not pid: return jsonify({"code": 400, "msg": "缺少ID"}) try: query("DELETE FROM new_products WHERE id=%s", (pid,)) return jsonify({"code": 0, "msg": "删除成功"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) NEW_UPLOAD_FOLDER = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'static', 'uploads', 'new_products') os.makedirs(NEW_UPLOAD_FOLDER, exist_ok=True) @products_bp.route('/api/new_products/upload_image', methods=['POST']) @login_required def new_products_upload_image(): if not operation_required('product_new', 'edit'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) """上传新品图片（产品效果图/升级前内料图/确版内料图凭证）""" file = request.files.get('file') if not file or not file.filename: return jsonify({"code": 400, "msg": "请选择文件"}) ext = os.path.splitext(file.filename)[1].lower() if ext not in ('.jpg', '.jpeg', '.png', '.gif', '.webp', '.bmp'): return jsonify({"code": 400, "msg": "仅支持图片格式（jpg/png/gif/webp/bmp）"}) safe_name = f"{uuid.uuid4().hex}{ext}" file.save(os.path.join(NEW_UPLOAD_FOLDER, safe_name)) return jsonify({"code": 0, "msg": "上传成功", "data": {"url": f"/static/uploads/new_products/{safe_name}"}}) @products_bp.route('/api/new_products/sync', methods=['POST']) @login_required def new_products_sync(): if not operation_required('product_new', 'sync'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) """同步新品成品信息、包材信息和 BOM 到产品信息管理。""" if not _check_product_edit_perm(): return jsonify({"code": 403, "msg": "无产品信息编辑权限"}) data = request.get_json(silent=True) or {} product_id = data.get('product_id') if not product_id: return jsonify({"code": 400, "msg": "缺少新品ID"}) try: product_id = int(product_id) except (TypeError, ValueError): return jsonify({"code": 400, "msg": "新品ID格式错误"}) conn = get_db() try: conn.begin() result = _sync_new_product_to_product_info(conn, product_id) conn.commit() add_operation_log( '新品资料同步', f"新品ID {product_id}：成品{result['product_created']}条新增/{result['product_updated']}条更新，" f"包材{result['materials_created']}条新增/{result['materials_updated']}条更新，" f"BOM{result['bom_created']}条新增/{result['bom_updated']}条更新" ) return jsonify({"code": 0, "msg": "同步成功", "data": result}) except ValueError as error: conn.rollback() return jsonify({"code": 400, "msg": str(error)}) except Exception as error: conn.rollback() return jsonify({"code": 500, "msg": f"同步失败：{error}"}) finally: conn.close() @products_bp.route('/api/new_products/category_search') @login_required def new_products_category_search(): if not operation_required('product_new', 'view'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) """三级类目模糊搜索：根据三级类目匹配对应一级/二级类目""" q = (request.args.get('q') or '').strip() if not q: return jsonify({"code": 0, "data": []}) rows = query( "SELECT DISTINCT category1, category2, category3 FROM product_categories WHERE category3 LIKE %s ORDER BY category3 LIMIT 20", (f"%{q}%",) ) return jsonify({"code": 0, "data": rows}) def _tax_included_value(value): """新品BOM税点展示规则：税点非0=是，税点为0或空=否。""" raw = str(value or '').strip().replace('%', '') try: return '是' if float(raw or 0) != 0 else '否' except (TypeError, ValueError): return '否' @products_bp.route('/api/new_products/bom') @login_required def new_products_bom(): if not operation_required('product_new', 'view'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) """新品BOM清单：查独立的新品BOM表 new_product_bom（不复用现有物料清单 bom 表）""" pid = (request.args.get('product_id') or '').strip() if not pid: return jsonify({"code": 400, "msg": "缺少新品ID"}) try: rows = query(""" SELECT id, material_code, material_name, usage_quantity, supplier_code, supplier, lead_time, tax_rate, unit_price, quantity, amount, payment_method, can_order, remark, common_code, common_name, order_date FROM new_product_bom WHERE product_id=%s ORDER BY sort_order, id """, (int(pid),)) data = [] for r in rows: data.append({ 'id': r['id'], 'material_code': r['material_code'] or '', 'material_name': r['material_name'] or '', 'usage_quantity': float(r['usage_quantity']) if r['usage_quantity'] is not None else 1.0, 'supplier_code': r['supplier_code'] or '', 'supplier': r['supplier'] or '', 'lead_time': r['lead_time'] if r['lead_time'] is not None else '', 'tax_rate': r['tax_rate'] or '', 'tax_included': _tax_included_value(r['tax_rate']), 'unit_price': float(r['unit_price'] or 0), 'quantity': float(r['quantity'] or 0), 'amount': float(r['amount'] or 0), 'payment_method': r['payment_method'] or '', 'can_order': r['can_order'] or '', 'remark': r['remark'] or '', 'common_code': r['common_code'] or '', 'common_name': r['common_name'] or '', 'order_date': str(r['order_date']) if r['order_date'] else '' }) return jsonify({"code": 0, "data": data}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/new_products/bom/save', methods=['POST']) @login_required def new_products_bom_save(): if not operation_required('product_new', 'edit'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) """保存新品BOM清单（全量替换该新品的物料行）""" data = request.get_json(silent=True) or {} pid = data.get('product_id') rows = data.get('rows') or [] if not pid: return jsonify({"code": 400, "msg": "缺少新品ID"}) try: pid = int(pid) # 过滤全空行（物料编码和包材明细都为空则跳过） valid = [] for r in rows: if not (r.get('material_code') or '').strip() and not (r.get('material_name') or '').strip(): continue valid.append(r) query("DELETE FROM new_product_bom WHERE product_id=%s", (pid,)) total_cost = 0.0 total_payment = 0.0 for i, r in enumerate(valid): unit_price = float(r.get('unit_price') or 0) quantity = float(r.get('quantity') or 0) raw_usage_quantity = str(r.get('usage_quantity') if r.get('usage_quantity') is not None else '').strip() usage_quantity = float(raw_usage_quantity) if raw_usage_quantity else 1.0 amount = round(unit_price * quantity, 2) total_cost += unit_price total_payment += amount order_date = (r.get('order_date') or '').strip() or None query(""" INSERT INTO new_product_bom (product_id, material_code, material_name, usage_quantity, supplier_code, supplier, lead_time, tax_rate, unit_price, quantity, amount, payment_method, can_order, remark, common_code, common_name, order_date, sort_order) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) """, (pid, (r.get('material_code') or '').strip(), (r.get('material_name') or '').strip(), usage_quantity, (r.get('supplier_code') or '').strip(), (r.get('supplier') or '').strip(), int(r.get('lead_time') or 0), (r.get('tax_rate') or '').strip(), unit_price, quantity, amount, (r.get('payment_method') or '').strip(), (r.get('can_order') or '').strip(), (r.get('remark') or '').strip(), (r.get('common_code') or '').strip(), (r.get('common_name') or '').strip(), order_date, i)) # 同步更新新品表：产品总成本=物料单价之和，产品总货款=物料货款之和（保留3位小数） total_cost = round(total_cost, 3) total_payment = round(total_payment, 3) query("UPDATE new_products SET total_cost=%s, total_payment=%s WHERE id=%s", (f"{total_cost:.3f}", f"{total_payment:.3f}", pid)) # 绑定BOM后按品牌与内料供应商匹配采购计划（海外品不动，由人工维护） matched_planner = resolve_new_product_planner(pid) return jsonify({"code": 0, "msg": "保存成功", "data": {"count": len(valid), "total_cost": total_cost, "total_payment": total_payment, "procurement_planner": matched_planner or ''}}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) def _new_packaging_where(product_filter='', material_filter='', supplier_filter=''): """新品包材列表与导出共用同一套筛选条件，确保两处数据口径一致。""" where = [] params = [] if product_filter: where.append('(p.product_code LIKE %s OR p.product_name LIKE %s)') params.extend([f'%{product_filter}%', f'%{product_filter}%']) if material_filter: where.append('(b.material_code LIKE %s OR b.material_name LIKE %s OR b.common_code LIKE %s OR b.common_name LIKE %s)') params.extend([f'%{material_filter}%'] * 4) if supplier_filter: where.append('(b.supplier LIKE %s OR b.supplier_code LIKE %s)') params.extend([f'%{supplier_filter}%', f'%{supplier_filter}%']) return (' AND '.join(where) if where else '1=1'), params @products_bp.route('/api/new_products/packaging') @login_required def new_products_packaging(): """新品包材管理：直接读取新品BOM弹窗保存的全部物料字段。""" if not operation_required('product_packaging', 'view'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) page = max(int(request.args.get('page', 1)), 1) size = min(max(int(request.args.get('size', 50)), 1), 200) offset = (page - 1) * size product_filter = (request.args.get('product') or '').strip() material_filter = (request.args.get('material') or '').strip() supplier_filter = (request.args.get('supplier') or '').strip() where_clause, params = _new_packaging_where(product_filter, material_filter, supplier_filter) try: total = query(f""" SELECT COUNT(*) AS cnt FROM new_product_bom b JOIN new_products p ON p.id = b.product_id WHERE {where_clause} """, params)[0]['cnt'] rows = query(f""" SELECT p.product_code AS finished_product_code, p.product_name AS finished_product_name, b.material_code, b.material_name, b.usage_quantity, b.supplier_code, b.supplier, b.lead_time, b.tax_rate, b.unit_price, b.quantity, b.amount, b.payment_method, b.can_order, b.remark, b.common_code, b.common_name, b.order_date FROM new_product_bom b JOIN new_products p ON p.id = b.product_id WHERE {where_clause} ORDER BY p.id DESC, b.sort_order ASC, b.id ASC LIMIT {size} OFFSET {offset} """, params) for row in rows: for key in ('unit_price', 'quantity', 'amount'): if row.get(key) is not None: row[key] = float(row[key]) row['usage_quantity'] = float(row['usage_quantity']) if row['usage_quantity'] is not None else 1.0 if row.get('order_date'): row['order_date'] = row['order_date'].isoformat() if hasattr(row['order_date'], 'isoformat') else str(row['order_date'])[:10] row['tax_included'] = _tax_included_value(row.get('tax_rate')) _filter_field_rows('product_packaging', rows, session.get('user_id'), session.get('role'), page_view=operation_required('product_packaging', 'view'), page_edit=operation_required('product_packaging', 'edit')) return jsonify({"code": 0, "data": {"rows": rows, "total": total, "page": page, "size": size}}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/new_products/packaging/export', methods=['POST']) @login_required def new_products_packaging_export(): """导出新品包材管理数据，字段由前端选择。""" if not operation_required('product_packaging', 'view'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) data = request.get_json(silent=True) or {} fields = data.get('fields') or [] product_filter = (data.get('product') or '').strip() material_filter = (data.get('material') or '').strip() supplier_filter = (data.get('supplier') or '').strip() where_clause, params = _new_packaging_where(product_filter, material_filter, supplier_filter) field_map = { 'finished_product_code': ('p.product_code', '成品编码'), 'finished_product_name': ('p.product_name', '成品名称'), 'material_code': ('b.material_code', '物料编码'), 'material_name': ('b.material_name', '物料名称'), 'usage_quantity': ('b.usage_quantity', '用量'), 'supplier_code': ('b.supplier_code', '供应商编码'), 'supplier': ('b.supplier', '供应商'), 'lead_time': ('b.lead_time', '货期/天'), 'tax_rate': ('b.tax_rate', '税点'), # PyMySQL 会先对 SQL 做 %-格式化，SQL 字面量百分号必须写成 %%。 'tax_included': ("CASE WHEN CAST(REPLACE(TRIM(COALESCE(b.tax_rate, '')), '%%', '') AS DECIMAL(12,4)) <> 0 THEN '是' ELSE '否' END", '是否含税'), 'unit_price': ('b.unit_price', '单价/元'), 'quantity': ('b.quantity', '数量'), 'amount': ('b.amount', '货款/元'), 'payment_method': ('b.payment_method', '结款方式'), 'can_order': ('b.can_order', '是否可下单'), 'remark': ('b.remark', '下单备注'), 'common_code': ('b.common_code', '通用物料编码'), 'common_name': ('b.common_name', '通用物料名称'), 'order_date': ('b.order_date', '下单日期') } selected = [f for f in fields if f in field_map] if not selected: return jsonify({"code": 400, "msg": "请至少选择一个导出字段"}) try: rows = query( 'SELECT ' + ','.join(f'{field_map[f][0]} AS `{f}`' for f in selected) + f' FROM new_product_bom b JOIN new_products p ON p.id=b.product_id' + f' WHERE {where_clause} ORDER BY p.id DESC,b.sort_order ASC,b.id ASC', tuple(params) ) import io import openpyxl from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter wb = openpyxl.Workbook(); ws = wb.active; ws.title = '新品包材管理' headers = [field_map[f][1] for f in selected]; ws.append(headers) thin = Side(style='thin', color='D9D9D9') for cell in ws[1]: cell.font = Font(bold=True, color='FFFFFF'); cell.fill = PatternFill('solid', fgColor='1890FF') cell.alignment = Alignment(horizontal='center'); cell.border = Border(left=thin, right=thin, top=thin, bottom=thin) for row in rows: vals = [] for f in selected: value = row.get(f, '') if value is None: value = '' if f in ('unit_price', 'quantity', 'amount') and value != '': try: value = float(value) except (TypeError, ValueError): pass vals.append(value) ws.append(vals) for row in ws.iter_rows(): for cell in row: cell.border = Border(left=thin, right=thin, top=thin, bottom=thin) for i, header in enumerate(headers, 1): ws.column_dimensions[get_column_letter(i)].width = min(max(len(str(header)) * 1.8, 12), 32) output = io.BytesIO(); wb.save(output); output.seek(0) from flask import send_file return send_file(output, as_attachment=True, download_name='新品包材管理.xlsx', mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet') except Exception as e: return jsonify({"code": 500, "msg": str(e)}) @products_bp.route('/api/new_products/brands') @login_required def new_products_brands(): if not operation_required('product_new', 'view'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有该操作权限"}) """品牌列表（现有品牌下拉）""" rows = query("SELECT DISTINCT brand FROM products WHERE brand IS NOT NULL AND brand != '' ORDER BY brand") return jsonify({"code": 0, "data": [r['brand'] for r in rows]}) @products_bp.route('/api/product/remark/save', methods=['POST']) @login_required def product_remark_save(): """保存产品批注（下单信息弹窗批注编辑框用，权限与下单信息编辑一致）""" if not operation_required('purchase', 'edit'): return jsonify({"code": 403, "msg": "权限不足，当前账号没有编辑下单信息权限"}) data = request.get_json(silent=True) or {} product_code = (data.get('product_code') or '').strip() remark = (data.get('remark') or '').strip() if not product_code: return jsonify({"code": 400, "msg": "缺少货品编号"}) query("UPDATE products SET remark=%s WHERE product_code=%s", (remark, product_code)) return jsonify({"code": 0, "msg": "保存成功"}) @products_bp.route('/api/bom_packaging/save', methods=['POST']) @login_required def bom_packaging_save(): """保存下单前/后包材数据到 bom_packaging 表""" if not (operation_required('planning', 'edit') or operation_required('purchase', 'edit')): return jsonify({"code": 403, "msg": "权限不足，当前账号没有编辑计划预测或待下单信息权限"}) data = request.get_json(silent=True) if not data: return jsonify({"code": 400, "msg": "缺少请求数据"}) product_code = data.get('product_code', '').strip() materials = data.get('materials', []) if not product_code or not materials: return jsonify({"code": 400, "msg": "缺少产品编号或物料数据"}) try: for m in materials: mc = (m.get('material_code') or '').strip() if not mc: continue pre_pack = int(m.get('pre_pack_qty') or 0) # 下单后包材由确认下单时按滚动逻辑计算，前端传入值不允许覆盖。 pkg_remark = str(m.get('pkg_remark') or '').strip() query("""INSERT INTO bom_packaging (product_code, material_code, pre_pack_qty, post_pack_qty, pkg_remark) VALUES (%s,%s,%s,0,%s) ON DUPLICATE KEY UPDATE pre_pack_qty=%s, pkg_remark=%s, updated_at=NOW()""", (product_code, mc, pre_pack, pkg_remark, pre_pack, pkg_remark)) return jsonify({"code": 0, "msg": "保存成功"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)})