archive_manager.sh•9.64 kB
#!/bin/bash
# MemOS归档管理脚本
# 管理30天未访问数据的自动归档和存储优化
set -e
MEMOS_DIR="/home/qqinshu/视频/MemOS"
# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m' # No Color
log_info() {
echo -e "${BLUE}ℹ️ $1${NC}"
}
log_success() {
echo -e "${GREEN}✅ $1${NC}"
}
log_warning() {
echo -e "${YELLOW}⚠️ $1${NC}"
}
log_error() {
echo -e "${RED}❌ $1${NC}"
}
# 显示帮助信息
show_help() {
echo "MemOS归档管理脚本"
echo ""
echo "用法: $0 [命令]"
echo ""
echo "命令:"
echo " status 显示存储和归档状态"
echo " analyze 分析存储使用情况"
echo " candidates 查看归档候选记忆"
echo " archive 执行30天未访问记忆归档"
echo " dry-run 模拟归档操作（不实际执行）"
echo " search 搜索已归档的记忆"
echo " restore 恢复归档的记忆"
echo " cleanup 清理旧数据"
echo " report 生成优化报告"
echo " help 显示此帮助信息"
echo ""
echo "示例:"
echo " $0 status # 查看当前状态"
echo " $0 dry-run # 模拟归档操作"
echo " $0 archive # 执行实际归档"
}
# 检查Python环境
check_python_env() {
if [ ! -d "$MEMOS_DIR/memos_venv" ]; then
log_error "虚拟环境不存在: $MEMOS_DIR/memos_venv"
exit 1
fi
if [ ! -f "$MEMOS_DIR/storage_optimizer.py" ]; then
log_error "存储优化器脚本不存在"
exit 1
fi
}
# 显示存储状态
show_status() {
log_info "显示存储和归档状态..."
cd "$MEMOS_DIR"
$MEMOS_DIR/memos_venv/bin/python -c "
from storage_optimizer import StorageOptimizer
optimizer = StorageOptimizer()
optimizer.print_status()
"
}
# 分析存储使用
analyze_storage() {
log_info "分析存储使用情况..."
cd "$MEMOS_DIR"
$MEMOS_DIR/memos_venv/bin/python -c "
from storage_optimizer import StorageOptimizer
import json
optimizer = StorageOptimizer()
analysis = optimizer.analyze_storage_usage()
print('📊 存储使用分析:')
print(f' 总存储大小: {analysis[\"total_size_mb\"]:.2f} MB')
print('')
for name, info in analysis['directories'].items():
print(f' {name}:')
print(f' 大小: {info[\"size_mb\"]:.2f} MB')
print(f' 文件数: {info[\"file_count\"]}')
print(f' 路径: {info[\"path\"]}')
print('')
if analysis['recommendations']:
print('💡 优化建议:')
for rec in analysis['recommendations']:
print(f' • {rec}')
"
}
# 查看归档候选
show_candidates() {
log_info "查看归档候选记忆..."
cd "$MEMOS_DIR"
$MEMOS_DIR/memos_venv/bin/python -c "
from storage_optimizer import StorageOptimizer
optimizer = StorageOptimizer()
candidates = optimizer.get_archival_candidates()
if not candidates:
print('✅ 没有需要归档的记忆')
else:
print(f'📋 找到 {len(candidates)} 个归档候选记忆:')
print('')
for i, candidate in enumerate(candidates[:10], 1):
print(f'{i:2d}. ID {candidate[\"memory_id\"]}:')
print(f' 最后访问: {candidate[\"last_access\"]}')
print(f' 未访问天数: {candidate[\"days_since_access\"]}')
print(f' 访问次数: {candidate[\"access_count\"]}')
print('')
if len(candidates) > 10:
print(f' ... 还有 {len(candidates) - 10} 条记忆')
"
}
# 执行归档
execute_archive() {
local dry_run=${1:-false}
if [ "$dry_run" = "true" ]; then
log_info "模拟归档操作（不实际执行）..."
else
log_info "执行30天未访问记忆归档..."
log_warning "这将从活跃存储中移除未访问的记忆"
read -p "确认继续？(y/N): " confirm
if [[ ! "$confirm" =~ ^[Yy]$ ]]; then
log_info "归档操作已取消"
return 0
fi
fi
cd "$MEMOS_DIR"
$MEMOS_DIR/memos_venv/bin/python -c "
from storage_optimizer import StorageOptimizer
import json
optimizer = StorageOptimizer()
result = optimizer.archive_inactive_memories(dry_run=$dry_run)
if result['success']:
if result['dry_run']:
print(f'🔍 模拟结果: 将归档 {result[\"archived_count\"]} 条记忆')
else:
print(f'✅ 归档成功: {result[\"archived_count\"]} 条记忆')
if result['archive_name']:
print(f' 归档文件: {result[\"archive_name\"]}.parquet')
else:
print('❌ 归档失败:')
for error in result['errors']:
print(f' • {error}')
"
}
# 搜索归档记忆
search_archived() {
if [ -z "$1" ]; then
echo "用法: $0 search <查询词>"
exit 1
fi
local query="$1"
log_info "搜索已归档的记忆: $query"
cd "$MEMOS_DIR"
$MEMOS_DIR/memos_venv/bin/python -c "
from storage_optimizer import StorageOptimizer
optimizer = StorageOptimizer()
results = optimizer.search_archived_memories('$query', max_results=10)
if not results:
print('❌ 未找到匹配的归档记忆')
else:
print(f'📋 找到 {len(results)} 条归档记忆:')
print('')
for i, result in enumerate(results, 1):
print(f'{i}. ID {result[\"memory_id\"]} (访问{result[\"access_count\"]}次):')
print(f' 内容: {result[\"content\"]}')
print(f' 标签: {result[\"tags\"]}')
print(f' 归档: {result[\"archive_name\"]}')
print(f' 最后访问: {result[\"last_access\"]}')
print('')
"
}
# 恢复归档记忆
restore_memory() {
if [ -z "$1" ]; then
echo "用法: $0 restore <记忆ID>"
exit 1
fi
local memory_id="$1"
log_info "恢复归档记忆: $memory_id"
cd "$MEMOS_DIR"
$MEMOS_DIR/memos_venv/bin/python -c "
from storage_optimizer import StorageOptimizer
optimizer = StorageOptimizer()
success = optimizer.restore_memory('$memory_id')
if success:
print('✅ 记忆恢复成功')
else:
print('❌ 记忆恢复失败')
"
}
# 清理旧数据
cleanup_old_data() {
log_info "清理旧数据..."
log_warning "这将删除90天以上的访问统计和365天以上的归档文件"
read -p "确认继续？(y/N): " confirm
if [[ ! "$confirm" =~ ^[Yy]$ ]]; then
log_info "清理操作已取消"
return 0
fi
cd "$MEMOS_DIR"
$MEMOS_DIR/memos_venv/bin/python -c "
from storage_optimizer import StorageOptimizer
optimizer = StorageOptimizer()
optimizer.cleanup_old_data()
print('✅ 旧数据清理完成')
"
}
# 生成优化报告
generate_report() {
log_info "生成存储优化报告..."
cd "$MEMOS_DIR"
$MEMOS_DIR/memos_venv/bin/python -c "
from storage_optimizer import StorageOptimizer
import json
from datetime import datetime
optimizer = StorageOptimizer()
report = optimizer.get_optimization_report()
print('📊 MemOS存储优化报告')
print('=' * 50)
print(f'生成时间: {report[\"timestamp\"]}')
print('')
# 存储使用情况
storage = report['storage_usage']
print(f'💾 存储使用: {storage[\"total_size_mb\"]:.2f} MB')
for name, info in storage['directories'].items():
print(f' {name}: {info[\"size_mb\"]:.2f} MB')
print('')
# 访问统计
access = report['access_statistics']
print(f'📈 访问统计:')
print(f' 追踪记忆: {access[\"total_memories_tracked\"]}')
print(f' 总访问次数: {access[\"total_accesses\"]}')
print(f' 30天未访问: {access[\"inactive_memories\"][\"30_days\"]}')
print('')
# 归档统计
archive = report['archive_statistics']
print(f'📦 归档统计:')
print(f' 归档文件: {archive[\"total_archives\"]}')
print(f' 已归档记忆: {archive[\"total_archived_memories\"]}')
print(f' 归档大小: {archive[\"total_size_mb\"]:.2f} MB')
print('')
# 优化机会
if report['optimization_opportunities']:
print('💡 优化机会:')
for opp in report['optimization_opportunities']:
print(f' • {opp[\"description\"]}')
print(f' 潜在收益: {opp[\"potential_savings\"]}')
print('')
print('📋 建议操作:')
inactive_30d = access['inactive_memories']['30_days']
if inactive_30d > 0:
print(f' 1. 执行归档: ./archive_manager.sh archive')
print(f' 可归档 {inactive_30d} 条30天未访问记忆')
total_size = storage['total_size_mb']
if total_size > 100:
print(f' 2. 定期清理: ./archive_manager.sh cleanup')
print(f' 当前总存储 {total_size:.1f}MB')
print(f' 3. 定期监控: ./archive_manager.sh status')
print(f' 建议每周检查一次存储状态')
"
}
# 主函数
main() {
echo "🗄️ MemOS归档管理脚本"
echo "===================="
check_python_env
case "${1:-status}" in
status)
show_status
;;
analyze)
analyze_storage
;;
candidates)
show_candidates
;;
archive)
execute_archive false
;;
dry-run)
execute_archive true
;;
search)
search_archived "$2"
;;
restore)
restore_memory "$2"
;;
cleanup)
cleanup_old_data
;;
report)
generate_report
;;
help|--help|-h)
show_help
;;
*)
log_error "未知命令: $1"
show_help
exit 1
;;
esac
}
# 运行主函数
main "$@"