Skip to content

A
aisbf
Commit c5907326 authored by Stefy Lanza (nextime / spora )'s avatar Stefy Lanza (nextime / spora )
Browse files
Options

feat: add marketplace publishing and settlement

parent 67149a8a
Hide whitespace changes
Inline Side-by-side
Showing with 2356 additions and 22 deletions
aisbf/database.py
View file @ c5907326
......@@ -25,12 +25,14 @@ import sqlite3
import json
import hashlib
import time
import copy
from pathlib import Path
from typing import Dict, List, Optional, Tuple, Any
from datetime import datetime, timedelta
import logging
import asyncio
from concurrent.futures import ThreadPoolExecutor
from decimal import Decimal, ROUND_HALF_UP
try:
import bcrypt as _bcrypt_lib
......@@ -171,6 +173,78 @@ class DatabaseManager:
def placeholder(self) -> str:
return '?' if self.db_type == 'sqlite' else '%s'
@staticmethod
def _quantize_money(value: Any) -> Decimal:
return Decimal(str(value or 0)).quantize(Decimal('0.01'), rounding=ROUND_HALF_UP)
@staticmethod
def _market_cache_key(consumer_user_id: int, listing_id: int, metadata: Optional[Dict[str, Any]]) -> Optional[str]:
if not isinstance(metadata, dict):
return None
request_id = metadata.get('market_request_id') or metadata.get('request_id')
if not request_id:
return None
return f"market:settlement:{consumer_user_id}:{listing_id}:{request_id}"
@staticmethod
def _extract_market_settlement_key(metadata: Optional[Dict[str, Any]]) -> Optional[str]:
if not isinstance(metadata, dict):
return None
key = metadata.get('market_settlement_key')
return str(key).strip() if key not in (None, '') else None
@staticmethod
def _sanitize_market_config(config: Dict[str, Any]) -> Dict[str, Any]:
secret_keys = {
'api_key', 'password', 'secret', 'token', 'access_token', 'refresh_token',
'client_secret', 'authorization', 'credentials', 'session_token'
}
def _sanitize(value):
if isinstance(value, dict):
sanitized = {}
for key, item in value.items():
lowered = str(key).lower()
if lowered in secret_keys or lowered.endswith('_token') or lowered.endswith('_secret'):
continue
if lowered in {'auth_files', 'credentials_file', 'cookie_file', 'oauth_file'}:
continue
sanitized[key] = _sanitize(item)
return sanitized
if isinstance(value, list):
return [_sanitize(item) for item in value]
return value
return _sanitize(copy.deepcopy(config or {}))
def _load_market_listing_row(self, row) -> Dict[str, Any]:
metadata = json.loads(row[14]) if row[14] else {}
config_snapshot = json.loads(row[15]) if row[15] else {}
return {
'id': row[0],
'owner_user_id': row[1],
'owner_username': row[2],
'source_scope': row[3],
'source_type': row[4],
'source_id': row[5],
'listing_key': row[6],
'title': row[7],
'description': row[8],
'provider_id': row[9],
'model_id': row[10],
'endpoint': row[11],
'currency_code': row[12],
'price_per_million_tokens': float(row[13] or 0),
'price_per_1000_requests': float(row[16] or 0),
'provider_price_per_million_tokens': float(row[17]) if row[17] is not None else None,
'provider_price_per_1000_requests': float(row[18]) if row[18] is not None else None,
'metadata': metadata,
'config_snapshot': config_snapshot,
'is_active': bool(row[19]),
'created_at': row[20],
'updated_at': row[21],
}
async def _run_in_executor(self, func, *args):
"""Run a blocking database operation in a thread pool executor."""
loop = asyncio.get_event_loop()
......@@ -2971,7 +3045,7 @@ class DatabaseManager:
SELECT id, name, description, price_monthly, price_yearly, is_default, is_active,
max_requests_per_day, max_requests_per_month, max_providers, max_rotations,
max_autoselections, max_rotation_models, max_autoselection_models,
created_at, updated_at, is_visible
created_at, updated_at, is_visible, market_fee_percentage
FROM account_tiers
ORDER BY price_monthly ASC
''')
......@@ -2995,7 +3069,8 @@ class DatabaseManager:
'max_autoselection_models': row[13],
'created_at': row[14],
'updated_at': row[15],
'is_visible': bool(row[16]) if len(row) > 16 else True
'is_visible': bool(row[16]) if len(row) > 16 else True,
'market_fee_percentage': float(row[17] or 10.0) if len(row) > 17 else 10.0,
})
return tiers
......@@ -3016,7 +3091,7 @@ class DatabaseManager:
SELECT id, name, description, price_monthly, price_yearly, is_default, is_active,
max_requests_per_day, max_requests_per_month, max_providers, max_rotations,
max_autoselections, max_rotation_models, max_autoselection_models,
created_at, updated_at, is_visible
created_at, updated_at, is_visible, market_fee_percentage
FROM account_tiers
WHERE id = {placeholder}
''', (tier_id,))
......@@ -3040,7 +3115,8 @@ class DatabaseManager:
'max_autoselection_models': row[13],
'created_at': row[14],
'updated_at': row[15],
'is_visible': bool(row[16]) if len(row) > 16 else True
'is_visible': bool(row[16]) if len(row) > 16 else True,
'market_fee_percentage': float(row[17] or 10.0) if len(row) > 17 else 10.0,
}
return None
......@@ -3048,7 +3124,8 @@ class DatabaseManager:
max_requests_per_day: int = -1, max_requests_per_month: int = -1,
max_providers: int = -1, max_rotations: int = -1,
max_autoselections: int = -1, max_rotation_models: int = -1,
max_autoselection_models: int = -1, is_active: bool = True, is_visible: bool = True) -> int:
max_autoselection_models: int = -1, market_fee_percentage: float = 10.0,
is_active: bool = True, is_visible: bool = True) -> int:
"""
Create a new account tier.
......@@ -3076,14 +3153,14 @@ class DatabaseManager:
cursor.execute(f'''
INSERT INTO account_tiers
(name, description, price_monthly, price_yearly, is_active, is_visible,
max_requests_per_day, max_requests_per_month, max_providers, max_rotations,
max_autoselections, max_rotation_models, max_autoselection_models)
max_requests_per_day, max_requests_per_month, max_providers, max_rotations,
max_autoselections, max_rotation_models, max_autoselection_models, market_fee_percentage)
VALUES ({placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder},
{placeholder}, {placeholder}, {placeholder}, {placeholder},
{placeholder}, {placeholder}, {placeholder})
{placeholder}, {placeholder}, {placeholder}, {placeholder})
''', (name, description, price_monthly, price_yearly, 1 if is_active else 0, 1 if is_visible else 0,
max_requests_per_day, max_requests_per_month, max_providers, max_rotations,
max_autoselections, max_rotation_models, max_autoselection_models))
max_autoselections, max_rotation_models, max_autoselection_models, market_fee_percentage))
conn.commit()
return cursor.lastrowid
......@@ -3108,7 +3185,7 @@ class DatabaseManager:
allowed_fields = ['name', 'description', 'price_monthly', 'price_yearly', 'is_active', 'is_visible',
'max_requests_per_day', 'max_requests_per_month', 'max_providers',
'max_rotations', 'max_autoselections', 'max_rotation_models',
'max_autoselection_models']
'max_autoselection_models', 'market_fee_percentage']
for field in allowed_fields:
if field in kwargs:
......@@ -3145,6 +3222,566 @@ class DatabaseManager:
''', (tier_id,))
conn.commit()
return cursor.rowcount > 0
def get_market_fee_percentage_for_user(self, user_id: int) -> float:
tier = self.get_user_tier(user_id)
if not tier:
tiers = self.get_all_tiers()
default_tier = next((t for t in tiers if t.get('is_default')), None)
if default_tier:
return float(default_tier.get('market_fee_percentage', 10.0) or 10.0)
return 10.0
return float(tier.get('market_fee_percentage', 10.0) or 10.0)
def upsert_market_listing(self, owner_user_id: int, owner_username: str, payload: Dict[str, Any]) -> int:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = self.placeholder
metadata_json = json.dumps(payload.get('metadata') or {})
snapshot_json = json.dumps(self._sanitize_market_config(payload.get('config_snapshot') or {}))
listing_key = payload['listing_key']
existing = None
cursor.execute(
f'''SELECT id FROM market_listings WHERE owner_user_id = {placeholder} AND listing_key = {placeholder}''',
(owner_user_id, listing_key)
)
existing = cursor.fetchone()
values = (
owner_user_id,
owner_username,
payload.get('source_scope', 'user'),
payload['source_type'],
payload['source_id'],
listing_key,
payload['title'],
payload.get('description'),
payload.get('provider_id'),
payload.get('model_id'),
payload.get('endpoint'),
payload.get('currency_code', 'USD'),
float(payload.get('price_per_million_tokens', 0) or 0),
metadata_json,
snapshot_json,
float(payload.get('price_per_1000_requests', 0) or 0),
payload.get('provider_price_per_million_tokens'),
payload.get('provider_price_per_1000_requests'),
1 if payload.get('is_active', True) else 0,
)
if existing:
cursor.execute(
f'''
UPDATE market_listings
SET owner_username = {placeholder},
source_scope = {placeholder},
source_type = {placeholder},
source_id = {placeholder},
title = {placeholder},
description = {placeholder},
provider_id = {placeholder},
model_id = {placeholder},
endpoint = {placeholder},
currency_code = {placeholder},
price_per_million_tokens = {placeholder},
metadata = {placeholder},
config_snapshot = {placeholder},
price_per_1000_requests = {placeholder},
provider_price_per_million_tokens = {placeholder},
provider_price_per_1000_requests = {placeholder},
is_active = {placeholder},
updated_at = CURRENT_TIMESTAMP
WHERE id = {placeholder}
''',
values[1:] + (existing[0],)
)
conn.commit()
return existing[0]
cursor.execute(
f'''
INSERT INTO market_listings (
owner_user_id, owner_username, source_scope, source_type, source_id, listing_key,
title, description, provider_id, model_id, endpoint, currency_code,
price_per_million_tokens, metadata, config_snapshot, price_per_1000_requests,
provider_price_per_million_tokens, provider_price_per_1000_requests, is_active,
created_at, updated_at
) VALUES (
{placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder},
{placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder},
{placeholder}, {placeholder}, {placeholder}, {placeholder},
{placeholder}, {placeholder}, {placeholder}, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
)
''',
values
)
conn.commit()
return cursor.lastrowid
def list_market_listings(self, active_only: bool = True) -> List[Dict[str, Any]]:
with self._get_connection() as conn:
cursor = conn.cursor()
query = '''
SELECT id, owner_user_id, owner_username, source_scope, source_type, source_id, listing_key,
title, description, provider_id, model_id, endpoint, currency_code,
price_per_million_tokens, metadata, config_snapshot, price_per_1000_requests,
provider_price_per_million_tokens, provider_price_per_1000_requests, is_active,
created_at, updated_at
FROM market_listings
'''
if active_only:
query += ' WHERE is_active = 1'
query += ' ORDER BY created_at DESC'
cursor.execute(query)
return [self._load_market_listing_row(row) for row in cursor.fetchall()]
def get_market_listing(self, listing_id: int) -> Optional[Dict[str, Any]]:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = self.placeholder
cursor.execute(
f'''
SELECT id, owner_user_id, owner_username, source_scope, source_type, source_id, listing_key,
title, description, provider_id, model_id, endpoint, currency_code,
price_per_million_tokens, metadata, config_snapshot, price_per_1000_requests,
provider_price_per_million_tokens, provider_price_per_1000_requests, is_active,
created_at, updated_at
FROM market_listings
WHERE id = {placeholder}
''',
(listing_id,)
)
row = cursor.fetchone()
return self._load_market_listing_row(row) if row else None
def set_market_listing_active(self, listing_id: int, owner_user_id: int, is_active: bool) -> bool:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = self.placeholder
cursor.execute(
f'''UPDATE market_listings SET is_active = {placeholder}, updated_at = CURRENT_TIMESTAMP WHERE id = {placeholder} AND owner_user_id = {placeholder}''',
(1 if is_active else 0, listing_id, owner_user_id)
)
conn.commit()
return cursor.rowcount > 0
def admin_set_market_listing_active(self, listing_id: int, is_active: bool) -> bool:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = self.placeholder
cursor.execute(
f'''UPDATE market_listings SET is_active = {placeholder}, updated_at = CURRENT_TIMESTAMP WHERE id = {placeholder}''',
(1 if is_active else 0, listing_id)
)
conn.commit()
return cursor.rowcount > 0
def upsert_market_vote(self, listing_id: int, voter_user_id: int, target_type: str, target_key: str, vote: int) -> bool:
vote = 1 if int(vote) > 0 else -1
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = self.placeholder
cursor.execute(
f'''SELECT id FROM market_votes WHERE listing_id = {placeholder} AND voter_user_id = {placeholder} AND target_type = {placeholder} AND target_key = {placeholder}''',
(listing_id, voter_user_id, target_type, target_key)
)
row = cursor.fetchone()
if row:
cursor.execute(
f'''UPDATE market_votes SET vote = {placeholder}, updated_at = CURRENT_TIMESTAMP WHERE id = {placeholder}''',
(vote, row[0])
)
else:
cursor.execute(
f'''
INSERT INTO market_votes (listing_id, voter_user_id, target_type, target_key, vote, created_at, updated_at)
VALUES ({placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
''',
(listing_id, voter_user_id, target_type, target_key, vote)
)
conn.commit()
return True
def get_market_vote_summary(self, listing_id: int) -> Dict[str, Dict[str, int]]:
summary = {
'listing': {'upvotes': 0, 'downvotes': 0, 'score': 0},
'provider': {'upvotes': 0, 'downvotes': 0, 'score': 0},
'model': {'upvotes': 0, 'downvotes': 0, 'score': 0},
'user': {'upvotes': 0, 'downvotes': 0, 'score': 0},
}
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = self.placeholder
cursor.execute(
f'''
SELECT target_type,
SUM(CASE WHEN vote > 0 THEN 1 ELSE 0 END),
SUM(CASE WHEN vote < 0 THEN 1 ELSE 0 END),
COALESCE(SUM(vote), 0)
FROM market_votes
WHERE listing_id = {placeholder}
GROUP BY target_type
''',
(listing_id,)
)
for row in cursor.fetchall():
target_type = row[0]
if target_type in summary:
summary[target_type] = {
'upvotes': int(row[1] or 0),
'downvotes': int(row[2] or 0),
'score': int(row[3] or 0),
}
return summary
def list_market_imports(self, user_id: int) -> List[Dict[str, Any]]:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = self.placeholder
cursor.execute(
f'''
SELECT i.id, i.listing_id, i.imported_config_type, i.imported_config_id, i.created_at,
l.title, l.owner_username
FROM market_imports i
JOIN market_listings l ON l.id = i.listing_id
WHERE i.user_id = {placeholder}
ORDER BY i.created_at DESC
''',
(user_id,)
)
return [
{
'id': row[0],
'listing_id': row[1],
'imported_config_type': row[2],
'imported_config_id': row[3],
'created_at': row[4],
'title': row[5],
'owner_username': row[6],
}
for row in cursor.fetchall()
]
def record_market_import(self, user_id: int, listing_id: int, imported_config_type: str, imported_config_id: str) -> int:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = self.placeholder
cursor.execute(
f'''
INSERT INTO market_imports (user_id, listing_id, imported_config_type, imported_config_id, created_at)
VALUES ({placeholder}, {placeholder}, {placeholder}, {placeholder}, CURRENT_TIMESTAMP)
''',
(user_id, listing_id, imported_config_type, imported_config_id)
)
conn.commit()
return cursor.lastrowid
def get_market_listing_for_share(self, owner_username: str, resource_type: str, resource_id: str) -> Optional[Dict[str, Any]]:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = self.placeholder
cursor.execute(
f'''
SELECT id, owner_user_id, owner_username, source_scope, source_type, source_id, listing_key,
title, description, provider_id, model_id, endpoint, currency_code,
price_per_million_tokens, metadata, config_snapshot, price_per_1000_requests,
provider_price_per_million_tokens, provider_price_per_1000_requests, is_active,
created_at, updated_at
FROM market_listings
WHERE owner_username = {placeholder} AND source_type = {placeholder} AND source_id = {placeholder} AND is_active = 1
''',
(owner_username, resource_type, resource_id)
)
row = cursor.fetchone()
return self._load_market_listing_row(row) if row else None
def get_market_listing_stats(self, listing_id: int) -> Dict[str, Any]:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = self.placeholder
cursor.execute(
f'''
SELECT COUNT(*),
COALESCE(SUM(total_tokens), 0),
COALESCE(SUM(requests_count), 0),
COALESCE(SUM(gross_amount), 0),
COALESCE(SUM(platform_fee), 0),
COALESCE(SUM(provider_amount), 0),
COALESCE(AVG(total_tokens), 0),
COUNT(settlement_key)
FROM market_usage_transactions
WHERE listing_id = {placeholder}
''',
(listing_id,)
)
row = cursor.fetchone() or (0, 0, 0, 0, 0, 0, 0, 0)
return {
'usage_events': int(row[0] or 0),
'total_tokens': int(row[1] or 0),
'total_requests': int(row[2] or 0),
'gross_revenue': float(row[3] or 0),
'platform_fees': float(row[4] or 0),
'provider_revenue': float(row[5] or 0),
'avg_tokens_per_request': float(row[6] or 0),
'settled_requests': int(row[7] or 0),
}
def _get_market_usage_transaction_by_settlement_key(self, settlement_key: Optional[str]) -> Optional[Dict[str, Any]]:
if not settlement_key:
return None
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = self.placeholder
cursor.execute(
f'''
SELECT gross_amount, platform_fee, provider_amount, currency_code, metadata
FROM market_usage_transactions
WHERE settlement_key = {placeholder}
''',
(settlement_key,)
)
row = cursor.fetchone()
if not row:
return None
metadata = {}
try:
metadata = json.loads(row[4]) if row[4] else {}
except Exception:
metadata = {}
return {
'gross_amount': Decimal(str(row[0] or 0)),
'platform_fee': Decimal(str(row[1] or 0)),
'provider_amount': Decimal(str(row[2] or 0)),
'currency_code': row[3],
'metadata': metadata,
}
def settle_market_usage(
self,
consumer_user_id: int,
listing_id: int,
prompt_tokens: int = 0,
completion_tokens: int = 0,
requests_count: int = 1,
metadata: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
metadata = metadata or {}
listing = self.get_market_listing(listing_id)
if not listing or not listing.get('is_active'):
raise ValueError('Market listing not available')
settlement_cache_key = self._market_cache_key(consumer_user_id, listing_id, metadata)
if settlement_cache_key:
metadata = {**metadata, 'market_settlement_key': settlement_cache_key}
if listing['owner_user_id'] == consumer_user_id:
return {
'charged_amount': Decimal('0.00'),
'seller_amount': Decimal('0.00'),
'platform_fee': Decimal('0.00'),
'platform_revenue': Decimal('0.00'),
'balance_after': None,
'listing': listing,
'self_use': True,
'charged': False,
}
total_tokens = max(int(prompt_tokens or 0) + int(completion_tokens or 0), 0)
request_units = Decimal(str(max(int(requests_count or 0), 0))) / Decimal('1000')
token_units = Decimal(str(total_tokens)) / Decimal('1000000')
token_price = Decimal(str(listing.get('price_per_million_tokens') or 0))
request_price = Decimal(str(listing.get('price_per_1000_requests') or 0))
gross_amount = self._quantize_money((token_units * token_price) + (request_units * request_price))
if gross_amount <= Decimal('0.00'):
return {
'charged_amount': Decimal('0.00'),
'seller_amount': Decimal('0.00'),
'platform_fee': Decimal('0.00'),
'platform_revenue': Decimal('0.00'),
'balance_after': None,
'listing': listing,
'self_use': False,
'charged': False,
}
owner_user_id = int(listing.get('owner_user_id') or 0)
is_platform_owned_listing = owner_user_id <= 0
fee_percentage = Decimal(str(self.get_market_fee_percentage_for_user(owner_user_id))) if owner_user_id > 0 else Decimal('0')
platform_fee = self._quantize_money((gross_amount * fee_percentage) / Decimal('100')) if owner_user_id > 0 else Decimal('0.00')
seller_amount = self._quantize_money(gross_amount - platform_fee) if owner_user_id > 0 else Decimal('0.00')
platform_revenue = self._quantize_money(gross_amount if is_platform_owned_listing else platform_fee)
def _build_deduplicated_result(existing_txn: Optional[Dict[str, Any]]) -> Dict[str, Any]:
balance_after = self.get_wallet_summary(consumer_user_id)
gross = Decimal(str((existing_txn or {}).get('gross_amount', 0) or 0))
fee = Decimal(str((existing_txn or {}).get('platform_fee', 0) or 0))
seller = Decimal(str((existing_txn or {}).get('provider_amount', 0) or 0))
return {
'charged_amount': gross,
'seller_amount': seller,
'platform_fee': fee,
'platform_revenue': gross if is_platform_owned_listing else fee,
'balance_after': self._quantize_money((balance_after or {}).get('balance', 0)) if balance_after else None,
'listing': listing,
'self_use': False,
'charged': False,
'deduplicated': True,
}
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = self.placeholder
if settlement_cache_key:
existing_txn = self._get_market_usage_transaction_by_settlement_key(settlement_cache_key)
if existing_txn:
return _build_deduplicated_result(existing_txn)
if self.db_type == 'mysql':
cursor.execute(f'''SELECT id, balance FROM user_wallets WHERE user_id = {placeholder} FOR UPDATE''', (consumer_user_id,))
else:
cursor.execute(f'''SELECT id, balance FROM user_wallets WHERE user_id = {placeholder}''', (consumer_user_id,))
consumer_wallet = cursor.fetchone()
if not consumer_wallet:
cursor.execute(
f'''INSERT INTO user_wallets (user_id, balance, currency_code, created_at, updated_at) VALUES ({placeholder}, 0.00, {placeholder}, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)''',
(consumer_user_id, listing.get('currency_code', 'USD'))
)
consumer_wallet_id = cursor.lastrowid
consumer_balance = Decimal('0.00')
else:
consumer_wallet_id = consumer_wallet[0]
consumer_balance = Decimal(str(consumer_wallet[1] or 0))
if consumer_balance < gross_amount:
raise ValueError('Insufficient wallet balance')
consumer_currency = None if not consumer_wallet else (self.get_wallet_summary(consumer_user_id) or {}).get('currency_code')
if consumer_currency and consumer_currency != listing.get('currency_code', 'USD'):
raise ValueError('Consumer wallet currency does not match listing currency')
owner_wallet_id = None
owner_balance = Decimal('0.00')
if owner_user_id > 0:
if self.db_type == 'mysql':
cursor.execute(f'''SELECT id, balance FROM user_wallets WHERE user_id = {placeholder} FOR UPDATE''', (owner_user_id,))
else:
cursor.execute(f'''SELECT id, balance FROM user_wallets WHERE user_id = {placeholder}''', (owner_user_id,))
owner_wallet = cursor.fetchone()
if not owner_wallet:
cursor.execute(
f'''INSERT INTO user_wallets (user_id, balance, currency_code, created_at, updated_at) VALUES ({placeholder}, 0.00, {placeholder}, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)''',
(owner_user_id, listing.get('currency_code', 'USD'))
)
owner_wallet_id = cursor.lastrowid
else:
owner_wallet_id = owner_wallet[0]
owner_balance = Decimal(str(owner_wallet[1] or 0))
new_consumer_balance = self._quantize_money(consumer_balance - gross_amount)
new_owner_balance = self._quantize_money(owner_balance + seller_amount)
cursor.execute(
f'''UPDATE user_wallets SET balance = {placeholder}, updated_at = CURRENT_TIMESTAMP WHERE id = {placeholder}''',
(float(new_consumer_balance), consumer_wallet_id)
)
if owner_wallet_id is not None:
cursor.execute(
f'''UPDATE user_wallets SET balance = {placeholder}, updated_at = CURRENT_TIMESTAMP WHERE id = {placeholder}''',
(float(new_owner_balance), owner_wallet_id)
)
listing_meta = {
'listing_id': listing_id,
'provider_id': listing.get('provider_id'),
'model_id': listing.get('model_id'),
'market': True,
**metadata,
}
cursor.execute(
f'''
INSERT INTO wallet_transactions
(user_id, wallet_id, amount, type, status, description, metadata, created_at)
VALUES ({placeholder}, {placeholder}, {placeholder}, 'debit', 'completed', {placeholder}, {placeholder}, CURRENT_TIMESTAMP)
''',
(
consumer_user_id,
consumer_wallet_id,
float(gross_amount),
f"Market usage: {listing['title']}",
json.dumps(listing_meta),
)
)
if owner_wallet_id is not None:
cursor.execute(
f'''
INSERT INTO wallet_transactions
(user_id, wallet_id, amount, type, status, description, metadata, created_at)
VALUES ({placeholder}, {placeholder}, {placeholder}, 'credit', 'completed', {placeholder}, {placeholder}, CURRENT_TIMESTAMP)
''',
(
owner_user_id,
owner_wallet_id,
float(seller_amount),
f"Market sale: {listing['title']}",
json.dumps({**listing_meta, 'platform_fee': float(platform_fee), 'platform_revenue': float(platform_revenue)}),
)
)
try:
cursor.execute(
f'''
INSERT INTO market_usage_transactions
(listing_id, consumer_user_id, provider_user_id, prompt_tokens, completion_tokens, total_tokens,
requests_count, gross_amount, platform_fee, provider_amount, currency_code, settlement_key, metadata, created_at)
VALUES ({placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder},
{placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, CURRENT_TIMESTAMP)
''',
(
listing_id,
consumer_user_id,
owner_user_id,
int(prompt_tokens or 0),
int(completion_tokens or 0),
total_tokens,
int(requests_count or 0),
float(gross_amount),
float(platform_fee),
float(seller_amount),
listing.get('currency_code', 'USD'),
settlement_cache_key,
json.dumps({**listing_meta, 'platform_fee': float(platform_fee), 'platform_revenue': float(platform_revenue), 'platform_owned_listing': is_platform_owned_listing}),
)
)
except Exception as exc:
if settlement_cache_key:
duplicate_error = 'unique' in str(exc).lower() or 'duplicate' in str(exc).lower() or 'constraint' in str(exc).lower()
if duplicate_error:
conn.rollback()
existing_txn = self._get_market_usage_transaction_by_settlement_key(settlement_cache_key)
if existing_txn:
return _build_deduplicated_result(existing_txn)
raise
conn.commit()
return {
'charged_amount': gross_amount,
'seller_amount': seller_amount if owner_wallet_id is not None else Decimal('0.00'),
'platform_fee': platform_fee,
'platform_revenue': platform_revenue,
'balance_after': new_consumer_balance,
'listing': listing,
'self_use': False,
'charged': True,
}
def get_wallet_summary(self, user_id: int) -> Optional[Dict[str, Any]]:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = self.placeholder
cursor.execute(f'''SELECT id, balance, currency_code FROM user_wallets WHERE user_id = {placeholder}''', (user_id,))
row = cursor.fetchone()
if not row:
return None
return {'id': row[0], 'balance': float(row[1] or 0), 'currency_code': row[2]}
def get_user_tier(self, user_id: int) -> Optional[Dict]:
"""
......@@ -3163,7 +3800,7 @@ class DatabaseManager:
SELECT t.id, t.name, t.description, t.price_monthly, t.price_yearly,
t.max_requests_per_day, t.max_requests_per_month, t.max_providers,
t.max_rotations, t.max_autoselections, t.max_rotation_models,
t.max_autoselection_models, t.is_default, t.is_active
t.max_autoselection_models, t.market_fee_percentage, t.is_default, t.is_active
FROM users u
JOIN account_tiers t ON u.tier_id = t.id
WHERE u.id = {placeholder}
......@@ -3184,8 +3821,9 @@ class DatabaseManager:
'max_autoselections': row[9],
'max_rotation_models': row[10],
'max_autoselection_models': row[11],
'is_default': bool(row[12]),
'is_active': bool(row[13]),
'market_fee_percentage': float(row[12] or 10.0),
'is_default': bool(row[13]),
'is_active': bool(row[14]),
}
return None
......@@ -3224,7 +3862,7 @@ class DatabaseManager:
SELECT id, name, description, price_monthly, price_yearly, is_default, is_active,
max_requests_per_day, max_requests_per_month, max_providers, max_rotations,
max_autoselections, max_rotation_models, max_autoselection_models,
created_at, updated_at, is_visible
created_at, updated_at, is_visible, market_fee_percentage
FROM account_tiers
WHERE is_visible = 1 AND is_active = 1
ORDER BY price_monthly ASC
......@@ -3249,7 +3887,8 @@ class DatabaseManager:
'max_autoselection_models': row[13],
'created_at': row[14],
'updated_at': row[15],
'is_visible': bool(row[16]) if len(row) > 16 else True
'is_visible': bool(row[16]) if len(row) > 16 else True,
'market_fee_percentage': float(row[17] or 10.0) if len(row) > 17 else 10.0,
})
return tiers
......@@ -3721,6 +4360,48 @@ class DatabaseManager:
return default_currency
def get_market_settings(self) -> Dict:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db_type == 'sqlite' else '%s'
default_settings = {
'enabled': False,
'allow_user_publish': True,
'allow_admin_publish': True,
'allow_import': True,
}
try:
cursor.execute(f'''
SELECT setting_value
FROM admin_settings
WHERE setting_key = {placeholder}
''', ('market',))
row = cursor.fetchone()
if row and row[0]:
loaded = json.loads(row[0])
if isinstance(loaded, dict):
default_settings.update(loaded)
except Exception as e:
logger.warning(f"Error loading market settings: {e}")
return default_settings
def save_market_settings(self, settings: Dict) -> bool:
with self._get_connection() as conn:
cursor = conn.cursor()
placeholder = '?' if self.db_type == 'sqlite' else '%s'
settings_json = json.dumps(settings or {})
try:
insert_syntax = 'INSERT OR REPLACE' if self.db_type == 'sqlite' else 'REPLACE'
cursor.execute(f'''
{insert_syntax} INTO admin_settings (setting_key, setting_value, updated_at)
VALUES ({placeholder}, {placeholder}, CURRENT_TIMESTAMP)
''', ('market', settings_json))
conn.commit()
return True
except Exception as e:
logger.error(f"Error saving market settings: {e}")
return False
def save_currency_settings(self, settings: Dict) -> bool:
"""Save currency settings to admin_settings table."""
with self._get_connection() as conn:
......@@ -4328,6 +5009,7 @@ def DatabaseManager__run_config_migrations(self, cursor, auto_increment, timesta
('max_autoselections', 'INTEGER DEFAULT -1'),
('max_rotation_models', 'INTEGER DEFAULT -1'),
('max_autoselection_models', 'INTEGER DEFAULT -1'),
('market_fee_percentage', 'DECIMAL(5,2) DEFAULT 10.0'),
('is_default', f'{boolean_type} DEFAULT 0'),
('is_active', f'{boolean_type} DEFAULT 1'),
('is_visible', f'{boolean_type} DEFAULT 1')
......@@ -4367,10 +5049,10 @@ def DatabaseManager__run_config_migrations(self, cursor, auto_increment, timesta
INSERT INTO account_tiers
(name, description, price_monthly, price_yearly, is_default, is_active,
max_requests_per_day, max_requests_per_month, max_providers, max_rotations,
max_autoselections, max_rotation_models, max_autoselection_models)
max_autoselections, max_rotation_models, max_autoselection_models, market_fee_percentage)
VALUES
('Free Tier', 'Default free account tier with unlimited access', 0.00, 0.00, 1, 1,
-1, -1, -1, -1, -1, -1, -1)
-1, -1, -1, -1, -1, -1, -1, 10.0)
''')
logger.info("✅ Migration: Inserted default free tier")
except Exception as e:
......@@ -4496,6 +5178,84 @@ def DatabaseManager__run_config_migrations(self, cursor, auto_increment, timesta
FOREIGN KEY (user_id) REFERENCES users(id),
UNIQUE(user_id, provider_id, model_name)
)
'''),
('market_listings', f'''
CREATE TABLE market_listings (
id INTEGER PRIMARY KEY {auto_increment},
owner_user_id INTEGER NOT NULL,
owner_username VARCHAR(255) NOT NULL,
source_scope VARCHAR(32) NOT NULL,
source_type VARCHAR(32) NOT NULL,
source_id VARCHAR(255) NOT NULL,
listing_key VARCHAR(255) NOT NULL,
title VARCHAR(255) NOT NULL,
description TEXT,
provider_id VARCHAR(255),
model_id VARCHAR(255),
endpoint TEXT,
currency_code VARCHAR(10) NOT NULL DEFAULT 'USD',
price_per_million_tokens DECIMAL(10,4) NOT NULL DEFAULT 0.0,
metadata TEXT,
config_snapshot TEXT,
price_per_1000_requests DECIMAL(10,4) NOT NULL DEFAULT 0.0,
provider_price_per_million_tokens DECIMAL(10,4),
provider_price_per_1000_requests DECIMAL(10,4),
is_active {boolean_type} DEFAULT 1,
created_at TIMESTAMP DEFAULT {timestamp_default},
updated_at TIMESTAMP DEFAULT {timestamp_default},
FOREIGN KEY (owner_user_id) REFERENCES users(id),
UNIQUE(owner_user_id, listing_key)
)
'''),
('market_votes', f'''
CREATE TABLE market_votes (
id INTEGER PRIMARY KEY {auto_increment},
listing_id INTEGER NOT NULL,
voter_user_id INTEGER NOT NULL,
target_type VARCHAR(32) NOT NULL,
target_key VARCHAR(255) NOT NULL,
vote INTEGER NOT NULL,
created_at TIMESTAMP DEFAULT {timestamp_default},
updated_at TIMESTAMP DEFAULT {timestamp_default},
FOREIGN KEY (listing_id) REFERENCES market_listings(id),
FOREIGN KEY (voter_user_id) REFERENCES users(id),
UNIQUE(listing_id, voter_user_id, target_type, target_key)
)
'''),
('market_imports', f'''
CREATE TABLE market_imports (
id INTEGER PRIMARY KEY {auto_increment},
user_id INTEGER NOT NULL,
listing_id INTEGER NOT NULL,
imported_config_type VARCHAR(32) NOT NULL,
imported_config_id VARCHAR(255) NOT NULL,
created_at TIMESTAMP DEFAULT {timestamp_default},
FOREIGN KEY (user_id) REFERENCES users(id),
FOREIGN KEY (listing_id) REFERENCES market_listings(id)
)
'''),
('market_usage_transactions', f'''
CREATE TABLE market_usage_transactions (
id INTEGER PRIMARY KEY {auto_increment},
listing_id INTEGER NOT NULL,
consumer_user_id INTEGER NOT NULL,
provider_user_id INTEGER NOT NULL,
prompt_tokens INTEGER DEFAULT 0,
completion_tokens INTEGER DEFAULT 0,
total_tokens INTEGER DEFAULT 0,
requests_count INTEGER DEFAULT 0,
gross_amount DECIMAL(10,2) NOT NULL DEFAULT 0.00,
platform_fee DECIMAL(10,2) NOT NULL DEFAULT 0.00,
provider_amount DECIMAL(10,2) NOT NULL DEFAULT 0.00,
currency_code VARCHAR(10) NOT NULL DEFAULT 'USD',
settlement_key VARCHAR(255),
metadata TEXT,
created_at TIMESTAMP DEFAULT {timestamp_default},
FOREIGN KEY (listing_id) REFERENCES market_listings(id),
FOREIGN KEY (consumer_user_id) REFERENCES users(id),
FOREIGN KEY (provider_user_id) REFERENCES users(id),
UNIQUE(settlement_key)
)
''')
]:
try:
......@@ -4699,6 +5459,33 @@ def DatabaseManager__run_config_migrations(self, cursor, auto_increment, timesta
except Exception as e:
logger.warning(f"Migration check for user_notifications table: {e}")
# Migration: add settlement_key to market_usage_transactions and index it for deduplication
try:
if self.db_type == 'sqlite':
cursor.execute("PRAGMA table_info(market_usage_transactions)")
columns = [row[1] for row in cursor.fetchall()]
if columns and 'settlement_key' not in columns:
cursor.execute("ALTER TABLE market_usage_transactions ADD COLUMN settlement_key VARCHAR(255)")
logger.info("✅ Migration: Added settlement_key column to market_usage_transactions")
cursor.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_market_usage_transactions_settlement_key ON market_usage_transactions(settlement_key)")
else:
cursor.execute("""
SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'market_usage_transactions' AND COLUMN_NAME = 'settlement_key'
""")
if not cursor.fetchone():
cursor.execute("ALTER TABLE market_usage_transactions ADD COLUMN settlement_key VARCHAR(255) NULL")
logger.info("✅ Migration: Added settlement_key column to market_usage_transactions")
cursor.execute("""
SELECT INDEX_NAME FROM INFORMATION_SCHEMA.STATISTICS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'market_usage_transactions' AND INDEX_NAME = 'idx_market_usage_transactions_settlement_key'
""")
if not cursor.fetchone():
cursor.execute("CREATE UNIQUE INDEX idx_market_usage_transactions_settlement_key ON market_usage_transactions(settlement_key)")
logger.info("✅ Migration: Added settlement_key unique index to market_usage_transactions")
except Exception as e:
logger.warning(f"Migration check for market_usage_transactions.settlement_key: {e}")
# Migration: Create context_dimensions table if missing
try:
if self.db_type == 'sqlite':
......
aisbf/handlers.py
View file @ c5907326
......@@ -24,6 +24,7 @@ Request handlers for AISBF.
"""
import asyncio
import base64
import json
import re
import uuid
import hashlib
......@@ -58,6 +59,7 @@ from .streaming_optimization import (
OptimizedTextAccumulator,
optimize_sse_chunk
)
from .database import DatabaseRegistry
_autoselect_result_cache: dict = {}
_autoselect_result_cache_ttl: int = 3600 # seconds
......@@ -183,6 +185,83 @@ class RequestHandler:
payload.pop('_studio_provider_endpoint', None)
return payload
def _get_market_source_details(self, provider_id: str):
provider_config = None
if self.user_id and provider_id in getattr(self, 'user_providers', {}):
provider_config = self.user_providers.get(provider_id)
elif provider_id in getattr(self.config, 'providers', {}):
cfg = self.config.get_provider(provider_id)
provider_config = cfg.model_dump() if hasattr(cfg, 'model_dump') else cfg
if isinstance(provider_config, dict):
return provider_config.get('market_source')
return None
@staticmethod
def _market_request_id(request_data: Optional[Dict], fallback_provider_id: str) -> str:
payload = {
'provider_id': fallback_provider_id,
'model': (request_data or {}).get('model'),
'messages': (request_data or {}).get('messages'),
'input': (request_data or {}).get('input'),
'prompt': (request_data or {}).get('prompt'),
'voice': (request_data or {}).get('voice'),
'endpoint_path': (request_data or {}).get('_market_endpoint_path'),
}
serialized = json.dumps(payload, sort_keys=True, default=str)
return hashlib.sha256(serialized.encode('utf-8')).hexdigest()
def _settle_market_result(self, provider_id: str, usage: Optional[Dict], requests_count: int = 1, metadata: Optional[Dict] = None, request_data: Optional[Dict] = None):
market_source = self._get_market_source_details(provider_id)
if not self.user_id or not market_source:
return None
listing_id = market_source.get('listing_id')
if not listing_id:
return None
db = DatabaseRegistry.get_config_database()
settlement_metadata = {
'provider_id': provider_id,
'market_imported': True,
'market_request_id': self._market_request_id(request_data, provider_id),
}
if metadata:
settlement_metadata.update(metadata)
return db.settle_market_usage(
consumer_user_id=self.user_id,
listing_id=int(listing_id),
prompt_tokens=(usage or {}).get('prompt_tokens', 0),
completion_tokens=(usage or {}).get('completion_tokens', 0),
requests_count=requests_count,
metadata=settlement_metadata,
)
def _extract_usage_from_sse_chunk(self, chunk_payload) -> Optional[Dict[str, int]]:
try:
if isinstance(chunk_payload, bytes):
chunk_payload = chunk_payload.decode('utf-8', errors='ignore')
if isinstance(chunk_payload, str):
for line in chunk_payload.splitlines():
line = line.strip()
if not line.startswith('data: '):
continue
raw = line[6:].strip()
if not raw or raw == '[DONE]':
continue
try:
parsed = json.loads(raw)
except Exception:
continue
if isinstance(parsed, dict) and isinstance(parsed.get('usage'), dict):
usage = parsed.get('usage') or {}
if usage.get('total_tokens') is not None or usage.get('prompt_tokens') is not None or usage.get('completion_tokens') is not None:
return usage
elif isinstance(chunk_payload, dict):
usage = chunk_payload.get('usage')
if isinstance(usage, dict):
return usage
except Exception:
return None
return None
def _load_user_configs(self):
"""Load user-specific configurations from database"""
self.reload_user_config()
......@@ -639,6 +718,12 @@ class RequestHandler:
logger.warning(f"Response cache set failed: {cache_error}")
handler.record_success()
if isinstance(response, dict):
try:
self._settle_market_result(provider_id, response.get('usage', {}), 1, {'kind': 'chat_completion'}, request_data=request_data)
except Exception as market_error:
logger.warning(f"Market settlement failed for {provider_id}: {market_error}")
# Record analytics for token usage
try:
......@@ -1847,6 +1932,11 @@ class RequestHandler:
content = resp.json()
except Exception:
content = {"detail": resp.text}
try:
usage = content.get('usage', {}) if isinstance(content, dict) else {}
self._settle_market_result(provider_id, usage, 1, {'kind': 'generic_proxy', 'endpoint_path': endpoint_path}, request_data=request_data)
except Exception as market_error:
logger.warning(f"Market settlement failed for generic proxy {provider_id}: {market_error}")
return JSONResponse(status_code=resp.status_code, content=content)
except Exception as e:
logger.error(f"Generic proxy error: {e}", exc_info=True)
......@@ -1876,6 +1966,11 @@ class RequestHandler:
await handler.apply_rate_limit()
result = await handler.handle_audio_transcription(form_data)
handler.record_success()
try:
usage = result.get('usage', {}) if isinstance(result, dict) else {}
self._settle_market_result(provider_id, usage, 1, {'kind': 'audio_transcription'}, request_data=request_data)
except Exception as market_error:
logger.warning(f"Market settlement failed for audio transcription {provider_id}: {market_error}")
return result
except Exception as e:
handler.record_failure()
......@@ -1905,6 +2000,11 @@ class RequestHandler:
await handler.apply_rate_limit()
result = await handler.handle_text_to_speech(request_data)
handler.record_success()
try:
usage = result.get('usage', {}) if isinstance(result, dict) else {}
self._settle_market_result(provider_id, usage, 1, {'kind': 'text_to_speech'}, request_data=request_data)
except Exception as market_error:
logger.warning(f"Market settlement failed for text to speech {provider_id}: {market_error}")
return result
except Exception as e:
handler.record_failure()
......@@ -1938,6 +2038,11 @@ class RequestHandler:
result = self._rewrite_content_urls(result, request)
handler.record_success()
try:
usage = result.get('usage', {}) if isinstance(result, dict) else {}
self._settle_market_result(provider_id, usage, 1, {'kind': 'image_generation'}, request_data=request_data)
except Exception as market_error:
logger.warning(f"Market settlement failed for image generation {provider_id}: {market_error}")
return result
except Exception as e:
handler.record_failure()
......@@ -1967,6 +2072,11 @@ class RequestHandler:
await handler.apply_rate_limit()
result = await handler.handle_embeddings(request_data)
handler.record_success()
try:
usage = result.get('usage', {}) if isinstance(result, dict) else {}
self._settle_market_result(provider_id, usage, 1, {'kind': 'embeddings'}, request_data=request_data)
except Exception as market_error:
logger.warning(f"Market settlement failed for embeddings {provider_id}: {market_error}")
return result
except Exception as e:
handler.record_failure()
......@@ -3070,6 +3180,11 @@ class RotationHandler:
logger.warning(f"Response cache set failed: {cache_error}")
logger.info("Returning non-streaming response")
try:
self._settle_market_result(provider_id, response.get('usage', {}) if isinstance(response, dict) else {}, 1, {'kind': 'rotation', 'rotation_id': rotation_id}, request_data=request_data)
except Exception as market_error:
logger.warning(f"Market settlement failed for rotation provider {provider_id}: {market_error}")
# Record analytics for token usage
try:
......@@ -3445,6 +3560,7 @@ class RotationHandler:
async def stream_generator(effective_context):
import json
accumulated_response_text = ""
final_usage = None
try:
if is_google_provider:
# Handle Google's streaming response
......@@ -3735,6 +3851,9 @@ class RotationHandler:
if data_str and data_str != '[DONE]':
try:
chunk_data = json.loads(data_str)
usage_data = chunk_data.get('usage') if isinstance(chunk_data, dict) else None
if isinstance(usage_data, dict):
final_usage = usage_data
choices = chunk_data.get('choices', [])
if choices:
delta = choices[0].get('delta', {})
......@@ -3778,10 +3897,13 @@ class RotationHandler:
logger.debug(f"Yielding raw bytes chunk: {len(chunk)} bytes")
yield chunk
elif isinstance(chunk, str):
final_usage = self._extract_usage_from_sse_chunk(chunk) or final_usage
yield chunk.encode('utf-8')
else:
# Fallback: treat as dict and serialize
chunk_dict = chunk.model_dump() if hasattr(chunk, 'model_dump') else chunk
if isinstance(chunk_dict, dict) and isinstance(chunk_dict.get('usage'), dict):
final_usage = chunk_dict.get('usage')
yield f"data: {json.dumps(chunk_dict)}\n\n".encode('utf-8')
except Exception as chunk_error:
error_msg = str(chunk_error)
......@@ -3838,6 +3960,7 @@ class RotationHandler:
chunk_dict['usage']['prompt_tokens'] = effective_context
chunk_dict['usage']['completion_tokens'] = completion_tokens
chunk_dict['usage']['total_tokens'] = total_tokens
final_usage = chunk_dict.get('usage')
yield f"data: {json.dumps(chunk_dict)}\n\n".encode('utf-8')
except Exception as chunk_error:
......@@ -3852,9 +3975,15 @@ class RotationHandler:
# Record analytics after stream completes
try:
analytics = get_analytics()
prompt_tokens = effective_context
completion_tokens_count = count_messages_tokens([{"role": "assistant", "content": accumulated_response_text}], model_name) if accumulated_response_text else 0
total_tokens = prompt_tokens + completion_tokens_count
prompt_tokens = (final_usage or {}).get('prompt_tokens')
completion_tokens_count = (final_usage or {}).get('completion_tokens')
total_tokens = (final_usage or {}).get('total_tokens')
if prompt_tokens is None:
prompt_tokens = effective_context
if completion_tokens_count is None:
completion_tokens_count = count_messages_tokens([{"role": "assistant", "content": accumulated_response_text}], model_name) if accumulated_response_text else 0
if total_tokens is None:
total_tokens = prompt_tokens + completion_tokens_count
latency_ms = (time.time() - request_start_time) * 1000 if request_start_time else 0
analytics.record_request(
provider_id=provider_id,
......@@ -3870,6 +3999,14 @@ class RotationHandler:
actual_cost=None,
analytics_kind='execution'
)
try:
self._settle_market_result(provider_id, {
'prompt_tokens': prompt_tokens,
'completion_tokens': completion_tokens_count,
'total_tokens': total_tokens,
}, 1, {'kind': 'streaming_chat_completion', 'rotation_id': rotation_id}, request_data=request_data)
except Exception as market_error:
logger.warning(f"Market settlement failed for streaming {provider_id}: {market_error}")
except Exception as analytics_error:
logger.warning(f"Analytics recording for streaming rotation failed: {analytics_error}")
......@@ -4992,6 +5129,15 @@ class AutoselectHandler:
logger.warning(f"Analytics recording for autoselect failed: {analytics_error}")
logger.info(f"=== AUTOSELECT REQUEST END ===")
if isinstance(response, dict):
selected_provider = None
if isinstance(selected_model_id, str) and '/' in selected_model_id:
selected_provider = selected_model_id.split('/', 1)[0]
try:
if selected_provider:
self._settle_market_result(selected_provider, response.get('usage', {}), 1, {'kind': 'autoselect', 'autoselect_id': autoselect_id}, request_data=request_data)
except Exception as market_error:
logger.warning(f"Market settlement failed for autoselect {autoselect_id}: {market_error}")
return response
async def handle_autoselect_streaming_request(self, autoselect_id: str, request_data: Dict):
......
aisbf/models.py
View file @ c5907326
......@@ -119,6 +119,7 @@ class AccountTier(BaseModel):
max_autoselections: int = -1
max_rotation_models: int = -1
max_autoselection_models: int = -1
market_fee_percentage: float = 10.0
created_at: Optional[int] = None
updated_at: Optional[int] = None
......@@ -165,3 +166,50 @@ class PaymentTransaction(BaseModel):
metadata: Optional[Dict] = None
created_at: Optional[int] = None
completed_at: Optional[int] = None
class MarketListing(BaseModel):
id: Optional[int] = None
owner_user_id: int
owner_username: str
source_scope: str
source_type: str
source_id: str
listing_key: str
title: str
description: Optional[str] = None
provider_id: Optional[str] = None
model_id: Optional[str] = None
endpoint: Optional[str] = None
currency_code: str = "USD"
price_per_million_tokens: float = 0.0
price_per_1000_requests: float = 0.0
provider_price_per_million_tokens: Optional[float] = None
provider_price_per_1000_requests: Optional[float] = None
metadata: Optional[Dict] = None
config_snapshot: Optional[Dict] = None
is_active: bool = True
created_at: Optional[int] = None
updated_at: Optional[int] = None
class MarketVote(BaseModel):
id: Optional[int] = None
listing_id: int
voter_user_id: int
target_type: str
target_key: str
vote: int
created_at: Optional[int] = None
updated_at: Optional[int] = None
class MarketUsageSettlementResult(BaseModel):
charged_amount: float = 0.0
seller_amount: float = 0.0
platform_fee: float = 0.0
platform_revenue: float = 0.0
balance_after: Optional[float] = None
listing_id: int
self_use: bool = False
charged: bool = False
aisbf/routes/api.py
View file @ c5907326
......@@ -23,6 +23,31 @@ def init(config, get_user_handler_fn, rotation_handler):
logger = logging.getLogger(__name__)
def _ensure_market_enabled():
settings = DatabaseRegistry.get_config_database().get_market_settings()
if not settings.get('enabled'):
raise HTTPException(status_code=403, detail='Market is disabled')
return settings
def _resolve_market_resource(model: str):
parts = (model or '').split('/')
if len(parts) < 3:
return None
username = parts[0].strip()
resource_type = parts[1].strip()
resource_id = parts[2].strip()
remainder = '/'.join(parts[3:]) if len(parts) > 3 else ''
if not username or resource_type not in {'rotation', 'rotations', 'autoselect', 'autoselections'} and not resource_id:
return None
return {
'username': username,
'resource_type': resource_type,
'resource_id': resource_id,
'model_name': remainder,
}
def parse_provider_from_model(model: str) -> tuple[str, str]:
if '/' in model:
parts = model.split('/', 1)
......@@ -425,6 +450,33 @@ async def embeddings(request: Request, body: dict):
def _resolve_provider(model: str, user_id=None, handler=None) -> tuple[str, str]:
"""Resolve provider_id and actual_model. Checks user providers when handler is given."""
market_resource = _resolve_market_resource(model)
if market_resource:
_ensure_market_enabled()
db = DatabaseRegistry.get_config_database()
owner = db.get_user_by_username(market_resource['username'])
if not owner:
raise HTTPException(status_code=404, detail=f"User '{market_resource['username']}' not found")
owner_user_id = owner['id']
if market_resource['resource_type'] in ('rotation', 'rotations'):
owner_rotation_handler = _get_user_handler('rotation', owner_user_id)
if market_resource['resource_id'] not in getattr(owner_rotation_handler, 'rotations', {}):
raise HTTPException(status_code=404, detail=f"Rotation '{market_resource['resource_id']}' not found")
provider_id, actual_model = owner_rotation_handler._select_provider_and_model(market_resource['resource_id'])
return provider_id, actual_model
if market_resource['resource_type'] in ('autoselect', 'autoselections'):
owner_autoselect_handler = _get_user_handler('autoselect', owner_user_id)
autoselect_cfg = getattr(owner_autoselect_handler, 'user_autoselects', {}).get(market_resource['resource_id'])
if not autoselect_cfg:
raise HTTPException(status_code=404, detail=f"Autoselect '{market_resource['resource_id']}' not found")
fallback = autoselect_cfg.get('fallback', '') if isinstance(autoselect_cfg, dict) else getattr(autoselect_cfg, 'fallback', '')
if '/' in fallback:
return tuple(fallback.split('/', 1))
if fallback:
owner_rotation_handler = _get_user_handler('rotation', owner_user_id)
provider_id, actual_model = owner_rotation_handler._select_provider_and_model(fallback)
return provider_id, actual_model
provider_id, actual_model = parse_provider_from_model(model)
if not provider_id:
raise HTTPException(status_code=400, detail="Model must be in format 'provider/model'")
......
aisbf/routes/dashboard/admin.py
View file @ c5907326
......@@ -457,6 +457,7 @@ async def api_create_tier(request: Request):
max_autoselections=body.get('max_autoselections', -1),
max_rotation_models=body.get('max_rotation_models', -1),
max_autoselection_models=body.get('max_autoselection_models', -1),
market_fee_percentage=body.get('market_fee_percentage', 10.0),
is_active=body.get('is_active', True),
is_visible=body.get('is_visible', True)
)
......@@ -502,6 +503,8 @@ async def api_update_tier(request: Request, tier_id: int):
update_kwargs['max_rotation_models'] = body['max_rotation_models']
if 'max_autoselection_models' in body:
update_kwargs['max_autoselection_models'] = body['max_autoselection_models']
if 'market_fee_percentage' in body:
update_kwargs['market_fee_percentage'] = body['market_fee_percentage']
if 'is_active' in body:
update_kwargs['is_active'] = body['is_active']
if 'is_visible' in body:
......@@ -603,6 +606,7 @@ async def dashboard_admin_tier_save(request: Request):
'max_autoselections': int(form.get('max_autoselections', -1)),
'max_rotation_models': int(form.get('max_rotation_models', -1)),
'max_autoselection_models': int(form.get('max_autoselection_models', -1)),
'market_fee_percentage': float(form.get('market_fee_percentage', 10.0)),
'is_active': form.get('is_active') == '1',
'is_visible': form.get('is_visible') == '1'
}
......@@ -1504,10 +1508,38 @@ async def dashboard_admin_payment_settings(request: Request):
context={
"request": request,
"session": request.session,
"currency_symbol": DatabaseRegistry.get_config_database().get_currency_settings().get('currency_symbol', '$')
"currency_symbol": DatabaseRegistry.get_config_database().get_currency_settings().get('currency_symbol', '$'),
"market_settings": DatabaseRegistry.get_config_database().get_market_settings(),
}
)
@router.get("/api/admin/settings/market")
async def api_get_market_settings(request: Request):
auth_check = require_api_admin(request)
if auth_check:
return auth_check
db = DatabaseRegistry.get_config_database()
return JSONResponse(db.get_market_settings())
@router.post("/api/admin/settings/market")
async def api_save_market_settings(request: Request):
auth_check = require_api_admin(request)
if auth_check:
return auth_check
db = DatabaseRegistry.get_config_database()
try:
body = await request.json()
db.save_market_settings({
'enabled': bool(body.get('enabled', False)),
'allow_user_publish': bool(body.get('allow_user_publish', True)),
'allow_admin_publish': bool(body.get('allow_admin_publish', True)),
'allow_import': bool(body.get('allow_import', True)),
})
return JSONResponse({'success': True, 'message': 'Market settings saved'})
except Exception as e:
logger.error(f"Error saving market settings: {e}")
return JSONResponse({'error': str(e)}, status_code=500)
@router.get("/dashboard/pricing")
async def dashboard_pricing(request: Request):
"""Pricing plans page for users"""
......
aisbf/routes/dashboard/market.py 0 → 100644
View file @ c5907326
from fastapi import APIRouter, Request, HTTPException
from fastapi.responses import HTMLResponse, JSONResponse
from aisbf.routes.auth import require_dashboard_auth, require_api_auth, require_admin
from aisbf.database import DatabaseRegistry
from aisbf.analytics import get_analytics
import json
import logging
router = APIRouter()
_templates = None
_config = None
logger = logging.getLogger(__name__)
def init(config, templates):
global _config, _templates
_config = config
_templates = templates
def _ensure_market_enabled(request: Request, require_admin_override: bool = False):
db = DatabaseRegistry.get_config_database()
settings = db.get_market_settings()
if settings.get('enabled'):
return None, settings
if require_admin_override and request.session.get('role') == 'admin':
return None, settings
return JSONResponse({'error': 'Market is disabled'}, status_code=403), settings
def _safe_provider_config(config_obj):
if hasattr(config_obj, 'model_dump'):
data = config_obj.model_dump()
elif isinstance(config_obj, dict):
data = dict(config_obj)
else:
data = vars(config_obj)
return DatabaseRegistry.get_config_database()._sanitize_market_config(data)
def _safe_config(config_obj):
return _safe_provider_config(config_obj)
def _listing_capabilities(listing: dict) -> list[str]:
metadata = listing.get('metadata') or {}
values = []
for key in ('capabilities', 'partial_capabilities'):
raw = metadata.get(key)
if isinstance(raw, list):
values.extend(str(item).strip().lower() for item in raw if str(item).strip())
model_meta = metadata.get('model') or {}
if isinstance(model_meta, dict):
for key in ('capabilities', 'partial_capabilities'):
raw = model_meta.get(key)
if isinstance(raw, list):
values.extend(str(item).strip().lower() for item in raw if str(item).strip())
return sorted(set(values))
def _listing_size_value(listing: dict):
metadata = listing.get('metadata') or {}
model_meta = metadata.get('model') or {}
for candidate in (
model_meta.get('size'),
model_meta.get('parameter_size'),
model_meta.get('model_size'),
metadata.get('size'),
metadata.get('parameter_size'),
metadata.get('model_size'),
):
if candidate not in (None, ''):
return str(candidate).strip().lower()
return ''
def _listing_model_type(listing: dict):
metadata = listing.get('metadata') or {}
model_meta = metadata.get('model') or {}
for candidate in (
model_meta.get('type'),
model_meta.get('family'),
metadata.get('provider_type'),
metadata.get('type'),
):
if candidate not in (None, ''):
return str(candidate).strip().lower()
return ''
def _build_model_listing_payload(owner_user_id: int, owner_username: str, provider_id: str, provider_config: dict, model_config: dict, source_scope: str, form: dict):
safe_provider = _safe_config(provider_config)
safe_model = _safe_config(model_config)
model_name = safe_model.get('name')
provider_price_tokens = float(form.get('provider_price_per_million_tokens') or form.get('price_per_million_tokens') or 0)
provider_price_requests = float(form.get('provider_price_per_1000_requests') or form.get('price_per_1000_requests') or 0)
model_price_tokens = float(form.get('price_per_million_tokens') or provider_price_tokens)
model_price_requests = float(form.get('price_per_1000_requests') or provider_price_requests)
return {
'owner_user_id': owner_user_id,
'owner_username': owner_username,
'source_scope': source_scope,
'source_type': 'model',
'source_id': f'{provider_id}/{model_name}',
'listing_key': f'model:{provider_id}/{model_name}',
'title': form.get('title') or f'{provider_id}/{model_name}',
'description': form.get('description') or safe_model.get('description') or f'Market model share for {provider_id}/{model_name}',
'provider_id': provider_id,
'model_id': model_name,
'endpoint': safe_provider.get('endpoint'),
'currency_code': form.get('currency_code', 'USD'),
'price_per_million_tokens': model_price_tokens,
'price_per_1000_requests': model_price_requests,
'provider_price_per_million_tokens': provider_price_tokens,
'provider_price_per_1000_requests': provider_price_requests,
'metadata': {
'provider_type': safe_provider.get('type'),
'model': safe_model,
'source_scope': source_scope,
},
'config_snapshot': {
'provider': safe_provider,
'model': safe_model,
},
'is_active': True,
}
def _build_rotation_listing_payload(owner_user_id: int, owner_username: str, rotation_id: str, rotation_config: dict, source_scope: str, form: dict):
safe_rotation = _safe_config(rotation_config)
return {
'owner_user_id': owner_user_id,
'owner_username': owner_username,
'source_scope': source_scope,
'source_type': 'rotation',
'source_id': rotation_id,
'listing_key': f'rotation:{rotation_id}',
'title': form.get('title') or rotation_id,
'description': form.get('description') or f'Market rotation share for {rotation_id}',
'provider_id': None,
'model_id': rotation_id,
'endpoint': None,
'currency_code': form.get('currency_code', 'USD'),
'price_per_million_tokens': float(form.get('price_per_million_tokens') or 0),
'price_per_1000_requests': float(form.get('price_per_1000_requests') or 0),
'provider_price_per_million_tokens': None,
'provider_price_per_1000_requests': None,
'metadata': {
'providers': safe_rotation.get('providers', []),
'model_name': safe_rotation.get('model_name', rotation_id),
'source_scope': source_scope,
},
'config_snapshot': safe_rotation,
'is_active': True,
}
def _build_autoselect_listing_payload(owner_user_id: int, owner_username: str, autoselect_id: str, autoselect_config: dict, source_scope: str, form: dict):
safe_autoselect = _safe_config(autoselect_config)
return {
'owner_user_id': owner_user_id,
'owner_username': owner_username,
'source_scope': source_scope,
'source_type': 'autoselect',
'source_id': autoselect_id,
'listing_key': f'autoselect:{autoselect_id}',
'title': form.get('title') or autoselect_id,
'description': form.get('description') or safe_autoselect.get('description') or f'Market autoselect share for {autoselect_id}',
'provider_id': None,
'model_id': autoselect_id,
'endpoint': None,
'currency_code': form.get('currency_code', 'USD'),
'price_per_million_tokens': float(form.get('price_per_million_tokens') or 0),
'price_per_1000_requests': float(form.get('price_per_1000_requests') or 0),
'provider_price_per_million_tokens': None,
'provider_price_per_1000_requests': None,
'metadata': {
'fallback': safe_autoselect.get('fallback'),
'available_models': safe_autoselect.get('available_models', []),
'source_scope': source_scope,
},
'config_snapshot': safe_autoselect,
'is_active': True,
}
def _attach_analytics_snapshot(listing: dict):
analytics = get_analytics()
source_type = listing.get('source_type')
source_id = listing.get('source_id')
snapshot = {'request_count': 0, 'avg_latency_ms': 0.0, 'error_rate': 0.0, 'total_tokens': 0}
try:
if source_type == 'provider' and listing.get('provider_id'):
stats = analytics.get_provider_stats(listing.get('provider_id'))
snapshot = {
'request_count': (stats.get('requests') or {}).get('total', 0),
'avg_latency_ms': stats.get('avg_latency_ms', 0.0),
'error_rate': stats.get('error_rate', 0.0),
'total_tokens': (stats.get('tokens') or {}).get('total', 0),
}
elif source_type == 'rotation':
rows = analytics.get_rotation_provider_breakdown()
entries = next((row.get('entries', []) for row in rows if row.get('rotation_id') == source_id), [])
if entries:
snapshot = {
'request_count': sum(int(item.get('requests', 0) or 0) for item in entries),
'avg_latency_ms': sum(float(item.get('avg_latency_ms', 0) or 0) for item in entries) / max(len(entries), 1),
'error_rate': 0.0,
'total_tokens': sum(int(item.get('tokens', 0) or 0) for item in entries),
}
elif source_type == 'autoselect':
rows = analytics.get_autoselect_selection_breakdown()
entries = next((row.get('entries', []) for row in rows if row.get('autoselect_id') == source_id), [])
if entries:
snapshot = {
'request_count': sum(int(item.get('requests', 0) or 0) for item in entries),
'avg_latency_ms': sum(float(item.get('avg_latency_ms', 0) or 0) for item in entries) / max(len(entries), 1),
'error_rate': 0.0,
'total_tokens': sum(int(item.get('tokens', 0) or 0) for item in entries),
}
except Exception as exc:
logger.debug(f"Could not attach analytics snapshot for listing {listing.get('id')}: {exc}")
listing['analytics'] = snapshot
return listing
def _apply_listing_derived_fields(listing: dict, db):
listing['votes'] = db.get_market_vote_summary(listing['id'])
listing['stats'] = db.get_market_listing_stats(listing['id'])
listing['capabilities'] = _listing_capabilities(listing)
listing['model_type_label'] = _listing_model_type(listing)
listing['size_label'] = _listing_size_value(listing)
listing['online'] = False
return _attach_analytics_snapshot(listing)
async def _build_provider_listing_payload(provider_id: str, provider_config, owner_user_id: int, owner_username: str, source_scope: str, form: dict):
safe_config = _safe_provider_config(provider_config)
models = []
configured_models = safe_config.get('models', []) or []
provider_price_tokens = float(form.get('price_per_million_tokens') or 0)
provider_price_requests = float(form.get('price_per_1000_requests') or 0)
for model in configured_models:
model_name = model.get('name') if isinstance(model, dict) else getattr(model, 'name', None)
if not model_name:
continue
model_prices = (form.get('model_prices') or {}).get(model_name, {})
model_token_price = model_prices.get('price_per_million_tokens')
model_request_price = model_prices.get('price_per_1000_requests')
models.append({
'name': model_name,
'price_per_million_tokens': float(model_token_price) if model_token_price not in (None, '') else provider_price_tokens,
'price_per_1000_requests': float(model_request_price) if model_request_price not in (None, '') else provider_price_requests,
})
return {
'owner_user_id': owner_user_id,
'owner_username': owner_username,
'source_scope': source_scope,
'source_type': 'provider',
'source_id': provider_id,
'listing_key': f'provider:{provider_id}',
'title': form.get('title') or provider_id,
'description': form.get('description') or f'Market provider share for {provider_id}',
'provider_id': provider_id,
'model_id': None,
'endpoint': safe_config.get('endpoint'),
'currency_code': form.get('currency_code', 'USD'),
'price_per_million_tokens': provider_price_tokens,
'price_per_1000_requests': provider_price_requests,
'provider_price_per_million_tokens': provider_price_tokens,
'provider_price_per_1000_requests': provider_price_requests,
'metadata': {
'provider_type': safe_config.get('type'),
'models': models,
'source_scope': source_scope,
},
'config_snapshot': safe_config,
'is_active': True,
}
@router.get('/dashboard/market', response_class=HTMLResponse)
async def dashboard_market(request: Request):
auth_check = require_dashboard_auth(request)
if auth_check:
return auth_check
market_check, market_settings = _ensure_market_enabled(request)
if market_check:
return market_check
db = DatabaseRegistry.get_config_database()
listings = db.list_market_listings(active_only=True)
def _listing_online(listing):
provider_id = listing.get('provider_id')
owner_user_id = listing.get('owner_user_id') if listing.get('source_scope') == 'user' else None
if not provider_id:
return True
if db.get_provider_disabled_until(owner_user_id, provider_id):
return False
usage = db.get_provider_usage(owner_user_id, provider_id)
return bool(usage and usage.get('usage_data') is not None)
enriched = []
current_user_id = request.session.get('user_id')
for listing in listings:
listing_copy = dict(listing)
_apply_listing_derived_fields(listing_copy, db)
listing_copy['online'] = _listing_online(listing_copy)
owner = db.get_user_by_id(listing['owner_user_id']) if listing.get('owner_user_id') else None
listing_copy['owner_display_name'] = (owner or {}).get('display_name') or listing.get('owner_username')
enriched.append(listing_copy)
return _templates.TemplateResponse(
request=request,
name='dashboard/market.html',
context={
'request': request,
'session': request.session,
'market_listings_json': json.dumps(enriched),
'currency_symbol': db.get_currency_settings().get('currency_symbol', '$'),
'current_user_id': current_user_id,
'market_settings_json': json.dumps(market_settings),
},
)
@router.get('/api/market/listings')
async def api_market_listings(request: Request):
auth_check = require_api_auth(request)
if auth_check:
return auth_check
market_check, _ = _ensure_market_enabled(request)
if market_check:
return market_check
db = DatabaseRegistry.get_config_database()
listings = db.list_market_listings(active_only=request.query_params.get('include_inactive') != '1')
query = (request.query_params.get('q') or '').strip().lower()
source_type = (request.query_params.get('source_type') or '').strip().lower()
capabilities = [item.strip().lower() for item in (request.query_params.get('capabilities') or '').split(',') if item.strip()]
model_type = (request.query_params.get('model_type') or '').strip().lower()
model_name = (request.query_params.get('model_name') or '').strip().lower()
size = (request.query_params.get('size') or '').strip().lower()
online_only = request.query_params.get('online_only') == '1'
sort_by = (request.query_params.get('sort_by') or 'newest').strip().lower()
def _listing_online(listing):
provider_id = listing.get('provider_id')
owner_user_id = listing.get('owner_user_id') if listing.get('source_scope') == 'user' else None
if not provider_id:
return True
if db.get_provider_disabled_until(owner_user_id, provider_id):
return False
usage = db.get_provider_usage(owner_user_id, provider_id)
return bool(usage and usage.get('usage_data') is not None)
filtered = []
for listing in listings:
listing = _apply_listing_derived_fields(dict(listing), db)
metadata = listing.get('metadata') or {}
capability_values = listing.get('capabilities') or []
listing_model_name = str(listing.get('model_id') or metadata.get('model_name') or (metadata.get('model') or {}).get('name') or '').lower()
listing_model_type = listing.get('model_type_label') or ''
listing_size = listing.get('size_label') or ''
haystack = ' '.join([
str(listing.get('title') or ''), str(listing.get('description') or ''),
str(listing.get('provider_id') or ''), str(listing.get('model_id') or ''),
str(metadata.get('provider_type') or ''), str(metadata.get('model_name') or ''),
listing_model_type, listing_size, ' '.join(capability_values),
]).lower()
if query and query not in haystack:
continue
if source_type and listing.get('source_type', '').lower() != source_type:
continue
if capabilities and not all(cap in capability_values or cap in haystack for cap in capabilities):
continue
if model_type and model_type not in listing_model_type and model_type not in haystack:
continue
if model_name and model_name not in listing_model_name and model_name not in haystack:
continue
if size and size not in listing_size and size not in haystack:
continue
listing['online'] = _listing_online(listing)
if online_only and not listing['online']:
continue
filtered.append(listing)
if sort_by == 'price_asc':
filtered.sort(key=lambda item: (item.get('price_per_million_tokens') or 0, item.get('price_per_1000_requests') or 0, -(item.get('votes', {}).get('listing', {}).get('score', 0))))
elif sort_by == 'price_desc':
filtered.sort(key=lambda item: (-(item.get('price_per_million_tokens') or 0), -(item.get('price_per_1000_requests') or 0)))
elif sort_by == 'upvotes':
filtered.sort(key=lambda item: (-(item.get('votes', {}).get('listing', {}).get('score', 0)), -(item.get('votes', {}).get('user', {}).get('score', 0))))
elif sort_by == 'performance':
filtered.sort(key=lambda item: ((item.get('analytics', {}).get('avg_latency_ms') or 10**9), -(item.get('analytics', {}).get('request_count') or 0)))
elif sort_by == 'requests':
filtered.sort(key=lambda item: (-(item.get('stats', {}).get('total_requests') or 0), -(item.get('analytics', {}).get('request_count') or 0)))
elif sort_by == 'revenue':
filtered.sort(key=lambda item: (-(item.get('stats', {}).get('gross_revenue') or 0), -(item.get('stats', {}).get('provider_revenue') or 0)))
else:
filtered.sort(key=lambda item: str(item.get('created_at') or ''), reverse=True)
return JSONResponse({'listings': filtered})
@router.post('/api/market/listings')
async def api_publish_market_listing(request: Request):
auth_check = require_api_auth(request)
if auth_check:
return auth_check
market_check, market_settings = _ensure_market_enabled(request, require_admin_override=True)
if market_check:
return market_check
db = DatabaseRegistry.get_config_database()
user_id = request.session.get('user_id')
is_admin = request.session.get('role') == 'admin'
if not user_id and not is_admin:
return JSONResponse({'error': 'Only authenticated configurations can be published to market'}, status_code=400)
if user_id:
if not market_settings.get('allow_user_publish', True):
return JSONResponse({'error': 'User publishing is disabled'}, status_code=403)
user = db.get_user_by_id(user_id)
owner_user_id = user_id
owner_username = user['username']
source_scope = 'user'
else:
if not market_settings.get('allow_admin_publish', True):
return JSONResponse({'error': 'Admin publishing is disabled'}, status_code=403)
user = {'username': 'admin'}
owner_user_id = 0
owner_username = 'admin'
source_scope = 'global'
body = await request.json()
source_type = body.get('source_type')
source_id = body.get('source_id')
if source_type == 'provider':
if source_scope == 'user':
user_provider = db.get_user_provider(user_id, source_id)
if not user_provider:
return JSONResponse({'error': 'Provider not found'}, status_code=404)
provider_config = user_provider['config']
else:
provider_obj = _config.get_provider(source_id)
if not provider_obj:
return JSONResponse({'error': 'Provider not found'}, status_code=404)
provider_config = provider_obj.model_dump() if hasattr(provider_obj, 'model_dump') else vars(provider_obj)
payload = await _build_provider_listing_payload(
source_id,
provider_config,
owner_user_id,
owner_username,
source_scope,
body,
)
elif source_type == 'model':
provider_id = body.get('provider_id')
if source_scope == 'user':
user_provider = db.get_user_provider(user_id, provider_id)
if not user_provider:
return JSONResponse({'error': 'Provider not found'}, status_code=404)
provider_config = user_provider['config']
else:
provider_obj = _config.get_provider(provider_id)
if not provider_obj:
return JSONResponse({'error': 'Provider not found'}, status_code=404)
provider_config = provider_obj.model_dump() if hasattr(provider_obj, 'model_dump') else vars(provider_obj)
model_name = body.get('model_id')
model_config = next(
(
m for m in (provider_config.get('models') or [])
if ((m.get('name') if isinstance(m, dict) else getattr(m, 'name', None)) == model_name)
),
None,
)
if not model_config:
return JSONResponse({'error': 'Model not found'}, status_code=404)
payload = _build_model_listing_payload(owner_user_id, owner_username, provider_id, provider_config, model_config, source_scope, body)
elif source_type == 'rotation':
if source_scope == 'user':
rotation = db.get_user_rotation(user_id, source_id)
if not rotation:
return JSONResponse({'error': 'Rotation not found'}, status_code=404)
rotation_config = rotation['config']
else:
rotation_obj = _config.get_rotation(source_id)
if not rotation_obj:
return JSONResponse({'error': 'Rotation not found'}, status_code=404)
rotation_config = rotation_obj.model_dump() if hasattr(rotation_obj, 'model_dump') else vars(rotation_obj)
payload = _build_rotation_listing_payload(owner_user_id, owner_username, source_id, rotation_config, source_scope, body)
elif source_type == 'autoselect':
if source_scope == 'user':
autoselect = db.get_user_autoselect(user_id, source_id)
if not autoselect:
return JSONResponse({'error': 'Autoselect not found'}, status_code=404)
autoselect_config = autoselect['config']
else:
autoselect_obj = _config.get_autoselect(source_id)
if not autoselect_obj:
return JSONResponse({'error': 'Autoselect not found'}, status_code=404)
autoselect_config = autoselect_obj.model_dump() if hasattr(autoselect_obj, 'model_dump') else vars(autoselect_obj)
payload = _build_autoselect_listing_payload(owner_user_id, owner_username, source_id, autoselect_config, source_scope, body)
else:
return JSONResponse({'error': 'Unsupported source_type'}, status_code=400)
listing_id = db.upsert_market_listing(owner_user_id, owner_username, payload)
return JSONResponse({'success': True, 'listing_id': listing_id})
@router.post('/api/market/listings/{listing_id}/import')
async def api_import_market_listing(request: Request, listing_id: int):
auth_check = require_api_auth(request)
if auth_check:
return auth_check
market_check, market_settings = _ensure_market_enabled(request)
if market_check:
return market_check
if not market_settings.get('allow_import', True):
return JSONResponse({'error': 'Market imports are disabled'}, status_code=403)
user_id = request.session.get('user_id')
if not user_id:
return JSONResponse({'error': 'User account required'}, status_code=400)
db = DatabaseRegistry.get_config_database()
listing = db.get_market_listing(listing_id)
if not listing or not listing.get('is_active'):
return JSONResponse({'error': 'Listing not found'}, status_code=404)
snapshot = listing.get('config_snapshot') or {}
source_type = listing.get('source_type')
source_id = listing.get('source_id')
owner_username = listing.get('owner_username')
import_id = f'market/{owner_username}/{source_id}'
if source_type == 'provider':
imported_config = dict(snapshot)
imported_config['market_source'] = {
'listing_id': listing_id,
'owner_user_id': listing['owner_user_id'],
'owner_username': owner_username,
'provider_id': listing.get('provider_id'),
'model_id': listing.get('model_id'),
'source_type': source_type,
}
db.save_user_provider(user_id, import_id, imported_config)
db.record_market_import(user_id, listing_id, 'provider', import_id)
return JSONResponse({'success': True, 'imported_config_type': 'provider', 'imported_config_id': import_id})
if source_type == 'model':
provider_snapshot = dict((snapshot.get('provider') or {}))
model_snapshot = dict((snapshot.get('model') or {}))
provider_snapshot['models'] = [model_snapshot]
provider_snapshot['market_source'] = {
'listing_id': listing_id,
'owner_user_id': listing['owner_user_id'],
'owner_username': owner_username,
'provider_id': listing.get('provider_id'),
'model_id': listing.get('model_id'),
'source_type': source_type,
}
model_import_id = f"market/{owner_username}/{listing.get('provider_id')}/{listing.get('model_id')}"
db.save_user_provider(user_id, model_import_id, provider_snapshot)
db.record_market_import(user_id, listing_id, 'provider', model_import_id)
return JSONResponse({'success': True, 'imported_config_type': 'provider', 'imported_config_id': model_import_id})
if source_type == 'rotation':
imported_rotation_id = f'market/{owner_username}/{source_id}'
imported_rotation = dict(snapshot)
imported_rotation['market_source'] = {
'listing_id': listing_id,
'owner_user_id': listing['owner_user_id'],
'owner_username': owner_username,
'source_type': source_type,
}
db.save_user_rotation(user_id, imported_rotation_id, imported_rotation)
db.record_market_import(user_id, listing_id, 'rotation', imported_rotation_id)
return JSONResponse({'success': True, 'imported_config_type': 'rotation', 'imported_config_id': imported_rotation_id})
if source_type == 'autoselect':
imported_autoselect_id = f'market/{owner_username}/{source_id}'
imported_autoselect = dict(snapshot)
imported_autoselect['market_source'] = {
'listing_id': listing_id,
'owner_user_id': listing['owner_user_id'],
'owner_username': owner_username,
'source_type': source_type,
}
db.save_user_autoselect(user_id, imported_autoselect_id, imported_autoselect)
db.record_market_import(user_id, listing_id, 'autoselect', imported_autoselect_id)
return JSONResponse({'success': True, 'imported_config_type': 'autoselect', 'imported_config_id': imported_autoselect_id})
return JSONResponse({'error': 'Unsupported listing type'}, status_code=400)
@router.post('/api/market/listings/{listing_id}/vote')
async def api_vote_market_listing(request: Request, listing_id: int):
auth_check = require_api_auth(request)
if auth_check:
return auth_check
market_check, _ = _ensure_market_enabled(request)
if market_check:
return market_check
user_id = request.session.get('user_id')
if not user_id:
return JSONResponse({'error': 'User account required'}, status_code=400)
body = await request.json()
vote = int(body.get('vote', 0))
target_type = body.get('target_type', 'listing')
db = DatabaseRegistry.get_config_database()
listing = db.get_market_listing(listing_id)
if not listing:
return JSONResponse({'error': 'Listing not found'}, status_code=404)
target_map = {
'listing': listing.get('listing_key'),
'provider': listing.get('provider_id') or listing.get('listing_key'),
'model': listing.get('model_id') or listing.get('listing_key'),
'user': listing.get('owner_username'),
}
if target_type not in target_map:
return JSONResponse({'error': 'Invalid vote target'}, status_code=400)
db.upsert_market_vote(listing_id, user_id, target_type, target_map[target_type], vote)
return JSONResponse({'success': True, 'votes': db.get_market_vote_summary(listing_id)})
@router.get('/api/market/me/exports')
async def api_my_market_exports(request: Request):
auth_check = require_api_auth(request)
if auth_check:
return auth_check
market_check, _ = _ensure_market_enabled(request, require_admin_override=True)
if market_check:
return market_check
user_id = request.session.get('user_id')
if not user_id:
return JSONResponse({'exports': []})
db = DatabaseRegistry.get_config_database()
listings = [listing for listing in db.list_market_listings(active_only=False) if listing.get('owner_user_id') == user_id]
return JSONResponse({'exports': listings})
@router.post('/api/market/listings/{listing_id}/toggle')
async def api_toggle_market_listing(request: Request, listing_id: int):
auth_check = require_api_auth(request)
if auth_check:
return auth_check
market_check, _ = _ensure_market_enabled(request, require_admin_override=True)
if market_check:
return market_check
user_id = request.session.get('user_id')
body = await request.json()
db = DatabaseRegistry.get_config_database()
if request.session.get('role') == 'admin' and (user_id is None or body.get('admin_override')):
success = db.admin_set_market_listing_active(listing_id, bool(body.get('is_active', True)))
else:
success = db.set_market_listing_active(listing_id, user_id, bool(body.get('is_active', True)))
if not success:
return JSONResponse({'error': 'Listing not found or not owned by user'}, status_code=404)
return JSONResponse({'success': True})
@router.get('/dashboard/admin/market', response_class=HTMLResponse)
async def dashboard_admin_market(request: Request):
auth_check = require_admin(request)
if auth_check:
return auth_check
db = DatabaseRegistry.get_config_database()
listings = db.list_market_listings(active_only=False)
enriched_listings = []
for listing in listings:
listing_copy = dict(listing)
_apply_listing_derived_fields(listing_copy, db)
enriched_listings.append(listing_copy)
return _templates.TemplateResponse(
request=request,
name='dashboard/admin_market.html',
context={
'request': request,
'session': request.session,
'market_listings': enriched_listings,
},
)
aisbf/routes/user_api.py
View file @ c5907326
......@@ -18,6 +18,31 @@ def init(config, get_user_handler_fn):
logger = logging.getLogger(__name__)
def _resolve_market_provider_alias(model: str):
parts = (model or '').split('/')
if len(parts) < 3:
return None
username = parts[0].strip()
provider_id = parts[1].strip()
model_id = '/'.join(parts[2:]).strip()
if not username or not provider_id or not model_id:
return None
return username, provider_id, model_id
def _resolve_market_share_alias(model: str):
parts = (model or '').split('/')
if len(parts) < 3:
return None
username = parts[0].strip()
share_type = parts[1].strip()
share_id = parts[2].strip()
trailing = '/'.join(parts[3:]).strip() if len(parts) > 3 else ''
if share_type not in {'provider', 'rotation', 'autoselect'}:
return None
return {'username': username, 'share_type': share_type, 'share_id': share_id, 'trailing': trailing}
def parse_provider_from_model(model: str) -> tuple[str, str]:
if '/' in model:
parts = model.split('/', 1)
......@@ -257,6 +282,67 @@ async def user_chat_completions(request: Request, username: str, body: ChatCompl
authenticated_user = db.get_user_by_id(user_id)
if authenticated_user and authenticated_user['username'] != username:
raise HTTPException(status_code=403, detail="Access denied. Username in URL must match authenticated user.")
market_share = _resolve_market_share_alias(body.model)
if market_share and market_share['username'] != username:
db = DatabaseRegistry.get_config_database()
owner_username = market_share['username']
owner = db.get_user_by_username(owner_username)
if not owner:
raise HTTPException(status_code=404, detail=f"User '{owner_username}' not found")
listing = db.get_market_listing_for_share(owner_username, market_share['share_type'], market_share['share_id'])
if not listing:
raise HTTPException(status_code=404, detail='Market listing not found for requested resource')
if not user_id:
raise HTTPException(status_code=401, detail='Authentication required for market usage')
body_dict = body.model_dump()
share_type = market_share['share_type']
actual_model = market_share['trailing']
if share_type == 'provider':
provider_id = market_share['share_id']
owner_provider = db.get_user_provider(owner['id'], provider_id)
if not owner_provider:
raise HTTPException(status_code=404, detail=f"Provider '{provider_id}' not found for user '{owner_username}'")
owner_handler = _get_user_handler('request', owner['id'])
body_dict['model'] = actual_model
result = await owner_handler.handle_chat_completion(request, provider_id, body_dict) if not body.stream else await owner_handler.handle_streaming_chat_completion(request, provider_id, body_dict)
elif share_type == 'rotation':
owner_handler = _get_user_handler('rotation', owner['id'])
body_dict['model'] = market_share['share_id']
result = await owner_handler.handle_rotation_request(market_share['share_id'], body_dict, owner['id'], getattr(request.state, 'token_id', None))
elif share_type == 'autoselect':
owner_handler = _get_user_handler('autoselect', owner['id'])
body_dict['model'] = market_share['share_id']
result = await owner_handler.handle_autoselect_request(market_share['share_id'], body_dict, owner['id'], getattr(request.state, 'token_id', None)) if not body.stream else await owner_handler.handle_autoselect_streaming_request(market_share['share_id'], body_dict)
else:
raise HTTPException(status_code=400, detail='Unsupported market share type')
return result
market_alias = _resolve_market_provider_alias(body.model)
if market_alias and market_alias[0] != username:
owner_username, provider_id, actual_model = market_alias
db = DatabaseRegistry.get_config_database()
owner = db.get_user_by_username(owner_username)
if not owner:
raise HTTPException(status_code=404, detail=f"User '{owner_username}' not found")
owner_provider = db.get_user_provider(owner['id'], provider_id)
if not owner_provider:
raise HTTPException(status_code=404, detail=f"Provider '{provider_id}' not found for user '{owner_username}'")
listing = db.get_market_listing_for_share(owner_username, 'provider', provider_id)
if not listing:
raise HTTPException(status_code=404, detail='Market listing not found for requested provider')
if not user_id:
raise HTTPException(status_code=401, detail='Authentication required for market usage')
owner_handler = _get_user_handler('request', owner['id'])
body_dict = body.model_dump()
body_dict['model'] = actual_model
provider_config = owner_provider['config']
result = await owner_handler.handle_chat_completion(request, provider_id, body_dict) if not body.stream else await owner_handler.handle_streaming_chat_completion(request, provider_id, body_dict)
return result
provider_id, actual_model = parse_provider_from_model(body.model)
if not provider_id:
raise HTTPException(status_code=400, detail="Model must be in format 'provider/model', 'rotation/name', 'autoselect/name', 'user-provider/model', 'user-rotation/name', or 'user-autoselect/name'")
......
main.py
View file @ c5907326
......@@ -196,6 +196,7 @@ import aisbf.routes.dashboard.settings as _dash_settings
import aisbf.routes.dashboard.admin as _dash_admin
import aisbf.routes.dashboard.payments as _dash_payments
import aisbf.routes.dashboard.provider_auth as _dash_provider_auth
import aisbf.routes.dashboard.market as _dash_market
app.include_router(_auth_routes.router)
app.include_router(_api_routes.router)
......@@ -207,6 +208,7 @@ app.include_router(_dash_settings.router)
app.include_router(_dash_admin.router)
app.include_router(_dash_payments.router)
app.include_router(_dash_provider_auth.router)
app.include_router(_dash_market.router)
# Wallet routes
try:
......@@ -230,6 +232,7 @@ def _init_all_routers():
_dash_admin.init(config, templates)
_dash_payments.init(config, templates)
_dash_provider_auth.init(config, templates)
_dash_market.init(config, templates)
# ---------------------------------------------------------------------------
......
templates/base.html
View file @ c5907326
......@@ -834,6 +834,7 @@ along with this program. If not, see <https://www.gnu.org/licenses/>.
<a href="{{ url_for(request, '/dashboard/rotations') }}" {% if '/rotations' in request.path %}class="active"{% endif %} data-i18n="nav.rotations">Rotations</a>
<a href="{{ url_for(request, '/dashboard/autoselect') }}" {% if '/autoselect' in request.path %}class="active"{% endif %} data-i18n="nav.autoselect">Autoselect</a>
<a href="{{ url_for(request, '/dashboard/studio') }}" {% if '/studio' in request.path %}class="active"{% endif %} data-i18n="nav.studio">Studio</a>
<a href="{{ url_for(request, '/dashboard/market') }}" {% if '/market' in request.path and '/admin/market' not in request.path %}class="active"{% endif %}>Market</a>
<a href="{{ url_for(request, '/dashboard/prompts') }}" {% if '/prompts' in request.path %}class="active"{% endif %} data-i18n="nav.prompts">Prompts</a>
<a href="{{ url_for(request, '/dashboard/analytics') }}" {% if '/analytics' in request.path %}class="active"{% endif %} data-i18n="nav.analytics">Analytics</a>
{% if request.session.user_id %}
......@@ -845,6 +846,7 @@ along with this program. If not, see <https://www.gnu.org/licenses/>.
<a href="{{ url_for(request, '/dashboard/users') }}" {% if '/users' in request.path %}class="active"{% endif %} data-i18n="nav.users">Users</a>
<a href="{{ url_for(request, '/dashboard/settings') }}" {% if '/settings' in request.path %}class="active"{% endif %} data-i18n="nav.settings">Settings</a>
<a href="{{ url_for(request, '/dashboard/admin/tiers') }}" {% if '/admin/tiers' in request.path %}class="active"{% endif %} data-i18n="nav.tiers">Tiers</a>
<a href="{{ url_for(request, '/dashboard/admin/market') }}" {% if '/admin/market' in request.path %}class="active"{% endif %}>Market Admin</a>
<a href="{{ url_for(request, '/dashboard/admin/payment-settings') }}" {% if '/admin/payment-settings' in request.path %}class="active"{% endif %} data-i18n="nav.payment_settings">Payment Settings</a>
{% endif %}
{% if show_upgrade_button %}
......
templates/dashboard/admin_market.html 0 → 100644
View file @ c5907326
{% extends "base.html" %}
{% block title %}Admin Market{% endblock %}
{% block content %}
<h2 style="margin-bottom: 20px;"><i class="fas fa-store-alt" style="color: var(--color-link);"></i> Market Administration</h2>
<div style="background: var(--bg-panel); border: 1px solid var(--color-border); border-radius: 10px; padding: 16px; overflow:auto;">
<table style="width:100%; border-collapse: collapse;">
<thead>
<tr style="background: var(--bg-accent);">
<th style="padding:10px; text-align:left;">ID</th>
<th style="padding:10px; text-align:left;">Title</th>
<th style="padding:10px; text-align:left;">Owner</th>
<th style="padding:10px; text-align:left;">Source</th>
<th style="padding:10px; text-align:left;">Status</th>
<th style="padding:10px; text-align:right;">1M Tokens</th>
<th style="padding:10px; text-align:right;">1K Requests</th>
<th style="padding:10px; text-align:right;">Revenue</th>
<th style="padding:10px; text-align:right;">Requests</th>
<th style="padding:10px; text-align:center;">Active</th>
</tr>
</thead>
<tbody>
{% for listing in market_listings %}
<tr style="border-top:1px solid var(--color-border);">
<td style="padding:10px;">{{ listing.id }}</td>
<td style="padding:10px;">{{ listing.title }}</td>
<td style="padding:10px;">{{ listing.owner_username }}</td>
<td style="padding:10px;">{{ listing.source_type }} / {{ listing.source_id }}</td>
<td style="padding:10px;">{{ 'Online' if listing.online else 'Offline' }}</td>
<td style="padding:10px; text-align:right;">{{ listing.price_per_million_tokens }}</td>
<td style="padding:10px; text-align:right;">{{ listing.price_per_1000_requests }}</td>
<td style="padding:10px; text-align:right;">{{ '%.2f'|format((listing.stats or {}).gross_revenue or 0) }}</td>
<td style="padding:10px; text-align:right;">{{ (listing.stats or {}).total_requests or 0 }}</td>
<td style="padding:10px; text-align:center;">{{ 'Yes' if listing.is_active else 'No' }}</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
{% endblock %}
templates/dashboard/admin_payment_settings.html
View file @ c5907326
......@@ -19,6 +19,31 @@
</div>
</div>
<div style="background: var(--bg-panel); border: 2px solid #2ecc71; border-radius: 8px; padding: 20px; margin-bottom: 20px;">
<h3 style="margin: 0 0 20px 0; color: #2ecc71;">
<i class="fas fa-store me-2"></i>Market Settings
</h3>
<form id="marketSettingsForm">
<div style="display:grid; grid-template-columns: repeat(auto-fit, minmax(240px, 1fr)); gap: 15px; margin-bottom: 16px;">
<label style="display:flex; align-items:center; gap:8px; color: var(--color-text);">
<input type="checkbox" id="marketEnabled" {% if market_settings.enabled %}checked{% endif %}> Enable market
</label>
<label style="display:flex; align-items:center; gap:8px; color: var(--color-text);">
<input type="checkbox" id="marketAllowUserPublish" {% if market_settings.allow_user_publish %}checked{% endif %}> Allow user publish
</label>
<label style="display:flex; align-items:center; gap:8px; color: var(--color-text);">
<input type="checkbox" id="marketAllowAdminPublish" {% if market_settings.allow_admin_publish %}checked{% endif %}> Allow admin publish
</label>
<label style="display:flex; align-items:center; gap:8px; color: var(--color-text);">
<input type="checkbox" id="marketAllowImport" {% if market_settings.allow_import %}checked{% endif %}> Allow import
</label>
</div>
<button type="submit" class="btn" style="background:#2ecc71; color:white;">
<i class="fas fa-save me-2"></i>Save Market Settings
</button>
</form>
</div>
<!-- Global Currency Settings -->
<div style="background: var(--bg-panel); border: 2px solid #17a2b8; border-radius: 8px; padding: 20px; margin-bottom: 20px;">
<h3 style="margin: 0 0 20px 0; color: var(--color-link);">
......@@ -1337,6 +1362,26 @@ document.getElementById('currencySettingsForm').addEventListener('submit', funct
});
});
document.getElementById('marketSettingsForm').addEventListener('submit', function(e) {
e.preventDefault();
fetch('{{ url_for(request, "/api/admin/settings/market") }}', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({
enabled: document.getElementById('marketEnabled').checked,
allow_user_publish: document.getElementById('marketAllowUserPublish').checked,
allow_admin_publish: document.getElementById('marketAllowAdminPublish').checked,
allow_import: document.getElementById('marketAllowImport').checked,
})
}).then(r => r.json()).then(data => {
if (data.success) {
showToast('Market settings saved', 'success');
} else {
showToast(data.error || 'Failed to save market settings', 'danger');
}
}).catch(error => showToast('Error saving market settings: ' + error, 'danger'));
});
// Refresh crypto prices manually
function refreshCryptoPrices() {
['Btc', 'Eth', 'Usdt', 'Usdc'].forEach(crypto => {
......
templates/dashboard/admin_tier_form.html
View file @ c5907326
......@@ -79,6 +79,10 @@
<label style="display: block; margin-bottom: 5px; color: var(--color-muted); font-size: 12px;">Max Models per Autoselection</label>
<input type="number" name="max_autoselection_models" style="width: 100%; padding: 12px; border: 1px solid var(--color-border); border-radius: 5px; background: var(--bg-panel); color: var(--color-text); font-size: 14px;" min="-1" value="{{ tier.max_autoselection_models if tier else '-1' }}">
</div>
<div>
<label style="display: block; margin-bottom: 5px; color: var(--color-muted); font-size: 12px;">Market Fee %</label>
<input type="number" name="market_fee_percentage" style="width: 100%; padding: 12px; border: 1px solid var(--color-border); border-radius: 5px; background: var(--bg-panel); color: var(--color-text); font-size: 14px;" min="0" max="100" step="0.01" value="{{ tier.market_fee_percentage if tier else '10.0' }}">
</div>
</div>
</div>
</div>
......
templates/dashboard/admin_tiers.html
View file @ c5907326
......@@ -32,6 +32,7 @@
<th style="padding: 12px; text-align: center; border-bottom:1px solid var(--color-border); font-weight: 600;" data-i18n="tiers_page.col_max_autoselections">Max Autoselections</th>
<th style="padding: 12px; text-align: center; border-bottom:1px solid var(--color-border); font-weight: 600;" data-i18n="tiers_page.col_max_models_rotation">Max Models / Rotation</th>
<th style="padding: 12px; text-align: center; border-bottom:1px solid var(--color-border); font-weight: 600;" data-i18n="tiers_page.col_max_models_autoselect">Max Models / Autoselect</th>
<th style="padding: 12px; text-align: center; border-bottom:1px solid var(--color-border); font-weight: 600;">Market Fee %</th>
<th style="padding: 12px; text-align: center; border-bottom:1px solid var(--color-border); font-weight: 600;" data-i18n="tiers_page.col_status">Status</th>
<th style="padding: 12px; text-align: center; border-bottom:1px solid var(--color-border); font-weight: 600;" data-i18n="tiers_page.col_actions">Actions</th>
</tr>
......@@ -50,6 +51,7 @@
<td>{% if tier.max_autoselections == -1 %}<span data-i18n="tiers_page.unlimited">Unlimited</span>{% elif tier.max_autoselections == 0 %}<span data-i18n="tiers_page.blocked">Blocked</span>{% else %}{{ tier.max_autoselections }}{% endif %}</td>
<td>{% if tier.max_rotation_models == -1 %}<span data-i18n="tiers_page.unlimited">Unlimited</span>{% elif tier.max_rotation_models == 0 %}<span data-i18n="tiers_page.blocked">Blocked</span>{% else %}{{ tier.max_rotation_models }}{% endif %}</td>
<td>{% if tier.max_autoselection_models == -1 %}<span data-i18n="tiers_page.unlimited">Unlimited</span>{% elif tier.max_autoselection_models == 0 %}<span data-i18n="tiers_page.blocked">Blocked</span>{% else %}{{ tier.max_autoselection_models }}{% endif %}</td>
<td>{{ '%.2f'|format(tier.market_fee_percentage or 10.0) }}</td>
<td>{% if tier.is_active %}<span class="badge bg-success" data-i18n="tiers_page.active">Active</span>{% else %}<span class="badge bg-secondary" data-i18n="tiers_page.inactive">Inactive</span>{% endif %}</td>
<td style="white-space: nowrap;">
<a href="{{ url_for(request, '/dashboard/admin/tiers/edit') }}/{{ tier.id }}" class="btn btn-sm btn-outline-primary me-1" title="Edit this tier">
......@@ -404,4 +406,4 @@ function showToast(message, type) {
}, 5000);
}
</script>
{% endblock %}
\ No newline at end of file
{% endblock %}
templates/dashboard/market.html 0 → 100644
View file @ c5907326
{% extends "base.html" %}
{% block title %}Market - AISBF Dashboard{% endblock %}
{% block content %}
<h2 style="margin-bottom: 20px;"><i class="fas fa-store" style="color: var(--color-link);"></i> Market</h2>
<p style="color: var(--color-muted); margin-bottom: 20px;">Browse shared providers and models, review pricing and reputation, and import market resources into your account.</p>
<div style="background: var(--bg-panel); border: 1px solid var(--color-border); border-radius: 10px; padding: 16px; margin-bottom: 18px; display:grid; grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); gap: 12px;">
<input id="market-search" placeholder="Search name, type, capability" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
<select id="market-source-type" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
<option value="">All resource types</option>
<option value="provider">Provider</option>
<option value="model">Model</option>
<option value="rotation">Rotation</option>
<option value="autoselect">Autoselect</option>
</select>
<input id="market-capabilities" placeholder="Capabilities comma-separated" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
<input id="market-model-type" placeholder="Model type / family" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
<input id="market-model-name" placeholder="Model name" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
<input id="market-size" placeholder="Size / params" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
<label style="display:flex; align-items:center; gap:8px; color:var(--color-text);"><input type="checkbox" id="market-online-only"> Online only</label>
<select id="market-sort" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
<option value="newest">Newest</option>
<option value="price_asc">Price low to high</option>
<option value="price_desc">Price high to low</option>
<option value="upvotes">Best rated</option>
<option value="performance">Best performance</option>
<option value="requests">Most used</option>
<option value="revenue">Top grossing</option>
</select>
<button class="btn btn-secondary" onclick="reloadMarket()">Apply Filters</button>
</div>
<div style="display:grid; grid-template-columns: repeat(auto-fit, minmax(320px, 1fr)); gap: 16px;" id="market-grid"></div>
<div style="margin-top: 30px; background: var(--bg-panel); border: 1px solid var(--color-border); border-radius: 10px; padding: 18px;">
<h3 style="margin-top:0;">Publish one of your resources</h3>
<div style="display:grid; grid-template-columns: 1fr 1fr; gap: 12px;">
<select id="publish-source-type" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
<option value="provider">Provider</option>
<option value="model">Model</option>
<option value="rotation">Rotation</option>
<option value="autoselect">Autoselect</option>
</select>
<input id="publish-source-id" placeholder="Provider / model / rotation / autoselect id" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
<input id="publish-title" placeholder="Listing title" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
<input id="publish-provider-id" placeholder="Provider id for model listings" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
<input id="publish-price-tokens" type="number" step="0.0001" min="0" placeholder="Price / 1M tokens" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
<input id="publish-price-requests" type="number" step="0.0001" min="0" placeholder="Price / 1000 requests" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
</div>
<textarea id="publish-description" placeholder="Description" style="margin-top:12px; width:100%; min-height:84px; padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);"></textarea>
<button class="btn" style="margin-top:12px;" onclick="publishListing()">Publish Listing</button>
</div>
{% endblock %}
{% block extra_js %}
<script>
const listings = {{ market_listings_json | safe }};
const currencySymbol = {{ currency_symbol | tojson }};
const marketSettings = {{ market_settings_json | safe }};
function voteSummary(votes, target) {
const item = (votes || {})[target] || {upvotes:0, downvotes:0, score:0};
return `+${item.upvotes} / -${item.downvotes} (score ${item.score})`;
}
function renderMarket() {
const root = document.getElementById('market-grid');
root.innerHTML = '';
listings.forEach(listing => {
const el = document.createElement('div');
el.style.cssText = 'background: var(--bg-panel); border: 1px solid var(--color-border); border-radius: 12px; padding: 16px; box-shadow: 0 10px 30px rgba(0,0,0,0.08);';
el.innerHTML = `
<div style="display:flex; justify-content:space-between; gap:10px; align-items:flex-start;">
<div>
<h3 style="margin:0 0 6px 0;">${listing.title}</h3>
<div style="color: var(--color-muted); font-size: 13px;">by ${listing.owner_display_name || listing.owner_username}</div>
</div>
<div style="text-align:right; font-size:13px; color: var(--color-link);">
<div>${currencySymbol}${Number(listing.price_per_million_tokens || 0).toFixed(4)} / 1M tok</div>
<div>${currencySymbol}${Number(listing.price_per_1000_requests || 0).toFixed(4)} / 1k req</div>
</div>
</div>
<p style="color: var(--color-text); margin: 12px 0;">${listing.description || ''}</p>
<div style="font-size:12px; color: var(--color-muted); margin-bottom: 10px;">Provider: ${listing.provider_id || '-'} ${listing.model_id ? ` | Model: ${listing.model_id}` : ''} ${listing.model_type_label ? ` | Type: ${listing.model_type_label}` : ''} ${listing.size_label ? ` | Size: ${listing.size_label}` : ''}</div>
<div style="font-size:12px; color: var(--color-muted); margin-bottom: 10px;">Status: ${listing.online ? 'online' : 'offline'}${(listing.capabilities || []).length ? ` | Capabilities: ${(listing.capabilities || []).join(', ')}` : ''}</div>
<div style="display:grid; grid-template-columns:1fr 1fr; gap:6px; font-size:12px; margin-bottom:12px;">
<div>Listing votes: ${voteSummary(listing.votes, 'listing')}</div>
<div>User votes: ${voteSummary(listing.votes, 'user')}</div>
<div>Provider votes: ${voteSummary(listing.votes, 'provider')}</div>
<div>Model votes: ${voteSummary(listing.votes, 'model')}</div>
</div>
<div style="display:grid; grid-template-columns:1fr 1fr; gap:6px; font-size:12px; margin-bottom:12px; color: var(--color-muted);">
<div>Usage events: ${(listing.stats || {}).usage_events || 0}</div>
<div>Total requests: ${(listing.stats || {}).total_requests || 0}</div>
<div>Total tokens: ${(listing.stats || {}).total_tokens || 0}</div>
<div>Avg tokens/req: ${Number((listing.stats || {}).avg_tokens_per_request || 0).toFixed(1)}</div>
<div>Gross: ${currencySymbol}${Number((listing.stats || {}).gross_revenue || 0).toFixed(2)}</div>
<div>Seller rev: ${currencySymbol}${Number((listing.stats || {}).provider_revenue || 0).toFixed(2)}</div>
<div>Analytics req: ${(listing.analytics || {}).request_count || 0}</div>
<div>Avg latency: ${Number((listing.analytics || {}).avg_latency_ms || 0).toFixed(1)} ms</div>
<div>Error rate: ${(Number((listing.analytics || {}).error_rate || 0) * 100).toFixed(1)}%</div>
<div>Analytics tokens: ${(listing.analytics || {}).total_tokens || 0}</div>
</div>
<div style="display:flex; flex-wrap:wrap; gap:8px;">
<button class="btn btn-secondary" onclick="importListing(${listing.id})">Import</button>
<button class="btn btn-secondary" onclick="voteListing(${listing.id}, 'listing', 1)">Upvote</button>
<button class="btn btn-secondary" onclick="voteListing(${listing.id}, 'listing', -1)">Downvote</button>
</div>
`;
root.appendChild(el);
});
}
async function importListing(id) {
const response = await fetch(`/api/market/listings/${id}/import`, {method:'POST', headers:{'Content-Type':'application/json'}});
const result = await response.json();
if (!response.ok) {
alert(result.error || 'Import failed');
return;
}
alert(`Imported as ${result.imported_config_type}: ${result.imported_config_id}`);
}
async function voteListing(id, targetType, vote) {
const response = await fetch(`/api/market/listings/${id}/vote`, {
method:'POST',
headers:{'Content-Type':'application/json'},
body: JSON.stringify({target_type: targetType, vote})
});
const result = await response.json();
if (!response.ok) {
alert(result.error || 'Vote failed');
return;
}
const listing = listings.find(item => item.id === id);
if (listing) listing.votes = result.votes;
renderMarket();
}
async function publishListing() {
const body = {
source_type: document.getElementById('publish-source-type').value,
provider_id: document.getElementById('publish-provider-id').value.trim(),
source_id: document.getElementById('publish-source-id').value.trim(),
model_id: document.getElementById('publish-source-type').value === 'model' ? document.getElementById('publish-source-id').value.trim() : undefined,
title: document.getElementById('publish-title').value.trim(),
description: document.getElementById('publish-description').value.trim(),
price_per_million_tokens: parseFloat(document.getElementById('publish-price-tokens').value || '0'),
price_per_1000_requests: parseFloat(document.getElementById('publish-price-requests').value || '0')
};
const response = await fetch('/api/market/listings', {
method:'POST',
headers:{'Content-Type':'application/json'},
body: JSON.stringify(body)
});
const result = await response.json();
if (!response.ok) {
alert(result.error || 'Publish failed');
return;
}
window.location.reload();
}
renderMarket();
async function reloadMarket() {
const params = new URLSearchParams();
const q = document.getElementById('market-search').value.trim();
const sourceType = document.getElementById('market-source-type').value;
const capabilities = document.getElementById('market-capabilities').value.trim();
const modelType = document.getElementById('market-model-type').value.trim();
const modelName = document.getElementById('market-model-name').value.trim();
const size = document.getElementById('market-size').value.trim();
const onlineOnly = document.getElementById('market-online-only').checked;
const sortBy = document.getElementById('market-sort').value;
if (q) params.set('q', q);
if (sourceType) params.set('source_type', sourceType);
if (capabilities) params.set('capabilities', capabilities);
if (modelType) params.set('model_type', modelType);
if (modelName) params.set('model_name', modelName);
if (size) params.set('size', size);
if (onlineOnly) params.set('online_only', '1');
if (sortBy) params.set('sort_by', sortBy);
const response = await fetch(`/api/market/listings?${params.toString()}`);
const result = await response.json();
if (!response.ok) {
alert(result.error || 'Failed to load market');
return;
}
listings.length = 0;
for (const listing of result.listings || []) listings.push(listing);
renderMarket();
}
</script>
{% endblock %}
tests/payments/test_integration.py
View file @ c5907326
......@@ -13,6 +13,7 @@ import asyncio
from decimal import Decimal
from datetime import datetime, timedelta
from unittest.mock import Mock, patch, AsyncMock
from pathlib import Path
@pytest.fixture
......@@ -40,6 +41,85 @@ def db_manager():
return db
@pytest.fixture
def market_db_manager(tmp_path):
"""Create file-backed database for concurrent market settlement tests."""
from aisbf.database import DatabaseManager
from aisbf.payments.migrations import PaymentMigrations
db_path = Path(tmp_path) / "market_settlement_test.db"
db = DatabaseManager({
'type': 'sqlite',
'sqlite_path': str(db_path),
})
migrations = PaymentMigrations(db)
migrations.run_migrations()
with db._get_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO users (username, email, password_hash, role, email_verified)
VALUES ('buyer', 'buyer@example.com', 'hash', 'user', 1)
"""
)
buyer_id = cursor.lastrowid
cursor.execute(
"""
INSERT INTO users (username, email, password_hash, role, email_verified)
VALUES ('seller', 'seller@example.com', 'hash', 'user', 1)
"""
)
seller_id = cursor.lastrowid
cursor.execute(
"""
INSERT INTO user_wallets (user_id, balance, currency_code, created_at, updated_at)
VALUES (?, 100.00, 'USD', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
""",
(buyer_id,)
)
cursor.execute(
"""
INSERT INTO user_wallets (user_id, balance, currency_code, created_at, updated_at)
VALUES (?, 0.00, 'USD', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
""",
(seller_id,)
)
conn.commit()
listing_id = db.upsert_market_listing(
seller_id,
'seller',
{
'source_scope': 'user',
'source_type': 'provider',
'source_id': 'shared-provider',
'listing_key': 'provider:shared-provider',
'title': 'Shared Provider',
'description': 'Concurrent settlement test listing',
'provider_id': 'shared-provider',
'model_id': None,
'endpoint': 'https://example.test',
'currency_code': 'USD',
'price_per_million_tokens': 5.0,
'price_per_1000_requests': 0.0,
'provider_price_per_million_tokens': 5.0,
'provider_price_per_1000_requests': 0.0,
'metadata': {'provider_type': 'openai'},
'config_snapshot': {'provider': {'type': 'openai'}},
'is_active': True,
},
)
return {
'db': db,
'buyer_id': buyer_id,
'seller_id': seller_id,
'listing_id': listing_id,
}
@pytest.fixture
def payment_config():
"""Payment service configuration"""
......@@ -423,5 +503,128 @@ class TestEndToEndFlow:
assert status == 'active'
class TestMarketSettlementIdempotency:
def test_duplicate_market_settlement_is_idempotent(self, market_db_manager):
db = market_db_manager['db']
buyer_id = market_db_manager['buyer_id']
seller_id = market_db_manager['seller_id']
listing_id = market_db_manager['listing_id']
async def run_duplicate_settlements():
metadata = {'market_request_id': 'shared-request-1', 'kind': 'chat_completion'}
return await asyncio.gather(
asyncio.to_thread(
db.settle_market_usage,
buyer_id,
listing_id,
500000,
500000,
1,
dict(metadata),
),
asyncio.to_thread(
db.settle_market_usage,
buyer_id,
listing_id,
500000,
500000,
1,
dict(metadata),
),
)
first, second = asyncio.run(run_duplicate_settlements())
assert sorted([first['charged'], second['charged']]) == [False, True]
deduplicated = first if not first['charged'] else second
charged = first if first['charged'] else second
assert deduplicated['deduplicated'] is True
assert charged['charged_amount'] == Decimal('5.00')
assert charged['seller_amount'] == Decimal('4.50')
assert charged['platform_fee'] == Decimal('0.50')
with db._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM market_usage_transactions WHERE listing_id = ?", (listing_id,))
usage_count = cursor.fetchone()[0]
cursor.execute(
"SELECT COUNT(*) FROM wallet_transactions WHERE user_id = ? AND type = 'debit' AND description = ?",
(buyer_id, 'Market usage: Shared Provider')
)
debit_count = cursor.fetchone()[0]
cursor.execute("SELECT balance FROM user_wallets WHERE user_id = ?", (buyer_id,))
buyer_balance = Decimal(str(cursor.fetchone()[0]))
cursor.execute("SELECT balance FROM user_wallets WHERE user_id = ?", (seller_id,))
seller_balance = Decimal(str(cursor.fetchone()[0]))
assert usage_count == 1
assert debit_count == 1
assert buyer_balance == Decimal('95.00')
assert seller_balance == Decimal('4.50')
def test_global_admin_listing_books_platform_revenue_without_seller_credit(self, market_db_manager):
db = market_db_manager['db']
buyer_id = market_db_manager['buyer_id']
admin_listing_id = db.upsert_market_listing(
0,
'admin',
{
'source_scope': 'global',
'source_type': 'provider',
'source_id': 'global-provider',
'listing_key': 'provider:global-provider',
'title': 'Global Provider',
'description': 'Global admin market listing',
'provider_id': 'global-provider',
'model_id': None,
'endpoint': 'https://example.test/global',
'currency_code': 'USD',
'price_per_million_tokens': 6.0,
'price_per_1000_requests': 0.0,
'provider_price_per_million_tokens': 6.0,
'provider_price_per_1000_requests': 0.0,
'metadata': {'provider_type': 'openai'},
'config_snapshot': {'provider': {'type': 'openai'}},
'is_active': True,
},
)
result = db.settle_market_usage(
consumer_user_id=buyer_id,
listing_id=admin_listing_id,
prompt_tokens=500000,
completion_tokens=500000,
requests_count=1,
metadata={'market_request_id': 'global-admin-request-1', 'kind': 'chat_completion'},
)
assert result['charged'] is True
assert result['charged_amount'] == Decimal('6.00')
assert result['seller_amount'] == Decimal('0.00')
assert result['platform_fee'] == Decimal('0.00')
assert result['platform_revenue'] == Decimal('6.00')
with db._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT balance FROM user_wallets WHERE user_id = ?", (buyer_id,))
buyer_balance = Decimal(str(cursor.fetchone()[0]))
cursor.execute("SELECT COUNT(*) FROM wallet_transactions WHERE description = ?", ('Market sale: Global Provider',))
seller_credit_count = cursor.fetchone()[0]
cursor.execute(
"SELECT gross_amount, platform_fee, provider_amount, metadata FROM market_usage_transactions WHERE listing_id = ?",
(admin_listing_id,)
)
txn = cursor.fetchone()
assert buyer_balance == Decimal('94.00')
assert seller_credit_count == 0
assert Decimal(str(txn[0])) == Decimal('6.00')
assert Decimal(str(txn[1])) == Decimal('0.00')
assert Decimal(str(txn[2])) == Decimal('0.00')
assert 'platform_owned_listing' in str(txn[3])
if __name__ == '__main__':
pytest.main([__file__, '-v'])
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment