diff --git a/routstr/core/admin.py b/routstr/core/admin.py index c0ccbe93..710626bc 100644 --- a/routstr/core/admin.py +++ b/routstr/core/admin.py @@ -456,18 +456,18 @@ async def batch_override_provider_models( logger.info( f"BATCH_OVERRIDE called: provider_id={provider_id}, count={len(payload.models)}" ) - + async with create_session() as session: provider = await session.get(UpstreamProviderRow, provider_id) if not provider: raise HTTPException(status_code=404, detail="Provider not found") - + overridden_count = 0 - + for model_data in payload.models: # Try to get existing model regardless of whether it's enabled or not existing_row = await session.get(ModelRow, (model_data.id, provider_id)) - + if existing_row: # Update existing existing_row.name = model_data.name @@ -483,7 +483,9 @@ async def batch_override_provider_models( else None ) existing_row.top_provider = ( - json.dumps(model_data.top_provider) if model_data.top_provider else None + json.dumps(model_data.top_provider) + if model_data.top_provider + else None ) existing_row.canonical_slug = model_data.canonical_slug existing_row.alias_ids = ( @@ -508,23 +510,32 @@ async def batch_override_provider_models( else None ), top_provider=( - json.dumps(model_data.top_provider) if model_data.top_provider else None + json.dumps(model_data.top_provider) + if model_data.top_provider + else None ), canonical_slug=model_data.canonical_slug, alias_ids=( - json.dumps(model_data.alias_ids) if model_data.alias_ids else None + json.dumps(model_data.alias_ids) + if model_data.alias_ids + else None ), upstream_provider_id=provider_id, enabled=model_data.enabled, ) session.add(row) - + overridden_count += 1 - + await session.commit() await refresh_model_maps() - return {"ok": True, "count": overridden_count, "message": f"Successfully batch overridden {overridden_count} models"} + return { + "ok": True, + "count": overridden_count, + "message": f"Successfully batch overridden {overridden_count} models", + } + class UpstreamProviderCreate(BaseModel): provider_type: str @@ -533,6 +544,7 @@ class UpstreamProviderCreate(BaseModel): api_version: str | None = None enabled: bool = True provider_fee: float = 1.01 + provider_settings: dict | None = None class UpstreamProviderUpdate(BaseModel): @@ -542,6 +554,7 @@ class UpstreamProviderUpdate(BaseModel): api_version: str | None = None enabled: bool | None = None provider_fee: float | None = None + provider_settings: dict | None = None @admin_router.get("/api/upstream-providers", dependencies=[Depends(require_admin_api)]) @@ -558,6 +571,9 @@ async def get_upstream_providers() -> list[dict[str, object]]: "api_version": p.api_version, "enabled": p.enabled, "provider_fee": p.provider_fee, + "provider_settings": json.loads(p.provider_settings) + if p.provider_settings + else None, } for p in providers ] @@ -585,6 +601,9 @@ async def create_upstream_provider( api_version=payload.api_version, enabled=payload.enabled, provider_fee=payload.provider_fee, + provider_settings=json.dumps(payload.provider_settings) + if payload.provider_settings + else None, ) session.add(provider) await session.commit() @@ -600,6 +619,7 @@ async def create_upstream_provider( "api_version": provider.api_version, "enabled": provider.enabled, "provider_fee": provider.provider_fee, + "provider_settings": payload.provider_settings, } @@ -619,6 +639,9 @@ async def get_upstream_provider(provider_id: int) -> dict[str, object]: "api_version": provider.api_version, "enabled": provider.enabled, "provider_fee": provider.provider_fee, + "provider_settings": json.loads(provider.provider_settings) + if provider.provider_settings + else None, } @@ -645,6 +668,8 @@ async def update_upstream_provider( provider.enabled = payload.enabled if payload.provider_fee is not None: provider.provider_fee = payload.provider_fee + if payload.provider_settings is not None: + provider.provider_settings = json.dumps(payload.provider_settings) session.add(provider) await session.commit() @@ -660,6 +685,9 @@ async def update_upstream_provider( "api_version": provider.api_version, "enabled": provider.enabled, "provider_fee": provider.provider_fee, + "provider_settings": json.loads(provider.provider_settings) + if provider.provider_settings + else None, } diff --git a/routstr/core/db.py b/routstr/core/db.py index 71834d5a..acfe876e 100644 --- a/routstr/core/db.py +++ b/routstr/core/db.py @@ -146,6 +146,9 @@ class UpstreamProviderRow(SQLModel, table=True): # type: ignore provider_fee: float = Field( default=1.01, description="Provider fee multiplier (default 1%)" ) + provider_settings: str | None = Field( + default=None, description="JSON string for provider-specific settings" + ) models: list["ModelRow"] = Relationship( back_populates="upstream_provider", sa_relationship_kwargs={"cascade": "all, delete-orphan"}, diff --git a/routstr/core/log_manager.py b/routstr/core/log_manager.py index 43920605..3498404e 100644 --- a/routstr/core/log_manager.py +++ b/routstr/core/log_manager.py @@ -19,6 +19,7 @@ class LogManager: specific_date: str | None = None, reverse_files: bool = False, max_files: int | None = None, + window_center: datetime | None = None, ) -> Iterator[dict[str, Any]]: """ Yields log entries from files. @@ -28,6 +29,7 @@ class LogManager: specific_date: specific date string (YYYY-MM-DD) to look at. reverse_files: if True, process files in reverse order (newest first). max_files: maximum number of log files to process (most recent if reverse_files is True). + window_center: datetime object to center a 5-month window around. """ if not self.logs_dir.exists(): return @@ -41,6 +43,36 @@ class LogManager: log_files.append(log_file) else: log_files = sorted(self.logs_dir.glob("app_*.log")) + + if window_center: + # Calculate the 5 months: [center-2, center-1, center, center+1, center+2] + allowed_month_years = [] + cur_m = window_center.month + cur_y = window_center.year + + for offset in range(-2, 3): + m = cur_m + offset + y = cur_y + while m <= 0: + m += 12 + y -= 1 + while m > 12: + m -= 12 + y += 1 + allowed_month_years.append(f"{y}-{m:02d}") + + filtered_files = [] + for log_path in log_files: + try: + # Stem is "app_YYYY-MM-DD" + file_date_str = log_path.stem.split("_")[1] + file_month_year = file_date_str[:7] # YYYY-MM + if file_month_year in allowed_month_years: + filtered_files.append(log_path) + except Exception: + continue + log_files = filtered_files + if reverse_files: log_files.reverse() @@ -217,11 +249,19 @@ class LogManager: return True def get_usage_summary(self, hours: int = 24) -> dict: - entries = list(self._yield_log_entries(hours_back=hours)) + entries = list( + self._yield_log_entries( + hours_back=hours, window_center=datetime.now(timezone.utc) + ) + ) return self._calculate_summary_stats(entries) def get_usage_metrics(self, interval: int = 15, hours: int = 24) -> dict: - entries = list(self._yield_log_entries(hours_back=hours)) + entries = list( + self._yield_log_entries( + hours_back=hours, window_center=datetime.now(timezone.utc) + ) + ) return self._aggregate_metrics_by_time(entries, interval, hours) def get_error_details(self, hours: int = 24, limit: int = 100) -> dict: @@ -236,7 +276,9 @@ class LogManager: # Let's just stick to PR 229 logic which filters 'ERROR' level. - entries = self._yield_log_entries(hours_back=hours) # oldest to newest + entries = self._yield_log_entries( + hours_back=hours, window_center=datetime.now(timezone.utc) + ) # oldest to newest for entry in entries: if entry.get("levelname", "").upper() == "ERROR": @@ -257,7 +299,11 @@ class LogManager: return {"errors": errors[:limit], "total_count": len(errors)} def get_revenue_by_model(self, hours: int = 24, limit: int = 20) -> dict: - entries = list(self._yield_log_entries(hours_back=hours)) + entries = list( + self._yield_log_entries( + hours_back=hours, window_center=datetime.now(timezone.utc) + ) + ) model_stats: dict[str, dict[str, int | float]] = defaultdict( lambda: { diff --git a/routstr/upstream/__init__.py b/routstr/upstream/__init__.py index 13c26791..c9156b10 100644 --- a/routstr/upstream/__init__.py +++ b/routstr/upstream/__init__.py @@ -10,6 +10,7 @@ from .openai import OpenAIUpstreamProvider from .openrouter import OpenRouterUpstreamProvider from .perplexity import PerplexityUpstreamProvider from .ppqai import PPQAIUpstreamProvider +from .routstr import RoutstrUpstreamProvider from .xai import XAIUpstreamProvider upstream_provider_classes: list[type[BaseUpstreamProvider]] = [ @@ -24,6 +25,7 @@ upstream_provider_classes: list[type[BaseUpstreamProvider]] = [ OpenRouterUpstreamProvider, PerplexityUpstreamProvider, PPQAIUpstreamProvider, + RoutstrUpstreamProvider, XAIUpstreamProvider, ] """List of all upstream classes""" diff --git a/routstr/upstream/routstr.py b/routstr/upstream/routstr.py new file mode 100644 index 00000000..7fdd99fb --- /dev/null +++ b/routstr/upstream/routstr.py @@ -0,0 +1,149 @@ +from typing import TYPE_CHECKING, Any, Mapping + +import httpx + +from ..core import get_logger +from ..payment.models import Model +from .base import BaseUpstreamProvider + +if TYPE_CHECKING: + from ..core.db import UpstreamProviderRow + +logger = get_logger(__name__) + + +class RoutstrUpstreamProvider(BaseUpstreamProvider): + """Upstream provider for communicating with another Routstr instance.""" + + provider_type = "routstr" + default_base_url = None + platform_url = None + + def __init__( + self, + base_url: str, + api_key: str, + provider_fee: float = 1.01, + provider_settings: dict | None = None, + ): + """Initialize Routstr provider. + + Args: + base_url: Base URL of the upstream Routstr instance + api_key: API key for the upstream Routstr instance + provider_fee: Provider fee multiplier + provider_settings: Provider-specific settings (auto-topup, etc.) + """ + # Ensure base_url doesn't end with /v1 as BaseUpstreamProvider appends it if needed + # but Routstr paths are usually absolute from base. + super().__init__( + base_url=base_url.rstrip("/"), + api_key=api_key, + provider_fee=provider_fee, + ) + self.settings = provider_settings or {} + + @classmethod + def from_db_row( + cls, provider_row: "UpstreamProviderRow" + ) -> "RoutstrUpstreamProvider": + import json + + settings = {} + if provider_row.provider_settings: + try: + settings = json.loads(provider_row.provider_settings) + except Exception: + pass + + return cls( + base_url=provider_row.base_url, + api_key=provider_row.api_key, + provider_fee=provider_row.provider_fee, + provider_settings=settings, + ) + + @classmethod + def get_provider_metadata(cls) -> dict[str, object]: + return { + "id": cls.provider_type, + "name": "Routstr Node", + "default_base_url": "", + "fixed_base_url": False, + "platform_url": cls.platform_url, + "can_create_account": False, + "can_topup": True, + "can_show_balance": True, + } + + async def get_balance(self) -> float | None: + """Fetch balance from the upstream Routstr node. + + Returns: + Balance in satoshis, or None if failed + """ + url = f"{self.base_url}/v1/balance/info" + headers = {"Authorization": f"Bearer {self.api_key}"} + + async with httpx.AsyncClient() as client: + try: + response = await client.get(url, headers=headers, timeout=10.0) + response.raise_for_status() + data = response.json() + # Routstr balance info usually contains 'balance' in msats or sats + # Check for msats and convert to sats + if "balance_msats" in data: + return float(data["balance_msats"]) / 1000.0 + return float(data.get("balance", 0)) + except Exception as e: + logger.error( + "Failed to fetch balance from upstream Routstr", + extra={"url": url, "error": str(e)}, + ) + return None + + async def topup(self, cashu_token: str) -> dict[str, Any]: + """Top up balance on the upstream Routstr node. + + Args: + cashu_token: Cashu token to deposit + + Returns: + Dict containing top-up result + """ + url = f"{self.base_url}/v1/balance/topup" + headers = {"Authorization": f"Bearer {self.api_key}"} + payload = {"cashu_token": cashu_token} + + async with httpx.AsyncClient() as client: + try: + response = await client.post( + url, headers=headers, json=payload, timeout=30.0 + ) + response.raise_for_status() + return response.json() + except Exception as e: + logger.error( + "Failed to topup upstream Routstr", + extra={"url": url, "error": str(e)}, + ) + return {"error": str(e)} + + async def fetch_models(self) -> list[Model]: + """Fetch models from the upstream Routstr node.""" + url = f"{self.base_url}/v1/models" + headers = {"Authorization": f"Bearer {self.api_key}"} + + async with httpx.AsyncClient() as client: + try: + response = await client.get(url, headers=headers, timeout=15.0) + response.raise_for_status() + data = response.json() + models = data.get("data", []) + return [Model(**m) for m in models] + except Exception as e: + logger.error( + "Failed to fetch models from upstream Routstr", + extra={"url": url, "error": str(e)}, + ) + return [] diff --git a/ui/app/page.tsx b/ui/app/page.tsx index ba5dff37..de744045 100644 --- a/ui/app/page.tsx +++ b/ui/app/page.tsx @@ -155,24 +155,30 @@ export default function DashboardPage() { - - Last Hour - Last 6 Hours - Last 24 Hours - Last 3 Days - Last Week - + + Last Hour + Last 6 Hours + Last 24 Hours + Last 3 Days + Last Week + Last Month + Last Year + +