from flask import Flask, render_template, request, jsonify, send_file import pandas as pd from openpyxl import load_workbook import os import shutil import requests import json import re import time from datetime import datetime 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) # ========== AI 配置 ========== AI_CONFIG = { 'api_url': 'https://ark.cn-beijing.volces.com/api/v3/responses', 'authorization': 'Bearer ark-f7930e50-4837-4f8f-96b4-bd2f56313300-b9dd7', 'vision_model': 'doubao-seed-2-1-pro-260628', 'extract_model': 'deepseek-v4-pro-260425', } # ========== AI 属性映射 ========== AI_FIELD_MAPPING = { 'Table Top Shape': 'Table Top Shape', 'Coffee Table Lift Top': 'Coffee Table Lift Top', 'Upholstered': 'Upholstered', 'Plug-In': 'Plug-In', 'Storage Included': 'Storage Included', 'Shelves Included': 'Shelves Included', 'Drawers Included': 'Drawers Included', 'Cabinets Included': 'Cabinets Included', 'Decal/Laminate Design': 'Decal/Laminate Design', 'Set Type': 'Set Type', 'Durability': 'Durability', 'Number of Tables Included': 'Number of Tables Included', 'CAL TB 117-2013 Compliant': 'CAL TB 117-2013 Compliant', 'SOFFA Compliant': 'SOFFA Compliant', } # ========== 全局日志存储 ========== ai_logs = [] def add_ai_log(message, level='info'): """添加AI日志""" timestamp = datetime.now().strftime('%H:%M:%S') log_entry = { 'time': timestamp, 'message': message, 'level': level } ai_logs.append(log_entry) print(f"[AI] {timestamp} - {message}") # 只保留最近200条 if len(ai_logs) > 200: ai_logs.pop(0) def get_ai_logs(): """获取AI日志""" return ai_logs def clear_ai_logs(): """清空AI日志""" global ai_logs ai_logs = [] # ========== 模板配置 ========== TEMPLATE_CONFIG = { 'wayfair': { 'file': 'wayfair模板.xlsx', 'source_header': 2, 'template_header': 4, 'name': 'Wayfair' }, 'amazon': { 'file': '亚马逊模板.xlsx', 'source_header': 1, 'template_header': 1, 'name': '亚马逊' } } def load_valid_values(wb): valid_data = {} if "Valid Values" not in wb.sheetnames: return valid_data ws_valid = wb["Valid Values"] 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 get_valid_value(valid_values, field_name, default=None): if field_name in valid_values and valid_values[field_name]: return valid_values[field_name][0] return default def call_vision_api(image_url, row_index): """调用豆包视觉模型识别图片（带详细日志）""" add_ai_log(f"🖼️ [第{row_index}行] 开始识别图片: {image_url[:60]}...", 'info') headers = { 'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate, br', 'Authorization': AI_CONFIG['authorization'], 'Connection': 'keep-alive', 'Content-Type': 'application/json', 'User-Agent': 'PostmanRuntime-ApipostRuntime/1.1.0' } data = { "model": AI_CONFIG['vision_model'], "input": [ { "role": "user", "content": [ { "type": "input_image", "image_url": image_url }, { "type": "input_text", "text": "请详细描述这张图片中的家具，包括：形状、颜色、材质、结构、功能特点、适用场景。描述要详细具体。" } ] } ] } add_ai_log(f"📤 [第{row_index}行] 调用视觉模型 ({AI_CONFIG['vision_model']})...", 'info') start_time = time.time() try: response = requests.post( AI_CONFIG['api_url'], headers=headers, json=data, timeout=120 ) elapsed = time.time() - start_time add_ai_log(f"⏱️ [第{row_index}行] 视觉识别耗时: {elapsed:.1f}秒", 'info') result = response.json() if 'output' in result: for item in result['output']: if item.get('type') == 'message' and 'content' in item: for content in item['content']: if content.get('type') == 'output_text': text = content.get('text', '') add_ai_log(f"✅ [第{row_index}行] 视觉识别成功，描述长度: {len(text)} 字符", 'success') return text add_ai_log(f"⚠️ [第{row_index}行] 视觉识别返回格式异常", 'warn') return None except requests.exceptions.Timeout: add_ai_log(f"❌ [第{row_index}行] 视觉识别超时 (120秒)", 'error') return None except Exception as e: add_ai_log(f"❌ [第{row_index}行] 视觉识别失败: {str(e)}", 'error') return None def call_extract_api(description, row_index): """调用DeepSeek提取属性（带详细日志）""" add_ai_log(f"🧠 [第{row_index}行] 开始属性提取...", 'info') headers = { 'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate, br', 'Authorization': AI_CONFIG['authorization'], 'Connection': 'keep-alive', 'Content-Type': 'application/json', 'User-Agent': 'PostmanRuntime-ApipostRuntime/1.1.0' } fields = [ 'Table Top Shape', 'Coffee Table Lift Top', 'Upholstered', 'Plug-In', 'Storage Included', 'Shelves Included', 'Drawers Included', 'Cabinets Included', 'Decal/Laminate Design', 'Set Type', 'Durability', 'Number of Tables Included', 'CAL TB 117-2013 Compliant', 'SOFFA Compliant' ] fields_str = ', '.join(fields) data = { "model": AI_CONFIG['extract_model'], "stream": False, "tools": [ { "type": "web_search", "max_keyword": 3 } ], "input": [ { "role": "user", "content": [ { "type": "input_text", "text": f"""请根据下面的家具描述，填写以下信息：{fields_str}。 返回格式要求：每个属性用方括号括起来，格式为 [属性名:值] 例如：[Table Top Shape:Rectangle with rounded corners] [Coffee Table Lift Top:No] 如果描述中没有相关信息，请填写 "Unknown"。 家具描述： {description}""" } ] } ] } add_ai_log(f"📤 [第{row_index}行] 调用提取模型 ({AI_CONFIG['extract_model']})...", 'info') start_time = time.time() try: response = requests.post( AI_CONFIG['api_url'], headers=headers, json=data, timeout=120 ) elapsed = time.time() - start_time add_ai_log(f"⏱️ [第{row_index}行] 属性提取耗时: {elapsed:.1f}秒", 'info') result = response.json() if 'output' in result: for item in result['output']: if item.get('type') == 'message' and 'content' in item: for content in item['content']: if content.get('type') == 'output_text': text = content.get('text', '') add_ai_log(f"✅ [第{row_index}行] 属性提取成功，结果长度: {len(text)} 字符", 'success') return text add_ai_log(f"⚠️ [第{row_index}行] 属性提取返回格式异常", 'warn') return None except requests.exceptions.Timeout: add_ai_log(f"❌ [第{row_index}行] 属性提取超时 (120秒)", 'error') return None except Exception as e: add_ai_log(f"❌ [第{row_index}行] 属性提取失败: {str(e)}", 'error') return None def parse_ai_result(text): """解析AI返回的 [属性:值] 格式""" result = {} if not text: return result pattern = r'\[([^:\]]+):([^\]]+)\]' matches = re.findall(pattern, text) for field, value in matches: field = field.strip() value = value.strip() if value and value not in ['Unknown', 'Not Applicable', 'N/A', '']: result[field] = value return result def get_ai_filled_data(image_url, row_index): """完整的AI填写流程（带详细日志）""" add_ai_log(f"🚀 [第{row_index}行] 开始AI处理", 'info') add_ai_log(f"📷 [第{row_index}行] 图片URL: {image_url[:80]}...", 'info') # 第一步：识别图片 description = call_vision_api(image_url, row_index) if not description: add_ai_log(f"⚠️ [第{row_index}行] 图片识别失败，跳过该行", 'warn') return {} # 显示描述前100字符 desc_preview = description[:100] + "..." if len(description) > 100 else description add_ai_log(f"📝 [第{row_index}行] 图片描述预览: {desc_preview}", 'info') # 第二步：提取属性 extract_result = call_extract_api(description, row_index) if not extract_result: add_ai_log(f"⚠️ [第{row_index}行] 属性提取失败，跳过该行", 'warn') return {} # 第三步：解析结果 parsed = parse_ai_result(extract_result) if parsed: add_ai_log(f"📋 [第{row_index}行] 解析成功，提取到 {len(parsed)} 个字段:", 'success') for field, value in parsed.items(): add_ai_log(f" └─ {field}: {value}", 'info') else: add_ai_log(f"⚠️ [第{row_index}行] 解析结果为空", 'warn') return parsed def fill_amazon_row(src_row, source_map, col_indices, valid_values, row_num, ws, suffix, mfr_suffix, ai_data=None): fill_count = 0 item_code = "" product_name = "" product_name_en = "" try: item_code = src_row.get("Item Code", "") product_name = src_row.get("产品名称", "") product_name_en = src_row.get("产品英文名称", "") if pd.isna(item_code): item_code = "" if pd.isna(product_name): product_name = "" if pd.isna(product_name_en): product_name_en = "" # SKU 相关 sku_value = f"{item_code}{product_name_en}{suffix}" if "SKU" in col_indices: ws.cell(row=row_num, column=col_indices["SKU"], value=sku_value) fill_count += 1 if "Model Number" in col_indices: ws.cell(row=row_num, column=col_indices["Model Number"], value=sku_value) fill_count += 1 if "Model Name" in col_indices: ws.cell(row=row_num, column=col_indices["Model Name"], value=sku_value) fill_count += 1 if "Part Number" in col_indices: ws.cell(row=row_num, column=col_indices["Part Number"], value=sku_value) fill_count += 1 if "Set Name" in col_indices: ws.cell(row=row_num, column=col_indices["Set Name"], value=product_name_en if product_name_en else product_name) fill_count += 1 # 从来源映射 field_map = { "Color": "颜色", "Base Color": "颜色", "Top Color": "颜色", "Frame Material": "材质", "Base Material": "材质", "Top Material": "材质", "Furniture Leg Material": "材质", "Upholstery Fabric Type": "材质", "Item Length": "组装长度", "Item Width": "组装宽度", "Item Height": "组装高度", "Item Weight": "产品重量", "Item Package Length": "包装尺寸-长度", "Item Package Width": "包装尺寸-宽度", "Item Package Height": "包装尺寸-高度", "Package Weight": "包装尺寸-重量", "List Price": "优惠单价", "Your Price USD": "优惠单价", "Unit Count": "包装尺寸-重量", "Item Length Longer Edge": "组装长度", "Item Width Shorter Edge": "组装宽度", "Length longer edge when assembled": "组装长度", "Width shorter edge when assembled": "组装宽度", "Height base to top when assembled": "组装高度", "Item Depth Front To Back": "组装宽度", "Item Height Floor To Top": "组装高度", "Item Width Side To Side": "组装长度", "Working Surface Length": "组装长度", "Working Surface Width": "组装宽度", "Country of Origin": "原产地", } source_col_map = { "颜色": "颜色", "材质": "材质", "组装长度": "组装长度(英寸)", "组装宽度": "组装宽度(英寸)", "组装高度": "组装高度(英寸)", "产品重量": "产品重量(磅)", "包装尺寸-长度": "包装尺寸-长度(英寸)", "包装尺寸-宽度": "包装尺寸-宽度(英寸)", "包装尺寸-高度": "包装尺寸-高度(英寸)", "包装尺寸-重量": "包装尺寸-重量(磅)", "优惠单价": "优惠单价", "产品名称": "产品名称", "产品英文名称": "产品英文名称", "Item Code": "Item Code", "店铺名称": "店铺名称", "原产地": "原产地", "产品主图": "产品主图", "上架图片1": "上架图片1", "上架图片2": "上架图片2", "上架图片3": "上架图片3", "上架图片4": "上架图片4", "上架图片5": "上架图片5", "上架图片6": "上架图片6", "上架图片7": "上架图片7", "上架图片8": "上架图片8", "产品描述": "产品描述", "产品特点1": "产品特点1", "产品特点2": "产品特点2", "产品特点3": "产品特点3", "产品特点4": "产品特点4", "产品特点5": "产品特点5", } for amazon_field, source_key in field_map.items(): if amazon_field not in col_indices: continue col_idx = col_indices[amazon_field] source_col_name = source_col_map.get(source_key, source_key) source_col = source_map.get(source_key, source_col_name) if source_col and source_col in src_row: val = src_row[source_col] if pd.notna(val): if amazon_field == "Country of Origin": val = str(val).strip() if val in ["CHN", "CN"]: val = "China" ws.cell(row=row_num, column=col_idx, value=val) fill_count += 1 # AI 填充 if ai_data: for ai_field, ai_value in ai_data.items(): target_field = AI_FIELD_MAPPING.get(ai_field) if target_field and target_field in col_indices: ws.cell(row=row_num, column=col_indices[target_field], value=ai_value) fill_count += 1 # 图片 image_fields = [ ("Main Image URL", "产品主图"), ("Other Image URL", "上架图片1"), ("Other Image URL", "上架图片2"), ("Other Image URL", "上架图片3"), ("Other Image URL", "上架图片4"), ("Other Image URL", "上架图片5"), ("Other Image URL", "上架图片6"), ("Other Image URL", "上架图片7"), ("Other Image URL", "上架图片8"), ] for field_name, source_key in image_fields: if field_name not in col_indices: continue col_idx = col_indices[field_name] source_col = source_col_map.get(source_key, source_key) if source_col in src_row: val = src_row[source_col] if pd.notna(val) and str(val).strip(): ws.cell(row=row_num, column=col_idx, value=val) fill_count += 1 # 固定值 fixed_values = { "Product Type": "TABLE", "Listing Action": "Create or Replace (Full Update)", "Package Level": "Unit", "Package Contains SKU Quantity": 1, "Number of Items": 1, "Item Package Quantity": 1, "Unit Count Type": "pound", "Item Thickness Decimal Value": 0.6, "Item Length Unit": "Inches", "Item Width Unit": "Inches", "Item Height Unit": "Inches", "Item Package Length Unit": "Inches", "Item Package Width Unit": "Inches", "Item Package Height Unit": "Inches", "Package Weight Unit": "Pounds", "Item Weight Unit": "Pounds", "Recommended Number of People for Assembly": 2, "Has Finished Back": "TRUE", "Includes All Assembly Tools": "Yes", "Wood Type": "Character", "Number of Players": 2, "Is Fragile?": "No", "Load Capacity Unit": "Pound", "Surface Texture": "wood grain", "Maximum Order Quantity": 1, "Handling Time (US)": 3, "Number of Boxes": 1, "Is This Product Subject To Buyer Age Restrictions": "No", "Safety Attestation": "Yes", "Ships Globally": "Yes", "Are batteries required?": "No", "Are batteries included?": "No", "Is OEM Sourced Product": "Yes", "Is Customizable?": "No", "Is Foldable": "No", "Is Stain Resistant": "Yes", "Tilting": "No", "Offering Can Be Gift Messaged": "Yes", "Is Gift Wrap Available": "Yes", "Accessories": "Generic", "Packaging": "OEM Original", "Fulfillment Channel Code (US)": "DEFAULT", "Inventory Always Available (US)": "Disabled", } for field_name, fixed_val in fixed_values.items(): if field_name in col_indices: ws.cell(row=row_num, column=col_indices[field_name], value=fixed_val) fill_count += 1 # Valid Values valid_fields = { "Product Type": "Product Type", "Listing Action": "Listing Action", "Brand Name": "Brand Name", "Product Id Type": "Product Id Type", "Item Type Keyword": "Item Type Keyword", "Package Level": "Package Level", "Warranty Type": "Warranty Type", "Theme": "Theme", "Frame Joint Type": "Frame Joint Type", "Assembly Instructions": "Assembly Instructions", "Specific Uses for Product": "Specific Uses for Product", "Room Type": "Room Type", "Maximum Weight Recommendation Unit": "Maximum Weight Recommendation Unit", "Leg Style": "Leg Style", "Indoor Outdoor Usage": "Indoor Outdoor Usage", "Table Design": "Table Design", "Furniture Base Movement": "Furniture Base Movement", "Item Condition": "Item Condition", "Natural Variation Type": "Natural Variation Type", "Includes All Assembly Tools": "Includes All Assembly Tools", "Dangerous Goods Regulations": "Dangerous Goods Regulations", "Warranty Description": "Warranty Description", } for field_name, valid_key in valid_fields.items(): if field_name in col_indices: val = get_valid_value(valid_values, valid_key) if val: ws.cell(row=row_num, column=col_indices[field_name], value=val) fill_count += 1 # Care Instructions care_vals = valid_values.get("Care Instructions", ["Wipe with Dry Cloth", "Wipe with Damp Cloth"]) for i in range(5): field = f"Care Instructions {i+1}" if i > 0 else "Care Instructions" if field in col_indices: if i % 2 == 0: val = care_vals[0] if care_vals else "Wipe with Dry Cloth" else: val = care_vals[1] if len(care_vals) > 1 else "Wipe with Damp Cloth" ws.cell(row=row_num, column=col_indices[field], value=val) fill_count += 1 # Theme theme_val = get_valid_value(valid_values, "Theme", "Space") for i in range(5): field = f"Theme {i+1}" if i > 0 else "Theme" if field in col_indices: ws.cell(row=row_num, column=col_indices[field], value=theme_val) fill_count += 1 # Warranty Description warranty_val = get_valid_value(valid_values, "Warranty Description", "1 Year Manufacturer") for i in range(5): field = f"Warranty Description {i+1}" if i > 0 else "Warranty Description" if field in col_indices: ws.cell(row=row_num, column=col_indices[field], value=warranty_val) fill_count += 1 # Dangerous Goods Regulations dgr_val = get_valid_value(valid_values, "Dangerous Goods Regulations", "Not Applicable") for i in range(5): field = f"Dangerous Goods Regulations {i+1}" if i > 0 else "Dangerous Goods Regulations" if field in col_indices: ws.cell(row=row_num, column=col_indices[field], value=dgr_val) fill_count += 1 # Tools tools = ["Hammer", "Screw Driver"] if "Tools Recommended For Assembly" in col_indices: for i, tool in enumerate(tools): if i == 0: field = "Tools Recommended For Assembly" else: field = f"Tools Recommended For Assembly {i+1}" if field in col_indices: ws.cell(row=row_num, column=col_indices[field], value=tool) fill_count += 1 # Components components = ["Installation Tool", "Assembly Guide"] if "Included Components" in col_indices: for i, comp in enumerate(components): if i == 0: field = "Included Components" else: field = f"Included Components {i+1}" if field in col_indices: ws.cell(row=row_num, column=col_indices[field], value=comp) fill_count += 1 except Exception as e: print(f"fill_amazon_row error: {e}") return fill_count @app.route('/') def index(): return render_template('index.html') @app.route('/get_config') def get_config(): return jsonify({ 'templates': [ {'key': 'wayfair', 'name': 'Wayfair'}, {'key': 'amazon', 'name': '亚马逊'} ], 'config': TEMPLATE_CONFIG }) @app.route('/get_ai_logs') def get_ai_logs_api(): """获取AI日志API""" return jsonify({ 'logs': ai_logs[-50:] # 返回最近50条 }) @app.route('/clear_ai_logs') def clear_ai_logs_api(): """清空AI日志""" clear_ai_logs() return jsonify({'success': True}) @app.route('/preview', methods=['POST']) def preview(): try: source_file = request.files.get('source_file') template_key = request.form.get('template_key') if not source_file: return jsonify({'error': '请选择来源文件'}), 400 if not template_key or template_key not in TEMPLATE_CONFIG: return jsonify({'error': '请选择有效的模板'}), 400 config = TEMPLATE_CONFIG[template_key] source_header = config['source_header'] 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) 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, 'config': config }) 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_key = request.form.get('template_key') use_ai = request.form.get('use_ai', 'false').lower() == 'true' if not source_path or not os.path.exists(source_path): return jsonify({'error': '来源文件不存在'}), 400 if not template_key or template_key not in TEMPLATE_CONFIG: return jsonify({'error': '请选择有效的模板'}), 400 # 清空之前的AI日志 clear_ai_logs() add_ai_log(f"🚀 开始生成任务，模板: {template_key}，AI: {'启用' if use_ai else '关闭'}", 'info') config = TEMPLATE_CONFIG[template_key] template_name = config['file'] source_header = config['source_header'] template_header_row = config['template_header'] template_path = os.path.join(TEMPLATES_FOLDER, template_name) if not os.path.exists(template_path): return jsonify({'error': f'模板文件不存在: {template_name}'}), 400 # 读取来源数据 add_ai_log(f"📖 读取来源数据: {os.path.basename(source_path)}", 'info') df = pd.read_excel(source_path, header=source_header - 1) total_rows = len(df) add_ai_log(f"📊 共 {total_rows} 行数据", 'info') # AI处理 ai_results = {} if use_ai and '产品主图' in df.columns: add_ai_log(f"🤖 开始AI识别，共 {total_rows} 行需要处理", 'info') # 统计有图片的行 has_image_count = 0 for idx, row in df.iterrows(): img_url = row.get('产品主图', '') if pd.notna(img_url) and str(img_url).strip(): has_image_count += 1 add_ai_log(f"📷 其中 {has_image_count} 行包含产品主图", 'info') processed = 0 success_count = 0 for idx, row in df.iterrows(): img_url = row.get('产品主图', '') if pd.notna(img_url) and str(img_url).strip(): processed += 1 add_ai_log(f"🔄 [第{idx+1}/{total_rows}行] 处理中 ({processed}/{has_image_count})...", 'info') ai_data = get_ai_filled_data(str(img_url).strip(), idx + 1) if ai_data: ai_results[idx] = ai_data success_count += 1 add_ai_log(f"✅ [第{idx+1}行] AI识别完成，提取 {len(ai_data)} 个字段", 'success') # 每处理5行输出一次汇总 if processed % 5 == 0: add_ai_log(f"📊 进度: {processed}/{has_image_count} 行已处理，成功 {success_count} 行", 'info') add_ai_log(f"🎉 AI处理完成！共处理 {processed} 行，成功 {success_count} 行", 'success') elif use_ai and '产品主图' not in df.columns: add_ai_log(f"⚠️ 来源数据中没有 '产品主图' 列，AI功能将跳过", 'warn') else: add_ai_log(f"⏭️ AI功能未启用", 'info') # 加载模板 add_ai_log(f"📖 加载模板: {template_name}", 'info') wb = load_workbook(template_path) ws = wb.active valid_values = load_valid_values(wb) add_ai_log(f"📋 Valid Values 加载完成，共 {len(valid_values)} 个字段", 'info') # 读取模板表头 col_indices = {} for col in range(1, ws.max_column + 1): cell_value = ws.cell(row=template_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 add_ai_log(f"📋 模板共 {len(col_indices)} 个列", 'info') # 找到已有数据的最后一行 start_row = 2 last_row = start_row for row in range(start_row, ws.max_row + 2): is_empty = True for col in range(1, min(10, 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 add_ai_log(f"📌 已有数据到第 {last_row - 1} 行，从第 {last_row} 行开始追加", 'info') # 匹配来源列 source_col_map = { "颜色": "颜色", "材质": "材质", "组装长度": "组装长度(英寸)", "组装宽度": "组装宽度(英寸)", "组装高度": "组装高度(英寸)", "产品重量": "产品重量(磅)", "包装尺寸-长度": "包装尺寸-长度(英寸)", "包装尺寸-宽度": "包装尺寸-宽度(英寸)", "包装尺寸-高度": "包装尺寸-高度(英寸)", "包装尺寸-重量": "包装尺寸-重量(磅)", "优惠单价": "优惠单价", "产品名称": "产品名称", "产品英文名称": "产品英文名称", "Item Code": "Item Code", "店铺名称": "店铺名称", "原产地": "原产地", "产品主图": "产品主图", "上架图片1": "上架图片1", "上架图片2": "上架图片2", "上架图片3": "上架图片3", "上架图片4": "上架图片4", "上架图片5": "上架图片5", "上架图片6": "上架图片6", "上架图片7": "上架图片7", "上架图片8": "上架图片8", "产品描述": "产品描述", "产品特点1": "产品特点1", "产品特点2": "产品特点2", "产品特点3": "产品特点3", "产品特点4": "产品特点4", "产品特点5": "产品特点5", } source_map = {} matched_count = 0 for key, expected in source_col_map.items(): found = False for col in df.columns: if col == expected: source_map[key] = col found = True matched_count += 1 break if not found: source_map[key] = None add_ai_log(f"🔗 来源列匹配: {matched_count}/{len(source_col_map)} 列匹配成功", 'info') # 填充数据 def get_key(row): item = row.get("Item Code", "") name = row.get("产品英文名称", "") if pd.isna(name) or not name: 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 add_ai_log(f"✍️ 开始填充数据，共 {total_rows} 行", 'info') for idx, (_, src_row) in enumerate(df.iterrows()): 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 = "" is_parent = (idx == 0) or (key_index[key] == 1) # Parentage Level if "Parentage Level" in col_indices: if not is_parent: ws.cell(row=row_num, column=col_indices["Parentage Level"], value="Child") # Parent SKU if "Parent SKU" in col_indices: if not is_parent: first_item = df.iloc[0].get("Item Code", "") first_name = df.iloc[0].get("产品英文名称", "") if pd.isna(first_name) or not first_name: first_name = df.iloc[0].get("产品名称", "") first_sku = f"{first_item}{first_name}" ws.cell(row=row_num, column=col_indices["Parent SKU"], value=first_sku) # Variation Theme Name if "Variation Theme Name" in col_indices: if key_count[key] > 1: ws.cell(row=row_num, column=col_indices["Variation Theme Name"], value="COLOR") # 获取AI数据 ai_data = ai_results.get(idx, {}) if use_ai else {} # 填充 fill_count = fill_amazon_row( src_row, source_map, col_indices, valid_values, row_num, ws, suffix, mfr_suffix, ai_data ) row_num += 1 filled_count += 1 # 每10行输出一次进度 if (idx + 1) % 10 == 0: add_ai_log(f"📊 填充进度: {idx+1}/{total_rows} 行", 'info') add_ai_log(f"✅ 数据填充完成，共 {filled_count} 行", 'success') # 保存 output_filename = f"output_{template_key}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" output_path = os.path.join(OUTPUT_FOLDER, output_filename) wb.save(output_path) add_ai_log(f"💾 文件保存成功: {output_filename}", 'success') try: os.remove(source_path) except: pass ai_info = f"，AI识别 {len(ai_results)} 行" if use_ai and ai_results else "" return jsonify({ 'success': True, 'filled_count': filled_count, 'download_url': f'/download/{output_filename}', 'message': f'成功追加 {filled_count} 行数据到 {config["name"]} 模板{ai_info}', 'ai_used': use_ai, 'ai_rows': len(ai_results) if ai_results else 0, 'logs': ai_logs[-30:] # 返回最近30条日志 }) except Exception as e: import traceback error_detail = traceback.format_exc() add_ai_log(f"❌ 生成失败: {str(e)}", 'error') print(error_detail) return jsonify({'error': str(e), 'detail': error_detail}), 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__': for key, config in TEMPLATE_CONFIG.items(): template_path = os.path.join(TEMPLATES_FOLDER, config['file']) if not os.path.exists(template_path): print(f'⚠️ 模板文件不存在: {template_path}') app.run(host='0.0.0.0', port=5002, debug=True)