iT邦幫忙

0

# Day 16 串接 LINE Location 訊息與經緯度周邊精準推薦

n8n
  • 分享至 

  • xImage
  •  

在昨天的文章中,我們完成了全雲原生架構的規劃與部署,讓整套系統具備了 24 小時在線的服務能力。

然而,目前使用者若想查詢附近的餐廳,依然必須手動輸入文字(例如「我在板橋車站附近」或「台中勤美周邊」)。在真實的日常場景中,人們往往站在十字路口或陌生巷弄,根本不清楚自己所在的確切地名或路名。這時候,如果能直接在 LINE 聊天室按一下「分享位置」,讓 Agent 自動讀取 GPS 座標並搜尋周邊店家,體驗將會更加直覺流暢!

今天 Day 16,我們將正式跨入進階功能模組:在 n8n 中解析 LINE Location Message(位置訊息),並在 FastAPI 中支援利用真實經緯度(Latitude & Longitude)進行半徑圓形搜尋!


一、LINE Location 訊息封包結構解析

當使用者點擊 LINE 輸入框旁的「+」號並選擇「位置資訊」發送時,LINE Webhook 傳送至 n8n 的封包類型不再是 text,而是 location:

{
  "type": "message",
  "message": {
    "type": "location",
    "id": "325708",
    "title": "我的目前位置",
    "address": "新北市板橋區文化路一段 188 巷",
    "latitude": 25.0195,
    "longitude": 121.4642
  }
}

關鍵欄位:
message.latitude:GPS 緯度(浮點數,例如 25.0195)。

message.longitude:GPS 經度(浮點數,例如 121.4642)。

message.address:LINE 自動逆向地理編碼出的地址字串。

二、升級 FastAPI 資料模型與搜尋邏輯

為了支援經緯度搜尋,我們先在後端新增座標專屬的查詢介面。

Step 1:擴充 Pydantic 模型 (app/schemas/food.py)

在 UserPreference 模型中加入選填的經緯度與搜尋半徑欄位:

from pydantic import BaseModel, Field
from typing import Optional

class UserPreference(BaseModel):
    user_id: str = Field(..., description="使用者唯一識別碼")
    cuisine: Optional[str] = Field(default="全部", description="料理類型")
    budget_max: int = Field(default=500, description="預算上限")
    location: Optional[str] = Field(default="板橋區", description="文字地名或商圈")
    
    # 新增 GPS 支援
    latitude: Optional[float] = Field(None, description="GPS 緯度")
    longitude: Optional[float] = Field(None, description="GPS 經度")
    radius_meters: int = Field(default=1000, description="搜尋半徑(公尺,預設 1 公里)")
    notes: Optional[str] = Field(None, description="特殊飲食備註")

Step 2:實作 Places Nearby Search 支援 (app/services/places.py)

Google Places API (New) 除了文字搜尋外,還支援以特定座標為中心的圓形範圍搜尋(locationRestriction):

import httpx
from typing import List, Optional
from app.config import GOOGLE_MAPS_API_KEY
from app.schemas.food import RestaurantItem, CuisineType

PLACES_TEXT_SEARCH_URL = "[https://places.googleapis.com/v1/places:searchText](https://places.googleapis.com/v1/places:searchText)"

PRICE_LEVEL_MAP = {
    "PRICE_LEVEL_FREE": 0,
    "PRICE_LEVEL_INEXPENSIVE": 150,
    "PRICE_LEVEL_MODERATE": 350,
    "PRICE_LEVEL_EXPENSIVE": 800,
    "PRICE_LEVEL_VERY_EXPENSIVE": 1500,
}

