Agent/examples/content_finder/memory/search_history.py

"""
搜索历史记录管理
"""

from typing import List, Dict, Any, Optional
from datetime import datetime
import json
from pathlib import Path


class SearchHistoryManager:
    """搜索历史记录管理器"""

    def __init__(self, storage_path: str):
        self.storage_path = Path(storage_path)
        self.storage_path.mkdir(parents=True, exist_ok=True)
        self.history_file = self.storage_path / "search_history.jsonl"

    async def save_search(
        self,
        keywords: List[str],
        platforms: List[str],
        filters: Dict[str, Any],
        results_count: int,
        trace_id: str,
    ) -> None:
        """保存搜索记录"""
        record = {
            "timestamp": datetime.now().isoformat(),
            "keywords": keywords,
            "platforms": platforms,
            "filters": filters,
            "results_count": results_count,
            "trace_id": trace_id,
        }

        with open(self.history_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")

    async def get_recent_searches(
        self,
        limit: int = 50,
        keyword_filter: Optional[str] = None,
    ) -> List[Dict[str, Any]]:
        """获取最近的搜索记录"""
        if not self.history_file.exists():
            return []

        records = []
        with open(self.history_file, "r", encoding="utf-8") as f:
            for line in f:
                if line.strip():
                    record = json.loads(line)
                    if keyword_filter:
                        if keyword_filter in record.get("keywords", []):
                            records.append(record)
                    else:
                        records.append(record)

        # 返回最近的记录
        return records[-limit:]

    async def get_similar_searches(
        self,
        keywords: List[str],
        limit: int = 10,
    ) -> List[Dict[str, Any]]:
        """获取相似的搜索记录"""
        all_searches = await self.get_recent_searches(limit=1000)

        # 计算相似度
        scored_searches = []
        for search in all_searches:
            search_keywords = set(search.get("keywords", []))
            input_keywords = set(keywords)
            similarity = len(search_keywords & input_keywords) / len(search_keywords | input_keywords)
            if similarity > 0:
                scored_searches.append((similarity, search))

        # 按相似度排序
        scored_searches.sort(key=lambda x: x[0], reverse=True)

        return [search for _, search in scored_searches[:limit]]