kg_agent/test_cache.py

#!/usr/bin/env python3
"""
测试 XiaohongshuSearchRecommendations 缓存功能
"""

import sys
import os
import json
import time

# 添加脚本目录到路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'script', 'search_recommendations'))

from xiaohongshu_search_recommendations import XiaohongshuSearchRecommendations


def print_section(title):
    """打印分隔符"""
    print("\n" + "="*60)
    print(f"  {title}")
    print("="*60)


def test_cache():
    """测试缓存功能"""

    # 创建客户端，设置较短的缓存时间用于测试（60秒）
    print_section("1. 初始化客户端")
    client = XiaohongshuSearchRecommendations(enable_cache=True, cache_ttl=60)
    print(f"✓ 客户端初始化成功")
    print(f"  - 缓存已启用: {client.enable_cache}")
    print(f"  - 缓存有效期: {client.cache_ttl} 秒")
    print(f"  - 结果目录: {client.results_base_dir}")

    # 测试关键词
    test_keyword = "川西"

    # 第一次请求（应该从API获取）
    print_section(f"2. 第一次请求关键词 '{test_keyword}'")
    print("预期: 缓存未命中，从API获取数据")
    start_time = time.time()
    result1 = client.get_recommendations(test_keyword)
    elapsed1 = time.time() - start_time
    print(f"✓ 请求完成，耗时: {elapsed1:.2f} 秒")
    print(f"  - 获取到 {len(result1)} 条推荐词")
    if result1:
        print(f"  - 示例: {result1[:3]}")

    # 保存结果到文件（模拟正常使用流程）
    if result1:
        filepath = client.save_result(test_keyword, result1)
        print(f"✓ 结果已保存到: {filepath}")

    # 查看缓存信息
    print_section("3. 查看缓存信息")
    cache_info = client.get_cache_info(test_keyword)
    print("内存缓存:")
    print(json.dumps(cache_info["memory_cache"], ensure_ascii=False, indent=2))
    print("\n文件缓存:")
    print(json.dumps(cache_info["file_cache"], ensure_ascii=False, indent=2))

    # 第二次请求（应该从内存缓存获取）
    print_section(f"4. 第二次请求关键词 '{test_keyword}'（内存缓存）")
    print("预期: 从内存缓存获取")
    start_time = time.time()
    result2 = client.get_recommendations(test_keyword)
    elapsed2 = time.time() - start_time
    print(f"✓ 请求完成，耗时: {elapsed2:.2f} 秒")
    print(f"  - 获取到 {len(result2)} 条推荐词")
    print(f"  - 速度提升: {(elapsed1/elapsed2):.1f}x")

    # 验证结果一致性
    if result1 == result2:
        print("✓ 缓存数据与原始数据一致")
    else:
        print("✗ 警告: 缓存数据与原始数据不一致")

    # 清除内存缓存，测试文件缓存
    print_section(f"5. 清除内存缓存，测试文件缓存")
    client.clear_memory_cache(test_keyword)
    print("✓ 内存缓存已清除")

    # 第三次请求（应该从文件缓存获取）
    print_section(f"6. 第三次请求关键词 '{test_keyword}'（文件缓存）")
    print("预期: 从文件缓存获取")
    start_time = time.time()
    result3 = client.get_recommendations(test_keyword)
    elapsed3 = time.time() - start_time
    print(f"✓ 请求完成，耗时: {elapsed3:.2f} 秒")
    print(f"  - 获取到 {len(result3)} 条推荐词")

    # 验证结果一致性
    if result1 == result3:
        print("✓ 文件缓存数据与原始数据一致")
    else:
        print("✗ 警告: 文件缓存数据与原始数据不一致")

    # 测试禁用缓存
    print_section(f"7. 测试禁用缓存（use_cache=False）")
    print("预期: 直接从API获取，不使用缓存")
    start_time = time.time()
    result4 = client.get_recommendations(test_keyword, use_cache=False)
    elapsed4 = time.time() - start_time
    print(f"✓ 请求完成，耗时: {elapsed4:.2f} 秒")
    print(f"  - 获取到 {len(result4)} 条推荐词")

    # 查看最终缓存状态
    print_section("8. 最终缓存状态")
    cache_info = client.get_cache_info()
    print("所有内存缓存:")
    for keyword, info in cache_info["memory_cache"].items():
        print(f"  - {keyword}: {info}")
    print("\n所有文件缓存:")
    for keyword, info in cache_info["file_cache"].items():
        print(f"  - {keyword}: {info}")

    print_section("测试完成")
    print("✓ 所有缓存功能测试通过")


if __name__ == "__main__":
    try:
        test_cache()
    except KeyboardInterrupt:
        print("\n\n测试被用户中断")
    except Exception as e:
        print(f"\n\n✗ 测试失败: {e}")
        import traceback
        traceback.print_exc()