Token 总数
+-
+未使用
+-
+限流中
+-
+失效
+-
+正常
+-
+总剩余
+-
+普通剩余
+-
+高级剩余
+-
+| + + | +Token | +类型 | +状态 | +普通剩余 | +高级剩余 | +创建时间 | +操作 | +
|---|
暂无数据
+diff --git a/.dockerignore b/.dockerignore
new file mode 100644
index 0000000..a3c2840
--- /dev/null
+++ b/.dockerignore
@@ -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/
\ No newline at end of file
diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml
new file mode 100644
index 0000000..633b413
--- /dev/null
+++ b/.github/workflows/docker.yml
@@ -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
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..fae2ca6
--- /dev/null
+++ b/.gitignore
@@ -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
\ No newline at end of file
diff --git a/Dockerfile b/Dockerfile
new file mode 100644
index 0000000..1992333
--- /dev/null
+++ b/Dockerfile
@@ -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"]
\ No newline at end of file
diff --git a/LICENSE b/LICENSE
index 50bb063..4bd60d2 100644
--- a/LICENSE
+++ b/LICENSE
@@ -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.
\ No newline at end of file
diff --git a/app/api/admin/manage.py b/app/api/admin/manage.py
new file mode 100644
index 0000000..28683f9
--- /dev/null
+++ b/app/api/admin/manage.py
@@ -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"}
+ )
diff --git a/app/api/v1/chat.py b/app/api/v1/chat.py
new file mode 100644
index 0000000..7c2c61d
--- /dev/null
+++ b/app/api/v1/chat.py
@@ -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"
+ }
+ }
+ )
diff --git a/app/api/v1/images.py b/app/api/v1/images.py
new file mode 100644
index 0000000..ef76851
--- /dev/null
+++ b/app/api/v1/images.py
@@ -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))
diff --git a/app/api/v1/models.py b/app/api/v1/models.py
new file mode 100644
index 0000000..f5f1a5a
--- /dev/null
+++ b/app/api/v1/models.py
@@ -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"
+ }
+ }
+ )
diff --git a/app/core/auth.py b/app/core/auth.py
new file mode 100644
index 0000000..fffaa9e
--- /dev/null
+++ b/app/core/auth.py
@@ -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()
\ No newline at end of file
diff --git a/app/core/config.py b/app/core/config.py
new file mode 100644
index 0000000..2a3963e
--- /dev/null
+++ b/app/core/config.py
@@ -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)
\ No newline at end of file
diff --git a/app/core/exception.py b/app/core/exception.py
new file mode 100644
index 0000000..52ac613
--- /dev/null
+++ b/app/core/exception.py
@@ -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)
\ No newline at end of file
diff --git a/app/core/logger.py b/app/core/logger.py
new file mode 100644
index 0000000..4bb282f
--- /dev/null
+++ b/app/core/logger.py
@@ -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()
diff --git a/app/models/grok_models.py b/app/models/grok_models.py
new file mode 100644
index 0000000..53f51f0
--- /dev/null
+++ b/app/models/grok_models.py
@@ -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
\ No newline at end of file
diff --git a/app/models/openai_schema.py b/app/models/openai_schema.py
new file mode 100644
index 0000000..74b721e
--- /dev/null
+++ b/app/models/openai_schema.py
@@ -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="响应选项")
\ No newline at end of file
diff --git a/app/services/grok/client.py b/app/services/grok/client.py
new file mode 100644
index 0000000..2f4443b
--- /dev/null
+++ b/app/services/grok/client.py
@@ -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}")
\ No newline at end of file
diff --git a/app/services/grok/image_cache.py b/app/services/grok/image_cache.py
new file mode 100644
index 0000000..300f11f
--- /dev/null
+++ b/app/services/grok/image_cache.py
@@ -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()
diff --git a/app/services/grok/processer.py b/app/services/grok/processer.py
new file mode 100644
index 0000000..42ab05f
--- /dev/null
+++ b/app/services/grok/processer.py
@@ -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"
+ else:
+ # 缓存失败,使用原始链接
+ content += f"\n"
+ except Exception as e:
+ logger.warning(f"[Processor] 缓存图片失败: {e}")
+ content += f"\n"
+
+ # 返回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"\n"
+ except Exception as e:
+ logger.warning(f"[Processor] 缓存图片失败: {e}")
+ content += f"\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"
Token 总数
+未使用
+限流中
+失效
+正常
+总剩余
+普通剩余
+高级剩余
+| + + | +Token | +类型 | +状态 | +普通剩余 | +高级剩余 | +创建时间 | +操作 | +
|---|
暂无数据
+管理员控制台
+Created By Chenyme © 2025
+