feat: 添加视频生成 API 支持和认证抽象层重构

- 新增 Video Generation API 路由和处理器（支持 Gemini Veo 和 OpenAI Sora 兼容格式）
- 新增 AuthHandler 策略模式，统一 API key 提取逻辑（Bearer/ApiKey/GoogApiKey/OAuth2/QueryKey）
- 新增 RequestContext 三维度检测（数据格式/端点类型/认证方式）
- 新增 EndpointType 和 AuthMethod 枚举
- Gemini/OpenAI normalizer 添加视频格式转换（InternalVideoRequest/Task/PollResult）
- 新增视频任务轮询服务和数据库迁移（video_tasks 表）
- 代码格式化：修复 black 行宽限制，调整 import 排序，target-version 降级至 py313

src/core/api_format/__init__.py

@@ -10,13 +10,26 @@ API 格式核心模块
 - utils.py: 工具函数（is_cli_format, get_base_format 等）
 - detection.py: 格式检测（从请求头、响应内容检测格式）
 """
+
+from src.core.api_format.auth import (
+    ApiKeyAuthHandler,
+    AuthHandler,
+    BearerAuthHandler,
+    GoogApiKeyAuthHandler,
+    OAuth2AuthHandler,
+    QueryKeyAuthHandler,
+    get_auth_handler,
+    get_default_auth_method,
+)
 from src.core.api_format.detection import (
+    RequestContext,
     detect_cli_format_from_path,
     detect_format_and_key_from_starlette,
     detect_format_from_request,
     detect_format_from_response,
+    detect_request_context,
 )
-from src.core.api_format.enums import APIFormat
+from src.core.api_format.enums import APIFormat, AuthMethod, EndpointType
 from src.core.api_format.headers import (
     CORE_REDACT_HEADERS,
     HOP_BY_HOP_HEADERS,
@@ -65,6 +78,8 @@ from src.core.api_format.utils import (
 __all__ = [
     # Enums
     "APIFormat",
+    "AuthMethod",
+    "EndpointType",
     # Metadata
     "ApiFormatDefinition",
     "API_FORMAT_DEFINITIONS",
@@ -111,4 +126,15 @@ __all__ = [
     "detect_format_and_key_from_starlette",
     "detect_format_from_response",
     "detect_cli_format_from_path",
+    "detect_request_context",
+    "RequestContext",
+    # Auth
+    "AuthHandler",
+    "BearerAuthHandler",
+    "ApiKeyAuthHandler",
+    "GoogApiKeyAuthHandler",
+    "OAuth2AuthHandler",
+    "QueryKeyAuthHandler",
+    "get_auth_handler",
+    "get_default_auth_method",
 ]

src/core/api_format/auth.py

@@ -0,0 +1,129 @@
+"""
+认证处理器
+
+将认证逻辑从 API 格式中解耦，支持多种认证方式。
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+from src.core.api_format.enums import APIFormat, AuthMethod
+
+if TYPE_CHECKING:
+    from starlette.requests import Request
+
+
+class AuthHandler(ABC):
+    """认证处理器基类"""
+
+    @abstractmethod
+    def extract_credentials(self, request: Request) -> str | None:
+        """从请求中提取凭证"""
+
+    @abstractmethod
+    def build_headers(self, credentials: str) -> dict[str, str]:
+        """构造上游请求的认证 Header"""
+
+
+class BearerAuthHandler(AuthHandler):
+    """Authorization: Bearer <token>"""
+
+    def extract_credentials(self, request: Request) -> str | None:
+        auth = request.headers.get("authorization", "")
+        if auth.lower().startswith("bearer "):
+            return auth[7:].strip()
+        return None
+
+    def build_headers(self, credentials: str) -> dict[str, str]:
+        return {"Authorization": f"Bearer {credentials}"}
+
+
+class ApiKeyAuthHandler(AuthHandler):
+    """x-api-key: <key>"""
+
+    def extract_credentials(self, request: Request) -> str | None:
+        return request.headers.get("x-api-key")
+
+    def build_headers(self, credentials: str) -> dict[str, str]:
+        return {"x-api-key": credentials}
+
+
+class GoogApiKeyAuthHandler(AuthHandler):
+    """x-goog-api-key: <key> (支持 ?key=)"""
+
+    def extract_credentials(self, request: Request) -> str | None:
+        return request.query_params.get("key") or request.headers.get("x-goog-api-key")
+
+    def build_headers(self, credentials: str) -> dict[str, str]:
+        return {"x-goog-api-key": credentials}
+
+
+class QueryKeyAuthHandler(AuthHandler):
+    """?key= 参数认证（仅提取，通常用于 Gemini）"""
+
+    def extract_credentials(self, request: Request) -> str | None:
+        return request.query_params.get("key")
+
+    def build_headers(self, credentials: str) -> dict[str, str]:
+        return {"x-goog-api-key": credentials}
+
+
+class OAuth2AuthHandler(AuthHandler):
+    """
+    Google OAuth2 / Service Account 认证
+
+    目前使用 Authorization: Bearer 透传 access token。
+    """
+
+    def extract_credentials(self, request: Request) -> str | None:
+        auth = request.headers.get("authorization", "")
+        if auth.lower().startswith("bearer "):
+            return auth[7:].strip()
+        return None
+
+    def build_headers(self, credentials: str) -> dict[str, str]:
+        return {"Authorization": f"Bearer {credentials}"}
+
+
+_AUTH_HANDLERS: dict[AuthMethod, AuthHandler] = {
+    AuthMethod.BEARER: BearerAuthHandler(),
+    AuthMethod.API_KEY: ApiKeyAuthHandler(),
+    AuthMethod.GOOG_API_KEY: GoogApiKeyAuthHandler(),
+    AuthMethod.OAUTH2: OAuth2AuthHandler(),
+    AuthMethod.QUERY_KEY: QueryKeyAuthHandler(),
+}
+
+
+def get_auth_handler(auth_method: AuthMethod) -> AuthHandler:
+    """获取认证处理器实例"""
+    handler = _AUTH_HANDLERS.get(auth_method)
+    if not handler:
+        raise ValueError(f"Unsupported auth method: {auth_method}")
+    return handler
+
+
+def get_default_auth_method(api_format: APIFormat) -> AuthMethod:
+    """从 APIFormat 推断默认 AuthMethod（兼容旧逻辑）"""
+    mapping = {
+        APIFormat.OPENAI: AuthMethod.BEARER,
+        APIFormat.OPENAI_CLI: AuthMethod.BEARER,
+        APIFormat.CLAUDE: AuthMethod.API_KEY,
+        APIFormat.CLAUDE_CLI: AuthMethod.BEARER,
+        APIFormat.GEMINI: AuthMethod.GOOG_API_KEY,
+        APIFormat.GEMINI_CLI: AuthMethod.GOOG_API_KEY,
+    }
+    return mapping.get(api_format, AuthMethod.BEARER)
+
+
+__all__ = [
+    "AuthHandler",
+    "BearerAuthHandler",
+    "ApiKeyAuthHandler",
+    "GoogApiKeyAuthHandler",
+    "OAuth2AuthHandler",
+    "QueryKeyAuthHandler",
+    "get_auth_handler",
+    "get_default_auth_method",
+]

src/core/api_format/conversion/internal_video.py

@@ -0,0 +1,84 @@
+"""
+视频格式转换内部表示（Internal Video Format）
+
+用于 Video API 的 Hub-and-Spoke 统一中间表示。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from enum import Enum
+from typing import Any
+
+
+class VideoStatus(str, Enum):
+    PENDING = "pending"
+    SUBMITTED = "submitted"
+    QUEUED = "queued"
+    PROCESSING = "processing"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+    EXPIRED = "expired"
+
+
+@dataclass
+class InternalVideoRequest:
+    """统一的视频生成请求格式"""
+
+    prompt: str
+    model: str = "sora-2"
+    duration_seconds: int = 4
+    aspect_ratio: str = "16:9"
+    resolution: str = "720p"
+    reference_image_url: str | None = None  # base64 或 URL
+    character_ids: list[str] = field(default_factory=list)
+    extra: dict[str, Any] = field(default_factory=dict)
+    preferred_provider: str | None = None
+    preferred_format: str | None = None
+
+
+@dataclass
+class InternalVideoTask:
+    """统一的视频任务状态"""
+
+    id: str
+    external_id: str | None = None
+    status: VideoStatus = VideoStatus.PENDING
+    progress_percent: int = 0
+    progress_message: str | None = None
+    video_url: str | None = None
+    video_urls: list[str] = field(default_factory=list)
+    thumbnail_url: str | None = None
+    video_duration_seconds: int | None = None
+    video_size_bytes: int | None = None
+    created_at: datetime | None = None
+    completed_at: datetime | None = None
+    expires_at: datetime | None = None
+    error_code: str | None = None
+    error_message: str | None = None
+    original_request: InternalVideoRequest | None = None
+    extra: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class InternalVideoPollResult:
+    """轮询结果"""
+
+    status: VideoStatus
+    progress_percent: int = 0
+    video_url: str | None = None
+    video_urls: list[str] = field(default_factory=list)
+    expires_at: datetime | None = None
+    error_code: str | None = None
+    error_message: str | None = None
+    raw_response: dict[str, Any] | None = None
+
+
+__all__ = [
+    "VideoStatus",
+    "InternalVideoRequest",
+    "InternalVideoTask",
+    "InternalVideoPollResult",
+]

src/core/api_format/conversion/normalizer.py

@@ -5,11 +5,11 @@
 再从 internal 输出到目标格式。
 """
 
-
 from abc import ABC, abstractmethod
 from typing import Any
 
 from .internal import FormatCapabilities, InternalError, InternalRequest, InternalResponse
+from .internal_video import InternalVideoPollResult, InternalVideoRequest, InternalVideoTask
 from .stream_events import InternalStreamEvent
 from .stream_state import StreamState
 
@@ -88,8 +88,29 @@ class FormatNormalizer(ABC):
         """将内部错误表示转换为格式特定错误"""
         raise NotImplementedError
 
+    # ============ 视频转换（可选） ============
+
+    def video_request_to_internal(self, request: dict[str, Any]) -> InternalVideoRequest:
+        """将视频请求转换为内部表示"""
+        raise NotImplementedError(f"{self.__class__.__name__} does not support video conversion")
+
+    def video_request_from_internal(self, internal: InternalVideoRequest) -> dict[str, Any]:
+        """将内部视频请求转换为格式特定请求"""
+        raise NotImplementedError(f"{self.__class__.__name__} does not support video conversion")
+
+    def video_task_to_internal(self, response: dict[str, Any]) -> InternalVideoTask:
+        """将视频任务响应转换为内部表示"""
+        raise NotImplementedError(f"{self.__class__.__name__} does not support video conversion")
+
+    def video_task_from_internal(self, internal: InternalVideoTask) -> dict[str, Any]:
+        """将内部视频任务转换为格式特定响应"""
+        raise NotImplementedError(f"{self.__class__.__name__} does not support video conversion")
+
+    def video_poll_to_internal(self, response: dict[str, Any]) -> InternalVideoPollResult:
+        """将视频轮询响应转换为内部表示"""
+        raise NotImplementedError(f"{self.__class__.__name__} does not support video conversion")
+
 
 __all__ = [
     "FormatNormalizer",
 ]
-

src/core/api_format/conversion/normalizers/gemini.py

@@ -1,4 +1,4 @@
-""" 
+"""
 Gemini (GenerateContent / streamGenerateContent) Normalizer
 
 负责：
