UX improvements: mobile bottom sheet, cuisine taxonomy, search enhancements

- Add BottomSheet component for Google Maps-style restaurant detail on mobile
  (3-snap drag: 40%/55%/92%, velocity-based close, backdrop overlay)
- Mobile map mode now full-screen with bottom sheet overlay for details
- Collapsible filter panel on mobile with active filter badge count
- Standardized cuisine taxonomy (46 categories: 한식|국밥, 일식|스시 etc.)
  with LLM remap endpoint and admin UI button
- Enhanced search: keyword search now includes foods_mentioned + video title
- Search results include channels array for frontend filtering
- Channel filter moved to frontend filteredRestaurants (not API-level)
- LLM extraction prompt updated for pipe-delimited region + cuisine taxonomy
- Vector rebuild endpoint with rich JSON chunks per restaurant
- Geolocation-based auto region selection on page load
- Desktop filters split into two clean rows

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

backend/api/routes/daemon.py

@@ -0,0 +1,98 @@
+"""Daemon config & manual trigger API routes."""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends
+from pydantic import BaseModel
+
+from api.deps import get_admin_user
+from core.db import conn
+from core import cache
+
+router = APIRouter()
+
+
+class DaemonConfigUpdate(BaseModel):
+    scan_enabled: bool | None = None
+    scan_interval_min: int | None = None
+    process_enabled: bool | None = None
+    process_interval_min: int | None = None
+    process_limit: int | None = None
+
+
+@router.get("/config")
+def get_config():
+    """Get daemon config (read-only for all authenticated users)."""
+    with conn() as c:
+        cur = c.cursor()
+        cur.execute(
+            "SELECT scan_enabled, scan_interval_min, process_enabled, process_interval_min, "
+            "process_limit, last_scan_at, last_process_at, updated_at "
+            "FROM daemon_config WHERE id = 1"
+        )
+        row = cur.fetchone()
+        if not row:
+            return {}
+        return {
+            "scan_enabled": bool(row[0]),
+            "scan_interval_min": row[1],
+            "process_enabled": bool(row[2]),
+            "process_interval_min": row[3],
+            "process_limit": row[4],
+            "last_scan_at": str(row[5]) if row[5] else None,
+            "last_process_at": str(row[6]) if row[6] else None,
+            "updated_at": str(row[7]) if row[7] else None,
+        }
+
+
+@router.put("/config")
+def update_config(body: DaemonConfigUpdate, _admin: dict = Depends(get_admin_user)):
+    """Update daemon schedule config (admin only)."""
+    sets = []
+    params: dict = {}
+    if body.scan_enabled is not None:
+        sets.append("scan_enabled = :se")
+        params["se"] = 1 if body.scan_enabled else 0
+    if body.scan_interval_min is not None:
+        sets.append("scan_interval_min = :si")
+        params["si"] = body.scan_interval_min
+    if body.process_enabled is not None:
+        sets.append("process_enabled = :pe")
+        params["pe"] = 1 if body.process_enabled else 0
+    if body.process_interval_min is not None:
+        sets.append("process_interval_min = :pi")
+        params["pi"] = body.process_interval_min
+    if body.process_limit is not None:
+        sets.append("process_limit = :pl")
+        params["pl"] = body.process_limit
+    if not sets:
+        return {"ok": True}
+    sets.append("updated_at = SYSTIMESTAMP")
+    sql = f"UPDATE daemon_config SET {', '.join(sets)} WHERE id = 1"
+    with conn() as c:
+        c.cursor().execute(sql, params)
+    return {"ok": True}
+
+
+@router.post("/run/scan")
+def run_scan(_admin: dict = Depends(get_admin_user)):
+    """Manually trigger channel scan (admin only)."""
+    from core.youtube import scan_all_channels
+    new_count = scan_all_channels()
+    with conn() as c:
+        c.cursor().execute("UPDATE daemon_config SET last_scan_at = SYSTIMESTAMP WHERE id = 1")
+    if new_count > 0:
+        cache.flush()
+    return {"ok": True, "new_videos": new_count}
+
+
+@router.post("/run/process")
+def run_process(limit: int = 10, _admin: dict = Depends(get_admin_user)):
+    """Manually trigger video processing (admin only)."""
+    from core.pipeline import process_pending
+    rest_count = process_pending(limit=limit)
+    with conn() as c:
+        c.cursor().execute("UPDATE daemon_config SET last_process_at = SYSTIMESTAMP WHERE id = 1")
+    if rest_count > 0:
+        cache.flush()
+    return {"ok": True, "restaurants_extracted": rest_count}

