feat(ssulbox): 공식 링크 오버레이 추가 및 AI 브리핑 대기 제거

placeDetail(base)의 homepages 필드는 스키마에 없어 전체 크롤링이
400으로 실패했다(컨테이너 검증 완료). 쿼리에서 제거하고, WTM 우회용
브라우저가 이미 로드하는 place 페이지의 __APOLLO_STATE__ JSON을 캡처해
대표(repr) 링크를 추출한다. 브라우저 캡처 실패 시 HTML 직접 파싱으로
보충한다(직접 요청은 429로 막히는 환경이 있어 best-effort).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVSeUT2pEmeHc6hG6gVXUS

app/archive/api/routers/v1/archive.py

@@ -16,6 +16,7 @@ from app.user.dependencies.auth import get_current_user
 from app.user.models import User
 from app.utils.logger import get_logger
 from app.utils.pagination import PaginatedResponse
+from app.utils.upload_blob_as_request import to_playback_url
 from app.video.models import Video
 from app.video.schemas.video_schema import VideoListItem
 from app.video.services import unified_list
@@ -93,7 +94,7 @@ async def get_videos(
                 # 썰박스는 task_id 개념이 없어 빈 문자열이다.
                 # 프론트는 반드시 (type, video_id) 쌍으로 식별할 것.
                 task_id=it.task_id,
-                result_movie_url=it.movie_url,
+                result_movie_url=to_playback_url(it.movie_url),
                 poster_url=it.poster_url,
                 title=it.title,
                 description=it.description,

app/database/session.py

@@ -80,7 +80,8 @@ async def create_db_tables():
     from app.home.models import Image, Project, MarketingIntel, ImageTag  # noqa: F401
     from app.lyric.models import Lyric  # noqa: F401
     from app.song.models import Song, SongTimestamp  # noqa: F401
-    from app.video.models import Video  # noqa: F401
+    from app.video.models import Video, VideoReaction  # noqa: F401
+    from app.comment.models import Comment  # noqa: F401
     from app.sns.models import SNSUploadTask  # noqa: F401
     from app.social.models import SocialUpload  # noqa: F401
     from app.dashboard.models import Dashboard  # noqa: F401
@@ -101,6 +102,8 @@ async def create_db_tables():
         Song.__table__,
         SongTimestamp.__table__,
         Video.__table__,
+        VideoReaction.__table__,
+        Comment.__table__,
         SNSUploadTask.__table__,
         SocialUpload.__table__,
         MarketingIntel.__table__,

app/home/api/routers/v1/home.py

@@ -417,8 +417,16 @@ async def _crawling_logic(url: str, session: AsyncSession):
             )
 
             # Step 4-3: 분석 결과 DB 저장 (industry는 Project로 흐르므로 여기엔 미저장)
+            # 공식 링크: 플레이스 홈페이지 항목 우선, 없으면 유저가 입력한 크롤링 소스 URL
+            # (컬럼 길이를 넘는 긴 검색 URL은 place_id 기반 표준 플레이스 URL로 대체)
+            official_site_url = scraper.official_site_url or url
+            if len(official_site_url) > 2048:
+                official_site_url = (
+                    f"https://map.naver.com/p/entry/place/{scraper.place_id[2:]}"
+                )
             marketing_intel = MarketingIntel(
                 place_id=scraper.place_id,
+                official_site_url=official_site_url,
                 intel_result=marketing_analysis.model_dump(),
             )
             session.add(marketing_intel)
@@ -486,6 +494,7 @@ async def _crawling_logic(url: str, session: AsyncSession):
 - **customer_name**: 업체명 / 브랜드명 (필수)
 - **address**: 도로명 또는 지번 주소 (필수)
 - **category**: 업종/카테고리 자유 입력 (선택, 예: 펜션, 카페). 비우면 업체명 기반 AI 분류
+- **official_site_url**: 업체 공식 홈페이지 링크 (선택, http/https만 허용, 최대 2048자). 영상 응답의 official_site_url로 노출
 
 ## 반환 정보
 - **processed_info**: 가공된 장소 정보 (customer_name, region, detail_region_info)
