from flask import Flask, render_template, request, jsonify, send_file import pandas as pd from openpyxl import load_workbook import os import shutil from datetime import datetime import re from collections import Counter app = Flask(__name__) app.config['MAX_CONTENT_LENGTH'] = 50 * 1024 * 1024 # 50MB # 配置路径 BASE_DIR = os.path.dirname(os.path.abspath(__file__)) UPLOAD_FOLDER = os.path.join(BASE_DIR, 'uploads') TEMPLATES_FOLDER = os.path.join(BASE_DIR, 'templates_excel') OUTPUT_FOLDER = os.path.join(BASE_DIR, 'output') # 创建必要的目录 for folder in [UPLOAD_FOLDER, TEMPLATES_FOLDER, OUTPUT_FOLDER]: if not os.path.exists(folder): os.makedirs(folder) def load_valid_values(wb): """从 Valid Values 工作表读取有效值""" valid_data = {} if "Valid Values" not in wb.sheetnames: return valid_data ws_valid = wb["Valid Values"] # 读取第1行作为列名 headers = [] for col in range(1, ws_valid.max_column + 1): val = ws_valid.cell(row=1, column=col).value if val: headers.append(str(val).strip()) else: headers.append(f"Column_{col}") # 读取数据行 for row in range(2, ws_valid.max_row + 1): for col in range(1, len(headers) + 1): val = ws_valid.cell(row=row, column=col).value if val: header = headers[col - 1] if header not in valid_data: valid_data[header] = [] if str(val).strip() not in valid_data[header]: valid_data[header].append(str(val).strip()) return valid_data def parse_excel_file(file_path, source_header_row): """读取来源Excel""" df = pd.read_excel(file_path, header=source_header_row - 1) return df def get_template_list(): """获取所有可用模板""" templates = [] for f in os.listdir(TEMPLATES_FOLDER): if f.endswith('.xlsx') or f.endswith('.xls'): templates.append(f) return templates @app.route('/') def index(): templates = get_template_list() return render_template('index.html', templates=templates) @app.route('/get_templates') def get_templates(): return jsonify({'templates': get_template_list()}) @app.route('/preview', methods=['POST']) def preview(): """预览来源数据""" try: source_file = request.files.get('source_file') template_name = request.form.get('template_name') source_header = int(request.form.get('source_header', 2)) if not source_file: return jsonify({'error': '请选择来源文件'}), 400 if not template_name: return jsonify({'error': '请选择模板'}), 400 # 保存来源文件 timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') source_path = os.path.join(UPLOAD_FOLDER, f'source_{timestamp}_{source_file.filename}') source_file.save(source_path) # 读取来源数据 df = pd.read_excel(source_path, header=source_header - 1) # 预览前10行 preview_data = df.head(10).fillna('').to_dict('records') columns = df.columns.tolist() return jsonify({ 'success': True, 'columns': columns, 'data': preview_data, 'total_rows': len(df), 'source_path': source_path }) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/generate', methods=['POST']) def generate(): """生成并下载""" try: source_path = request.form.get('source_path') template_name = request.form.get('template_name') source_header = int(request.form.get('source_header', 2)) header_row = int(request.form.get('header_row', 4)) if not source_path or not os.path.exists(source_path): return jsonify({'error': '来源文件不存在'}), 400 if not template_name: return jsonify({'error': '请选择模板'}), 400 template_path = os.path.join(TEMPLATES_FOLDER, template_name) if not os.path.exists(template_path): return jsonify({'error': '模板文件不存在'}), 400 # ========== 读取来源数据 ========== df = pd.read_excel(source_path, header=source_header - 1) # ========== 加载模板 ========== wb = load_workbook(template_path) ws = wb.active # 加载 Valid Values valid_values = load_valid_values(wb) # ========== 读取模板表头 ========== col_indices = {} for col in range(1, ws.max_column + 1): cell_value = ws.cell(row=header_row, column=col).value if cell_value: col_name = str(cell_value).strip() if col_name and not col_name.startswith("Unnamed"): col_indices[col_name] = col # ========== 找到已有数据的最后一行 ========== last_row = 5 for row in range(5, ws.max_row + 2): is_empty = True for col in range(1, min(20, ws.max_column + 1)): if ws.cell(row=row, column=col).value is not None: is_empty = False break if is_empty: last_row = row break else: last_row = ws.max_row + 1 # ========== 匹配来源列 ========== source_mapping = { "Item Code": "Item Code", "产品名称": "产品名称", "优惠单价": "优惠单价", "产品重量": "产品重量(磅)", "包装尺寸-重量": "包装尺寸-重量(磅)", "包装尺寸-高度": "包装尺寸-高度(英寸)", "包装尺寸-宽度": "包装尺寸-宽度(英寸)", "包装尺寸-长度": "包装尺寸-长度(英寸)", "颜色": "颜色", "材质": "材质", "组装长度": "组装长度(英寸)", "组装高度": "组装高度(英寸)", "组装宽度": "组装宽度(英寸)", "产品类目": "产品类目", } source_map = {} for key, expected in source_mapping.items(): found = False for col in df.columns: if col == expected: source_map[key] = col found = True break if not found: source_map[key] = None # ========== 默认值映射 ========== default_mapping = { "Variant Type": ("fixed", "Not Variant"), "Minimum Order Quantity": ("fixed", 1), "Force Quantity Multiplier": ("fixed", 1), "Display Set Quantity": ("fixed", 1), "Ship Type": ("valid", "Ship Type"), "Lead Time": ("fixed", 48), "Replacement Lead Time": ("fixed", 120), "Flat Pack": ("fixed", "Yes"), "Assembly Required": ("fixed", "Yes"), "Canada Product Restriction": ("fixed", "Yes"), "CARB Phase II Compliant (formaldehyde emissions)": ("fixed", "Yes"), "Commercial Warranty": ("fixed", "Yes"), "CANFER Compliant": ("fixed", "Yes"), "Commercial Warranty Length": ("fixed", "1 Years"), "Composite Wood Product (CWP)": ("fixed", "Yes"), "Country Of Manufacturer": ("fixed", "China"), "General Certificate of Conformity (GCC)": ("fixed", "Yes"), "Hazard Class(es)": ("fixed", "Does Not Apply"), "Hazardous Material / Dangerous Goods": ("fixed", "No"), "Hazardous Material Weight": ("fixed", "Does Not Apply"), "Level of Assembly": ("fixed", "Full Assembly Needed"), "Packing Group": ("fixed", "Does Not Apply"), "Reason for Restriction": ("fixed", "Does Not Apply"), "Supplier Intended and Approved Use": ("fixed", "Residential Use"), "TSCA Title VI Compliant (formaldehyde emissions)": ("fixed", "Yes"), "UN or ID number": ("fixed", "Does Not Apply"), "Uniform Packaging and Labeling Regulations (UPLR) Compliant": ("fixed", "Yes"), "Warning Required": ("fixed", "No"), "Warranty Length": ("fixed", "1 Years"), "Battery or Batteries Included": ("fixed", "No"), "Product Type": ("source", "产品类目"), "Brand": ("valid", "Brand"), } # ========== 填充数据 ========== def get_key(row): item = row.get("Item Code", "") name = row.get("产品名称", "") return f"{item}_{name}" keys = [get_key(row) for _, row in df.iterrows()] key_count = Counter(keys) key_index = Counter() row_num = last_row filled_count = 0 for idx, (_, src_row) in enumerate(df.iterrows()): item_code = src_row.get("Item Code", "") product_name = src_row.get("产品名称", "") if pd.isna(item_code): item_code = "" if pd.isna(product_name): product_name = "" key = get_key(src_row) if key_count[key] > 1: key_index[key] += 1 suffix = f".{key_index[key] - 1}" mfr_suffix = f".{key_index[key]}" else: suffix = "" mfr_suffix = "" fill_count = 0 # Supplier Part Number if "Supplier Part Number" in col_indices: val = f"{item_code}{product_name}{suffix}" ws.cell(row=row_num, column=col_indices["Supplier Part Number"], value=val) fill_count += 1 # Manufacturer Part Number if "Manufacturer Part Number" in col_indices: val = f"{item_code}{product_name}{mfr_suffix}" ws.cell(row=row_num, column=col_indices["Manufacturer Part Number"], value=val) fill_count += 1 # Base Cost if "Base Cost" in col_indices and source_map.get("优惠单价"): col = source_map["优惠单价"] val = src_row.get(col, "") if pd.notna(val): ws.cell(row=row_num, column=col_indices["Base Cost"], value=val) fill_count += 1 # MSRP if "Manufacturer Suggested Retail Price" in col_indices and source_map.get("优惠单价"): col = source_map["优惠单价"] val = src_row.get(col, 0) try: num = float(val) if pd.notna(val) else 0 ws.cell(row=row_num, column=col_indices["Manufacturer Suggested Retail Price"], value=num + 70) fill_count += 1 except: pass # Product Weight if "Product Weight" in col_indices and source_map.get("产品重量"): col = source_map["产品重量"] val = src_row.get(col, "") if pd.notna(val): ws.cell(row=row_num, column=col_indices["Product Weight"], value=val) fill_count += 1 # Carton carton_map = [ ("Carton Weight 1", "包装尺寸-重量"), ("Carton Height 1", "包装尺寸-高度"), ("Carton Width 1", "包装尺寸-宽度"), ("Carton Depth 1", "包装尺寸-长度"), ] for target, src_name in carton_map: if target in col_indices and source_map.get(src_name): col = source_map[src_name] val = src_row.get(col, "") if pd.notna(val): ws.cell(row=row_num, column=col_indices[target], value=val) fill_count += 1 # Color/Material cm_map = [ ("Base Color", "颜色"), ("Base Material", "材质"), ("Color", "颜色"), ("Material", "材质"), ("Top Color", "颜色"), ("Top Material", "材质"), ] for target, src_name in cm_map: if target in col_indices and source_map.get(src_name): col = source_map[src_name] val = src_row.get(col, "") if pd.notna(val): ws.cell(row=row_num, column=col_indices[target], value=val) fill_count += 1 # Overall overall_map = [ ("Overall Depth - Front to Back", "组装长度"), ("Overall Height - Top to Bottom", "组装高度"), ("Overall Product Weight", "产品重量"), ("Overall Width - Side to Side", "组装宽度"), ] for target, src_name in overall_map: if target in col_indices and source_map.get(src_name): col = source_map[src_name] val = src_row.get(col, "") if pd.notna(val): ws.cell(row=row_num, column=col_indices[target], value=val) fill_count += 1 # 默认值字段 variant_type = "Not Variant" for target_col, (source_type, source_val) in default_mapping.items(): if target_col not in col_indices: continue col_idx = col_indices[target_col] if source_type == "fixed": ws.cell(row=row_num, column=col_idx, value=source_val) fill_count += 1 elif source_type == "valid": if source_val in valid_values and valid_values[source_val]: val = valid_values[source_val][0] ws.cell(row=row_num, column=col_idx, value=val) fill_count += 1 else: ws.cell(row=row_num, column=col_idx, value="Not Variant") fill_count += 1 elif source_type == "source": if source_map.get(source_val): col = source_map[source_val] val = src_row.get(col, "") if pd.notna(val): ws.cell(row=row_num, column=col_idx, value=val) fill_count += 1 # Variant Grouping 特殊逻辑 if "Variant Grouping 1" in col_indices: if variant_type != "Not Variant": ws.cell(row=row_num, column=col_indices["Variant Grouping 1"], value="Color") fill_count += 1 if "Variant Attribute Name On Site 1" in col_indices: if variant_type != "Not Variant": ws.cell(row=row_num, column=col_indices["Variant Attribute Name On Site 1"], value="Finish") fill_count += 1 if "Variant Grouping 2" in col_indices: if variant_type != "Not Variant": ws.cell(row=row_num, column=col_indices["Variant Grouping 2"], value="Size") fill_count += 1 if "Variant Attribute Name On Site 2" in col_indices: if variant_type != "Not Variant": ws.cell(row=row_num, column=col_indices["Variant Attribute Name On Site 2"], value="Dimensions") fill_count += 1 row_num += 1 filled_count += 1 # ========== 保存输出文件 ========== output_filename = f"output_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" output_path = os.path.join(OUTPUT_FOLDER, output_filename) wb.save(output_path) # 清理上传文件 try: os.remove(source_path) except: pass return jsonify({ 'success': True, 'filled_count': filled_count, 'download_url': f'/download/{output_filename}', 'message': f'成功追加 {filled_count} 行数据' }) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/download/') def download(filename): """下载生成的文件""" file_path = os.path.join(OUTPUT_FOLDER, filename) if not os.path.exists(file_path): return '文件不存在', 404 return send_file(file_path, as_attachment=True, download_name=filename) if __name__ == '__main__': # 确保有默认模板 default_template = os.path.join(TEMPLATES_FOLDER, '模板.xlsx') if not os.path.exists(default_template): # 创建空模板提示 print(f'⚠️ 请将模板文件放入 {TEMPLATES_FOLDER} 目录') app.run(host='0.0.0.0', port=5000, debug=True)