261 lines
8.5 KiB
Python
261 lines
8.5 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
🦐 皮皮虾记忆管理工具
|
||
自动化记忆维护：整理、清理、提炼、归档
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
import re
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
|
||
# 配置
|
||
WORKSPACE = Path.home() / ".openclaw" / "workspace"
|
||
MEMORY_DIR = WORKSPACE / "memory"
|
||
MEMORY_MD = WORKSPACE / "MEMORY.md"
|
||
RETENTION_DAYS = 30 # 短期记忆保留天数
|
||
|
||
# ANSI 颜色
|
||
class Colors:
|
||
GREEN = '\033[92m'
|
||
YELLOW = '\033[93m'
|
||
RED = '\033[91m'
|
||
BLUE = '\033[94m'
|
||
END = '\033[0m'
|
||
BOLD = '\033[1m'
|
||
|
||
def print_header(text):
|
||
print(f"\n{Colors.BOLD}{Colors.BLUE}{'='*60}{Colors.END}")
|
||
print(f"{Colors.BOLD}{Colors.BLUE} {text}{Colors.END}")
|
||
print(f"{Colors.BOLD}{Colors.BLUE}{'='*60}{Colors.END}\n")
|
||
|
||
def print_success(text):
|
||
print(f"{Colors.GREEN}✅ {text}{Colors.END}")
|
||
|
||
def print_warning(text):
|
||
print(f"{Colors.YELLOW}⚠️ {text}{Colors.END}")
|
||
|
||
def print_info(text):
|
||
print(f"{Colors.BLUE}📌 {text}{Colors.END}")
|
||
|
||
def get_memory_files():
|
||
"""获取所有记忆文件"""
|
||
if not MEMORY_DIR.exists():
|
||
return []
|
||
return sorted([f for f in MEMORY_DIR.glob("*.md") if f.is_file()])
|
||
|
||
def get_daily_logs():
|
||
"""获取所有每日日志文件"""
|
||
files = get_memory_files()
|
||
daily_logs = []
|
||
for f in files:
|
||
if re.match(r'^\d{4}-\d{2}-\d{2}\.md$', f.name):
|
||
daily_logs.append(f)
|
||
return daily_logs
|
||
|
||
def parse_log_date(filename):
|
||
"""从文件名解析日期"""
|
||
match = re.match(r'^(\d{4}-\d{2}-\d{2})\.md$', filename)
|
||
if match:
|
||
return datetime.strptime(match.group(1), '%Y-%m-%d')
|
||
return None
|
||
|
||
def get_file_size_kb(filepath):
|
||
"""获取文件大小（KB）"""
|
||
return round(filepath.stat().st_size / 1024, 2)
|
||
|
||
def read_file_content(filepath):
|
||
"""读取文件内容"""
|
||
try:
|
||
with open(filepath, 'r', encoding='utf-8') as f:
|
||
return f.read()
|
||
except Exception as e:
|
||
print_warning(f"读取失败 {filepath.name}: {e}")
|
||
return None
|
||
|
||
def write_file_content(filepath, content):
|
||
"""写入文件内容"""
|
||
try:
|
||
with open(filepath, 'w', encoding='utf-8') as f:
|
||
f.write(content)
|
||
return True
|
||
except Exception as e:
|
||
print_warning(f"写入失败 {filepath.name}: {e}")
|
||
return False
|
||
|
||
def extract_important_content(content, filepath):
|
||
"""从每日日志中提取重要内容"""
|
||
important_items = []
|
||
filename = filepath.stem # 不含扩展名的文件名
|
||
|
||
# 提取会话记录中的关键事件
|
||
session_matches = re.findall(r'###.*?—.*?\n(.*?)(?=###|$)', content, re.DOTALL)
|
||
for session in session_matches[:5]: # 最多 5 个会话
|
||
# 提取用户请求
|
||
user_req = re.search(r'用户请求：(.*?)(?:\n|$)', session)
|
||
# 提取结果
|
||
result = re.search(r'结果：(✅.*?)(?:\n|$)', session)
|
||
|
||
if user_req and result:
|
||
important_items.append(f"**{filename}** - {user_req.group(1).strip()} → {result.group(1).strip()}")
|
||
elif user_req:
|
||
important_items.append(f"**{filename}** - {user_req.group(1).strip()}")
|
||
|
||
# 提取重要性评分记录中的高分项
|
||
rating_matches = re.findall(r'\|\s*([^\|]+)\s*\|\s*([45])\s*\|\s*(✅.*?)(?:\n|$)', content)
|
||
for item, score, action in rating_matches[:3]:
|
||
important_items.append(f"**高分记忆** - {item.strip()} (评分{score}) → {action.strip()}")
|
||
|
||
return important_items
|
||
|
||
def update_memory_md(important_items):
|
||
"""更新 MEMORY.md 文件"""
|
||
if not MEMORY_MD.exists():
|
||
print_warning("MEMORY.md 不存在，跳过更新")
|
||
return
|
||
|
||
content = read_file_content(MEMORY_MD)
|
||
if not content:
|
||
return
|
||
|
||
# 在"核心记忆内容"部分添加新内容
|
||
today = datetime.now().strftime('%Y-%m-%d')
|
||
new_section = f"\n### {today} 自动提炼\n\n"
|
||
for item in important_items[:10]: # 最多添加 10 条
|
||
new_section += f"{item}\n"
|
||
new_section += "\n"
|
||
|
||
# 找到"核心记忆内容"部分并插入
|
||
if "## 📝 核心记忆内容" in content:
|
||
parts = content.split("## 📝 核心记忆内容")
|
||
if len(parts) == 2:
|
||
new_content = parts[0] + "## 📝 核心记忆内容" + new_section + parts[1]
|
||
# 更新最后更新时间
|
||
new_content = re.sub(
|
||
r'\*最后更新：\d{4}-\d{2}-\d{2}',
|
||
f'*最后更新：{today}',
|
||
new_content
|
||
)
|
||
if write_file_content(MEMORY_MD, new_content):
|
||
print_success("MEMORY.md 已更新")
|
||
else:
|
||
print_warning("未找到核心记忆内容部分")
|
||
|
||
def archive_old_logs(old_logs):
|
||
"""归档旧日志"""
|
||
if not old_logs:
|
||
return
|
||
|
||
archive_dir = MEMORY_DIR / "archive"
|
||
archive_dir.mkdir(exist_ok=True)
|
||
|
||
for log_file in old_logs:
|
||
# 移动到 archive 目录
|
||
dest = archive_dir / log_file.name
|
||
try:
|
||
log_file.rename(dest)
|
||
print_success(f"已归档：{log_file.name} → archive/")
|
||
except Exception as e:
|
||
print_warning(f"归档失败 {log_file.name}: {e}")
|
||
|
||
def cleanup_empty_files():
|
||
"""清理空文件"""
|
||
cleaned = 0
|
||
for f in get_memory_files():
|
||
if f.stat().st_size == 0:
|
||
try:
|
||
f.unlink()
|
||
print_success(f"已删除空文件：{f.name}")
|
||
cleaned += 1
|
||
except Exception as e:
|
||
print_warning(f"删除失败 {f.name}: {e}")
|
||
return cleaned
|
||
|
||
def generate_report(daily_logs, old_logs, important_items):
|
||
"""生成维护报告"""
|
||
today = datetime.now()
|
||
|
||
print_header("📊 记忆维护报告")
|
||
|
||
print(f"{Colors.BOLD}记忆文件统计：{Colors.END}")
|
||
print(f" - 每日日志总数：{len(daily_logs)}")
|
||
print(f" - 长期记忆文件：{len(get_memory_files()) - len(daily_logs)}")
|
||
|
||
print(f"\n{Colors.BOLD}清理结果：{Colors.END}")
|
||
if old_logs:
|
||
print(f" - 超过{RETENTION_DAYS}天的日志：{len(old_logs)} 个")
|
||
print(f" - 已归档到 memory/archive/")
|
||
else:
|
||
print(f" - 无需清理（所有日志都在{RETENTION_DAYS}天内）")
|
||
|
||
print(f"\n{Colors.BOLD}内容提炼：{Colors.END}")
|
||
if important_items:
|
||
print(f" - 提炼重要内容：{len(important_items)} 条")
|
||
print(f" - 已添加到 MEMORY.md")
|
||
else:
|
||
print(f" - 暂无可提炼内容")
|
||
|
||
print(f"\n{Colors.BOLD}下次维护：{Colors.END}")
|
||
next_check = today + timedelta(days=7)
|
||
print(f" - 建议下次检查：{next_check.strftime('%Y-%m-%d')}")
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print_header("🦐 皮皮虾记忆维护工具")
|
||
|
||
# 1. 获取所有每日日志
|
||
daily_logs = get_daily_logs()
|
||
print_info(f"找到 {len(daily_logs)} 个每日日志文件")
|
||
|
||
# 2. 识别过期日志（超过 RETENTION_DAYS 天）
|
||
today = datetime.now()
|
||
cutoff_date = today - timedelta(days=RETENTION_DAYS)
|
||
old_logs = []
|
||
|
||
for log_file in daily_logs:
|
||
log_date = parse_log_date(log_file.name)
|
||
if log_date and log_date < cutoff_date:
|
||
old_logs.append(log_file)
|
||
|
||
if old_logs:
|
||
print_warning(f"发现 {len(old_logs)} 个过期日志（超过{RETENTION_DAYS}天）")
|
||
else:
|
||
print_success(f"所有日志都在{RETENTION_DAYS}天内，无需清理")
|
||
|
||
# 3. 从最近的日志中提取重要内容
|
||
important_items = []
|
||
recent_logs = sorted(daily_logs, key=lambda x: x.name, reverse=True)[:3]
|
||
|
||
for log_file in recent_logs:
|
||
content = read_file_content(log_file)
|
||
if content:
|
||
items = extract_important_content(content, log_file)
|
||
important_items.extend(items)
|
||
|
||
if important_items:
|
||
print_success(f"从最近日志中提取了 {len(important_items)} 条重要内容")
|
||
|
||
# 4. 更新 MEMORY.md
|
||
if important_items:
|
||
update_memory_md(important_items)
|
||
|
||
# 5. 归档过期日志（询问用户确认）
|
||
if old_logs:
|
||
print(f"\n{Colors.YELLOW}准备归档 {len(old_logs)} 个过期日志到 memory/archive/{Colors.END}")
|
||
# 自动执行归档
|
||
archive_old_logs(old_logs)
|
||
|
||
# 6. 清理空文件
|
||
cleaned = cleanup_empty_files()
|
||
if cleaned:
|
||
print_success(f"清理了 {cleaned} 个空文件")
|
||
|
||
# 7. 生成报告
|
||
generate_report(daily_logs, old_logs, important_items)
|
||
|
||
print(f"\n{Colors.GREEN}🦐 记忆维护完成！{Colors.END}\n")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|