@@ -529,6 +538,7 @@ async def manual_marketing(
         # Step 3: 분석 결과 DB 저장 (place_id=None — 네이버 장소와 연결되지 않음)
         marketing_intel = MarketingIntel(
             place_id=None,
+            official_site_url=request_body.official_site_url,
             intel_result=marketing_analysis.model_dump(),
         )
         session.add(marketing_intel)

app/home/models.py

@@ -274,6 +274,7 @@ class MarketingIntel(Base):
     Attributes:
         id: 고유 식별자 (자동 증가)
         place_id : 데이터 소스별 식별자
+        official_site_url : 업체 공식 링크 (플레이스 홈페이지 항목, 없으면 크롤링 소스 URL)
         intel_result : 마케팅 분석 결과물 json
         created_at: 생성 일시 (자동 설정)
     """
@@ -302,6 +303,12 @@ class MarketingIntel(Base):
         comment="매장 소스별 고유 식별자 (네이버 크롤링 시 'nv{id}' 형식; 직접 입력 시 NULL)",
     )
 
+    official_site_url: Mapped[Optional[str]] = mapped_column(
+        String(2048),
+        nullable=True,
+        comment="업체 공식 링크 (플레이스 홈페이지 항목 우선, 없으면 크롤링 소스 URL; 직접 입력 시 NULL)",
+    )
+
     intel_result : Mapped[dict[str, Any]] = mapped_column(
         JSON,
         nullable=False,

app/home/schemas/home_schema.py

@@ -1,6 +1,6 @@
 from typing import Literal, Optional
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 from app.utils.prompts.schemas import MarketingPromptOutput
 
 class CrawlingRequest(BaseModel):
@@ -261,6 +261,7 @@ class ManualMarketingRequest(BaseModel):
                 "store_name": "스테이 머뭄",
                 "address": "전북특별자치도 군산시 절골길 18",
                 "category": "펜션",
+                "official_site_url": "https://www.staymeomoom.com",
             }
         }
     )
@@ -268,6 +269,23 @@ class ManualMarketingRequest(BaseModel):
     store_name: str = Field(..., description="업체명 / 브랜드명")
     address: str = Field(..., description="도로명 또는 지번 주소")
     category: str = Field(default="", description="업체 업종/카테고리 자유 입력 (예: 펜션, 카페). 크롤링 경로와 동일하게 AI가 8개 industry enum으로 자동 분류하는 데 사용. 비우면 업체명 기반 AI 분류")
+    official_site_url: Optional[str] = Field(
+        default=None,
+        max_length=2048,
+        description="업체 공식 홈페이지 링크 (선택, http/https만 허용). 영상 응답의 official_site_url로 노출됨",
+    )
+
+    @field_validator("official_site_url")
+    @classmethod
+    def _validate_official_site_url(cls, v: Optional[str]) -> Optional[str]:
+        if v is None:
+            return None
+        v = v.strip()
+        if not v:
+            return None
+        if not v.startswith(("http://", "https://")):
+            raise ValueError("official_site_url은 http:// 또는 https://로 시작해야 합니다.")
+        return v
 
 
 class ErrorResponse(BaseModel):

app/ssulbox/api/routers/v1/content.py

@@ -83,7 +83,6 @@ async def create_ssul(
                 session,
                 user_uuid=current_user.user_uuid,
                 scenario=body.scenario,
-                input_text=body.input,
                 scenes=body.scenes,
                 seconds=body.seconds,
                 store_name=body.store_name,
@@ -300,6 +299,7 @@ async def get_content_detail(
         description=row.description,
         store_name=row.store_name or None,
         region=row.region,
+        official_site_url=row.official_site_url,
         created_at=row.created_at,
         like_count=like_count,
         is_liked_by_me=is_liked,

app/ssulbox/models.py

@@ -110,10 +110,14 @@ class SsulContent(Base):
         comment="시나리오 코드 (joseon/samgukji/greek/odyssey)",
     )
 
-    input: Mapped[str] = mapped_column(
-        Text,
-        nullable=False,
-        comment="입력값 (네이버 지도 URL 또는 업장명)",
+    # castad `marketing.official_site_url` 과 같은 의미다 — 영상 종료 직전 오버레이의 링크.
+    # 생성 요청 시점에는 알 수 없어 NULL 로 시작하고, 워커가 place 페이지를 크롤링할 때
+    # 채운다(홈페이지 항목 우선, 없으면 네이버 플레이스 URL). 업장명만 입력해 place URL
+    # 해석까지 실패하면 끝까지 NULL 이고, 그때는 프론트가 오버레이를 그리지 않는다.
+    official_site_url: Mapped[Optional[str]] = mapped_column(
+        String(2048),
+        nullable=True,
+        comment="업체 공식 링크 (플레이스 홈페이지 항목 우선, 없으면 네이버 플레이스 URL; 미확보 시 NULL)",
     )
 
     # scenes / seconds 는 DB 기본값을 두지 않는다. 기본값(9 / 30)과 허용 범위

app/ssulbox/schemas/ssulbox_schema.py

@@ -119,6 +119,13 @@ class SsulDetailResponse(BaseModel):
     description: Optional[str] = Field(None, description="SNS 업로드 설명")
     store_name: Optional[str] = Field(None, description="업장명")
     region: Optional[str] = Field(None, description="지역명")
+    official_site_url: Optional[str] = Field(
+        None,
+        description=(
+            "업체 공식 링크 (플레이스 홈페이지 항목 우선, 없으면 네이버 플레이스 URL). "
+            "영상 종료 직전 오버레이에 쓰며, 미확보 시 null 이라 오버레이를 그리지 않는다"
+        ),
+    )
     created_at: datetime = Field(..., description="생성 일시")
     like_count: int = Field(..., description="좋아요 수")
     is_liked_by_me: bool = Field(..., description="현재 로그인 사용자가 좋아요를 눌렀는지")

app/ssulbox/services/place_service.py

@@ -141,6 +141,10 @@ async def _extract_detail_from_page(page, tries: int = 12) -> dict | None:
                 continue
             detail = _detail_from_state(state)
             if detail:
+                # 공식 링크는 apollo state 의 파싱된 dict 가 아니라 원문 HTML 에서 뽑는다
+                # (`homepages` 는 GraphQL placeDetail 이 노출하지 않아 castad 도 같은 방식).
+                # 이미 받아 둔 html 을 재사용하므로 추가 요청·왕복이 없다.
+                detail["homepage"] = NvMapScraper._extract_homepage_from_html(html)
                 return detail
         await page.wait_for_timeout(700)
     return None

app/ssulbox/services/task_service.py

@@ -42,7 +42,6 @@ async def create_task(
     *,
     user_uuid: str,
     scenario: str,
-    input_text: str,
     scenes: int,
     seconds: int,
     store_name: Optional[str] = None,
@@ -65,7 +64,6 @@ async def create_task(
     row = SsulContent(
         user_uuid=user_uuid,
         scenario=scenario,
-        input=input_text,
         scenes=scenes,
         seconds=seconds,
         status=SsulTaskStatus.QUEUED.value,
@@ -217,15 +215,15 @@ async def fail_task(
 
 async def get_place_info(
     session: AsyncSession, content_id: int
-) -> tuple[Optional[str], Optional[str], Optional[str]]:
-    """(store_name, region, detail_region_info).
+) -> tuple[Optional[str], Optional[str], Optional[str], Optional[str]]:
+    """(store_name, region, detail_region_info, official_site_url).
 
     크롤링으로 채울 값이 남았는지 판단하는 데 쓴다.
     """
     row = await session.get(SsulContent, content_id)
     if row is None:
