landmark_data_manager.py

import logging
import traceback
from typing import List, Dict, Tuple, Optional, Union, Any

from landmark_data import ALL_LANDMARKS, get_all_landmark_prompts
from landmark_activities import LANDMARK_ACTIVITIES

class LandmarkDataManager:
    """
    專門處理地標數據的載入、管理和查詢功能，包括地標信息、提示詞和活動建議
    """

    def __init__(self):
        """
        initialize landmark related
        """
        self.logger = logging.getLogger(__name__)
        self.landmark_data = {}
        self.landmark_prompts = []
        self.landmark_id_to_index = {}
        self.is_enabled = False

        self._load_landmark_data()

    def _load_landmark_data(self):
        """
        載入地標數據和相關資訊
        """
        try:
            self.landmark_data = ALL_LANDMARKS
            self.landmark_prompts = get_all_landmark_prompts()
            self.logger.info(f"Loaded {len(self.landmark_prompts)} landmark prompts for classification")

            # 創建地標ID到索引的映射，可快速查找
            self.landmark_id_to_index = {landmark_id: i for i, landmark_id in enumerate(ALL_LANDMARKS.keys())}

            self.is_enabled = True
            self.logger.info(f"Successfully loaded landmark data with {len(self.landmark_data)} landmarks")

        except ImportError:
            self.logger.warning("landmark_data.py not found. Landmark classification will be limited")
            self.landmark_data = {}
            self.landmark_prompts = []
            self.landmark_id_to_index = {}
            self.is_enabled = False
        except Exception as e:
            self.logger.error(f"Error loading landmark data: {e}")
            self.logger.error(traceback.format_exc())
            self.landmark_data = {}
            self.landmark_prompts = []
            self.landmark_id_to_index = {}
            self.is_enabled = False

    def get_landmark_prompts(self) -> List[str]:
        """
        獲取所有地標提示詞

        Returns:
            List[str]: 地標提示詞列表
        """
        return self.landmark_prompts

    def get_landmark_by_id(self, landmark_id: str) -> Dict[str, Any]:
        """
        根據地標ID獲取地標信息

        Args:
            landmark_id: Landmark ID

        Returns:
            Dict[str, Any]: 地標詳細信息
        """
        return self.landmark_data.get(landmark_id, {})

    def get_landmark_by_index(self, index: int) -> Tuple[str, Dict[str, Any]]:
        """
        根據索引獲取地標信息

        Args:
            index: 地標在列表中的索引

        Returns:
            Tuple[str, Dict[str, Any]]: (地標ID, 地標info)
        """
        try:
            landmark_ids = list(self.landmark_data.keys())
            if 0 <= index < len(landmark_ids):
                landmark_id = landmark_ids[index]
                return landmark_id, self.landmark_data[landmark_id]
            else:
                self.logger.warning(f"Index {index} out of range for landmark data")
                return None, {}
        except Exception as e:
            self.logger.error(f"Error getting landmark by index {index}: {e}")
            self.logger.error(traceback.format_exc())
            return None, {}

    def get_landmark_index(self, landmark_id: str) -> Optional[int]:
        """
        獲取地標ID對應的index

        Args:
            landmark_id: 地標ID

        Returns:
            Optional[int]: 索引，如果不存在則返回None
        """
        return self.landmark_id_to_index.get(landmark_id)

    def determine_landmark_type(self, landmark_id: str) -> str:
        """
        自動判斷地標類型，基於地標數據和命名

        Args:
            landmark_id: 地標ID

        Returns:
            str: 地標類型，用於調整閾值
        """
        if not landmark_id:
            return "building"  # 預設類型

        try:
            # 獲取地標詳細數據
            landmark_info = self.landmark_data.get(landmark_id, {})

            # 獲取地標相關文本
            landmark_id_lower = landmark_id.lower()
            landmark_name = landmark_info.get("name", "").lower()
            landmark_location = landmark_info.get("location", "").lower()
            landmark_aliases = [alias.lower() for alias in landmark_info.get("aliases", [])]

            # 合併所有文本數據用於特徵判斷
            combined_text = " ".join([landmark_id_lower, landmark_name] + landmark_aliases)

            # 地標類型的特色特徵
            type_features = {
                "skyscraper": ["skyscraper", "tall", "tower", "高樓", "摩天", "大厦", "タワー"],
                "tower": ["tower", "bell", "clock", "塔", "鐘樓", "タワー", "campanile"],
                "monument": ["monument", "memorial", "statue", "紀念", "雕像", "像", "memorial"],
                "natural": ["mountain", "lake", "canyon", "falls", "beach", "山", "湖", "峽谷", "瀑布", "海灘"],
                "temple": ["temple", "shrine", "寺", "神社", "廟"],
                "palace": ["palace", "castle", "宮", "城", "皇宮", "宫殿"],
                "distinctive": ["unique", "leaning", "slanted", "傾斜", "斜", "獨特", "傾く"]
            }

            # 檢查是否位於亞洲地區
            asian_regions = ["china", "japan", "korea", "taiwan", "singapore", "vietnam", "thailand",
                            "hong kong", "中國", "日本", "韓國", "台灣", "新加坡", "越南", "泰國", "香港"]
            is_asian = any(region in landmark_location for region in asian_regions)

            # 判斷地標類型
            best_type = None
            max_matches = 0

            for type_name, features in type_features.items():
                # 計算特徵詞匹配數量
                matches = sum(1 for feature in features if feature in combined_text)
                if matches > max_matches:
                    max_matches = matches
                    best_type = type_name

            # 處理亞洲地區特例
            if is_asian and best_type == "tower":
                best_type = "skyscraper"  # 亞洲地區的塔型建築閾值較低

            # 特例處理：檢測傾斜建築
            if any(term in combined_text for term in ["leaning", "slanted", "tilt", "inclined", "斜", "傾斜"]):
                return "distinctive"  # 傾斜建築需要特殊處理

            return best_type if best_type and max_matches > 0 else "building"  # 預設為一般建築

        except Exception as e:
            self.logger.error(f"Error determining landmark type for {landmark_id}: {e}")
            self.logger.error(traceback.format_exc())
            return "building"

    def extract_landmark_specific_info(self, landmark_id: str) -> Dict[str, Any]:
        """
        提取特定地標的詳細信息，包括特色模板和活動建議

        Args:
            landmark_id: 地標ID

        Returns:
            Dict[str, Any]: 地標特定信息
        """
        if not landmark_id or landmark_id == "unknown":
            return {"has_specific_activities": False}

        specific_info = {"has_specific_activities": False}

        try:
            # 從 landmark_data 中提取基本信息
            landmark_data_source = self.landmark_data.get(landmark_id)

            # 處理地標基本數據
            if landmark_data_source:
                # 提取正確的地標名稱
                if "name" in landmark_data_source:
                    specific_info["landmark_name"] = landmark_data_source["name"]

                # 提取所有可用的 prompts 作為特色模板
                if "prompts" in landmark_data_source:
                    specific_info["feature_templates"] = landmark_data_source["prompts"][:5]
                    specific_info["primary_template"] = landmark_data_source["prompts"][0]

                # 提取別名info
                if "aliases" in landmark_data_source:
                    specific_info["aliases"] = landmark_data_source["aliases"]

                # 提取位置信息
                if "location" in landmark_data_source:
                    specific_info["location"] = landmark_data_source["location"]

                # 提取其他相關信息
                for key in ["year_built", "architectural_style", "significance", "description"]:
                    if key in landmark_data_source:
                        specific_info[key] = landmark_data_source[key]

            # 嘗試從 LANDMARK_ACTIVITIES 中提取活動建議
            try:
                if landmark_id in LANDMARK_ACTIVITIES:
                    activities = LANDMARK_ACTIVITIES[landmark_id]
                    specific_info["landmark_specific_activities"] = activities
                    specific_info["has_specific_activities"] = True
                    self.logger.info(f"Found {len(activities)} specific activities for landmark {landmark_id}")
                else:
                    self.logger.info(f"No specific activities found for landmark {landmark_id} in LANDMARK_ACTIVITIES")
                    specific_info["has_specific_activities"] = False
            except ImportError:
                self.logger.warning("Could not import LANDMARK_ACTIVITIES from landmark_activities")
                specific_info["has_specific_activities"] = False
            except Exception as e:
                self.logger.error(f"Error loading landmark activities for {landmark_id}: {e}")
                self.logger.error(traceback.format_exc())
                specific_info["has_specific_activities"] = False

        except Exception as e:
            self.logger.error(f"Error extracting landmark specific info for {landmark_id}: {e}")
            self.logger.error(traceback.format_exc())

        return specific_info

    def get_landmark_count(self) -> int:
        """
        獲取地標總數

        Returns:
            int: 地標數量
        """
        return len(self.landmark_data)

    def is_landmark_enabled(self) -> bool:
        """
        檢查地標功能是否啟用

        Returns:
            bool: 地標功能狀態
        """
        return self.is_enabled

    def get_all_landmark_ids(self) -> List[str]:
        """
        獲取所有地標ID列表

        Returns:
            List[str]: 地標ID列表
        """
        return list(self.landmark_data.keys())

    def validate_landmark_id(self, landmark_id: str) -> bool:
        """
        驗證地標ID是否有效

        Args:
            landmark_id: 要驗證的地標ID

        Returns:
            bool: ID是否有效
        """
        return landmark_id in self.landmark_data