Member Manager

~/.openclaw/workspace/users/{SESSION_KEY}/members.json
~/.openclaw/workspace/users/{SESSION_KEY}/reminders.json

import os, json, re

def sanitize_session_key(raw_key: str) -> str:
    """清洗 session key，只允許字母數字、連字號和底線，防止目錄遍歷"""
    sanitized = re.sub(r'[^a-zA-Z0-9_-]', '', raw_key)
    if not sanitized:
        raise ValueError(f"Invalid session key: {raw_key!r}")
    return sanitized

# 方法 1：從環境變數取得
raw_id = os.environ.get("OPENCLAW_SESSION_KEY", "default")

# 方法 2：如果是在 Telegram bot 場景，嘗試讀取 session 檔案
session_file = os.path.expanduser("~/.openclaw/session.json")
if os.path.exists(session_file):
    with open(session_file) as f:
        session = json.load(f)
    raw_id = session.get("user_id", session.get("session_key", "default"))

user_id = sanitize_session_key(str(raw_id))

WORKSPACE_ROOT = os.path.expanduser("~/.openclaw/workspace/users")
base_dir = os.path.join(WORKSPACE_ROOT, user_id)

# 二次驗證：確保解析後的路徑仍在 workspace 目錄內
real_base = os.path.realpath(base_dir)
real_root = os.path.realpath(WORKSPACE_ROOT)
if not real_base.startswith(real_root + os.sep):
    raise ValueError(f"Path traversal detected: {base_dir}")

members_path = os.path.join(base_dir, "members.json")
reminders_path = os.path.join(base_dir, "reminders.json")

# 初始化新用戶
os.makedirs(base_dir, exist_ok=True)
if not os.path.exists(members_path):
    with open(members_path, "w") as f:
        json.dump([], f)
if not os.path.exists(reminders_path):
    with open(reminders_path, "w") as f:
        json.dump([], f)

[
  {
    "id": "uuid-v4",
    "name": "張三",
    "group": "家庭",
    "relationship": "父親",
    "phone": "0912345678",
    "birthday_solar": "1955-03-15",
    "birthday_lunar": {
      "month": 2,
      "day": 13,
      "is_leap_month": false
    },
    "birthday_lunar_solar_this_year": "2025-03-12",
    "anniversaries": [
      {
        "id": "uuid-v4",
        "label": "結婚紀念日",
        "date_solar": "1980-06-20",
        "recurring": true
      }
    ],
    "notes": "喜歡喝普洱茶",
    "created_at": "2025-01-01T00:00:00Z",
    "updated_at": "2025-01-01T00:00:00Z"
  }
]

[
  {
    "id": "uuid-v4",
    "member_id": "對應 member 的 id",
    "type": "birthday_solar",
    "label": "張三 新曆生日",
    "advance_days": 3,
    "scheduled_task_id": "member-birthday-張三-solar",
    "enabled": true
  }
]

# pip install lunardate
from lunardate import LunarDate
from datetime import date, timedelta

def lunar_to_solar(year: int, lunar_month: int, lunar_day: int, is_leap: bool = False) -> date:
    """將農曆日期轉換為新曆日期"""
    try:
        lunar = LunarDate(year, lunar_month, lunar_day, is_leap)
        return lunar.toSolarDate()
    except ValueError:
        # 若該年無此農曆日期（例如閏月不存在），嘗試非閏月
        if is_leap:
            lunar = LunarDate(year, lunar_month, lunar_day, False)
            return lunar.toSolarDate()
        raise

def calculate_reminder_date(target_date: date, advance_days: int) -> date:
    """計算提前提醒的日期，正確處理跨月情況"""
    return target_date - timedelta(days=advance_days)

def update_all_lunar_birthdays(base_dir: str, year: int):
    """更新所有成員農曆生日在指定年份的新曆日期"""
    members_path = os.path.join(base_dir, "members.json")
    with open(members_path) as f:
        members = json.load(f)

    updated = []
    for m in members:
        lunar = m.get("birthday_lunar")
        if lunar:
            try:
                solar = lunar_to_solar(
                    year, lunar["month"], lunar["day"],
                    lunar.get("is_leap_month", False)
                )
                m["birthday_lunar_solar_this_year"] = str(solar)
                updated.append(m["name"])
            except ValueError as e:
                # 記錄錯誤但繼續處理其他成員
                pass
        m["updated_at"] = datetime.utcnow().isoformat() + "Z"

    with open(members_path, "w") as f:
        json.dump(members, f, ensure_ascii=False, indent=2)

    return updated

