From 15cb154a19cabb5c3c66491a61f215ef104d5280 Mon Sep 17 00:00:00 2001
From: TevinClaw <510129976@qq.com>
Date: Sat, 14 Mar 2026 13:03:06 +0800
Subject: [PATCH] feat(早报): 增加昨日总结模块
---
workspace/skills/memory-management/scripts/weekly_maintenance.py | 207 +++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 207 insertions(+), 0 deletions(-)
diff --git a/workspace/skills/memory-management/scripts/weekly_maintenance.py b/workspace/skills/memory-management/scripts/weekly_maintenance.py
new file mode 100755
index 0000000..e4cee14
--- /dev/null
+++ b/workspace/skills/memory-management/scripts/weekly_maintenance.py
@@ -0,0 +1,207 @@
+#!/usr/bin/env python3
+"""
+每周维护脚本
+周一早上9:30执行，负责：
+1. 运行memory-merger整理L2→L1
+2. 检查L0大小
+3. 生成周报
+4. 发送报告（可选）
+"""
+
+import os
+import sys
+import subprocess
+from datetime import datetime, timedelta
+from pathlib import Path
+
+
+def get_workspace_path() -> Path:
+ """获取workspace路径。"""
+ return Path.home() / ".openclaw" / "workspace"
+
+
+def run_memory_merger() -> tuple:
+ """运行memory-merger技能。"""
+ workspace = get_workspace_path()
+ merger_path = workspace / ".agents" / "skills" / "memory-merger"
+
+ if not merger_path.exists():
+ return False, "memory-merger技能未安装"
+
+ # 运行memory-merger
+ try:
+ result = subprocess.run(
+ ["python", str(merger_path / "scripts" / "merge.py"), "memory-management"],
+ capture_output=True,
+ text=True,
+ timeout=60
+ )
+ if result.returncode == 0:
+ return True, result.stdout
+ else:
+ return False, result.stderr
+ except Exception as e:
+ return False, str(e)
+
+
+def check_l0_size() -> dict:
+ """检查L0状态。"""
+ workspace = get_workspace_path()
+ memory_file = workspace / "MEMORY.md"
+
+ if not memory_file.exists():
+ return {"exists": False, "size": 0, "status": "missing"}
+
+ size = memory_file.stat().st_size
+ kb = size / 1024
+
+ if size > 4096:
+ status = "over_limit"
+ elif size > 3500:
+ status = "warning"
+ else:
+ status = "ok"
+
+ return {
+ "exists": True,
+ "size": size,
+ "size_kb": kb,
+ "status": status
+ }
+
+
+def count_journal_files() -> int:
+ """统计本周L2文件数量。"""
+ workspace = get_workspace_path()
+ journal_dir = workspace / "memory" / "journal"
+
+ if not journal_dir.exists():
+ return 0
+
+ # 获取本周日期范围
+ today = datetime.now()
+ start_of_week = today - timedelta(days=today.weekday())
+
+ count = 0
+ for f in journal_dir.glob("*.md"):
+ try:
+ file_date = datetime.strptime(f.stem, "%Y-%m-%d")
+ if start_of_week <= file_date <= today:
+ count += 1
+ except ValueError:
+ continue
+
+ return count
+
+
+def count_milestone_files() -> int:
+ """统计L1里程碑文件数量。"""
+ workspace = get_workspace_path()
+ milestones_dir = workspace / "memory" / "milestones"
+
+ if not milestones_dir.exists():
+ return 0
+
+ return len(list(milestones_dir.glob("*.md")))
+
+
+def generate_report() -> str:
+ """生成周报内容。"""
+ today_str = datetime.now().strftime("%Y-%m-%d")
+ week_start = (datetime.now() - timedelta(days=datetime.now().weekday())).strftime("%Y-%m-%d")
+
+ report = []
+ report.append("# 📊 记忆管理周报")
+ report.append(f"**周期**: {week_start} ~ {today_str}")
+ report.append(f"**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M')}")
+ report.append("")
+ report.append("---")
+ report.append("")
+
+ # L0状态
+ l0_status = check_l0_size()
+ report.append("## 📋 L0层 (MEMORY.md)")
+ if l0_status["exists"]:
+ report.append(f"- **大小**: {l0_status['size_kb']:.1f}KB / 4KB")
+ if l0_status["status"] == "ok":
+ report.append("- **状态**: ✅ 正常")
+ elif l0_status["status"] == "warning":
+ report.append("- **状态**: ⚠️ 接近限制，建议归档")
+ else:
+ report.append("- **状态**: 🚨 超过红线，需要立即归档")
+ else:
+ report.append("- **状态**: ❌ 文件不存在")
+ report.append("")
+
+ # L2统计
+ journal_count = count_journal_files()
+ report.append("## 📝 L2层 (Journal)")
+ report.append(f"- **本周新增**: {journal_count} 条记录")
+ report.append("")
+
+ # L1统计
+ milestone_count = count_milestone_files()
+ report.append("## 🗂️ L1层 (Milestones)")
+ report.append(f"- **里程碑总数**: {milestone_count} 个主题")
+ report.append("")
+
+ # 维护任务
+ report.append("## 🔧 本周维护任务")
+
+ # 尝试运行memory-merger
+ success, output = run_memory_merger()
+ if success:
+ report.append("- ✅ L2→L1合并完成")
+ if output.strip():
+ report.append(f"- 📄 合并详情:\n```\n{output}\n```")
+ else:
+ report.append(f"- ❌ L2→L1合并失败: {output}")
+
+ if l0_status["status"] in ["warning", "over_limit"]:
+ report.append("- ⚠️ L0层需要归档整理")
+
+ report.append("")
+ report.append("---")
+ report.append("")
+ report.append("*由memory-management技能自动生成*")
+
+ return "\n".join(report)
+
+
+def main():
+ """主函数。"""
+ import argparse
+ parser = argparse.ArgumentParser(description="三层记忆每周维护")
+ parser.add_argument("--send-report", action="store_true", help="发送报告到飞书")
+ parser.add_argument("--output", type=str, help="报告输出文件路径")
+ args = parser.parse_args()
+
+ print("🔄 开始执行每周维护...")
+ print("=" * 50)
+
+ # 生成报告
+ report = generate_report()
+
+ # 输出到文件
+ if args.output:
+ with open(args.output, 'w', encoding='utf-8') as f:
+ f.write(report)
+ print(f"✅ 报告已保存到: {args.output}")
+
+ # 打印报告
+ print("\n" + report)
+
+ # 发送到飞书（如果需要）
+ if args.send_report:
+ print("\n📤 发送到飞书...")
+ # 这里会调用message工具，但在脚本中我们通过stdout返回
+ print(report)
+
+ print("\n" + "=" * 50)
+ print("✅ 每周维护完成")
+
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
--
Gitblit v1.9.1