Files
token-analytics/jsonl-sync/export.mjs
T

142 lines
4.1 KiB
JavaScript

import fs from 'node:fs';
import path from 'node:path';
import { fileURLToPath } from 'node:url';
import Database from 'better-sqlite3';
const __dirname = path.dirname(fileURLToPath(import.meta.url));
import os from 'node:os';
const SESSIONS_DIR = process.env.SESSIONS_DIR || path.join(os.homedir(), '.openclaw/agents/main/sessions');
const DB_FILE = path.join(__dirname, 'sessions.db');
const db = new Database(DB_FILE);
// Schema
db.exec(`
CREATE TABLE IF NOT EXISTS entries (
id TEXT PRIMARY KEY,
session TEXT NOT NULL,
type TEXT,
timestamp TEXT,
data TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_session ON entries (session);
CREATE INDEX IF NOT EXISTS idx_type ON entries (type);
CREATE INDEX IF NOT EXISTS idx_timestamp ON entries (timestamp);
CREATE TABLE IF NOT EXISTS state (
file TEXT PRIMARY KEY,
lines INTEGER NOT NULL DEFAULT 0
);
`);
// Add chainId column if missing (migration)
try { db.exec('ALTER TABLE entries ADD COLUMN chainId TEXT'); } catch {}
try { db.exec('CREATE INDEX IF NOT EXISTS idx_chainId ON entries (chainId)'); } catch {}
const getState = db.prepare('SELECT lines FROM state WHERE file = ?');
const setState = db.prepare('INSERT OR REPLACE INTO state (file, lines) VALUES (?, ?)');
const insertEntry = db.prepare(`
INSERT OR IGNORE INTO entries (id, session, type, timestamp, data)
VALUES (@id, @session, @type, @timestamp, @data)
`);
const getEntry = db.prepare('SELECT id, data FROM entries WHERE id = ?');
const setChainId = db.prepare('UPDATE entries SET chainId = @chainId WHERE id = @id');
const getUnchained = db.prepare(`
SELECT id, data FROM entries
WHERE chainId IS NULL
ORDER BY timestamp ASC
`);
// --- Step 1: Import new entries ---
const jsonlFiles = fs.readdirSync(SESSIONS_DIR)
.filter(f => f.endsWith('.jsonl') && !f.includes('.deleted'));
let exported = 0;
const runImport = db.transaction(() => {
for (const file of jsonlFiles) {
const filePath = path.join(SESSIONS_DIR, file);
const sessionName = file.replace('.jsonl', '');
const lines = fs.readFileSync(filePath, 'utf-8').split('\n').filter(l => l.trim());
const lastProcessed = getState.get(file)?.lines ?? 0;
for (let i = lastProcessed; i < lines.length; i++) {
try {
const entry = JSON.parse(lines[i]);
if (!entry.timestamp) continue;
const id = entry.id || `${sessionName}-${i}`;
insertEntry.run({
id,
session: sessionName,
type: entry.type ?? 'unknown',
timestamp: entry.timestamp,
data: JSON.stringify(entry),
});
exported++;
} catch { /* skip */ }
}
setState.run(file, lines.length);
}
});
runImport();
// --- Step 2: Compute chainId for entries that don't have one ---
// chainId = the nearest user message when walking up the parentId chain.
// If no user message found, chainId = own id.
let chained = 0;
const chainCache = new Map(); // id -> chainId
function resolveChainId(id) {
if (chainCache.has(id)) return chainCache.get(id);
const visited = new Set();
let currentId = id;
while (currentId && !visited.has(currentId)) {
visited.add(currentId);
if (chainCache.has(currentId) && currentId !== id) {
// Already resolved - reuse
const result = chainCache.get(currentId);
chainCache.set(id, result);
return result;
}
const row = getEntry.get(currentId);
if (!row) break;
const d = JSON.parse(row.data);
// If this is a user message, it's the chain root
if (d.message?.role === 'user') {
chainCache.set(id, currentId);
return currentId;
}
currentId = d.parentId || null;
}
// No user message found - self is root
chainCache.set(id, id);
return id;
}
const runChaining = db.transaction(() => {
const unchained = getUnchained.all();
for (const row of unchained) {
const chainId = resolveChainId(row.id);
setChainId.run({ id: row.id, chainId });
chained++;
}
});
runChaining();
db.close();
console.log(`Exportiert: ${exported} neue, ${chained} verkettet → ${DB_FILE}`);