refactor: 提取 is_codex_url 到公共 utils 模块

将重复的 Codex URL 判断逻辑提取到 src/utils/url_utils.py,
避免 transport.py 和 adapter.py 中的代码重复。
This commit is contained in:
fawney19
2026-02-05 11:17:00 +08:00
parent bb125a1ddc
commit e72e5370c4
3 changed files with 29 additions and 18 deletions

View File

@@ -16,6 +16,7 @@ from src.api.handlers.base.cli_handler_base import CliMessageHandlerBase
from src.api.handlers.openai.adapter import OpenAIChatAdapter
from src.config.settings import config
from src.core.api_format import ApiFamily
from src.utils.url_utils import is_codex_url
@register_cli_adapter
@@ -75,7 +76,7 @@ class OpenAICliAdapter(CliAdapterBase):
"""
base_url = base_url.rstrip("/")
# Codex OAuth 端点chatgpt.com/backend-api/codex -> /responses
if cls._is_codex_url(base_url):
if is_codex_url(base_url):
return f"{base_url}/responses"
# 标准 OpenAI API
if base_url.endswith("/v1"):
@@ -83,11 +84,6 @@ class OpenAICliAdapter(CliAdapterBase):
else:
return f"{base_url}/v1/responses"
@classmethod
def _is_codex_url(cls, base_url: str) -> bool:
"""判断是否是 Codex OAuth 端点"""
return "/backend-api/codex" in base_url or base_url.endswith("/codex")
# build_request_body 使用基类实现
# OpenAI CLI normalizer 会自动添加 instructions 字段
@@ -101,7 +97,7 @@ class OpenAICliAdapter(CliAdapterBase):
"""构建测试请求体Codex 端点需要强制 stream=true 等特性)"""
from src.api.handlers.base.request_builder import build_test_request_body
target_variant = "codex" if base_url and cls._is_codex_url(base_url) else None
target_variant = "codex" if base_url and is_codex_url(base_url) else None
return build_test_request_body(
cls.FORMAT_ID,
request_data,
@@ -129,7 +125,7 @@ class OpenAICliAdapter(CliAdapterBase):
headers["User-Agent"] = cli_user_agent
# 仅 Codex 端点添加特定头部
if base_url and cls._is_codex_url(base_url):
if base_url and is_codex_url(base_url):
headers["x-oai-web-search-eligible"] = "true"
headers["session_id"] = str(uuid.uuid4())
headers["accept"] = "text/event-stream"

View File

@@ -20,6 +20,7 @@ from src.core.api_format import (
)
from src.core.logger import logger
from src.services.provider.format import normalize_endpoint_signature
from src.utils.url_utils import is_codex_url
if TYPE_CHECKING:
from src.models.database import ProviderAPIKey, ProviderEndpoint
@@ -148,7 +149,7 @@ def build_provider_url(
path = _resolve_default_path(endpoint_sig)
# Codex OAuth 端点chatgpt.com/backend-api/codex使用 /responses 而非 /v1/responses
base_url = getattr(endpoint, "base_url", "") or ""
if endpoint_sig == "openai:cli" and _is_codex_url(base_url):
if endpoint_sig == "openai:cli" and is_codex_url(base_url):
path = "/responses"
if effective_path_params:
try:
@@ -195,15 +196,6 @@ def _resolve_default_path(endpoint_sig: str | None) -> str:
return "/"
def _is_codex_url(base_url: str) -> bool:
"""判断是否是 Codex OAuth 端点(如 chatgpt.com/backend-api/codex
Codex 端点不走标准 /v1 前缀,直接使用 /responses。
"""
url = base_url.rstrip("/")
return "/backend-api/codex" in url or url.endswith("/codex")
# ==============================================================================
# Vertex AI 配置
# ==============================================================================

23
src/utils/url_utils.py Normal file
View File

@@ -0,0 +1,23 @@
"""
URL 处理工具函数
提供 URL 模式检测和处理功能。
"""
from __future__ import annotations
def is_codex_url(base_url: str) -> bool:
"""判断是否是 Codex OAuth 端点。
Codex OAuth 端点(如 chatgpt.com/backend-api/codex不走标准 /v1 前缀,
直接使用 /responses 而非 /v1/responses。
Args:
base_url: 端点基础 URL
Returns:
bool: 是否是 Codex 端点
"""
url = base_url.rstrip("/")
return "/backend-api/codex" in url or url.endswith("/codex")