from __future__ import annotations import importlib.util import io import json import logging import sys import types from pathlib import Path from unittest import mock def _load_product_enrich_module(): if "dotenv" not in sys.modules: fake_dotenv = types.ModuleType("dotenv") fake_dotenv.load_dotenv = lambda *args, **kwargs: None sys.modules["dotenv"] = fake_dotenv if "redis" not in sys.modules: fake_redis = types.ModuleType("redis") class _FakeRedisClient: def __init__(self, *args, **kwargs): pass def ping(self): return True fake_redis.Redis = _FakeRedisClient sys.modules["redis"] = fake_redis repo_root = Path(__file__).resolve().parents[1] if str(repo_root) not in sys.path: sys.path.insert(0, str(repo_root)) module_path = repo_root / "indexer" / "product_enrich.py" spec = importlib.util.spec_from_file_location("product_enrich_under_test", module_path) module = importlib.util.module_from_spec(spec) assert spec and spec.loader spec.loader.exec_module(module) return module product_enrich = _load_product_enrich_module() def _attach_stream(logger_obj: logging.Logger): stream = io.StringIO() handler = logging.StreamHandler(stream) handler.setFormatter(logging.Formatter("%(message)s")) logger_obj.addHandler(handler) return stream, handler def test_create_prompt_splits_shared_context_and_localized_tail(): products = [ {"id": "1", "title": "dress"}, {"id": "2", "title": "linen shirt"}, ] shared_zh, user_zh, prefix_zh = product_enrich.create_prompt(products, target_lang="zh") shared_en, user_en, prefix_en = product_enrich.create_prompt(products, target_lang="en") assert shared_zh == shared_en assert "Analyze each input product text" in shared_zh assert "1. dress" in shared_zh assert "2. linen shirt" in shared_zh assert "Product list" not in user_zh assert "Product list" not in user_en assert "specified language" in user_zh assert "Language: Chinese" in user_zh assert "Language: English" in user_en assert prefix_zh.startswith("| 序号 | 商品标题 | 品类路径 |") assert prefix_en.startswith("| No. | Product title | Category path |") def test_create_prompt_supports_taxonomy_analysis_kind(): products = [{"id": "1", "title": "linen dress"}] shared_zh, user_zh, prefix_zh = product_enrich.create_prompt( products, target_lang="zh", analysis_kind="taxonomy", ) shared_fr, user_fr, prefix_fr = product_enrich.create_prompt( products, target_lang="fr", analysis_kind="taxonomy", ) assert "apparel attribute taxonomy" in shared_zh assert "1. linen dress" in shared_zh assert "Language: Chinese" in user_zh assert "Language: French" in user_fr assert prefix_zh.startswith("| 序号 | 品类 | 目标性别 |") assert prefix_fr.startswith("| No. | Product Type | Target Gender |") def test_call_llm_logs_shared_context_once_and_verbose_contains_full_requests(): payloads = [] response_bodies = [ { "choices": [ { "message": { "content": ( "| 1 | 连衣裙 | 女装>连衣裙 | 法式,收腰 | 年轻女性 | " "通勤,约会 | 春季,夏季 | 中长款 | 聚酯纤维 | 透气 | " "修身显瘦 | 法式收腰连衣裙 |\n" ) } } ], "usage": {"prompt_tokens": 120, "completion_tokens": 45, "total_tokens": 165}, }, { "choices": [ { "message": { "content": ( "| 1 | Dress | Women>Dress | French,Waisted | Young women | " "Commute,Date | Spring,Summer | Midi | Polyester | Breathable | " "Slim fit | French waisted dress |\n" ) } } ], "usage": {"prompt_tokens": 118, "completion_tokens": 43, "total_tokens": 161}, }, ] class _FakeResponse: def __init__(self, body): self.body = body def raise_for_status(self): return None def json(self): return self.body class _FakeSession: trust_env = True def post(self, url, headers=None, json=None, timeout=None, proxies=None): del url, headers, timeout, proxies payloads.append(json) return _FakeResponse(response_bodies[len(payloads) - 1]) def close(self): return None product_enrich.reset_logged_shared_context_keys() main_stream, main_handler = _attach_stream(product_enrich.logger) verbose_stream, verbose_handler = _attach_stream(product_enrich.verbose_logger) try: with mock.patch.object(product_enrich, "API_KEY", "fake-key"), mock.patch.object( product_enrich.requests, "Session", lambda: _FakeSession(), ): zh_shared, zh_user, zh_prefix = product_enrich.create_prompt( [{"id": "1", "title": "dress"}], target_lang="zh", ) en_shared, en_user, en_prefix = product_enrich.create_prompt( [{"id": "1", "title": "dress"}], target_lang="en", ) zh_markdown, zh_raw = product_enrich.call_llm( zh_shared, zh_user, zh_prefix, target_lang="zh", ) en_markdown, en_raw = product_enrich.call_llm( en_shared, en_user, en_prefix, target_lang="en", ) finally: product_enrich.logger.removeHandler(main_handler) product_enrich.verbose_logger.removeHandler(verbose_handler) assert zh_shared == en_shared assert len(payloads) == 2 assert len(payloads[0]["messages"]) == 3 assert payloads[0]["messages"][1]["role"] == "user" assert "1. dress" in payloads[0]["messages"][1]["content"] assert "Language: Chinese" in payloads[0]["messages"][1]["content"] assert "Language: English" in payloads[1]["messages"][1]["content"] assert payloads[0]["messages"][-1]["partial"] is True assert payloads[1]["messages"][-1]["partial"] is True main_log = main_stream.getvalue() verbose_log = verbose_stream.getvalue() assert main_log.count("LLM Shared Context") == 1 assert main_log.count("LLM Request Variant") == 2 assert "Localized Requirement" in main_log assert "Shared Context" in main_log assert verbose_log.count("LLM Request [model=") == 2 assert verbose_log.count("LLM Response [model=") == 2 assert '"partial": true' in verbose_log assert "Combined User Prompt" in verbose_log assert "French waisted dress" in verbose_log assert "法式收腰连衣裙" in verbose_log assert zh_markdown.startswith(zh_prefix) assert en_markdown.startswith(en_prefix) assert json.loads(zh_raw)["usage"]["total_tokens"] == 165 assert json.loads(en_raw)["usage"]["total_tokens"] == 161 def test_process_batch_reads_result_and_validates_expected_fields(): merged_markdown = """| 序号 | 商品标题 | 品类路径 | 细分标签 | 适用人群 | 使用场景 | 适用季节 | 关键属性 | 材质说明 | 功能特点 | 锚文本 | |----|----|----|----|----|----|----|----|----|----|----| | 1 | 法式连衣裙 | 女装>连衣裙 | 法式,收腰 | 年轻女性 | 通勤,约会 | 春季,夏季 | 中长款 | 聚酯纤维 | 透气 | 法式收腰连衣裙 | """ with mock.patch.object( product_enrich, "call_llm", return_value=(merged_markdown, json.dumps({"choices": [{"message": {"content": "stub"}}]})), ): results = product_enrich.process_batch( [{"id": "sku-1", "title": "dress"}], batch_num=1, target_lang="zh", ) assert len(results) == 1 row = results[0] assert row["id"] == "sku-1" assert row["lang"] == "zh" assert row["title_input"] == "dress" assert row["title"] == "法式连衣裙" assert row["category_path"] == "女装>连衣裙" assert row["tags"] == "法式,收腰" assert row["target_audience"] == "年轻女性" assert row["usage_scene"] == "通勤,约会" assert row["season"] == "春季,夏季" assert row["key_attributes"] == "中长款" assert row["material"] == "聚酯纤维" assert row["features"] == "透气" assert row["anchor_text"] == "法式收腰连衣裙" def test_process_batch_reads_taxonomy_result_with_schema_specific_fields(): merged_markdown = """| 序号 | 品类 | 目标性别 | 年龄段 | 适用季节 | 版型 | 廓形 | 领型 | 袖长类型 | 袖型 | 肩带设计 | 腰型 | 裤型 | 裙型 | 长度类型 | 闭合方式 | 设计细节 | 面料 | 成分 | 面料特性 | 服装特征 | 功能 | 主颜色 | 色系 | 印花 / 图案 | 适用场景 | 风格 | |----|----|----|----|----|----|----|----|----|----|----|----|----|----|----|----|----|----|----|----|----|----|----|----|----|----|----| | 1 | 连衣裙 | 女 | 成人 | 春季,夏季 | 修身 | A字 | V领 | 无袖 | | 细肩带 | 高腰 | | A字裙 | 中长款 | 拉链 | 褶皱 | 梭织 | 聚酯纤维,氨纶 | 轻薄,透气 | 有内衬 | 易打理 | 酒红色 | 红色 | 纯色 | 约会,度假 | 浪漫 | """ with mock.patch.object( product_enrich, "call_llm", return_value=(merged_markdown, json.dumps({"choices": [{"message": {"content": "stub"}}]})), ): results = product_enrich.process_batch( [{"id": "sku-1", "title": "dress"}], batch_num=1, target_lang="zh", analysis_kind="taxonomy", ) assert len(results) == 1 row = results[0] assert row["id"] == "sku-1" assert row["lang"] == "zh" assert row["title_input"] == "dress" assert row["product_type"] == "连衣裙" assert row["target_gender"] == "女" assert row["age_group"] == "成人" assert row["sleeve_length_type"] == "无袖" assert row["material_composition"] == "聚酯纤维,氨纶" assert row["occasion_end_use"] == "约会,度假" assert row["style_aesthetic"] == "浪漫" def test_analyze_products_uses_product_level_cache_across_batch_requests(): cache_store = {} process_calls = [] def _cache_key(product, target_lang): return ( target_lang, product.get("title", ""), product.get("brief", ""), product.get("description", ""), product.get("image_url", ""), ) def fake_get_cached_analysis_result(product, target_lang, analysis_kind="content"): assert analysis_kind == "content" return cache_store.get(_cache_key(product, target_lang)) def fake_set_cached_analysis_result(product, target_lang, result, analysis_kind="content"): assert analysis_kind == "content" cache_store[_cache_key(product, target_lang)] = result def fake_process_batch(batch_data, batch_num, target_lang="zh", analysis_kind="content"): assert analysis_kind == "content" process_calls.append( { "batch_num": batch_num, "target_lang": target_lang, "titles": [item["title"] for item in batch_data], } ) return [ { "id": item["id"], "lang": target_lang, "title_input": item["title"], "title": f"normalized:{item['title']}", "category_path": "cat", "tags": "tags", "target_audience": "audience", "usage_scene": "scene", "season": "season", "key_attributes": "attrs", "material": "material", "features": "features", "anchor_text": f"anchor:{item['title']}", } for item in batch_data ] products = [ {"id": "1", "title": "dress"}, {"id": "2", "title": "shirt"}, ] with mock.patch.object(product_enrich, "API_KEY", "fake-key"), mock.patch.object( product_enrich, "_get_cached_analysis_result", side_effect=fake_get_cached_analysis_result, ), mock.patch.object( product_enrich, "_set_cached_analysis_result", side_effect=fake_set_cached_analysis_result, ), mock.patch.object( product_enrich, "process_batch", side_effect=fake_process_batch, ): first = product_enrich.analyze_products( [products[0]], target_lang="zh", tenant_id="170", ) second = product_enrich.analyze_products( products, target_lang="zh", tenant_id="999", ) third = product_enrich.analyze_products( products, target_lang="zh", tenant_id="170", ) assert [row["title_input"] for row in first] == ["dress"] assert [row["title_input"] for row in second] == ["dress", "shirt"] assert [row["title_input"] for row in third] == ["dress", "shirt"] assert process_calls == [ {"batch_num": 1, "target_lang": "zh", "titles": ["dress"]}, {"batch_num": 1, "target_lang": "zh", "titles": ["shirt"]}, ] assert second[0]["anchor_text"] == "anchor:dress" assert second[1]["anchor_text"] == "anchor:shirt" assert third[0]["anchor_text"] == "anchor:dress" assert third[1]["anchor_text"] == "anchor:shirt" def test_analyze_products_reuses_cached_content_with_current_product_identity(): cached_result = { "id": "1165", "lang": "zh", "title_input": "old-title", "title": "法式连衣裙", "category_path": "女装>连衣裙", "enriched_tags": "法式,收腰", "target_audience": "年轻女性", "usage_scene": "通勤,约会", "season": "春季,夏季", "key_attributes": "中长款", "material": "聚酯纤维", "features": "透气", "anchor_text": "法式收腰连衣裙", } products = [{"id": "69960", "title": "dress"}] with mock.patch.object(product_enrich, "API_KEY", "fake-key"), mock.patch.object( product_enrich, "_get_cached_analysis_result", wraps=lambda product, target_lang, analysis_kind="content": product_enrich._normalize_analysis_result( cached_result, product=product, target_lang=target_lang, schema=product_enrich._get_analysis_schema("content"), ), ), mock.patch.object( product_enrich, "process_batch", side_effect=AssertionError("process_batch should not be called on cache hit"), ): result = product_enrich.analyze_products( products, target_lang="zh", tenant_id="170", ) assert result == [ { "id": "69960", "lang": "zh", "title_input": "dress", "title": "法式连衣裙", "category_path": "女装>连衣裙", "tags": "法式,收腰", "target_audience": "年轻女性", "usage_scene": "通勤,约会", "season": "春季,夏季", "key_attributes": "中长款", "material": "聚酯纤维", "features": "透气", "anchor_text": "法式收腰连衣裙", } ] def test_build_index_content_fields_maps_internal_tags_to_enriched_tags_output(): def fake_analyze_products( products, target_lang="zh", batch_size=None, tenant_id=None, analysis_kind="content", ): if analysis_kind == "taxonomy": return [ { "id": products[0]["id"], "lang": target_lang, "title_input": products[0]["title"], "product_type": f"{target_lang}-dress", "target_gender": f"{target_lang}-women", "age_group": "", "season": f"{target_lang}-summer", "fit": "", "silhouette": "", "neckline": "", "sleeve_length_type": "", "sleeve_style": "", "strap_type": "", "rise_waistline": "", "leg_shape": "", "skirt_shape": "", "length_type": "", "closure_type": "", "design_details": "", "fabric": "", "material_composition": "", "fabric_properties": "", "clothing_features": "", "functional_benefits": "", "color": "", "color_family": "", "print_pattern": "", "occasion_end_use": "", "style_aesthetic": "", } ] return [ { "id": products[0]["id"], "lang": target_lang, "title_input": products[0]["title"], "title": products[0]["title"], "category_path": "玩具>滑行玩具", "tags": f"{target_lang}-tag1,{target_lang}-tag2", "target_audience": f"{target_lang}-audience", "usage_scene": "", "season": "", "key_attributes": "", "material": "", "features": "", "anchor_text": f"{target_lang}-anchor", } ] with mock.patch.object( product_enrich, "analyze_products", side_effect=fake_analyze_products, ): result = product_enrich.build_index_content_fields( items=[{"spu_id": "69960", "title": "dress"}], tenant_id="170", ) assert result == [ { "id": "69960", "qanchors": {"zh": ["zh-anchor"], "en": ["en-anchor"]}, "enriched_tags": {"zh": ["zh-tag1", "zh-tag2"], "en": ["en-tag1", "en-tag2"]}, "enriched_attributes": [ { "name": "enriched_tags", "value": { "zh": ["zh-tag1", "zh-tag2"], "en": ["en-tag1", "en-tag2"], }, }, {"name": "target_audience", "value": {"zh": ["zh-audience"], "en": ["en-audience"]}}, ], "enriched_taxonomy_attributes": [ { "name": "Product Type", "value": {"zh": ["zh-dress"], "en": ["en-dress"]}, }, { "name": "Target Gender", "value": {"zh": ["zh-women"], "en": ["en-women"]}, }, { "name": "Season", "value": {"zh": ["zh-summer"], "en": ["en-summer"]}, }, ], } ] def test_anchor_cache_key_depends_on_product_input_not_identifiers(): product_a = { "id": "1", "spu_id": "1001", "title": "dress", "brief": "soft cotton", "description": "summer dress", "image_url": "https://img/a.jpg", } product_b = { "id": "2", "spu_id": "9999", "title": "dress", "brief": "soft cotton", "description": "summer dress", "image_url": "https://img/a.jpg", } product_c = { "id": "1", "spu_id": "1001", "title": "dress", "brief": "soft cotton updated", "description": "summer dress", "image_url": "https://img/a.jpg", } key_a = product_enrich._make_anchor_cache_key(product_a, "zh") key_b = product_enrich._make_anchor_cache_key(product_b, "zh") key_c = product_enrich._make_anchor_cache_key(product_c, "zh") assert key_a == key_b assert key_a != key_c def test_build_prompt_input_text_appends_brief_and_description_for_short_title(): product = { "title": "T恤", "brief": "夏季透气纯棉短袖，舒适亲肤", "description": "100%棉，圆领版型，适合日常通勤与休闲穿搭。", } text = product_enrich._build_prompt_input_text(product) assert text.startswith("T恤") assert "夏季透气纯棉短袖" in text assert "100%棉" in text def test_build_prompt_input_text_truncates_non_cjk_by_words(): product = { "title": "dress", "brief": " ".join(f"brief{i}" for i in range(50)), "description": " ".join(f"desc{i}" for i in range(50)), } text = product_enrich._build_prompt_input_text(product) assert len(text.split()) <= product_enrich.PROMPT_INPUT_MAX_WORDS