backend/api/routes/search.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from fastapi import APIRouter, Query
 
-from core import restaurant, vector
+from core import restaurant, vector, cache
 from core.db import conn
 
 router = APIRouter()
@@ -17,8 +17,15 @@ def search_restaurants(
     limit: int = Query(20, le=100),
 ):
     """Search restaurants by keyword, semantic similarity, or hybrid."""
+    key = cache.make_key("search", f"q={q}", f"m={mode}", f"l={limit}")
+    cached = cache.get(key)
+    if cached is not None:
+        return cached
+
     if mode == "semantic":
-        return _semantic_search(q, limit)
+        result = _semantic_search(q, limit)
+        cache.set(key, result)
+        return result
     elif mode == "hybrid":
         kw = _keyword_search(q, limit)
         sem = _semantic_search(q, limit)
@@ -29,21 +36,31 @@ def search_restaurants(
             if r["id"] not in seen:
                 merged.append(r)
                 seen.add(r["id"])
-        return merged[:limit]
+        result = merged[:limit]
+        cache.set(key, result)
+        return result
     else:
-        return _keyword_search(q, limit)
+        result = _keyword_search(q, limit)
+        cache.set(key, result)
+        return result
 
 
 def _keyword_search(q: str, limit: int) -> list[dict]:
+    # JOIN video_restaurants to also search foods_mentioned and video title
     sql = """
-        SELECT id, name, address, region, latitude, longitude,
-               cuisine_type, price_range
-        FROM restaurants
-        WHERE latitude IS NOT NULL
-          AND (UPPER(name) LIKE UPPER(:q)
-               OR UPPER(address) LIKE UPPER(:q)
-               OR UPPER(region) LIKE UPPER(:q)
-               OR UPPER(cuisine_type) LIKE UPPER(:q))
+        SELECT DISTINCT r.id, r.name, r.address, r.region, r.latitude, r.longitude,
+               r.cuisine_type, r.price_range, r.google_place_id,
+               r.business_status, r.rating, r.rating_count
+        FROM restaurants r
+        JOIN video_restaurants vr ON vr.restaurant_id = r.id
+        JOIN videos v ON v.id = vr.video_id
+        WHERE r.latitude IS NOT NULL
+          AND (UPPER(r.name) LIKE UPPER(:q)
+               OR UPPER(r.address) LIKE UPPER(:q)
+               OR UPPER(r.region) LIKE UPPER(:q)
+               OR UPPER(r.cuisine_type) LIKE UPPER(:q)
+               OR UPPER(vr.foods_mentioned) LIKE UPPER(:q)
+               OR UPPER(v.title) LIKE UPPER(:q))
         FETCH FIRST :lim ROWS ONLY
     """
     pattern = f"%{q}%"
@@ -51,18 +68,56 @@ def _keyword_search(q: str, limit: int) -> list[dict]:
         cur = c.cursor()
         cur.execute(sql, {"q": pattern, "lim": limit})
         cols = [d[0].lower() for d in cur.description]
-        return [dict(zip(cols, row)) for row in cur.fetchall()]
+        rows = [dict(zip(cols, row)) for row in cur.fetchall()]
+
+    # Attach channel names
+    if rows:
+        _attach_channels(rows)
+    return rows
 
 
 def _semantic_search(q: str, limit: int) -> list[dict]:
-    similar = vector.search_similar(q, top_k=limit)
+    similar = vector.search_similar(q, top_k=max(30, limit * 3))
     if not similar:
         return []
 
-    rest_ids = list({s["restaurant_id"] for s in similar})
+    # Deduplicate by restaurant_id, preserving distance order (best first)
+    seen: set[str] = set()
+    ordered_ids: list[str] = []
+    for s in similar:
+        rid = s["restaurant_id"]
+        if rid not in seen:
+            seen.add(rid)
+            ordered_ids.append(rid)
+
     results = []
-    for rid in rest_ids[:limit]:
+    for rid in ordered_ids[:limit]:
         r = restaurant.get_by_id(rid)
         if r and r.get("latitude"):
             results.append(r)
+
+    if results:
+        _attach_channels(results)
     return results
+
+
+def _attach_channels(rows: list[dict]):
+    """Attach channel names to each restaurant dict."""
+    ids = [r["id"] for r in rows]
+    placeholders = ", ".join(f":id{i}" for i in range(len(ids)))
+    sql = f"""
+        SELECT DISTINCT vr.restaurant_id, c.channel_name
+        FROM video_restaurants vr
+        JOIN videos v ON v.id = vr.video_id
+        JOIN channels c ON c.id = v.channel_id
+        WHERE vr.restaurant_id IN ({placeholders})
+    """
+    params = {f"id{i}": rid for i, rid in enumerate(ids)}
+    ch_map: dict[str, list[str]] = {}
+    with conn() as c:
+        cur = c.cursor()
+        cur.execute(sql, params)
+        for row in cur.fetchall():
+            ch_map.setdefault(row[0], []).append(row[1])
+    for r in rows:
+        r["channels"] = ch_map.get(r["id"], [])

backend/api/routes/videos.py

@@ -9,11 +9,14 @@ import random
 import time
 from concurrent.futures import ThreadPoolExecutor
 
-from fastapi import APIRouter, Query
+from fastapi import APIRouter, Depends, Query
 from fastapi.responses import StreamingResponse
 
+from api.deps import get_admin_user
+
 from core.db import conn
 from core.pipeline import process_pending
+from core import cache
 
 logger = logging.getLogger(__name__)
 router = APIRouter()
@@ -23,11 +26,9 @@ _executor = ThreadPoolExecutor(max_workers=4)
 @router.get("")
 def list_videos(
     status: str | None = None,
-    limit: int = Query(50, le=500),
-    offset: int = Query(0, ge=0),
 ):
     conditions = []
-    params: dict = {"lim": limit, "off": offset}
+    params: dict = {}
     if status:
         conditions.append("v.status = :st")
         params["st"] = status
@@ -44,7 +45,6 @@ def list_videos(
         JOIN channels c ON c.id = v.channel_id
         {where}
         ORDER BY v.published_at DESC NULLS LAST
-        OFFSET :off ROWS FETCH NEXT :lim ROWS ONLY
     """
     with conn() as c:
         cur = c.cursor()
@@ -100,7 +100,7 @@ def bulk_extract_pending_count():
 
 
 @router.post("/bulk-extract")
-def bulk_extract():
+def bulk_extract(_admin: dict = Depends(get_admin_user)):
     """Process all unextracted videos with random delays. Streams SSE progress."""
     from core.pipeline import process_video_extract
 
@@ -131,6 +131,8 @@ def bulk_extract():
                 logger.error("Bulk extract error for %s: %s", v["video_id"], e)
                 yield f"data: {_json.dumps({'type': 'error', 'index': i, 'title': v['title'], 'message': str(e)})}\n\n"
 
+        if total_restaurants > 0:
+            cache.flush()
         yield f"data: {_json.dumps({'type': 'complete', 'total': total, 'total_restaurants': total_restaurants})}\n\n"
 
     return StreamingResponse(generate(), media_type="text/event-stream")
@@ -159,7 +161,7 @@ def bulk_transcript_pending_count():
 
 
 @router.post("/bulk-transcript")
-def bulk_transcript():
+def bulk_transcript(_admin: dict = Depends(get_admin_user)):
     """Fetch transcripts for all videos missing them. Streams SSE progress."""
     from core.youtube import get_transcript
 
@@ -196,11 +198,133 @@ def bulk_transcript():
                 logger.error("Bulk transcript error for %s: %s", v["video_id"], e)
                 yield f"data: {_json.dumps({'type': 'error', 'index': i, 'title': v['title'], 'message': str(e)})}\n\n"
 
+        if success > 0:
+            cache.flush()
         yield f"data: {_json.dumps({'type': 'complete', 'total': total, 'success': success})}\n\n"
 
     return StreamingResponse(generate(), media_type="text/event-stream")
 
 
+@router.post("/remap-cuisine")
+def remap_cuisine(_admin: dict = Depends(get_admin_user)):
+    """Remap all restaurant cuisine_type using LLM. Streams SSE progress."""
+    from core.cuisine import build_remap_prompt, CUISINE_TYPES, VALID_PREFIXES
+    from core.extractor import _llm, _parse_json
+    from core.db import conn as db_conn
+
+    BATCH = 20  # restaurants per LLM call (smaller for better accuracy)
+
+    def _apply_batch(batch: list[dict], valid_set: set[str]) -> tuple[int, list[dict]]:
+        """Run LLM on a batch. Returns (updated_count, missed_items)."""
+        prompt = build_remap_prompt(batch)
+        raw = _llm(prompt, max_tokens=4096)
+        result = _parse_json(raw)
+        if not isinstance(result, list):
+            result = []
+
+        result_map = {}
+        for item in result:
+            rid = item.get("id")
+            new_type = item.get("cuisine_type")
+            if rid and new_type:
+                result_map[rid] = new_type
+
+        updated = 0
+        missed = []
+        for r in batch:
+            rid = r["id"]
+            new_type = result_map.get(rid)
+            if not new_type:
+                missed.append(r)
+                continue
+            # Accept if exact match or valid prefix
+            if new_type not in valid_set and not new_type.startswith(VALID_PREFIXES):
+                missed.append(r)
+                continue
+            with db_conn() as c:
+                c.cursor().execute(
+                    "UPDATE restaurants SET cuisine_type = :ct WHERE id = :id",
+                    {"ct": new_type, "id": rid},
+                )
+                updated += 1
+
+        return updated, missed
+
+    def generate():
+        sql = """
+            SELECT r.id, r.name, r.cuisine_type,
+                   (SELECT LISTAGG(vr.foods_mentioned, '|') WITHIN GROUP (ORDER BY vr.id)
+                    FROM video_restaurants vr WHERE vr.restaurant_id = r.id) AS foods
+            FROM restaurants r
+            WHERE EXISTS (SELECT 1 FROM video_restaurants vr2 WHERE vr2.restaurant_id = r.id)
+            ORDER BY r.name
+        """
+        with db_conn() as c:
+            cur = c.cursor()
+            cur.execute(sql)
+            rows = []
+            for row in cur.fetchall():
+                foods_raw = row[3].read() if hasattr(row[3], "read") else row[3]
+                rows.append({"id": row[0], "name": row[1], "cuisine_type": row[2], "foods_mentioned": foods_raw})
+
+        total = len(rows)
+        yield f"data: {_json.dumps({'type': 'start', 'total': total})}\n\n"
+
+        valid_set = set(CUISINE_TYPES)
+        updated = 0
+        all_missed: list[dict] = []
+
+        # Pass 1: process all in batches
+        for i in range(0, total, BATCH):
+            batch = rows[i : i + BATCH]
+            yield f"data: {_json.dumps({'type': 'processing', 'current': min(i + BATCH, total), 'total': total, 'pass': 1})}\n\n"
+            try:
+                cnt, missed = _apply_batch(batch, valid_set)
+                updated += cnt
+                all_missed.extend(missed)
+                yield f"data: {_json.dumps({'type': 'batch_done', 'current': min(i + BATCH, total), 'total': total, 'updated': updated, 'missed': len(all_missed)})}\n\n"
+            except Exception as e:
+                logger.error("Remap batch error at %d: %s", i, e, exc_info=True)
+                all_missed.extend(batch)
+                yield f"data: {_json.dumps({'type': 'error', 'message': str(e), 'current': i})}\n\n"
+
+        # Pass 2: retry missed items (smaller batches for accuracy)
+        if all_missed:
+            yield f"data: {_json.dumps({'type': 'retry', 'missed': len(all_missed)})}\n\n"
+            RETRY_BATCH = 10
+            for i in range(0, len(all_missed), RETRY_BATCH):
+                batch = all_missed[i : i + RETRY_BATCH]
+                try:
+                    cnt, _ = _apply_batch(batch, valid_set)
+                    updated += cnt
+                    yield f"data: {_json.dumps({'type': 'batch_done', 'current': min(i + RETRY_BATCH, len(all_missed)), 'total': len(all_missed), 'updated': updated, 'pass': 2})}\n\n"
+                except Exception as e:
+                    logger.error("Remap retry error at %d: %s", i, e, exc_info=True)
+
+        cache.flush()
+        yield f"data: {_json.dumps({'type': 'complete', 'total': total, 'updated': updated})}\n\n"
+
+    return StreamingResponse(generate(), media_type="text/event-stream")
+
+
+@router.post("/rebuild-vectors")
+def rebuild_vectors(_admin: dict = Depends(get_admin_user)):
+    """Rebuild all restaurant vector embeddings. Streams SSE progress."""
+    from core import vector
+
+    def generate():
+        yield f"data: {_json.dumps({'type': 'start'})}\n\n"
+        try:
+            for progress in vector.rebuild_all_vectors():
+                yield f"data: {_json.dumps({'type': progress.get('status', 'progress'), **progress})}\n\n"
+            cache.flush()
+        except Exception as e:
+            logger.error("Rebuild vectors error: %s", e, exc_info=True)
+            yield f"data: {_json.dumps({'type': 'error', 'message': str(e)})}\n\n"
+
+    return StreamingResponse(generate(), media_type="text/event-stream")
+
+
 @router.get("/extract/prompt")
 def get_extract_prompt():
     """Get the current LLM extraction prompt template."""
@@ -209,11 +333,14 @@ def get_extract_prompt():
 
 
 def _do_process(limit: int):
-    return {"restaurants_extracted": process_pending(limit)}
+    result = process_pending(limit)
+    if result > 0:
+        cache.flush()
+    return {"restaurants_extracted": result}
 
 
 @router.post("/process")
-async def trigger_processing(limit: int = Query(5, le=20)):
+async def trigger_processing(limit: int = Query(5, le=20), _admin: dict = Depends(get_admin_user)):
     """Manually trigger processing of pending videos (non-blocking)."""
     loop = asyncio.get_event_loop()
     return await loop.run_in_executor(_executor, _do_process, limit)
@@ -318,11 +445,12 @@ def _do_fetch_transcript(video_db_id: str, mode: str):
             {"txt": transcript, "vid": video_db_id},
         )
 
+    cache.flush()
     return {"ok": True, "length": len(transcript), "source": source}
 
 
 @router.post("/{video_db_id}/fetch-transcript")
-async def fetch_transcript(video_db_id: str, mode: str = Query("auto")):
+async def fetch_transcript(video_db_id: str, mode: str = Query("auto"), _admin: dict = Depends(get_admin_user)):
     """Fetch and save transcript for a video (non-blocking)."""
     from fastapi import HTTPException
 
@@ -359,11 +487,12 @@ def _do_extract(video_db_id: str, custom_prompt: str | None):
         transcript,
         custom_prompt=custom_prompt,
     )
+    cache.flush()
     return {"ok": True, "restaurants_extracted": count}
 
 
 @router.post("/{video_db_id}/extract")
-async def extract_restaurants_from_video(video_db_id: str, body: dict = None):
+async def extract_restaurants_from_video(video_db_id: str, body: dict = None, _admin: dict = Depends(get_admin_user)):
     """Run LLM extraction on an existing transcript (non-blocking)."""
     from fastapi import HTTPException
     custom_prompt = body.get("prompt") if body else None
@@ -375,7 +504,7 @@ async def extract_restaurants_from_video(video_db_id: str, body: dict = None):
 
 
 @router.post("/{video_db_id}/skip")
-def skip_video(video_db_id: str):
+def skip_video(video_db_id: str, _admin: dict = Depends(get_admin_user)):
     """Mark a video as skipped."""
     from fastapi import HTTPException
     with conn() as c:
@@ -386,11 +515,12 @@ def skip_video(video_db_id: str):
         )
         if cur.rowcount == 0:
             raise HTTPException(404, "Video not found")
+    cache.flush()
     return {"ok": True}
 
 
 @router.delete("/{video_db_id}")
-def delete_video(video_db_id: str):
+def delete_video(video_db_id: str, _admin: dict = Depends(get_admin_user)):
     """Delete a video and its related data."""
     from core.db import conn as get_conn
     with get_conn() as c:
@@ -441,11 +571,12 @@ def delete_video(video_db_id: str):
         if cur.rowcount == 0:
             from fastapi import HTTPException
             raise HTTPException(404, "Video not found")
+    cache.flush()
     return {"ok": True}
 
 
 @router.put("/{video_db_id}")
-def update_video(video_db_id: str, body: dict):
+def update_video(video_db_id: str, body: dict, _admin: dict = Depends(get_admin_user)):
     """Update video title."""
     from fastapi import HTTPException
     title = body.get("title")
@@ -459,11 +590,12 @@ def update_video(video_db_id: str, body: dict):
         )
         if cur.rowcount == 0:
             raise HTTPException(404, "Video not found")
+    cache.flush()
     return {"ok": True}
 
 
 @router.delete("/{video_db_id}/restaurants/{restaurant_id}")
-def delete_video_restaurant(video_db_id: str, restaurant_id: str):
+def delete_video_restaurant(video_db_id: str, restaurant_id: str, _admin: dict = Depends(get_admin_user)):
     """Delete a video-restaurant mapping. Also cleans up orphaned restaurant."""
     from fastapi import HTTPException
     with conn() as c:
@@ -487,11 +619,12 @@ def delete_video_restaurant(video_db_id: str, restaurant_id: str):
             DELETE FROM restaurants WHERE id = :rid
             AND NOT EXISTS (SELECT 1 FROM video_restaurants WHERE restaurant_id = :rid)
         """, {"rid": restaurant_id})
+    cache.flush()
     return {"ok": True}
 
 
 @router.post("/{video_db_id}/restaurants/manual")
-def add_manual_restaurant(video_db_id: str, body: dict):
+def add_manual_restaurant(video_db_id: str, body: dict, _admin: dict = Depends(get_admin_user)):
     """Manually add a restaurant and link it to a video."""
     from fastapi import HTTPException
     from core import restaurant as rest_mod
@@ -538,11 +671,12 @@ def add_manual_restaurant(video_db_id: str, body: dict):
         guests=guests if isinstance(guests, list) else [],
     )
 