@@ -11,7 +11,6 @@ Gemini (GenerateContent / streamGenerateContent) Normalizer
 - 响应/流式通常为 camelCase（candidates/finishReason/usageMetadata/modelVersion）。
 """
 
-
 import json
 from typing import Any
 
@@ -43,6 +42,12 @@ from src.core.api_format.conversion.internal import (
     UnknownBlock,
     UsageInfo,
 )
+from src.core.api_format.conversion.internal_video import (
+    InternalVideoPollResult,
+    InternalVideoRequest,
+    InternalVideoTask,
+    VideoStatus,
+)
 from src.core.api_format.conversion.normalizer import FormatNormalizer
 from src.core.api_format.conversion.stream_events import (
     ContentBlockStartEvent,
@@ -111,7 +116,9 @@ class GeminiNormalizer(FormatNormalizer):
         if isinstance(contents, list):
             for content in contents:
                 if not isinstance(content, dict):
-                    dropped["gemini_content_non_dict"] = dropped.get("gemini_content_non_dict", 0) + 1
+                    dropped["gemini_content_non_dict"] = (
+                        dropped.get("gemini_content_non_dict", 0) + 1
+                    )
                     continue
                 imsg, md = self._content_to_internal_message(content)
                 self._merge_dropped(dropped, md)
@@ -128,9 +135,7 @@ class GeminiNormalizer(FormatNormalizer):
             else None
         )
         temperature = self._optional_float(
-            generation_config.get("temperature")
-            if isinstance(generation_config, dict)
-            else None
+            generation_config.get("temperature") if isinstance(generation_config, dict) else None
         )
         top_p = self._optional_float(
             generation_config.get("top_p") if isinstance(generation_config, dict) else None
@@ -144,9 +149,7 @@ class GeminiNormalizer(FormatNormalizer):
 
         tools = self._gemini_tools_to_internal(request.get("tools"))
         tool_choice = self._gemini_tool_config_to_tool_choice(
-            request.get("tool_config")
-            if "tool_config" in request
-            else request.get("toolConfig")
+            request.get("tool_config") if "tool_config" in request else request.get("toolConfig")
         )
 
         # 构建 extra，保留原始 gemini 字段
@@ -253,9 +256,15 @@ class GeminiNormalizer(FormatNormalizer):
             orig_gc = gemini_extra.get("generation_config") or gemini_extra.get("generationConfig")
             if isinstance(orig_gc, dict):
                 # responseModalities
-                if "responseModalities" in orig_gc and "responseModalities" not in generation_config:
+                if (
+                    "responseModalities" in orig_gc
+                    and "responseModalities" not in generation_config
+                ):
                     generation_config["responseModalities"] = orig_gc["responseModalities"]
-                if "response_modalities" in orig_gc and "responseModalities" not in generation_config:
+                if (
+                    "response_modalities" in orig_gc
+                    and "responseModalities" not in generation_config
+                ):
                     generation_config["responseModalities"] = orig_gc["response_modalities"]
                 # thinkingConfig
                 if "thinkingConfig" in orig_gc and "thinkingConfig" not in generation_config:
@@ -370,14 +379,19 @@ class GeminiNormalizer(FormatNormalizer):
 
         finish_reason = None
         if internal.stop_reason is not None:
-            finish_reason = STOP_REASON_MAPPINGS.get("GEMINI", {}).get(internal.stop_reason.value, "OTHER")
+            finish_reason = STOP_REASON_MAPPINGS.get("GEMINI", {}).get(
+                internal.stop_reason.value, "OTHER"
+            )
 
         usage_metadata: dict[str, Any] = {}
         if internal.usage:
             usage_metadata = {
                 "promptTokenCount": int(internal.usage.input_tokens),
                 "candidatesTokenCount": int(internal.usage.output_tokens),
-                "totalTokenCount": int(internal.usage.total_tokens or (internal.usage.input_tokens + internal.usage.output_tokens)),
+                "totalTokenCount": int(
+                    internal.usage.total_tokens
+                    or (internal.usage.input_tokens + internal.usage.output_tokens)
+                ),
             }
             if internal.usage.cache_read_tokens:
                 usage_metadata["cachedContentTokenCount"] = int(internal.usage.cache_read_tokens)
@@ -410,7 +424,9 @@ class GeminiNormalizer(FormatNormalizer):
     # Streaming
     # =========================
 
-    def stream_chunk_to_internal(self, chunk: dict[str, Any], state: StreamState) -> list[InternalStreamEvent]:
+    def stream_chunk_to_internal(
+        self, chunk: dict[str, Any], state: StreamState
+    ) -> list[InternalStreamEvent]:
         ss = state.substate(self.FORMAT_ID)
         events: list[InternalStreamEvent] = []
 
@@ -457,7 +473,9 @@ class GeminiNormalizer(FormatNormalizer):
                     if delta:
                         if not ss.get("text_block_started"):
                             ss["text_block_started"] = True
-                            events.append(ContentBlockStartEvent(block_index=0, block_type=ContentType.TEXT))
+                            events.append(
+                                ContentBlockStartEvent(block_index=0, block_type=ContentType.TEXT)
+                            )
                         events.append(ContentDeltaEvent(block_index=0, text_delta=delta))
                     continue
 
@@ -500,7 +518,9 @@ class GeminiNormalizer(FormatNormalizer):
                     inline_data = part.get("inline_data")
 
                 if isinstance(inline_data, dict):
-                    mime_type = str(inline_data.get("mimeType") or inline_data.get("mime_type") or "").strip()
+                    mime_type = str(
+                        inline_data.get("mimeType") or inline_data.get("mime_type") or ""
+                    ).strip()
                     data = str(inline_data.get("data") or "").strip()
 
                     # 确保 mime_type 和 data 都非空
@@ -635,7 +655,9 @@ class GeminiNormalizer(FormatNormalizer):
         if isinstance(event, MessageStopEvent):
             finish_reason = None
             if event.stop_reason is not None:
-                finish_reason = STOP_REASON_MAPPINGS.get("GEMINI", {}).get(event.stop_reason.value, "OTHER")
+                finish_reason = STOP_REASON_MAPPINGS.get("GEMINI", {}).get(
+                    event.stop_reason.value, "OTHER"
+                )
 
             chunk: dict[str, Any] = base_chunk([])
             if finish_reason is not None:
@@ -645,10 +667,15 @@ class GeminiNormalizer(FormatNormalizer):
                 chunk["usageMetadata"] = {
                     "promptTokenCount": int(event.usage.input_tokens),
                     "candidatesTokenCount": int(event.usage.output_tokens),
-                    "totalTokenCount": int(event.usage.total_tokens or (event.usage.input_tokens + event.usage.output_tokens)),
+                    "totalTokenCount": int(
+                        event.usage.total_tokens
+                        or (event.usage.input_tokens + event.usage.output_tokens)
+                    ),
                 }
                 if event.usage.cache_read_tokens:
-                    chunk["usageMetadata"]["cachedContentTokenCount"] = int(event.usage.cache_read_tokens)
+                    chunk["usageMetadata"]["cachedContentTokenCount"] = int(
+                        event.usage.cache_read_tokens
+                    )
 
             out.append(chunk)
             return out
@@ -698,11 +725,171 @@ class GeminiNormalizer(FormatNormalizer):
         }
         return {"error": payload}
 
+    # =========================
+    # Video conversion
+    # =========================
+
+    def video_request_to_internal(self, request: dict[str, Any]) -> InternalVideoRequest:
+        instances = request.get("instances")
+        if not instances or not isinstance(instances, list) or len(instances) == 0:
+            raise ValueError("Video request requires at least one instance")
+        instance = instances[0] if isinstance(instances[0], dict) else {}
+        params = request.get("parameters") or {}
+
+        image = instance.get("image") if isinstance(instance, dict) else None
+        image_ref = None
+        if isinstance(image, dict):
+            image_ref = image.get("bytesBase64Encoded")
+
+        prompt = instance.get("prompt") if isinstance(instance, dict) else None
+        prompt_str = str(prompt).strip() if prompt else ""
+        if not prompt_str:
+            raise ValueError("Video prompt is required")
+
+        duration_raw = params.get("durationSeconds")
+        sample_count_raw = params.get("sampleCount")
+
+        try:
+            duration_seconds = int(duration_raw) if duration_raw else 8
+        except (ValueError, TypeError):
+            duration_seconds = 8
+
+        # 安全解析 sampleCount
+        try:
+            sample_count = int(sample_count_raw) if sample_count_raw else 1
+        except (ValueError, TypeError):
+            sample_count = 1
+
+        return InternalVideoRequest(
+            prompt=prompt_str,
+            model=str(request.get("model") or "veo-3.1-generate-preview"),
+            duration_seconds=duration_seconds,
+            aspect_ratio=str(params.get("aspectRatio") or "16:9"),
+            resolution=str(params.get("resolution") or "720p"),
+            reference_image_url=image_ref,
+            extra={
+                "personGeneration": params.get("personGeneration"),
+                "sampleCount": sample_count,
+            },
+        )
+
+    def video_request_from_internal(self, internal: InternalVideoRequest) -> dict[str, Any]:
+        instance: dict[str, Any] = {"prompt": internal.prompt}
+        if internal.reference_image_url:
+            instance["image"] = {"bytesBase64Encoded": internal.reference_image_url}
+
+        parameters: dict[str, Any] = {
+            "aspectRatio": internal.aspect_ratio,
+            "resolution": internal.resolution,
+            "durationSeconds": internal.duration_seconds,
+        }
+        for key in ["personGeneration", "sampleCount"]:
+            if key in internal.extra:
+                parameters[key] = internal.extra[key]
+
+        return {
+            "instances": [instance],
+            "parameters": parameters,
+        }
+
+    def video_task_to_internal(self, response: dict[str, Any]) -> InternalVideoTask:
+        operation_name = str(response.get("name") or "")
+        done = bool(response.get("done"))
+
+        if done:
+            video_response = response.get("response", {}).get("generateVideoResponse", {})
+            samples = video_response.get("generatedSamples", [])
+            video_urls = [
+                s.get("video", {}).get("uri")
+                for s in samples
+                if isinstance(s, dict) and s.get("video", {}).get("uri")
+            ]
+            return InternalVideoTask(
+                id=operation_name.replace("operations/", ""),
+                external_id=operation_name,
+                status=VideoStatus.COMPLETED,
+                progress_percent=100,
+                video_url=video_urls[0] if video_urls else None,
+                video_urls=video_urls,
+                extra={"raw_response": video_response},
+            )
+
+        metadata = response.get("metadata", {})
+        return InternalVideoTask(
+            id=operation_name.replace("operations/", ""),
+            external_id=operation_name,
+            status=VideoStatus.PROCESSING,
+            progress_percent=50,
+            extra={"metadata": metadata},
+        )
+
+    def video_task_from_internal(self, internal: InternalVideoTask) -> dict[str, Any]:
+        # 优先使用 external_id（上游返回的 operation name），否则用内部 id
+        operation_name = internal.external_id or f"operations/{internal.id}"
+        if not operation_name.startswith("operations/"):
+            operation_name = f"operations/{operation_name}"
+
+        if internal.status == VideoStatus.COMPLETED:
+            urls = internal.video_urls or ([internal.video_url] if internal.video_url else [])
+            return {
+                "name": operation_name,
+                "done": True,
+                "response": {
+                    "generateVideoResponse": {
+                        "generatedSamples": [
+                            {"video": {"uri": url, "mimeType": "video/mp4"}} for url in urls
+                        ]
+                    }
+                },
+            }
+
+        return {
+            "name": operation_name,
+            "done": False,
+            "metadata": internal.extra.get("metadata", {}),
+        }
+
+    def video_poll_to_internal(self, response: dict[str, Any]) -> InternalVideoPollResult:
+        done = bool(response.get("done"))
+        if done:
+            error = response.get("error")
+            if error:
+                return InternalVideoPollResult(
+                    status=VideoStatus.FAILED,
+                    error_code=str(error.get("code", "unknown")),
+                    error_message=error.get("message", "Unknown error"),
+                    raw_response=response,
+                )
+
+            video_response = response.get("response", {}).get("generateVideoResponse", {})
+            samples = video_response.get("generatedSamples", [])
+            video_urls = [
+                s.get("video", {}).get("uri")
+                for s in samples
+                if isinstance(s, dict) and s.get("video", {}).get("uri")
+            ]
+            video_url = video_urls[0] if video_urls else None
+            return InternalVideoPollResult(
+                status=VideoStatus.COMPLETED,
+                progress_percent=100,
+                video_url=video_url,
+                video_urls=video_urls,
+                raw_response=response,
+            )
+
+        return InternalVideoPollResult(
+            status=VideoStatus.PROCESSING,
+            progress_percent=50,
+            raw_response=response,
+        )
+
     # =========================
     # Helpers
     # =========================
 
-    def _content_to_internal_message(self, content: dict[str, Any]) -> tuple[InternalMessage | None, dict[str, int]]:
+    def _content_to_internal_message(
+        self, content: dict[str, Any]
+    ) -> tuple[InternalMessage | None, dict[str, int]]:
         dropped: dict[str, int] = {}
 
         role_raw = str(content.get("role") or "user")
@@ -749,12 +936,16 @@ class GeminiNormalizer(FormatNormalizer):
             if inline is None:
                 inline = part.get("inlineData")
             if isinstance(inline, dict):
-                mime_type = inline.get("mime_type") if "mime_type" in inline else inline.get("mimeType")
+                mime_type = (
+                    inline.get("mime_type") if "mime_type" in inline else inline.get("mimeType")
+                )
                 data = inline.get("data")
                 if isinstance(mime_type, str) and mime_type and isinstance(data, str) and data:
                     blocks.append(ImageBlock(data=data, media_type=mime_type))
                 else:
-                    dropped["gemini_inline_data_invalid"] = dropped.get("gemini_inline_data_invalid", 0) + 1
+                    dropped["gemini_inline_data_invalid"] = (
+                        dropped.get("gemini_inline_data_invalid", 0) + 1
+                    )
                     blocks.append(UnknownBlock(raw_type="inline_data", payload=part))
                 continue
 
@@ -859,7 +1050,9 @@ class GeminiNormalizer(FormatNormalizer):
 
         return {"role": role, "parts": parts}
 
-    def _collapse_system_instruction(self, system_instruction: Any) -> tuple[str | None, dict[str, int]]:
+    def _collapse_system_instruction(
+        self, system_instruction: Any
+    ) -> tuple[str | None, dict[str, int]]:
         dropped: dict[str, int] = {}
         if system_instruction is None:
             return None, dropped
@@ -875,12 +1068,18 @@ class GeminiNormalizer(FormatNormalizer):
                 joined = "".join(texts)
                 return (joined or None), dropped
 
-        dropped["gemini_system_instruction_unsupported"] = dropped.get("gemini_system_instruction_unsupported", 0) + 1
+        dropped["gemini_system_instruction_unsupported"] = (
+            dropped.get("gemini_system_instruction_unsupported", 0) + 1
+        )
         return None, dropped
 
     def _get_generation_config(self, request: dict[str, Any]) -> dict[str, Any]:
         # 兼容 snake_case 与 camelCase
-        gc = request.get("generation_config") if "generation_config" in request else request.get("generationConfig")
+        gc = (
+            request.get("generation_config")
+            if "generation_config" in request
+            else request.get("generationConfig")
+        )
         if not isinstance(gc, dict):
             return {}
 
@@ -936,8 +1135,16 @@ class GeminiNormalizer(FormatNormalizer):
                     ToolDefinition(
                         name=name,
                         description=decl.get("description"),
-                        parameters=decl.get("parameters") if isinstance(decl.get("parameters"), dict) else None,
-                        extra={"gemini_function_declaration": self._extract_extra(decl, {"name", "description", "parameters"})},
+                        parameters=(
+                            decl.get("parameters")
+                            if isinstance(decl.get("parameters"), dict)
+                            else None
+                        ),
+                        extra={
+                            "gemini_function_declaration": self._extract_extra(
+                                decl, {"name", "description", "parameters"}
+                            )
+                        },
                     )
                 )
 
@@ -966,7 +1173,11 @@ class GeminiNormalizer(FormatNormalizer):
         if mode in ("ANY", "REQUIRED"):
             return ToolChoice(type=ToolChoiceType.REQUIRED, extra={"gemini": tool_config})
         if isinstance(allowed, list) and len(allowed) == 1:
-            return ToolChoice(type=ToolChoiceType.TOOL, tool_name=str(allowed[0] or ""), extra={"gemini": tool_config})
+            return ToolChoice(
+                type=ToolChoiceType.TOOL,
+                tool_name=str(allowed[0] or ""),
+                extra={"gemini": tool_config},
+            )
 
         return ToolChoice(type=ToolChoiceType.AUTO, extra={"gemini": tool_config})
 
@@ -1010,7 +1221,9 @@ class GeminiNormalizer(FormatNormalizer):
                 pass
 
         if "total_tokens" not in fields:
-            fields["total_tokens"] = int(fields.get("input_tokens", 0) + fields.get("output_tokens", 0))
+            fields["total_tokens"] = int(
+                fields.get("input_tokens", 0) + fields.get("output_tokens", 0)
+            )
 
         return UsageInfo(
             input_tokens=int(fields.get("input_tokens", 0)),

src/core/api_format/conversion/normalizers/openai.py

@@ -7,12 +7,11 @@ OpenAI Chat Completions Normalizer
 - 可选：OpenAI error <-> InternalError
 """
 
