Bridge script: export OpenClaw SQLite cron data to legacy flat files for ClawMetry

clawmetry-cron-bridge.py

#!/usr/bin/env python3
"""
Bridge: export OpenClaw SQLite cron data → legacy flat files for ClawMetry.
OpenClaw v2026.6.5+ stores everything in state/openclaw.sqlite.
ClawMetry v0.12.535 still reads cron/jobs.json and cron/runs/*.jsonl.
Run this periodically (e.g., every 15 min) to keep ClawMetry in sync.
"""

import sqlite3, json, os

OC_DIR = os.path.expanduser("~/.openclaw")
STATE_DB = os.path.join(OC_DIR, "state", "openclaw.sqlite")

# ── 1. Export cron jobs → ~/.openclaw/cron/jobs.json ──────────────
def export_jobs():
    conn = sqlite3.connect(STATE_DB)
    conn.row_factory = sqlite3.Row
    rows = conn.execute("SELECT * FROM cron_jobs").fetchall()
    conn.close()

    jobs = []
    for r in rows:
        job = json.loads(r["job_json"])
        state = json.loads(r["state_json"]) if r["state_json"] else {}
        job["state"] = {
            "lastStatus": state.get("lastStatus"),
            "lastRunAtMs": state.get("lastRunAtMs"),
            "nextRunAtMs": state.get("nextRunAtMs"),
            "lastDurationMs": state.get("lastDurationMs"),
            "lastError": state.get("lastError"),
            "consecutiveFailures": state.get("consecutiveFailures"),
        }
        jobs.append(job)

    path = os.path.join(OC_DIR, "cron", "jobs.json")
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w") as f:
        json.dump({"jobs": jobs}, f, indent=2)
    print(f"Exported {len(jobs)} cron jobs")

# ── 2. Export cron run logs → ~/.openclaw/cron/runs/<job_id>.jsonl ─
def export_runs():
    conn = sqlite3.connect(STATE_DB)
    conn.row_factory = sqlite3.Row
    rows = conn.execute("SELECT * FROM cron_run_logs ORDER BY ts DESC").fetchall()
    conn.close()

    runs_dir = os.path.join(OC_DIR, "cron", "runs")
    os.makedirs(runs_dir, exist_ok=True)

    by_job = {}
    for r in rows:
        jid = r["job_id"]
        by_job.setdefault(jid, [])
        entry = json.loads(r["entry_json"]) if r["entry_json"] else {}
        entry["ts"] = r["ts"]
        entry["status"] = r["status"]
        by_job[jid].append(entry)

    total = 0
    for jid, entries in by_job.items():
        entries.sort(key=lambda x: x.get("ts", 0))
        with open(os.path.join(runs_dir, f"{jid}.jsonl"), "w") as f:
            for e in entries:
                f.write(json.dumps(e) + "\n")
        total += len(entries)
    print(f"Exported {total} run logs across {len(by_job)} jobs")

# ── 3. Export task_runs → ~/.openclaw/tasks/runs.sqlite ────────────
def export_task_runs():
    src = STATE_DB
    dst = os.path.join(OC_DIR, "tasks", "runs.sqlite")
    os.makedirs(os.path.dirname(dst), exist_ok=True)

    conn_src = sqlite3.connect(f"file:{src}?mode=ro", uri=True)
    conn_dst = sqlite3.connect(dst)
    conn_dst.execute("DROP TABLE IF EXISTS task_runs")

    schema = conn_src.execute(
        "SELECT sql FROM sqlite_master WHERE name='task_runs'"
    ).fetchone()[0]
    conn_dst.execute(schema)

    rows = conn_src.execute("SELECT * FROM task_runs").fetchall()
    cols = [
        str(d[1])
        for d in conn_src.execute("PRAGMA table_info(task_runs)").fetchall()
    ]
    placeholders = ",".join(["?"] * len(cols))
    cols_str = ",".join(cols)
    conn_dst.executemany(
        f"INSERT INTO task_runs ({cols_str}) VALUES ({placeholders})", rows
    )
    conn_dst.commit()
    cnt = conn_dst.execute("SELECT COUNT(*) FROM task_runs").fetchone()[0]
    conn_src.close()
    conn_dst.close()
    print(f"Exported {cnt} task runs")

# ── Run all ────────────────────────────────────────────────────────
if __name__ == "__main__":
    export_jobs()
    export_runs()
    export_task_runs()
    print("ClawMetry cron data refresh complete")