laantungir/routstr-core
1
0
Fork 0
mirror of https://github.com/Routstr/routstr-core.git synced 2026-08-09 02:54:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #93 from Routstr/main

Browse Source 
update prod
This commit is contained in:
shroominic
2025-08-05 14:27:47 -03:00
committed by GitHub
parent 331c323e70 92318f0b00
commit 451fdd08e9
10 changed files with 132 additions and 282 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+6
View File
@@ -4,6 +4,7 @@ DESCRIPTION = "A short Description"
# Any openai-compatible api endpoint
UPSTREAM_BASE_URL="https://api.openai.com/v1"
UPSTREAM_API_KEY="sk-21212121212121212121212121212121"
# UPSTREAM_PROVIDER_FEE=1 # 1 = no fees, 1.05 = 5% fees
# Lightning address used to receive funds
RECEIVE_LN_ADDRESS="shroominic@walletofsatoshi.com"
@@ -20,6 +21,7 @@ MODEL_BASED_PRICING = "true"
# COST_PER_REQUEST="10"
# COST_PER_1K_INPUT_TOKENS = "0"
# COST_PER_1K_OUTPUT_TOKENS = "0"
# EXCHANGE_FEE = "1.005" # 0.5 % currency exchange fee
# password used to log into admin interface
ADMIN_PASSWORD="CHANGE-THIS"
@@ -32,3 +34,7 @@ ONION_URL=".onion"
RELAYS="wss://relay.routstr.com,wss://relay.nostr.band"
CASHU_MINTS="https://mint.minibits.cash/Bitcoin,https://mint.cubabitcoin.org"
# Development
# DEBUG=TRUE
# LOG_LEVEL=TRACE
router/auth.py
+7 -8
View File
@@ -370,7 +370,7 @@ async def revert_pay_for_request(
async def adjust_payment_for_tokens(
key: ApiKey, response_data: dict, session: AsyncSession
key: ApiKey, response_data: dict, session: AsyncSession, deducted_max_cost: int
) -> dict:
"""
Adjusts the payment based on token usage in the response.
@@ -378,20 +378,19 @@ async def adjust_payment_for_tokens(
Returns cost data to be included in the response.
"""
model = response_data.get("model", "unknown")
max_cost = get_max_cost_for_model(model=model)
logger.debug(
"Starting payment adjustment for tokens",
extra={
"key_hash": key.hashed_key[:8] + "...",
"model": model,
"max_cost": max_cost,
"deducted_max_cost": deducted_max_cost,
"current_balance": key.balance,
"has_usage": "usage" in response_data,
},
)
match calculate_cost(response_data, max_cost):
match calculate_cost(response_data, deducted_max_cost):
case MaxCostData() as cost:
logger.debug(
"Using max cost data (no token adjustment)",
@@ -406,7 +405,7 @@ async def adjust_payment_for_tokens(
case CostData() as cost:
# If token-based pricing is enabled and base cost is 0, use token-based cost
# Otherwise, token cost is additional to the base cost
cost_difference = cost.total_msats - max_cost
cost_difference = cost.total_msats - deducted_max_cost
logger.info(
"Calculated token-based cost",
@@ -414,7 +413,7 @@ async def adjust_payment_for_tokens(
"key_hash": key.hashed_key[:8] + "...",
"model": model,
"token_cost": cost.total_msats,
"max_cost": max_cost,
"deducted_max_cost": deducted_max_cost,
"cost_difference": cost_difference,
"input_msats": cost.input_msats,
"output_msats": cost.output_msats,
@@ -468,7 +467,7 @@ async def adjust_payment_for_tokens(
await session.commit()
if result.rowcount:
cost.total_msats = max_cost + cost_difference
cost.total_msats = deducted_max_cost + cost_difference
await session.refresh(key)
logger.info(
@@ -513,7 +512,7 @@ async def adjust_payment_for_tokens(
)
await session.exec(refund_stmt) # type: ignore[call-overload]
await session.commit()
cost.total_msats = max_cost - refund
cost.total_msats = deducted_max_cost - refund
await session.refresh(key)
logger.info(
router/core/logging.py
+20 -1
View File
@@ -10,6 +10,20 @@ from typing import Any
from pythonjsonlogger import jsonlogger
from rich.logging import RichHandler
# Define custom TRACE level
TRACE_LEVEL = 5
logging.addLevelName(TRACE_LEVEL, "TRACE")
def trace(self: logging.Logger, message: str, *args: Any, **kwargs: Any) -> None:
"""Log with TRACE level"""
if self.isEnabledFor(TRACE_LEVEL):
self._log(TRACE_LEVEL, message, args, **kwargs)
# Add the trace method to Logger class
setattr(logging.Logger, "trace", trace)
class DailyRotatingFileHandler(logging.handlers.TimedRotatingFileHandler):
"""Custom TimedRotatingFileHandler that creates date-based filenames."""
@@ -150,7 +164,12 @@ class SecurityFilter(logging.Filter):
def get_log_level() -> str:
"""Get log level from environment variable."""
return os.environ.get("LOG_LEVEL", "INFO").upper()
level = os.environ.get("LOG_LEVEL", "INFO").upper()
# Validate log level - if invalid, default to INFO
valid_levels = {"TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
if level not in valid_levels:
level = "INFO"
return level
def should_enable_console_logging() -> bool:
router/payment/cost_caculation.py
+6 -5
View File
@@ -1,3 +1,4 @@
import math
import os
from pydantic import BaseModel
@@ -145,9 +146,9 @@ def calculate_cost(
input_tokens = response_data.get("usage", {}).get("prompt_tokens", 0)
output_tokens = response_data.get("usage", {}).get("completion_tokens", 0)
input_msats = int(round(input_tokens / 1000 * MSATS_PER_1K_INPUT_TOKENS, 0))
output_msats = int(round(output_tokens / 1000 * MSATS_PER_1K_OUTPUT_TOKENS, 0))
token_based_cost = int(round(input_msats + output_msats, 0))
input_msats = round(input_tokens / 1000 * MSATS_PER_1K_INPUT_TOKENS, 3)
output_msats = round(output_tokens / 1000 * MSATS_PER_1K_OUTPUT_TOKENS, 3)
token_based_cost = math.ceil(input_msats + output_msats)
logger.info(
"Calculated token-based cost",
@@ -163,7 +164,7 @@ def calculate_cost(
return CostData(
base_msats=0,
input_msats=input_msats,
output_msats=output_msats,
input_msats=int(input_msats),
output_msats=int(output_msats),
total_msats=token_based_cost,
)
router/payment/helpers.py
+15 -175
View File
@@ -1,12 +1,10 @@
import base64
import json
import os
from typing import Literal
import cbor2
from fastapi import HTTPException, Response
from ..core import get_logger
from ..wallet import deserialize_token_from_string
from .cost_caculation import COST_PER_REQUEST, MODEL_BASED_PRICING
from .models import MODELS
@@ -49,17 +47,7 @@ def get_cost_per_request(model: str | None = None) -> int:
return COST_PER_REQUEST
def check_token_balance(headers: dict, body: dict) -> Literal["sat", "msat"]:
"""Check if the provided token has sufficient balance."""
logger.debug(
"Checking token balance",
extra={
"has_x_cashu": "x-cashu" in headers,
"has_authorization": "authorization" in headers,
"model": body.get("model", "unknown"),
},
)
def check_token_balance(headers: dict, body: dict, max_cost_for_model: int) -> None:
if x_cashu := headers.get("x-cashu", None):
cashu_token = x_cashu
logger.debug(
@@ -100,173 +88,25 @@ def check_token_balance(headers: dict, body: dict) -> Literal["sat", "msat"]:
# Handle regular API keys (sk-*)
if cashu_token.startswith("sk-"):
logger.debug(
"Regular API key detected", extra={"key_preview": cashu_token[:10] + "..."}
)
return "sat"
return
cost = get_cost_per_request(model=body.get("model", None))
token_obj = deserialize_token_from_string(cashu_token)
if cashu_token.startswith("cashuA"):
logger.debug("Processing CashuA token", extra={"required_cost_msats": cost})
amount_msat = (
token_obj.amount if token_obj.unit == "msat" else token_obj.amount * 1000
)
try:
_token = base64_token_json(cashu_token)
amount = sum(p["amount"] for t in _token["token"] for p in t["proofs"])
unit: Literal["sat", "msat"] = _token.get("unit", "sat")
if unit == "sat":
amount *= 1000
logger.info(
"CashuA token parsed successfully",
extra={
"amount": amount,
"unit": unit,
"amount_msats": amount,
"required_cost_msats": cost,
"sufficient_balance": amount >= cost,
},
)
if amount < cost:
logger.warning(
"Insufficient token balance",
extra={
"amount_msats": amount,
"required_msats": cost,
"shortfall_msats": cost - amount,
"unit": unit,
},
)
raise HTTPException(status_code=413, detail="Insufficient balance")
except Exception as e:
logger.error(
"Failed to parse CashuA token",
extra={
"error": str(e),
"error_type": type(e).__name__,
"token_preview": cashu_token[:20] + "...",
},
)
raise HTTPException(status_code=401, detail="Invalid token format")
elif cashu_token.startswith("cashuB"):
logger.debug("Processing CashuB token", extra={"required_cost_msats": cost})
try:
_token = base64_token_cbor(cashu_token)
amount = sum(p["a"] for t in _token["t"] for p in t["p"])
unit = _token["u"]
if unit == "sat":
amount *= 1000
logger.info(
"CashuB token parsed successfully",
extra={
"amount": amount,
"unit": unit,
"amount_msats": amount,
"required_cost_msats": cost,
"sufficient_balance": amount >= cost,
},
)
if amount < cost:
logger.warning(
"Insufficient token balance",
extra={
"amount_msats": amount,
"required_msats": cost,
"shortfall_msats": cost - amount,
"unit": unit,
},
)
raise HTTPException(status_code=413, detail="Insufficient balance")
except Exception as e:
logger.error(
"Failed to parse CashuB token",
extra={
"error": str(e),
"error_type": type(e).__name__,
"token_preview": cashu_token[:20] + "...",
},
)
raise HTTPException(status_code=401, detail="Invalid token format")
else:
logger.error(
"Unknown token format",
extra={"token_prefix": cashu_token[:10] if cashu_token else "empty"},
)
raise HTTPException(status_code=401, detail="Unauthorized")
return unit
def base64_token_json(cashu_token: str) -> dict:
"""Decode a CashuA (JSON) token."""
logger.debug("Decoding CashuA token", extra={"token_length": len(cashu_token)})
try:
# Version 3 - JSON format
encoded = cashu_token[6:] # Remove "cashuA"
# Add correct padding – (-len) % 4 equals 0,1,2,3
encoded += "=" * ((-len(encoded)) % 4)
decoded = base64.urlsafe_b64decode(encoded).decode()
token_data = json.loads(decoded)
logger.debug(
"CashuA token decoded successfully",
extra={
"token_proofs_count": sum(
len(t.get("proofs", [])) for t in token_data.get("token", [])
),
"unit": token_data.get("unit", "unknown"),
if max_cost_for_model > amount_msat:
raise HTTPException(
status_code=413,
detail={
"reason": "Insufficient balance",
"amount_required_msat": max_cost_for_model,
"model": body.get("model", "unknown"),
"type": "minimum_balance_required",
},
)
return token_data
except Exception as e:
logger.error(
"Failed to decode CashuA token",
extra={"error": str(e), "error_type": type(e).__name__},
)
raise
def base64_token_cbor(cashu_token: str) -> dict:
"""Decode a CashuB (CBOR) token."""
logger.debug("Decoding CashuB token", extra={"token_length": len(cashu_token)})
try:
encoded = cashu_token[6:] # Remove "cashuB"
encoded += "=" * ((-len(encoded)) % 4)
decoded_bytes = base64.urlsafe_b64decode(encoded)
token_data = cbor2.loads(decoded_bytes)
logger.debug(
"CashuB token decoded successfully",
extra={
"token_proofs_count": sum(
len(t.get("p", [])) for t in token_data.get("t", [])
),
"unit": token_data.get("u", "unknown"),
},
)
return token_data
except Exception as e:
logger.error(
"Failed to decode CashuB token",
extra={"error": str(e), "error_type": type(e).__name__},
)
raise
def get_max_cost_for_model(model: str) -> int:
"""Get the maximum cost for a specific model."""
router/payment/models.py
+14 -34
View File
@@ -89,43 +89,23 @@ async def update_sats_pricing() -> None:
model.sats_pricing = Pricing(
**{k: v / sats_to_usd for k, v in model.pricing.dict().items()}
)
mspp = model.sats_pricing.prompt
mspc = model.sats_pricing.completion
if model.top_provider:
if (
model.top_provider.context_length
and model.top_provider.max_completion_tokens
if (cl := model.top_provider.context_length) and (
mct := model.top_provider.max_completion_tokens
):
max_context_cost = (
model.top_provider.context_length
* model.sats_pricing.prompt
)
max_completion_cost = (
model.top_provider.max_completion_tokens
* model.sats_pricing.completion
)
model.sats_pricing.max_cost = (
max_context_cost + max_completion_cost
)
elif model.top_provider.context_length:
max_context_cost = (
model.top_provider.context_length
* model.sats_pricing.prompt
)
max_completion_cost = 32_000 * model.sats_pricing.completion
model.sats_pricing.max_cost = (
max_context_cost + max_completion_cost
)
elif model.top_provider.max_completion_tokens:
max_completion_cost = (
model.top_provider.max_completion_tokens
* model.sats_pricing.completion
)
max_context_cost = 1_048_576 * model.sats_pricing.prompt
model.sats_pricing.max_cost = max_completion_cost
model.sats_pricing.max_cost = (cl - mct) * mspp + mct * mspc
elif cl := model.top_provider.context_length:
model.sats_pricing.max_cost = cl * 0.8 * mspp + cl * 0.2 * mspc
elif mct := model.top_provider.max_completion_tokens:
model.sats_pricing.max_cost = mct * 4 * mspp + mct * mspc
else:
model.sats_pricing.max_cost = (
1_048_576 * model.sats_pricing.prompt
+ 32_000 * model.sats_pricing.completion
)
model.sats_pricing.max_cost = 1_000_000 * mspp + 32_000 * mspc
elif model.context_length:
model.sats_pricing.max_cost = (
model.sats_pricing.prompt * model.context_length * 0.8
) + (model.sats_pricing.completion * model.context_length * 0.2)
else:
p = model.sats_pricing.prompt * 1_000_000
c = model.sats_pricing.completion * 32_000
router/payment/price.py
+4 -3
View File
@@ -9,8 +9,9 @@ logger = get_logger(__name__)
# artifical spread to cover conversion fees
EXCHANGE_FEE = float(os.environ.get("EXCHANGE_FEE", "1.005")) # 0.5% default
logger.info("Price module initialized", extra={"exchange_fee": EXCHANGE_FEE})
UPSTREAM_PROVIDER_FEE = float(
os.environ.get("UPSTREAM_PROVIDER_FEE", "1.05")
) # 5% default (e.g. openrouter charges 5% margin)
async def kraken_btc_usd(client: httpx.AsyncClient) -> float | None:
@@ -98,7 +99,7 @@ async def btc_usd_ask_price() -> float:
raise ValueError("Unable to fetch BTC price from any exchange")
max_price = max(valid_prices)
final_price = max_price * EXCHANGE_FEE
final_price = max_price * EXCHANGE_FEE * UPSTREAM_PROVIDER_FEE
return final_price
router/proxy.py
+25 -21
View File
@@ -19,6 +19,7 @@ from .payment.helpers import (
UPSTREAM_BASE_URL,
check_token_balance,
create_error_response,
get_cost_per_request,
prepare_upstream_headers,
)
from .payment.x_cashu import x_cashu_handler
@@ -28,7 +29,7 @@ proxy_router = APIRouter()
async def handle_streaming_chat_completion(
response: httpx.Response, key: ApiKey, session: AsyncSession
response: httpx.Response, key: ApiKey, max_cost_for_model: int
) -> StreamingResponse:
"""Handle streaming chat completion responses with token-based pricing."""
logger.info(
@@ -40,7 +41,7 @@ async def handle_streaming_chat_completion(
},
)
async def stream_with_cost() -> AsyncGenerator[bytes, None]:
async def stream_with_cost(max_cost_for_model: int) -> AsyncGenerator[bytes, None]:
# Store all chunks to analyze
stored_chunks = []
@@ -103,7 +104,10 @@ async def handle_streaming_chat_completion(
if fresh_key:
try:
cost_data = await adjust_payment_for_tokens(
fresh_key, data, new_session
fresh_key,
data,
new_session,
max_cost_for_model,
)
logger.info(
"Token adjustment completed for streaming",
@@ -140,14 +144,17 @@ async def handle_streaming_chat_completion(
)
return StreamingResponse(
stream_with_cost(),
stream_with_cost(max_cost_for_model),
status_code=response.status_code,
headers=dict(response.headers),
)
async def handle_non_streaming_chat_completion(
response: httpx.Response, key: ApiKey, session: AsyncSession
response: httpx.Response,
key: ApiKey,
session: AsyncSession,
deducted_max_cost: int,
) -> Response:
"""Handle non-streaming chat completion responses with token-based pricing."""
logger.info(
@@ -172,7 +179,9 @@ async def handle_non_streaming_chat_completion(
},
)
cost_data = await adjust_payment_for_tokens(key, response_json, session)
cost_data = await adjust_payment_for_tokens(
key, response_json, session, deducted_max_cost
)
response_json["cost"] = cost_data
logger.info(
@@ -239,6 +248,7 @@ async def forward_to_upstream(
headers: dict,
request_body: bytes | None,
key: ApiKey,
max_cost_for_model: int,
session: AsyncSession,
) -> Response | StreamingResponse:
"""Forward request to upstream and handle the response."""
@@ -338,7 +348,9 @@ async def forward_to_upstream(
if is_streaming and response.status_code == 200:
# Process streaming response and extract cost from the last chunk
result = await handle_streaming_chat_completion(response, key, session)
result = await handle_streaming_chat_completion(
response, key, max_cost_for_model
)
background_tasks = BackgroundTasks()
background_tasks.add_task(response.aclose)
background_tasks.add_task(client.aclose)
@@ -349,7 +361,7 @@ async def forward_to_upstream(
# Handle non-streaming response
try:
return await handle_non_streaming_chat_completion(
response, key, session
response, key, session, max_cost_for_model
)
finally:
await response.aclose()
@@ -479,18 +491,10 @@ async def proxy(
media_type="application/json",
)
# Check token balance for all requests to get currency unit
try:
unit = check_token_balance(headers, request_body_dict)
logger.debug(
"Token balance check completed", extra={"path": path, "unit": unit}
)
except HTTPException as e:
logger.warning(
"Token balance check failed",
extra={"path": path, "status_code": e.status_code, "detail": str(e.detail)},
)
raise
max_cost_for_model = get_cost_per_request(
model=request_body_dict.get("model", None)
)
check_token_balance(headers, request_body_dict, max_cost_for_model)
# Handle authentication
if x_cashu := headers.get("x-cashu", None):
@@ -571,7 +575,7 @@ async def proxy(
# Forward to upstream and handle response
response = await forward_to_upstream(
request, path, headers, request_body, key, session
request, path, headers, request_body, key, max_cost_for_model, session
)
if response.status_code != 200:
router/wallet.py
+18 -22
View File
@@ -1,8 +1,8 @@
import os
from typing import Literal
from cashu.core.base import Token, Unit
from cashu.wallet.helpers import deserialize_token_from_string, receive, send
from cashu.core.base import Token
from cashu.wallet.helpers import deserialize_token_from_string, send
from cashu.wallet.wallet import Wallet
from .core import db, get_logger
@@ -18,7 +18,6 @@ PRIMARY_MINT_URL = TRUSTED_MINTS[0]
async def get_balance(unit: CurrencyUnit) -> int:
wallet = await Wallet.with_db(
PRIMARY_MINT_URL,
# DATABASE_URL,
db=".wallet",
load_all_keysets=True,
unit=unit,
@@ -30,21 +29,22 @@ async def get_balance(unit: CurrencyUnit) -> int:
async def recieve_token(
token: str,
) -> tuple[int, CurrencyUnit, str]: # amount, unit, mint_url
# trusted_mints = os.environ["CASHU_MINTS"].split(",")
token_obj = deserialize_token_from_string(token)
if len(token_obj.keysets) > 1:
raise ValueError("Multiple keysets per token currently not supported")
wallet = await Wallet.with_db(
token_obj.mint,
db=".wallet",
load_all_keysets=True,
unit=token_obj.unit,
)
await wallet.load_mint(token_obj.keysets[0])
if token_obj.mint != PRIMARY_MINT_URL:
raise ValueError(
f"This mint is not supported, please use {PRIMARY_MINT_URL} instead"
)
if token_obj.mint not in TRUSTED_MINTS:
return await swap_to_primary_mint(token_obj, wallet)
await receive(wallet, token_obj)
await wallet.redeem(token_obj.proofs)
return token_obj.amount, token_obj.unit, token_obj.mint
@@ -80,28 +80,24 @@ async def swap_to_primary_mint(
raise ValueError("Invalid unit")
estimated_fee_sat = max(amount_msat // 1000 * 0.01, 2)
amount_msat_after_fee = amount_msat - estimated_fee_sat * 1000
print(f"amount_msat_after_fee: {amount_msat_after_fee}")
primary_wallet = await Wallet.with_db(
PRIMARY_MINT_URL, db=".temp", load_all_keysets=True, unit="sat"
PRIMARY_MINT_URL, db=".wallet", load_all_keysets=True, unit="sat"
)
await primary_wallet.load_mint_keysets()
mint_quote = await primary_wallet.mint_quote(
amount_msat_after_fee // 1000, Unit.sat
)
print(f"mint_quote: {mint_quote}")
await primary_wallet.load_mint()
minted_amount = amount_msat_after_fee // 1000
mint_quote = await primary_wallet.request_mint(minted_amount)
melt_quote = await token_wallet.melt_quote(mint_quote.request)
print(f"melt_quote: {melt_quote}")
melt_quote_resp = await token_wallet.melt(
_ = await token_wallet.melt(
proofs=token_obj.proofs,
invoice=mint_quote.request,
fee_reserve_sat=melt_quote.fee_reserve,
quote_id=melt_quote.quote,
)
print(f"melt_quote_resp: {melt_quote_resp}")
_ = await primary_wallet.mint(minted_amount, quote_id=mint_quote.quote)
_ = await primary_wallet.mint(amount_msat_after_fee // 1000, mint_quote.quote)
return amount_msat_after_fee // 1000, "sat", PRIMARY_MINT_URL
return minted_amount, "sat", PRIMARY_MINT_URL
async def credit_balance(
tests/test_models.py
+17 -13
View File
@@ -87,11 +87,18 @@ async def test_update_sats_pricing_calculation(sample_model: Model) -> None:
0.001 / 0.0001
) # 10 sats
# Verify max_cost calculation for model with top_provider
expected_max_context = 4096 * sample_model.sats_pricing.prompt
expected_max_completion = 2048 * sample_model.sats_pricing.completion
assert sample_model.top_provider is not None
assert sample_model.top_provider.context_length is not None
assert sample_model.top_provider.max_completion_tokens is not None
assert sample_model.sats_pricing.max_cost == pytest.approx(
expected_max_context + expected_max_completion
(
sample_model.top_provider.context_length
- sample_model.top_provider.max_completion_tokens
)
* sample_model.sats_pricing.prompt
+ sample_model.top_provider.max_completion_tokens
* sample_model.sats_pricing.completion
)
# Cancel and await the task
@@ -159,16 +166,13 @@ async def test_update_sats_pricing_without_top_provider() -> None:
assert model_without_top.sats_pricing is not None
# Verify the fallback max_cost calculation
p = model_without_top.sats_pricing.prompt * 1_000_000
c = model_without_top.sats_pricing.completion * 32_000
r = model_without_top.sats_pricing.request * 100_000
i = model_without_top.sats_pricing.image * 100
w = model_without_top.sats_pricing.web_search * 1000
ir = model_without_top.sats_pricing.internal_reasoning * 100
expected_max = p + c + r + i + w + ir
assert model_without_top.sats_pricing.max_cost == pytest.approx(
expected_max
model_without_top.context_length
* 0.8
* model_without_top.sats_pricing.prompt
+ model_without_top.context_length
* 0.2
* model_without_top.sats_pricing.completion
)
# Cancel and await the task