import json, os, uuid
from datetime import datetime

def add_member(base_dir, name, group, relationship="", birthday_solar=None,
               birthday_lunar=None, phone="", notes=""):
    path = os.path.join(base_dir, "members.json")
    with open(path) as f:
        members = json.load(f)

    # 檢查是否已有同名同組成員
    for existing in members:
        if existing["name"] == name and existing["group"] == group:
            return None, f"已存在同名成員：{name}（{group}）"

    now = datetime.utcnow().isoformat() + "Z"
    member = {
        "id": str(uuid.uuid4()),
        "name": name,
        "group": group,
        "relationship": relationship,
        "phone": phone,
        "birthday_solar": birthday_solar,
        "birthday_lunar": birthday_lunar,
        "birthday_lunar_solar_this_year": None,
        "anniversaries": [],
        "notes": notes,
        "created_at": now,
        "updated_at": now
    }

    # 如果有農曆生日，計算當年新曆日期
    if birthday_lunar:
        from lunardate import LunarDate
        from datetime import date
        try:
            solar = lunar_to_solar(
                date.today().year,
                birthday_lunar["month"],
                birthday_lunar["day"],
                birthday_lunar.get("is_leap_month", False)
            )
            member["birthday_lunar_solar_this_year"] = str(solar)
        except ValueError:
            pass

    members.append(member)
    with open(path, "w") as f:
        json.dump(members, f, ensure_ascii=False, indent=2)

    return member, None

def update_member(base_dir, member_id, **updates):
    """更新成員資料。支持更新任意欄位。"""
    path = os.path.join(base_dir, "members.json")
    with open(path) as f:
        members = json.load(f)

    for m in members:
        if m["id"] == member_id:
            for key, value in updates.items():
                if key in m and key not in ("id", "created_at"):
                    m[key] = value
            m["updated_at"] = datetime.utcnow().isoformat() + "Z"

            # 如果更新了農曆生日，重新計算新曆日期
            if "birthday_lunar" in updates and updates["birthday_lunar"]:
                from datetime import date
                try:
                    lunar = updates["birthday_lunar"]
                    solar = lunar_to_solar(
                        date.today().year,
                        lunar["month"], lunar["day"],
                        lunar.get("is_leap_month", False)
                    )
                    m["birthday_lunar_solar_this_year"] = str(solar)
                except ValueError:
                    pass

            with open(path, "w") as f:
                json.dump(members, f, ensure_ascii=False, indent=2)
            return m
    return None

def delete_member(base_dir, member_id):
    """刪除成員及其所有提醒"""
    # 刪除成員
    members_path = os.path.join(base_dir, "members.json")
    with open(members_path) as f:
        members = json.load(f)

    member = None
    new_members = []
    for m in members:
        if m["id"] == member_id:
            member = m
        else:
            new_members.append(m)

    if not member:
        return None, []

    with open(members_path, "w") as f:
        json.dump(new_members, f, ensure_ascii=False, indent=2)

    # 清理該成員的所有提醒
    reminders_path = os.path.join(base_dir, "reminders.json")
    with open(reminders_path) as f:
        reminders = json.load(f)

    removed_task_ids = []
    new_reminders = []
    for r in reminders:
        if r["member_id"] == member_id:
            removed_task_ids.append(r.get("scheduled_task_id"))
        else:
            new_reminders.append(r)

    with open(reminders_path, "w") as f:
        json.dump(new_reminders, f, ensure_ascii=False, indent=2)

    return member, removed_task_ids

def add_anniversary(base_dir, member_id, label, date_solar, recurring=True):
    """為成員添加紀念日"""
    path = os.path.join(base_dir, "members.json")
    with open(path) as f:
        members = json.load(f)

    for m in members:
        if m["id"] == member_id:
            anniversary = {
                "id": str(uuid.uuid4()),
                "label": label,
                "date_solar": date_solar,
                "recurring": recurring
            }
            m["anniversaries"].append(anniversary)
            m["updated_at"] = datetime.utcnow().isoformat() + "Z"

            with open(path, "w") as f:
                json.dump(members, f, ensure_ascii=False, indent=2)
            return anniversary
    return None