-
 import json
 import time
+from datetime import datetime, timezone
 from typing import Any
 
-from src.core.logger import logger
 from src.core.api_format.conversion.field_mappings import (
     ERROR_TYPE_MAPPINGS,
     RETRYABLE_ERROR_TYPES,
@@ -39,6 +38,12 @@ from src.core.api_format.conversion.internal import (
     UnknownBlock,
     UsageInfo,
 )
+from src.core.api_format.conversion.internal_video import (
+    InternalVideoPollResult,
+    InternalVideoRequest,
+    InternalVideoTask,
+    VideoStatus,
+)
 from src.core.api_format.conversion.normalizer import FormatNormalizer
 from src.core.api_format.conversion.stream_events import (
     ContentBlockStartEvent,
@@ -51,6 +56,7 @@ from src.core.api_format.conversion.stream_events import (
     ToolCallDeltaEvent,
 )
 from src.core.api_format.conversion.stream_state import StreamState
+from src.core.logger import logger
 
 
 class OpenAINormalizer(FormatNormalizer):
@@ -81,6 +87,28 @@ class OpenAINormalizer(FormatNormalizer):
         StopReason.UNKNOWN: "stop",
     }
 
+    # 视频尺寸映射: (resolution, aspect_ratio) -> size
+    _VIDEO_SIZE_MAP: dict[tuple[str, str], str] = {
+        ("480p", "16:9"): "854x480",
+        ("480p", "9:16"): "480x854",
+        ("480p", "1:1"): "480x480",
+        ("720p", "16:9"): "1280x720",
+        ("720p", "9:16"): "720x1280",
+        ("720p", "1:1"): "720x720",
+        ("1080p", "16:9"): "1920x1080",
+        ("1080p", "9:16"): "1080x1920",
+        ("1080p", "1:1"): "1080x1080",
+    }
+    # OpenAI Sora 特定尺寸（非标准分辨率，需单独处理）
+    _SORA_SIZE_REVERSE: dict[str, tuple[str, str]] = {
+        "1792x1024": ("1080p", "16:9"),
+        "1024x1792": ("1080p", "9:16"),
+    }
+    _VIDEO_SIZE_REVERSE: dict[str, tuple[str, str]] = {
+        **{value: key for key, value in _VIDEO_SIZE_MAP.items()},
+        **_SORA_SIZE_REVERSE,
+    }
+
     # InternalError.type -> OpenAI error.type（最佳努力）
     _ERROR_TYPE_TO_OPENAI: dict[ErrorType, str] = {
         ErrorType.INVALID_REQUEST: "invalid_request_error",
@@ -139,9 +167,7 @@ class OpenAINormalizer(FormatNormalizer):
 
         # 兼容新旧参数名：优先使用 max_completion_tokens，回退到 max_tokens
         mct = request.get("max_completion_tokens")
-        max_tokens_value = self._optional_int(
-            mct if mct is not None else request.get("max_tokens")
-        )
+        max_tokens_value = self._optional_int(mct if mct is not None else request.get("max_tokens"))
 
         # 构建 extra，保留未识别字段
         extra: dict[str, Any] = {"openai": self._extract_extra(request, {"messages"})}
@@ -304,7 +330,9 @@ class OpenAINormalizer(FormatNormalizer):
             message["content"] = content_value
 
         if tool_blocks:
-            message["tool_calls"] = [self._tool_use_block_to_openai_call(b, idx) for idx, b in enumerate(tool_blocks)]
+            message["tool_calls"] = [
+                self._tool_use_block_to_openai_call(b, idx) for idx, b in enumerate(tool_blocks)
+            ]
 
         finish_reason = None
         if internal.stop_reason is not None:
@@ -334,7 +362,9 @@ class OpenAINormalizer(FormatNormalizer):
     # Streaming
     # =========================
 
-    def stream_chunk_to_internal(self, chunk: dict[str, Any], state: StreamState) -> list[InternalStreamEvent]:
+    def stream_chunk_to_internal(
+        self, chunk: dict[str, Any], state: StreamState
+    ) -> list[InternalStreamEvent]:
         ss = state.substate(self.FORMAT_ID)
         events: list[InternalStreamEvent] = []
 
@@ -393,7 +423,9 @@ class OpenAINormalizer(FormatNormalizer):
                 tc_name = str(fn.get("name") or "")
                 tc_args = fn.get("arguments")
 
-                block_index = self._ensure_tool_block_index(ss, tc_id or str(tool_call.get("index") or ""))
+                block_index = self._ensure_tool_block_index(
+                    ss, tc_id or str(tool_call.get("index") or "")
+                )
 
                 # tool start（只在首次见到该 tool_id 时发）
                 started_key = f"tool_started:{block_index}"
@@ -493,7 +525,12 @@ class OpenAINormalizer(FormatNormalizer):
             image_data = event.extra.get("image_data")
             image_media_type = event.extra.get("image_media_type")
             # 确保图片数据有效（base64 数据至少有一定长度）
-            if image_data and image_media_type and isinstance(image_data, str) and len(image_data) > 10:
+            if (
+                image_data
+                and image_media_type
+                and isinstance(image_data, str)
+                and len(image_data) > 10
+            ):
                 # 构造 data URL 格式的图片
                 data_url = f"data:{image_media_type};base64,{image_data}"
                 # 存储图片数据，在 ContentBlockStopEvent 时输出
@@ -602,11 +639,171 @@ class OpenAINormalizer(FormatNormalizer):
             payload["param"] = internal.param
         return {"error": payload}
 
+    # =========================
+    # Video conversion
+    # =========================
+
+    def video_request_to_internal(self, request: dict[str, Any]) -> InternalVideoRequest:
+        prompt = str(request.get("prompt") or "").strip()
+        if not prompt:
+            raise ValueError("Video prompt is required")
+
+        size = str(request.get("size") or "720x1280")
+        resolution, aspect_ratio = self._VIDEO_SIZE_REVERSE.get(size, ("720p", "9:16"))
+
+        input_reference = request.get("input_reference")
+        reference_url = str(input_reference) if input_reference else None
+
+        # 安全解析 seconds 字段
+        seconds_raw = request.get("seconds")
+        try:
+            duration_seconds = int(seconds_raw) if seconds_raw else 4
+        except (ValueError, TypeError):
+            duration_seconds = 4
+
+        return InternalVideoRequest(
+            prompt=prompt,
+            model=str(request.get("model") or "sora-2"),
+            duration_seconds=duration_seconds,
+            resolution=resolution,
+            aspect_ratio=aspect_ratio,
+            character_ids=request.get("character_ids") or [],
+            reference_image_url=reference_url,
+            extra={"original_size": size},
+        )
+
+    def video_request_from_internal(self, internal: InternalVideoRequest) -> dict[str, Any]:
+        size = self._VIDEO_SIZE_MAP.get((internal.resolution, internal.aspect_ratio), "720x1280")
+        payload: dict[str, Any] = {
+            "prompt": internal.prompt,
+            "model": internal.model,
+            "seconds": internal.duration_seconds,
+            "size": size,
+            "character_ids": internal.character_ids,
+        }
+        if internal.reference_image_url:
+            payload["input_reference"] = internal.reference_image_url
+        return payload
+
+    def video_task_to_internal(self, response: dict[str, Any]) -> InternalVideoTask:
+        status_map = {
+            "queued": VideoStatus.QUEUED,
+            "processing": VideoStatus.PROCESSING,
+            "completed": VideoStatus.COMPLETED,
+            "failed": VideoStatus.FAILED,
+        }
+        status = status_map.get(str(response.get("status") or ""), VideoStatus.PENDING)
+
+        error = response.get("error") or {}
+        error_code = error.get("code") if isinstance(error, dict) else None
+        error_message = error.get("message") if isinstance(error, dict) else None
+
+        created_at = response.get("created_at")
+        completed_at = response.get("completed_at")
+        expires_at = response.get("expires_at")
+
+        return InternalVideoTask(
+            id=str(response.get("id") or ""),
+            status=status,
+            progress_percent=int(response.get("progress") or 0),
+            created_at=datetime.fromtimestamp(created_at, tz=timezone.utc) if created_at else None,
+            completed_at=(
+                datetime.fromtimestamp(completed_at, tz=timezone.utc) if completed_at else None
+            ),
+            expires_at=datetime.fromtimestamp(expires_at, tz=timezone.utc) if expires_at else None,
+            error_code=error_code,
+            error_message=error_message,
+            extra={
+                "object": response.get("object"),
+                "model": response.get("model"),
+                "size": response.get("size"),
+                "seconds": response.get("seconds"),
+            },
+        )
+
+    def video_task_from_internal(self, internal: InternalVideoTask) -> dict[str, Any]:
+        status_map = {
+            VideoStatus.PENDING: "queued",
+            VideoStatus.SUBMITTED: "queued",
+            VideoStatus.QUEUED: "queued",
+            VideoStatus.PROCESSING: "processing",
+            VideoStatus.COMPLETED: "completed",
+            VideoStatus.FAILED: "failed",
+            VideoStatus.CANCELLED: "failed",
+            VideoStatus.EXPIRED: "failed",
+        }
+        payload: dict[str, Any] = {
+            "id": internal.id,
+            "object": "video",
+            "status": status_map.get(internal.status, "queued"),
+            "progress": internal.progress_percent,
+        }
+
+        if internal.created_at:
+            payload["created_at"] = int(internal.created_at.timestamp())
+        if internal.completed_at:
+            payload["completed_at"] = int(internal.completed_at.timestamp())
+        if internal.expires_at:
+            payload["expires_at"] = int(internal.expires_at.timestamp())
+        if internal.error_code:
+            payload["error"] = {
+                "code": internal.error_code,
+                "message": internal.error_message,
+            }
+
+        for key in ["model", "size", "seconds"]:
+            if key in internal.extra:
+                payload[key] = internal.extra[key]
+
+        return payload
+
+    def video_poll_to_internal(self, response: dict[str, Any]) -> InternalVideoPollResult:
+        status = str(response.get("status") or "")
+        task_id = response.get("id")
+
+        if status == "completed":
+            expires_at = response.get("expires_at")
+            # 使用任务 ID 构建内容路径，由调用方拼接完整 URL
+            # 如果 task_id 不存在，说明上游响应异常
+            video_url = f"videos/{task_id}/content" if task_id else None
+            if not video_url:
+                return InternalVideoPollResult(
+                    status=VideoStatus.FAILED,
+                    error_code="missing_task_id",
+                    error_message="Upstream response missing task id",
+                    raw_response=response,
+                )
+            return InternalVideoPollResult(
+                status=VideoStatus.COMPLETED,
+                progress_percent=100,
+                video_url=video_url,
+                expires_at=(
+                    datetime.fromtimestamp(expires_at, tz=timezone.utc) if expires_at else None
+                ),
+                raw_response=response,
+            )
+        if status == "failed":
+            error = response.get("error") or {}
+            return InternalVideoPollResult(
+                status=VideoStatus.FAILED,
+                error_code=error.get("code") if isinstance(error, dict) else None,
+                error_message=error.get("message") if isinstance(error, dict) else None,
+                raw_response=response,
+            )
+
+        return InternalVideoPollResult(
+            status=VideoStatus.PROCESSING,
+            progress_percent=int(response.get("progress") or 0),
+            raw_response=response,
+        )
+
     # =========================
     # Helpers
     # =========================
 
-    def _openai_message_to_internal(self, msg: dict[str, Any]) -> tuple[InternalMessage | None, dict[str, int]]:
+    def _openai_message_to_internal(
+        self, msg: dict[str, Any]
+    ) -> tuple[InternalMessage | None, dict[str, int]]:
         dropped: dict[str, int] = {}
 
         role_raw = str(msg.get("role") or "unknown")
@@ -620,7 +817,11 @@ class OpenAINormalizer(FormatNormalizer):
             if tr_block is None:
                 return None, dropped
             return (
-                InternalMessage(role=Role.USER, content=[tr_block], extra=self._extract_extra(msg, {"role", "content"})),
+                InternalMessage(
+                    role=Role.USER,
+                    content=[tr_block],
+                    extra=self._extract_extra(msg, {"role", "content"}),
+                ),
                 dropped,
             )
 
@@ -668,24 +869,34 @@ class OpenAINormalizer(FormatNormalizer):
         blocks: list[ContentBlock] = []
         for part in content:
             if not isinstance(part, dict):
-                dropped["openai_content_part_non_dict"] = dropped.get("openai_content_part_non_dict", 0) + 1
+                dropped["openai_content_part_non_dict"] = (
+                    dropped.get("openai_content_part_non_dict", 0) + 1
+                )
                 continue
 
             ptype = str(part.get("type") or "unknown")
             if ptype == "text":
                 text = str(part.get("text") or "")
                 if text:
-                    blocks.append(TextBlock(text=text, extra=self._extract_extra(part, {"type", "text"})))
+                    blocks.append(
+                        TextBlock(text=text, extra=self._extract_extra(part, {"type", "text"}))
+                    )
                 continue
 
             if ptype == "image_url":
-                url = (part.get("image_url") or {}).get("url") if isinstance(part.get("image_url"), dict) else None
+                url = (
+                    (part.get("image_url") or {}).get("url")
+                    if isinstance(part.get("image_url"), dict)
+                    else None
+                )
                 if isinstance(url, str) and url:
                     img = self._image_url_to_block(url)
                     img.extra.update(self._extract_extra(part, {"type", "image_url"}))
                     blocks.append(img)
                 else:
-                    dropped["openai_image_url_missing"] = dropped.get("openai_image_url_missing", 0) + 1
+                    dropped["openai_image_url_missing"] = (
+                        dropped.get("openai_image_url_missing", 0) + 1
+                    )
                     blocks.append(UnknownBlock(raw_type="image_url", payload=part))
                 continue
 
@@ -731,7 +942,9 @@ class OpenAINormalizer(FormatNormalizer):
                     parameters=params_raw if isinstance(params_raw, dict) else None,
                     extra={
                         "openai_tool": self._extract_extra(tool, {"type", "function"}),
-                        "openai_function": self._extract_extra(function, {"name", "description", "parameters"}),
+                        "openai_function": self._extract_extra(
+                            function, {"name", "description", "parameters"}
+                        ),
                     },
                 )
             )
