#!/usr/bin/env python3 """ 测试产品主表生成功能 验证从原始订单直接生成产品主表的完整流程 """ import asyncio import sys import os from pathlib import Path # 添加项目根目录到 Python 路径 sys.path.append(str(Path(__file__).parent)) from app.core.database import get_db, init_db from app.services.products_master_service import ProductsMasterService from app.services.image_download_service import ImageDownloadService from sqlalchemy import text, select, func from app.models.products_master import ProductMaster, ProductSourceMapping, ImageDownloadLog from app.models.raw_orders import RawOrder async def test_products_master_generation(): """测试产品主表生成功能""" print("=" * 60) print("📦 产品主表生成测试") print("=" * 60) # 1. 初始化数据库 print("\n1. 初始化数据库...") try: await init_db() print("✅ 数据库初始化成功") except Exception as e: print(f"❌ 数据库初始化失败: {e}") return async for db in get_db(): try: # 2. 检查原始订单数据 print("\n2. 检查原始订单数据...") raw_orders_count = await db.execute( select(func.count(RawOrder.id)) .where(RawOrder.原始订单编号.is_not(None)) .where(RawOrder.线上宝贝名称.is_not(None)) ) count = raw_orders_count.scalar() print(f" 原始订单总数: {count}") if count == 0: print("⚠️ 没有原始订单数据，请先导入订单数据") return # 3. 显示示例原始订单 print("\n3. 示例原始订单数据:") sample_orders = await db.execute( select(RawOrder.id, RawOrder.线上宝贝名称, RawOrder.线上销售属性, RawOrder.订单单价, RawOrder.数量) .where(RawOrder.原始订单编号.is_not(None)) .where(RawOrder.线上宝贝名称.is_not(None)) .limit(5) ) for row in sample_orders.fetchall(): print(f" ID: {row.id}") print(f" 商品名称: {str(row.线上宝贝名称)[:50]}...") print(f" 销售属性: {row.线上销售属性}") print(f" 单价: {row.订单单价}, 数量: {row.数量}") print(" " + "-" * 50) # 4. 检查现有产品主表 print("\n4. 检查现有产品主表...") existing_products = await db.execute(select(func.count(ProductMaster.id))) existing_count = existing_products.scalar() print(f" 现有产品数: {existing_count}") # 5. 生成产品主表 print("\n5. 开始生成产品主表...") service = ProductsMasterService() # 小批量测试 result = await service.generate_products_master( db=db, batch_size=50, force_rebuild=False # 增量生成 ) print(f"\n📊 生成结果统计:") print(f" 处理订单数: {result['processed']}") print(f" 创建产品数: {result['created']}") print(f" 更新产品数: {result['updated']}") print(f" 失败订单数: {result['failed']}") print(f" 跳过订单数: {result['skipped']}") print(f" 最终产品总数: {result['products_count']}") if result['errors']: print(f"\n❌ 错误信息:") for error in result['errors'][:5]: # 只显示前5个错误 print(f" • {error}") # 6. 验证生成的产品主表数据 print("\n6. 验证生成的产品数据...") await verify_products_master_data(db) # 7. 测试图片下载调度 print("\n7. 测试图片下载调度...") await test_image_download_scheduling(db) print(f"\n✅ 产品主表生成测试完成!") except Exception as e: print(f"❌ 测试过程中发生错误: {e}") import traceback traceback.print_exc() break async def verify_products_master_data(db): """验证产品主表数据质量""" print(" 验证数据质量...") # 统计各个字段的完整性 stats_query = await db.execute(text(""" SELECT COUNT(*) as total_products, COUNT(品牌) as has_brand, COUNT(货号) as has_product_code, COUNT(颜色) as has_color, COUNT(尺寸) as has_size, COUNT(original_image_urls) as has_images, COUNT(CASE WHEN image_download_status = 'pending' THEN 1 END) as pending_downloads, AVG(total_quantity) as avg_quantity, AVG(avg_price) as avg_price FROM products_master WHERE is_active = true """)) stats = stats_query.fetchone() print(f" 📈 数据质量统计:") print(f" 总产品数: {stats.total_products}") print(f" 有品牌: {stats.has_brand} ({stats.has_brand/stats.total_products*100:.1f}%)") print(f" 有货号: {stats.has_product_code} ({stats.has_product_code/stats.total_products*100:.1f}%)") print(f" 有颜色: {stats.has_color} ({stats.has_color/stats.total_products*100:.1f}%)") print(f" 有尺寸: {stats.has_size} ({stats.has_size/stats.total_products*100:.1f}%)") print(f" 有图片: {stats.has_images} ({stats.has_images/stats.total_products*100:.1f}%)") print(f" 待下载图片: {stats.pending_downloads}") print(f" 平均销量: {stats.avg_quantity:.1f}") print(f" 平均价格: ¥{stats.avg_price:.2f}") # 显示示例产品 print(f"\n 🔍 示例产品数据:") sample_products = await db.execute( select(ProductMaster.id, ProductMaster.线上宝贝名称, ProductMaster.线上销售属性, ProductMaster.品牌, ProductMaster.货号, ProductMaster.颜色, ProductMaster.尺寸, ProductMaster.procurement_method, ProductMaster.total_quantity, ProductMaster.avg_price) .where(ProductMaster.is_active == True) .order_by(ProductMaster.total_quantity.desc()) .limit(3) ) for i, product in enumerate(sample_products.fetchall(), 1): print(f" 产品 {i}:") print(f" 商品名称: {str(product.线上宝贝名称)[:40]}...") print(f" 销售属性: {product.线上销售属性}") print(f" 品牌: {product.品牌}") print(f" 货号: {product.货号}") print(f" 颜色: {product.颜色}") print(f" 尺寸: {product.尺寸}") print(f" 采购方式: {product.procurement_method}") print(f" 销量: {product.total_quantity}, 均价: ¥{product.avg_price:.2f}") print() async def test_image_download_scheduling(db): """测试图片下载调度""" print(" 测试图片下载调度...") # 统计图片下载状态 image_stats = await db.execute(text(""" SELECT image_download_status, COUNT(*) as count FROM products_master WHERE original_image_urls IS NOT NULL GROUP BY image_download_status """)) print(f" 📷 图片下载状态:") for row in image_stats.fetchall(): print(f" {row.image_download_status}: {row.count} 个产品") # 创建图片下载服务 download_service = ImageDownloadService() # 调度小批量图片下载测试 (仅测试5个产品) print(f" 开始小批量图片下载测试...") try: download_result = await download_service.download_product_images( db=db, limit=5 # 仅下载5个产品的图片进行测试 ) print(f" 📊 下载测试结果:") print(f" 处理产品: {download_result['processed_products']}") print(f" 总图片数: {download_result['total_images']}") print(f" 下载成功: {download_result['downloaded_images']}") print(f" 下载失败: {download_result['failed_images']}") if download_result['errors']: print(f" ⚠️ 下载错误 (前3个):") for error in download_result['errors'][:3]: print(f" • {error}") except Exception as e: print(f" ❌ 图片下载测试失败: {e}") async def main(): """主函数""" await test_products_master_generation() if __name__ == "__main__": asyncio.run(main())