def remove_anniversary(base_dir, member_id, anniversary_id):
    """刪除成員的某個紀念日"""
    path = os.path.join(base_dir, "members.json")
    with open(path) as f:
        members = json.load(f)

    for m in members:
        if m["id"] == member_id:
            m["anniversaries"] = [
                a for a in m["anniversaries"] if a["id"] != anniversary_id
            ]
            m["updated_at"] = datetime.utcnow().isoformat() + "Z"

            with open(path, "w") as f:
                json.dump(members, f, ensure_ascii=False, indent=2)
            return True
    return False

from datetime import date, timedelta
import calendar

def _safe_replace_year(d: date, year: int) -> date:
    """安全替換年份，處理 2/29 在非閏年的情況（fallback 到 2/28）"""
    if d.month == 2 and d.day == 29 and not calendar.isleap(year):
        return date(year, 2, 28)
    return d.replace(year=year)

def upcoming_events(base_dir, days_ahead=30):
    """查詢未來 N 天內的生日和紀念日"""
    path = os.path.join(base_dir, "members.json")
    with open(path) as f:
        members = json.load(f)

    today = date.today()
    end_date = today + timedelta(days=days_ahead)
    events = []

    for m in members:
        # 新曆生日
        if m.get("birthday_solar"):
            bday = date.fromisoformat(m["birthday_solar"])
            this_year = _safe_replace_year(bday, today.year)
            # 處理跨年情況
            if this_year < today:
                this_year = _safe_replace_year(bday, today.year + 1)
            if today <= this_year <= end_date:
                age = this_year.year - bday.year
                events.append({
                    "date": str(this_year),
                    "name": m["name"],
                    "group": m["group"],
                    "type": "新曆生日",
                    "label": f"{m['name']} 的生日（{age} 歲）",
                    "member_id": m["id"]
                })

        # 農曆生日（使用預計算的當年新曆日期）
        if m.get("birthday_lunar_solar_this_year"):
            lunar_solar = date.fromisoformat(m["birthday_lunar_solar_this_year"])
            if today <= lunar_solar <= end_date:
                events.append({
                    "date": str(lunar_solar),
                    "name": m["name"],
                    "group": m["group"],
                    "type": "農曆生日",
                    "label": f"{m['name']} 的農曆生日（農曆{m['birthday_lunar']['month']}月{m['birthday_lunar']['day']}日）",
                    "member_id": m["id"]
                })

        # 紀念日
        for ann in m.get("anniversaries", []):
            if ann.get("recurring") and ann.get("date_solar"):
                aday = date.fromisoformat(ann["date_solar"])
                this_year = _safe_replace_year(aday, today.year)
                if this_year < today:
                    this_year = _safe_replace_year(aday, today.year + 1)
                if today <= this_year <= end_date:
                    years = this_year.year - aday.year
                    events.append({
                        "date": str(this_year),
                        "name": m["name"],
                        "group": m["group"],
                        "type": "紀念日",
                        "label": f"{m['name']} — {ann['label']}（第 {years} 年）",
                        "member_id": m["id"]
                    })

    return sorted(events, key=lambda e: e["date"])

def search_members(base_dir, query=None, group=None):
    """搜索成員。支持按名字模糊搜索和按組別篩選。"""
    path = os.path.join(base_dir, "members.json")
    with open(path) as f:
        members = json.load(f)

    results = members
    if group:
        results = [m for m in results if m["group"] == group]
    if query:
        query_lower = query.lower()
        results = [m for m in results if
                   query_lower in m["name"].lower() or
                   query_lower in m.get("relationship", "").lower() or
                   query_lower in m.get("notes", "").lower()]
    return results

def list_groups(base_dir):
    """列出所有組別及成員數量"""
    path = os.path.join(base_dir, "members.json")
    with open(path) as f:
        members = json.load(f)

    groups = {}
    for m in members:
        g = m["group"]
        groups[g] = groups.get(g, 0) + 1
    return groups

步驟：
1. 確定生日日期（新曆或農曆轉新曆）
2. 計算提醒日期 = 生日日期 - advance_days
3. 使用 mcp__scheduled-tasks__create_scheduled_task 創建任務
4. 將提醒信息寫入 reminders.json

taskId: "member-birthday-{member_name}-solar"
cronExpression: "0 9 {remind_day} {remind_month} *"
prompt: "🎂 提醒：{member_name} 的生日在 {advance_days} 天後（{birthday_month}/{birthday_day}）！請準備祝福！\n\n成員資料：\n- 關係：{relationship}\n- 組別：{group}\n- 備註：{notes}"
description: "{member_name} 生日提醒（提前 {advance_days} 天）"