@@ -756,7 +969,9 @@ class OpenAINormalizer(FormatNormalizer):
             fn_raw = tool_choice.get("function")
             fn: dict[str, Any] = fn_raw if isinstance(fn_raw, dict) else {}
             name = str(fn.get("name") or "")
-            return ToolChoice(type=ToolChoiceType.TOOL, tool_name=name, extra={"openai": tool_choice})
+            return ToolChoice(
+                type=ToolChoiceType.TOOL, tool_name=name, extra={"openai": tool_choice}
+            )
 
         return ToolChoice(type=ToolChoiceType.AUTO, extra={"openai": tool_choice})
 
@@ -771,7 +986,9 @@ class OpenAINormalizer(FormatNormalizer):
             return {"type": "function", "function": {"name": tool_choice.tool_name or ""}}
         return "auto"
 
-    def _openai_tool_call_to_block(self, tool_call: Any) -> tuple[ToolUseBlock | None, dict[str, int]]:
+    def _openai_tool_call_to_block(
+        self, tool_call: Any
+    ) -> tuple[ToolUseBlock | None, dict[str, int]]:
         dropped: dict[str, int] = {}
         if not isinstance(tool_call, dict):
             dropped["openai_tool_call_non_dict"] = dropped.get("openai_tool_call_non_dict", 0) + 1
