"""영상 공유 링크용 Open Graph HTML 생성 서비스."""
from collections.abc import Mapping
from dataclasses import dataclass
from html import escape
from urllib.parse import urlsplit, urlunsplit
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.home.models import Project
from app.video.models import Video
FALLBACK_FRONTEND_URL = "https://ado2.o2osolution.ai"
DEFAULT_SHARE_IMAGE_PATH = "/assets/images/ado2_image.png"
DEFAULT_SHARE_IMAGE_STATIC_PATH = "/static/images/ado2_image.png"
@dataclass(frozen=True, slots=True)
class VideoShareData:
"""공유 페이지에 필요한 영상 및 프로젝트 정보."""
video_id: int
poster_url: str | None
store_name: str
region: str
title: str | None = None
description: str | None = None
async def get_video_share_data(
session: AsyncSession,
video_id: int,
) -> VideoShareData | None:
"""공유 가능한 완료 영상을 프로젝트 정보와 함께 조회합니다."""
result = await session.execute(
select(
Video.id,
Video.poster_url,
Video.title,
Video.description,
Project.store_name,
Project.region,
)
.join(Project, Video.project_id == Project.id)
.where(
Video.id == video_id,
Video.status == "completed",
Video.is_deleted.is_(False),
Project.is_deleted.is_(False),
)
)
row = result.one_or_none()
if row is None:
return None
return VideoShareData(
video_id=row.id,
poster_url=row.poster_url,
store_name=row.store_name,
region=row.region,
title=row.title,
description=row.description,
)
def build_video_share_html(
data: VideoShareData,
*,
share_url: str,
frontend_base_url: str,
configured_default_image_url: str = "",
) -> str:
"""영상별 OG 메타데이터와 상세 화면 이동 기능을 포함한 HTML을 생성합니다."""
frontend_base = _normalise_frontend_base_url(frontend_base_url)
detail_url = f"{frontend_base}/video/{data.video_id}"
fallback_image_url = _resolve_default_image_url(
configured_default_image_url,
frontend_base,
share_url=share_url,
)
image_url = _absolute_http_url(data.poster_url) or fallback_image_url
title = _share_title(data.title, data.store_name)
description = _share_description(data.description)
canonical_tags = _canonical_tags(_absolute_http_url(share_url))
image_size_tags = _og_image_size_tags(image_url, fallback_image_url)
escaped_title = escape(title, quote=True)
escaped_description = escape(description, quote=True)
escaped_image_url = escape(image_url, quote=True)
escaped_detail_url = escape(detail_url, quote=True)
return f"""
{escaped_title}
{canonical_tags}
{image_size_tags}
{escaped_title}
{escaped_description}
영상 보기
"""
def _normalise_text(value: str | None, fallback: str) -> str:
"""메타데이터용 텍스트에서 불필요한 공백을 제거합니다."""
normalised = " ".join((value or "").split())
return normalised or fallback
_OG_DESCRIPTION_MAX_LEN = 300
def _share_title(title: str | None, store_name: str) -> str:
"""저장된 SNS 제목을 쓰고, 없으면 가게명 폴백을 사용합니다."""
stored = _normalise_text(title, "")
if stored:
return stored
name = _normalise_text(store_name, "ADO2 영상")
return f"{name} | ADO2"
def _share_description(description: str | None) -> str:
"""저장된 SNS 설명을 쓰고, 없으면 기본 문구를 사용합니다."""
stored = _normalise_text(description, "")
if stored:
if len(stored) > _OG_DESCRIPTION_MAX_LEN:
return stored[: _OG_DESCRIPTION_MAX_LEN - 1].rstrip() + "…"
return stored
return "ADO2 AI 마케팅 영상"
def resolve_frontend_base_url(headers: Mapping[str, str], fallback: str) -> str:
"""프록시가 넘긴 프론트 호스트를 우선해 canonical 기준 URL을 정합니다."""
forwarded_host = (headers.get("x-forwarded-host") or "").split(",")[0].strip()
if not forwarded_host:
return _normalise_frontend_base_url(fallback)
forwarded_proto = (headers.get("x-forwarded-proto") or "https").split(",")[0].strip().lower()
if forwarded_proto not in {"http", "https"}:
forwarded_proto = "https"
return _normalise_frontend_base_url(f"{forwarded_proto}://{forwarded_host}")
def resolve_share_url(
headers: Mapping[str, str],
request_url: str,
configured_api_base_url: str = "",
) -> str:
"""크롤러가 다시 읽어도 같은 OG 페이지가 나오는 공유 URL을 만듭니다.
nginx가 ``/api`` prefix를 떼고 넘기면 ``request.url``에는 그 prefix가 없어,
그대로 쓰면 프론트 SPA 주소가 된다. 복원할 근거가 없으면 빈 문자열을 돌려
호출부가 canonical/og:url을 생략하도록 한다.
"""
path = urlsplit(request_url).path
configured_base = _absolute_http_url(configured_api_base_url)
if configured_base:
return f"{configured_base.rstrip('/')}{path}"
origin = _forwarded_origin(headers) or _origin_from_url(request_url)
if not origin:
return ""
forwarded_prefix = (headers.get("x-forwarded-prefix") or "").strip().rstrip("/")
if forwarded_prefix:
return f"{origin}{forwarded_prefix}{path}"
if headers.get("x-forwarded-host"):
# 프록시 뒤인데 prefix 를 못 받았다. 잘못된 URL 을 내보내지 않는다.
return ""
return f"{origin}{path}"
def _forwarded_origin(headers: Mapping[str, str]) -> str | None:
"""프록시가 넘긴 외부 호스트 기준 origin을 만듭니다."""
forwarded_host = (headers.get("x-forwarded-host") or "").split(",")[0].strip()
if not forwarded_host:
return None
forwarded_proto = (headers.get("x-forwarded-proto") or "https").split(",")[0].strip().lower()
if forwarded_proto not in {"http", "https"}:
forwarded_proto = "https"
return f"{forwarded_proto}://{forwarded_host}"
def _canonical_tags(canonical_url: str | None) -> str:
"""공유 URL을 확신할 때만 canonical/og:url을 붙입니다.
잘못된 og:url을 내보내면 크롤러가 그 주소를 다시 읽어, OG 메타가 없는
프론트 SPA 문서를 미리보기로 쓴다. 확신이 없으면 크롤러가 실제로 받은
URL을 쓰도록 태그 자체를 생략한다.
"""
if not canonical_url:
return ""
escaped = escape(canonical_url, quote=True)
return (
f' \n'
f' \n'
)
_DEFAULT_OG_IMAGE_SIZE = (385, 385)
def _og_image_size_tags(image_url: str, fallback_image_url: str) -> str:
"""폴백 로고처럼 크기를 아는 이미지에만 width/height 메타를 붙입니다."""
if image_url != fallback_image_url:
return ""
width, height = _DEFAULT_OG_IMAGE_SIZE
return (
f' \n'
f' \n'
)
def _normalise_frontend_base_url(value: str) -> str:
"""프론트엔드 기준 URL을 안전한 절대 HTTP(S) URL로 정규화합니다."""
absolute_url = _absolute_http_url(value) or FALLBACK_FRONTEND_URL
parts = urlsplit(absolute_url)
path = parts.path.rstrip("/")
return urlunsplit((parts.scheme, parts.netloc, path, "", ""))
def _resolve_default_image_url(
configured_url: str,
frontend_base: str,
*,
share_url: str = "",
) -> str:
"""포스터가 없을 때 사용할 기본 OG 이미지 URL을 반환합니다.
우선순위:
1. ``SHARE_DEFAULT_IMAGE_URL`` (.env)
2. 공유 URL과 같은 API 베이스 ``.../static/images/ado2_image.png``
(``/api/video/share/1`` 이면 ``/api/static/...``)
3. ``SHARE_FRONTEND_URL`` + ``/assets/images/ado2_image.png``
"""
configured_absolute_url = _absolute_http_url(configured_url)
if configured_absolute_url:
return configured_absolute_url
share_api_base = _api_base_from_share_url(share_url)
if share_api_base:
return f"{share_api_base}{DEFAULT_SHARE_IMAGE_STATIC_PATH}"
return f"{frontend_base}{DEFAULT_SHARE_IMAGE_PATH}"
def _api_base_from_share_url(share_url: str) -> str | None:
"""공유 URL에서 API 베이스를 만듭니다. ``/api/video/share/1`` → ``https://host/api``."""
absolute_url = _absolute_http_url(share_url)
if not absolute_url:
return None
parts = urlsplit(absolute_url)
origin = urlunsplit((parts.scheme, parts.netloc, "", "", ""))
idx = (parts.path or "").find("/video/share/")
if idx < 0:
return origin
prefix = parts.path[:idx].rstrip("/")
return f"{origin}{prefix}" if prefix else origin
def _origin_from_url(value: str) -> str | None:
"""URL에서 scheme + host(+port) origin만 추출합니다."""
absolute_url = _absolute_http_url(value)
if not absolute_url:
return None
parts = urlsplit(absolute_url)
return urlunsplit((parts.scheme, parts.netloc, "", "", ""))
def _absolute_http_url(value: str | None) -> str | None:
"""값이 절대 HTTP(S) URL인 경우에만 정리된 문자열을 반환합니다."""
candidate = (value or "").strip()
if not candidate:
return None
try:
parts = urlsplit(candidate)
except ValueError:
return None
if parts.scheme.lower() not in {"http", "https"} or not parts.hostname:
return None
return candidate