|
|
|
|
|
|
| """Request source detection helpers."""
|
|
|
| from __future__ import annotations
|
|
|
| import re
|
| from dataclasses import dataclass
|
| from typing import Optional
|
|
|
| from fastapi import Request
|
|
|
|
|
| @dataclass(frozen=True)
|
| class RequestSourceInfo:
|
| """Normalized request-source metadata for logging."""
|
|
|
| source: str
|
| protocol: str
|
| client_name: str
|
| endpoint: str
|
| user_agent: str
|
|
|
|
|
| def _normalize_source_name(value: str) -> str:
|
| normalized = re.sub(r"[^a-zA-Z0-9._-]+", "_", value.strip().lower())
|
| return normalized.strip("_") or "unknown"
|
|
|
|
|
| def detect_request_source(
|
| request: Request,
|
| protocol_hint: Optional[str] = None,
|
| ) -> RequestSourceInfo:
|
| """Detect the request source from headers and path."""
|
| headers = request.headers
|
| endpoint = request.url.path
|
| user_agent = (headers.get("user-agent") or "").strip()
|
| user_agent_normalized = user_agent.casefold()
|
|
|
| protocol = (protocol_hint or "").strip().lower()
|
| if not protocol:
|
| protocol = "openai" if endpoint.startswith("/v1/") else "unknown"
|
|
|
| explicit_source = headers.get("x-request-source") or headers.get("x-client-source")
|
| if explicit_source:
|
| return RequestSourceInfo(
|
| source=_normalize_source_name(explicit_source),
|
| protocol=protocol,
|
| client_name=explicit_source.strip(),
|
| endpoint=endpoint,
|
| user_agent=user_agent,
|
| )
|
|
|
| if "openai" in user_agent_normalized:
|
| source = "openai_sdk"
|
| client_name = "OpenAI SDK"
|
| elif "curl/" in user_agent_normalized:
|
| source = "curl"
|
| client_name = "curl"
|
| elif any(
|
| token in user_agent_normalized
|
| for token in ("python-httpx", "httpx/", "python-requests", "requests/")
|
| ):
|
| source = "custom_http_client"
|
| client_name = "HTTP Client"
|
| elif "mozilla/" in user_agent_normalized:
|
| source = "browser"
|
| client_name = "Browser"
|
| elif protocol == "openai":
|
| source = "openai_compatible"
|
| client_name = "OpenAI Compatible"
|
| else:
|
| source = "unknown"
|
| client_name = "Unknown"
|
|
|
| return RequestSourceInfo(
|
| source=source,
|
| protocol=protocol,
|
| client_name=client_name,
|
| endpoint=endpoint,
|
| user_agent=user_agent,
|
| )
|
|
|
|
|
| def format_request_source(info: RequestSourceInfo) -> str:
|
| """Render request-source metadata into a compact log prefix."""
|
| return (
|
| f"[source={info.source}]"
|
| f"[protocol={info.protocol}]"
|
| f"[client={info.client_name}]"
|
| f"[endpoint={info.endpoint}]"
|
| )
|
|
|