async def search_places(
    query_location: str,
    query_cuisine: str,
    budget_max: int,
    latitude: Optional[float] = None,
    longitude: Optional[float] = None,
    radius: int = 1000
) -> List[RestaurantItem]:
    """
    支援文字地名或經緯度半徑圓形搜尋
    """
    if not GOOGLE_MAPS_API_KEY:
        return []

    headers = {
        "Content-Type": "application/json",
        "X-Goog-Api-Key": GOOGLE_MAPS_API_KEY,
        "X-Goog-FieldMask": "places.id,places.displayName,places.formattedAddress,places.rating,places.priceLevel,places.editorialSummary"
    }

    payload = {
        "textQuery": f"{query_cuisine} 餐廳" if query_cuisine != "全部" else "熱門美食餐廳",
        "languageCode": "zh-TW",
        "maxResultCount": 5
    }

    # 若有傳入 GPS 座標,加入圓形限制以鎖定周邊店家
    if latitude is not None and longitude is not None:
        payload["locationBias"] = {
            "circle": {
                "center": {
                    "latitude": latitude,
                    "longitude": longitude
                },
                "radius": float(radius)
            }
        }
    else:
        # 降級使用純文字地點搜尋
        payload["textQuery"] = f"{query_location} {query_cuisine}" if query_cuisine != "全部" else f"{query_location} 美食"

    results: List[RestaurantItem] = []

    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.post(PLACES_TEXT_SEARCH_URL, headers=headers, json=payload)
        if response.status_code != 200:
            return []

        places = response.json().get("places", [])
        for p in places:
            price_level_str = p.get("priceLevel", "PRICE_LEVEL_MODERATE")
            estimated_price = PRICE_LEVEL_MAP.get(price_level_str, 300)

            if estimated_price > budget_max:
                continue

            summary_dict = p.get("editorialSummary", {})
            reason_text = summary_dict.get("text", "定位周邊高評價精選餐廳!")

            item = RestaurantItem(
                id=p.get("id", "places_id"),
                name=p.get("displayName", {}).get("text", "周邊推薦餐廳"),
                cuisine=CuisineType.ALL,
                price_avg=estimated_price,
                location=p.get("formattedAddress", "距離您不遠處"),
                rating=float(p.get("rating", 4.0)),
                signature_dish=f"在地人氣餐點",
                reason=reason_text
            )
            results.append(item)

    return results

三、在 n8n 中分流文字與位置事件

在 n8n 的訊息抽取節點中,我們必須判斷進來的事件是「文字」還是「位置」,並進行分流處理。

修改 Extract LINE Event 節點 (Code 節點)

const event = $input.first().json.body.events[0];

if (!event || event.type !== 'message') {
  return [];
}

const baseData = {
  reply_token: event.replyToken,
  user_id: event.source.userId,
  msg_type: event.message.type
};

// 處理文字訊息
if (event.message.type === 'text') {
  return {
    ...baseData,
    user_message: event.message.text,
    latitude: null,
    longitude: null
  };
}

// 處理位置訊息
if (event.message.type === 'location') {
  return {
    ...baseData,
    user_message: `我在 ${event.message.address || '目前位置'},推薦附近好吃的`,
    latitude: event.message.latitude,
    longitude: event.message.longitude
  };
}

return [];

四、條件分支(Switch)與組裝呼叫

在工作流畫布中新增一個 Switch 節點:

條件 1 (is_location):當 {{ $json.msg_type }} 等於 location。

直接跳過 Dify:因為使用者已經提供了確切經緯度,無需多消耗一次 LLM Token 去抽取地點,直接將 latitude 與 longitude 傳給 FastAPI!

條件 2 (is_text):當 {{ $json.msg_type }} 等於 text。

走原先流程:進入 Dify 抽取意圖 ➜ 調用 FastAPI。

這樣做不僅能大幅提升定位查詢的反應速度,還能幫我們省下寶貴的 LLM 呼叫次數!

五、實機操作驗證

1.打開手機 LINE 聊天室。

2.點擊左下角「+」號 ➜ 點選 「位置資訊」。

3.拖動地圖或直接點擊右上角的 「分享這處地點」。

4.幾秒之內,聊天室便會跳出以你當前座標為中心、半徑 1 公里內的熱門餐廳 Flex Message 輪播卡片!


*提醒邦友,使用第三方服務/API 時,請務必評估資安風險與隱私保護
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言