""" Phase 2 标准化功能测试 """ import tempfile import shutil from pathlib import Path from datetime import datetime import pytest from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker from app.core.database import Base from app.models.raw_orders import RawOrder from app.models.normalized_orders import OrderItemNorm, BrandAlias, ColorAlias, SizeAlias from app.services.alias_service import AliasService from app.services.normalization_service import NormalizationService from app.utils.text_parser import BrandExtractor, ProductCodeExtractor, AttributeExtractor, generate_sku_id, slug from app.utils.image_processor import ImageProcessor @pytest.fixture async def test_db(): """创建测试数据库""" engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) async_session = async_sessionmaker(engine, expire_on_commit=False) async with async_session() as session: yield session await engine.dispose() @pytest.fixture async def sample_raw_orders(test_db): """创建示例原始订单数据""" raw_orders = [ RawOrder( id=1, file_name="test1.xlsx", file_hash="hash1", row_idx=2, 原始订单编号="ORDER001", 网店名称="测试店铺", 交易状态="交易成功", 付款时间=datetime(2025, 1, 15, 10, 30, 0), 线上宝贝名称="Nike 运动鞋 AJ1234 Low", 线上销售属性="颜色:黑色;尺码:42", 线上商家编码="NK001", 商品编号="SP001", SKU编号="SKU001", 图片="http://img1.jpg,http://img2.jpg", 数量=1, 订单单价=599.0 ), RawOrder( id=2, file_name="test1.xlsx", file_hash="hash1", row_idx=3, 原始订单编号="ORDER002", 网店名称="测试店铺", 交易状态="交易成功", 付款时间=datetime(2025, 1, 16, 14, 20, 0), 线上宝贝名称="Adidas 外套 2025款", 线上销售属性="颜色:蓝色;尺码:XL", 线上商家编码="AD001", 商品编号="SP002", SKU编号="SKU002", 图片="http://img3.jpg;http://img4.jpg", 数量=1, 订单单价=899.0 ) ] for order in raw_orders: test_db.add(order) await test_db.commit() return raw_orders def test_slug_function(): """测试 slug 函数""" assert slug("Nike Air Max") == "nike_air_max" assert slug("ADIDAS-SHOES") == "adidas_shoes" assert slug("Product 123!@#") == "product_123" assert slug(" Space Test ") == "space_test" assert slug("") == "" def test_brand_extractor(): """测试品牌提取器""" aliases = { "nike": "NIKE", "adidas": "ADIDAS", "puma": "PUMA" } extractor = BrandExtractor(aliases) # 测试别名匹配 assert extractor.extract_brand("Nike 运动鞋") == "NIKE" assert extractor.extract_brand("Adidas外套") == "ADIDAS" # 测试首词提取 assert extractor.extract_brand("Puma 背包") == "PUMA" # 测试括号提取 assert extractor.extract_brand("运动鞋（Nike）经典款") == "NIKE" # 测试未匹配情况 brand = extractor.extract_brand("UnknownBrand 商品") assert brand == "UnknownBrand" def test_product_code_extractor(): """测试货号提取器""" extractor = ProductCodeExtractor() # 测试基本提取 assert extractor.extract_product_code("Nike AJ1234 Low") == "AJ1234" assert extractor.extract_product_code("Adidas BOOST350 V2") == "BOOST350" # 测试带品牌过滤 assert extractor.extract_product_code("Nike AJ1234 Low", "Nike") == "AJ1234" # 测试数字开头 assert extractor.extract_product_code("商品 123ABC 款式") == "123ABC" def test_attribute_extractor(): """测试属性提取器""" color_aliases = {"黑": "黑色", "白": "白色", "blue": "蓝色"} size_aliases = {"L": "L", "XL": "XL", "42": "42"} extractor = AttributeExtractor(color_aliases, size_aliases) # 测试结构化属性提取 attrs = extractor.extract_attributes( "Nike 运动鞋", "颜色:黑色;尺码:42" ) assert attrs["color"] == "黑色" assert attrs["size"] == "42" # 测试别名匹配 attrs = extractor.extract_attributes( "商品名称 黑 L码", "" ) assert attrs["color"] == "黑色" assert attrs["size"] == "L" def test_generate_sku_id(): """测试 SKU ID 生成""" sku_id = generate_sku_id("NK001", "SP001", "SKU001", "黑色", "42") assert "nk001" in sku_id assert "黑色" in sku_id or "hei_se" in sku_id assert "42" in sku_id # 测试空值处理 sku_id = generate_sku_id(None, None, None, "红色", "L") assert "unknown" in sku_id def test_image_processor(): """测试图片处理器""" processor = ImageProcessor() # 测试分隔符处理 images = processor.process_image_links("http://img1.jpg,http://img2.jpg") assert len(images) == 2 assert "http://img1.jpg" in images # 测试不同分隔符 images = processor.process_image_links("http://img1.jpg;http://img2.jpg|http://img3.jpg") assert len(images) == 3 # 测试无效URL过滤 images = processor.process_image_links("http://img1.jpg,invalid_url,http://img2.jpg") assert len(images) == 2 # 测试JSON序列化 json_str = processor.serialize_images(["http://img1.jpg", "http://img2.jpg"]) assert "http://img1.jpg" in json_str # 测试JSON反序列化 images = processor.deserialize_images(json_str) assert len(images) == 2 @pytest.mark.asyncio async def test_alias_service(test_db): """测试别名服务""" alias_service = AliasService() # 测试添加品牌别名 brand_alias = await alias_service.add_brand_alias( test_db, "nike", "NIKE", 100, auto_confirm=True ) assert brand_alias.alias == "nike" assert brand_alias.canonical_brand == "NIKE" assert brand_alias.is_confirmed == "confirmed" # 测试加载别名映射 aliases = await alias_service.load_brand_aliases(test_db) assert aliases["nike"] == "NIKE" # 测试颜色别名 await alias_service.add_color_alias(test_db, "黑", "黑色", 100) color_aliases = await alias_service.load_color_aliases(test_db) assert color_aliases["黑"] == "黑色" # 测试尺码别名 await alias_service.add_size_alias(test_db, "L", "L", 100) size_aliases = await alias_service.load_size_aliases(test_db) assert size_aliases["L"] == "L" @pytest.mark.asyncio async def test_normalization_service(test_db, sample_raw_orders): """测试标准化服务""" # 首先添加一些别名数据 alias_service = AliasService() await alias_service.add_brand_alias(test_db, "Nike", "NIKE", 100, True) await alias_service.add_brand_alias(test_db, "Adidas", "ADIDAS", 100, True) await alias_service.add_color_alias(test_db, "黑色", "黑色", 100) await alias_service.add_color_alias(test_db, "蓝色", "蓝色", 100) await alias_service.add_size_alias(test_db, "42", "42", 100) await alias_service.add_size_alias(test_db, "XL", "XL", 100) # 测试标准化服务 norm_service = NormalizationService() result = await norm_service.normalize_raw_orders(test_db) assert result["total_raw_orders"] == 2 assert result["processed"] == 2 assert result["created"] == 2 assert result["failed"] == 0 # 验证标准化结果 from sqlalchemy import select result = await test_db.execute(select(OrderItemNorm)) norm_orders = result.scalars().all() assert len(norm_orders) == 2 # 检查第一个订单的标准化结果 nike_order = next(order for order in norm_orders if "Nike" in order.线上宝贝名称) assert nike_order.品牌 == "NIKE" assert nike_order.货号 == "AJ1234" assert nike_order.颜色 == "黑色" assert nike_order.尺寸 == "42" assert nike_order.sku_id is not None assert nike_order.图片数量 == 2 # 检查第二个订单 adidas_order = next(order for order in norm_orders if "Adidas" in order.线上宝贝名称) assert adidas_order.品牌 == "ADIDAS" assert adidas_order.颜色 == "蓝色" assert adidas_order.尺寸 == "XL" @pytest.mark.asyncio async def test_normalization_deduplication(test_db): """测试标准化去重功能""" # 创建两个相同产品键的原始订单 raw_order1 = RawOrder( id=10, file_name="test.xlsx", file_hash="hash1", row_idx=2, 原始订单编号="ORDER001", 线上商家编码="NK001", SKU编号="SKU001", 商品编号="SP001", 订单单价=599.0, 数量=1, 线上宝贝名称="Nike 运动鞋", 交易状态="交易成功" ) raw_order2 = RawOrder( id=11, file_name="test.xlsx", file_hash="hash1", row_idx=3, 原始订单编号="ORDER001", # 相同的产品键字段 线上商家编码="NK001", SKU编号="SKU001", 商品编号="SP001", 订单单价=599.0, 数量=1, 线上宝贝名称="Nike 运动鞋", 交易状态="交易关闭", # 不同的可变字段 卖家备注="更新的备注" ) test_db.add_all([raw_order1, raw_order2]) await test_db.commit() # 运行标准化 norm_service = NormalizationService() result = await norm_service.normalize_raw_orders(test_db) # 应该只创建一个标准化记录 from sqlalchemy import select result = await test_db.execute(select(OrderItemNorm)) norm_orders = result.scalars().all() assert len(norm_orders) == 1 # 第二次处理应该更新可变字段 result2 = await norm_service.normalize_raw_orders(test_db) assert result2["updated"] >= 1 @pytest.mark.asyncio async def test_brand_alias_suggestions(test_db): """测试品牌别名建议""" # 添加一些现有品牌 alias_service = AliasService() await alias_service.add_brand_alias(test_db, "Nike", "NIKE", 100, True) await alias_service.add_brand_alias(test_db, "Adidas", "ADIDAS", 100, True) # 创建包含相似品牌的订单 raw_order = RawOrder( id=20, file_name="test.xlsx", file_hash="hash1", row_idx=2, 原始订单编号="ORDER003", 线上宝贝名称="nike 运动鞋", # 小写，应该匹配到 NIKE 数量=1, 订单单价=599.0 ) test_db.add(raw_order) await test_db.commit() # 运行标准化 norm_service = NormalizationService() await norm_service.normalize_raw_orders(test_db) # 检查是否正确匹配品牌 from sqlalchemy import select result = await test_db.execute( select(OrderItemNorm).where(OrderItemNorm.原始订单编号 == "ORDER003") ) norm_order = result.scalar_one() assert norm_order.品牌 == "NIKE" if __name__ == '__main__': pytest.main([__file__, '-v'])