taskId: "member-birthday-{member_name}-lunar"
cronExpression: "0 9 {remind_day} {remind_month} *"
prompt: "🎂 提醒：{member_name} 的農曆生日在 {advance_days} 天後（農曆{lunar_month}月{lunar_day}日，今年新曆{solar_date}）！\n\n成員資料：\n- 關係：{relationship}\n- 組別：{group}\n- 備註：{notes}"
description: "{member_name} 農曆生日提醒（提前 {advance_days} 天）"

taskId: "member-anniversary-{member_name}-{anniversary_label}"
cronExpression: "0 9 {remind_day} {remind_month} *"
prompt: "💝 提醒：{member_name} 的{anniversary_label}在 {advance_days} 天後（{date}）！\n\n成員資料：\n- 關係：{relationship}\n- 備註：{notes}"
description: "{member_name} {anniversary_label}提醒"

from datetime import date, timedelta
import calendar

def get_reminder_cron(target_month, target_day, advance_days):
    """計算提醒的 cron expression，正確處理跨月和跨年"""
    today = date.today()
    year = today.year
    m, d = int(target_month), int(target_day)

    # 處理 2/29 非閏年 fallback
    if m == 2 and d == 29 and not calendar.isleap(year):
        d = 28

    target = date(year, m, d)
    # 如果今年的目標日期已過，使用明年
    if target < today:
        year += 1
        if m == 2 and int(target_day) == 29 and not calendar.isleap(year):
            d = 28
        target = date(year, m, d)

    remind_date = target - timedelta(days=advance_days)
    # 返回 cron: 分 時 日 月 星期
    return f"0 9 {remind_date.day} {remind_date.month} *"

def save_reminder(base_dir, member_id, reminder_type, label, advance_days, task_id):
    """保存提醒記錄"""
    path = os.path.join(base_dir, "reminders.json")
    with open(path) as f:
        reminders = json.load(f)

    reminder = {
        "id": str(uuid.uuid4()),
        "member_id": member_id,
        "type": reminder_type,
        "label": label,
        "advance_days": advance_days,
        "scheduled_task_id": task_id,
        "enabled": True
    }
    reminders.append(reminder)

    with open(path, "w") as f:
        json.dump(reminders, f, ensure_ascii=False, indent=2)
    return reminder

taskId: "member-lunar-annual-update"
cronExpression: "0 10 1 1 *"
prompt: "執行 member-manager 年度農曆更新：
1. 讀取 members.json 中所有有農曆生日的成員
2. 使用 lunardate 將每個農曆生日轉換為今年的新曆日期
3. 更新 birthday_lunar_solar_this_year 欄位
4. 更新所有農曆生日提醒的 cron expression（因為新曆日期變了）
5. 回報更新結果"
description: "每年自動更新農曆生日對應的新曆日期"

📋 {group} 成員列表（共 N 人）

1. {name}（{relationship}）
   🎂 生日：{birthday_solar}（新曆）/ 農曆{lunar_month}月{lunar_day}日
   📞 {phone}
   📝 {notes}

2. ...

📅 未來 {days} 天的重要日子：

🎂 {date} — {name} 的生日（{age} 歲）
💝 {date} — {name} — {anniversary_label}（第 N 年）
...

共 N 個事件。

✅ 已添加成員：{name}（{group} / {relationship}）
   生日：{birthday_info}
   已設置提醒：提前 {advance_days} 天

✅ 已刪除成員：{name}
   已清理 {N} 個相關提醒

✅ 已更新 {N} 位成員的農曆生日新曆日期

用戶：幫我加一個成員，我媽媽，農曆三月初五生日

AI：
1. 解析信息：姓名=媽媽，關係=母親，農曆生日=三月初五
2. 計算今年新曆日期
3. 確認：
   「確認添加以下成員：
    - 姓名：媽媽
    - 組別：家庭
    - 關係：母親
    - 農曆生日：三月初五（今年新曆 {date}）
    是否正確？要設置生日提醒嗎？」
4. 用戶確認後存檔
5. 如果要提醒，詢問提前幾天（默認 3 天）
6. 創建 scheduled task + 保存 reminder

# 初始化時設置目錄權限
import stat
os.makedirs(base_dir, exist_ok=True)
os.chmod(base_dir, stat.S_IRWXU)  # 700: 僅 owner 可讀寫執行