#!/usr/bin/env python3 """ Memory Reminder Script for Personal Relationship Management. Checks upcoming events and follow-ups from the memory system and sends notifications via Feishu. Usage: python scripts/memory_reminder.py Cron example (daily at 9:00 AM CST): 0 9 * * * cd /root/projects/butler && /root/projects/butler/venv/bin/python /root/projects/butler/scripts/memory_reminder.py >> /root/projects/butler/logs/memory_reminder.log 2>&1 """ import json import os import sys from datetime import date, datetime, timedelta from pathlib import Path # Add project root to path sys.path.insert(0, str(Path(__file__).parent.parent)) # OpenClaw workspace path WORKSPACE = Path("/root/.openclaw/workspace") MEMORY_DIR = WORKSPACE / "memory/life-memories" def load_json(filepath: Path) -> dict: """Load and parse a JSON file.""" if not filepath.exists(): return {} with open(filepath, "r", encoding="utf-8") as f: return json.load(f) def save_json(filepath: Path, data: dict): """Save data to a JSON file.""" filepath.parent.mkdir(parents=True, exist_ok=True) with open(filepath, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) def get_today() -> date: """Get today's date in China timezone (GMT+8).""" # Simple approach: use UTC date and adjust if needed # For cron running at 9:00 AM CST (1:00 AM UTC), this should work return date.today() def days_until(target_date: str) -> int: """Calculate days until target date (YYYY-MM-DD format).""" try: target = date.fromisoformat(target_date) today = get_today() return (target - today).days except (ValueError, TypeError): return 9999 # Return a large number for non-date strings def check_upcoming_events() -> list[dict]: """Check for events that need reminders.""" events_file = MEMORY_DIR / "upcoming-events.json" data = load_json(events_file) events = data.get("events", []) reminders = [] today = get_today() for event in events: if event.get("reminded", False): continue try: event_date = date.fromisoformat(event["date"]) days = (event_date - today).days # Remind 2 days before, 1 day before, or same day if 0 <= days <= 2: reminders.append({ "type": "event", "title": event["title"], "date": event["date"], "description": event.get("description", ""), "people": event.get("people", []), "event_id": event["id"] }) except (ValueError, TypeError): continue # Skip events with invalid date formats return reminders def check_follow_ups() -> list[dict]: """Check for follow-ups that are due or overdue.""" followups_file = MEMORY_DIR / "follow-ups.json" data = load_json(followups_file) followups = data.get("followUps", []) reminders = [] today = get_today() for followup in followups: if followup.get("status") == "completed": continue try: deadline = date.fromisoformat(followup["deadline"]) days = (deadline - today).days # Remind if due today, overdue, or due tomorrow if days <= 1: reminders.append({ "type": "followup", "person": followup.get("person", ""), "task": followup.get("task", ""), "deadline": followup["deadline"], "priority": followup.get("priority", "normal"), "id": followup["id"] }) except (ValueError, TypeError): # Skip follow-ups with invalid deadline formats like "待定" continue return reminders def generate_message(reminders: list[dict]) -> str: """Generate Feishu message from reminders.""" if not reminders: return None lines = ["📋 **今日提醒事项**\n"] events = [r for r in reminders if r["type"] == "event"] followups = [r for r in reminders if r["type"] == "followup"] if events: lines.append("📅 **即将到来的事件：**\n") for e in events: days = days_until(e["date"]) time_text = "今天" if days == 0 else "明天" lines.append(f"• **{e['title']}** - {time_text}\n") if e["description"]: lines.append(f" {e['description']}\n") if e["people"]: lines.append(f" 相关人员: {', '.join(e['people'])}\n") lines.append("") if followups: lines.append("⏰ **待跟进事项：**\n") for f in followups: deadline_days = days_until(f["deadline"]) if deadline_days == 9999: time_text = f['deadline'] # Show original string if not a date elif deadline_days == 0: time_text = "今天截止" elif deadline_days < 0: time_text = f"已超期 {abs(deadline_days)} 天" else: time_text = f"还有 {deadline_days} 天" priority_emoji = "🔴" if f["priority"] == "high" else "🟡" lines.append(f"{priority_emoji} • {f['task']}\n") lines.append(f" 截止: {time_text} ({f['deadline']})\n") if f["person"]: lines.append(f" 人物: {f['person']}\n") lines.append("") lines.append("---\n") lines.append("💡 记得处理这些事项哦！有问题随时问我。") return "".join(lines) def send_feishu_message(text: str): """Send message to Feishu using OpenClaw's message tool. This uses the openclaw CLI to send the message to the configured Feishu user. """ import subprocess # Load the target user from feishu-allowFrom.json allow_from_file = Path("/root/.openclaw/credentials/feishu-allowFrom.json") target_user = None if allow_from_file.exists(): allow_data = load_json(allow_from_file) allow_list = allow_data.get("allowFrom", []) if allow_list: target_user = allow_list[0] # Use the first allowed user if not target_user: print("ERROR: No Feishu target user configured in feishu-allowFrom.json") return False print(f"[FEISHU MESSAGE]\n{text}\n[/FEISHU MESSAGE]") # Save to a file for debugging msg_file = WORKSPACE / ".feishu_outbox.txt" msg_file.write_text(text, encoding="utf-8") print(f"Message saved to {msg_file}") # Send via OpenClaw CLI cmd = [ "/root/.nvm/versions/node/v22.22.0/bin/openclaw", "message", "send", "--channel", "feishu", "--target", f"user:{target_user}", "--message", text ] print(f"Running: {' '.join(cmd)}") try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=30 ) if result.returncode == 0: print(f"✅ Message sent successfully to Feishu") return True else: print(f"❌ Failed to send message:") print(f" stdout: {result.stdout}") print(f" stderr: {result.stderr}") return False except subprocess.TimeoutExpired: print("❌ Message send timed out after 30 seconds") return False except Exception as e: print(f"❌ Error sending message: {e}") return False def mark_events_as_reminded(event_ids: list[str]): """Mark events as reminded.""" events_file = MEMORY_DIR / "upcoming-events.json" data = load_json(events_file) events = data.get("events", []) for event in events: if event.get("id") in event_ids: event["reminded"] = True data["events"] = events data["lastUpdated"] = datetime.now().isoformat() save_json(events_file, data) def main() -> int: """Main entry point.""" print(f"[{datetime.now().isoformat()}] Memory Reminder Started") # Check reminders events = check_upcoming_events() followups = check_follow_ups() print(f"Found {len(events)} upcoming events, {len(followups)} follow-ups") all_reminders = events + followups if all_reminders: message = generate_message(all_reminders) print(f"Generated message:\n{message}\n") send_feishu_message(message) # Mark events as reminded event_ids = [e["event_id"] for e in events] if event_ids: mark_events_as_reminded(event_ids) print(f"Marked {len(event_ids)} events as reminded") print("Reminder sent successfully") return 0 else: print("No reminders due today") return 0 if __name__ == "__main__": sys.exit(main())