@@ -808,12 +1025,16 @@ class OpenAINormalizer(FormatNormalizer):
             dropped,
         )
 
-    def _legacy_function_call_to_block(self, func_call: dict[str, Any]) -> tuple[ToolUseBlock | None, dict[str, int]]:
+    def _legacy_function_call_to_block(
+        self, func_call: dict[str, Any]
+    ) -> tuple[ToolUseBlock | None, dict[str, int]]:
         dropped: dict[str, int] = {}
         name = str(func_call.get("name") or "")
         args_str = str(func_call.get("arguments") or "")
         if not name:
-            dropped["openai_function_call_missing_name"] = dropped.get("openai_function_call_missing_name", 0) + 1
+            dropped["openai_function_call_missing_name"] = (
+                dropped.get("openai_function_call_missing_name", 0) + 1
+            )
             return None, dropped
 
         tool_input: dict[str, Any]
@@ -844,7 +1065,12 @@ class OpenAINormalizer(FormatNormalizer):
         dropped: dict[str, int] = {}
         content = msg.get("content")
         if content is None:
-            return ToolResultBlock(tool_use_id=tool_call_id, output=None, content_text=None, extra={"openai": msg}), dropped
+            return (
+                ToolResultBlock(
+                    tool_use_id=tool_call_id, output=None, content_text=None, extra={"openai": msg}
+                ),
+                dropped,
+            )
 
         if isinstance(content, str):
             parsed: Any = None
