"""콘텐츠 공유 링크용 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.ssulbox.models import SsulContent from app.video.models import Video from app.video.services.share_image import og_image_size FALLBACK_FRONTEND_URL = "https://ado2.o2osolution.ai" DEFAULT_SHARE_IMAGE_PATH = "/assets/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 @dataclass(frozen=True, slots=True) class SsulShareData: """공유 페이지에 필요한 썰박스 콘텐츠 정보.""" content_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, ) async def get_ssul_share_data( session: AsyncSession, content_id: int, ) -> SsulShareData | None: """공유 가능한 완료 썰박스 콘텐츠를 조회합니다.""" result = await session.execute( select( SsulContent.id, SsulContent.poster_url, SsulContent.title, SsulContent.description, SsulContent.store_name, SsulContent.region, ).where( SsulContent.id == content_id, SsulContent.status == "done", SsulContent.is_deleted.is_(False), SsulContent.video_url.is_not(None), SsulContent.video_url != "", ) ) row = result.one_or_none() if row is None: return None return SsulShareData( content_id=row.id, poster_url=row.poster_url, store_name=row.store_name, region=row.region or "", 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을 생성합니다.""" return _build_share_html( detail_path=f"/video/{data.video_id}", poster_url=data.poster_url, title=_share_title(data.title, data.store_name, fallback_store="ADO2 영상"), description=_share_description(data.description), share_url=share_url, frontend_base_url=frontend_base_url, configured_default_image_url=configured_default_image_url, ) def build_ssul_share_html( data: SsulShareData, *, share_url: str, frontend_base_url: str, configured_default_image_url: str = "", ) -> str: """썰박스 OG 메타데이터와 상세 화면 이동 기능을 포함한 HTML을 생성합니다.""" return _build_share_html( detail_path=f"/ssul/{data.content_id}", poster_url=data.poster_url, title=_share_title(data.title, data.store_name, fallback_store="ADO2 썰"), description=_share_description(data.description), share_url=share_url, frontend_base_url=frontend_base_url, configured_default_image_url=configured_default_image_url, ) def _build_share_html( *, detail_path: str, poster_url: str | None, title: str, description: str, 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}{detail_path}" fallback_image_url = _resolve_default_image_url( configured_default_image_url, frontend_base, ) canonical_url = _absolute_http_url(share_url) image_url = _resolve_og_image_url( canonical_url, poster_url=poster_url, fallback_image_url=fallback_image_url, ) image_size_tags = _og_image_size_tags(image_url, canonical_url) canonical_tags = _canonical_tags(canonical_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, *, fallback_store: str) -> str: """저장된 SNS 제목을 쓰고, 없으면 가게명 폴백을 사용합니다.""" stored = _normalise_text(title, "") if stored: return stored name = _normalise_text(store_name, fallback_store) 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 _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 _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) -> str: """공유 URL을 몰라 카드 엔드포인트를 못 쓸 때의 기본 이미지 URL입니다. 우선순위: 1. ``SHARE_DEFAULT_IMAGE_URL`` (.env) 2. ``SHARE_FRONTEND_URL`` + ``/assets/images/ado2_image.png`` """ configured_absolute_url = _absolute_http_url(configured_url) if configured_absolute_url: return configured_absolute_url return f"{frontend_base}{DEFAULT_SHARE_IMAGE_PATH}" 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' ) #: 공유 URL 뒤에 붙는 OG 카드 이미지 경로. 라우터의 엔드포인트 경로와 같아야 한다. OG_IMAGE_PATH_SUFFIX = "/og.jpg" def _resolve_og_image_url( canonical_url: str | None, *, poster_url: str | None, fallback_image_url: str, ) -> str: """카드 비율을 보장하는 OG 이미지 URL을 고릅니다. 공유 URL을 알면 원본을 1200x630으로 합성해 주는 엔드포인트를 쓴다. 포스터가 세로형이어도 플랫폼이 가운데를 잘라내지 않는다. 공유 URL을 모르면 절대 주소를 만들 수 없으므로 원본 포스터를 그대로 내보낸다. """ if not canonical_url: return _absolute_http_url(poster_url) or fallback_image_url parts = urlsplit(canonical_url) path = f"{parts.path.rstrip('/')}{OG_IMAGE_PATH_SUFFIX}" return urlunsplit((parts.scheme, parts.netloc, path, "", "")) def _og_image_size_tags(image_url: str, canonical_url: str | None) -> str: """크기를 아는 합성 카드에만 width/height 메타를 붙입니다.""" if not canonical_url or not image_url.endswith(OG_IMAGE_PATH_SUFFIX): return "" width, height = og_image_size() return ( f' \n' f' \n' ) 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