import json import time import random import pandas as pd import numpy as np import re #提取 POI 經緯度 extract_poi_coordinates = lambda poi_data: np.array([ [poi["location"]["latitude"], poi["location"]["longitude"]] for poi in poi_data if poi.get("location") and "latitude" in poi["location"] and "longitude" in poi["location"] ]) def filter_pois(poi_with_distances, parsed_targets): """執行一系列過濾操作""" #以目標過濾 filtered_target_pois = filter_by_target(poi_with_distances, parsed_targets) # 以距離 (distance) 過濾 filtered_distance_pois = filter_by_distance(filtered_target_pois, parsed_targets) # 以營業時間 (open) 過濾 filtered_open_pois = filter_by_open(filtered_distance_pois, parsed_targets) # 以評分 (rating) 過濾或排序 filtered_rating_pois = filter_or_sort_by_rating(filtered_open_pois, parsed_targets) # 過濾 POI 列表中的營業時間，僅保留今天的時間段。 top_filtered_pois = filter_today_opening_hours(filtered_rating_pois) return top_filtered_pois #過濾掉位於 user_location 西邊（longitude < user_lon）的 POIs，依據每個POI 的distance 排序後取前3。 def filter_and_get_top_pois(user_location, top_filtered_pois): _, user_lon = user_location # 過濾出在東邊的 POIs (longitude > user_lon) east_pois = [ poi for poi in top_filtered_pois if poi['location']['longitude'] > user_lon ] # 依據 POI 本身的 distance 屬性進行排序 east_pois.sort(key=lambda x: x['distance']) # 取前 3 個 return east_pois[:3] def filter_poi_list(data_list): needed_keys = [ "englishName", "types", "primaryType", "rating", "location", "formattedAddress", "userRatingCount", "editorialSummary" ] filtered = [] for poi in data_list: filtered_dict = {key: poi.get(key) for key in needed_keys} regular_hours = poi.get("regularOpeningHours") if regular_hours and "periods" in regular_hours: filtered_dict["regularOpeningHours"] = { "periods": regular_hours["periods"] } else: # 如果沒有 regularOpeningHours 或沒有 periods，就給個 None 或空字典 filtered_dict["regularOpeningHours"] = None filtered.append(filtered_dict) return filtered def get_coordinates(location_data, video_tps_value): """ 從 DataFrame 中檢索特定 video_tps 的經緯度座標。 """ matched_row = location_data[location_data['video_tps'] == video_tps_value] if matched_row.empty: # 如果沒有匹配的行，返回空列表 return [] # 提取經緯度 latitude = matched_row['latitude'].values[0] longitude = matched_row['longitude'].values[0] return (latitude, longitude) # 計算兩個地點之間的距離（Haversine formula）返回公尺 def calculate_distances(point, points_array): lat1, lon1 = point lat2 = points_array[:, 0] lon2 = points_array[:, 1] # 將經緯度轉換為弧度 lon1, lat1, lon2, lat2 = map(np.radians, [lon1, lat1, lon2, lat2]) # 計算經緯度的差值 dlon = lon2 - lon1 dlat = lat2 - lat1 # Haversine formula 計算大圓距離 a = np.sin(dlat / 2.0) ** 2 + np.cos(lat1) * np.cos(lat2) * np.sin(dlon / 2.0) ** 2 c = 2 * np.arcsin(np.sqrt(a)) r = 6371000 # 地球半徑，單位：公尺 distances_in_meters = c * r # 以公尺計算距離 # 四捨五入到小數點後兩位 return np.round(distances_in_meters, 2) def add_distances_to_poi(poi_data, distances): # 新增 distance 欄位 poi_with_distances = [] for i, poi in enumerate(poi_data): # 複製 POI 資料，新增 distance 欄位 poi_with_distance = poi.copy() poi_with_distance['distance'] = distances[i] poi_with_distances.append(poi_with_distance) return poi_with_distances def parse_ner_targets(parsed_entities): """ 將 parsed_entities 中的 target 拆解為 primarytype 和 displayname， """ # 定義主要類別 primary_categories = {"restaurant", "shopping_mall", "hospital", "parking", "tourist_attraction", "convenience_store", "bank", "car_repair", "electric_vehicle_charging_station", "gas_station", "movie_theater"} parsed_result = { "primarytype": '', "displayname": '', "distance": parsed_entities.get("distance", []), "open": parsed_entities.get("open", []), "rank": parsed_entities.get("rank", []) } targets = parsed_entities.get("target", []) for target in targets: if target in primary_categories: parsed_result["primarytype"] = target else: parsed_result["displayname"] = target return parsed_result def filter_by_target(poi_with_distances, parsed_targets): """ 用 parsed_targets 的內容篩選 poi_with_distances。 1. 如果 parsed_targets['displayname'] 不為空字串 (''), 以正則匹配完整詞，不被其他字母數字相鄰 2. 若 parsed_targets['displayname'] 為空字串 (''), 則以 parsed_targets['primarytype'] 檢查 poi_with_distances['types'] 中是否包含 """ target_displayname = parsed_targets.get("displayname", "").strip() target_type = parsed_targets.get("primarytype", "").strip() if target_displayname: print('target_displayname') pattern = re.compile(rf'\b{re.escape(target_displayname.lower())}\b') filtered_pois = [ poi for poi in poi_with_distances if pattern.search(poi['englishName'].lower()) ] else: # displayname為空，就看types裡面是否含有target_type filtered_pois = [ poi for poi in poi_with_distances if target_type in poi.get("types", []) ] return filtered_pois # 以距離 (distance) 過濾 def filter_by_distance(poi_list, parsed_targets): """ 1) 先按照 distance 升冪(由近到遠)進行排序 2) 若 parsed_targets['distance'] = ['500']，則只保留 distance < 500 的項目 若無 distance 或 distance 為空，則返回原清單（但也還是已經排序好） """ # 先依距離由小到大排序 sorted_by_distance = sorted( poi_list, key=lambda x: x.get("distance", float("inf")) # 如果沒有 distance，當作無限大 ) # 若沒有傳入 distance 閾值，直接回傳「已排序」的結果 if not parsed_targets.get("distance"): return sorted_by_distance # 假設只取第一個 distance 閾值 distance_str = parsed_targets["distance"][0] distance_threshold = float(distance_str) # 再針對距離做篩選 filtered_result = [ poi for poi in sorted_by_distance if poi.get("distance") is not None and poi["distance"] < distance_threshold ] return filtered_result # 以評分 (rating) 過濾或排序 def filter_or_sort_by_rating(poi_list, parsed_targets): """ 若 parsed_targets['rank'] 為小數點字串 (例如 "4.0") => 過濾 rating >= 該值 若 parsed_targets['rank'] 為 "highest" => 依 rating 高到低排序 其他情況 (例如沒有 rank 或 rank 為空) 則不處理，直接返回原清單 """ # 如果沒有 rank 或 rank 為空 list，就直接回傳原清單 if not parsed_targets.get("rank"): return poi_list rank_value = parsed_targets["rank"][0] # 只取第一個 # Case 1：若為 "highest" => 依 rating 高到低排序 if rank_value.lower() == "highest": return sorted(poi_list, key=lambda x: x["rating"] if x["rating"] is not None else 0, reverse=True) # Case 2：否則視為數值字符串 (如 "4.5") rank_float = float(rank_value) # 可直接轉 float，因為不會有其他無法轉換的情況 return [ poi for poi in poi_list if poi.get("rating") and poi["rating"] >= rank_float ] # 以營業時間 (open) 過濾 def filter_by_open(poi_with_distances, parsed_targets): """ 不動原始 poi_with_distances，只回傳一個「新清單」。 在新清單的每個 POI dict 中，新增/更新 key: "isOpenNow" (True/False)， 表示「在當下星期 + parsed_targets['open'] 時間下」是否營業。 - 若 parsed_targets['open'] 是空，則直接將 isOpenNow 設為 None （不判斷）。 - 其餘欄位皆照原資料保留。 """ # 如果沒有 open，則不判斷，整份清單都 isOpenNow = None if not parsed_targets.get("open"): new_list = [] for poi in poi_with_distances: poi_copy = poi.copy() # 淺拷貝，避免改到原資料 poi_copy["isOpenNow"] = None # 不判斷，固定 None new_list.append(poi_copy) return new_list # 取得使用者指定的小時，例如 15 / 21 target_hour = float(parsed_targets["open"][0]) # 假設只取第一個 # 系統當下是星期幾 (0=星期一, 6=星期日) current_weekday_py = time.localtime().tm_wday # 轉成 POI 的 day： (0=星期日, 1=星期一, ... 6=星期六) current_weekday_poi = (current_weekday_py + 1) % 7 new_list = [] for poi in poi_with_distances: poi_copy = poi.copy() # 建立副本，不動原資料 # 預設 isOpenNow = False poi_copy["isOpenNow"] = False regular_hours = poi_copy.get("regularOpeningHours") if regular_hours and "periods" in regular_hours: periods = regular_hours["periods"] # 檢查當天 (current_weekday_poi) 時段 for p in periods: if p["open"]["day"] == current_weekday_poi: open_hour = p["open"]["hour"] close_hour = p["close"]["hour"] # 分鐘先忽略，只看整點 if open_hour <= target_hour < close_hour: poi_copy["isOpenNow"] = True break new_list.append(poi_copy) return new_list def filter_today_opening_hours(poi_list): """ 過濾 POI 列表中的營業時間，僅保留今天的時間段。 """ # 取得今天的星期 (0 表示星期一, 6 表示星期日) today = time.localtime().tm_wday # 遍歷 POI 資料 for poi in poi_list: # 如果 regularOpeningHours 為 None，跳過處理 if poi.get('regularOpeningHours') is None or 'periods' not in poi['regularOpeningHours']: continue # 過濾只保留今天的營業時間 poi['regularOpeningHours']['periods'] = [ period for period in poi['regularOpeningHours']['periods'] if period['open']['day'] == today ] return poi_list def filter_pois_by_rating_count(pois, user_rating_threshold=1000): filtered_pois = [ poi for poi in pois if (poi.get('userRatingCount') or 0) >= user_rating_threshold ] return filtered_pois # 帶入pois 過濾poi type def filter_pois_by_type(pois, poi_type): """ Filters a list of POIs based on a specified type. """ return [poi for poi in pois if poi_type in poi.get("types", [])] def format_english_name(english_name): """ 接收 englishname 字串，並根據分隔符號處理字串。 """ # 使用正則表達式拆分字符串，根據 '|', '/', '\' 作為分隔符 parts = re.split(r'[|｜/／\\＼]', english_name) # 確保至少有兩個部分可以處理 if len(parts) > 1: main_part = parts[0].strip() # 取第一部分並去除前後空格 first_segment = parts[1].strip() # 取第二部分並去除前後空格 return f"{main_part} ({first_segment})" return english_name # 若無有效分隔符，則返回原始字符串 def truncate_and_fix_brackets(input_str, max_length=70): """ 簡化並修正括號的字串處理 1. 若字串超過指定長度（預設 70 字），則確保不在單字中間截斷，並適當延伸到完整單字結束後。 2. 截斷後，若發現未完整的括號（如 `(` `（` `{` `[`），則自動補齊對應的右括號。 """ # 若字串長度在允許範圍內，直接回傳 if len(input_str) <= max_length: return input_str # **步驟 1：確保不截斷單字** truncated_str = input_str[:max_length] # 初步截斷 remaining_str = input_str[max_length:] # 搜尋下一個可用的空格或標點，避免單字被切開 match = re.search(r'[\s,!)\]\}，）]', remaining_str) # 尋找最近的空格或標點 if match: truncated_str += remaining_str[:match.start()] # 延伸至完整的單字結束後 else: truncated_str = input_str # 若沒有適合的分隔符，則不截斷 # **步驟 2：修正未關閉的括號** bracket_pairs = {'(': ')', '（': '）', '[': ']', '{': '}'} open_brackets = [] # 用來追蹤未關閉的括號 for char in truncated_str: if char in bracket_pairs: # 若遇到左括號，記錄對應的右括號 open_brackets.append(bracket_pairs[char]) elif char in bracket_pairs.values(): # 若遇到右括號，確認是否匹配 if open_brackets and open_brackets[-1] == char: open_brackets.pop() # 依照順序補齊缺少的右括號 for missing in reversed(open_brackets): truncated_str += missing return truncated_str def generate_tts_message(top_filtered_pois): """ 生成禮貌 自然語氣的TTS。輸出找到的店名，並隨機選擇一種禮貌模板。 """ num_places = len(top_filtered_pois) if num_places == 0: return "I'm sorry, I couldn't find any matching places nearby. Please try adjusting your preferences." # 五種模板 templates = [ "I found {num_places} places that might interest you nearby: {details} Have a wonderful day ahead!", "Here are {num_places} nearby places you might like: {details} Hope this helps you!", "I've identified {num_places} places nearby: {details} Wishing you a pleasant journey!", "There are {num_places} places nearby that you might enjoy: {details} Have a great time!", "Nearby, I found {num_places} places worth checking out: {details} Enjoy exploring!" ] # 選擇隨機模板 chosen_template = random.choice(templates) # 構建內容，直接列出店名 details = ", ".join( [f"***{format_english_name(truncate_and_fix_brackets(poi.get('englishName', 'Unknown Place')))}***" for poi in top_filtered_pois] ) + "." # 使用模板生成 final_message = chosen_template.format(num_places=num_places, details=details) return final_message def generate_display_message(top_filtered_pois): """ 生成自然語言的顯示消息，包含店名、評分、距離（整數公尺）、與英文簡介。 使用多個模板，隨機選擇一種生成消息。 """ num_places = len(top_filtered_pois) if num_places == 0: return "I'm sorry, I couldn't find any places that match your request in the nearby area. Please try expanding the search radius or adjusting your preferences." # 定義五種模板 templates = [ "I found {num_places} places nearby. \n{details} Enjoy exploring!", "Here are {num_places} nearby options: \n{details} Hope you find this helpful!", "There are {num_places} places worth checking out. \n{details} Have a great time!", "Nearby, I identified {num_places} options you might like. \n{details} Wishing you a pleasant experience!", "I discovered {num_places} places close by. \n{details} Enjoy your visit!" ] # 隨機選擇模板 chosen_template = random.choice(templates) # 構建詳細內容 details = "\n".join([ f"{idx}. {format_english_name(truncate_and_fix_brackets(poi.get('englishName', 'Unknown Place')))}:\n" f" - Rating: {poi.get('rating', 'No rating')}/5\n" f" - Distance: {int(poi.get('distance', 0))} meters\n" f" - Summary: {poi.get('editorialSummary', {}).get('englishSummary', 'No summary available.') if isinstance(poi.get('editorialSummary'), dict) else 'No summary available.'}" for idx, poi in enumerate(top_filtered_pois, start=1) ]) # 使用模板生成最終消息 final_message = chosen_template.format(num_places=num_places, details=details) return final_message