@@ -906,7 +1132,9 @@ class OpenAINormalizer(FormatNormalizer):
             extra={"openai": extra} if extra else {},
         )
 
-    def _blocks_to_openai_content(self, blocks: list[ContentBlock]) -> str | list[dict[str, Any]] | None:
+    def _blocks_to_openai_content(
+        self, blocks: list[ContentBlock]
+    ) -> str | list[dict[str, Any]] | None:
         parts: list[dict[str, Any]] = []
         text_parts: list[str] = []
 
@@ -946,7 +1174,9 @@ class OpenAINormalizer(FormatNormalizer):
         # OpenAI content 可以是空字符串；但作为响应 message.content 通常允许为 ""/None。
         return ""
 
-    def _split_blocks(self, blocks: list[ContentBlock]) -> tuple[list[ContentBlock], list[ToolUseBlock]]:
+    def _split_blocks(
+        self, blocks: list[ContentBlock]
+    ) -> tuple[list[ContentBlock], list[ToolUseBlock]]:
         content_blocks: list[ContentBlock] = []
         tool_blocks: list[ToolUseBlock] = []
         for b in blocks:
@@ -979,7 +1209,13 @@ class OpenAINormalizer(FormatNormalizer):
         def flush_user() -> None:
             nonlocal pending
             # 丢弃 Unknown/Tool blocks（tool_result 在 flush 时不会出现）
-            content = self._blocks_to_openai_content([b for b in pending if not isinstance(b, (UnknownBlock, ToolUseBlock, ToolResultBlock))])
+            content = self._blocks_to_openai_content(
+                [
+                    b
+                    for b in pending
+                    if not isinstance(b, (UnknownBlock, ToolUseBlock, ToolResultBlock))
+                ]
+            )
             if content is None:
                 content = ""
             out.append({"role": "user", "content": content})
@@ -1025,7 +1261,9 @@ class OpenAINormalizer(FormatNormalizer):
         out["content"] = content_value if content_value is not None else ""
 
         if tool_blocks:
-            out["tool_calls"] = [self._tool_use_block_to_openai_call(b, idx) for idx, b in enumerate(tool_blocks)]
+            out["tool_calls"] = [
+                self._tool_use_block_to_openai_call(b, idx) for idx, b in enumerate(tool_blocks)
+            ]
 
         return out
 

src/core/api_format/detection.py

@@ -6,13 +6,13 @@ API 格式检测
 
 from __future__ import annotations
 
-
+from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from starlette.requests import Request
 
-from src.core.api_format.enums import APIFormat
+from src.core.api_format.enums import APIFormat, AuthMethod, EndpointType
 from src.core.api_format.metadata import API_FORMAT_DEFINITIONS, ApiFormatDefinition
 
 
