kg_agent/script/search/test_xiaohongshu_search_api.py

#!/usr/bin/env python3
"""
小红书搜索API测试脚本
用于测试和调试 xhs_note_search API 接口
"""

import requests
import json
import time
import sys
from datetime import datetime
from typing import Dict, Any, List, Tuple
from xiaohongshu_search import XiaohongshuSearch


# 颜色输出
class Colors:
    """终端颜色输出"""
    GREEN = '\033[92m'
    RED = '\033[91m'
    YELLOW = '\033[93m'
    BLUE = '\033[94m'
    CYAN = '\033[96m'
    RESET = '\033[0m'
    BOLD = '\033[1m'


class TestResult:
    """测试结果记录"""
    def __init__(self):
        self.total = 0
        self.passed = 0
        self.failed = 0
        self.errors = []
        self.details = []


class XiaohongshuAPITester:
    """小红书API测试器"""

    BASE_URL = "http://47.84.182.56:8001"
    API_ENDPOINT = "/tools/call/xhs_note_search"

    def __init__(self, verbose: bool = True):
        """
        初始化测试器

        Args:
            verbose: 是否输出详细日志
        """
        self.verbose = verbose
        self.result = TestResult()
        self.client = XiaohongshuSearch()

    def log(self, message: str, color: str = Colors.RESET):
        """输出日志"""
        if self.verbose:
            print(f"{color}{message}{Colors.RESET}")

    def log_section(self, title: str):
        """输出章节标题"""
        print(f"\n{Colors.BOLD}{Colors.CYAN}{'='*70}{Colors.RESET}")
        print(f"{Colors.BOLD}{Colors.CYAN}{title}{Colors.RESET}")
        print(f"{Colors.BOLD}{Colors.CYAN}{'='*70}{Colors.RESET}\n")

    def test_raw_api_request(
        self,
        keyword: str,
        content_type: str = "不限",
        sort_type: str = "综合",
        publish_time: str = "不限",
        cursor: str = "",
        timeout: int = 30
    ) -> Tuple[bool, Dict[str, Any], float]:
        """
        直接测试原始API请求（不使用封装类）

        Returns:
            (是否成功, 响应数据, 响应时间)
        """
        url = f"{self.BASE_URL}{self.API_ENDPOINT}"
        payload = {
            "keyword": keyword,
            "content_type": content_type,
            "sort_type": sort_type,
            "publish_time": publish_time,
            "cursor": cursor
        }

        self.log(f"\n{Colors.YELLOW}[请求详情]{Colors.RESET}")
        self.log(f"URL: {url}")
        self.log(f"Method: POST")
        self.log(f"Headers: {{'Content-Type': 'application/json'}}")
        self.log(f"Payload: {json.dumps(payload, ensure_ascii=False, indent=2)}")

        start_time = time.time()
        try:
            response = requests.post(
                url,
                json=payload,
                timeout=timeout,
                headers={"Content-Type": "application/json"}
            )
            elapsed = time.time() - start_time

            self.log(f"\n{Colors.YELLOW}[响应详情]{Colors.RESET}")
            self.log(f"状态码: {response.status_code}")
            self.log(f"响应时间: {elapsed:.2f}秒")
            self.log(f"响应头: {dict(response.headers)}")

            # 尝试解析JSON
            try:
                result = response.json()
                self.log(f"响应体预览: {json.dumps(result, ensure_ascii=False, indent=2)[:500]}...")

                # 检查HTTP状态码
                if response.status_code == 200:
                    return True, result, elapsed
                else:
                    self.log(f"{Colors.RED}HTTP错误: {response.status_code}{Colors.RESET}")
                    return False, {"error": f"HTTP {response.status_code}", "detail": result}, elapsed

            except json.JSONDecodeError as e:
                self.log(f"{Colors.RED}JSON解析失败: {e}{Colors.RESET}")
                self.log(f"原始响应: {response.text[:500]}")
                return False, {"error": "JSON解析失败", "detail": str(e), "raw": response.text}, elapsed

        except requests.exceptions.Timeout:
            elapsed = time.time() - start_time
            self.log(f"{Colors.RED}请求超时（{timeout}秒）{Colors.RESET}")
            return False, {"error": "Timeout", "timeout": timeout}, elapsed

        except requests.exceptions.ConnectionError as e:
            elapsed = time.time() - start_time
            self.log(f"{Colors.RED}连接错误: {e}{Colors.RESET}")
            return False, {"error": "ConnectionError", "detail": str(e)}, elapsed

        except requests.exceptions.RequestException as e:
            elapsed = time.time() - start_time
            self.log(f"{Colors.RED}请求异常: {e}{Colors.RESET}")
            return False, {"error": "RequestException", "detail": str(e)}, elapsed

    def run_test(self, test_name: str, test_func, *args, **kwargs) -> bool:
        """
        运行单个测试

        Returns:
            是否通过
        """
        self.result.total += 1
        self.log(f"\n{Colors.BLUE}[测试 {self.result.total}] {test_name}{Colors.RESET}")

        try:
            success, data, elapsed = test_func(*args, **kwargs)

            if success:
                self.result.passed += 1
                self.log(f"{Colors.GREEN}✓ 通过{Colors.RESET} (耗时: {elapsed:.2f}秒)")
                self.result.details.append({
                    "name": test_name,
                    "status": "PASS",
                    "elapsed": elapsed
                })
                return True
            else:
                self.result.failed += 1
                error_info = {
                    "name": test_name,
                    "error": data.get("error", "Unknown"),
                    "detail": data.get("detail", "")
                }
                self.result.errors.append(error_info)
                self.result.details.append({
                    "name": test_name,
                    "status": "FAIL",
                    "elapsed": elapsed,
                    "error": error_info
                })
                self.log(f"{Colors.RED}✗ 失败{Colors.RESET}")
                self.log(f"{Colors.RED}错误: {error_info}{Colors.RESET}")
                return False

        except Exception as e:
            self.result.failed += 1
            error_info = {
                "name": test_name,
                "error": "Exception",
                "detail": str(e)
            }
            self.result.errors.append(error_info)
            self.result.details.append({
                "name": test_name,
                "status": "ERROR",
                "error": error_info
            })
            self.log(f"{Colors.RED}✗ 异常: {e}{Colors.RESET}")
            return False

    def test_basic_keywords(self):
        """测试基础关键词搜索"""
        self.log_section("1. 基础关键词测试")

        # 测试简单关键词
        self.run_test(
            "简单关键词: 美食",
            self.test_raw_api_request,
            keyword="美食"
        )

        # 测试复杂中文关键词
        self.run_test(
            "复杂关键词: 表情包怎么制作",
            self.test_raw_api_request,
            keyword="表情包怎么制作"
        )

        # 测试英文关键词
        self.run_test(
            "英文关键词: food",
            self.test_raw_api_request,
            keyword="food"
        )

        # 测试带特殊符号的关键词
        self.run_test(
            "特殊符号: Python编程#入门",
            self.test_raw_api_request,
            keyword="Python编程#入门"
        )

    def test_content_types(self):
        """测试不同内容类型"""
        self.log_section("2. 内容类型测试")

        keyword = "美食"

        for content_type in ["不限", "视频", "图文"]:
            self.run_test(
                f"内容类型: {content_type}",
                self.test_raw_api_request,
                keyword=keyword,
                content_type=content_type
            )

    def test_sort_types(self):
        """测试不同排序方式"""
        self.log_section("3. 排序方式测试")

        keyword = "旅游"

        for sort_type in ["综合", "最新", "最多点赞", "最多评论"]:
            self.run_test(
                f"排序方式: {sort_type}",
                self.test_raw_api_request,
                keyword=keyword,
                sort_type=sort_type
            )

    def test_publish_times(self):
        """测试不同发布时间筛选"""
        self.log_section("4. 发布时间筛选测试")

        keyword = "健身"

        for publish_time in ["不限", "一天内", "一周内", "半年内"]:
            self.run_test(
                f"发布时间: {publish_time}",
                self.test_raw_api_request,
                keyword=keyword,
                publish_time=publish_time
            )

    def test_edge_cases(self):
        """测试边缘情况"""
        self.log_section("5. 边缘情况测试")

        # 空关键词
        self.run_test(
            "空关键词",
            self.test_raw_api_request,
            keyword=""
        )

        # 超长关键词
        self.run_test(
            "超长关键词（100字符）",
            self.test_raw_api_request,
            keyword="如何" * 50
        )

        # 纯空格
        self.run_test(
            "纯空格关键词",
            self.test_raw_api_request,
            keyword="   "
        )

        # 超时测试（设置1秒超时）
        self.run_test(
            "超时测试（1秒）",
            self.test_raw_api_request,
            keyword="美食",
            timeout=1
        )

    def test_invalid_parameters(self):
        """测试无效参数"""
        self.log_section("6. 无效参数测试")

        # 无效的content_type
        self.run_test(
            "无效content_type: 音频",
            self.test_raw_api_request,
            keyword="音乐",
            content_type="音频"
        )

        # 无效的sort_type
        self.run_test(
            "无效sort_type: 随机",
            self.test_raw_api_request,
            keyword="音乐",
            sort_type="随机"
        )

    def test_using_wrapper_class(self):
        """测试使用封装类"""
        self.log_section("7. 封装类测试")

        def test_wrapper(keyword: str, **kwargs) -> Tuple[bool, Dict, float]:
            """使用XiaohongshuSearch类进行测试"""
            start_time = time.time()
            try:
                result = self.client.search(keyword, **kwargs)
                elapsed = time.time() - start_time
                return True, result, elapsed
            except Exception as e:
                elapsed = time.time() - start_time
                return False, {"error": str(e)}, elapsed

        self.run_test(
            "封装类: 基础搜索",
            test_wrapper,
            keyword="美食"
        )

        self.run_test(
            "封装类: 完整参数",
            test_wrapper,
            keyword="旅游",
            content_type="视频",
            sort_type="最新",
            publish_time="一周内"
        )

    def generate_report(self) -> Dict[str, Any]:
        """生成测试报告"""
        self.log_section("测试报告")

        report = {
            "timestamp": datetime.now().isoformat(),
            "summary": {
                "total": self.result.total,
                "passed": self.result.passed,
                "failed": self.result.failed,
                "pass_rate": f"{(self.result.passed/self.result.total*100):.1f}%" if self.result.total > 0 else "N/A"
            },
            "details": self.result.details,
            "errors": self.result.errors
        }

        # 打印摘要
        print(f"\n{Colors.BOLD}测试摘要:{Colors.RESET}")
        print(f"  总计: {self.result.total}")
        print(f"  {Colors.GREEN}通过: {self.result.passed}{Colors.RESET}")
        print(f"  {Colors.RED}失败: {self.result.failed}{Colors.RESET}")
        print(f"  通过率: {report['summary']['pass_rate']}")

        if self.result.errors:
            print(f"\n{Colors.RED}{Colors.BOLD}失败详情:{Colors.RESET}")
            for i, error in enumerate(self.result.errors, 1):
                print(f"{Colors.RED}  {i}. {error['name']}{Colors.RESET}")
                print(f"     错误: {error['error']}")
                if error.get('detail'):
                    print(f"     详情: {error['detail'][:200]}")

        return report

    def save_report(self, filepath: str = None):
        """保存测试报告"""
        if filepath is None:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filepath = f"test_report_{timestamp}.json"

        report = self.generate_report()

        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(report, f, ensure_ascii=False, indent=2)

        print(f"\n{Colors.GREEN}报告已保存: {filepath}{Colors.RESET}")
        return filepath

    def run_all_tests(self):
        """运行所有测试"""
        print(f"{Colors.BOLD}{Colors.CYAN}")
        print("=" * 70)
        print("小红书搜索API测试")
        print(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print("=" * 70)
        print(Colors.RESET)

        # 运行所有测试
        self.test_basic_keywords()
        self.test_content_types()
        self.test_sort_types()
        self.test_publish_times()
        self.test_edge_cases()
        self.test_invalid_parameters()
        self.test_using_wrapper_class()

        # 生成并保存报告
        return self.save_report()


def run_quick_test(keyword: str = "表情包怎么制作"):
    """
    快速测试单个关键词

    Args:
        keyword: 要测试的关键词
    """
    tester = XiaohongshuAPITester(verbose=True)

    print(f"{Colors.BOLD}{Colors.CYAN}")
    print("=" * 70)
    print(f"快速测试: {keyword}")
    print("=" * 70)
    print(Colors.RESET)

    tester.run_test(
        f"测试关键词: {keyword}",
        tester.test_raw_api_request,
        keyword=keyword
    )

    tester.generate_report()


def main():
    """主函数"""
    import argparse

    parser = argparse.ArgumentParser(description='小红书搜索API测试工具')
    parser.add_argument(
        '--mode',
        type=str,
        choices=['full', 'quick'],
        default='quick',
        help='测试模式: full(完整测试) 或 quick(快速测试)'
    )
    parser.add_argument(
        '--keyword',
        type=str,
        default='表情包怎么制作',
        help='快速测试的关键词'
    )
    parser.add_argument(
        '--verbose',
        action='store_true',
        default=True,
        help='输出详细日志'
    )

    args = parser.parse_args()

    if args.mode == 'full':
        tester = XiaohongshuAPITester(verbose=args.verbose)
        tester.run_all_tests()
    else:
        run_quick_test(args.keyword)


if __name__ == "__main__":
    main()