325 lines
12 KiB
C#
325 lines
12 KiB
C#
using Dapper;
|
|
using Microsoft.Data.Sqlite;
|
|
|
|
var builder = WebApplication.CreateBuilder(args);
|
|
|
|
builder.Configuration.AddJsonFile("appsettings.user.json", optional: true, reloadOnChange: true);
|
|
|
|
var dbPath = builder.Configuration["DbPath"]!;
|
|
|
|
builder.Services.AddSingleton<Func<SqliteConnection>>(() =>
|
|
new SqliteConnection($"Data Source={dbPath}"));
|
|
|
|
var app = builder.Build();
|
|
|
|
app.UseDefaultFiles();
|
|
app.UseStaticFiles();
|
|
|
|
// GET /api/settings
|
|
app.MapGet("/api/settings", (IConfiguration config) =>
|
|
{
|
|
var anchorDay = int.TryParse(config["AnchorDay"], out var d) ? d : 0;
|
|
var anchorHour = int.TryParse(config["AnchorHour"], out var h) ? h : 0;
|
|
return Results.Ok(new { anchorDay, anchorHour });
|
|
});
|
|
|
|
// GET /api/cost-buckets?from=&to=&bucketMinutes=
|
|
// Returns N buckets of aggregated cost
|
|
app.MapGet("/api/cost-buckets", (Func<SqliteConnection> factory, string from, string to, int bucketMinutes) =>
|
|
{
|
|
using var db = factory();
|
|
var rows = db.Query(@"
|
|
SELECT
|
|
CAST((strftime('%s', timestamp) - strftime('%s', @from)) / (@bucketSec) AS INTEGER) AS bucket,
|
|
SUM(CAST(json_extract(data, '$.message.usage.cost.total') AS REAL)) AS cost,
|
|
COUNT(*) AS count,
|
|
MIN(timestamp) AS bucketStart,
|
|
MAX(timestamp) AS bucketEnd
|
|
FROM entries
|
|
WHERE json_extract(data, '$.message.usage.cost.total') IS NOT NULL
|
|
AND timestamp >= @from AND timestamp < @to
|
|
GROUP BY bucket
|
|
ORDER BY bucket ASC",
|
|
new { from, to, bucketSec = bucketMinutes * 60 });
|
|
|
|
return Results.Ok(rows);
|
|
});
|
|
|
|
// GET /api/cost-entries?from=&to=
|
|
app.MapGet("/api/cost-entries", (Func<SqliteConnection> factory, string from, string to) =>
|
|
{
|
|
using var db = factory();
|
|
var rows = db.Query(@"
|
|
SELECT id, session, type, timestamp,
|
|
json_extract(data, '$.message.usage.cost.total') AS cost,
|
|
json_extract(data, '$.message.usage.input') AS inputTokens,
|
|
json_extract(data, '$.message.usage.output') AS outputTokens,
|
|
json_extract(data, '$.message.model') AS model,
|
|
chainId,
|
|
data
|
|
FROM entries
|
|
WHERE json_extract(data, '$.message.usage.cost.total') IS NOT NULL
|
|
AND timestamp >= @from AND timestamp < @to
|
|
ORDER BY timestamp ASC",
|
|
new { from, to });
|
|
|
|
return Results.Ok(rows);
|
|
});
|
|
|
|
// GET /api/cost-chains?from=&to=
|
|
// Returns chains with aggregated cost, sorted by total cost desc
|
|
app.MapGet("/api/cost-chains", (Func<SqliteConnection> factory, string from, string to, int limit = 50, int offset = 0) =>
|
|
{
|
|
using var db = factory();
|
|
var rows = db.Query(@"
|
|
SELECT c.chainId,
|
|
c.totalCost,
|
|
c.count,
|
|
c.first,
|
|
c.last,
|
|
e.data AS chainData
|
|
FROM (
|
|
SELECT chainId,
|
|
SUM(CAST(json_extract(data, '$.message.usage.cost.total') AS REAL)) AS totalCost,
|
|
COUNT(*) AS count,
|
|
MIN(timestamp) AS first,
|
|
MAX(timestamp) AS last
|
|
FROM entries
|
|
WHERE json_extract(data, '$.message.usage.cost.total') IS NOT NULL
|
|
AND timestamp >= @from AND timestamp < @to
|
|
AND chainId IS NOT NULL
|
|
GROUP BY chainId
|
|
ORDER BY totalCost DESC
|
|
LIMIT @limit OFFSET @offset
|
|
) c
|
|
LEFT JOIN entries e ON e.id = c.chainId",
|
|
new { from, to, limit, offset });
|
|
|
|
var result = rows.Select((dynamic r) =>
|
|
{
|
|
string? userMsg = null;
|
|
try
|
|
{
|
|
string raw = r.chainData;
|
|
var doc = System.Text.Json.JsonDocument.Parse(raw);
|
|
var msg = doc.RootElement.GetProperty("message");
|
|
var content = msg.GetProperty("content");
|
|
string fullText;
|
|
if (content.ValueKind == System.Text.Json.JsonValueKind.String)
|
|
fullText = content.GetString()!;
|
|
else
|
|
{
|
|
fullText = string.Join("\n", content.EnumerateArray()
|
|
.Where(c => c.TryGetProperty("type", out var t) && t.GetString() == "text")
|
|
.Select(c => c.GetProperty("text").GetString()));
|
|
}
|
|
|
|
userMsg = StripSystemPrefixes(fullText);
|
|
if (userMsg?.Length > 200) userMsg = userMsg[..200] + "...";
|
|
}
|
|
catch { }
|
|
|
|
return new
|
|
{
|
|
chainId = (string)r.chainId,
|
|
totalCost = (double)r.totalCost,
|
|
count = (long)r.count,
|
|
first = (string)r.first,
|
|
last = (string)r.last,
|
|
userMessage = userMsg ?? ""
|
|
};
|
|
});
|
|
|
|
return Results.Ok(result);
|
|
});
|
|
|
|
// GET /api/cost-summary?from=&to=
|
|
app.MapGet("/api/cost-summary", (Func<SqliteConnection> factory, string? from, string? to) =>
|
|
{
|
|
using var db = factory();
|
|
var conditions = new List<string> { "json_extract(data, '$.message.usage.cost.total') IS NOT NULL" };
|
|
if (from != null) conditions.Add("timestamp >= @from");
|
|
if (to != null) conditions.Add("timestamp < @to");
|
|
var where = string.Join(" AND ", conditions);
|
|
|
|
var row = db.QuerySingle($@"
|
|
SELECT COALESCE(SUM(CAST(json_extract(data, '$.message.usage.cost.total') AS REAL)), 0) AS totalCost,
|
|
COUNT(*) AS totalEntries
|
|
FROM entries
|
|
WHERE {where}",
|
|
new { from, to });
|
|
|
|
return Results.Ok(row);
|
|
});
|
|
|
|
// GET /api/entry-chain/{id} - full chain: walk back to user message, then forward to end
|
|
app.MapGet("/api/entry-chain/{id}", (string id, Func<SqliteConnection> factory) =>
|
|
{
|
|
using var db = factory();
|
|
|
|
// Step 1: Walk backwards to find the user message (chain root)
|
|
var backChain = new List<dynamic>();
|
|
var currentId = id;
|
|
var seen = new HashSet<string>();
|
|
string? rootId = null;
|
|
|
|
while (currentId != null && seen.Add(currentId))
|
|
{
|
|
var row = db.QueryFirstOrDefault(
|
|
"SELECT id, session, type, timestamp, data, json_extract(data, '$.parentId') AS parentId FROM entries WHERE id = @id",
|
|
new { id = currentId });
|
|
|
|
if (row == null) break;
|
|
|
|
backChain.Add(row);
|
|
|
|
string dataStr = row.data;
|
|
try
|
|
{
|
|
var parsed = System.Text.Json.JsonDocument.Parse(dataStr);
|
|
var role = parsed.RootElement.TryGetProperty("message", out var msg)
|
|
&& msg.TryGetProperty("role", out var r) ? r.GetString() : null;
|
|
if (role == "user")
|
|
{
|
|
rootId = (string)row.id;
|
|
break;
|
|
}
|
|
}
|
|
catch { }
|
|
|
|
currentId = (string?)row.parentId;
|
|
}
|
|
|
|
backChain.Reverse();
|
|
|
|
// Step 2: Walk forwards from the last entry in backChain until we hit the next user message
|
|
var chainIds = new HashSet<string>(backChain.Select(r => (string)r.id));
|
|
var forwardEntries = new List<dynamic>();
|
|
|
|
var lastId = (string)backChain.Last().id;
|
|
var maxIter = 100;
|
|
var currentId2 = lastId;
|
|
while (maxIter-- > 0)
|
|
{
|
|
var child = db.QueryFirstOrDefault(
|
|
"SELECT id, session, type, timestamp, data, json_extract(data, '$.message.role') AS role FROM entries WHERE json_extract(data, '$.parentId') = @pid",
|
|
new { pid = currentId2 });
|
|
|
|
if (child == null) break;
|
|
|
|
string childRole = child.role;
|
|
// Stop when we reach the next user message (new turn)
|
|
if (childRole == "user") break;
|
|
|
|
string childId = child.id;
|
|
if (!chainIds.Add(childId)) break;
|
|
|
|
forwardEntries.Add(new { id = childId, session = (string)child.session, type = (string)child.type, timestamp = (string)child.timestamp, data = (string)child.data });
|
|
currentId2 = childId;
|
|
}
|
|
|
|
// Combine and sort chronologically
|
|
var fullChain = backChain.Concat(forwardEntries)
|
|
.OrderBy(r => (string)r.timestamp)
|
|
.Select(r => new { id = (string)r.id, session = (string)r.session, type = (string)r.type, timestamp = (string)r.timestamp, data = (string)r.data })
|
|
.ToList();
|
|
|
|
return Results.Ok(fullChain);
|
|
});
|
|
|
|
// GET /api/entry-chains?from=&to= - returns cost entries with cached chainId
|
|
app.MapGet("/api/entry-chains", (Func<SqliteConnection> factory, string from, string to) =>
|
|
{
|
|
using var db = factory();
|
|
var rows = db.Query(@"
|
|
SELECT id, session, type, timestamp,
|
|
json_extract(data, '$.message.usage.cost.total') AS cost,
|
|
json_extract(data, '$.message.usage.input') AS inputTokens,
|
|
json_extract(data, '$.message.usage.output') AS outputTokens,
|
|
json_extract(data, '$.message.model') AS model,
|
|
chainId,
|
|
data
|
|
FROM entries
|
|
WHERE json_extract(data, '$.message.usage.cost.total') IS NOT NULL
|
|
AND timestamp >= @from AND timestamp < @to
|
|
ORDER BY timestamp ASC",
|
|
new { from, to }).AsList();
|
|
|
|
// Map chainId to sequential index (for coloring)
|
|
var chainIndex = new Dictionary<string, int>();
|
|
int nextChain = 0;
|
|
var result = rows.Select(e =>
|
|
{
|
|
string cid = e.chainId ?? e.id;
|
|
if (!chainIndex.ContainsKey(cid)) chainIndex[cid] = nextChain++;
|
|
return new
|
|
{
|
|
e.id, e.session, e.type, e.timestamp,
|
|
e.cost, e.inputTokens, e.outputTokens, e.model,
|
|
e.data,
|
|
chainId = chainIndex[cid]
|
|
};
|
|
});
|
|
|
|
return Results.Ok(result);
|
|
});
|
|
|
|
// POST /api/sync - run export.mjs to import new entries
|
|
app.MapPost("/api/sync", async () =>
|
|
{
|
|
var config = app.Configuration;
|
|
var wd = config["SyncWorkingDirectory"] ?? "/opt/token-analytics/jsonl-sync";
|
|
var fileName = config["SyncCommandFileName"] ?? "node";
|
|
var arguments = config["SyncCommandArguments"] ?? "/opt/token-analytics/jsonl-sync/export.mjs";
|
|
var psi = new System.Diagnostics.ProcessStartInfo(fileName, arguments)
|
|
{
|
|
RedirectStandardOutput = true,
|
|
RedirectStandardError = true,
|
|
WorkingDirectory = wd
|
|
};
|
|
var proc = System.Diagnostics.Process.Start(psi)!;
|
|
var stdout = await proc.StandardOutput.ReadToEndAsync();
|
|
var stderr = await proc.StandardError.ReadToEndAsync();
|
|
await proc.WaitForExitAsync();
|
|
return Results.Ok(new { exitCode = proc.ExitCode, output = (stdout + stderr).Trim() });
|
|
});
|
|
|
|
app.Run();
|
|
|
|
static string StripSystemPrefixes(string text)
|
|
{
|
|
var lines = text.Split('\n').ToList();
|
|
var inCodeBlock = false;
|
|
|
|
// Strip from top
|
|
while (lines.Count > 0)
|
|
{
|
|
var t = lines[0].Trim();
|
|
if (string.IsNullOrEmpty(t)) { lines.RemoveAt(0); continue; }
|
|
// [Direct ...], [Current ...], [System ...], [message_id ...], [message-id ...], [Chat messages ...], [Queued messages ...]
|
|
if (t.StartsWith('[') && (t.StartsWith("[Direct ") || t.StartsWith("[Current ") ||
|
|
t.StartsWith("[System ") || t.StartsWith("[message_id:") || t.StartsWith("[message-id ") ||
|
|
t.StartsWith("[Chat messages ") || t.StartsWith("[Queued messages ")))
|
|
{ lines.RemoveAt(0); continue; }
|
|
// [Telegram ... id:xxx ...] msg - strip only the bracket prefix, keep the message after ]
|
|
if (t.StartsWith("[Telegram ") && t.Contains(']'))
|
|
{
|
|
var afterBracket = t[(t.IndexOf(']') + 1)..].Trim();
|
|
if (!string.IsNullOrEmpty(afterBracket)) { lines[0] = afterBracket; break; }
|
|
lines.RemoveAt(0); continue;
|
|
}
|
|
// "Conversation info", "Queued #N", "---", "Assistant:"
|
|
if (t.StartsWith("Conversation info") || t.StartsWith("Queued #") ||
|
|
t.StartsWith("Assistant:") || t == "---") { lines.RemoveAt(0); continue; }
|
|
// Code block boundaries and JSON content inside
|
|
if (t == "```json" || t == "```") { inCodeBlock = !inCodeBlock; lines.RemoveAt(0); continue; }
|
|
if (inCodeBlock) { lines.RemoveAt(0); continue; }
|
|
// "User: " prefix
|
|
if (t.StartsWith("User: ")) { lines[0] = t["User: ".Length..]; break; }
|
|
break;
|
|
}
|
|
|
|
var result = string.Join("\n", lines).Trim();
|
|
return string.IsNullOrEmpty(result) ? text.Split('\n').LastOrDefault(l => !string.IsNullOrWhiteSpace(l))?.Trim() ?? "" : result;
|
|
}
|