@@ -64,6 +64,78 @@ def _extract_api_key_by_definition(
         return header_value, "header"
 
 
+@dataclass(frozen=True)
+class RequestContext:
+    """请求上下文 - 三维度信息"""
+
+    data_format: APIFormat
+    endpoint_type: EndpointType
+    auth_method: AuthMethod
+    credentials: str | None
+
+
+def _detect_endpoint_type(path: str) -> EndpointType:
+    normalized = path.lower()
+
+    if normalized.startswith("/upload/v1beta/files") or normalized.startswith("/v1beta/files"):
+        return EndpointType.FILES
+    if normalized.startswith("/v1/videos") or (
+        normalized.startswith("/v1beta/") and "predictlongrunning" in normalized
+    ):
+        return EndpointType.VIDEO
+    # Gemini operations (视频轮询) 也归类为 VIDEO
+    if normalized.startswith("/v1beta/operations"):
+        return EndpointType.VIDEO
+    if normalized.startswith("/v1/models"):
+        return EndpointType.MODELS
+    if "/embeddings" in normalized:
+        return EndpointType.EMBEDDING
+    if "/images" in normalized:
+        return EndpointType.IMAGE
+    if "/audio" in normalized:
+        return EndpointType.AUDIO
+    return EndpointType.CHAT
+
+
+def _detect_data_format(
+    path: str, headers: dict[str, str], query_params: dict[str, str] | None
+) -> APIFormat:
+    normalized = path.lower()
+
+    if normalized.startswith("/v1/messages"):
+        return APIFormat.CLAUDE
+    if normalized.startswith("/v1beta/") or normalized.startswith("/upload/v1beta/"):
+        return APIFormat.GEMINI
+    if normalized.startswith("/v1/chat/completions") or normalized.startswith("/v1/videos"):
+        return APIFormat.OPENAI
+
+    api_format, _api_key, _auth_method = detect_format_from_request(headers, query_params)
+    return api_format
+
+
+def _detect_auth_method(
+    headers: dict[str, str], query_params: dict[str, str] | None
+) -> tuple[AuthMethod, str | None]:
+    # Query key (Gemini) has highest priority
+    query_key = query_params.get("key") if query_params else None
+    if query_key:
+        return AuthMethod.QUERY_KEY, query_key
+
+    x_goog_key = headers.get("x-goog-api-key")
+    if x_goog_key:
+        return AuthMethod.GOOG_API_KEY, x_goog_key
+
+    x_api_key = headers.get("x-api-key")
+    if x_api_key:
+        return AuthMethod.API_KEY, x_api_key
+
+    auth_header = headers.get("authorization", "")
+    if auth_header.lower().startswith("bearer "):
+        return AuthMethod.BEARER, auth_header[7:].strip()
+
+    return AuthMethod.BEARER, None
+
+
 def detect_format_from_request(
     headers: dict[str, str],
     query_params: dict[str, str] | None = None,
@@ -86,20 +158,26 @@ def detect_format_from_request(
     """
     # Claude: x-api-key + anthropic-version (必须同时存在)
     claude_def = API_FORMAT_DEFINITIONS[APIFormat.CLAUDE]
-    claude_key, claude_auth_method = _extract_api_key_by_definition(headers, query_params, claude_def)
+    claude_key, claude_auth_method = _extract_api_key_by_definition(
+        headers, query_params, claude_def
+    )
     if claude_key and headers.get("anthropic-version"):
         return APIFormat.CLAUDE, claude_key, claude_auth_method
 
     # Gemini: x-goog-api-key (header 类型) 或 ?key=
     gemini_def = API_FORMAT_DEFINITIONS[APIFormat.GEMINI]
-    gemini_key, gemini_auth_method = _extract_api_key_by_definition(headers, query_params, gemini_def)
+    gemini_key, gemini_auth_method = _extract_api_key_by_definition(
+        headers, query_params, gemini_def
+    )
     if gemini_key:
         return APIFormat.GEMINI, gemini_key, gemini_auth_method
 
     # OpenAI: Authorization: Bearer (默认)
     # 注意: 如果只有 x-api-key 但没有 anthropic-version，也走 OpenAI 格式
     openai_def = API_FORMAT_DEFINITIONS[APIFormat.OPENAI]
-    openai_key, openai_auth_method = _extract_api_key_by_definition(headers, query_params, openai_def)
+    openai_key, openai_auth_method = _extract_api_key_by_definition(
+        headers, query_params, openai_def
+    )
     # 如果 OpenAI 格式没有 key，但有 x-api-key，也用它（兼容）
     if not openai_key and claude_key:
         openai_key = claude_key
@@ -134,6 +212,28 @@ def detect_format_and_key_from_starlette(
     return format_name, api_key, auth_method
 
 
+def detect_request_context(request: Request) -> RequestContext:
+    """
+    从 Request 中检测三维度信息
+
+    Returns:
+        RequestContext(data_format, endpoint_type, auth_method, credentials)
+    """
+    headers = {k.lower(): v for k, v in request.headers.items()}
+    query_params = dict(request.query_params)
+
+    endpoint_type = _detect_endpoint_type(request.url.path)
+    data_format = _detect_data_format(request.url.path, headers, query_params)
+    auth_method, credentials = _detect_auth_method(headers, query_params)
+
+    return RequestContext(
+        data_format=data_format,
+        endpoint_type=endpoint_type,
+        auth_method=auth_method,
+        credentials=credentials,
+    )
+
+
 def detect_format_from_response(
     response_data: dict,
 ) -> APIFormat | None:
@@ -197,4 +297,6 @@ __all__ = [
     "detect_format_and_key_from_starlette",
     "detect_format_from_response",
     "detect_cli_format_from_path",
+    "detect_request_context",
+    "RequestContext",
 ]

src/core/api_format/enums.py

@@ -18,4 +18,26 @@ class APIFormat(Enum):
     GEMINI_CLI = "GEMINI_CLI"  # Gemini CLI API 格式
 
 
-__all__ = ["APIFormat"]
+class AuthMethod(str, Enum):
+    """认证方式 - 决定如何构造认证 Header"""
+
+    BEARER = "bearer"  # Authorization: Bearer {token}
+    API_KEY = "api_key"  # x-api-key: {key}
+    GOOG_API_KEY = "goog_key"  # x-goog-api-key: {key}
+    OAUTH2 = "oauth2"  # Google OAuth2 / Service Account
+    QUERY_KEY = "query_key"  # ?key={key} (Gemini 备用)
+
+
+class EndpointType(str, Enum):
+    """端点类型 - 决定 API 功能类别"""
+
+    CHAT = "chat"  # Chat/Completion API
+    VIDEO = "video"  # Video Generation API
+    FILES = "files"  # Files API
+    IMAGE = "image"  # Image Generation API
+    AUDIO = "audio"  # Audio API
+    EMBEDDING = "embedding"  # Embedding API
+    MODELS = "models"  # Models API
+
+
+__all__ = ["APIFormat", "AuthMethod", "EndpointType"]