kg_agent/knowledge_v2/clear_cache.py

"""
清除缓存工具
"""

import os
import sys
import shutil

# 添加路径
current_dir = os.path.dirname(os.path.abspath(__file__))
root_dir = os.path.dirname(current_dir)
sys.path.insert(0, root_dir)

from knowledge_v2.cache_manager import CacheManager

def clear_all_cache():
    """清除所有缓存"""
    cache = CacheManager()
    
    if os.path.exists(cache.base_cache_dir):
        print(f"缓存目录: {cache.base_cache_dir}")
        
        # 统计信息
        total_size = 0
        file_count = 0
        for root, dirs, files in os.walk(cache.base_cache_dir):
            for file in files:
                file_path = os.path.join(root, file)
                total_size += os.path.getsize(file_path)
                file_count += 1
        
        print(f"文件数量: {file_count}")
        print(f"总大小: {total_size / 1024:.2f} KB")
        
        # 确认删除
        response = input("\n确认清除所有缓存？(yes/no): ")
        if response.lower() == 'yes':
            shutil.rmtree(cache.base_cache_dir)
            os.makedirs(cache.base_cache_dir)
            print("✓ 已清除所有缓存")
        else:
            print("取消操作")
    else:
        print("缓存目录不存在")

def clear_question_cache(question: str):
    """清除特定问题的缓存"""
    cache = CacheManager()
    
    import hashlib
    question_hash = hashlib.md5(question.encode('utf-8')).hexdigest()[:12]
    question_dir = os.path.join(cache.base_cache_dir, question_hash)
    
    if os.path.exists(question_dir):
        # 显示问题文本
        question_file = os.path.join(question_dir, 'question.txt')
        if os.path.exists(question_file):
            with open(question_file, 'r', encoding='utf-8') as f:
                cached_question = f.read()
            print(f"找到缓存的问题: {cached_question}")
        
        # 确认删除
        response = input("\n确认清除此问题的缓存？(yes/no): ")
        if response.lower() == 'yes':
            shutil.rmtree(question_dir)
            print("✓ 已清除该问题的缓存")
        else:
            print("取消操作")
    else:
        print("未找到该问题的缓存")

def list_cached_questions():
    """列出所有缓存的问题"""
    cache = CacheManager()
    
    if not os.path.exists(cache.base_cache_dir):
        print("缓存目录不存在")
        return
    
    print("缓存的问题列表:")
    print("=" * 60)
    
    count = 0
    for dir_name in os.listdir(cache.base_cache_dir):
        question_dir = os.path.join(cache.base_cache_dir, dir_name)
        if os.path.isdir(question_dir):
            question_file = os.path.join(question_dir, 'question.txt')
            if os.path.exists(question_file):
                with open(question_file, 'r', encoding='utf-8') as f:
                    question = f.read()
                count += 1
                print(f"{count}. [{dir_name}] {question[:50]}...")
    
    print("=" * 60)
    print(f"总计: {count} 个缓存问题")

if __name__ == "__main__":
    print("缓存管理工具")
    print("=" * 60)
    print("1. 列出所有缓存")
    print("2. 清除所有缓存")
    print("3. 清除特定问题的缓存")
    print("=" * 60)
    
    choice = input("请选择操作 (1/2/3): ")
    
    if choice == "1":
        list_cached_questions()
    elif choice == "2":
        clear_all_cache()
    elif choice == "3":
        question = input("请输入问题的完整文本（需要与原问题完全一致）: ")
        clear_question_cache(question)
    else:
        print("无效的选择")