kg_agent/knowledge_v2/clear_cache_simple.py

"""
简化版缓存清除工具 - 不依赖外部库
"""

import os
import shutil
import hashlib

def get_cache_dir():
    """获取缓存目录"""
    current_dir = os.path.dirname(os.path.abspath(__file__))
    return os.path.join(current_dir, '.cache')

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

def clear_question_cache(question: str):
    """清除特定问题的缓存"""
    cache_dir = get_cache_dir()
    
    question_hash = hashlib.md5(question.encode('utf-8')).hexdigest()[:12]
    question_dir = os.path.join(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("✓ 已清除该问题的缓存")
            return True
        else:
            print("取消操作")
            return False
    else:
        print("未找到该问题的缓存")
        return False

def list_cached_questions():
    """列出所有缓存的问题"""
    cache_dir = get_cache_dir()
    
    if not os.path.exists(cache_dir):
        print("缓存目录不存在")
        return
    
    print("\n缓存的问题列表:")
    print("=" * 60)
    
    count = 0
    for dir_name in os.listdir(cache_dir):
        question_dir = os.path.join(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):
                try:
                    with open(question_file, 'r', encoding='utf-8') as f:
                        question = f.read()
                    count += 1
                    print(f"{count}. [{dir_name}] {question[:50]}...")
                except:
                    pass
    
    print("=" * 60)
    print(f"总计: {count} 个缓存问题\n")

if __name__ == "__main__":
    print("=" * 60)
    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("无效的选择")