-        return None, None, None
-    return row.store_name, row.region, row.detail_region_info
+        return None, None, None, None
+    return row.store_name, row.region, row.detail_region_info, row.official_site_url
 
 
 async def set_place_info(
@@ -235,10 +233,14 @@ async def set_place_info(
     store_name: Optional[str] = None,
     region: Optional[str] = None,
     detail_region_info: Optional[str] = None,
+    official_site_url: Optional[str] = None,
 ) -> Optional[SsulContent]:
     """크롤링으로 얻은 업장 정보를 채운다. **이미 있는 값은 덮지 않는다.**
 
     사용자가 검색으로 직접 고른 값이 크롤링 추정치보다 정확하므로 우선한다.
+
+    `official_site_url` 만 예외로 덮어쓴다 — 호출부가 플레이스 URL 을 먼저 폴백으로
+    넣어 두고 크롤링에 성공하면 진짜 홈페이지로 승급시키기 때문이다.
     """
     row = await session.get(SsulContent, content_id)
     if row is None:
@@ -249,9 +251,12 @@ async def set_place_info(
         row.region = region
     if detail_region_info and not row.detail_region_info:
         row.detail_region_info = detail_region_info
+    if official_site_url:
+        row.official_site_url = official_site_url[:2048]
     logger.info(
         f"[set_place_info] id={content_id} store={row.store_name!r} "
-        f"region={row.region!r} detail={(row.detail_region_info or '')[:30]!r}"
+        f"region={row.region!r} detail={(row.detail_region_info or '')[:30]!r} "
+        f"site={(row.official_site_url or '')[:60]!r}"
     )
     return row
 

app/ssulbox/worker/job_manager.py

@@ -144,14 +144,18 @@ async def _update_step(content_id: int, step: int) -> None:
 
 async def _get_place_info(
     content_id: int,
-) -> tuple[Optional[str], Optional[str], Optional[str]]:
-    """현재 저장된 (store_name, region, detail_region_info)."""
+) -> tuple[Optional[str], Optional[str], Optional[str], Optional[str]]:
+    """현재 저장된 (store_name, region, detail_region_info, official_site_url)."""
     async with BackgroundSessionLocal() as session:
         return await task_service.get_place_info(session, content_id)
 
 
 async def _save_place(
-    content_id: int, store_name: str, region: str, detail: str
+    content_id: int,
+    store_name: str,
+    region: str,
+    detail: str,
+    official_site_url: Optional[str] = None,
 ) -> None:
     async with BackgroundSessionLocal() as session:
         await task_service.set_place_info(
@@ -160,6 +164,7 @@ async def _save_place(
             store_name=store_name,
             region=region,
             detail_region_info=detail,
+            official_site_url=official_site_url,
         )
         await session.commit()
 
@@ -249,23 +254,29 @@ def _is_place_url(text: str) -> bool:
 
 
 def _collect_place_info(content_id: int, job: dict) -> None:
-    """비어 있는 업장명·지역을 크롤링으로 채운다(있는 값은 유지).
+    """비어 있는 업장명·지역·공식 링크를 크롤링으로 채운다(있는 값은 유지).
 
-    **빠진 값이 있을 때만 크롤링한다.** 검색으로 고른 경우 create 시점에 둘 다
-    채워져 있으므로 브라우저를 띄우지 않는다. 다만 검색 경로에서도 지역만 비는
-    경우(주소에서 시/군을 못 뽑는 등)가 있어, 존재 여부를 실제로 확인하고 판단한다.
+    **빠진 값이 있을 때만 크롤링한다.** 검색으로 고른 경우 create 시점에 업장명·지역이
+    채워져 있으나 공식 링크는 늘 비어 있으므로, 그 경로에서도 이 함수가 크롤링한다
+    (Playwright 1회, 최대 120초). 링크를 못 얻어도 place URL 폴백은 남는다.
 
     수집 실패는 삼킨다. 이 정보가 없어도 생성은 place_url 만으로 진행된다.
     """
-    if not _is_place_url(job.get("input", "")):
-        return
+    place_url = job.get("input", "")
+    if not _is_place_url(place_url):
+        return  # 업장명 해석 실패 — 링크로 쓸 값이 없다
     try:
-        store_name, region, detail = _run_db(_get_place_info(content_id))
-        if store_name and region and detail:
+        store_name, region, detail, site_url = _run_db(_get_place_info(content_id))
+        if store_name and region and detail and site_url:
             return  # 채울 것이 없다
 
+        # 크롤링이 실패해도 오버레이가 뜨도록 place URL 을 먼저 폴백으로 저장한다.
+        # castad 가 `official_site_url or 크롤링 소스 URL` 로 폴백하는 것과 같은 규칙.
+        if not site_url:
+            _run_db(_save_place(content_id, "", "", "", place_url))
+
         detail = _run_db(
-            place_service.fetch_place_detail(job["input"]),
+            place_service.fetch_place_detail(place_url),
             # Playwright 기동 + 상세 파싱까지 DB 위임 기본 타임아웃(60s)보다 길 수 있다
             timeout=120,
         )
@@ -275,14 +286,16 @@ def _collect_place_info(content_id: int, job: dict) -> None:
         title = detail.get("title") or ""
         road = detail.get("roadAddress") or ""
         jibun = detail.get("address") or ""
+        homepage = detail.get("homepage") or ""
         # castad `/home/crawl` 과 동일하게 **도로명·지번을 모두** 넘긴다.
         # 도로명에서 시/군 추출이 실패하면 지번으로 재시도한다(한쪽만 넘기면 놓친다).
         new_region = extract_region_from_address(road or None, jibun or None)
         new_detail = road or jibun  # 도로명 우선, 없으면 지번
         if title:
             job["store_name"] = title
-        if title or new_region or new_detail:
-            _run_db(_save_place(content_id, title, new_region, new_detail))
+        if title or new_region or new_detail or homepage:
+            # homepage 가 있으면 위에서 넣어 둔 place URL 폴백을 진짜 홈페이지로 승급시킨다.
+            _run_db(_save_place(content_id, title, new_region, new_detail, homepage))
     except Exception as e:
         logger.warning(
             f"[ssul {content_id}] 업장 정보 수집 실패(생성은 계속): "

app/utils/nvMapPwScraper.py

@@ -131,7 +131,8 @@ if (originalQuery) {
         cls,
         place_id: str,
         payloads: list[dict],
-    ) -> list[dict | None] | None:
+        capture_apollo: bool = False,
+    ) -> list[dict | None] | None | tuple[list[dict | None] | None, str | None]:
         """실제 브라우저로 네이버 WTM 안티봇 캡차를 통과해 GraphQL 쿼리를 실행한다.
 
         네이버 pcmap GraphQL은 두 헤더를 검사한다:
@@ -144,12 +145,19 @@ if (originalQuery) {
         Args:
             place_id: 네이버 place ID
             payloads: GraphQL POST 본문 목록
+            capture_apollo: True면 place 페이지에 인라인된 __APOLLO_STATE__
+                JSON 문자열도 함께 캡처해 (results, apollo_json) 튜플로 반환.
+                (GraphQL base가 노출하지 않는 homepages 등 SSR 캐시 전용 필드용)
         Returns:
             payload별 파싱 JSON 목록(실패 항목은 None). 토큰 캡처 실패 시 None.
+            capture_apollo=True면 (위 결과, apollo_json 또는 None) 튜플.
         """
+        def _ret(results, apollo=None):
+            return (results, apollo) if capture_apollo else results
+
         if not cls.is_ready:
             logger.warning("[NvMapPwScraper] fetch_graphql: scraper가 초기화되지 않았습니다")
-            return None
+            return _ret(None)
 
         page = await cls._new_stealth_page()
         captured: dict = {}
@@ -185,7 +193,17 @@ if (originalQuery) {
 
             if not captured.get("tok"):
                 logger.warning("[NvMapPwScraper] WTM 토큰 캡처 실패")
-                return None
+                return _ret(None)
+
+            apollo_json: str | None = None
+            if capture_apollo:
+                try:
+                    apollo_json = await page.evaluate(
+                        "() => { try { return JSON.stringify(window.__APOLLO_STATE__ || null); }"
+                        " catch (e) { return null; } }"
+                    )
+                except Exception as e:
+                    logger.warning(f"[NvMapPwScraper] __APOLLO_STATE__ 캡처 실패: {e}")
 
             headers = {
                 "Content-Type": "application/json",
@@ -220,11 +238,11 @@ if (originalQuery) {
                     results.append(None)
                 else:
                     results.append(r)
-            return results
+            return _ret(results, apollo_json)
 
         except Exception as e:
             logger.error(f"[NvMapPwScraper] fetch_graphql 오류: {e}")
-            return None
+            return _ret(None)
         finally:
             await page.close()
     

app/utils/nvMapScraper.py

@@ -112,6 +112,7 @@ query getVisitorReviewStats($id: String!) {
         self.facility_info: str | None = None
         self.voted_keyword_stats: list[dict] | None = None  # 키워드 투표 집계 (displayName, count)
         self.menu_info: list[dict] | None = None  # 메뉴 목록 (name, price, description, recommend)
+        self.official_site_url: str | None = None  # 업체 공식 링크 (base.homepages 대표 URL)
 
     def _get_request_headers(self) -> dict:
         headers = self.DEFAULT_HEADERS.copy()
@@ -256,9 +257,11 @@ query getVisitorReviewStats($id: String!) {
         #     self.scrap_type = "GraphQL-Browser"
 
         # ── 실제 브라우저로 WTM 캡차 우회 ──
-        data, stats_data, extra_photo_urls, biz_photo_urls = await self._scrap_via_browser(place_id)
+        data, stats_data, extra_photo_urls, biz_photo_urls, homepage_url = await self._scrap_via_browser(place_id)
         # 편의시설은 HTML 페이지 파싱이라 GraphQL(WTM) 차단과 별개로 직접 시도 (best-effort, 실패 시 None)
-        fac_data = await self._get_facility_string(place_id)
+        # 홈페이지 링크는 브라우저 캡처가 실패한 경우에만 HTML 경로로 보충한다.
+        fac_data, html_homepage = await self._get_facility_and_homepage(place_id)
+        homepage_url = homepage_url or html_homepage
         self.scrap_type = "GraphQL-Browser"
 
         self.rawdata = data
@@ -297,14 +300,43 @@ query getVisitorReviewStats($id: String!) {
         self.facility_info = fac_data
         self.voted_keyword_stats = stats_data
         self.menu_info = business.get("menus") or None
+        self.official_site_url = homepage_url
 
         return
 
-    async def _scrap_via_browser(self, place_id: str) -> tuple[dict, list[dict] | None, list[dict], list[dict]]:
+    @staticmethod
+    def _extract_homepage_from_html(html: str) -> str | None:
+        """플레이스 페이지 HTML의 __APOLLO_STATE__에서 홈페이지 링크를 추출한다.
+
+        GraphQL placeDetail(base)는 homepages 필드를 노출하지 않으므로(400),
+        SSR 페이지에 인라인된 Apollo 캐시의 "homepages" 객체를 직접 파싱한다.
+        대표(repr) 링크 우선, 죽은 링크(isDeadUrl)는 제외. 홈페이지 항목은
+        자체 홈페이지 외에 인스타그램/블로그 등일 수도 있다 — 업체가 대표로
+        등록한 링크를 그대로 신뢰한다.
+        """
+        decoder = json.JSONDecoder()
+        search_from = 0
+        while True:
+            idx = html.find('"homepages":', search_from)
+            if idx == -1:
+                return None
+            search_from = idx + 1
+            try:
+                homepages, _ = decoder.raw_decode(html, idx + len('"homepages":'))
+            except ValueError:
+                continue
+            if not isinstance(homepages, dict):
+                continue
+            candidates = [homepages.get("repr"), *(homepages.get("etc") or [])]
+            for item in candidates:
+                if isinstance(item, dict) and item.get("url") and not item.get("isDeadUrl"):
+                    return item["url"]
+
+    async def _scrap_via_browser(self, place_id: str) -> tuple[dict, list[dict] | None, list[dict], list[dict], str | None]:
         """직접 호출이 WTM 캡차에 막힌 경우, 실제 브라우저로 GraphQL을 호출한다.
 
         Returns:
-            (overview_data, review_stats_details, extra_photo_urls, biz_photo_urls)
+            (overview_data, review_stats_details, extra_photo_urls, biz_photo_urls, homepage_url)
 
         Raises:
             GraphQLException: 브라우저 폴백마저 실패한 경우
@@ -382,10 +414,14 @@ query getVisitorReviewStats($id: String!) {
         payloads = [overview_payload, stats_payload, interior_payload, exterior_payload, review_payload, *biz_payloads]
         MAX_RETRY = 3
         results = None
+        apollo_json: str | None = None
         last_error: Exception | None = None
         for attempt in range(1, MAX_RETRY + 1):
             try:
-                results = await NvMapPwScraper.fetch_graphql(place_id, payloads)
+                results, captured_apollo = await NvMapPwScraper.fetch_graphql(
+                    place_id, payloads, capture_apollo=True
+                )
+                apollo_json = apollo_json or captured_apollo
             except Exception as e:
                 last_error = e
                 logger.warning(f"[NvMapScraper] 브라우저 폴백 시도 {attempt}/{MAX_RETRY} 오류: {e}")
@@ -439,8 +475,17 @@ query getVisitorReviewStats($id: String!) {
             f"리뷰:{len(review_urls)} / 업체(biz):{len(biz_urls)}"
         )
 
-        logger.info(f"[NvMapScraper] 브라우저 폴백 SUCCESS - place_id: {place_id}")
-        return data, stats_data, extra_photo_urls, biz_urls
+        # 홈페이지 링크: GraphQL base는 homepages를 노출하지 않으므로(400),
+        # 브라우저가 로드한 place 페이지의 __APOLLO_STATE__ JSON에서 추출한다.
+        homepage_url = (
+            self._extract_homepage_from_html(apollo_json) if apollo_json else None
+        )
+
+        logger.info(
+            f"[NvMapScraper] 브라우저 폴백 SUCCESS - place_id: {place_id}, "
+            f"homepage: {homepage_url or '없음'}"
+        )
+        return data, stats_data, extra_photo_urls, biz_urls, homepage_url
 
     async def _call_get_accommodation(self, place_id: str) -> dict:
         """GraphQL API를 호출하여 숙소 정보를 가져옵니다.
@@ -521,29 +566,39 @@ query getVisitorReviewStats($id: String!) {
             logger.warning(f"[NvMapScraper] Failed to get review stats: {e}")
             return None
 
-    async def _get_facility_string(self, place_id: str) -> str | None:
-        """장소 페이지에서 편의시설 정보를 크롤링합니다. 숙소, 음식점 순으로 시도합니다.
+    async def _get_facility_and_homepage(self, place_id: str) -> tuple[str | None, str | None]:
+        """장소 페이지에서 편의시설 정보와 홈페이지 링크를 크롤링합니다. 숙소, 음식점 순으로 시도합니다.
 
         Args:
             place_id: 네이버 지도 장소 ID
 
         Returns:
-            편의시설 정보 문자열 또는 None
+            (편의시설 정보 문자열 또는 None, 홈페이지 링크 또는 None)
         """
+        facility: str | None = None
+        homepage: str | None = None
         place_types = ["place", "accommodation", "restaurant"]
         try:
             async with aiohttp.ClientSession() as session:
                 for place_type in place_types:
                     url = f"https://pcmap.place.naver.com/{place_type}/{place_id}/home"
                     async with session.get(url, headers=self._get_request_headers()) as response:
-                        soup = bs4.BeautifulSoup(await response.read(), "html.parser")
+                        raw = await response.read()
+                    if homepage is None:
+                        homepage = self._extract_homepage_from_html(
+                            raw.decode("utf-8", errors="ignore")
+                        )
+                    if facility is None:
+                        soup = bs4.BeautifulSoup(raw, "html.parser")
                         c_elem = soup.find("span", "place_blind", string="편의")
                         if c_elem:
-                            return c_elem.parent.parent.find("div").string
-            return None
+                            facility = c_elem.parent.parent.find("div").string
+                    if facility is not None and homepage is not None:
+                        break
+            return facility, homepage
         except Exception as e:
-            logger.warning(f"[NvMapScraper] Failed to get facility info: {e}")
-            return None
+            logger.warning(f"[NvMapScraper] Failed to get facility/homepage info: {e}")
+            return facility, homepage
 
 
 # if __name__ == "__main__":

app/utils/upload_blob_as_request.py

@@ -90,6 +90,23 @@ async def close_shared_blob_client() -> None:
         logger.info("[AzureBlobUploader] Shared HTTP client closed")
 
 
+def to_playback_url(blob_url: str | None) -> str | None:
+    """DB에 저장된 SAS 미포함 공개 URL에 읽기용 SAS 토큰을 붙여 반환합니다.
+
+    Azure Blob 익명(공개) 접근은 x-ms-version 헤더가 없어 Range 요청을 지원하지
+    않는 구버전 API로 처리되어 브라우저에서 영상/오디오 seek이 동작하지 않는다.
+    SAS 토큰(sv= 포함)을 붙이면 최신 API 버전으로 처리되어 Range/Accept-Ranges가
+    정상 동작한다. DB에는 SAS 없는 URL을 그대로 저장하고, 응답 시점에만 붙인다.
+    """
+    if not blob_url:
+        return blob_url
+    sas_token = azure_blob_settings.AZURE_BLOB_SAS_TOKEN.strip("?'\"")
+    if not sas_token:
+        return blob_url
+    separator = "&" if "?" in blob_url else "?"
+    return f"{blob_url}{separator}{sas_token}"
+
+
 class AzureBlobUploader:
     """Azure Blob Storage 업로드 클래스
 

app/video/api/routers/v1/video.py

@@ -30,6 +30,7 @@ from app.home.api.routers.v1.home import _extract_region_from_address
 from app.lyric.models import Lyric
 from app.song.models import Song, SongTimestamp
 from app.utils.creatomate import CreatomateService, LANGUAGE_FONT_MAP
+from app.utils.upload_blob_as_request import to_playback_url
 
 from app.database.like_cache import (
     backfill_user_set,
@@ -79,6 +80,55 @@ VIDEO_CREDIT_COST = 1
 router = APIRouter(prefix="/video", tags=["Video"])
 
 
+def _place_id_to_site_url(place_id: str | None) -> str | None:
+    """MarketingIntel.place_id("nv{네이버 place ID}")를 네이버 플레이스 URL로 변환한다.
+
+    크롤링 없이 직접 입력된 업체는 place_id가 없으므로 None을 반환한다.
+    """
+    if place_id and place_id.startswith("nv") and place_id[2:].isdigit():
+        return f"https://map.naver.com/p/entry/place/{place_id[2:]}"
+    return None
+
+
+async def _get_official_site_urls(
+    session: AsyncSession, projects: list[Project]
+) -> dict[int, str | None]:
+    """프로젝트 목록에 대해 {project_id: 공식 페이지 URL(or None)}을 일괄 조회한다.
+
+    Project.marketing_intelligence(문자열로 저장된 MarketingIntel.id)를 경유해
+    저장된 official_site_url을 우선 사용하고, 컬럼 도입 전 기존 행은
+    place_id 기반 네이버 플레이스 URL로 폴백한다.
+    """
+    m_id_by_project: dict[int, int] = {}
+    for p in projects:
+        try:
+            if p.marketing_intelligence is not None:
+                m_id_by_project[p.id] = int(p.marketing_intelligence)
+        except (TypeError, ValueError):
+            continue
+
+    url_by_project: dict[int, str | None] = {p.id: None for p in projects}
+    if not m_id_by_project:
+        return url_by_project
+
+    rows = (
+        await session.execute(
+            select(
+                MarketingIntel.id,
+                MarketingIntel.place_id,
+                MarketingIntel.official_site_url,
+            ).where(MarketingIntel.id.in_(set(m_id_by_project.values())))
+        )
+    ).all()
+    intel_by_m_id = {m_id: (place_id, site_url) for m_id, place_id, site_url in rows}
+
+    for project_id, m_id in m_id_by_project.items():
+        place_id, site_url = intel_by_m_id.get(m_id, (None, None))
+        url_by_project[project_id] = site_url or _place_id_to_site_url(place_id)
+    return url_by_project
+
+
+
 
 @router.get(
     "/generate/{task_id}",
@@ -892,7 +942,7 @@ async def download_video(
             store_name=project.store_name if project else None,
             region=project.region or _extract_region_from_address(project.detail_region_info) if project else None,
             task_id=task_id,
-            result_movie_url=video.result_movie_url,
+            result_movie_url=to_playback_url(video.result_movie_url),
             created_at=video.created_at,
         )
 
@@ -962,7 +1012,7 @@ async def get_all_videos(
                     type=it.ctype,
                     video_id=it.id,
                     store_name=it.store_name,
-                    result_movie_url=it.movie_url,
+                    result_movie_url=to_playback_url(it.movie_url),
                     poster_url=it.poster_url,
                     title=it.title,
                     description=it.description,
@@ -1203,10 +1253,12 @@ async def get_video_detail(
                     liked = False
             is_liked_by_me = liked
 
+        official_site_url_map = await _get_official_site_urls(session, [project])
+
         logger.info(f"[get_video_detail] SUCCESS - video_id: {video_id}")
         return VideoDetailResponse(
             video_id=video.id,
-            result_movie_url=video.result_movie_url,
+            result_movie_url=to_playback_url(video.result_movie_url),
             poster_url=video.poster_url,
             store_name=project.store_name,
             region=project.region or _extract_region_from_address(project.detail_region_info),
@@ -1215,6 +1267,7 @@ async def get_video_detail(
             created_at=video.created_at,
             like_count=like_count,
             is_liked_by_me=is_liked_by_me,
+            official_site_url=official_site_url_map.get(project.id),
         )
 
     except HTTPException:

app/video/schemas/video_schema.py

@@ -207,6 +207,10 @@ class VideoThumbnailItem(BaseModel):
     like_count: int = Field(..., description="좋아요 수")
     is_liked_by_me: bool = Field(..., description="현재 로그인 사용자가 좋아요를 눌렀는지 (비로그인은 항상 false)")
     comment_count: int = Field(..., description="댓글 수 (대댓글 포함)")
+    official_site_url: Optional[str] = Field(
+        None,
+        description="업체 공식 링크 (플레이스 홈페이지 항목 우선, 없으면 크롤링 소스 URL; 직접 입력 생성 영상만 null)",
+    )
 
 
 class VideoDetailResponse(BaseModel):
@@ -226,6 +230,10 @@ class VideoDetailResponse(BaseModel):
     created_at: datetime = Field(..., description="생성 일시")
     like_count: int = Field(..., description="좋아요 수")
     is_liked_by_me: bool = Field(..., description="현재 로그인 사용자가 좋아요를 눌렀는지 (비로그인은 항상 false)")
+    official_site_url: Optional[str] = Field(
+        None,
+        description="업체 공식 링크 (플레이스 홈페이지 항목 우선, 없으면 크롤링 소스 URL; 직접 입력 생성 영상만 null)",
+    )
 
 
 class LikeToggleResponse(BaseModel):

docs/database-schema/migration_2026_08_24_ssul_official_site_url.sql

@@ -0,0 +1,21 @@
+-- ============================================================
+-- Migration: ssul_content.input → official_site_url 로 전환
+-- Date: 2026-08-24
+-- Description: 썰박스 상세에도 ADO2 영상과 동일한 '공식 링크 오버레이'를 적용한다.
+--              `input`(요청 원본: place URL 또는 업장명)은 INSERT 후 어디서도
+--              읽지 않는 쓰기 전용 컬럼이었다 — 워커는 인메모리 job dict 를 쓴다.
+--              그 자리를 공식 링크(플레이스 홈페이지 우선, 없으면 플레이스 URL)로 돌린다.
+--              관련 코드: app/ssulbox/worker/job_manager.py(_collect_place_info),
+--                        app/ssulbox/services/place_service.py(_extract_detail_from_page)
+-- ============================================================
+
+ALTER TABLE `ssul_content`
+    CHANGE COLUMN `input` `official_site_url` VARCHAR(2048) NULL
+    COMMENT '업체 공식 링크 (플레이스 홈페이지 항목 우선, 없으면 네이버 플레이스 URL; 미확보 시 NULL)';
+
+-- ⚠️ 반드시 함께 실행할 것.
+-- CHANGE 는 기존 `input` 값을 새 컬럼으로 그대로 옮긴다. 업장명만 입력해 만든 행은
+-- URL 이 아닌 문자열("스테이 머뭄" 등)을 갖게 되고, 그대로 두면 상세 화면의 링크
+-- href 에 그 값이 들어간다. 기존 콘텐츠는 오버레이 대상이 아니므로(신규 생성분부터
+-- 적용) 전부 비운다.
+UPDATE `ssul_content` SET `official_site_url` = NULL;

generator/naver.py

@@ -339,6 +339,11 @@ def _parse_ai_briefing(txt):
 
 
 AI_BRIEFING_TIMEOUT = 18     # getAiBriefing 응답 대기(초). 정상은 ~6s, 이 안에 안 오면 스로틀/지연으로 본다.
+# place 홈 진입 후 getAiBriefing '요청' 발사를 기다리는 창(초).
+# 브리핑이 없는 업체엔 네이버가 요청 자체를 만들지 않는다 — 실측(2026-08-24, zzz/probe_ai_briefing.py):
+# 브리핑 있는 업체는 진입 0.8초 뒤 발사·1.1초에 응답, 없는 업체는 45초를 봐도 0회.
+# 이 창을 넘기면 '없음'이 확정이므로 응답을 기다릴 이유도, 새 세션으로 재시도할 이유도 없다.
+AI_BRIEFING_REQUEST_WINDOW = 6
 AI_BRIEFING_RETRIES = 2      # 응답이 끝내 안 올 때(kind='timeout') 새 세션으로 재시도할 횟수.
 AI_BRIEFING_RETRY_DELAY = 4  # 재시도 전 대기(초) — 연속 타격으로 스로틀이 굳는 걸 늦춘다.
 
@@ -348,8 +353,8 @@ async def _fetch(url, headful=False):
     끝내 안 오면(kind='timeout', 스로틀/지연 추정) 새 세션으로 최대 AI_BRIEFING_RETRIES 회 재시도한다.
     반환은 공개 API 형식인 3-튜플 (store, bullets, cat) — 내부 kind 는 여기서 소비.
       kind='briefing' → 브리핑 확보(bullets 채움)
-      kind='none'     → 브리핑 없는 업체 확정(bullets=None → 호출부가 리뷰 폴백)
-      kind='timeout'  → 응답 보류(재시도 가치 있음); 소진 시 bullets=None(→ 리뷰 폴백)."""
+      kind='none'     → 브리핑 없는 업체 확정(요청 미발사 또는 빈 응답; bullets=None → 호출부가 리뷰 폴백)
+      kind='timeout'  → 요청은 나갔으나 응답 보류(재시도 가치 있음); 소진 시 bullets=None(→ 리뷰 폴백)."""
     last_store, last_cat = None, None
     for i in range(AI_BRIEFING_RETRIES + 1):
         store, bullets, cat, kind = await _with_profile_retry(
@@ -369,15 +374,16 @@ async def _fetch_once(url, profile, headful=False):
     """네이버 플레이스 진입 → getAiBriefing GraphQL 응답을 가로채 브리핑을 파싱.
     반환 4-튜플 (store, bullets, cat, kind):
       kind='briefing' → bullets=[불릿...]  (브리핑 존재)
-      kind='none'     → bullets=None       (aiBriefing null/빈 = 브리핑 없는 업체)
-      kind='timeout'  → bullets=None       (응답이 AI_BRIEFING_TIMEOUT 내 안 옴 = 스로틀/지연)."""
+      kind='none'     → bullets=None       (요청 미발사 또는 aiBriefing null/빈 = 브리핑 없는 업체)
+      kind='timeout'  → bullets=None       (요청은 나갔으나 응답이 AI_BRIEFING_TIMEOUT 내 안 옴 = 스로틀/지연)."""
     async with async_playwright() as p:
         browser, ctx, page = await _new_page(p, headful=headful, profile=profile)
 
-        # getAiBriefing 응답을 진입 전에 미리 리스닝(진입 직후 발사되므로 놓치지 않게).
+        # getAiBriefing 요청/응답을 진입 전에 미리 리스닝(진입 직후 발사되므로 놓치지 않게).
         #   parsed: [불릿...] 또는 [](없음 확정) 이면 done, None(무관/파싱실패)이면 계속 대기.
         captured = {"bullets": None}
         done = asyncio.Event()
+        requested = asyncio.Event()     # getAiBriefing 요청이 실제로 나갔는가
 
         async def on_response(resp):
             if done.is_set() or not _is_ai_briefing_response(resp):
@@ -392,6 +398,13 @@ async def _fetch_once(url, profile, headful=False):
             captured["bullets"] = parsed                # [불릿...] 또는 [](없음) → 확정
             done.set()
 
+        def on_request(req):
+            if requested.is_set() or "graphql" not in req.url or req.method != "POST":
+                return
+            if "getAiBriefing" in (req.post_data or ""):
+                requested.set()
+
+        page.on("request", on_request)
         page.on("response", lambda r: asyncio.ensure_future(on_response(r)))
 
         try:
@@ -412,7 +425,15 @@ async def _fetch_once(url, profile, headful=False):
             await browser.close()
             raise
 
-        # getAiBriefing 응답을 기다린다(스크롤 불필요 — 진입 직후 XHR 로 발사됨).
+        # 요청이 창 안에 안 나가면 브리핑 미제공 업체로 확정한다. 예전에는 이 경우도
+        # 'timeout' 으로 보고 새 세션 재시도를 2회 돌아 업체당 55~65초를 버렸다.
+        try:
+            await asyncio.wait_for(requested.wait(), timeout=AI_BRIEFING_REQUEST_WINDOW)
+        except asyncio.TimeoutError:
+            await browser.close()
+            return store, None, cat, "none"
+
+        # 요청은 나갔다 → 응답을 기다린다(스크롤 불필요 — 진입 직후 XHR 로 발사됨).
         got = True
         try:
             await asyncio.wait_for(done.wait(), timeout=AI_BRIEFING_TIMEOUT)