openclaw-ops/scripts/google-sync.py

#!/usr/bin/env python3
import json
import re
import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import Any

WORKSPACE = Path('/home/node/.openclaw/workspace')
STATE_PATH = WORKSPACE / 'state' / 'projects.json'
REMINDER_LIST = Path('/home/node/.openclaw/skills/reminder/scripts/list.sh')


@dataclass
class Reminder:
    when_local: str
    message: str
    reminder_id: str


def run(cmd: list[str]) -> str:
    res = subprocess.run(cmd, text=True, capture_output=True)
    if res.returncode != 0:
        raise RuntimeError((res.stderr or res.stdout).strip())
    return res.stdout


def load_state() -> dict[str, Any]:
    return json.loads(STATE_PATH.read_text())


def save_state(state: dict[str, Any]) -> None:
    STATE_PATH.write_text(json.dumps(state, indent=2) + "\n")


def slug(text: str) -> str:
    return re.sub(r'[^a-z0-9]+', '-', text.lower()).strip('-')


def parse_reminders() -> list[Reminder]:
    out = run(['bash', str(REMINDER_LIST)])
    reminders: list[Reminder] = []
    current_when = None
    current_msg = None
    current_id = None
    for line in out.splitlines():
        if line.startswith('⏰ '):
            current_when = line.replace('⏰ ', '', 1).strip()
            current_msg = None
            current_id = None
        elif line.strip().startswith('ID:'):
            current_id = line.split('ID:', 1)[1].strip()
            if current_when and current_msg and current_id:
                reminders.append(Reminder(when_local=current_when, message=current_msg, reminder_id=current_id))
                current_when = None
                current_msg = None
                current_id = None
        elif current_when and line.startswith('   ') and current_msg is None and line.strip() and not line.strip().startswith('ID:'):
            current_msg = line.strip()
    return reminders


def ensure_followup_task(state: dict[str, Any], title: str, notes: str = '') -> str:
    tasklist = state['google']['tasklists']['claw_follow_ups']['id']
    out = run([
        'gws', 'tasks', 'tasks', 'insert',
        '--params', json.dumps({'tasklist': tasklist}),
        '--json', json.dumps({'title': title, 'notes': notes})
    ])
    obj = json.loads(out)
    return obj['id']


def ensure_calendar_event(state: dict[str, Any], summary: str, start_utc: str, end_utc: str, description: str = '') -> str:
    cal_id = state['google']['calendar']['claw_ops']['id']
    existing_raw = run([
        'gws', 'calendar', 'events', 'list',
        '--params', json.dumps({'calendarId': cal_id})
    ])
    existing = json.loads(existing_raw)
    for item in existing.get('items', []):
        if item.get('summary') == summary and item.get('start', {}).get('dateTime') == start_utc:
            return item['id']
    out = run([
        'gws', 'calendar', 'events', 'insert',
        '--params', json.dumps({'calendarId': cal_id}),
        '--json', json.dumps({
            'summary': summary,
            'description': description,
            'start': {'dateTime': start_utc, 'timeZone': 'UTC'},
            'end': {'dateTime': end_utc, 'timeZone': 'UTC'}
        })
    ])
    obj = json.loads(out)
    return obj['id']


def sync_projects(state: dict[str, Any]) -> dict[str, Any]:
    project_list_id = state['google']['tasklists']['claw_projects']['id']
    created = []
    for project in state['projects']:
        ids = set(project.get('task_ids', []))
        if not ids:
            title = f"[Project] {project['title']}"
            notes = f"Project ID: {project['id']}\nStatus: {project['status']}\nNext action: {project.get('next_action', '')}\nNotes ref: {project.get('notes_ref', '')}"
            out = run([
                'gws', 'tasks', 'tasks', 'insert',
                '--params', json.dumps({'tasklist': project_list_id}),
                '--json', json.dumps({'title': title, 'notes': notes})
            ])
            obj = json.loads(out)
            project.setdefault('task_ids', []).append(obj['id'])
            created.append({'type': 'project-task', 'project_id': project['id'], 'task_id': obj['id']})
        if project.get('next_action') and len(project.get('task_ids', [])) < 2:
            title = f"[{project['title']}] {project['next_action']}"
            notes = f"Project ID: {project['id']}\nSource: {project.get('notes_ref', '')}"
            out = run([
                'gws', 'tasks', 'tasks', 'insert',
                '--params', json.dumps({'tasklist': project_list_id}),
                '--json', json.dumps({'title': title, 'notes': notes})
            ])
            obj = json.loads(out)
            project.setdefault('task_ids', []).append(obj['id'])
            created.append({'type': 'next-action-task', 'project_id': project['id'], 'task_id': obj['id']})
    return {'created': created}


def sync_reminders(state: dict[str, Any]) -> dict[str, Any]:
    reminder_state = state.setdefault('reminders', {'synced': {}})
    synced = reminder_state.setdefault('synced', {})
    created = []
    skipped = []
    for rem in parse_reminders():
        if rem.reminder_id in synced:
            continue
        msg = rem.message.replace('\\n', '\n')
        lower = msg.lower()
        if 'vehicle registration renewal' in lower:
            # known time conversion from existing reminder local labels
            if '09:00 cdt' in rem.when_local.lower():
                start, end = '2026-04-15T14:00:00Z', '2026-04-15T14:30:00Z'
            elif '16:00 cdt' in rem.when_local.lower():
                start, end = '2026-04-15T21:00:00Z', '2026-04-15T21:30:00Z'
            else:
                skipped.append({'reminder_id': rem.reminder_id, 'reason': 'unsupported-known-reminder-time'})
                continue
            event_id = ensure_calendar_event(state, 'Vehicle registration renewal - Tesla Model Y (VJF3166)', start, end, msg)
            synced[rem.reminder_id] = {'kind': 'calendar', 'event_id': event_id}
            created.append({'reminder_id': rem.reminder_id, 'calendar_event_id': event_id})
        else:
            skipped.append({'reminder_id': rem.reminder_id, 'reason': 'past-or-unscheduled-manual-review'})
    return {'created': created, 'skipped': skipped}


def main() -> None:
    state = load_state()
    result = {
        'projects': sync_projects(state),
        'reminders': sync_reminders(state),
    }
    save_state(state)
    print(json.dumps(result, indent=2))


if __name__ == '__main__':
    main()