+    cache.flush()
     return {"ok": True, "restaurant_id": rid, "link_id": link_id}
 
 
 @router.put("/{video_db_id}/restaurants/{restaurant_id}")
-def update_video_restaurant(video_db_id: str, restaurant_id: str, body: dict):
+def update_video_restaurant(video_db_id: str, restaurant_id: str, body: dict, _admin: dict = Depends(get_admin_user)):
     """Update restaurant info linked to a video.
 
     If name changed, re-geocode and remap to a new restaurant record.
@@ -552,6 +686,9 @@ def update_video_restaurant(video_db_id: str, restaurant_id: str, body: dict):
 
     # Check if name changed — need to remap
     new_name = body.get("name", "").strip() if "name" in body else None
+    name_changed = False
+    active_rid = restaurant_id
+
     if new_name:
         with conn() as c:
             cur = c.cursor()
@@ -560,101 +697,126 @@ def update_video_restaurant(video_db_id: str, restaurant_id: str, body: dict):
             old_name = row[0] if row else ""
 
         if old_name != new_name:
-            # Name changed: geocode new restaurant, remap
+            name_changed = True
             from core import restaurant as rest_mod
             from core.geocoding import geocode_restaurant
 
-            address = body.get("address", "").strip() or body.get("region", "").strip() or ""
+            address = (body.get("address") or "").strip() or (body.get("region") or "").strip() or ""
             geo = geocode_restaurant(new_name, address)
             if not geo:
-                raise HTTPException(400, f"'{new_name}' 위치를 찾을 수 없습니다.")
-
-            new_rid = rest_mod.upsert(
-                name=new_name,
-                address=geo.get("formatted_address") or body.get("address"),
-                region=body.get("region"),
-                latitude=geo["latitude"],
-                longitude=geo["longitude"],
-                cuisine_type=body.get("cuisine_type"),
-                price_range=body.get("price_range"),
-                google_place_id=geo.get("google_place_id"),
-                phone=geo.get("phone"),
-                website=geo.get("website"),
-                business_status=geo.get("business_status"),
-                rating=geo.get("rating"),
-                rating_count=geo.get("rating_count"),
-            )
-
-            # Read existing mapping data, delete old, create new
-            with conn() as c:
-                cur = c.cursor()
-                cur.execute(
-                    "SELECT foods_mentioned, evaluation, guests FROM video_restaurants WHERE video_id = :vid AND restaurant_id = :rid",
-                    {"vid": video_db_id, "rid": restaurant_id},
-                )
-                old_vr = cur.fetchone()
-
-                cur.execute(
-                    "DELETE FROM video_restaurants WHERE video_id = :vid AND restaurant_id = :rid",
-                    {"vid": video_db_id, "rid": restaurant_id},
+                # Geocode failed — just rename in place without remapping
+                with conn() as c:
+                    cur = c.cursor()
+                    cur.execute("UPDATE restaurants SET name = :name, updated_at = SYSTIMESTAMP WHERE id = :rid",
+                                {"name": new_name, "rid": restaurant_id})
+            else:
+                new_rid = rest_mod.upsert(
+                    name=new_name,
+                    address=geo.get("formatted_address") or body.get("address"),
+                    region=body.get("region"),
+                    latitude=geo["latitude"],
+                    longitude=geo["longitude"],
+                    cuisine_type=body.get("cuisine_type"),
+                    price_range=body.get("price_range"),
+                    google_place_id=geo.get("google_place_id"),
+                    phone=geo.get("phone"),
+                    website=geo.get("website"),
+                    business_status=geo.get("business_status"),
+                    rating=geo.get("rating"),
+                    rating_count=geo.get("rating_count"),
                 )
 
-            # Build new mapping values from body or old data
-            def _parse(val, default):
-                if val is None:
-                    return default
-                if hasattr(val, "read"):
-                    val = val.read()
-                if isinstance(val, (list, dict)):
-                    return val
-                try:
-                    return _json.loads(val)
-                except Exception:
-                    return default
+                # Read existing mapping data, delete old, create new
+                with conn() as c:
+                    cur = c.cursor()
+                    cur.execute(
+                        "SELECT foods_mentioned, evaluation, guests FROM video_restaurants WHERE video_id = :vid AND restaurant_id = :rid",
+                        {"vid": video_db_id, "rid": restaurant_id},
+                    )
+                    old_vr = cur.fetchone()
 
-            old_foods = _parse(old_vr[0], []) if old_vr else []
-            old_eval = _parse(old_vr[1], {}) if old_vr else {}
-            old_guests = _parse(old_vr[2], []) if old_vr else []
+                    cur.execute(
+                        "DELETE FROM video_restaurants WHERE video_id = :vid AND restaurant_id = :rid",
+                        {"vid": video_db_id, "rid": restaurant_id},
+                    )
 
-            foods = body.get("foods_mentioned", old_foods)
-            evaluation = body.get("evaluation", old_eval)
-            guests = body.get("guests", old_guests)
+                def _parse(val, default):
+                    if val is None:
+                        return default
+                    if hasattr(val, "read"):
+                        val = val.read()
+                    if isinstance(val, (list, dict)):
+                        return val
+                    try:
+                        return _json.loads(val)
+                    except Exception:
+                        return default
 
-            eval_text = evaluation.get("text", "") if isinstance(evaluation, dict) else str(evaluation or "")
+                old_foods = _parse(old_vr[0], []) if old_vr else []
+                old_eval = _parse(old_vr[1], {}) if old_vr else {}
+                old_guests = _parse(old_vr[2], []) if old_vr else []
 
-            rest_mod.link_video_restaurant(
-                video_db_id=video_db_id,
-                restaurant_id=new_rid,
-                foods=foods if isinstance(foods, list) else [],
-                evaluation=eval_text or None,
-                guests=guests if isinstance(guests, list) else [],
-            )
+                foods = body.get("foods_mentioned", old_foods)
+                evaluation = body.get("evaluation", old_eval)
+                guests = body.get("guests", old_guests)
 
-            return {"ok": True, "remapped": True, "new_restaurant_id": new_rid}
+                eval_text = evaluation.get("text", "") if isinstance(evaluation, dict) else str(evaluation or "")
 
-    # No name change — update in place
-    with conn() as c:
-        cur = c.cursor()
-        r_sets = []
-        r_params: dict = {"rid": restaurant_id}
-        for field in ("name", "address", "region", "cuisine_type", "price_range"):
-            if field in body:
-                r_sets.append(f"{field} = :{field}")
-                r_params[field] = body[field]
-        if r_sets:
-            r_sets.append("updated_at = SYSTIMESTAMP")
-            sql = f"UPDATE restaurants SET {', '.join(r_sets)} WHERE id = :rid"
-            cur.execute(sql, r_params)
+                rest_mod.link_video_restaurant(
+                    video_db_id=video_db_id,
+                    restaurant_id=new_rid,
+                    foods=foods if isinstance(foods, list) else [],
+                    evaluation=eval_text or None,
+                    guests=guests if isinstance(guests, list) else [],
+                )
 
-        vr_params: dict = {"vid": video_db_id, "rid": restaurant_id}
-        vr_sets = []
-        for field in ("foods_mentioned", "evaluation", "guests"):
-            if field in body:
-                vr_sets.append(f"{field} = :{field}")
-                val = body[field]
-                vr_params[field] = _json.dumps(val, ensure_ascii=False) if isinstance(val, (list, dict)) else val
-        if vr_sets:
-            sql = f"UPDATE video_restaurants SET {', '.join(vr_sets)} WHERE video_id = :vid AND restaurant_id = :rid"
-            cur.execute(sql, vr_params)
+                active_rid = new_rid
 
-    return {"ok": True}
+                # 기존 식당이 다른 영상 매핑이 없으면 고아 → 삭제
+                if new_rid != restaurant_id:
+                    with conn() as c:
+                        cur = c.cursor()
+                        cur.execute(
+                            "SELECT COUNT(*) FROM video_restaurants WHERE restaurant_id = :rid",
+                            {"rid": restaurant_id},
+                        )
+                        remaining = cur.fetchone()[0]
+                        if remaining == 0:
+                            cur.execute("DELETE FROM restaurant_vectors WHERE restaurant_id = :rid", {"rid": restaurant_id})
+                            cur.execute("DELETE FROM user_reviews WHERE restaurant_id = :rid", {"rid": restaurant_id})
+                            cur.execute("DELETE FROM user_favorites WHERE restaurant_id = :rid", {"rid": restaurant_id})
+                            cur.execute("DELETE FROM restaurants WHERE id = :rid", {"rid": restaurant_id})
+
+    # Update remaining fields in place (skip name if already remapped)
+    if not name_changed:
+        with conn() as c:
+            cur = c.cursor()
+            r_sets = []
+            r_params: dict = {"rid": active_rid}
+            for field in ("address", "region", "cuisine_type", "price_range"):
+                if field in body:
+                    r_sets.append(f"{field} = :{field}")
+                    r_params[field] = body[field]
+            if r_sets:
+                r_sets.append("updated_at = SYSTIMESTAMP")
+                sql = f"UPDATE restaurants SET {', '.join(r_sets)} WHERE id = :rid"
+                cur.execute(sql, r_params)
+
+            vr_params: dict = {"vid": video_db_id, "rid": active_rid}
+            vr_sets = []
+            for field in ("foods_mentioned", "evaluation", "guests"):
+                if field in body:
+                    vr_sets.append(f"{field} = :{field}")
+                    val = body[field]
+                    vr_params[field] = _json.dumps(val, ensure_ascii=False) if isinstance(val, (list, dict)) else val
+            if vr_sets:
+                sql = f"UPDATE video_restaurants SET {', '.join(vr_sets)} WHERE video_id = :vid AND restaurant_id = :rid"
+                cur.execute(sql, vr_params)
+
+    cache.flush()
+    result: dict = {"ok": True}
+    if name_changed:
+        result["remapped"] = active_rid != restaurant_id
+        if active_rid != restaurant_id:
+            result["new_restaurant_id"] = active_rid
+    return result