alvaro
abd03ed07d
Rename the project directories, Python distribution and import package, and npm package. Update launcher, Docker, tooling, tests, lockfiles, and documentation while preserving existing API contracts. Make AGENTS.md self-contained and remove external skill and workflow references.
54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
|
|
|
|
from fastapi import Request, WebSocket
|
|
|
|
from vocero_api.core import config as settings
|
|
|
|
|
|
def _url_with_query(url: str, **params: str) -> str:
|
|
target = urlsplit(url)
|
|
merged_query = dict(parse_qsl(target.query, keep_blank_values=True))
|
|
for key, value in params.items():
|
|
if value:
|
|
merged_query[key] = value
|
|
return urlunsplit(
|
|
(
|
|
target.scheme,
|
|
target.netloc,
|
|
target.path,
|
|
urlencode(merged_query, doseq=True),
|
|
target.fragment,
|
|
)
|
|
)
|
|
|
|
|
|
def _proxy_public_url(request: Request) -> str:
|
|
if not settings.S2S_PROXY_TARGET:
|
|
return ""
|
|
forwarded_proto = (request.headers.get("x-forwarded-proto") or request.url.scheme).split(",")[0].strip()
|
|
forwarded_host = (
|
|
(request.headers.get("x-forwarded-host") or request.headers.get("host") or request.url.netloc)
|
|
.split(",")[0]
|
|
.strip()
|
|
)
|
|
ws_scheme = "wss" if forwarded_proto in {"https", "wss"} else "ws"
|
|
root_path = (request.scope.get("root_path") or "").rstrip("/")
|
|
return f"{ws_scheme}://{forwarded_host}{root_path}/v1/realtime"
|
|
|
|
|
|
def _proxy_upstream_url(ws: WebSocket) -> str:
|
|
target = urlsplit(settings.S2S_PROXY_TARGET)
|
|
merged_query = dict(parse_qsl(target.query, keep_blank_values=True))
|
|
merged_query.update(parse_qsl(ws.url.query, keep_blank_values=True))
|
|
return urlunsplit(
|
|
(
|
|
target.scheme,
|
|
target.netloc,
|
|
target.path or "/v1/realtime",
|
|
urlencode(merged_query, doseq=True),
|
|
target.fragment,
|
|
)
|
|
)
|