服务状态
状态:
运行中
版本:
1.0.0
令牌状态:
{('有效' if token_valid else '无效')}
令牌过期时间:
{expiry_readable}
SambaNova 凭据:
{('已配置' if settings.SAMBA_EMAIL and settings.SAMBA_PASSWORD else '未配置')}
本地API密钥:
{('已配置' if settings.LOCAL_API_KEY else '未配置')}
"""
return html_content
class SambaAuthAsync:
def __init__(self, email, password):
self.email = email
self.password = password
self.client = httpx.AsyncClient()
self.ua = UserAgent()
self.base_headers = {
"accept": "*/*",
"accept-language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7",
"origin": "https://cloud.sambanova.ai",
"referer": "https://cloud.sambanova.ai/",
"user-agent": self.ua.random
}
self.config = None
self.nonce = None # 确保nonce属性存在
async def _get_config(self):
"""获取动态配置信息"""
config_url = "https://cloud.sambanova.ai/api/config"
response = await self.client.get(config_url, headers=self.base_headers)
response.raise_for_status()
self.config = response.json()
print(f"[配置获取成功] ClientID: {self.config['clientId']}")
async def _get_login_ticket(self):
"""获取登录票据"""
auth_url = f"https://{self.config['issuerBaseUrl']}/co/authenticate"
payload = {
"client_id": self.config["clientId"],
"username": self.email,
"password": self.password,
"realm": "Username-Password-Authentication",
"credential_type": "http://auth0.com/oauth/grant-type/password-realm"
}
headers = {**self.base_headers, "content-type": "application/json"}
response = await self.client.post(auth_url, headers=headers, json=payload)
response.raise_for_status()
return response.json()["login_ticket"]
async def _get_auth_code(self, login_ticket: str):
"""获取授权码"""
state = secrets.token_urlsafe(32)
self.nonce = secrets.token_urlsafe(32) # 保存nonce到实例变量
params = {
"client_id": self.config["clientId"],
"response_type": "code",
"redirect_uri": self.config["redirectURL"],
"scope": "openid profile email",
"nonce": self.nonce,
"state": state,
"login_ticket": login_ticket,
"realm": "Username-Password-Authentication",
"auth0Client": "eyJuYW1lIjoibG9jay5qcyIsInZlcnNpb24iOiIxMi4zLjAiLCJlbnYiOnsiYXV0aDAuanMiOiI5LjIyLjEiLCJhdXRoMC5qcy11bHAiOiI5LjIyLjEifX0="
}
auth_url = f"https://{self.config['issuerBaseUrl']}/authorize"
response = await self.client.get(
auth_url,
params=params,
follow_redirects=False
)
if response.status_code == 302:
location = response.headers["location"]
parsed = urllib.parse.urlparse(location)
query = urllib.parse.parse_qs(parsed.query)
return query.get("code", [None])[0], state
raise Exception(f"未收到302重定向，实际状态码：{response.status_code}")
async def _exchange_token(self, code: str, state: str):
"""交换访问令牌"""
# 设置必要的cookies
self.client.cookies.set("nonce", self.nonce, domain="cloud.sambanova.ai")
callback_url = f"{self.config['redirectURL']}?code={code}&state={state}"
response = await self.client.get(
callback_url,
headers={
**self.base_headers,
"sec-fetch-site": "same-site",
"sec-fetch-mode": "navigate",
"sec-fetch-user": "?1",
"sec-fetch-dest": "document"
},
follow_redirects=True
)
# 从cookies中提取access_token
for cookie in self.client.cookies.jar:
if cookie.name == "access_token" and "sambanova.ai" in cookie.domain:
return cookie.value
raise Exception("未找到access_token")
async def login(self):
"""完整登录流程"""
try:
await self._get_config()
login_ticket = await self._get_login_ticket()
print(f"[登录票据获取成功] 完整票据: {login_ticket}")
auth_code, state = await self._get_auth_code(login_ticket)
if not auth_code:
raise Exception("授权码获取失败")
print(f"[授权码获取成功] 完整授权码: {auth_code}")
print(f"[授权状态] state: {state}")
token = await self._exchange_token(auth_code, state)
print(f"[令牌获取成功] 完整令牌: {token}")
return token
except Exception as e:
print(f"[登录失败] 详细错误: {str(e)}")
return None
finally:
await self.client.aclose()
@app.on_event("startup")
async def startup_event():
"""应用启动时预获取令牌"""
print("\n" + "="*50)
print("[启动] SambaNova OpenAI 代理服务启动")
print("="*50)
# 检查环境变量
print(f"[环境] SAMBA_EMAIL: {'已设置' if settings.SAMBA_EMAIL else '未设置'}")
print(f"[环境] SAMBA_PASSWORD: {'已设置' if settings.SAMBA_PASSWORD else '未设置'}")
print(f"[环境] LOCAL_API_KEY: {'已设置' if settings.LOCAL_API_KEY else '未设置'}")
# 尝试直接登录
print("[登录] 开始尝试登录...")
try:
auth = SambaAuthAsync(settings.SAMBA_EMAIL, settings.SAMBA_PASSWORD)
token = await auth.login()
if token:
global access_token, token_expiry
access_token = token
token_expiry = time.time() + settings.TOKEN_CACHE_TIME
print(f"[登录] 登录成功! 令牌: {token}")
print(f"[登录] 令牌将在 {settings.TOKEN_CACHE_TIME} 秒后过期")
else:
print("[登录] 登录失败，未获取到令牌")
except Exception as e:
print(f"[登录] 登录过程发生异常: {str(e)}")
print("="*50 + "\n")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860)