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_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_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_anchor_result(product, target_lang):
        return cache_store.get(_cache_key(product, target_lang))

    def fake_set_cached_anchor_result(product, target_lang, result):
        cache_store[_cache_key(product, target_lang)] = result

    def fake_process_batch(batch_data, batch_num, target_lang="zh"):
        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_anchor_result",
        side_effect=fake_get_cached_anchor_result,
    ), mock.patch.object(
        product_enrich,
        "_set_cached_anchor_result",
        side_effect=fake_set_cached_anchor_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_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