#!/usr/bin/env python3
|
"""
|
hot-memory 每日维护脚本
|
维护 memory.md 文件的最近7天事件流水和重要事件
|
"""
|
|
import os
|
import re
|
import sys
|
from datetime import datetime, timedelta
|
from pathlib import Path
|
|
# 配置
|
MEMORY_FILE = Path.home() / ".openclaw" / "workspace" / "MEMORY.md"
|
MAX_SIZE_BYTES = 4 * 1024 # 4KB 限制
|
DAYS_TO_KEEP = 7
|
|
|
def read_memory_file():
|
"""读取 memory.md 文件内容"""
|
if not MEMORY_FILE.exists():
|
return ""
|
return MEMORY_FILE.read_text(encoding='utf-8')
|
|
|
def write_memory_file(content):
|
"""写入 memory.md 文件"""
|
MEMORY_FILE.parent.mkdir(parents=True, exist_ok=True)
|
MEMORY_FILE.write_text(content, encoding='utf-8')
|
|
|
def get_file_size():
|
"""获取文件大小（字节）"""
|
if not MEMORY_FILE.exists():
|
return 0
|
return MEMORY_FILE.stat().st_size
|
|
|
def parse_date_from_line(line):
|
"""从行中提取日期（YYYY-MM-DD 格式）"""
|
match = re.search(r'(\d{4}-\d{2}-\d{2})', line)
|
if match:
|
try:
|
return datetime.strptime(match.group(1), '%Y-%m-%d').date()
|
except ValueError:
|
return None
|
return None
|
|
|
def is_within_last_7_days(line):
|
"""检查行是否在最近7天内"""
|
date = parse_date_from_line(line)
|
if not date:
|
return True # 无日期的行保留
|
cutoff_date = (datetime.now() - timedelta(days=DAYS_TO_KEEP)).date()
|
return date >= cutoff_date
|
|
|
def extract_event_entries(content):
|
"""
|
从内容中提取事件条目
|
返回: (important_events, daily_events) 元组
|
"""
|
important_events = []
|
daily_events = []
|
|
lines = content.split('\n')
|
in_important = False
|
in_daily = False
|
|
for line in lines:
|
stripped = line.strip()
|
|
# 检测区块
|
if '重要事件' in stripped and stripped.startswith('#'):
|
in_important = True
|
in_daily = False
|
continue
|
elif ('最近7天事件流水' in stripped or '事件流水' in stripped) and stripped.startswith('#'):
|
in_important = False
|
in_daily = True
|
continue
|
elif stripped.startswith('#') and in_important:
|
in_important = False
|
elif stripped.startswith('#') and in_daily:
|
in_daily = False
|
|
# 收集事件
|
if in_important and stripped and not stripped.startswith('-') and not stripped.startswith('*'):
|
if len(stripped) > 10: # 过滤空行和短行
|
important_events.append(line)
|
elif in_daily and (stripped.startswith('-') or stripped.startswith('*')):
|
if is_within_last_7_days(line):
|
daily_events.append(line)
|
|
return important_events, daily_events
|
|
|
def generate_default_content():
|
"""生成默认的 memory.md 内容"""
|
today = datetime.now().strftime('%Y-%m-%d')
|
return f"""# MEMORY.md - 热记忆 / 活跃记忆
|
|
> 本文件记录近期发生的重要事情，详细信息可通过记忆检索获取。
|
|
---
|
|
## 🔔 重要事件
|
|
> 记录具有全局长期性影响的重要决策和事件。
|
> 添加重要事件时会告知用户。
|
|
<!-- 重要事件在此添加 -->
|
|
---
|
|
## 📅 最近7天事件流水
|
|
> 按天分组，每天主要事情的概要。
|
> 自动维护，8天前的记录会被移除。
|
|
### {{today}}
|
|
- {today} 10:00 | memory.md 初始化 | 热记忆系统
|
|
---
|
|
*文件大小: ~0.5KB | 限制: 4KB*
|
*维护脚本: `hot-memory/scripts/daily_maintenance.py`*
|
""".format(today=today)
|
|
|
def update_memory_file(new_daily_events=None, important_event=None):
|
"""
|
更新 memory.md 文件
|
|
Args:
|
new_daily_events: 新的每日事件列表 [(date, time, summary, keywords), ...]
|
important_event: 重要事件元组 (date, time, summary, keywords) 或 None
|
|
Returns:
|
dict: 操作结果
|
"""
|
result = {
|
'success': True,
|
'added_daily': 0,
|
'added_important': False,
|
'removed_old': 0,
|
'size_warning': False,
|
'current_size_kb': 0,
|
'messages': []
|
}
|
|
# 读取现有内容
|
content = read_memory_file()
|
if not content:
|
content = generate_default_content()
|
|
# 提取现有事件
|
existing_important, existing_daily = extract_event_entries(content)
|
|
# 移除8天前的日常事件
|
cutoff_date = (datetime.now() - timedelta(days=DAYS_TO_KEEP)).date()
|
filtered_daily = []
|
for line in existing_daily:
|
date = parse_date_from_line(line)
|
if date and date < cutoff_date:
|
result['removed_old'] += 1
|
else:
|
filtered_daily.append(line)
|
|
# 添加新的日常事件
|
if new_daily_events:
|
for event in new_daily_events:
|
date, time, summary, keywords = event
|
entry = f"- {date} {time} | {summary} | {keywords}"
|
# 检查是否已存在
|
if not any(entry.strip() in existing.strip() for existing in filtered_daily):
|
filtered_daily.append(entry)
|
result['added_daily'] += 1
|
|
# 添加重要事件
|
new_important = list(existing_important)
|
if important_event:
|
date, time, summary, keywords = important_event
|
entry = f"- {date} {time} | {summary} | {keywords}"
|
new_important.append(entry)
|
result['added_important'] = True
|
result['messages'].append(f"重要事件已记录: {summary}")
|
|
# 重新构建文件内容
|
today = datetime.now().strftime('%Y-%m-%d')
|
|
# 按日期分组日常事件
|
daily_by_date = {}
|
for line in filtered_daily:
|
date = parse_date_from_line(line)
|
if date:
|
date_str = date.strftime('%Y-%m-%d')
|
if date_str not in daily_by_date:
|
daily_by_date[date_str] = []
|
daily_by_date[date_str].append(line)
|
|
# 构建日常事件区块
|
daily_sections = []
|
for date_str in sorted(daily_by_date.keys(), reverse=True):
|
daily_sections.append(f"\n### {date_str}\n")
|
daily_sections.extend([f"{line}\n" for line in daily_by_date[date_str]])
|
|
daily_content = ''.join(daily_sections) if daily_sections else f"\n### {today}\n\n- {today} --:-- | 暂无记录 | --\n"
|
|
# 构建重要事件区块
|
important_content = '\n'.join(new_important) if new_important else "<!-- 重要事件在此添加 -->"
|
|
# 组装最终内容
|
current_size = get_file_size()
|
new_content = f"""# MEMORY.md - 热记忆 / 活跃记忆
|
|
> 本文件记录近期发生的重要事情，详细信息可通过记忆检索获取。
|
|
---
|
|
## 🔔 重要事件
|
|
> 记录具有全局长期性影响的重要决策和事件。
|
> 添加重要事件时会告知用户。
|
|
{important_content}
|
|
---
|
|
## 📅 最近7天事件流水
|
|
> 按天分组，每天主要事情的概要。
|
> 自动维护，8天前的记录会被移除。
|
{daily_content}
|
---
|
|
*文件大小: ~{current_size / 1024:.1f}KB | 限制: 4KB*
|
*维护脚本: `hot-memory/scripts/daily_maintenance.py`*
|
"""
|
|
# 写入文件
|
write_memory_file(new_content)
|
|
# 检查文件大小
|
new_size = get_file_size()
|
result['current_size_kb'] = round(new_size / 1024, 2)
|
|
if new_size > MAX_SIZE_BYTES:
|
result['size_warning'] = True
|
result['messages'].append(f"警告: 文件大小 ({result['current_size_kb']}KB) 超过 4KB 限制，需要瘦身")
|
|
return result
|
|
|
def check_size():
|
"""检查文件大小并返回状态"""
|
size = get_file_size()
|
return {
|
'size_bytes': size,
|
'size_kb': round(size / 1024, 2),
|
'limit_bytes': MAX_SIZE_BYTES,
|
'limit_kb': 4,
|
'exceeded': size > MAX_SIZE_BYTES,
|
'percentage': round((size / MAX_SIZE_BYTES) * 100, 1)
|
}
|
|
|
def main():
|
"""主函数 - 供命令行调用"""
|
if len(sys.argv) < 2:
|
print("用法: python daily_maintenance.py <command> [args]")
|
print("")
|
print("命令:")
|
print(" update - 执行每日更新")
|
print(" check-size - 检查文件大小")
|
print(" add-daily <date> <time> <summary> <keywords>")
|
print(" - 添加日常事件")
|
print(" add-important <date> <time> <summary> <keywords>")
|
print(" - 添加重要事件")
|
print("")
|
print("示例:")
|
print(' python daily_maintenance.py add-daily 2026-03-16 "10:30" "测试脚本" "开发,测试"')
|
sys.exit(1)
|
|
command = sys.argv[1]
|
|
if command == "update":
|
result = update_memory_file()
|
print(f"更新完成:")
|
print(f" - 添加日常事件: {result['added_daily']}")
|
print(f" - 移除旧事件: {result['removed_old']}")
|
print(f" - 添加重要事件: {'是' if result['added_important'] else '否'}")
|
print(f" - 当前大小: {result['current_size_kb']}KB")
|
for msg in result['messages']:
|
print(f" - {msg}")
|
|
elif command == "check-size":
|
status = check_size()
|
print(f"文件大小检查:")
|
print(f" - 当前: {status['size_kb']}KB ({status['percentage']}%)")
|
print(f" - 限制: {status['limit_kb']}KB")
|
print(f" - 状态: {'⚠️ 超限' if status['exceeded'] else '✅ 正常'}")
|
|
elif command == "add-daily" and len(sys.argv) >= 6:
|
event = (sys.argv[2], sys.argv[3], sys.argv[4], sys.argv[5])
|
result = update_memory_file(new_daily_events=[event])
|
print(f"已添加日常事件: {event}")
|
if result['size_warning']:
|
print(f"⚠️ 警告: 文件大小 {result['current_size_kb']}KB 超过限制")
|
|
elif command == "add-important" and len(sys.argv) >= 6:
|
event = (sys.argv[2], sys.argv[3], sys.argv[4], sys.argv[5])
|
result = update_memory_file(important_event=event)
|
print(f"已添加重要事件: {event}")
|
print(f"ℹ️ 请知悉用户: 已记录重要事件 '{sys.argv[4]}'")
|
if result['size_warning']:
|
print(f"⚠️ 警告: 文件大小 {result['current_size_kb']}KB 超过限制")
|
|
else:
|
print(f"未知命令或参数不足: {command}")
|
sys.exit(1)
|
|
|
if __name__ == "__main__":
|
main()
|
