mirror of
https://github.com/Routstr/routstr-core.git
synced 2026-10-05 20:28:23 +00:00
403 lines
13 KiB
Python
403 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Nostr usage analytics publisher.
|
|
Publishes a single replaceable analytics snapshot for each provider.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import time
|
|
from typing import Any
|
|
|
|
from ..core import get_logger
|
|
from ..core.log_manager import log_manager
|
|
from ..core.settings import settings
|
|
from .listing import nsec_to_keypair, publish_to_relay
|
|
from .sdk import create_signed_event
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
ANALYTICS_KIND = 38422
|
|
ANALYTICS_SCHEMA = "routstr.analytics.snapshot.v1"
|
|
DEFAULT_RELAYS = [
|
|
"wss://relay.nostr.band",
|
|
"wss://relay.damus.io",
|
|
"wss://relay.routstr.com",
|
|
"wss://nos.lol",
|
|
]
|
|
PUBLISH_INTERVAL_SECONDS = 15 * 60
|
|
DISABLED_POLL_SECONDS = 60
|
|
DASHBOARD_WINDOW_HOURS = 24
|
|
DASHBOARD_INTERVAL_MINUTES = 60
|
|
MODEL_LIMIT = 20
|
|
WINDOW_DEFINITIONS: tuple[tuple[str, int, int], ...] = (
|
|
("24h", 24, 60),
|
|
("7d", 7 * 24, 6 * 60),
|
|
("30d", 30 * 24, 24 * 60),
|
|
("3m", 90 * 24, 24 * 60),
|
|
("1y", 365 * 24, 7 * 24 * 60),
|
|
)
|
|
|
|
|
|
def _resolve_provider_id(public_key_hex: str) -> str:
|
|
explicit_provider_id = (settings.provider_id or "").strip()
|
|
if explicit_provider_id:
|
|
return explicit_provider_id
|
|
return public_key_hex[:12]
|
|
|
|
|
|
def _resolve_endpoint_urls() -> list[str]:
|
|
urls: list[str] = []
|
|
http_url = (settings.http_url or "").strip()
|
|
onion_url = (settings.onion_url or "").strip()
|
|
|
|
if http_url and http_url != "http://localhost:8000":
|
|
urls.append(http_url)
|
|
|
|
if onion_url:
|
|
if onion_url.endswith(".onion") and not (
|
|
onion_url.startswith("http://") or onion_url.startswith("https://")
|
|
):
|
|
onion_url = f"http://{onion_url}"
|
|
urls.append(onion_url)
|
|
|
|
return urls
|
|
|
|
|
|
def _resolve_relays() -> list[str]:
|
|
configured = [url.strip() for url in settings.relays if url.strip()]
|
|
return configured if configured else list(DEFAULT_RELAYS)
|
|
|
|
|
|
def _to_int(value: Any) -> int:
|
|
if isinstance(value, bool):
|
|
return int(value)
|
|
if isinstance(value, int):
|
|
return value
|
|
if isinstance(value, float):
|
|
return int(value)
|
|
if isinstance(value, str):
|
|
try:
|
|
return int(float(value))
|
|
except ValueError:
|
|
return 0
|
|
return 0
|
|
|
|
|
|
def _to_float(value: Any) -> float:
|
|
if isinstance(value, bool):
|
|
return float(int(value))
|
|
if isinstance(value, (int, float)):
|
|
return float(value)
|
|
if isinstance(value, str):
|
|
try:
|
|
return float(value)
|
|
except ValueError:
|
|
return 0.0
|
|
return 0.0
|
|
|
|
|
|
def _aggregate_top_model_usage(
|
|
model_usage_mix: dict[str, Any],
|
|
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
|
|
top_models_raw = model_usage_mix.get("top_models", [])
|
|
mix_metrics_raw = model_usage_mix.get("metrics", [])
|
|
|
|
top_models = [model for model in top_models_raw if isinstance(model, str)]
|
|
metrics = [row for row in mix_metrics_raw if isinstance(row, dict)]
|
|
|
|
model_totals: dict[str, dict[str, float | int]] = {
|
|
model: {
|
|
"successful_requests": 0,
|
|
"revenue_msats": 0.0,
|
|
"total_tokens": 0,
|
|
}
|
|
for model in top_models
|
|
}
|
|
others = {
|
|
"successful_requests": 0,
|
|
"revenue_msats": 0.0,
|
|
"total_tokens": 0,
|
|
}
|
|
|
|
for metric in metrics:
|
|
model_counts = metric.get("model_counts", {})
|
|
model_revenue = metric.get("model_revenue_msats", {})
|
|
model_tokens = metric.get("model_tokens", {})
|
|
|
|
if isinstance(model_counts, dict):
|
|
for model, count in model_counts.items():
|
|
if model in model_totals:
|
|
model_totals[model]["successful_requests"] += _to_int(count)
|
|
|
|
if isinstance(model_revenue, dict):
|
|
for model, amount in model_revenue.items():
|
|
if model in model_totals:
|
|
model_totals[model]["revenue_msats"] += _to_float(amount)
|
|
|
|
if isinstance(model_tokens, dict):
|
|
for model, token_count in model_tokens.items():
|
|
if model in model_totals:
|
|
model_totals[model]["total_tokens"] += _to_int(token_count)
|
|
|
|
others["successful_requests"] += _to_int(metric.get("others", 0))
|
|
others["revenue_msats"] += _to_float(metric.get("others_revenue_msats", 0.0))
|
|
others["total_tokens"] += _to_int(metric.get("others_tokens", 0))
|
|
|
|
model_rows = [
|
|
{
|
|
"model": model,
|
|
"successful_requests": int(values["successful_requests"]),
|
|
"revenue_msats": float(values["revenue_msats"]),
|
|
"total_tokens": int(values["total_tokens"]),
|
|
}
|
|
for model, values in model_totals.items()
|
|
]
|
|
model_rows.sort(
|
|
key=lambda row: _to_int(row.get("successful_requests", 0)),
|
|
reverse=True,
|
|
)
|
|
|
|
return model_rows, others
|
|
|
|
|
|
def _build_summary_payload(summary: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"total_requests": _to_int(summary.get("total_requests", 0)),
|
|
"successful_chat_completions": _to_int(
|
|
summary.get("successful_chat_completions", 0)
|
|
),
|
|
"failed_requests": _to_int(summary.get("failed_requests", 0)),
|
|
"success_rate": _to_float(summary.get("success_rate", 0.0)),
|
|
"unique_models_count": _to_int(summary.get("unique_models_count", 0)),
|
|
"input_tokens": _to_int(summary.get("input_tokens", 0)),
|
|
"output_tokens": _to_int(summary.get("output_tokens", 0)),
|
|
"total_tokens": _to_int(summary.get("total_tokens", 0)),
|
|
"revenue_msats": _to_float(summary.get("revenue_msats", 0.0)),
|
|
"refunds_msats": _to_float(summary.get("refunds_msats", 0.0)),
|
|
"net_revenue_msats": _to_float(summary.get("net_revenue_msats", 0.0)),
|
|
"revenue_sats": _to_float(summary.get("revenue_sats", 0.0)),
|
|
"refunds_sats": _to_float(summary.get("refunds_sats", 0.0)),
|
|
"net_revenue_sats": _to_float(summary.get("net_revenue_sats", 0.0)),
|
|
}
|
|
|
|
|
|
def _build_window_payload(
|
|
*,
|
|
hours: int,
|
|
interval_minutes: int,
|
|
model_limit: int,
|
|
) -> dict[str, Any]:
|
|
dashboard = log_manager.get_usage_dashboard(
|
|
interval=interval_minutes,
|
|
hours=hours,
|
|
error_limit=1,
|
|
model_limit=model_limit,
|
|
)
|
|
|
|
summary = dashboard.get("summary", {})
|
|
model_usage_mix = dashboard.get("model_usage_mix", {})
|
|
|
|
summary_payload = _build_summary_payload(summary if isinstance(summary, dict) else {})
|
|
usage_mix_payload = model_usage_mix if isinstance(model_usage_mix, dict) else {}
|
|
top_model_usage, others_usage = _aggregate_top_model_usage(usage_mix_payload)
|
|
|
|
return {
|
|
"window_hours": hours,
|
|
"interval_minutes": interval_minutes,
|
|
"summary": summary_payload,
|
|
"model_usage_mix": usage_mix_payload,
|
|
"top_model_usage": top_model_usage,
|
|
"others_usage": others_usage,
|
|
}
|
|
|
|
|
|
def build_stats_snapshot_payload(
|
|
provider_id: str,
|
|
*,
|
|
public_key_hex: str,
|
|
generated_at: int,
|
|
window_hours: int = DASHBOARD_WINDOW_HOURS,
|
|
interval_minutes: int = DASHBOARD_INTERVAL_MINUTES,
|
|
model_limit: int = MODEL_LIMIT,
|
|
) -> dict[str, Any]:
|
|
_ = (window_hours, interval_minutes)
|
|
windows: dict[str, dict[str, Any]] = {}
|
|
for key, hours, window_interval_minutes in WINDOW_DEFINITIONS:
|
|
windows[key] = _build_window_payload(
|
|
hours=hours,
|
|
interval_minutes=window_interval_minutes,
|
|
model_limit=model_limit,
|
|
)
|
|
|
|
primary_window = windows.get("24h", {})
|
|
summary_payload = (
|
|
primary_window.get("summary", {})
|
|
if isinstance(primary_window.get("summary", {}), dict)
|
|
else {}
|
|
)
|
|
usage_mix_payload = (
|
|
primary_window.get("model_usage_mix", {})
|
|
if isinstance(primary_window.get("model_usage_mix", {}), dict)
|
|
else {}
|
|
)
|
|
top_model_usage = (
|
|
primary_window.get("top_model_usage", [])
|
|
if isinstance(primary_window.get("top_model_usage", []), list)
|
|
else []
|
|
)
|
|
others_usage = (
|
|
primary_window.get("others_usage", {})
|
|
if isinstance(primary_window.get("others_usage", {}), dict)
|
|
else {}
|
|
)
|
|
|
|
return {
|
|
"schema": ANALYTICS_SCHEMA,
|
|
"generated_at": generated_at,
|
|
"provider_id": provider_id,
|
|
"pubkey": public_key_hex,
|
|
"npub": settings.npub or "",
|
|
"endpoint_urls": _resolve_endpoint_urls(),
|
|
"window_hours": DASHBOARD_WINDOW_HOURS,
|
|
"interval_minutes": DASHBOARD_INTERVAL_MINUTES,
|
|
"summary": summary_payload,
|
|
"model_usage_mix": usage_mix_payload,
|
|
"top_model_usage": top_model_usage,
|
|
"others_usage": others_usage,
|
|
"windows": windows,
|
|
}
|
|
|
|
|
|
def create_stats_snapshot_event(
|
|
private_key_hex: str,
|
|
provider_id: str,
|
|
payload_json: str,
|
|
*,
|
|
d_tag: str,
|
|
) -> dict[str, Any]:
|
|
tags = [
|
|
["d", d_tag],
|
|
["provider", provider_id],
|
|
["schema", ANALYTICS_SCHEMA],
|
|
]
|
|
|
|
return create_signed_event(
|
|
private_key_hex,
|
|
kind=ANALYTICS_KIND,
|
|
content=payload_json,
|
|
tags=tags,
|
|
)
|
|
|
|
|
|
def _fingerprint_payload(payload: dict[str, Any]) -> str:
|
|
normalized = dict(payload)
|
|
# Ignore generated timestamp for semantic dedupe.
|
|
normalized.pop("generated_at", None)
|
|
payload_json = json.dumps(normalized, separators=(",", ":"), sort_keys=True)
|
|
return hashlib.sha256(payload_json.encode("utf-8")).hexdigest()
|
|
|
|
|
|
async def publish_usage_analytics() -> None:
|
|
last_payload_hash: str | None = None
|
|
|
|
parsed_nsec: str | None = None
|
|
private_key_hex: str | None = None
|
|
public_key_hex: str | None = None
|
|
provider_id: str | None = None
|
|
warned_missing_nsec = False
|
|
|
|
logger.info("Usage analytics sharing task started")
|
|
|
|
while True:
|
|
try:
|
|
if not settings.enable_analytics_sharing:
|
|
await asyncio.sleep(DISABLED_POLL_SECONDS)
|
|
continue
|
|
|
|
nsec = (settings.nsec or "").strip()
|
|
if not nsec:
|
|
if not warned_missing_nsec:
|
|
logger.info("NSEC is not configured; skipping analytics sharing to Nostr")
|
|
warned_missing_nsec = True
|
|
await asyncio.sleep(DISABLED_POLL_SECONDS)
|
|
continue
|
|
|
|
warned_missing_nsec = False
|
|
if nsec != parsed_nsec or private_key_hex is None or public_key_hex is None:
|
|
keypair = nsec_to_keypair(nsec)
|
|
if not keypair:
|
|
logger.error("Invalid NSEC; analytics sharing is paused")
|
|
await asyncio.sleep(DISABLED_POLL_SECONDS)
|
|
continue
|
|
private_key_hex, public_key_hex = keypair
|
|
parsed_nsec = nsec
|
|
provider_id = _resolve_provider_id(public_key_hex)
|
|
last_payload_hash = None
|
|
|
|
if private_key_hex is None or public_key_hex is None:
|
|
await asyncio.sleep(DISABLED_POLL_SECONDS)
|
|
continue
|
|
|
|
relay_urls = _resolve_relays()
|
|
if not relay_urls:
|
|
logger.warning("No Nostr relays configured; analytics sharing skipped")
|
|
await asyncio.sleep(DISABLED_POLL_SECONDS)
|
|
continue
|
|
|
|
resolved_provider_id = provider_id or _resolve_provider_id(public_key_hex)
|
|
now_ts = int(time.time())
|
|
payload = build_stats_snapshot_payload(
|
|
resolved_provider_id,
|
|
public_key_hex=public_key_hex,
|
|
generated_at=now_ts,
|
|
)
|
|
|
|
payload_hash = _fingerprint_payload(payload)
|
|
if last_payload_hash == payload_hash:
|
|
await asyncio.sleep(PUBLISH_INTERVAL_SECONDS)
|
|
continue
|
|
|
|
payload_json = json.dumps(payload, separators=(",", ":"), sort_keys=True)
|
|
d_tag = f"{resolved_provider_id}:stats"
|
|
event = create_stats_snapshot_event(
|
|
private_key_hex,
|
|
resolved_provider_id,
|
|
payload_json,
|
|
d_tag=d_tag,
|
|
)
|
|
|
|
success_count = 0
|
|
for relay_url in relay_urls:
|
|
if await publish_to_relay(relay_url, event):
|
|
success_count += 1
|
|
|
|
if success_count > 0:
|
|
last_payload_hash = payload_hash
|
|
|
|
logger.info(
|
|
"Published analytics snapshot (success=%s/%s provider=%s)",
|
|
success_count,
|
|
len(relay_urls),
|
|
resolved_provider_id,
|
|
extra={
|
|
"relay_success_count": success_count,
|
|
"relay_total": len(relay_urls),
|
|
"provider_id": resolved_provider_id,
|
|
},
|
|
)
|
|
await asyncio.sleep(PUBLISH_INTERVAL_SECONDS)
|
|
|
|
except asyncio.CancelledError:
|
|
logger.info("Usage analytics sharing task cancelled")
|
|
break
|
|
except Exception as e:
|
|
logger.error(
|
|
"Usage analytics sharing error",
|
|
extra={"error": str(e), "error_type": type(e).__name__},
|
|
)
|
|
await asyncio.sleep(DISABLED_POLL_SECONDS)
|