#!/usr/bin/env python3 """ 生成待采购订单数据 从标准化订单表中生成待采购订单 """ import asyncio import logging from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker from sqlalchemy import text # 设置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) DATABASE_URL = "sqlite+aiosqlite:///./ordersys.db" async def generate_procurement_orders(): """从标准化订单生成待采购订单""" # 切换到后端目录 import os os.chdir('/Users/jinjunqian/PycharmProjects/订单处理新版/ordersys/backend') engine = create_async_engine(DATABASE_URL) async_session = async_sessionmaker(engine, expire_on_commit=False) async with async_session() as session: # 1. 清空现有的待采购订单 logger.info("清空现有待采购订单...") await session.execute(text("DELETE FROM procurement_orders")) await session.commit() # 2. 查询符合采购条件的标准化订单 logger.info("查询符合采购条件的标准化订单...") result = await session.execute(text(""" SELECT id, 原始订单编号, 线上宝贝名称, 线上销售属性, 品牌, 货号, 颜色, 尺寸, 数量, 订单单价, 订单金额, 付款时间, 交易状态, 退款状态, 网店名称 FROM order_items_norm WHERE 付款时间 IS NOT NULL AND (交易状态 = '已付款' OR 交易状态 = '部分发货') AND 退款状态 IN ('正常', '退款中') ORDER BY 付款时间 DESC """)) orders = result.fetchall() logger.info(f"找到 {len(orders)} 个符合条件的订单") if not orders: logger.warning("没有找到符合条件的订单") return # 3. 生成待采购订单 logger.info("生成待采购订单...") created_count = 0 batch_size = 100 for i, order in enumerate(orders): # 计算产品SKU Key (使用产品主表的逻辑) import hashlib sku_components = [ str(order.线上宝贝名称) if order.线上宝贝名称 else "", str(order.线上销售属性) if order.线上销售属性 else "" ] sku_key = '|'.join(sku_components) product_key = hashlib.sha256(sku_key.encode('utf-8')).hexdigest()[:16] # 确定采购方式（使用修正后的逻辑） procurement_method = determine_procurement_method( order.线上宝贝名称 or "", order.线上销售属性 or "", order.品牌 or "" ) # 计算建议采购价 suggested_price = None if order.订单单价: try: suggested_price = float(order.订单单价) * 0.7 except: suggested_price = None # 处理付款时间 payment_time = order.付款时间 if isinstance(payment_time, str): # 如果是字符串，保持原样 payment_time_str = payment_time else: # 如果是datetime，转换为字符串 payment_time_str = payment_time.isoformat() if payment_time else None # 插入待采购订单 await session.execute(text(""" INSERT INTO procurement_orders ( original_order_id, 原始订单编号, 线上宝贝名称, 线上销售属性, 品牌, 货号, 颜色, 尺寸, 数量, 订单单价, 订单金额, 付款时间, 交易状态, 退款状态, 网店名称, procurement_status, procurement_method, product_key, 建议采购价, is_urgent, priority_score ) VALUES ( :original_order_id, :原始订单编号, :线上宝贝名称, :线上销售属性, :品牌, :货号, :颜色, :尺寸, :数量, :订单单价, :订单金额, :付款时间, :交易状态, :退款状态, :网店名称, 'PENDING', :procurement_method, :product_key, :建议采购价, 0, 0 ) """), { 'original_order_id': order.id, '原始订单编号': order.原始订单编号, '线上宝贝名称': order.线上宝贝名称, '线上销售属性': order.线上销售属性, '品牌': order.品牌, '货号': order.货号, '颜色': order.颜色, '尺寸': order.尺寸, '数量': order.数量, '订单单价': order.订单单价, '订单金额': order.订单金额, '付款时间': payment_time_str, '交易状态': order.交易状态, '退款状态': order.退款状态, '网店名称': order.网店名称, 'procurement_method': procurement_method, 'product_key': product_key, '建议采购价': suggested_price }) created_count += 1 # 批量提交 if (i + 1) % batch_size == 0: await session.commit() logger.info(f"已创建 {created_count} 个待采购订单...") # 最终提交 await session.commit() logger.info(f"✅ 待采购订单生成完成！共创建 {created_count} 个订单") # 4. 统计结果 stats_result = await session.execute(text(""" SELECT procurement_method, COUNT(*) as count FROM procurement_orders GROUP BY procurement_method ORDER BY count DESC """)) logger.info("\n📊 采购方式分布:") for row in stats_result: logger.info(f" {row.procurement_method}: {row.count} 个订单") def determine_procurement_method(product_name: str, sales_attributes: str = "", brand: str = "") -> str: """采购方式判断逻辑（与产品主表保持一致）""" import re product_name = product_name.strip() if product_name else "" sales_attributes = sales_attributes.strip() if sales_attributes else "" brand = brand.strip() if brand else "" # 转为大写便于匹配 product_name_upper = product_name.upper() sales_attributes_upper = sales_attributes.upper() brand_upper = brand.upper() # 规则1: MC库存 if (product_name_upper.endswith(' MC') or product_name_upper.endswith('MC') or 'MC' in sales_attributes_upper): return "MC" # 规则2: AT/ATM Atelier if ((product_name_upper.endswith(' AT') or product_name_upper.endswith('AT') or product_name_upper.endswith(' ATM') or product_name_upper.endswith('ATM')) and 'MC' not in sales_attributes_upper): return "AT" # 规则3: AP/APM 昌昌 if ((product_name_upper.endswith(' AP') or product_name_upper.endswith('AP') or product_name_upper.endswith(' APM') or product_name_upper.endswith('APM')) and 'MC' not in sales_attributes_upper): return "AP" # 规则4: LA/LAM - 商品名称以LA/LAM结尾 if (product_name_upper.endswith(' LA') or product_name_upper.endswith('LA') or product_name_upper.endswith(' LAM') or product_name_upper.endswith('LAM')): return "LA" # 规则5: SS/SSM Saks if ((product_name_upper.endswith(' SS') or product_name_upper.endswith('SS') or product_name_upper.endswith(' SSM') or product_name_upper.endswith('SSM')) and 'MC' not in sales_attributes_upper): return "SS" # 规则6: 销售属性有独立的LA suffixes_to_exclude = ['LA', 'MC', 'LAM', 'SS', 'SSM', 'AT', 'ATM', 'AP', 'APM'] has_excluded_suffix = any( product_name_upper.endswith(f' {suffix}') or product_name_upper.endswith(suffix) for suffix in suffixes_to_exclude ) if not has_excluded_suffix: pattern = r'LA(?=_|\s|$|\d|\[)' if re.search(pattern, sales_attributes_upper): return "LA" # 规则7: 国内现货 if ('现货' in product_name or 'GN现货' in product_name or any(brand in brand_upper for brand in ['GRUNDZERO', 'JANSPORT', 'EMU'])): return "GN" # 规则8: 默认NY return "NY" if __name__ == "__main__": asyncio.run(generate_procurement_orders())