release: the first version

This commit is contained in:
Chenyme committed 2025-10-10 16:59:56 +08:00
1 parent 59b1560b36
commit ba617455f0
30 files changed
+3944 -1

No files matched your search

+136
View File
@@ -0,0 +1,136 @@
# Python 运行时和缓存文件
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
*.egg-info/
dist/
build/
*.egg
.eggs/
pip-log.txt
pip-delete-this-directory.txt
# 虚拟环境
venv/
env/
ENV/
.venv/
.virtualenv/
pipenv/
poetry.lock
Pipfile.lock
# IDE 和编辑器
.vscode/
.idea/
*.swp
*.swo
*.swn
.DS_Store
*~
.project
.classpath
.settings/
*.sublime-project
*.sublime-workspace
# Git
.git/
.gitignore
.gitattributes
.gitmodules
# Docker 相关
Dockerfile*
docker-compose*.yml
.dockerignore
.docker/
# 文档和说明
*.md
!data/*.md
README.md
docs/
LICENSE
CHANGELOG*
CONTRIBUTING*
# 测试和质量检查
tests/
test/
*.pytest_cache/
.coverage
.coverage.*
htmlcov/
.tox/
.mypy_cache/
.ruff_cache/
.pytest_cache/
.hypothesis/
.pytype/
coverage.xml
*.cover
.cache/
nosetests.xml
# 日志和运行时数据
logs/
*.log
*.log.*
# 数据库文件
*.db
*.sqlite
*.sqlite3
# 临时文件
tmp/
temp/
*.tmp
*.temp
*.bak
*.orig
*.rej
*.swp
# CI/CD
.github/
.gitlab-ci.yml
.travis.yml
.circleci/
azure-pipelines.yml
.jenkins/
Jenkinsfile
# 环境变量和密钥
.env
.env.*
*.key
*.pem
*.crt
secrets/
# 媒体和大文件
*.mp4
*.avi
*.mov
*.zip
*.tar
*.tar.gz
*.rar
# Node.js
node_modules/
npm-debug.log*
yarn-debug.log*
yarn-error.log*
package-lock.json
yarn.lock
# 其他
*.pyc
.Python
.sass-cache/
.ipynb_checkpoints/
+58
View File
@@ -0,0 +1,58 @@
name: Build Docker Image
on:
push:
branches:
- main
tags:
- 'v*'
pull_request:
branches:
- main
env:
REGISTRY: ghcr.io
IMAGE_NAME: ${{ github.repository }}
jobs:
build-and-push:
runs-on: ubuntu-latest
permissions:
contents: read
packages: write
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Log in to Container Registry
uses: docker/login-action@v3
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Extract metadata
id: meta
uses: docker/metadata-action@v5
with:
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
tags: |
type=ref,event=branch
type=ref,event=pr
type=semver,pattern={{version}}
type=semver,pattern={{major}}.{{minor}}
type=raw,value=latest,enable={{is_default_branch}}
- name: Build and push Docker image
uses: docker/build-push-action@v5
with:
context: .
push: true
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
cache-from: type=gha
cache-to: type=gha,mode=max
+184
View File
@@ -0,0 +1,184 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
.idea
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# UV
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
#uv.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/latest/usage/project/#working-with-version-control
.pdm.toml
.pdm-python
.pdm-build/
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
logs/*
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
# Ruff stuff:
.ruff_cache/
# PyPI configuration file
.pypirc
# Cursor
# Cursor is an AI-powered code editor.`.cursorignore` specifies files/directories to
# exclude from AI features like autocomplete and code analysis. Recommended for sensitive data
# refer to https://docs.cursor.com/context/ignore-files
.cursorignore
.cursorindexingignore
+37
View File
@@ -0,0 +1,37 @@
FROM python:3.11-slim AS base
WORKDIR /app
# 安装依赖阶段
FROM base AS dependencies
# 安装运行时需要的系统依赖
RUN apt-get update && \
apt-get install -y --no-install-recommends \
gcc \
&& rm -rf /var/lib/apt/lists/*
# 复制并安装 Python 依赖
COPY requirements.txt .
RUN pip install --no-cache-dir --upgrade pip && \
pip install --no-cache-dir -r requirements.txt
# 最终运行阶段
FROM python:3.11-slim AS runtime
WORKDIR /app
# 复制必要的 Python 依赖
COPY --from=dependencies /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY --from=dependencies /usr/local/bin /usr/local/bin
# 复制应用代码
COPY . .
# 创建必要的目录和文件
RUN mkdir -p /app/logs /app/data/temp && \
echo '{"ssoNormal": {}, "ssoSuper": {}}' > /app/data/token.json
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
+1 -1
View File
@@ -18,4 +18,4 @@ FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
SOFTWARE.
+594
View File
@@ -0,0 +1,594 @@
"""
管理接口模块
提供Token管理功能,包括登录验证、Token增删查等操作。
"""
import secrets
from typing import Dict, Any, List, Optional
from datetime import datetime, timedelta
from pathlib import Path
from fastapi import APIRouter, HTTPException, Depends, Header
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from app.core.config import setting
from app.core.logger import logger
from app.services.grok.token import token_manager
from app.models.grok_models import TokenType
# 创建路由器
router = APIRouter(tags=["管理"])
# 静态文件路径
STATIC_DIR = Path(__file__).parents[2] / "template"
# 简单的会话存储
_sessions: Dict[str, datetime] = {}
# === 请求/响应模型 ===
class LoginRequest(BaseModel):
"""登录请求"""
username: str
password: str
class LoginResponse(BaseModel):
"""登录响应"""
success: bool
token: Optional[str] = None
message: str
class AddTokensRequest(BaseModel):
"""批量添加Token请求"""
tokens: List[str]
token_type: str # "sso" 或 "ssoSuper"
class DeleteTokensRequest(BaseModel):
"""批量删除Token请求"""
tokens: List[str]
token_type: str # "sso" 或 "ssoSuper"
class TokenInfo(BaseModel):
"""Token信息"""
token: str
token_type: str
created_time: Optional[int] = None
remaining_queries: int
heavy_remaining_queries: int
status: str # "未使用"、"限流中"、"失效"、"正常"
class TokenListResponse(BaseModel):
"""Token列表响应"""
success: bool
data: List[TokenInfo]
total: int
# === 辅助函数 ===
def validate_token_type(token_type_str: str) -> TokenType:
"""验证并转换Token类型字符串为枚举"""
if token_type_str not in ["sso", "ssoSuper"]:
raise HTTPException(
status_code=400,
detail={"error": "无效的Token类型,必须是 'sso' 或 'ssoSuper'", "code": "INVALID_TYPE"}
)
return TokenType.NORMAL if token_type_str == "sso" else TokenType.SUPER
def parse_created_time(created_time) -> Optional[int]:
"""解析创建时间,统一处理不同格式"""
if isinstance(created_time, str):
return int(created_time) if created_time else None
elif isinstance(created_time, int):
return created_time
return None
def calculate_token_stats(tokens: Dict[str, Any], token_type: str) -> Dict[str, int]:
"""计算Token统计信息"""
total = len(tokens)
expired = sum(1 for t in tokens.values() if t.get("status") == "expired")
if token_type == "normal":
unused = sum(1 for t in tokens.values()
if t.get("status") != "expired" and t.get("remainingQueries", -1) == -1)
limited = sum(1 for t in tokens.values()
if t.get("status") != "expired" and t.get("remainingQueries", -1) == 0)
active = sum(1 for t in tokens.values()
if t.get("status") != "expired" and t.get("remainingQueries", -1) > 0)
else: # super token
unused = sum(1 for t in tokens.values()
if t.get("status") != "expired" and
t.get("remainingQueries", -1) == -1 and t.get("heavyremainingQueries", -1) == -1)
limited = sum(1 for t in tokens.values()
if t.get("status") != "expired" and
(t.get("remainingQueries", -1) == 0 or t.get("heavyremainingQueries", -1) == 0))
active = sum(1 for t in tokens.values()
if t.get("status") != "expired" and
(t.get("remainingQueries", -1) > 0 or t.get("heavyremainingQueries", -1) > 0))
return {
"total": total,
"unused": unused,
"limited": limited,
"expired": expired,
"active": active
}
def verify_admin_session(authorization: Optional[str] = Header(None)) -> bool:
"""验证管理员会话"""
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(
status_code=401,
detail={"error": "未授权访问", "code": "UNAUTHORIZED"}
)
token = authorization[7:] # 移除 "Bearer " 前缀
# 检查token是否存在且未过期
if token not in _sessions:
raise HTTPException(
status_code=401,
detail={"error": "会话已过期或无效", "code": "SESSION_INVALID"}
)
# 检查会话是否过期(24小时)
if datetime.now() > _sessions[token]:
del _sessions[token]
raise HTTPException(
status_code=401,
detail={"error": "会话已过期", "code": "SESSION_EXPIRED"}
)
return True
def get_token_status(token_data: Dict[str, Any], token_type: str) -> str:
"""获取Token状态"""
# 首先检查是否失效(来自 token.json 的 status 字段)
if token_data.get("status") == "expired":
return "失效"
# 获取剩余次数
remaining_queries = token_data.get("remainingQueries", -1)
heavy_remaining = token_data.get("heavyremainingQueries", -1)
# 根据token类型选择正确的字段
if token_type == "ssoSuper":
# Super token 可能使用 heavy 模型
relevant_remaining = max(remaining_queries, heavy_remaining)
else:
# 普通token主要看 remaining_queries
relevant_remaining = remaining_queries
if relevant_remaining == -1:
return "未使用"
elif relevant_remaining == 0:
return "限流中"
else:
return "正常"
# === 页面路由 ===
@router.get("/login", response_class=HTMLResponse)
async def login_page():
"""登录页面"""
login_html = STATIC_DIR / "login.html"
if login_html.exists():
return login_html.read_text(encoding="utf-8")
raise HTTPException(status_code=404, detail="登录页面不存在")
@router.get("/manage", response_class=HTMLResponse)
async def manage_page():
"""管理页面"""
admin_html = STATIC_DIR / "admin.html"
if admin_html.exists():
return admin_html.read_text(encoding="utf-8")
raise HTTPException(status_code=404, detail="管理页面不存在")
# === API端点 ===
@router.post("/api/login", response_model=LoginResponse)
async def admin_login(request: LoginRequest) -> LoginResponse:
"""
管理员登录
验证用户名和密码,成功后返回会话token。
"""
try:
logger.debug(f"[Admin] 管理员登录尝试 - 用户名: {request.username}")
# 验证用户名和密码
expected_username = setting.global_config.get("admin_username", "")
expected_password = setting.global_config.get("admin_password", "")
if request.username != expected_username or request.password != expected_password:
logger.warning(f"[Admin] 登录失败: 用户名或密码错误 - 用户名: {request.username}")
return LoginResponse(
success=False,
message="用户名或密码错误"
)
# 生成会话token
session_token = secrets.token_urlsafe(32)
# 设置会话过期时间(24小时)
expire_time = datetime.now() + timedelta(hours=24)
_sessions[session_token] = expire_time
logger.debug(f"[Admin] 管理员登录成功 - 用户名: {request.username}")
return LoginResponse(
success=True,
token=session_token,
message="登录成功"
)
except Exception as e:
logger.error(f"[Admin] 登录处理异常 - 用户名: {request.username}, 错误: {str(e)}")
raise HTTPException(
status_code=500,
detail={"error": f"登录失败: {str(e)}", "code": "LOGIN_ERROR"}
)
@router.post("/api/logout")
async def admin_logout(authenticated: bool = Depends(verify_admin_session),
authorization: Optional[str] = Header(None)) -> Dict[str, Any]:
"""
管理员登出
清除会话token。
"""
try:
if authorization and authorization.startswith("Bearer "):
token = authorization[7:]
if token in _sessions:
del _sessions[token]
logger.debug("[Admin] 管理员登出成功")
return {"success": True, "message": "登出成功"}
logger.warning("[Admin] 登出失败: 无效或缺失的会话Token")
return {"success": False, "message": "无效的会话"}
except Exception as e:
logger.error(f"[Admin] 登出处理异常 - 错误: {str(e)}")
raise HTTPException(
status_code=500,
detail={"error": f"登出失败: {str(e)}", "code": "LOGOUT_ERROR"}
)
@router.get("/api/tokens", response_model=TokenListResponse)
async def list_tokens(authenticated: bool = Depends(verify_admin_session)) -> TokenListResponse:
"""
获取所有Token列表
返回系统中所有Token及其状态信息。
"""
try:
logger.debug("[Admin] 开始获取Token列表")
all_tokens_data = token_manager.get_all_token()
token_list: List[TokenInfo] = []
# 处理普通Token
normal_tokens = all_tokens_data.get(TokenType.NORMAL.value, {})
for token, data in normal_tokens.items():
token_list.append(TokenInfo(
token=token,
token_type="sso",
created_time=parse_created_time(data.get("createdTime")),
remaining_queries=data.get("remainingQueries", -1),
heavy_remaining_queries=data.get("heavyremainingQueries", -1),
status=get_token_status(data, "sso")
))
# 处理Super Token
super_tokens = all_tokens_data.get(TokenType.SUPER.value, {})
for token, data in super_tokens.items():
token_list.append(TokenInfo(
token=token,
token_type="ssoSuper",
created_time=parse_created_time(data.get("createdTime")),
remaining_queries=data.get("remainingQueries", -1),
heavy_remaining_queries=data.get("heavyremainingQueries", -1),
status=get_token_status(data, "ssoSuper")
))
normal_count = len(normal_tokens)
super_count = len(super_tokens)
total_count = len(token_list)
logger.debug(f"[Admin] Token列表获取成功 - 普通Token: {normal_count}, Super Token: {super_count}, 总计: {total_count}")
return TokenListResponse(
success=True,
data=token_list,
total=total_count
)
except Exception as e:
logger.error(f"[Admin] 获取Token列表异常 - 错误: {str(e)}")
raise HTTPException(
status_code=500,
detail={"error": f"获取Token列表失败: {str(e)}", "code": "LIST_ERROR"}
)
@router.post("/api/tokens/add")
async def add_tokens(request: AddTokensRequest,
authenticated: bool = Depends(verify_admin_session)) -> Dict[str, Any]:
"""
批量添加Token
支持添加普通Token(sso)和Super Token(ssoSuper)。
"""
try:
logger.debug(f"[Admin] 批量添加Token - 类型: {request.token_type}, 数量: {len(request.tokens)}")
# 验证并转换token类型
token_type = validate_token_type(request.token_type)
# 添加Token
token_manager.add_token(request.tokens, token_type)
logger.debug(f"[Admin] Token添加成功 - 类型: {request.token_type}, 数量: {len(request.tokens)}")
return {
"success": True,
"message": f"成功添加 {len(request.tokens)} 个Token",
"count": len(request.tokens)
}
except HTTPException:
raise
except Exception as e:
logger.error(f"[Admin] Token添加异常 - 类型: {request.token_type}, 数量: {len(request.tokens)}, 错误: {str(e)}")
raise HTTPException(
status_code=500,
detail={"error": f"添加Token失败: {str(e)}", "code": "ADD_ERROR"}
)
@router.post("/api/tokens/delete")
async def delete_tokens(request: DeleteTokensRequest,
authenticated: bool = Depends(verify_admin_session)) -> Dict[str, Any]:
"""
批量删除Token
支持删除普通Token(sso)和Super Token(ssoSuper)。
"""
try:
logger.debug(f"[Admin] 批量删除Token - 类型: {request.token_type}, 数量: {len(request.tokens)}")
# 验证并转换token类型
token_type = validate_token_type(request.token_type)
# 删除Token
token_manager.delete_token(request.tokens, token_type)
logger.debug(f"[Admin] Token删除成功 - 类型: {request.token_type}, 数量: {len(request.tokens)}")
return {
"success": True,
"message": f"成功删除 {len(request.tokens)} 个Token",
"count": len(request.tokens)
}
except HTTPException:
raise
except Exception as e:
logger.error(f"[Admin] Token删除异常 - 类型: {request.token_type}, 数量: {len(request.tokens)}, 错误: {str(e)}")
raise HTTPException(
status_code=500,
detail={"error": f"删除Token失败: {str(e)}", "code": "DELETE_ERROR"}
)
@router.get("/api/settings")
async def get_settings(authenticated: bool = Depends(verify_admin_session)) -> Dict[str, Any]:
"""获取全局配置"""
try:
logger.debug("[Admin] 获取全局配置")
return {
"success": True,
"data": {
"global": setting.global_config,
"grok": setting.grok_config
}
}
except Exception as e:
logger.error(f"[Admin] 获取配置失败: {str(e)}")
raise HTTPException(status_code=500, detail={"error": f"获取配置失败: {str(e)}", "code": "GET_SETTINGS_ERROR"})
class UpdateSettingsRequest(BaseModel):
"""更新配置请求"""
global_config: Optional[Dict[str, Any]] = None
grok_config: Optional[Dict[str, Any]] = None
@router.post("/api/settings")
async def update_settings(request: UpdateSettingsRequest, authenticated: bool = Depends(verify_admin_session)) -> Dict[str, Any]:
"""更新全局配置"""
try:
import toml
logger.debug("[Admin] 更新全局配置")
# 读取现有配置
with open(setting.config_path, "r", encoding="utf-8") as f:
config = toml.load(f)
# 更新配置
if request.global_config:
config["global"].update(request.global_config)
if request.grok_config:
config["grok"].update(request.grok_config)
# 写回配置文件
with open(setting.config_path, "w", encoding="utf-8") as f:
toml.dump(config, f)
# 重新加载配置
setting.global_config = setting.load("global")
setting.grok_config = setting.load("grok")
logger.debug("[Admin] 配置更新成功")
return {"success": True, "message": "配置更新成功"}
except Exception as e:
logger.error(f"[Admin] 更新配置失败: {str(e)}")
raise HTTPException(status_code=500, detail={"error": f"更新配置失败: {str(e)}", "code": "UPDATE_SETTINGS_ERROR"})
@router.get("/api/cache/size")
async def get_cache_size(authenticated: bool = Depends(verify_admin_session)) -> Dict[str, Any]:
"""
获取缓存大小
计算/data/temp目录中所有文件的大小。
"""
try:
logger.debug("[Admin] 开始获取缓存大小")
# 获取temp目录路径
temp_dir = Path(__file__).parents[3] / "data" / "temp"
if not temp_dir.exists():
logger.warning(f"[Admin] 缓存目录不存在: {temp_dir}")
return {
"success": True,
"data": {"size": "0 MB"}
}
# 计算目录大小
total_size = 0
for file_path in temp_dir.iterdir():
if file_path.is_file():
try:
total_size += file_path.stat().st_size
except Exception as e:
logger.error(f"[Admin] 获取文件大小失败: {file_path.name}, 错误: {str(e)}")
# 转换为MB并格式化
size_mb = total_size / (1024 * 1024)
if size_mb < 1:
size_kb = total_size / 1024
size_str = f"{size_kb:.1f} KB"
else:
size_str = f"{size_mb:.1f} MB"
logger.debug(f"[Admin] 缓存大小获取完成 - 大小: {size_str}")
return {
"success": True,
"data": {"size": size_str}
}
except Exception as e:
logger.error(f"[Admin] 获取缓存大小异常 - 错误: {str(e)}")
raise HTTPException(
status_code=500,
detail={"error": f"获取缓存大小失败: {str(e)}", "code": "CACHE_SIZE_ERROR"}
)
@router.post("/api/cache/clear")
async def clear_cache(authenticated: bool = Depends(verify_admin_session)) -> Dict[str, Any]:
"""
清理缓存
删除/data/temp目录中的所有文件。
"""
try:
logger.debug("[Admin] 开始清理缓存")
# 获取temp目录路径
temp_dir = Path(__file__).parents[3] / "data" / "temp"
if not temp_dir.exists():
logger.warning(f"[Admin] 缓存目录不存在: {temp_dir}")
return {
"success": True,
"message": "缓存目录不存在,无需清理",
"data": {"deleted_count": 0}
}
# 获取所有文件
deleted_count = 0
for file_path in temp_dir.iterdir():
if file_path.is_file():
try:
file_path.unlink()
deleted_count += 1
logger.debug(f"[Admin] 删除缓存文件: {file_path.name}")
except Exception as e:
logger.error(f"[Admin] 删除缓存文件失败: {file_path.name}, 错误: {str(e)}")
logger.debug(f"[Admin] 缓存清理完成 - 删除文件数量: {deleted_count}")
return {
"success": True,
"message": f"成功清理缓存,删除 {deleted_count} 个文件",
"data": {"deleted_count": deleted_count}
}
except Exception as e:
logger.error(f"[Admin] 清理缓存异常 - 错误: {str(e)}")
raise HTTPException(
status_code=500,
detail={"error": f"清理缓存失败: {str(e)}", "code": "CACHE_CLEAR_ERROR"}
)
@router.get("/api/stats")
async def get_stats(authenticated: bool = Depends(verify_admin_session)) -> Dict[str, Any]:
"""
获取统计信息
返回Token的统计数据。
"""
try:
logger.debug("[Admin] 开始获取统计信息")
all_tokens_data = token_manager.get_all_token()
# 统计普通Token
normal_tokens = all_tokens_data.get(TokenType.NORMAL.value, {})
normal_stats = calculate_token_stats(normal_tokens, "normal")
# 统计Super Token
super_tokens = all_tokens_data.get(TokenType.SUPER.value, {})
super_stats = calculate_token_stats(super_tokens, "super")
total_count = normal_stats["total"] + super_stats["total"]
stats = {
"success": True,
"data": {
"normal": normal_stats,
"super": super_stats,
"total": total_count
}
}
logger.debug(f"[Admin] 统计信息获取成功 - 普通Token: {normal_stats['total']}, Super Token: {super_stats['total']}, 总计: {total_count}")
return stats
except Exception as e:
logger.error(f"[Admin] 获取统计信息异常 - 错误: {str(e)}")
raise HTTPException(
status_code=500,
detail={"error": f"获取统计信息失败: {str(e)}", "code": "STATS_ERROR"}
)
+87
View File
@@ -0,0 +1,87 @@
# -*- coding: utf-8 -*-
"""
聊天API路由模块
提供OpenAI兼容的聊天API接口,支持与Grok模型的交互。
"""
from fastapi import APIRouter, Depends, HTTPException
from typing import Optional
from fastapi.responses import StreamingResponse
from app.core.auth import auth_manager
from app.core.exception import GrokApiException
from app.core.logger import logger
from app.services.grok.client import GrokClient
from app.models.openai_schema import OpenAIChatRequest
# 聊天路由
router = APIRouter(prefix="/chat", tags=["聊天"])
@router.post("/completions", response_model=None)
async def chat_completions(
request: OpenAIChatRequest,
authenticated: Optional[str] = Depends(auth_manager.verify)
):
"""
创建聊天补全
兼容OpenAI聊天API的端点,支持流式和非流式响应。
Args:
request: OpenAI格式的聊天请求
authenticated: 认证状态(由依赖注入)
Returns:
OpenAIChatCompletionResponse: 非流式响应
StreamingResponse: 流式响应
Raises:
HTTPException: 当请求处理失败时
"""
try:
logger.info(f"[Chat] 聊天请求 - 模型: {request.model}")
# 调用Grok客户端处理请求
result = await GrokClient.openai_to_grok(request.model_dump())
# 如果是流式响应,GrokClient已经返回了Iterator,直接包装为StreamingResponse
if request.stream:
return StreamingResponse(
content=result,
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no"
}
)
# 非流式响应直接返回
return result
except GrokApiException as e:
logger.error(f"[Chat] Grok API错误: {str(e)}", extra={"details": e.details})
raise HTTPException(
status_code=500,
detail={
"error": {
"message": str(e),
"type": e.error_code or "grok_api_error",
"code": e.error_code or "unknown"
}
}
)
except Exception as e:
logger.error(f"[Chat] 聊天请求处理失败: {str(e)}", exc_info=True)
raise HTTPException(
status_code=500,
detail={
"error": {
"message": "服务器内部错误",
"type": "internal_error",
"code": "internal_server_error"
}
}
)
+49
View File
@@ -0,0 +1,49 @@
"""图片服务API路由"""
from fastapi import APIRouter, HTTPException
from fastapi.responses import FileResponse
from app.core.logger import logger
from app.services.grok.image_cache import image_cache_service
router = APIRouter()
@router.get("/images/{img_path:path}")
async def get_image(img_path: str):
"""获取缓存的图片
Args:
img_path: 图片路径,格式如 users-xxx-generated-xxx-image.jpg
Returns:
图片文件响应
"""
try:
# 将路径转换回原始格式(短横线转斜杠)
original_path = "/" + img_path.replace('-', '/')
# 检查缓存是否存在
cache_path = image_cache_service.get_cached_image(original_path)
if cache_path and cache_path.exists():
logger.debug(f"[ImageAPI] 返回缓存图片: {cache_path}")
return FileResponse(
path=str(cache_path),
media_type="image/jpeg",
headers={
"Cache-Control": "public, max-age=86400",
"Access-Control-Allow-Origin": "*"
}
)
# 图片不存在
logger.warning(f"[ImageAPI] 图片未找到: {original_path}")
raise HTTPException(status_code=404, detail="Image not found")
except HTTPException:
raise
except Exception as e:
logger.error(f"[ImageAPI] 获取图片失败: {e}")
raise HTTPException(status_code=500, detail=str(e))
+153
View File
@@ -0,0 +1,153 @@
"""
模型接口模块
提供 OpenAI 兼容的 /v1/models 端点,返回系统支持的所有模型列表。
"""
import time
from typing import Dict, Any, List, Optional
from fastapi import APIRouter, HTTPException, Depends
from app.models.grok_models import Models
from app.core.auth import auth_manager
from app.core.logger import logger
# 配置日志
# 创建路由器
router = APIRouter(tags=["模型"])
@router.get("/models")
async def list_models(authenticated: Optional[str] = Depends(auth_manager.verify)) -> Dict[str, Any]:
"""
获取可用模型列表
返回 OpenAI 兼容的模型列表格式,包含系统支持的所有 Grok 模型的详细信息。
Args:
authenticated: 认证状态(由依赖注入)
Returns:
Dict[str, Any]: 包含模型列表的响应数据
"""
try:
logger.debug("[Models] 请求获取模型列表")
# 获取当前时间戳
current_timestamp = int(time.time())
# 构建模型数据列表
model_data: List[Dict[str, Any]] = []
for model in Models:
model_id = model.value
config = Models.get_model_info(model_id)
# 基础信息
model_info = {
"id": model_id,
"object": "model",
"created": current_timestamp,
"owned_by": "x-ai",
"display_name": config.get("display_name", model_id),
"description": config.get("description", ""),
"raw_model_path": config.get("raw_model_path", f"xai/{model_id}"),
"default_temperature": config.get("default_temperature", 1.0),
"default_max_output_tokens": config.get("default_max_output_tokens", 8192),
"supported_max_output_tokens": config.get("supported_max_output_tokens", 131072),
"default_top_p": config.get("default_top_p", 0.95)
}
model_data.append(model_info)
# 构建响应
response = {
"object": "list",
"data": model_data
}
logger.debug(f"[Models] 成功返回 {len(model_data)} 个模型")
return response
except Exception as e:
logger.error(f"[Models] 获取模型列表时发生错误: {str(e)}")
raise HTTPException(
status_code=500,
detail={
"error": {
"message": f"Failed to retrieve models: {str(e)}",
"type": "internal_error",
"code": "model_list_error"
}
}
)
@router.get("/models/{model_id}")
async def get_model(model_id: str, authenticated: Optional[str] = Depends(auth_manager.verify)) -> Dict[str, Any]:
"""
获取特定模型信息
Args:
model_id (str): 模型ID
authenticated: 认证状态(由依赖注入)
Returns:
Dict[str, Any]: 模型详细信息
"""
try:
logger.debug(f"[Models] 请求获取模型信息: {model_id}")
# 验证模型是否存在
if not Models.is_valid_model(model_id):
logger.warning(f"[Models] 请求的模型不存在: {model_id}")
raise HTTPException(
status_code=404,
detail={
"error": {
"message": f"Model '{model_id}' not found",
"type": "invalid_request_error",
"code": "model_not_found"
}
}
)
# 获取当前时间戳
current_timestamp = int(time.time())
# 获取模型配置
config = Models.get_model_info(model_id)
# 构建模型信息
model_info = {
"id": model_id,
"object": "model",
"created": current_timestamp,
"owned_by": "x-ai",
"display_name": config.get("display_name", model_id),
"description": config.get("description", ""),
"raw_model_path": config.get("raw_model_path", f"xai/{model_id}"),
"default_temperature": config.get("default_temperature", 1.0),
"default_max_output_tokens": config.get("default_max_output_tokens", 8192),
"supported_max_output_tokens": config.get("supported_max_output_tokens", 131072),
"default_top_p": config.get("default_top_p", 0.95)
}
logger.debug(f"[Models] 成功返回模型信息: {model_id}")
return model_info
except HTTPException:
raise
except Exception as e:
logger.error(f"[Models] 获取模型信息时发生错误: {str(e)}")
raise HTTPException(
status_code=500,
detail={
"error": {
"message": f"Failed to retrieve model: {str(e)}",
"type": "internal_error",
"code": "model_retrieve_error"
}
}
)
+53
View File
@@ -0,0 +1,53 @@
"""认证模块"""
from typing import Optional
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from app.core.config import setting
from app.core.logger import logger
security = HTTPBearer(auto_error=False)
class AuthManager:
"""认证管理器"""
@staticmethod
def verify(credentials: Optional[HTTPAuthorizationCredentials] = Depends(security)) -> Optional[str]:
"""验证认证令牌"""
api_key = setting.grok_config.get("api_key")
if not api_key:
logger.debug("[Auth] 未设置API_KEY,跳过验证。")
return credentials.credentials if credentials else None
if not credentials:
raise HTTPException(
status_code=401,
detail={
"error": {
"message": "缺少认证令牌",
"type": "authentication_error",
"code": "missing_token"
}
}
)
if credentials.credentials != api_key:
raise HTTPException(
status_code=401,
detail={
"error": {
"message": f"令牌无效,长度: {len(credentials.credentials)}",
"type": "authentication_error",
"code": "invalid_token"
}
}
)
logger.debug("[Auth] 令牌认证成功。")
return credentials.credentials
auth_manager = AuthManager()
+32
View File
@@ -0,0 +1,32 @@
"""配置管理器"""
import toml
from pathlib import Path
from typing import Dict, Any
class ConfigManager:
"""配置管理器"""
def __init__(self) -> None:
"""初始化"""
# 加载环境变量
self.config_path: Path = Path(__file__).parents[2] / "data" / "setting.toml"
self.global_config: Dict[str, Any] = self.load("global")
self.grok_config: Dict[str, Any] = self.load("grok")
def load(self, section: str) -> Dict[str, Any]:
"""配置加载器"""
try:
with open(self.config_path, "r", encoding="utf-8") as f:
return toml.load(f)[section]
except Exception as e:
raise Exception(f"[Setting] 配置加载失败: {e}")
# 全局设置
setting = ConfigManager()
if __name__ == "__main__":
print(setting.global_config)
print(setting.grok_config)
+120
View File
@@ -0,0 +1,120 @@
"""异常处理器"""
from fastapi import Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
class GrokApiException(Exception):
"""Grok API 业务异常"""
def __init__(self, message: str, error_code: str = None, details: dict = None):
self.message = message
self.error_code = error_code
self.details = details or {}
super().__init__(self.message)
def build_error_response(message: str, error_type: str, code: str = None, param: str = None) -> dict:
"""构建OpenAI兼容的错误响应"""
error = {
"message": message,
"type": error_type,
}
if code:
error["code"] = code
if param:
error["param"] = param
return {"error": error}
async def http_exception_handler(_: Request, exc: StarletteHTTPException) -> JSONResponse:
"""处理HTTP异常"""
error_map = {
400: ("invalid_request_error", "请求格式错误或缺少必填参数。"),
401: ("invalid_request_error", "令牌认证失败。"),
403: ("permission_error", "没有权限访问此资源。"),
404: ("invalid_request_error", "请求的资源不存在。"),
429: ("rate_limit_error", "请求频率超出限制,请稍后再试。"),
500: ("api_error", "内部服务器错误。"),
503: ("api_error", "服务暂时不可用。"),
}
error_type, default_message = error_map.get(exc.status_code, ("api_error", str(exc.detail)))
message = str(exc.detail) if exc.detail else default_message
return JSONResponse(
status_code=exc.status_code,
content=build_error_response(message, error_type)
)
async def validation_exception_handler(_: Request, exc: RequestValidationError) -> JSONResponse:
"""处理验证错误"""
errors = exc.errors()
param = errors[0]["loc"][-1] if errors and errors[0].get("loc") else None
message = errors[0]["msg"] if errors and errors[0].get("msg") else "请求参数错误。"
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content=build_error_response(message, "invalid_request_error", param=param)
)
async def grok_api_exception_handler(_: Request, exc: GrokApiException) -> JSONResponse:
"""处理Grok API业务异常"""
# 根据错误码映射HTTP状态码
status_code_map = {
"NO_AUTH_TOKEN": status.HTTP_401_UNAUTHORIZED,
"INVALID_TOKEN": status.HTTP_401_UNAUTHORIZED,
"HTTP_ERROR": status.HTTP_502_BAD_GATEWAY,
"NETWORK_ERROR": status.HTTP_503_SERVICE_UNAVAILABLE,
"JSON_ERROR": status.HTTP_502_BAD_GATEWAY,
"API_ERROR": status.HTTP_502_BAD_GATEWAY,
"STREAM_ERROR": status.HTTP_502_BAD_GATEWAY,
"NO_RESPONSE": status.HTTP_502_BAD_GATEWAY,
"TOKEN_SAVE_ERROR": status.HTTP_500_INTERNAL_SERVER_ERROR,
"NO_AVAILABLE_TOKEN": status.HTTP_503_SERVICE_UNAVAILABLE,
}
http_status = status_code_map.get(exc.error_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
error_type_map = {
"NO_AUTH_TOKEN": "authentication_error",
"INVALID_TOKEN": "authentication_error",
"HTTP_ERROR": "api_error",
"NETWORK_ERROR": "api_error",
"JSON_ERROR": "api_error",
"API_ERROR": "api_error",
"STREAM_ERROR": "api_error",
"NO_RESPONSE": "api_error",
"TOKEN_SAVE_ERROR": "api_error",
"NO_AVAILABLE_TOKEN": "api_error",
}
error_type = error_type_map.get(exc.error_code, "api_error")
return JSONResponse(
status_code=http_status,
content=build_error_response(exc.message, error_type, exc.error_code)
)
async def global_exception_handler(_: Request) -> JSONResponse:
"""处理未捕获异常"""
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content=build_error_response(
"服务器遇到意外错误,请重试。",
"api_error"
)
)
def register_exception_handlers(app) -> None:
"""注册OpenAI兼容的异常处理器"""
app.add_exception_handler(StarletteHTTPException, http_exception_handler)
app.add_exception_handler(RequestValidationError, validation_exception_handler)
app.add_exception_handler(GrokApiException, grok_api_exception_handler)
app.add_exception_handler(Exception, global_exception_handler)
+74
View File
@@ -0,0 +1,74 @@
"""全局日志模块"""
import sys
import logging
from pathlib import Path
from logging.handlers import RotatingFileHandler
from app.core.config import setting
class LoggerManager:
"""日志管理器"""
_initialized = False
def __init__(self):
"""初始化日志"""
if LoggerManager._initialized:
return
# 日志配置
log_dir = Path(__file__).parents[2] / "logs"
log_dir.mkdir(exist_ok=True)
log_level = setting.global_config.get("log_level", "INFO").upper()
log_format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
log_file = log_dir / "app.log"
# 配置根日志器
self.logger = logging.getLogger()
self.logger.setLevel(log_level)
# 避免重复添加处理器
if self.logger.handlers:
return
# 控制台处理器
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(log_level)
console_handler.setFormatter(logging.Formatter(log_format))
# 文件处理器
file_handler = RotatingFileHandler(
log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8"
)
file_handler.setLevel(log_level)
file_handler.setFormatter(logging.Formatter(log_format))
self.logger.addHandler(console_handler)
self.logger.addHandler(file_handler)
LoggerManager._initialized = True
def debug(self, msg: str) -> None:
"""调试日志"""
self.logger.debug(msg)
def info(self, msg: str) -> None:
"""信息日志"""
self.logger.info(msg)
def warning(self, msg: str) -> None:
"""警告日志"""
self.logger.warning(msg)
def error(self, msg: str) -> None:
"""错误日志"""
self.logger.error(msg)
def critical(self, msg: str) -> None:
"""严重错误日志"""
self.logger.critical(msg)
# 全局日志器实例
logger = LoggerManager()
+108
View File
@@ -0,0 +1,108 @@
from enum import Enum
# 模型配置字典
_MODEL_CONFIG = {
"grok-3-fast": {
"grok_model": ("grok-3", "MODEL_MODE_FAST"),
"rate_limit_model": "grok-3",
"cost": {"type": "low_cost", "multiplier": 1, "description": "计1次调用"},
"requires_super": False,
"display_name": "Grok 3 Fast",
"description": "Fast and efficient Grok 3 model",
"raw_model_path": "xai/grok-3",
"default_temperature": 1.0,
"default_max_output_tokens": 8192,
"supported_max_output_tokens": 131072,
"default_top_p": 0.95
},
"grok-4-fast": {
"grok_model": ("grok-4-mini-thinking-tahoe", "MODEL_MODE_GROK_4_MINI_THINKING"),
"rate_limit_model": "grok-4-mini-thinking-tahoe",
"cost": {"type": "low_cost", "multiplier": 1, "description": "计1次调用"},
"requires_super": False,
"display_name": "Grok 4 Fast",
"description": "Fast version of Grok 4 with mini thinking capabilities",
"raw_model_path": "xai/grok-4-mini-thinking-tahoe",
"default_temperature": 1.0,
"default_max_output_tokens": 8192,
"supported_max_output_tokens": 131072,
"default_top_p": 0.95
},
"grok-4-fast-expert": {
"grok_model": ("grok-4-mini-thinking-tahoe", "MODEL_MODE_EXPERT"),
"rate_limit_model": "grok-4-mini-thinking-tahoe",
"cost": {"type": "high_cost", "multiplier": 4, "description": "计4次调用"},
"requires_super": False,
"display_name": "Grok 4 Fast Expert",
"description": "Expert mode of Grok 4 Fast with enhanced reasoning",
"raw_model_path": "xai/grok-4-mini-thinking-tahoe",
"default_temperature": 1.0,
"default_max_output_tokens": 32768,
"supported_max_output_tokens": 131072,
"default_top_p": 0.95
},
"grok-4-expert": {
"grok_model": ("grok-4", "MODEL_MODE_EXPERT"),
"rate_limit_model": "grok-4",
"cost": {"type": "high_cost", "multiplier": 4, "description": "计4次调用"},
"requires_super": False,
"display_name": "Grok 4 Expert",
"description": "Full Grok 4 model with expert mode capabilities",
"raw_model_path": "xai/grok-4",
"default_temperature": 1.0,
"default_max_output_tokens": 32768,
"supported_max_output_tokens": 131072,
"default_top_p": 0.95
},
"grok-4-heavy": {
"grok_model": ("grok-4-heavy", "MODEL_MODE_HEAVY"),
"rate_limit_model": "grok-4-heavy",
"cost": {"type": "independent", "multiplier": 1, "description": "独立计费,只有Super用户可用"},
"requires_super": True,
"display_name": "Grok 4 Heavy",
"description": "Most powerful Grok 4 model with heavy computational capabilities. Requires Super Token for access.",
"raw_model_path": "xai/grok-4-heavy",
"default_temperature": 1.0,
"default_max_output_tokens": 65536,
"supported_max_output_tokens": 131072,
"default_top_p": 0.95
}
}
class TokenType(Enum):
"""Token类型枚举"""
NORMAL = "ssoNormal" # 普通用户Token
SUPER = "ssoSuper" # 超级用户Token
class Models(Enum):
"""支持的模型枚举"""
GROK_3_FAST = "grok-3-fast"
GROK_4_FAST = "grok-4-fast"
GROK_4_FAST_EXPERT = "grok-4-fast-expert"
GROK_4_EXPERT = "grok-4-expert"
GROK_4_HEAVY = "grok-4-heavy"
@classmethod
def get_model_info(cls, model: str) -> dict:
"""获取模型的完整配置信息"""
return _MODEL_CONFIG.get(model, {})
@classmethod
def is_valid_model(cls, model: str) -> bool:
"""检查模型是否有效"""
return model in _MODEL_CONFIG
@classmethod
def to_grok(cls, model: str) -> tuple[str, str]:
"""转换为Grok内部模型名和模式类型"""
config = _MODEL_CONFIG.get(model)
if config:
return config["grok_model"]
return model, "MODEL_MODE_FAST"
@classmethod
def to_rate_limit(cls, model: str) -> str:
"""转换为速率限制接口模型名"""
config = _MODEL_CONFIG.get(model)
return config["rate_limit_model"] if config else model
+115
View File
@@ -0,0 +1,115 @@
"""OpenAI 请求-响应模型"""
from fastapi import HTTPException
from typing import Optional, List, Union, Dict, Any
from pydantic import BaseModel, Field, field_validator
class OpenAIChatRequest(BaseModel):
"""OpenAI 聊天请求模型"""
model: str = Field(..., description="模型名称", min_length=1)
messages: List[Dict[str, Any]] = Field(..., description="消息列表", min_length=1)
stream: bool = Field(False, description="启用流式响应")
temperature: Optional[float] = Field(0.7, ge=0, le=2, description="采样温度")
max_tokens: Optional[int] = Field(None, ge=1, le=100000, description="最大Token数")
top_p: Optional[float] = Field(1.0, ge=0, le=1, description="采样参数")
@field_validator('messages')
@classmethod
def validate_messages(cls, v):
"""验证消息格式"""
if not v:
raise HTTPException(
status_code=400,
detail="消息列表不能为空"
)
for msg in v:
if not isinstance(msg, dict):
raise HTTPException(
status_code=400,
detail="每个消息必须是一个字典"
)
if 'role' not in msg:
raise HTTPException(
status_code=400,
detail="消息缺少必填字段 'role'"
)
if 'content' not in msg:
raise HTTPException(
status_code=400,
detail="消息缺少必填字段 'content'"
)
if msg['role'] not in ['system', 'user', 'assistant']:
raise HTTPException(
status_code=400,
detail=f"无效的角色 '{msg['role']}', 必须是 'system', 'user' 或 'assistant'"
)
return v
@field_validator('model')
@classmethod
def validate_model(cls, v):
"""验证模型名称"""
allowed_models = [
'grok-3-fast', 'grok-4-fast', 'grok-4-fast-expert',
'grok-4-expert', 'grok-4-heavy'
]
if v not in allowed_models:
raise HTTPException(
status_code=400,
detail=f"不支持的模型 '{v}', 支持的模型: {', '.join(allowed_models)}"
)
return v
class OpenAIChatCompletionMessage(BaseModel):
"""聊天完成消息"""
role: str = Field(..., description="角色")
content: str = Field(..., description="消息内容")
reference_id: Optional[str] = Field(None, description="参考ID")
annotations: Optional[List[str]] = Field(None, description="注释")
class OpenAIChatCompletionChoice(BaseModel):
"""聊天完成选项"""
index: int = Field(..., description="选项索引")
message: OpenAIChatCompletionMessage = Field(..., description="响应消息")
logprobs: Optional[float] = Field(None, description="对数概率")
finish_reason: str = Field("stop", description="完成原因")
class OpenAIChatCompletionResponse(BaseModel):
"""聊天完成响应"""
id: str = Field(..., description="响应ID")
object: str = Field("chat.completion", description="对象类型")
created: int = Field(..., description="创建时间戳")
model: str = Field(..., description="使用的模型")
choices: List[OpenAIChatCompletionChoice] = Field(..., description="响应选项")
usage: Optional[Dict[str, Any]] = Field(None, description="令牌使用")
class OpenAIChatCompletionChunkMessage(BaseModel):
"""流式响应消息片段"""
role: str = Field(..., description="角色")
content: str = Field(..., description="消息内容")
class OpenAIChatCompletionChunkChoice(BaseModel):
"""流式响应选项"""
index: int = Field(..., description="选项索引")
delta: Optional[Union[Dict[str, Any], OpenAIChatCompletionChunkMessage]] = Field(
None, description="Delta数据"
)
finish_reason: Optional[str] = Field(None, description="完成原因")
class OpenAIChatCompletionChunkResponse(BaseModel):
"""流式聊天完成响应"""
id: str = Field(..., description="响应ID")
object: str = Field("chat.completion.chunk", description="对象类型")
created: int = Field(..., description="创建时间戳")
model: str = Field(..., description="使用的模型")
system_fingerprint: Optional[str] = Field(None, description="系统指纹")
choices: List[OpenAIChatCompletionChunkChoice] = Field(..., description="响应选项")
+224
View File
@@ -0,0 +1,224 @@
"""Grok API 客户端模块"""
import asyncio
import json
import threading
from typing import Dict, List, Tuple, Any
from curl_cffi import requests as curl_requests
from app.core.config import setting
from app.core.logger import logger
from app.models.grok_models import Models
from app.services.grok.processer import GrokResponseProcessor
from app.services.grok.statsig import get_dynamic_headers
from app.services.grok.token import token_manager
from app.services.grok.upload import ImageUploadManager
from app.core.exception import GrokApiException
# 常量定义
GROK_API_ENDPOINT = "https://grok.com/rest/app-chat/conversations/new"
REQUEST_TIMEOUT = 120
IMPERSONATE_BROWSER = "chrome133a"
class GrokClient:
"""Grok API 客户端"""
@staticmethod
async def openai_to_grok(openai_request: dict):
"""转换OpenAI请求为Grok请求并处理响应"""
model = openai_request["model"]
messages = openai_request["messages"]
stream = openai_request.get("stream", False)
logger.debug(f"[Client] 处理请求 - 模型:{model}, 消息数:{len(messages)}, 流式:{stream}")
# 提取消息内容和图片URL
content, image_urls = GrokClient._extract_message_content(messages)
# 获取认证令牌和模型信息
auth_token = token_manager.get_token(model)
model_name, model_mode = Models.to_grok(model)
# 上传图片并获取附件ID列表
image_attachments = GrokClient._upload_images(image_urls, auth_token)
# 构建Grok请求载荷
payload = GrokClient._build_payload(content, model_name, model_mode, image_attachments)
return await GrokClient._send_request(payload, auth_token, model, stream)
@staticmethod
def _extract_message_content(messages: List[Dict]) -> Tuple[str, List[str]]:
"""提取消息内容和图片URL"""
content_parts = []
image_urls = []
for msg in messages:
msg_content = msg.get("content", "")
# 处理复杂消息格式(包含文本和图片)
if isinstance(msg_content, list):
for item in msg_content:
item_type = item.get("type")
if item_type == "text":
content_parts.append(item.get("text", ""))
elif item_type == "image_url":
url = item.get("image_url", {}).get("url", "")
if url:
image_urls.append(url)
# 处理纯文本消息
else:
content_parts.append(msg_content)
return "".join(content_parts), image_urls
@staticmethod
def _upload_images(image_urls: List[str], auth_token: str) -> List[str]:
"""上传图片并返回附件ID列表"""
image_attachments = []
for url in image_urls:
try:
image_id = ImageUploadManager.upload(url, auth_token)
if image_id:
image_attachments.append(image_id)
except Exception as e:
logger.warning(f"[Client] 图片上传失败: {url}, 错误: {e}")
return image_attachments
@staticmethod
def _build_payload(content: str, model_name: str, model_mode: str, image_attachments: List[str]) -> Dict[str, Any]:
"""构建Grok API请求载荷"""
return {
"temporary": setting.grok_config.get("temporary", True),
"modelName": model_name,
"message": content,
"fileAttachments": image_attachments,
"imageAttachments": [],
"disableSearch": False,
"enableImageGeneration": True,
"returnImageBytes": False,
"returnRawGrokInXaiRequest": False,
"enableImageStreaming": True,
"imageGenerationCount": 2,
"forceConcise": False,
"toolOverrides": {},
"enableSideBySide": True,
"sendFinalMetadata": True,
"isReasoning": False,
"webpageUrls": [],
"disableTextFollowUps": True,
"responseMetadata": {"requestModelDetails": {"modelId": model_name}},
"disableMemory": False,
"forceSideBySide": False,
"modelMode": model_mode,
"isAsyncChat": False
}
@staticmethod
async def _send_request(payload: dict, auth_token: str, model: str, stream: bool):
"""发送HTTP请求到Grok API"""
# 验证认证令牌
if not auth_token:
raise GrokApiException("认证令牌缺失", "NO_AUTH_TOKEN")
try:
# 准备请求头和Cookie
headers = GrokClient._prepare_headers(auth_token)
# 准备代理配置
proxies = GrokClient._get_proxy_config()
# 在线程池中执行同步HTTP请求,避免阻塞事件循环
response = await asyncio.to_thread(
curl_requests.post,
GROK_API_ENDPOINT,
headers=headers,
data=json.dumps(payload),
impersonate=IMPERSONATE_BROWSER,
timeout=REQUEST_TIMEOUT,
stream=True,
**proxies
)
logger.debug(f"[Client] API响应状态码: {response.status_code}")
# 处理非成功响应
if response.status_code != 200:
GrokClient._handle_error_response(response, auth_token)
# 请求成功,重置失败计数
token_manager.reset_token_failure(auth_token)
# 处理并返回响应
return GrokClient._process_response(response, auth_token, model, stream)
except curl_requests.RequestsError as e:
raise GrokApiException(f"网络错误: {e}", "NETWORK_ERROR") from e
except json.JSONDecodeError as e:
raise GrokApiException(f"JSON解析错误: {e}", "JSON_ERROR") from e
@staticmethod
def _prepare_headers(auth_token: str) -> Dict[str, str]:
"""准备请求头"""
headers = get_dynamic_headers("/rest/app-chat/conversations/new")
# 构建Cookie
cf_clearance = setting.grok_config.get("cf_clearance", "")
headers["Cookie"] = f"{auth_token};{cf_clearance}" if cf_clearance else auth_token
return headers
@staticmethod
def _get_proxy_config() -> Dict[str, str]:
"""获取代理配置"""
proxy_url = setting.grok_config.get("proxy_url", "")
if proxy_url:
return {"http": proxy_url, "https": proxy_url}
return {}
@staticmethod
def _handle_error_response(response, auth_token: str):
"""处理错误响应"""
try:
error_data = response.json()
error_message = str(error_data)
except Exception:
error_data = response.text
error_message = error_data[:200] if error_data else "未知错误"
# 记录Token失败
token_manager.record_token_failure(auth_token, response.status_code, error_message)
raise GrokApiException(
f"请求失败: {response.status_code} - {error_message}",
"HTTP_ERROR",
{"status": response.status_code, "data": error_data}
)
@staticmethod
def _process_response(response, auth_token: str, model: str, stream: bool):
"""处理API响应"""
if stream:
# 流式响应:异步更新速率限制
result = GrokResponseProcessor.process_stream(response, auth_token)
threading.Thread(
target=lambda: GrokClient._update_rate_limits(auth_token, model),
daemon=True
).start()
else:
# 非流式响应:同步更新速率限制
result = GrokResponseProcessor.process_response(response, auth_token)
GrokClient._update_rate_limits(auth_token, model)
return result
@staticmethod
def _update_rate_limits(auth_token: str, model: str):
"""更新速率限制信息"""
try:
token_manager.check_limits(auth_token, model)
except Exception as e:
logger.error(f"[Client] 更新速率限制失败: {e}")
+151
View File
@@ -0,0 +1,151 @@
"""图片缓存服务模块"""
import asyncio
from pathlib import Path
from typing import Optional
from curl_cffi.requests import AsyncSession
from app.core.config import setting
from app.core.logger import logger
from app.services.grok.statsig import get_dynamic_headers
class ImageCacheService:
"""图片缓存服务"""
def __init__(self):
self.cache_dir = Path("data/temp")
self.cache_dir.mkdir(parents=True, exist_ok=True)
@staticmethod
def _get_cache_filename(image_path: str) -> str:
"""将图片路径转换为缓存文件名"""
# 移除开头的斜杠并替换所有斜杠为短横线
filename = image_path.lstrip('/').replace('/', '-')
return filename
def _get_cache_path(self, image_path: str) -> Path:
"""获取缓存文件的完整路径"""
filename = self._get_cache_filename(image_path)
return self.cache_dir / filename
async def download_image(self, image_path: str, auth_token: str) -> Optional[Path]:
"""下载并缓存图片
Args:
image_path: 图片路径,如 /users/xxx/generated/xxx/image.jpg
auth_token: 认证令牌
Returns:
缓存文件路径,下载失败返回 None
"""
cache_path = self._get_cache_path(image_path)
if cache_path.exists():
logger.debug(f"[ImageCache] 图片已缓存: {cache_path}")
return cache_path
image_url = f"https://assets.grok.com{image_path}"
try:
# 构建 Cookie
cf_clearance = setting.grok_config.get("cf_clearance", "")
cookie = f"{auth_token};{cf_clearance}" if cf_clearance else auth_token
# 构建请求头
headers = {
**get_dynamic_headers(pathname=image_path),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
"Sec-Fetch-Dest": "document",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-Site": "same-site",
"Sec-Fetch-User": "?1",
"Upgrade-Insecure-Requests": "1",
"Referer": "https://grok.com/",
"Cookie": cookie
}
# 代理配置
proxy_url = setting.grok_config.get("proxy_url")
proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None
async with AsyncSession() as session:
logger.debug(f"[ImageCache] 开始下载图片: {image_url}")
response = await session.get(
image_url,
headers=headers,
proxies=proxies,
timeout=30.0,
allow_redirects=True,
impersonate="chrome133a"
)
response.raise_for_status()
cache_path.write_bytes(response.content)
logger.debug(f"[ImageCache] 图片已缓存: {cache_path} ({len(response.content)} bytes)")
asyncio.create_task(self.cleanup_cache())
return cache_path
except Exception as e:
logger.error(f"[ImageCache] 下载图片失败: {e}")
return None
def get_cached_image(self, image_path: str) -> Optional[Path]:
"""获取缓存的图片路径
Args:
image_path: 图片路径
Returns:
缓存文件路径,不存在返回 None
"""
cache_path = self._get_cache_path(image_path)
return cache_path if cache_path.exists() else None
async def cleanup_cache(self):
"""清理缓存目录,确保不超过配置的大小限制"""
try:
# 获取配置的最大缓存大小(MB)
max_size_mb = setting.global_config.get("temp_max_size_mb", 500)
max_size_bytes = max_size_mb * 1024 * 1024
# 获取所有缓存文件及其大小和修改时间
files = []
total_size = 0
for file_path in self.cache_dir.glob("*"):
if file_path.is_file():
size = file_path.stat().st_size
mtime = file_path.stat().st_mtime
files.append((file_path, size, mtime))
total_size += size
# 如果总大小未超限,无需清理
if total_size <= max_size_bytes:
logger.debug(f"[ImageCache] 缓存大小 {total_size / 1024 / 1024:.2f}MB,未超限")
return
logger.info(f"[ImageCache] 缓存大小 {total_size / 1024 / 1024:.2f}MB 超过限制 {max_size_mb}MB,开始清理")
# 按修改时间排序(最旧的在前)
files.sort(key=lambda x: x[2])
# 删除最旧的文件直到总大小低于限制
for file_path, size, _ in files:
if total_size <= max_size_bytes:
break
file_path.unlink()
total_size -= size
logger.debug(f"[ImageCache] 已删除缓存文件: {file_path}")
logger.info(f"[ImageCache] 缓存清理完成,当前大小 {total_size / 1024 / 1024:.2f}MB")
except Exception as e:
logger.error(f"[ImageCache] 清理缓存失败: {e}")
# 全局实例
image_cache_service = ImageCacheService()
+272
View File
@@ -0,0 +1,272 @@
"""Grok API 响应处理器模块"""
import json
import uuid
import time
import asyncio
from typing import Iterator
from app.core.config import setting
from app.core.exception import GrokApiException
from app.core.logger import logger
from app.models.openai_schema import (
OpenAIChatCompletionResponse,
OpenAIChatCompletionChoice,
OpenAIChatCompletionMessage,
OpenAIChatCompletionChunkResponse,
OpenAIChatCompletionChunkChoice,
OpenAIChatCompletionChunkMessage
)
from app.services.grok.image_cache import image_cache_service
def _safe_run_async(coro):
"""安全地运行异步协程,无论是否有事件循环"""
try:
loop = asyncio.get_event_loop()
if loop.is_running():
# 如果循环正在运行,使用 ensure_future 调度任务
return asyncio.ensure_future(coro)
else:
# 循环存在但未运行,直接运行
return loop.run_until_complete(coro)
except RuntimeError:
# 没有事件循环,创建新的
return asyncio.run(coro)
class GrokResponseProcessor:
"""Grok API 响应处理器"""
@staticmethod
def process_response(response, auth_token: str) -> OpenAIChatCompletionResponse:
"""处理非流式响应"""
try:
for chunk in response.iter_lines():
if not chunk:
continue
data = json.loads(chunk.decode("utf-8"))
# 错误检查
if error := data.get("error"):
raise GrokApiException(
f"API错误: {error.get('message', '未知错误')}",
"API_ERROR",
{"code": error.get("code")}
)
# 提取模型响应
model_response = data.get("result", {}).get("response", {}).get("modelResponse")
if not model_response:
continue
# 检查 modelResponse 中的错误
if error_msg := model_response.get("error"):
raise GrokApiException(
f"模型响应错误: {error_msg}",
"MODEL_ERROR"
)
# 构建响应内容
model = model_response.get("model")
content = model_response.get("message", "")
# 添加生成的图片
if images := model_response.get("generatedImageUrls"):
for img in images:
# 下载并缓存图片
try:
cache_path = _safe_run_async(image_cache_service.download_image(f"/{img}", auth_token))
if cache_path:
# 使用本地缓存路径
img_path = img.replace('/', '-')
base_url = setting.global_config.get("base_url", "")
img_url = f"{base_url}/images/{img_path}" if base_url else f"/images/{img_path}"
content += f"\n![Generated Image]({img_url})"
else:
# 缓存失败,使用原始链接
content += f"\n![Generated Image](https://assets.grok.com/{img})"
except Exception as e:
logger.warning(f"[Processor] 缓存图片失败: {e}")
content += f"\n![Generated Image](https://assets.grok.com/{img})"
# 返回OpenAI格式响应
return OpenAIChatCompletionResponse(
id=f"chatcmpl-{uuid.uuid4()}",
object="chat.completion",
created=int(time.time()),
model=model,
choices=[OpenAIChatCompletionChoice(
index=0,
message=OpenAIChatCompletionMessage(
role="assistant",
content=content
),
finish_reason="stop"
)],
usage=None
)
raise GrokApiException("无响应数据", "NO_RESPONSE")
except json.JSONDecodeError as e:
raise GrokApiException(f"JSON解析失败: {e}", "JSON_ERROR") from e
@staticmethod
def process_stream(response, auth_token: str) -> Iterator[str]:
"""处理流式响应"""
is_image = False
is_thinking = False
thinking_finished = False
chunk_index = 0
model = None
filtered_tags = setting.grok_config.get("filtered_tags", "").split(",")
def make_chunk(content: str, finish: str = None):
"""生成OpenAI格式的响应块"""
chunk_data = OpenAIChatCompletionChunkResponse(
id=f"chatcmpl-{uuid.uuid4()}",
created=int(time.time()),
model=model or "grok-4-mini-thinking-tahoe",
choices=[OpenAIChatCompletionChunkChoice(
index=chunk_index,
delta=OpenAIChatCompletionChunkMessage(
role="assistant",
content=content
) if content else {},
finish_reason=finish
)]
).model_dump()
# SSE 格式返回
return f"data: {json.dumps(chunk_data)}\n\n"
try:
for chunk in response.iter_lines():
logger.debug(f"[Processor] 接收到数据块: {len(chunk)} bytes")
if not chunk:
continue
try:
data = json.loads(chunk.decode("utf-8"))
# 错误检查
if error := data.get("error"):
error_msg = error.get('message', '未知错误')
logger.error(f"[Processor] Grok API返回错误: {error_msg}")
yield make_chunk(f"Error: {error_msg}", "stop")
yield "data: [DONE]\n\n"
return
# 提取响应数据
grok_resp = data.get("result", {}).get("response", {})
logger.debug(f"[Processor] 解析响应数据: {len(grok_resp)} 字段")
if not grok_resp:
continue
# 更新模型名称
if user_resp := grok_resp.get("userResponse"):
if m := user_resp.get("model"):
model = m
# 检查生成模式
if grok_resp.get("imageAttachmentInfo"):
is_image = True
# 获取token
token = grok_resp.get("token", "")
# 图片模式
if is_image:
if model_resp := grok_resp.get("modelResponse"):
# 生成图片链接并缓存
content = ""
for img in model_resp.get("generatedImageUrls", []):
try:
# 异步下载并缓存图片(不阻塞)
_safe_run_async(image_cache_service.download_image(f"/{img}", auth_token))
# 使用本地缓存路径
img_path = img.replace('/', '-')
base_url = setting.global_config.get("base_url", "")
img_url = f"{base_url}/images/{img_path}" if base_url else f"/images/{img_path}"
content += f"![Generated Image]({img_url})\n"
except Exception as e:
logger.warning(f"[Processor] 缓存图片失败: {e}")
content += f"![Generated Image](https://assets.grok.com/{img})\n"
yield make_chunk(content.strip(), "stop")
return
elif token:
yield make_chunk(token)
chunk_index += 1
# 对话模式
else:
# 过滤 list 格式的 token
if isinstance(token, list):
continue
# 过滤特定标签
if any(tag in token for tag in filtered_tags if token):
continue
# 获取当前状态
current_is_thinking = grok_resp.get("isThinking", False)
message_tag = grok_resp.get("messageTag")
# 跳过后续的 thinking
if thinking_finished and current_is_thinking:
continue
# 检查 toolUsageCardId
if grok_resp.get("toolUsageCardId"):
if web_search := grok_resp.get("webSearchResults"):
if current_is_thinking:
# 添加搜索结果到 token
for result in web_search.get("results", []):
title = result.get("title", "")
url = result.get("url", "")
preview = result.get("preview", "")
preview_clean = preview.replace("\n", "") if isinstance(preview, str) else ""
token += f'\n- [{title}]({url} "{preview_clean}")'
token += "\n"
else:
# 有 webSearchResults 但 isThinking 为 false
continue
else:
# 没有 webSearchResults
continue
if token:
content = token
# header 在 token 后换行
if message_tag == "header":
content = f"\n\n{token}\n\n"
# is_thinking 状态切换
if not is_thinking and current_is_thinking:
content = f"<think>\n{content}"
elif is_thinking and not current_is_thinking:
content = f"\n</think>\n{content}"
thinking_finished = True
yield make_chunk(content)
chunk_index += 1
is_thinking = current_is_thinking
except (json.JSONDecodeError, UnicodeDecodeError) as e:
logger.warning(f"[Processor] 解析chunk失败: {e}")
continue
except Exception as e:
logger.warning(f"[Processor] 处理chunk出错: {e}")
continue
# 发送结束块
yield make_chunk("", "stop")
# 发送流结束标记
yield "data: [DONE]\n\n"
except Exception as e:
logger.error(f"[Processor] 流式处理严重错误: {e}")
yield make_chunk(f"处理错误: {e}", "error")
# 发送流结束标记
yield "data: [DONE]\n\n"
+44
View File
@@ -0,0 +1,44 @@
"""Grok 请求头管理模块"""
import uuid
from typing import Dict
from app.core.config import setting
def get_dynamic_headers(pathname: str = "/rest/app-chat/conversations/new") -> Dict[str, str]:
"""获取请求头
Args:
pathname: 请求路径
Returns:
请求头字典
"""
# 获取配置的 x-statsig-id
statsig_id = setting.grok_config.get("x_statsig_id")
if not statsig_id:
raise ValueError("配置文件中未设置 x_statsig_id")
# 构建基础请求头
headers = {
"Accept": "*/*",
"Accept-Language": "zh-CN,zh;q=0.9",
"Accept-Encoding": "gzip, deflate, br, zstd",
"Content-Type": "application/json" if "upload-file" not in pathname else "text/plain;charset=UTF-8",
"Connection": "keep-alive",
"Origin": "https://grok.com",
"Priority": "u=1, i",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36",
"Sec-Ch-Ua": '"Not(A:Brand";v="99", "Google Chrome";v="133", "Chromium";v="133"',
"Sec-Ch-Ua-Mobile": "?0",
"Sec-Ch-Ua-Platform": '"macOS"',
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"Baggage": "sentry-public_key=b311e0f2690c81f25e2c4cf6d4f7ce1c",
"x-statsig-id": statsig_id,
"x-xai-request-id": str(uuid.uuid4())
}
return headers
+438
View File
@@ -0,0 +1,438 @@
"""Grok Token 管理器模块"""
import json
import time
import threading
from pathlib import Path
from typing import Dict, Any, Optional, Tuple
from curl_cffi import requests as curl_requests
from app.models.grok_models import TokenType, Models
from app.core.exception import GrokApiException
from app.core.logger import logger
from app.core.config import setting
from app.services.grok.statsig import get_dynamic_headers
# 常量定义
RATE_LIMIT_ENDPOINT = "https://grok.com/rest/rate-limits"
REQUEST_TIMEOUT = 30
IMPERSONATE_BROWSER = "chrome133a"
MAX_FAILURE_COUNT = 3
TOKEN_INVALID_CODE = 401 # SSO Token失效
STATSIG_INVALID_CODE = 403 # x-statsig-id失效
class GrokTokenManager:
"""
Grok Token管理器
单例模式的Token管理器,负责:
- Token文件的读写操作
- Token负载均衡
- Token状态管理
- 支持普通Token和Super Token
"""
_instance: Optional['GrokTokenManager'] = None
_lock = threading.Lock()
def __new__(cls) -> 'GrokTokenManager':
"""单例模式实现"""
if cls._instance is None:
with cls._lock:
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
"""初始化Token管理器"""
if hasattr(self, '_initialized'):
return
self.token_file = Path(__file__).parents[3] / "data" / "token.json"
self._file_lock = threading.Lock()
self.token_file.parent.mkdir(parents=True, exist_ok=True)
self._load_token_data()
self._initialized = True
logger.debug(f"[Token] 管理器初始化完成,文件: {self.token_file}")
def _load_token_data(self) -> None:
"""加载Token数据"""
default_data = {
TokenType.NORMAL.value: {},
TokenType.SUPER.value: {}
}
try:
if self.token_file.exists():
with open(self.token_file, "r", encoding="utf-8") as f:
self.token_data = json.load(f)
else:
self.token_data = default_data
self._save_token_data()
logger.debug("[Token] 创建新的Token数据文件")
except (json.JSONDecodeError, IOError) as e:
logger.error(f"[Token] 加载Token数据失败: {str(e)}")
self.token_data = default_data
def _save_token_data(self) -> None:
"""保存Token数据到文件"""
try:
with self._file_lock:
with open(self.token_file, "w", encoding="utf-8") as f:
json.dump(self.token_data, f, indent=2, ensure_ascii=False)
except IOError as e:
logger.error(f"[Token] 保存Token数据失败: {str(e)}")
raise GrokApiException(
f"Token数据保存失败: {str(e)}",
"TOKEN_SAVE_ERROR",
{"file_path": str(self.token_file)}
)
@staticmethod
def _extract_sso_value(auth_token: str) -> Optional[str]:
"""从认证令牌中提取SSO值"""
if "sso=" in auth_token:
return auth_token.split("sso=")[1].split(";")[0]
logger.warning("[Token] 无法从认证令牌中提取SSO值")
return None
def _find_token_data(self, sso_value: str) -> Tuple[Optional[str], Optional[Dict[str, Any]]]:
"""查找Token数据,返回(token_type, token_data)"""
for token_type in [TokenType.NORMAL.value, TokenType.SUPER.value]:
if sso_value in self.token_data[token_type]:
return token_type, self.token_data[token_type][sso_value]
return None, None
@staticmethod
def _get_proxy_config() -> Dict[str, str]:
"""获取代理配置"""
proxy_url = setting.grok_config.get("proxy_url", "")
if proxy_url:
return {"http": proxy_url, "https": proxy_url}
return {}
def add_token(self, tokens: list[str], token_type: TokenType) -> None:
"""
添加Token到管理器
Args:
tokens: Token列表
token_type: Token类型
"""
if not tokens:
logger.debug("[Token] 尝试添加空的Token列表")
return
added_count = 0
for token in tokens:
if not token or not token.strip():
logger.debug("[Token] 跳过空的Token")
continue
self.token_data[token_type.value][token] = {
"createdTime": int(time.time() * 1000),
"remainingQueries": -1,
"heavyremainingQueries": -1,
"status": "active",
"failedCount": 0,
"lastFailureTime": None,
"lastFailureReason": None
}
added_count += 1
self._save_token_data()
logger.info(f"[Token] 成功添加 {added_count} 个 {token_type.value} Token")
def delete_token(self, tokens: list[str], token_type: TokenType) -> None:
"""
删除指定的Token
Args:
tokens: 要删除的Token列表
token_type: Token类型
"""
if not tokens:
logger.debug("[Token] 尝试删除空的Token列表")
return
deleted_count = 0
for token in tokens:
if token in self.token_data[token_type.value]:
del self.token_data[token_type.value][token]
deleted_count += 1
else:
logger.debug(f"[Token] Token不存在: {token[:10]}...")
self._save_token_data()
logger.info(f"[Token] 成功删除 {deleted_count} 个 {token_type.value} Token")
def get_all_token(self) -> Dict[str, Any]:
"""获取所有Token数据"""
return self.token_data.copy()
def get_token(self, model: str) -> str:
"""
获取指定模型的Token
Args:
model: 模型名称
Returns:
str: 格式化的Cookie字符串
"""
jwt_token = self.token_balancer(model)
return f"sso-rw={jwt_token};sso={jwt_token}"
def token_balancer(self, model: str) -> str:
"""
Token负载均衡器 - 根据模型类型和剩余次数选择最优Token
选择策略:
1. 跳过 status=expired 的 token
2. 跳过 remaining=0 的 token(已限流)
3. 优先选择 remaining=-1 的 token(未使用)
4. 如果没有 -1,选择 remaining 最大的 token(剩余最多)
"""
def select_best_token(tokens_dict):
"""从 token 字典中选择最佳 token"""
unused_tokens = [] # remaining = -1 的 token
used_tokens = [] # remaining > 0 的 token
for token_key, token_data in tokens_dict.items():
# 跳过已失效的Token
if token_data.get("status") == "expired":
continue
remaining = int(token_data.get(remaining_field, -1))
# 跳过已限流的 token (remaining = 0)
if remaining == 0:
continue
# 分类存储
if remaining == -1:
unused_tokens.append(token_key)
elif remaining > 0:
used_tokens.append((token_key, remaining))
# 优先返回未使用的 token
if unused_tokens:
return unused_tokens[0], -1
# 否则返回剩余次数最多的 token
if used_tokens:
used_tokens.sort(key=lambda x: x[1], reverse=True)
return used_tokens[0][0], used_tokens[0][1]
return None, None
max_token_key = None
max_remaining = None
if model == "grok-4-heavy":
# grok-4-heavy 只能使用Super Token,且使用 heavyremainingQueries
remaining_field = "heavyremainingQueries"
max_token_key, max_remaining = select_best_token(self.token_data[TokenType.SUPER.value])
else:
# 其他模型使用 remainingQueries(对应API的remainingTokens)
remaining_field = "remainingQueries"
# 优先使用普通Token
max_token_key, max_remaining = select_best_token(self.token_data[TokenType.NORMAL.value])
# 如果普通Token没有可用的,尝试使用Super Token
if max_token_key is None:
max_token_key, max_remaining = select_best_token(self.token_data[TokenType.SUPER.value])
if max_token_key is None:
raise GrokApiException(
f"没有可用Token用于模型 {model}",
"NO_AVAILABLE_TOKEN",
{
"model": model,
"normal_count": len(self.token_data[TokenType.NORMAL.value]),
"super_count": len(self.token_data[TokenType.SUPER.value])
}
)
status_text = "未使用" if max_remaining == -1 else f"剩余{max_remaining}次"
logger.debug(f"[Token] 为模型 {model} 选择Token ({status_text})")
return max_token_key
def check_limits(self, auth_token: str, model: str) -> Optional[Dict[str, Any]]:
"""检查并更新模型速率限制"""
try:
rate_limit_model_name = Models.to_rate_limit(model)
logger.debug(f"[Token] 检查模型 {model} (接口模型: {rate_limit_model_name}) 的速率限制")
# 准备请求
payload = {"requestKind": "DEFAULT", "modelName": rate_limit_model_name}
cf_clearance = setting.grok_config.get("cf_clearance", "")
cookie = f"{auth_token};{cf_clearance}" if cf_clearance else auth_token
headers = get_dynamic_headers("/rest/rate-limits")
headers["Cookie"] = cookie
# 发送请求
response = curl_requests.post(
RATE_LIMIT_ENDPOINT,
headers=headers,
json=payload,
impersonate=IMPERSONATE_BROWSER,
timeout=REQUEST_TIMEOUT,
**self._get_proxy_config()
)
if response.status_code == 200:
rate_limit_data = response.json()
logger.debug(f"[Token] 成功获取速率限制信息")
self._save_to_storage(auth_token, model, rate_limit_data)
return rate_limit_data
else:
logger.warning(f"[Token] 获取速率限制失败,状态码: {response.status_code}")
return None
except Exception as e:
logger.error(f"[Token] 检查速率限制时发生错误: {str(e)}")
return None
@staticmethod
def get_remaining(data: Dict[str, Any], model: str) -> int:
"""获取剩余Token数量"""
try:
if model == "grok-4-heavy":
return data.get("remainingQueries", 0)
else:
return data.get("remainingTokens", 0)
except Exception as e:
logger.error(f"[Token] 解析剩余Token数量时发生错误: {str(e)}")
return 0
def _save_to_storage(self, auth_token: str, model: str, data: Dict[str, Any]) -> None:
"""存储速率限制信息到token.json"""
try:
sso_value = self._extract_sso_value(auth_token)
if not sso_value:
return
# 根据模型类型获取剩余数量
if model == "grok-4-heavy":
normal_remaining = -1
heavy_remaining = data.get("remainingQueries", -1)
else:
normal_remaining = data.get("remainingTokens", -1)
heavy_remaining = -1
self.update_limits(sso_value, normal_remaining, heavy_remaining)
logger.info(f"[Token] 已更新限制: sso={sso_value[:10]}..., 通用={normal_remaining}, heavy={heavy_remaining}")
except Exception as e:
logger.error(f"[Token] 存储速率限制信息时发生错误: {str(e)}")
def update_limits(self, sso_value: str, normal: int = None, heavy: int = None) -> None:
"""更新Token限制信息"""
try:
for token_type in [TokenType.NORMAL.value, TokenType.SUPER.value]:
if sso_value in self.token_data[token_type]:
if normal is not None:
self.token_data[token_type][sso_value]["remainingQueries"] = normal
if heavy is not None:
self.token_data[token_type][sso_value]["heavyremainingQueries"] = heavy
self._save_token_data()
logger.info(f"[Token] 已更新Token {sso_value[:10]}... 的限制信息")
return
logger.warning(f"[Token] 未找到SSO值为 {sso_value[:10]}... 的Token")
except Exception as e:
logger.error(f"[Token] 更新Token限制时发生错误: {str(e)}")
def record_token_failure(self, auth_token: str, status_code: int, error_message: str) -> None:
"""记录Token失败信息
错误码说明:
- 401: SSO Token失效,会标记Token为expired
- 403: x-statsig-id失效,不影响Token状态
Args:
auth_token: 完整的认证Token (格式: sso-rw=xxx;sso=xxx)
status_code: HTTP状态码
error_message: 错误信息
"""
try:
# 403错误是x-statsig-id失效,不是Token问题
if status_code == STATSIG_INVALID_CODE:
logger.warning(f"[Token] x-statsig-id失效 (403),需要更新配置文件中的x_statsig_id")
return
sso_value = self._extract_sso_value(auth_token)
if not sso_value:
return
_, token_data = self._find_token_data(sso_value)
if not token_data:
logger.warning(f"[Token] 未找到SSO值为 {sso_value[:10]}... 的Token")
return
# 更新失败计数
token_data["failedCount"] = token_data.get("failedCount", 0) + 1
token_data["lastFailureTime"] = int(time.time() * 1000)
token_data["lastFailureReason"] = f"{status_code}: {error_message}"
logger.warning(
f"[Token] Token {sso_value[:10]}... 失败 (状态码: {status_code}), "
f"失败次数: {token_data['failedCount']}/{MAX_FAILURE_COUNT}, "
f"原因: {error_message}"
)
# 只有401错误(SSO Token失效)且失败次数达到上限时,标记为失效
if status_code == TOKEN_INVALID_CODE and token_data["failedCount"] >= MAX_FAILURE_COUNT:
token_data["status"] = "expired"
logger.error(
f"[Token] SSO Token {sso_value[:10]}... 已被标记为失效 "
f"(连续401错误{token_data['failedCount']}次)"
)
self._save_token_data()
except Exception as e:
logger.error(f"[Token] 记录Token失败信息时发生错误: {str(e)}")
def reset_token_failure(self, auth_token: str) -> None:
"""重置Token失败计数
当Token成功完成请求时调用此方法,用于清除失败记录。
Args:
auth_token: 完整的认证Token (格式: sso-rw=xxx;sso=xxx)
"""
try:
sso_value = self._extract_sso_value(auth_token)
if not sso_value:
return
_, token_data = self._find_token_data(sso_value)
if not token_data:
logger.warning(f"[Token] 未找到SSO值为 {sso_value[:10]}... 的Token")
return
# 只有在有失败记录时才重置并保存
if token_data.get("failedCount", 0) > 0:
token_data["failedCount"] = 0
token_data["lastFailureTime"] = None
token_data["lastFailureReason"] = None
self._save_token_data()
logger.info(f"[Token] Token {sso_value[:10]}... 失败计数已重置")
except Exception as e:
logger.error(f"[Token] 重置Token失败计数时发生错误: {str(e)}")
# 全局Token管理器实例
token_manager = GrokTokenManager()
+134
View File
@@ -0,0 +1,134 @@
"""图片上传管理器"""
import base64
import re
from urllib.parse import urlparse
from curl_cffi import requests as curl_requests
from app.services.grok.statsig import get_dynamic_headers
from app.core.exception import GrokApiException
from app.core.config import setting
from app.core.logger import logger
class ImageUploadManager:
"""
Grok图片上传管理器
提供图片上传功能,支持:
- Base64格式图片上传
- URL图片下载并上传
- 多种图片格式支持
"""
@staticmethod
def upload(image_input: str, auth_token: str) -> str:
"""上传图片到Grok,支持Base64或URL"""
try:
if ImageUploadManager._is_url(image_input):
# 下载 URL 图片
image_buffer, mime_type = ImageUploadManager._download(image_input)
# 获取图片信息
file_name, _ = ImageUploadManager._get_info("", mime_type)
else:
# 处理 base64 数据
image_buffer = image_input.split(",")[1] if "data:image" in image_input else image_input
# 获取图片信息
file_name, mime_type = ImageUploadManager._get_info(image_input)
# 构建上传数据
upload_data = {
"fileName": file_name,
"fileMimeType": mime_type,
"content": image_buffer,
}
# 获取认证令牌
if not auth_token:
raise GrokApiException("认证令牌缺失或为空", "NO_AUTH_TOKEN")
cf_clearance = setting.grok_config.get("cf_clearance", "")
cookie = f"{auth_token};{cf_clearance}" if cf_clearance else auth_token
proxy_config = {"http": setting.grok_config.get("proxy_url", ""), "https": setting.grok_config.get("proxy_url", "")} if setting.grok_config.get("proxy_url") else {}
# 发送请求
response = curl_requests.post(
"https://grok.com/rest/app-chat/upload-file",
headers={
**get_dynamic_headers("/rest/app-chat/upload-file"),
"Cookie": cookie,
},
json=upload_data,
impersonate="chrome133a",
timeout=30,
**proxy_config,
)
# 检查响应
if response.status_code == 200:
result = response.json()
file_id = result.get("fileMetadataId", "")
logger.debug(f"[Upload] 图片上传成功,文件ID: {file_id}")
return file_id
return ""
except Exception as e:
logger.warning(f"[Upload] 上传图片失败: {e}")
return ""
@staticmethod
def _is_url(image_input: str) -> bool:
try:
result = urlparse(image_input)
return all([result.scheme, result.netloc]) and result.scheme in ['http', 'https']
except Exception as e:
logger.warning(f"[Upload] 上传图片失败: {e}")
return False
@staticmethod
def _download(url: str) -> tuple[str, str]:
try:
response = curl_requests.get(url, timeout=5)
response.raise_for_status()
# 获取内容类型
content_type = response.headers.get('content-type', 'image/jpeg')
if not content_type.startswith('image/'):
content_type = 'image/jpeg'
# 转换为 Base64
image_base64 = base64.b64encode(response.content).decode('utf-8')
return image_base64, content_type
except Exception as e:
logger.warning(f"[Upload] 下载图片失败: {e}")
return "", ""
@staticmethod
def _get_info(image_data: str, mime_type: str = None) -> tuple[str, str]:
# mime_type 有值,直接使用
if mime_type:
extension = mime_type.split("/")[1] if "/" in mime_type else "jpg"
file_name = f"image.{extension}"
return file_name, mime_type
# mime_type 没有值,使用默认值
mime_type = "image/jpeg"
extension = "jpg"
# 从 Base64 数据中提取 MIME 类型
if "data:image" in image_data:
match = re.search(r"data:([a-zA-Z0-9]+/[a-zA-Z0-9-.+]+);base64,", image_data)
if match:
mime_type = match.group(1)
extension = mime_type.split("/")[1]
file_name = f"image.{extension}"
return file_name, mime_type
+493
View File
@@ -0,0 +1,493 @@
<!DOCTYPE html>
<html lang="zh-CN" class="h-full">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>管理控制台 - Grok2API</title>
<script src="https://cdn.tailwindcss.com"></script>
<style>
@keyframes slide-up {
from {
transform: translateY(100%);
opacity: 0;
}
to {
transform: translateY(0);
opacity: 1;
}
}
.animate-slide-up {
animation: slide-up 0.3s ease-out;
}
.tab-btn {
transition: all 0.2s ease;
}
[title] {
position: relative;
}
[title]:hover::after {
content: attr(title);
position: absolute;
bottom: 100%;
left: 50%;
transform: translateX(-50%) translateY(-4px);
background: hsl(0 0% 3.9%);
color: hsl(0 0% 98%);
padding: 4px 8px;
border-radius: 4px;
font-size: 11px;
white-space: nowrap;
z-index: 1000;
pointer-events: none;
box-shadow: 0 2px 8px rgba(0,0,0,0.15);
}
[title]:hover::before {
content: '';
position: absolute;
bottom: 100%;
left: 50%;
transform: translateX(-50%);
border: 4px solid transparent;
border-top-color: hsl(0 0% 3.9%);
z-index: 1000;
}
</style>
<script>
tailwind.config = {
theme: {
extend: {
colors: {
border: "hsl(0 0% 89%)",
input: "hsl(0 0% 89%)",
ring: "hsl(0 0% 3.9%)",
background: "hsl(0 0% 100%)",
foreground: "hsl(0 0% 3.9%)",
primary: {
DEFAULT: "hsl(0 0% 9%)",
foreground: "hsl(0 0% 98%)",
},
secondary: {
DEFAULT: "hsl(0 0% 96.1%)",
foreground: "hsl(0 0% 9%)",
},
muted: {
DEFAULT: "hsl(0 0% 96.1%)",
foreground: "hsl(0 0% 45.1%)",
},
accent: {
DEFAULT: "hsl(0 0% 96.1%)",
foreground: "hsl(0 0% 9%)",
},
destructive: {
DEFAULT: "hsl(0 84.2% 60.2%)",
foreground: "hsl(0 0% 98%)",
},
},
}
}
}
</script>
</head>
<body class="h-full bg-background text-foreground antialiased">
<!-- 导航栏 -->
<header class="sticky top-0 z-50 w-full border-b border-border/40 bg-background/95 backdrop-blur supports-[backdrop-filter]:bg-background/60">
<div class="mx-auto flex h-14 max-w-7xl items-center px-6">
<div class="mr-4 flex">
<span class="font-bold text-lg">Grok2API</span>
</div>
<div class="flex flex-1 items-center justify-between space-x-2 md:justify-end">
<nav class="flex items-center space-x-2">
<button
onclick="logout()"
class="inline-flex items-center justify-center text-sm font-medium transition-colors focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-ring disabled:pointer-events-none disabled:opacity-50 hover:bg-accent hover:text-accent-foreground h-9 px-4 py-2"
>
退出
</button>
</nav>
</div>
</div>
</header>
<main class="mx-auto max-w-7xl px-6 py-6">
<!-- Tab 导航 -->
<div class="border-b border-border mb-6">
<nav class="flex space-x-8">
<button onclick="switchTab('tokens')" id="tabTokens" class="tab-btn border-b-2 border-primary text-sm font-medium py-3 px-1">Token 管理</button>
<button onclick="switchTab('settings')" id="tabSettings" class="tab-btn border-b-2 border-transparent text-sm font-medium py-3 px-1">Setting 配置</button>
</nav>
</div>
<!-- Token 管理面板 -->
<div id="panelTokens">
<!-- 统计卡片 -->
<div class="grid gap-4 grid-cols-2 md:grid-cols-4 mb-6">
<div class="rounded-lg border border-border bg-background p-4">
<p class="text-sm font-medium text-muted-foreground mb-2">Token 总数</p>
<h3 class="text-xl font-bold" id="statTotal">-</h3>
</div>
<div class="rounded-lg border border-border bg-background p-4">
<p class="text-sm font-medium text-muted-foreground mb-2">未使用</p>
<h3 class="text-xl font-bold text-gray-500" id="statUnused">-</h3>
</div>
<div class="rounded-lg border border-border bg-background p-4">
<p class="text-sm font-medium text-muted-foreground mb-2">限流中</p>
<h3 class="text-xl font-bold text-orange-600" id="statLimited">-</h3>
</div>
<div class="rounded-lg border border-border bg-background p-4">
<p class="text-sm font-medium text-muted-foreground mb-2">失效</p>
<h3 class="text-xl font-bold text-destructive" id="statExpired">-</h3>
</div>
<div class="rounded-lg border border-border bg-background p-4">
<p class="text-sm font-medium text-muted-foreground mb-2">正常</p>
<h3 class="text-xl font-bold text-green-600" id="statActive">-</h3>
</div>
<div class="rounded-lg border border-border bg-background p-4">
<p class="text-sm font-medium text-muted-foreground mb-2">总剩余</p>
<h3 class="text-xl font-bold" id="statTotalRemaining">-</h3>
</div>
<div class="rounded-lg border border-border bg-background p-4">
<p class="text-sm font-medium text-muted-foreground mb-2">普通剩余</p>
<h3 class="text-xl font-bold text-blue-600" id="statNormalRemaining">-</h3>
</div>
<div class="rounded-lg border border-border bg-background p-4">
<p class="text-sm font-medium text-muted-foreground mb-2">高级剩余</p>
<h3 class="text-xl font-bold text-purple-600" id="statHeavyRemaining">-</h3>
</div>
</div>
<!-- Token 列表 -->
<div class="rounded-lg border border-border bg-background">
<!-- 工具栏 -->
<div class="flex items-center justify-between gap-4 p-4 border-b border-border">
<div class="flex items-center gap-3 flex-1">
<div class="flex items-center gap-2">
<select id="filterType" onchange="filterTokens()" class="h-8 px-3 text-sm rounded-md border border-input bg-background focus:outline-none focus:ring-1 focus:ring-ring">
<option value="all">全部类型</option>
<option value="sso">SSO</option>
<option value="ssoSuper">SuperSSO</option>
</select>
<select id="filterStatus" onchange="filterTokens()" class="h-8 px-3 text-sm rounded-md border border-input bg-background focus:outline-none focus:ring-1 focus:ring-ring">
<option value="all">全部状态</option>
<option value="未使用">未使用</option>
<option value="限流中">限流中</option>
<option value="失效">失效</option>
<option value="正常">正常</option>
</select>
</div>
</div>
<div class="flex items-center gap-2">
<button
onclick="refreshTokens()"
class="inline-flex items-center justify-center rounded-md transition-colors focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-ring hover:bg-accent hover:text-accent-foreground h-8 w-8"
title="刷新列表"
>
<svg class="h-4 w-4" xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2">
<polyline points="23 4 23 10 17 10"/>
<polyline points="1 20 1 14 7 14"/>
<path d="M3.51 9a9 9 0 0 1 14.85-3.36L23 10M1 14l4.64 4.36A9 9 0 0 0 20.49 15"/>
</svg>
</button>
<div id="batchActions" class="hidden items-center gap-2">
<button
onclick="exportSelected()"
class="inline-flex items-center justify-center rounded-md transition-colors focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-ring hover:bg-accent hover:text-accent-foreground h-8 w-8"
title="导出选中项"
>
<svg class="h-4 w-4" xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2">
<path d="M21 15v4a2 2 0 0 1-2 2H5a2 2 0 0 1-2-2v-4"/>
<polyline points="7 10 12 15 17 10"/>
<line x1="12" y1="15" x2="12" y2="3"/>
</svg>
</button>
<button
onclick="batchDelete()"
class="inline-flex items-center justify-center rounded-md transition-colors focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-ring hover:bg-destructive/10 hover:text-destructive h-8 w-8"
title="批量删除"
>
<svg class="h-4 w-4" xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2">
<polyline points="3 6 5 6 21 6"/>
<path d="M19 6v14a2 2 0 0 1-2 2H7a2 2 0 0 1-2-2V6m3 0V4a2 2 0 0 1 2-2h4a2 2 0 0 1 2 2v2"/>
<line x1="10" y1="11" x2="10" y2="17"/>
<line x1="14" y1="11" x2="14" y2="17"/>
</svg>
</button>
</div>
<button
onclick="openAddModal()"
class="inline-flex items-center justify-center rounded-md transition-colors focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-ring bg-primary text-primary-foreground hover:bg-primary/90 h-8 px-3"
title="添加 Token"
>
<svg class="h-4 w-4 mr-2" xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2">
<line x1="12" y1="5" x2="12" y2="19"/>
<line x1="5" y1="12" x2="19" y2="12"/>
</svg>
<span class="text-sm font-medium">新增</span>
</button>
</div>
</div>
<!-- 表格 -->
<div class="relative w-full overflow-auto">
<table class="w-full text-sm table-fixed">
<thead>
<tr class="border-b border-border">
<th class="h-10 px-3 text-left align-middle font-medium w-12">
<input
type="checkbox"
id="selectAll"
onchange="toggleSelectAll()"
class="h-3.5 w-3.5 rounded border border-input focus:ring-1 focus:ring-ring"
>
</th>
<th class="h-10 px-3 text-left align-middle text-sm font-medium text-muted-foreground w-72">Token</th>
<th class="h-10 px-3 text-left align-middle text-sm font-medium text-muted-foreground w-20">类型</th>
<th class="h-10 px-3 text-left align-middle text-sm font-medium text-muted-foreground w-20">状态</th>
<th class="h-10 px-3 text-left align-middle text-sm font-medium text-muted-foreground w-20">普通剩余</th>
<th class="h-10 px-3 text-left align-middle text-sm font-medium text-muted-foreground w-20">高级剩余</th>
<th class="h-10 px-3 text-left align-middle text-sm font-medium text-muted-foreground w-40">创建时间</th>
<th class="h-10 px-3 text-right align-middle text-sm font-medium text-muted-foreground w-20">操作</th>
</tr>
</thead>
<tbody id="tokenTableBody" class="divide-y divide-border">
<!-- 动态填充 -->
</tbody>
</table>
</div>
<div id="emptyState" class="hidden flex flex-col items-center justify-center py-12">
<svg class="h-10 w-10 text-muted-foreground/50 mb-3" xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2">
<rect x="3" y="3" width="18" height="18" rx="2"/>
<path d="M9 9h6v6H9z"/>
</svg>
<p class="text-sm text-muted-foreground">暂无数据</p>
</div>
</div>
</div>
<!-- 全局设置面板 -->
<div id="panelSettings" class="hidden">
<div class="grid gap-6 lg:grid-cols-2">
<!-- 全局配置 -->
<div class="rounded-lg border border-border bg-background p-6">
<h3 class="text-lg font-semibold mb-4">全局配置</h3>
<div class="space-y-4">
<div>
<label class="text-sm font-medium flex items-center gap-1 mb-2">
管理账户
<span class="inline-flex items-center justify-center w-3.5 h-3.5 rounded-full border border-muted-foreground text-muted-foreground cursor-help" style="font-size:10px;line-height:1" title="登录管理后台的用户名">?</span>
</label>
<input id="cfgAdminUser" class="flex h-9 w-full rounded-md border border-input bg-background px-3 py-2 text-sm" placeholder="admin">
</div>
<div>
<label class="text-sm font-medium flex items-center gap-1 mb-2">
管理密码
<span class="inline-flex items-center justify-center w-3.5 h-3.5 rounded-full border border-muted-foreground text-muted-foreground cursor-help" style="font-size:10px;line-height:1" title="登录管理后台的密码,留空表示不修改当前密码">?</span>
</label>
<input id="cfgAdminPass" type="password" class="flex h-9 w-full rounded-md border border-input bg-background px-3 py-2 text-sm" placeholder="留空则不修改">
</div>
<div>
<label class="text-sm font-medium flex items-center gap-1 mb-2">
日志级别
<span class="inline-flex items-center justify-center w-3.5 h-3.5 rounded-full border border-muted-foreground text-muted-foreground cursor-help" style="font-size:10px;line-height:1" title="日志详细程度。DEBUG:最详细 | INFO:一般信息 | WARNING:警告 | ERROR:仅错误">?</span>
</label>
<select id="cfgLogLevel" class="flex h-9 w-full rounded-md border border-input bg-background px-3 py-2 text-sm">
<option>DEBUG</option><option>INFO</option><option>WARNING</option><option>ERROR</option>
</select>
</div>
<div>
<label class="text-sm font-medium flex items-center gap-1 mb-2">
缓存上限 (MB)
<span class="inline-flex items-center justify-center w-3.5 h-3.5 rounded-full border border-muted-foreground text-muted-foreground cursor-help" style="font-size:10px;line-height:1" title="图片缓存目录的最大容量,超过后自动删除最旧的缓存文件">?</span>
</label>
<input id="cfgTempMaxSize" type="number" class="flex h-9 w-full rounded-md border border-input bg-background px-3 py-2 text-sm" placeholder="500">
</div>
<div>
<label class="text-sm font-medium flex items-center gap-1 mb-2">
缓存大小
<span class="inline-flex items-center justify-center w-3.5 h-3.5 rounded-full border border-muted-foreground text-muted-foreground cursor-help" style="font-size:10px;line-height:1" title="当前缓存目录的大小">?</span>
</label>
<div class="flex gap-2">
<input id="cacheSize" readonly class="flex h-9 flex-1 rounded-md border border-input bg-muted px-3 py-2 text-sm" placeholder="0 MB">
<button onclick="clearCache()" class="inline-flex items-center justify-center rounded-md text-sm font-medium bg-orange-600 text-white hover:bg-orange-700 h-9 px-3 transition-colors">
<svg class="h-4 w-4 mr-1" xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2">
<path d="M3 6h18"/>
<path d="M19 6v14a2 2 0 0 1-2 2H7a2 2 0 0 1-2-2V6m3 0V4a2 2 0 0 1 2-2h4a2 2 0 0 1 2 2v2"/>
<line x1="10" y1="11" x2="10" y2="17"/>
<line x1="14" y1="11" x2="14" y2="17"/>
</svg>
清除
</button>
</div>
</div>
<div>
<label class="text-sm font-medium flex items-center gap-1 mb-2">
服务网址
<span class="inline-flex items-center justify-center w-3.5 h-3.5 rounded-full border border-muted-foreground text-muted-foreground cursor-help" style="font-size:10px;line-height:1" title="用于生成图片链接的服务地址。建议使用域名以保护服务器IP,如无图片需求可留空">?</span>
</label>
<input id="cfgBaseUrl" class="flex h-9 w-full rounded-md border border-input bg-background px-3 py-2 text-sm" placeholder="http://localhost:8000">
</div>
</div>
<div class="mt-6 flex justify-end">
<button onclick="saveGlobalSettings()" class="inline-flex items-center justify-center rounded-md text-sm font-medium bg-primary text-primary-foreground hover:bg-primary/90 h-9 px-6 transition-colors">保存配置</button>
</div>
</div>
<!-- Grok 配置 -->
<div class="rounded-lg border border-border bg-background p-6">
<h3 class="text-lg font-semibold mb-4">Grok 配置</h3>
<div class="space-y-4">
<div>
<label class="text-sm font-medium flex items-center gap-1 mb-2">
API Key
<span class="inline-flex items-center justify-center w-3.5 h-3.5 rounded-full border border-muted-foreground text-muted-foreground cursor-help" style="font-size:10px;line-height:1" title="API访问密钥。建议填写以提高安全性,可留空">?</span>
</label>
<input id="cfgApiKey" class="flex h-9 w-full rounded-md border border-input bg-background px-3 py-2 text-sm" placeholder="">
</div>
<div>
<label class="text-sm font-medium flex items-center gap-1 mb-2">
Proxy Url
<span class="inline-flex items-center justify-center w-3.5 h-3.5 rounded-full border border-muted-foreground text-muted-foreground cursor-help" style="font-size:10px;line-height:1" title="HTTP代理服务器地址,用于访问Grok服务,可留空">?</span>
</label>
<input id="cfgProxyUrl" class="flex h-9 w-full rounded-md border border-input bg-background px-3 py-2 text-sm" placeholder="http://127.0.0.1:7890">
</div>
<div>
<label class="text-sm font-medium flex items-center gap-1 mb-2">
CF Clearance
<span class="inline-flex items-center justify-center w-3.5 h-3.5 rounded-full border border-muted-foreground text-muted-foreground cursor-help" style="font-size:10px;line-height:1" title="Cloudflare安全令牌,用于绕过人机验证,可留空">?</span>
</label>
<input id="cfgCfClearance" class="flex h-9 w-full rounded-md border border-input bg-background px-3 py-2 text-sm" placeholder="">
</div>
<div>
<label class="text-sm font-medium flex items-center gap-1 mb-2">
X Statsig ID
<span class="inline-flex items-center justify-center w-3.5 h-3.5 rounded-full border border-muted-foreground text-muted-foreground cursor-help" style="font-size:10px;line-height:1" title="Grok反机器人检测的唯一标识符,必填">?</span>
</label>
<input id="cfgStatsigId" class="flex h-9 w-full rounded-md border border-input bg-background px-3 py-2 text-sm" placeholder="">
</div>
<div>
<label class="text-sm font-medium flex items-center gap-1 mb-2">
Filtered_tags
<span class="inline-flex items-center justify-center w-3.5 h-3.5 rounded-full border border-muted-foreground text-muted-foreground cursor-help" style="font-size:10px;line-height:1" title="过滤Grok响应中的指定标签,多个标签用逗号分隔">?</span>
</label>
<input id="cfgFilteredTags" class="flex h-9 w-full rounded-md border border-input bg-background px-3 py-2 text-sm" placeholder="xaiartifact,xai:tool_usage_card">
</div>
<div>
<label class="text-sm font-medium flex items-center gap-1 mb-2">
临时会话
<span class="inline-flex items-center justify-center w-3.5 h-3.5 rounded-full border border-muted-foreground text-muted-foreground cursor-help" style="font-size:10px;line-height:1" title="选择会话模式:标准模式保留上下文,临时模式每次对话独立">?</span>
</label>
<select id="cfgTemporary" class="flex h-9 w-full rounded-md border border-input bg-background px-3 py-2 text-sm focus:outline-none focus:ring-1 focus:ring-ring">
<option value="false">关闭</option>
<option value="true">开启</option>
</select>
</div>
</div>
<div class="mt-6 flex justify-end">
<button onclick="saveGrokSettings()" class="inline-flex items-center justify-center rounded-md text-sm font-medium bg-primary text-primary-foreground hover:bg-primary/90 h-9 px-6 transition-colors">保存配置</button>
</div>
</div>
</div>
</div>
</main>
<!-- 添加 Token 模态框 -->
<div id="addModal" class="hidden fixed inset-0 z-50 bg-black/80 flex items-center justify-center p-4">
<div class="bg-background rounded-lg border border-border w-full max-w-2xl max-h-[90vh] overflow-y-auto shadow-xl">
<div class="flex items-center justify-between p-5 border-b border-border">
<h3 class="text-lg font-semibold">添加 Token</h3>
<button onclick="closeAddModal()" class="text-muted-foreground hover:text-foreground transition-colors">
<svg class="h-3.5 w-3.5" xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2">
<line x1="18" y1="6" x2="6" y2="18"/>
<line x1="6" y1="6" x2="18" y2="18"/>
</svg>
</button>
</div>
<div class="p-5 space-y-4">
<div class="space-y-2">
<label class="text-sm font-medium">Token 类型</label>
<select id="addTokenType" class="flex h-9 w-full rounded-md border border-input bg-background px-3 py-2 text-sm focus:outline-none focus:ring-1 focus:ring-ring">
<option value="sso">SSO</option>
<option value="ssoSuper">SuperSSO</option>
</select>
</div>
<div class="space-y-2">
<label class="text-sm font-medium">Token 列表 <span class="text-muted-foreground">(每行一个)</span></label>
<textarea
id="addTokenList"
rows="12"
placeholder="请输入 Token,每行一个"
class="flex w-full rounded-md border border-input bg-background px-3 py-2 text-sm focus:outline-none focus:ring-1 focus:ring-ring font-mono resize-none"
></textarea>
</div>
</div>
<div class="flex items-center justify-end gap-3 p-5 border-t border-border bg-muted/30">
<button
onclick="closeAddModal()"
class="inline-flex items-center justify-center rounded-md text-sm font-medium transition-colors focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-ring disabled:pointer-events-none disabled:opacity-50 border border-input bg-background hover:bg-accent h-9 px-5"
>
取消
</button>
<button
onclick="submitAddTokens()"
class="inline-flex items-center justify-center rounded-md text-sm font-medium transition-colors focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-ring disabled:pointer-events-none disabled:opacity-50 bg-primary text-primary-foreground hover:bg-primary/90 h-9 px-5"
>
添加
</button>
</div>
</div>
</div>
<script>
let allTokens=[],filteredTokens=[],selectedTokens=new Set();
const $=(id)=>document.getElementById(id),
checkAuth=()=>{const t=localStorage.getItem('adminToken');return t||(location.href='/login',null),t},
apiRequest=async(url,opts={})=>{const t=checkAuth();if(!t)return null;const r=await fetch(url,{...opts,headers:{...opts.headers,Authorization:`Bearer ${t}`,'Content-Type':'application/json'}});return r.status===401?(localStorage.removeItem('adminToken'),location.href='/login',null):r},
loadStats=async()=>{try{const r=await apiRequest('/api/stats');if(!r)return;const d=await r.json();if(d.success){const s=d.data;$('statTotal').textContent=s.total||0;['Unused','Limited','Expired','Active'].forEach((k,i)=>$(`stat${k}`).textContent=(s.normal?.[k.toLowerCase()]||0)+(s.super?.[k.toLowerCase()]||0))}}catch(e){console.error('加载统计失败:',e)}},
calcRemaining=()=>{let n=0,h=0;allTokens.forEach(t=>{if(t.remaining_queries>0)n+=t.remaining_queries;if(t.heavy_remaining_queries>0)h+=t.heavy_remaining_queries});return{normal:n,heavy:h,total:n+h}},
loadTokens=async()=>{try{const r=await apiRequest('/api/tokens');if(!r)return;const d=await r.json();d.success&&(allTokens=d.data,filteredTokens=allTokens,selectedTokens.clear(),renderTokens(),updateRemaining())}catch(e){console.error('加载列表失败:',e)}},
updateRemaining=()=>{const r=calcRemaining();['Total','Normal','Heavy'].forEach(k=>$(`stat${k}Remaining`).textContent=r[k.toLowerCase()]===0?'-':r[k.toLowerCase()].toLocaleString())}
const renderTokens=()=>{const tb=$('tokenTableBody'),es=$('emptyState'),ss={'未使用':'bg-muted text-muted-foreground','限流中':'bg-orange-50 text-orange-700 border-orange-200','失效':'bg-destructive/10 text-destructive border-destructive/20','正常':'bg-green-50 text-green-700 border-green-200'},ts={sso:'bg-blue-50 text-blue-700 border-blue-200',ssoSuper:'bg-purple-50 text-purple-700 border-purple-200'},tl={sso:'SSO',ssoSuper:'SuperSSO'};if(!filteredTokens.length){tb.innerHTML='';es.classList.remove('hidden');$('selectAll').checked=false;return updateBatchActions()}es.classList.add('hidden');tb.innerHTML=filteredTokens.map(t=>`
<tr class="transition-colors">
<td class="py-2.5 px-3 align-middle w-12"><input type="checkbox" class="token-checkbox h-3.5 w-3.5 rounded border border-input focus:ring-1 focus:ring-ring" data-token="${t.token}" data-type="${t.token_type}" ${selectedTokens.has(t.token)?'checked':''} onchange="toggleToken('${t.token}')"></td>
<td class="py-2.5 px-3 align-middle w-80"><div class="flex items-center gap-2"><span class="font-mono text-xs">${t.token.substring(0,30)}...</span><button onclick="copyToken('${t.token.replace(/'/g,"\\'")}',event)" class="inline-flex items-center justify-center rounded-md transition-colors focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-ring hover:bg-accent h-6 w-6" title="复制完整 Token"><svg class="h-3 w-3 text-muted-foreground" xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"><rect x="9" y="9" width="13" height="13" rx="2" ry="2"/><path d="M5 15H4a2 2 0 0 1-2-2V4a2 2 0 0 1 2-2h9a2 2 0 0 1 2 2v1"/></svg></button></div></td>
<td class="py-2.5 px-3 align-middle w-20"><span class="inline-flex items-center rounded-full px-1.5 py-0.5 text-xs font-medium border ${ts[t.token_type]}">${tl[t.token_type]}</span></td>
<td class="py-2.5 px-3 align-middle w-20"><span class="inline-flex items-center rounded-full px-1.5 py-0.5 text-xs font-medium border ${ss[t.status]}">${t.status}</span></td>
<td class="py-2.5 px-3 align-middle w-20 text-xs tabular-nums">${t.remaining_queries===-1?'-':t.remaining_queries}</td>
<td class="py-2.5 px-3 align-middle w-20 text-xs tabular-nums">${t.heavy_remaining_queries===-1?'-':t.heavy_remaining_queries}</td>
<td class="py-2.5 px-3 align-middle w-32 text-xs text-muted-foreground">${t.created_time?new Date(t.created_time).toLocaleString('zh-CN',{dateStyle:'short',timeStyle:'short'}):'-'}</td>
<td class="py-2.5 px-3 align-middle text-right w-16"><button onclick="deleteToken('${t.token}','${t.token_type}')" class="inline-flex items-center justify-center rounded-md text-xs font-medium transition-colors focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-ring hover:bg-destructive/10 hover:text-destructive h-7 w-7"><svg class="h-3.5 w-3.5" xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"><polyline points="3 6 5 6 21 6"/><path d="M19 6v14a2 2 0 0 1-2 2H7a2 2 0 0 1-2-2V6m3 0V4a2 2 0 0 1 2-2h4a2 2 0 0 1 2 2v2"/></svg></button></td>
</tr>`).join('');updateBatchActions()},
toggleToken=t=>selectedTokens[selectedTokens.has(t)?'delete':'add'](t)||updateBatchActions(),
toggleSelectAll=()=>{const sa=$('selectAll');sa.checked?filteredTokens.forEach(t=>selectedTokens.add(t.token)):selectedTokens.clear();renderTokens()},
updateBatchActions=()=>{const ba=$('batchActions'),sc=$('selectedCount'),c=selectedTokens.size;ba.classList[c>0?'add':'remove']('flex');ba.classList[c>0?'remove':'add']('hidden');c>0&&(sc.textContent=`已选择 ${c} 项`);$('selectAll').checked=filteredTokens.length>0&&c===filteredTokens.length},
filterTokens=()=>{const tf=$('filterType').value,sf=$('filterStatus').value;filteredTokens=allTokens.filter(t=>(tf==='all'||t.token_type===tf)&&(sf==='all'||t.status===sf));selectedTokens.clear();renderTokens()},
refreshTokens=async()=>{await loadTokens();await loadStats()},
openAddModal=()=>$('addModal').classList.remove('hidden'),
closeAddModal=()=>{$('addModal').classList.add('hidden');$('addTokenList').value=''},
deleteToken=async(t,tt)=>{if(!confirm('确定要删除这个 Token 吗?'))return;try{const r=await apiRequest('/api/tokens/delete',{method:'POST',body:JSON.stringify({tokens:[t],token_type:tt})});if(!r)return;const d=await r.json();d.success?await refreshTokens():alert('删除失败: '+(d.error||'未知错误'))}catch(e){alert('删除失败: '+e.message)}},
batchDelete=async()=>{if(!selectedTokens.size||!confirm(`确定要删除选中的 ${selectedTokens.size} 个 Token 吗?此操作不可恢复!`))return;const tbt={sso:[],ssoSuper:[]};document.querySelectorAll('.token-checkbox:checked').forEach(cb=>tbt[cb.dataset.type].push(cb.dataset.token));try{const ps=[];['sso','ssoSuper'].forEach(k=>tbt[k].length&&ps.push(apiRequest('/api/tokens/delete',{method:'POST',body:JSON.stringify({tokens:tbt[k],token_type:k})})));await Promise.all(ps);await refreshTokens()}catch(e){alert('批量删除失败: '+e.message)}},
submitAddTokens=async()=>{const tt=$('addTokenType').value,tks=$('addTokenList').value.split('\n').map(t=>t.trim()).filter(t=>t);if(!tks.length)return alert('请输入至少一个 Token');try{const r=await apiRequest('/api/tokens/add',{method:'POST',body:JSON.stringify({tokens:tks,token_type:tt})});if(!r)return;const d=await r.json();d.success?(closeAddModal(),await refreshTokens()):alert('添加失败: '+(d.error||'未知错误'))}catch(e){alert('添加失败: '+e.message)}},
copyToken=async(t,e)=>{e.stopPropagation();try{await navigator.clipboard.writeText(t);showToast('Token 已复制到剪贴板','success')}catch(err){console.error('复制失败:',err);showToast('复制失败,请手动复制','error')}}
const exportSelected=()=>{if(!selectedTokens.size)return showToast('请先选择要导出的 Token','error');const sd=allTokens.filter(t=>selectedTokens.has(t.token)),csv=[['Token','类型','状态','普通调用剩余','高级调用剩余','创建时间'].join(','),...sd.map(t=>[`"${t.token}"`,t.token_type==='sso'?'SSO':'SuperSSO',t.status,t.remaining_queries===-1?'未使用':t.remaining_queries,t.heavy_remaining_queries===-1?'未使用':t.heavy_remaining_queries,`"${t.created_time?new Date(t.created_time).toLocaleString('zh-CN'):'-'}"`].join(','))].join('\n'),l=document.createElement('a');l.href=URL.createObjectURL(new Blob(['\uFEFF'+csv],{type:'text/csv;charset=utf-8;'}));l.download=`grok_tokens_${new Date().toISOString().slice(0,10)}.csv`;l.style.visibility='hidden';document.body.appendChild(l);l.click();document.body.removeChild(l);URL.revokeObjectURL(l.href);showToast(`已导出 ${selectedTokens.size} 个 Token`,'success')},
showToast=(m,t='info')=>{const d=document.createElement('div'),bc={success:'bg-green-600',error:'bg-destructive',info:'bg-primary'};d.className=`fixed bottom-4 right-4 ${bc[t]||bc.info} text-white px-4 py-2.5 rounded-lg shadow-lg text-sm font-medium z-50 animate-slide-up`;d.textContent=m;document.body.appendChild(d);setTimeout(()=>{d.style.opacity='0';d.style.transition='opacity 0.3s';setTimeout(()=>d.parentNode&&document.body.removeChild(d),300)},2000)},
logout=async()=>{if(!confirm('确定要退出登录吗?'))return;try{await apiRequest('/api/logout',{method:'POST'})}catch(e){console.error('登出失败:',e)}finally{localStorage.removeItem('adminToken');location.href='/login'}},
switchTab=t=>{['tokens','settings'].forEach(n=>{$(`panel${n.charAt(0).toUpperCase()+n.slice(1)}`).classList[n===t?'remove':'add']('hidden');$(`tab${n.charAt(0).toUpperCase()+n.slice(1)}`).classList[n===t?'add':'remove']('border-primary','text-primary');$(`tab${n.charAt(0).toUpperCase()+n.slice(1)}`).classList[n===t?'remove':'add']('border-transparent','text-muted-foreground')});t==='settings'&&loadSettings()},
loadSettings=async()=>{try{const r=await apiRequest('/api/settings');if(!r)return;const d=await r.json();if(d.success){const g=d.data.global,k=d.data.grok;$('cfgAdminUser').value=g.admin_username||'';$('cfgAdminPass').value='';$('cfgLogLevel').value=g.log_level||'DEBUG';$('cfgTempMaxSize').value=g.temp_max_size_mb||500;$('cfgBaseUrl').value=g.base_url||'';$('cfgApiKey').value=k.api_key||'';$('cfgProxyUrl').value=k.proxy_url||'';$('cfgCfClearance').value=k.cf_clearance||'';$('cfgStatsigId').value=k.x_statsig_id||'';$('cfgFilteredTags').value=k.filtered_tags||'';$('cfgTemporary').value=k.temporary!==false?'true':'false';await loadCacheSize()}}catch(e){console.error('加载配置失败:',e);showToast('加载配置失败','error')}},
loadCacheSize=async()=>{try{const r=await apiRequest('/api/cache/size');if(!r)return;const d=await r.json();if(d.success){$('cacheSize').value=d.data.size||'0 MB'}}catch(e){console.error('加载缓存大小失败:',e);$('cacheSize').value='0 MB'}},
clearCache=async()=>{if(!confirm('确定要清理缓存吗?此操作将删除 /data/temp 目录中的所有文件!'))return;try{const r=await apiRequest('/api/cache/clear',{method:'POST'});if(!r)return;const d=await r.json();if(d.success){showToast(`缓存清理完成,已删除 ${d.data.deleted_count||0} 个文件`,'success');await loadCacheSize()}else{showToast('清理失败: '+(d.error||'未知错误'),'error')}}catch(e){showToast('清理失败: '+e.message,'error')}},
saveGlobalSettings=async()=>{const gc={admin_username:$('cfgAdminUser').value,log_level:$('cfgLogLevel').value,temp_max_size_mb:parseInt($('cfgTempMaxSize').value)||500,base_url:$('cfgBaseUrl').value};if($('cfgAdminPass').value)gc.admin_password=$('cfgAdminPass').value;try{const r=await apiRequest('/api/settings');if(!r)return;const d=await r.json();if(!d.success)return showToast('加载配置失败','error');const s=await apiRequest('/api/settings',{method:'POST',body:JSON.stringify({global_config:gc,grok_config:d.data.grok})});if(!s)return;const sd=await s.json();sd.success?(showToast('全局配置保存成功','success'),$('cfgAdminPass').value=''):showToast('保存失败: '+(sd.error||'未知错误'),'error')}catch(e){showToast('保存失败: '+e.message,'error')}},
saveGrokSettings=async()=>{const kc={api_key:$('cfgApiKey').value,proxy_url:$('cfgProxyUrl').value,cf_clearance:$('cfgCfClearance').value,x_statsig_id:$('cfgStatsigId').value,filtered_tags:$('cfgFilteredTags').value,temporary:$('cfgTemporary').value==='true'};try{const r=await apiRequest('/api/settings');if(!r)return;const d=await r.json();if(!d.success)return showToast('加载配置失败','error');const s=await apiRequest('/api/settings',{method:'POST',body:JSON.stringify({global_config:d.data.global,grok_config:kc})});if(!s)return;const sd=await s.json();sd.success?showToast('Grok配置保存成功','success'):showToast('保存失败: '+(sd.error||'未知错误'),'error')}catch(e){showToast('保存失败: '+e.message,'error')}};
window.addEventListener('DOMContentLoaded',()=>{checkAuth();refreshTokens();setInterval(()=>{loadStats();updateRemaining()},30000)});
</script>
</body>
</html>
+180
View File
@@ -0,0 +1,180 @@
<!DOCTYPE html>
<html lang="zh-CN" class="h-full">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>登录 - Grok2API</title>
<script src="https://cdn.tailwindcss.com"></script>
<script>
tailwind.config = {
theme: {
extend: {
colors: {
border: "hsl(0 0% 89%)",
input: "hsl(0 0% 89%)",
ring: "hsl(0 0% 3.9%)",
background: "hsl(0 0% 100%)",
foreground: "hsl(0 0% 3.9%)",
primary: {
DEFAULT: "hsl(0 0% 9%)",
foreground: "hsl(0 0% 98%)",
},
secondary: {
DEFAULT: "hsl(0 0% 96.1%)",
foreground: "hsl(0 0% 9%)",
},
muted: {
DEFAULT: "hsl(0 0% 96.1%)",
foreground: "hsl(0 0% 45.1%)",
},
destructive: {
DEFAULT: "hsl(0 84.2% 60.2%)",
foreground: "hsl(0 0% 98%)",
},
},
}
}
}
</script>
<style>
@keyframes slide-up {
from {
transform: translateY(100%);
opacity: 0;
}
to {
transform: translateY(0);
opacity: 1;
}
}
.animate-slide-up {
animation: slide-up 0.3s ease-out;
}
</style>
</head>
<body class="h-full bg-background text-foreground antialiased">
<div class="flex min-h-full flex-col justify-center py-12 px-4 sm:px-6 lg:px-8">
<div class="sm:mx-auto sm:w-full sm:max-w-md">
<div class="text-center">
<h1 class="text-4xl font-bold">Grok2API</h1>
<p class="mt-2 text-sm text-muted-foreground">管理员控制台</p>
</div>
</div>
<div class="sm:mx-auto sm:w-full sm:max-w-md">
<div class="bg-background py-8 px-4 sm:px-10 rounded-lg">
<form id="loginForm" class="space-y-6">
<!-- 账户 -->
<div class="space-y-2">
<label for="username" class="text-sm font-medium leading-none peer-disabled:cursor-not-allowed peer-disabled:opacity-70">
账户
</label>
<input
type="text"
id="username"
name="username"
required
class="flex h-10 w-full rounded-md border border-input bg-background px-3 py-2 text-sm ring-offset-background file:border-0 file:bg-transparent file:text-sm file:font-medium placeholder:text-muted-foreground focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring focus-visible:ring-offset-0 disabled:cursor-not-allowed disabled:opacity-50 transition-colors"
placeholder="请输入用户名"
>
</div>
<!-- 密码 -->
<div class="space-y-2">
<label for="password" class="text-sm font-medium leading-none peer-disabled:cursor-not-allowed peer-disabled:opacity-70">
密码
</label>
<input
type="password"
id="password"
name="password"
required
class="flex h-10 w-full rounded-md border border-input bg-background px-3 py-2 text-sm ring-offset-background file:border-0 file:bg-transparent file:text-sm file:font-medium placeholder:text-muted-foreground focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring focus-visible:ring-offset-0 disabled:cursor-not-allowed disabled:opacity-50 transition-colors"
placeholder="请输入密码"
>
</div>
<!-- 登录按钮 -->
<button
type="submit"
id="loginButton"
class="inline-flex items-center justify-center whitespace-nowrap rounded-md font-medium transition-colors focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring focus-visible:ring-offset-0 disabled:pointer-events-none disabled:opacity-50 bg-primary text-primary-foreground hover:bg-primary/90 h-10 px-4 py-2 w-full text-sm"
>
登录
</button>
</form>
<div class="mt-6 text-center text-xs text-muted-foreground">
<p>Created By Chenyme © 2025</p>
</div>
</div>
</div>
</div>
<script>
const loginForm = document.getElementById('loginForm');
const loginButton = document.getElementById('loginButton');
loginForm.addEventListener('submit', async (e) => {
e.preventDefault();
loginButton.disabled = true;
loginButton.textContent = '登录中...';
try {
const formData = new FormData(loginForm);
const response = await fetch('/api/login', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
username: formData.get('username'),
password: formData.get('password')
})
});
const data = await response.json();
if (data.success) {
localStorage.setItem('adminToken', data.token);
window.location.href = '/manage';
} else {
showToast(data.message || '登录失败', 'error');
}
} catch (error) {
showToast('网络错误,请稍后重试', 'error');
} finally {
loginButton.disabled = false;
loginButton.textContent = '登录';
}
});
function showToast(message, type = 'error') {
const toast = document.createElement('div');
const bgColors = {
success: 'bg-green-600',
error: 'bg-destructive',
info: 'bg-primary'
};
toast.className = `fixed bottom-4 right-4 ${bgColors[type] || bgColors.error} text-white px-4 py-2.5 rounded-lg shadow-lg text-sm font-medium z-50 animate-slide-up`;
toast.textContent = message;
document.body.appendChild(toast);
setTimeout(() => {
toast.style.opacity = '0';
toast.style.transition = 'opacity 0.3s';
setTimeout(() => toast.parentNode && document.body.removeChild(toast), 300);
}, 2000);
}
window.addEventListener('DOMContentLoaded', () => {
const token = localStorage.getItem('adminToken');
if (token) {
fetch('/api/stats', {
headers: { 'Authorization': `Bearer ${token}` }
}).then(response => {
if (response.ok) window.location.href = '/manage';
});
}
});
</script>
</body>
</html>
+14
View File
@@ -0,0 +1,14 @@
[global]
admin_username = "admin"
admin_password = "admin"
log_level = "INFO"
temp_max_size_mb = 500
base_url = ""
[grok]
api_key = ""
proxy_url = ""
cf_clearance = ""
temporary = true
filtered_tags = "xaiartifact,xai:tool_usage_card,grok:render"
x_statsig_id = "ZTpUeXBlRXJyb3I6IENhbm5vdCByZWFkIHByb3BlcnRpZXMgb2YgdW5kZWZpbmVkIChyZWFkaW5nICdjaGlsZE5vZGVzJyk="
View File
Whitespace-only changes.
+4
View File
@@ -0,0 +1,4 @@
{
"ssoNormal": {},
"ssoSuper": {}
}
+11
View File
@@ -0,0 +1,11 @@
services:
grok2api:
image: ghcr.io/chenyme/grok2api:latest
ports:
- "8000:8000"
volumes:
- grok_data:/app/data
- ./logs:/app/logs
volumes:
grok_data:
+53
View File
@@ -0,0 +1,53 @@
"""FastAPI应用主入口"""
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from app.core.logger import logger
from app.core.exception import register_exception_handlers
from app.api.v1.chat import router as chat_router
from app.api.v1.models import router as models_router
from app.api.v1.images import router as images_router
from app.api.admin.manage import router as admin_router
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用生命周期管理"""
logger.debug("[Web2API] 应用启动成功")
yield
logger.info("[Web2API] 应用关闭成功")
# 初始化日志
logger.info("[Web2API] 应用正在启动...")
# 创建FastAPI应用
app = FastAPI(
title="Web2API",
description="Web服务API",
version="1.0.0",
lifespan=lifespan
)
# 注册全局异常处理器
register_exception_handlers(app)
# 注册路由
app.include_router(chat_router, prefix="/v1")
app.include_router(models_router, prefix="/v1")
app.include_router(images_router)
app.include_router(admin_router)
# 挂载静态文件(注意:这个应该在API路由之后,避免拦截API请求)
app.mount("/static", StaticFiles(directory="app/template"), name="template")
@app.get("/")
async def root():
"""根路径"""
return {"message": "Welcome to Web2API"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8001)
+117
View File
@@ -0,0 +1,117 @@
# Grok2API
基于 FastAPI 重构的 Grok2API,适配最新的Web调用格式,支持流式对话、图像生成、图像编辑、联网搜索、深度思考,支持号池并发调用和自动负载均衡。
### 接口说明
> 同 OpenAI 官方接口一致,API 请求使用通过 Authorization header 进行认证
| 方法 | 端点 | 描述 | 认证 |
|------|------|------|------|
| POST | `/v1/chat/completions` | 创建聊天对话,支持流式和非流式响应,兼容OpenAI格式 | ✅ |
| GET | `/v1/models` | 获取支持的所有模型列表 | ✅ |
| GET | `/v1/models/{model_id}` | 获取特定模型信息 | ✅ |
| GET | `/images/{img_path}` | 获取生成的图片文件 | ❌ |
<details>
<summary>其余管理接口</summary>
| 方法 | 端点 | 描述 | 认证 |
|------|------|------|------|
| GET | /login | 管理员登录页面 | ❌ |
| GET | /manage | 管理控制台页面 | ❌ |
| POST | /api/login | 管理员登录认证 | ❌ |
| POST | /api/logout | 管理员登出 | ✅ |
| GET | /api/tokens | 获取Token列表 | ✅ |
| POST | /api/tokens/add | 批量添加Token | ✅ |
| POST | /api/tokens/delete | 批量删除Token | ✅ |
| GET | /api/settings | 获取系统配置 | ✅ |
| POST | /api/settings | 更新系统配置 | ✅ |
| GET | /api/cache/size | 获取缓存大小 | ✅ |
| POST | /api/cache/clear | 清理图片缓存 | ✅ |
| GET | /api/stats | 获取统计信息 | ✅ |
</details>
<br>
### 使用说明
#### 调用次数说明
- 普通账号(Basic)可免费使用 **80 次/20 小时**,Super 账号配额尚未确定(我没有号就没测)。
- 系统自动监控并负载均衡各账号调用次数,可在**管理页面**直观查看各账户实时用量和状态。
#### 图像生成说明
- 只需在对话内容中输入如“给我画一个月亮”即可自动触发生成图片。
- 每次生成会以 **Markdown 格式返回两张图片**,计 4 次额度。
- **由于 Grok 的图片直链受限(403),系统会自动将图片缓存到本地。要确保图片能正常显示,请务必正确设置 Base Url。**
#### 关于 x_statsig_id
- `x_statsig_id` 是 Grok 用于识别机器人请求的 Token,目前网络上已经有相关的逆向分析资料,有兴趣可自行查阅。
- 如果您不确定如何获取 x_statsig_id,建议不要修改配置文件中的该参数,保持默认值即可,无需关注。
- 项目最初借鉴了 [VeroFess/grok2api](https://github.com/VeroFess/grok2api) 的实现,曾通过 Camoufox 尝试绕过原项目的 Playwright 获取 x_statsig_id 时触发的 403。起初尚可使用,但近期 Grok 已对未登录用户的 x_statsig_id 做了限制,因此已弃用改方法,当前直接使用固定值以兼容调用。
#### docker-compose 部署
```
services:
grok2api:
image: ghcr.io/chenyme/grok2api:latest
ports:
- "8000:8000"
volumes:
- grok_data:/app/data
- ./logs:/app/logs
volumes:
grok_data:
```
<br>
### 可用模型
> 所有模型均可进行图像生成,每次计次为 4 次
| 模型名称 | 调用计次 | 可用账户 | 图像生成/编辑 | 深度思考 | 联网搜索 |
|----------------------|---------|--------------|--------------|----------|----------|
| `grok-3-fast` | 1 | Basic/Super | ✅ | ❌ | ✅ |
| `grok-4-fast` | 1 | Basic/Super | ✅ | ✅ | ✅ |
| `grok-4-fast-expert` | 4 | Basic/Super | ✅ | ✅ | ✅ |
| `grok-4-expert` | 4 | Basic/Super | ✅ | ✅ | ✅ |
| `grok-4-heavy` | 1 | Super | ✅ | ✅ | ✅ |
<br>
### 参数说明
> 所有参数请在启动服务后,进入 `/login` 路由后登陆管理进行配置
| 参数名 | 位置 | 必填 | 说明 | 默认值 |
|---------|---------|------|------|--------|
| admin_username | global | 否 | 管理后台登录用户名 | "admin" |
| admin_password | global | 否 | 管理后台登录密码 | "admin" |
| log_level | global | 否 | 日志级别:DEBUG/INFO/WARNING/ERROR | "INFO" |
| temp_max_size_mb | global | 否 | 图片缓存目录最大容量(MB),超过后自动清理 | 500 |
| base_url | global | 否 | 服务基础URL,用于生成图片链接,保护服务器IP | "" |
| api_key | grok | 否 | API访问密钥,提高安全性,可留空 | "" |
| proxy_url | grok | 否 | HTTP代理服务器地址,访问Grok服务 | "" |
| cf_clearance | grok | 否 | Cloudflare安全令牌,绕过人机验证 | "" |
| x_statsig_id | grok | 是 | Grok反机器人检测唯一标识符 | "ZTpUeXBlRXJyb3I6IENhbm5vdCByZWFkIHByb3BlcnRpZXMgb2YgdW5kZWZpbmVkIChyZWFkaW5nICdjaGlsZE5vZGVzJyk=" |
| filtered_tags | grok | 否 | 过滤响应标签,多个用逗号分隔 | "xaiartifact,xai:tool_usage_card,grok:render" |
| temporary | grok | 否 | 会话模式:true(临时)/false | true |
<br>
### 注意事项
⚠️ 本项目仅供学习和研究目的,请遵守相关使用条款。
<br>
---
<br>
此项目基于其他 Grok2API 项目进行学习重构,感谢 [LINUX DO](https://linux.do)、[VeroFess/grok2api](https://github.com/VeroFess/grok2api)、[xLmiler/grok2api_python](https://github.com/xLmiler/grok2api_python) 的项目大佬
+8
View File
@@ -0,0 +1,8 @@
toml==0.10.2
fastapi==0.118.2
uvicorn==0.37.0
python-dotenv==1.1.1
curl_cffi==0.13.0
requests==2.32.5
starlette==0.48.0
pydantic==2.12.0