125 lines
4.2 KiB
Python
125 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
from app.config import settings
|
|
|
|
|
|
STATIC_ROOT = Path(__file__).resolve().parent.parent / "static"
|
|
|
|
|
|
def _normalize_url(value: str) -> str:
|
|
url = str(value or "").strip()
|
|
if not url:
|
|
return ""
|
|
if url.startswith("//"):
|
|
return f"https:{url}"
|
|
if url.startswith("/"):
|
|
return url
|
|
if url.startswith(("http:/", "https:/")) and not url.startswith(("http://", "https://")):
|
|
fixed = url.replace(":/", "://", 1)
|
|
parsed = urlsplit(fixed)
|
|
if parsed.netloc in {"api", "static"}:
|
|
return parsed.path or "/"
|
|
if parsed.netloc and parsed.path.startswith("/v1/static/"):
|
|
return parsed.path.replace("/v1/", "/api/v1/", 1)
|
|
return urlunsplit(parsed)
|
|
return url
|
|
|
|
|
|
def _canonical_static_path(path: str) -> str:
|
|
normalized = _normalize_url(path)
|
|
if normalized.startswith("/api/v1/static/"):
|
|
return normalized
|
|
if normalized.startswith("/static/"):
|
|
return f"{settings.api_v1_prefix}{normalized}"
|
|
if normalized.startswith("/v1/static/"):
|
|
return normalized.replace("/v1/", "/api/v1/", 1)
|
|
return f"{settings.api_v1_prefix}/static/{normalized.lstrip('/')}"
|
|
|
|
|
|
def _local_static_file_exists(url: str) -> bool:
|
|
normalized = _normalize_url(url)
|
|
static_prefixes = (f"{settings.api_v1_prefix}/static/", "/static/", "/v1/static/")
|
|
if not normalized.startswith(static_prefixes):
|
|
return True
|
|
relative_path = normalized
|
|
if normalized.startswith(f"{settings.api_v1_prefix}/static/"):
|
|
relative_path = normalized.removeprefix(f"{settings.api_v1_prefix}/static/")
|
|
elif normalized.startswith("/v1/static/"):
|
|
relative_path = normalized.removeprefix("/v1/")
|
|
else:
|
|
relative_path = normalized.removeprefix("/static/")
|
|
file_path = STATIC_ROOT / Path(relative_path)
|
|
return file_path.is_file()
|
|
|
|
|
|
def _build_absolute_media_url(url: str) -> str:
|
|
normalized = _normalize_url(url)
|
|
base = _normalize_public_base_url(settings.public_site_url)
|
|
normalized = _canonical_static_path(normalized)
|
|
if not base:
|
|
return normalized
|
|
return f"{base}{normalized}"
|
|
|
|
|
|
def _normalize_public_base_url(value: str | None) -> str:
|
|
base = str(value or "").strip().rstrip("/")
|
|
if not base:
|
|
return ""
|
|
if base.startswith("//"):
|
|
base = f"https:{base}"
|
|
if base.startswith(("http:/", "https:/")) and not base.startswith(("http://", "https://")):
|
|
base = base.replace(":/", "://", 1)
|
|
if base.startswith(("http://", "https://")):
|
|
parsed = urlsplit(base)
|
|
if parsed.scheme and parsed.netloc:
|
|
return f"{parsed.scheme}://{parsed.netloc}{parsed.path.rstrip('/')}"
|
|
return ""
|
|
return ""
|
|
|
|
|
|
def sanitize_media_url(url: str | None) -> str | None:
|
|
if not url:
|
|
return None
|
|
normalized = _normalize_url(url)
|
|
if not normalized:
|
|
return None
|
|
if normalized.startswith((f"{settings.api_v1_prefix}/static/", "/static/", "/v1/static/")) and not _local_static_file_exists(normalized):
|
|
return None
|
|
return _build_absolute_media_url(normalized) if normalized.startswith("/") else normalized
|
|
|
|
|
|
def sanitize_media_list(urls: list[str] | None) -> list[str]:
|
|
if not isinstance(urls, list):
|
|
return []
|
|
items: list[str] = []
|
|
for item in urls:
|
|
if not isinstance(item, str):
|
|
continue
|
|
safe_url = sanitize_media_url(item)
|
|
if safe_url:
|
|
items.append(safe_url)
|
|
return items
|
|
|
|
|
|
def sanitize_user_media_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
|
data = dict(payload)
|
|
avatar_url = sanitize_media_url(data.get("avatar_url"))
|
|
avatar_blur_url = sanitize_media_url(data.get("avatar_blur_url"))
|
|
profile_images = sanitize_media_list(data.get("profile_images"))
|
|
|
|
if not avatar_url and profile_images:
|
|
avatar_url = profile_images[0]
|
|
if not avatar_blur_url:
|
|
avatar_blur_url = avatar_url
|
|
if avatar_url and not profile_images:
|
|
profile_images = [avatar_url]
|
|
|
|
data["avatar_url"] = avatar_url
|
|
data["avatar_blur_url"] = avatar_blur_url
|
|
data["profile_images"] = profile_images
|
|
return data
|