feat(ssulbox): 업장 검색을 ADO2 방식으로 전환하고 URL 해석을 워커로 이관

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

@@ -5,10 +5,11 @@ castad 는 `/api/*` prefix 를 쓰지 않고 도메인별 prefix 를 쓰므로 `
 인증은 castad `get_current_user` 를 그대로 쓴다(원본의 auth 라우터·JWT 는 폐기).
 """
 
-from fastapi import APIRouter, Depends, Query
+from fastapi import APIRouter, Depends, HTTPException
-from sqlalchemy import select
+from sqlalchemy import func, select
 
 from app.credit.exceptions import InsufficientCreditError
+from app.database.like_cache import get_like_count, is_user_liked, set_like_count
 from app.database.session import AsyncSessionLocal, get_session
 from app.ssulbox.constants import ORPHAN_STATUSES, is_generation_available
 from app.ssulbox.exceptions import GenerationUnavailableError, TaskNotFoundError
@@ -17,13 +18,16 @@ from app.ssulbox.schemas.ssulbox_schema import (
     SsulActiveTasksResponse,
     SsulCreateRequest,
     SsulCreateResponse,
-    SsulPlaceSearchResponse,
+    SsulDeleteResponse,
+    SsulDetailResponse,
     SsulTaskStatus,
 )
-from app.ssulbox.services import place_service, task_service
+from app.ssulbox.services import task_service
 from app.ssulbox.worker import job_manager
-from app.user.dependencies.auth import get_current_user
+from app.user.dependencies.auth import get_current_user, get_current_user_optional
 from app.user.models import User
+# 좋아요는 castad video_reaction 에 병합돼 있다 (썰박스 행은 content_id 가 채워짐)
+from app.video.models import VideoReaction
 from app.utils.logger import get_logger
 from config import ssulbox_settings
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -33,33 +37,6 @@ logger = get_logger("ssulbox")
 router = APIRouter(prefix="/ssul", tags=["Ssulbox"])
 
 
-@router.get(
-    "/search/place",
-    response_model=SsulPlaceSearchResponse,
-    summary="업장 검색 (네이버 지도)",
-    description="""
-업장명으로 네이버 지도 후보를 검색합니다.
-
-castad `/search/accommodation`(네이버 **검색 API**)과 달리 **`place_url` 을 포함**합니다.
-생성 파이프라인이 네이버 지도 place 페이지를 크롤링하므로 이 URL 이 필요합니다.
-
-- 동명 업장은 주소로 구분됩니다.
-- 검색 실패·타임아웃 시 빈 목록을 반환합니다(사용자는 네이버 링크를 직접 붙여넣을 수 있습니다).
-- Playwright 크롤링이라 수 초 걸립니다.
-""",
-    responses={
-        200: {"description": "검색 성공 (결과 없음도 200)"},
-        401: {"description": "인증 실패"},
-    },
-)
-async def search_place(
-    query: str = Query(..., min_length=2, description="업장명"),
-    limit: int = Query(default=8, ge=1, le=20),
-    current_user: User = Depends(get_current_user),
-) -> SsulPlaceSearchResponse:
-    items = await place_service.search_places(query, limit=limit)
-    return SsulPlaceSearchResponse(query=query, count=len(items), items=items)
-
 
 @router.post(
     "/create",
@@ -120,8 +97,16 @@ async def create_ssul(
 
     # 커밋이 끝난 뒤에 큐잉한다 — 워커가 아직 없는 행을 조회하면 안 된다.
     # create_job 은 앱 이벤트 루프를 캡처하므로 반드시 요청 핸들러에서 호출한다.
+    # place URL 해석은 워커가 한다 — 10초 이상 걸려 요청을 막으면 안 된다.
     job_manager.create_job(
-        content_id, body.scenario, body.input, body.scenes, body.seconds
+        content_id,
+        body.scenario,
+        body.input,
+        body.scenes,
+        body.seconds,
+        store_name=body.store_name,
+        road_address=body.road_address,
+        address=body.address,
     )
 
     return SsulCreateResponse(
@@ -192,3 +177,124 @@ async def get_task(
     if row is None or row.user_uuid != current_user.user_uuid:
         raise TaskNotFoundError()
     return SsulTaskStatus.model_validate(row)
+
+
+@router.get(
+    "/{content_id}",
+    response_model=SsulDetailResponse,
+    summary="썰박스 콘텐츠 공개 상세",
+    description="""
+완성된 썰박스 콘텐츠의 상세 정보를 반환합니다. **비로그인도 접근 가능**합니다
+(공유 링크 `/ssul/{content_id}` 가 이 API 를 씁니다 — castad `/video/{video_id}` 와 동일한 패턴).
+
+- `status=done` 이고 삭제되지 않은 콘텐츠만 조회됩니다.
+- `is_liked_by_me` 는 비로그인이면 항상 false 입니다.
+- 댓글은 `GET /comment/video/{content_id}?type=ssul` 로 따로 조회합니다.
+""",
+    responses={
+        200: {"description": "조회 성공"},
+        404: {"description": "콘텐츠를 찾을 수 없음"},
+    },
+)
+async def get_content_detail(
+    content_id: int,
+    current_user: User | None = Depends(get_current_user_optional),
+    session: AsyncSession = Depends(get_session),
+) -> SsulDetailResponse:
+    row = await session.get(SsulContent, content_id)
+    if (
+        row is None
+        or row.is_deleted
+        or row.status != "done"
+        or not row.video_url
+    ):
+        raise TaskNotFoundError()
+
+    # 좋아요 수: Redis 우선, 미스면 DB 집계 후 캐시에 채운다 (castad 상세와 동일)
+    like_count = await get_like_count(content_id, ctype="ssul")
+    if like_count is None:
+        like_count = (
+            await session.execute(
+                select(func.count(VideoReaction.id)).where(
+                    VideoReaction.content_id == content_id
+                )
+            )
+        ).scalar() or 0
+        await set_like_count(content_id, like_count, ctype="ssul")
+
+    is_liked = False
+    if current_user:
+        cached = await is_user_liked(
+            content_id, current_user.user_uuid, ctype="ssul"
+        )
+        if cached is None:
+            # user-set 콜드 미스 — 응답 정확성만 필요하므로 DB 존재 확인으로 대체
+            is_liked = (
+                await session.execute(
+                    select(VideoReaction.id).where(
+                        VideoReaction.content_id == content_id,
+                        VideoReaction.user_uuid == current_user.user_uuid,
+                    )
+                )
+            ).scalar_one_or_none() is not None
+        else:
+            is_liked = cached
+
+    return SsulDetailResponse(
+        content_id=row.id,
+        scenario=row.scenario,
+        video_url=row.video_url,
+        store_name=row.store_name or None,
+        region=row.region,
+        created_at=row.created_at,
+        like_count=like_count,
+        is_liked_by_me=is_liked,
+    )
+
+
+@router.delete(
+    "/{content_id}",
+    response_model=SsulDeleteResponse,
+    summary="썰박스 콘텐츠 소프트 삭제",
+    description="""
+본인 소유의 썰박스 콘텐츠를 소프트 삭제합니다 (`is_deleted=True`, 데이터는 유지).
+castad `DELETE /archive/videos/{video_id}` 와 동일한 정책입니다.
+
+- 진행 중(queued/running)인 잡은 삭제할 수 없습니다 — 완료·실패 후에 지워야
+  워커·크레딧 환불 경로와 충돌하지 않습니다.
+""",
+    responses={
+        200: {"description": "삭제 성공"},
+        401: {"description": "인증 실패"},
+        404: {"description": "콘텐츠를 찾을 수 없음 (남의 것 포함)"},
+        409: {"description": "진행 중인 잡은 삭제 불가"},
+    },
+)
+async def delete_content(
+    content_id: int,
+    current_user: User = Depends(get_current_user),
+    session: AsyncSession = Depends(get_session),
+) -> SsulDeleteResponse:
+    row = await session.get(SsulContent, content_id)
+    # 남의 것이면 존재 여부를 노출하지 않고 동일하게 404
+    if row is None or row.is_deleted or row.user_uuid != current_user.user_uuid:
+        raise TaskNotFoundError()
+
+    # 진행 중인 잡을 지우면 워커가 완료 시점에 삭제된 행을 done 으로 되살리거나,
+    # 실패 환불 경로와 꼬인다. 터미널 상태에서만 허용한다.
+    if row.status in ("queued", "running"):
+        raise HTTPException(
+            status_code=409,
+            detail="생성이 진행 중입니다. 완료된 뒤에 삭제해주세요.",
+        )
+
+    row.is_deleted = True
+    await session.commit()
+    logger.info(
+        f"[delete_content] id={content_id} user={current_user.user_uuid}"
+    )
+    return SsulDeleteResponse(
+        success=True,
+        content_id=content_id,
+        message="콘텐츠가 삭제되었습니다.",
+    )

app/ssulbox/constants.py

@@ -24,6 +24,15 @@ SCENARIO_ENGINE: Final[dict[str, str]] = {
 
 ENGINE_SCENARIO: Final[dict[str, str]] = {v: k for k, v in SCENARIO_ENGINE.items()}
 
+#: 시나리오 한글 표시명. SNS 업로드 SEO 프롬프트 입력({scenario_name})에 쓴다.
+#: 프론트 i18n(ssulbox.scenario.*.name)과 값을 맞춘다.
+SCENARIO_NAMES: Final[dict[str, str]] = {
+    "joseon": "조선왕",
+    "samgukji": "삼국지",
+    "greek": "그리스·로마 신화",
+    "odyssey": "오디세이",
+}
+
 SCENARIOS: Final[tuple[str, ...]] = tuple(SCENARIO_ENGINE)
 
 

app/ssulbox/schemas/ssulbox_schema.py

@@ -10,27 +10,9 @@ from app.ssulbox.constants import SCENARIOS
 ScenarioLiteral = Literal["joseon", "samgukji", "greek", "odyssey"]
 
 
-# =============================================================================
+# 장소 검색 스키마는 두지 않는다 — 업장 검색은 ADO2 와 동일하게
-# 장소 검색
+# `/search/accommodation`(네이버 검색 API)을 쓰고, place URL 은 생성 요청 시
-# =============================================================================
+# 서버가 해석한다(2026-07-31 전환). 썰박스 전용 검색 엔드포인트는 제거됐다.
-class SsulPlaceItem(BaseModel):
-    """네이버 지도 검색 후보.
-
-    castad `/search/accommodation` 응답과 달리 **place_url 을 포함**한다.
-    generator 가 이 URL 로 place 페이지를 크롤링하기 때문이다.
-    """
-
-    title: str = Field(..., description="업장명")
-    category: str = Field(default="", description="업종")
-    address: str = Field(default="", description="표시용 주소 (동명 업장 구분에 사용)")
-    roadAddress: str = Field(default="", description="도로명 주소")
-    place_url: str = Field(..., description="네이버 지도 place URL (생성 파이프라인 입력)")
-
-
-class SsulPlaceSearchResponse(BaseModel):
-    query: str
-    count: int
-    items: list[SsulPlaceItem]
 
 
 # =============================================================================
@@ -48,13 +30,18 @@ class SsulCreateRequest(BaseModel):
         ...,
         min_length=2,
         max_length=500,
-        description="네이버 지도 place URL 또는 업장명",
+        description=(
+            "네이버 지도 place URL 또는 업장명. "
+            "자동완성으로 고른 경우 store_name/address 가 함께 오며, "
+            "그때는 서버가 ADO2 와 동일한 방식으로 place URL 을 해석한다."
+        ),
     )
     scenes: int = Field(default=9, ge=4, le=20, description="장면 수")
     seconds: int = Field(default=30, ge=20, le=90, description="장면당 초 길이")
 
-    # 검색으로 업장을 고른 경우 프론트가 함께 보낸다(`/ssul/search/place` 결과).
+    # 자동완성(`/search/accommodation`)으로 업장을 고른 경우 프론트가 함께 보낸다.
-    # 통합 목록의 업장명 표시와 store_name/region 필터가 이 값에 의존한다.
+    # 세 가지에 쓰인다: ① place URL 해석 ② 통합 목록의 업장명 표시
+    # ③ store_name/region 필터.
     # place URL 을 직접 붙여넣은 경우에는 없으며, 그때 store_name 은 생성 로그의
     # `■ 가게:` 마커로 뒤늦게 채운다(region 은 주소가 없어 채울 수 없다).
     store_name: str | None = Field(
@@ -105,13 +92,39 @@ class SsulActiveTasksResponse(BaseModel):
     items: list[SsulTaskStatus]
 
 
+class SsulDetailResponse(BaseModel):
+    """`GET /ssul/{content_id}` 공개 상세 응답.
+
+    castad `VideoDetailResponse` 와 같은 필드 구성에 `scenario` 만 더했다
+    (프론트가 시나리오 표지·라벨을 그리는 데 쓴다). 공유 링크로 들어온
+    **비로그인 사용자도 볼 수 있다** — `is_liked_by_me` 는 비로그인이면 항상 False.
+    """
+
+    content_id: int = Field(..., description="콘텐츠 고유 ID")
+    scenario: str = Field(..., description="시나리오 코드")
+    video_url: str = Field(..., description="완성 영상 URL")
+    store_name: Optional[str] = Field(None, description="업장명")
+    region: Optional[str] = Field(None, description="지역명")
+    created_at: datetime = Field(..., description="생성 일시")
+    like_count: int = Field(..., description="좋아요 수")
+    is_liked_by_me: bool = Field(..., description="현재 로그인 사용자가 좋아요를 눌렀는지")
+
+
+class SsulDeleteResponse(BaseModel):
+    """`DELETE /ssul/{content_id}` 응답. castad 삭제 응답과 같은 형태."""
+
+    success: bool
+    content_id: int
+    message: str
+
+
 __all__ = [
     "SCENARIOS",
     "ScenarioLiteral",
     "SsulActiveTasksResponse",
     "SsulCreateRequest",
     "SsulCreateResponse",
-    "SsulPlaceItem",
+    "SsulDeleteResponse",
-    "SsulPlaceSearchResponse",
+    "SsulDetailResponse",
     "SsulTaskStatus",
 ]

app/ssulbox/services/place_service.py

@@ -1,26 +1,27 @@
 # -*- coding: utf-8 -*-
-"""네이버 지도 업장 검색 (키 불필요, Playwright).
+"""네이버 지도 업장 정보 조회 (Playwright).
 
-`map.naver.com/p/search/{query}` 의 검색 리스트 iframe(`pcmap.place.naver.com/place/list`)
+두 가지를 한다:
-안에 있는 `__APOLLO_STATE__` 를 파싱해, 동명 업장들을 주소로 구분한 후보 목록을 만든다.
+  1. `resolve_place_url` — 업장명(+주소) → place URL.
-각 후보는 `place_url`(map.naver.com/p/entry/place/{id})을 가지므로, 사용자가 하나를 고르면
+     **castad `NvMapPwScraper.get_place_id_url` 을 재사용**한다(ADO2 자동완성과 동일 경로).
-그 URL 을 그대로 생성 파이프라인(generator/naver.py)에 넘겨 정확히 그 가게를 크롤링한다.
+  2. `fetch_place_detail` — place URL → 업장명·주소.
+     생성 시 store_name/region 을 채우는 데 쓴다. place 상세의 `__APOLLO_STATE__`
+     안 `PlaceDetailBase:{id}` 를 파싱한다.
 
-**castad `/search/accommodation` 으로 대체할 수 없다.** 그쪽은 네이버 *검색 API* 라
+**후보 목록 검색은 여기 없다.** 예전에는 지도 검색 iframe 을 크롤링해 후보를
-`title`/`address`/`roadAddress` 만 주고 `place_url` 이 없는데, generator 가 place 페이지를
+받았지만 **17초**가 걸려, ADO2 와 동일하게 네이버 검색 API(`/search/accommodation`,
-크롤링하므로 URL 이 반드시 필요하다. (castad `NvMapPwScraper` 는 후보=NAVER API,
+~0.2초)로 자동완성하고 place URL 은 제출 시 1건만 해석하도록 바꿨다(2026-07-31).
-place_id 해석=Playwright 로 2단계를 밟지만 여기서는 지도만으로 한 번에 얻는다.)
 """
 import asyncio
 import json
 import sys
-from urllib.parse import quote
 
 from playwright.async_api import async_playwright
 
 from app.utils.logger import get_logger
 # URL 정규화(단축링크 해석·place_id 추출)는 castad 크롤러 것을 그대로 쓴다.
 # 무거운 scrap() 은 쓰지 않는다 — fetch_place_detail docstring 참조.
+from app.utils.nvMapPwScraper import NvMapPwScraper
 from app.utils.nvMapScraper import NvMapScraper, URLNotFoundException
 
 logger = get_logger("ssulbox")
@@ -66,87 +67,42 @@ def _extract_apollo_state(html: str) -> dict | None:
     return None
 
 
-def _candidates_from_state(state: dict, limit: int) -> list[dict]:
+async def resolve_place_url(
-    out: list[dict] = []
+    title: str, address: str = "", road_address: str = ""
-    for key, obj in state.items():
+) -> str | None:
-        if not key.startswith("PlaceListBusinessesItem:"):
+    """업장명(+주소)으로 네이버 지도 place URL 을 해석한다.
-            continue
-        if not isinstance(obj, dict):
-            continue
-        name = obj.get("name")
-        pid = obj.get("id")
-        if not (name and pid and str(pid).isdigit()):
-            continue
-        out.append(
-            {
-                "title": name,
-                "category": obj.get("category") or "",
-                # 표시용 주소는 '서울 성동구 금호동3가' 같은 commonAddress 를 우선(동명 구분에 최적)
-                "address": obj.get("commonAddress") or obj.get("fullAddress") or "",
-                "roadAddress": obj.get("fullAddress") or obj.get("roadAddress") or "",
-                "place_url": f"https://map.naver.com/p/entry/place/{pid}",
-            }
-        )
-        if len(out) >= limit:
-            break
-    return out
 
+    **castad `NvMapPwScraper.get_place_id_url` 을 그대로 재사용한다** — ADO2 자동완성이
+    쓰는 바로 그 경로다(`app/home` 의 `_autocomplete_logic`). allSearch API 응답을
+    캡처해 이름·주소 유사도로 후보를 고르고, 실패하면 정제주소 → 업체명 단독 →
+    컨텍스트 재생성 순으로 3단 재시도한다.
 
-async def _search(query: str, limit: int) -> list[dict]:
+    썰박스 자체 지도 크롤링(`search_places`)을 쓰지 않는 이유: 그쪽은 후보 목록을
-    async with async_playwright() as p:
+    받으려고 iframe 렌더를 기다려 **17초**가 걸린다. ADO2 는 후보 탐색을 네이버
-        browser = await p.chromium.launch(
+    검색 API(0.2초)로 하고 여기서는 확정된 1건만 해석하므로 훨씬 빠르고,
-            headless=True,
+    안티봇 차단 대응까지 이미 들어 있다.
-            args=["--disable-blink-features=AutomationControlled", "--no-sandbox"],
+
+    실패하면 None. 호출부는 사용자가 링크를 직접 붙여넣는 우회 경로를 남겨둔다.
+    """
+    title = (title or "").strip()
+    if not title:
+        return None
+    try:
+        selected = {
+            "title": title,
+            "address": address or "",
+            "roadAddress": road_address or address or "",
+        }
+        async with NvMapPwScraper() as scraper:
+            url = await scraper.get_place_id_url(selected)
+        logger.info(f"[resolve_place_url] {title!r} → {url}")
+        return url
+    except Exception as e:
+        logger.error(
+            f"[resolve_place_url] FAILED {title!r} - {type(e).__name__}: {e}",
+            exc_info=True,
         )
-        try:
+        return None
-            ctx = await browser.new_context(
-                user_agent=DESKTOP_UA, locale="ko-KR", timezone_id="Asia/Seoul",
-                viewport={"width": 1280, "height": 800},
-                extra_http_headers={"Accept-Language": "ko-KR,ko;q=0.9"},
-            )
-            page = await ctx.new_page()
-            await page.goto(
-                f"https://map.naver.com/p/search/{quote(query)}",
-                wait_until="domcontentloaded", timeout=40000,
-            )
-            # 검색 리스트 iframe 이 뜰 때까지 대기(최대 ~12초)
-            frame = None
-            for _ in range(24):
-                for f in page.frames:
-                    if "pcmap.place.naver.com/place/list" in f.url:
-                        frame = f
-                        break
-                if frame:
-                    break
-                # 단일 결과면 곧바로 place 상세로 리다이렉트됨 → 후보 1개로 처리.
-                # 이때도 **주소를 반드시 채운다**. 예전에는 빈 문자열을 돌려줬는데,
-                # 그러면 프론트가 주소를 못 보내고 지역(region)이 NULL 이 되어
-                # 그 콘텐츠가 통합 목록의 지역 필터에서 영구 제외된다.
-                if "/place/" in page.url and "/search/" not in page.url:
-                    detail = await _extract_detail_from_page(page)
-                    if detail:
-                        detail["place_url"] = page.url
-                        return [detail]
-                    # 상세 파싱까지 실패하면 최소 정보라도 준다(생성은 가능해야 한다)
-                    return [{"title": query, "category": "", "address": "",
-                             "roadAddress": "", "place_url": page.url}]
-                await page.wait_for_timeout(500)
-            if not frame:
-                return []
-            # apollo state 가 채워질 시간을 조금 더 준다
-            html = await frame.content()
-            state = _extract_apollo_state(html)
-            for _ in range(6):
-                if state and any(k.startswith("PlaceListBusinessesItem:") for k in state):
-                    break
-                await page.wait_for_timeout(600)
-                html = await frame.content()
-                state = _extract_apollo_state(html)
-            if not state:
-                return []
-            return _candidates_from_state(state, limit)
-        finally:
-            await browser.close()
 
 
 def _detail_from_state(state: dict) -> dict | None:
@@ -284,44 +240,3 @@ async def fetch_place_detail(
         return None
 
 
-def _search_blocking(query: str, limit: int) -> list[dict]:
-    """별도 스레드에서 자체 이벤트 루프로 Playwright 실행.
-
-    Windows 의 웹서버 루프(Selector)는 서브프로세스를 못 띄워 Playwright 가 즉시 실패한다.
-    → 이 함수는 워커 스레드에서 Proactor 루프(윈도우) 를 새로 만들어 그 위에서 돌린다.
-    """
-    if sys.platform == "win32":
-        loop = asyncio.ProactorEventLoop()
-    else:
-        loop = asyncio.new_event_loop()
-    try:
-        return loop.run_until_complete(_search(query, limit))
-    finally:
-        loop.close()
-
-
-async def search_places(query: str, limit: int = 8, timeout: float = 35.0) -> list[dict]:
-    """업장명으로 네이버 지도 후보 목록을 반환. 실패/타임아웃이면 빈 리스트.
-
-    실패해도 예외를 올리지 않는다 — 사용자에게는 네이버 링크를 직접 붙여넣는
-    우회 경로가 있으므로 흐름을 막지 않는다. 다만 원본은 예외를 통째로 삼켜
-    디버깅이 불가능했으므로 로그는 남긴다.
-    """
-    query = (query or "").strip()
-    if len(query) < 2:
-        return []
-    try:
-        results = await asyncio.wait_for(
-            asyncio.to_thread(_search_blocking, query, limit), timeout=timeout
-        )
-        logger.info(f"[search_places] query='{query}' → {len(results)}건")
-        return results
-    except asyncio.TimeoutError:
-        logger.warning(f"[search_places] TIMEOUT ({timeout}s) query='{query}'")
-        return []
-    except Exception as e:
-        logger.error(
-            f"[search_places] FAILED query='{query}' - {type(e).__name__}: {e}",
-            exc_info=True,
-        )
-        return []

app/ssulbox/worker/job_manager.py

@@ -58,7 +58,15 @@ _loop: Optional[asyncio.AbstractEventLoop] = None
 
 
 def create_job(
-    content_id: int, scenario: str, input_text: str, scenes: int, seconds: int
+    content_id: int,
+    scenario: str,
+    input_text: str,
+    scenes: int,
+    seconds: int,
+    *,
+    store_name: str | None = None,
+    road_address: str | None = None,
+    address: str | None = None,
 ) -> None:
     """잡을 큐에 넣는다.
 
@@ -79,6 +87,11 @@ def create_job(
         "output": None,
         "job_dir": None,  # generator 가 stdout 으로 알려준다. 실패 정리 대상
         "store_name": None,  # 검색을 안 거친 경우 생성 로그에서 주워온다
+        # 자동완성으로 고른 값. place URL 해석(_resolve_input_url)에만 쓴다 —
+        # DB 저장은 create_task 가 이미 했으므로 여기서는 힌트일 뿐이다.
+        "store_name_hint": store_name,
+        "road_address_hint": road_address,
+        "address_hint": address,
         "error": None,
         "timings": {},
     }
@@ -262,6 +275,40 @@ def _collect_place_info(content_id: int, job: dict) -> None:
         )
 
 
+def _resolve_input_url(job: dict) -> None:
+    """업장명 입력을 네이버 지도 place URL 로 바꾼다(자동완성 선택 경로).
+
+    generator 는 place 페이지를 크롤링하므로 URL 이 있어야 정확한 가게를 잡는다.
+    해석은 castad `NvMapPwScraper` 경로를 재사용한다(ADO2 자동완성과 동일).
+
+    실패하면 원래 입력(업장명)을 그대로 둔다 — generator 가 업장명만으로도
+    네이버 브리핑을 시도한다. 정확도는 떨어지지만 생성을 막지는 않는다.
+    """
+    raw = (job.get("input") or "").strip()
+    store = (job.get("store_name_hint") or "").strip()
+    if not store or _is_place_url(raw):
+        return
+    try:
+        url = _run_db(
+            place_service.resolve_place_url(
+                store,
+                address=job.get("address_hint") or "",
+                road_address=job.get("road_address_hint") or "",
+            ),
+            # Playwright 기동 + 최대 3단 재시도라 DB 위임 기본 타임아웃보다 길다
+            timeout=120,
+        )
+        if url:
+            job["input"] = url
+            logger.info(f"[_resolve_input_url] {store!r} → {url}")
+        else:
+            logger.warning(f"[_resolve_input_url] 해석 실패, 업장명으로 진행 - {store!r}")
+    except Exception as e:
+        logger.warning(
+            f"[_resolve_input_url] 실패(업장명으로 진행): {type(e).__name__}: {e}"
+        )
+
+
 def _run(content_id: int) -> None:
     """워커 스레드 본체. subprocess 를 감독하며 stdout 마커로 진행을 파싱한다."""
     global _running
@@ -269,6 +316,23 @@ def _run(content_id: int) -> None:
     proc: Optional[subprocess.Popen] = None
 
     try:
+        _run_db(_mark_running(content_id))
+        job["status"] = "running"
+
+        # 업장명만 들어온 경우(자동완성 선택) place URL 로 바꾼다.
+        # **요청 핸들러가 아니라 여기서 하는 이유**: 해석에 10초 이상 걸려
+        # `/ssul/create` 응답이 그만큼 막히고 사용자가 폼에 묶인다.
+        # 워커로 옮기면 요청은 즉시 끝나고 진행 화면이 바로 뜬다.
+        _resolve_input_url(job)
+
+        # 업장명·주소를 **항상** 확보한다. 검색으로 고르지 않고 place URL 을
+        # 붙여넣은 경우 create 시점에 아무것도 없으므로 여기서 크롤링한다.
+        # 생성 **전에** 하는 이유: 목록에 이름 없는 카드가 뜨는 구간을 없앤다.
+        # 실패해도 생성은 계속한다(목록 표시·필터용 부가 정보다).
+        _collect_place_info(content_id, job)
+
+        # ⚠️ 커맨드는 위 두 단계가 끝난 **뒤에** 만든다 — job["input"] 이 갱신되므로
+        # 먼저 만들면 해석 전 값(업장명)이 그대로 엔진에 넘어간다.
         cmd, engine_dir = _build_command(job)
 
         env = dict(os.environ)
@@ -277,15 +341,6 @@ def _run(content_id: int) -> None:
         if key:
             env["GEMINI_API_KEY"] = key
 
-        _run_db(_mark_running(content_id))
-        job["status"] = "running"
-
-        # 업장명·주소를 **항상** 확보한다. 검색으로 고르지 않고 place URL 을
-        # 붙여넣은 경우 create 시점에 아무것도 없으므로 여기서 크롤링한다.
-        # 생성 **전에** 하는 이유: 목록에 이름 없는 카드가 뜨는 구간을 없앤다.
-        # 실패해도 생성은 계속한다(목록 표시·필터용 부가 정보다).
-        _collect_place_info(content_id, job)
-
         proc = subprocess.Popen(
             cmd,
             cwd=str(engine_dir),