core/database/base.py

import os
import sys
import logging
from datetime import datetime
from typing import Any, Dict, List, Optional
from dotenv import load_dotenv
import firebase_admin
from firebase_admin import credentials, firestore
from google.cloud.firestore import DocumentReference, CollectionReference, FieldFilter
from google.cloud.firestore_v1 import ArrayUnion, Increment
import asyncio
import hashlib

# 設置日誌
LOG_LEVEL_NAME = os.getenv("BLOOMWARE_LOG_LEVEL", "WARNING").upper()
LOG_LEVEL = getattr(logging, LOG_LEVEL_NAME, logging.WARNING)
logging.basicConfig(level=LOG_LEVEL, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger("Firestore")
logger.setLevel(LOG_LEVEL)

# 載入環境變數
load_dotenv()

# 統一配置管理
from core.config import settings

# 全局變數
firestore_db = None
users_collection = None
chats_collection = None
messages_collection = None
memories_collection = None
health_data_collection = None
device_bindings_collection = None
geo_cache_collection = None
route_cache_collection = None

# 記憶儲存相關設定
MAX_MEMORIES_PER_USER = 500


def _serialize_firestore_data(data: Any) -> Any:
    """
    遞迴轉換 Firestore 資料中的 DatetimeWithNanoseconds 物件為 ISO 字串

    Args:
        data: Firestore 回傳的資料（可能包含 DatetimeWithNanoseconds）

    Returns:
        JSON 可序列化的資料
    """
    from google.cloud.firestore_v1._helpers import DatetimeWithNanoseconds

    if isinstance(data, DatetimeWithNanoseconds):
        # 轉成 ISO 8601 字串
        return data.isoformat()
    elif isinstance(data, datetime):
        # 一般 Python datetime 也轉成字串
        return data.isoformat()
    elif isinstance(data, dict):
        # 遞迴處理字典
        return {k: _serialize_firestore_data(v) for k, v in data.items()}
    elif isinstance(data, list):
        # 遞迴處理列表
        return [_serialize_firestore_data(item) for item in data]
    else:
        # 其他型別直接回傳
        return data


def _get_user_doc_ref(user_id: str) -> DocumentReference:
    if users_collection is None:
        raise RuntimeError("Firestore尚未連接，無法操作使用者資料")
    return users_collection.document(user_id)


def _get_user_memories_collection(user_id: str) -> CollectionReference:
    return _get_user_doc_ref(user_id).collection("memories")


def _get_chat_messages_collection(chat_id: str) -> CollectionReference:
    if chats_collection is None:
        raise RuntimeError("Firestore尚未連接，無法取得對話消息集合")
    return chats_collection.document(chat_id).collection("messages")

def connect_to_firestore():
    """初始化 Firebase Firestore 連接"""
    global firestore_db, messages_collection, users_collection, chats_collection, memories_collection, health_data_collection, device_bindings_collection

    firebase_project_id = settings.FIREBASE_PROJECT_ID

    if not firebase_project_id:
        logger.error("Firebase專案ID未正確設置，請在.env文件中設置FIREBASE_PROJECT_ID環境變數")
        logger.error("\n❌ 錯誤: Firebase專案ID未設置！請在.env文件中設置FIREBASE_PROJECT_ID\n")
        return False

    try:
        logger.info("正在嘗試連接Firebase Firestore...")
        logger.info("\n🔄 正在連接Firebase Firestore數據庫...\n")

        # 檢查是否已經初始化 Firebase
        try:
            firebase_admin.get_app()
            logger.info("Firebase 已初始化，跳過重複初始化")
        except ValueError:
            # 從統一配置取得 Firebase 憑證（支援環境變數或檔案）
            try:
                firebase_creds_dict = settings.get_firebase_credentials()
                cred = credentials.Certificate(firebase_creds_dict)
                firebase_admin.initialize_app(cred, {
                    'projectId': firebase_project_id,
                })
                logger.info(f"Firebase 初始化成功（專案ID：{firebase_project_id}）")
            except ValueError as e:
                logger.error(f"Firebase 憑證載入失敗: {e}")
                logger.error(f"\n❌ 錯誤: Firebase 憑證載入失敗！{e}\n")
                return False
        
        # 初始化 Firestore 客戶端
        firestore_db = firestore.client()
        
        # 測試連接
        test_doc = firestore_db.collection('_test_connection').document('test')
        test_doc.set({'timestamp': datetime.now(), 'test': True})
        test_doc.delete()  # 清理測試文檔
        
        # 初始化集合引用
        messages_collection = firestore_db.collection('messages')
        users_collection = firestore_db.collection('users')
        chats_collection = firestore_db.collection('chats')
        health_data_collection = firestore_db.collection('health_data')
        device_bindings_collection = firestore_db.collection('device_bindings')
        
        # 其他集合
        global geo_cache_collection, route_cache_collection
        geo_cache_collection = firestore_db.collection('geo_cache')
        route_cache_collection = firestore_db.collection('route_cache')
        
        logger.info(f"✅ Firestore連接成功，專案ID：{firebase_project_id}")
        logger.info(f"\n✅ Firebase Firestore連接成功！專案ID：{firebase_project_id}\n")
        return True
        
    except Exception as e:
        logger.error(f"Firebase Firestore連接失敗：{e}")
        logger.error(f"\n❌ Firebase Firestore連接失敗：{e}\n")
        logger.error("🔧 故障排除建議：")
        logger.error("1. 檢查網絡連接")
        logger.error("2. 確認Firebase服務帳戶金鑰文件路徑正確")
        logger.error("3. 驗證Firebase專案ID是否正確")
        logger.error("4. 確保Firestore Database已在Firebase Console中啟用")
        logger.error("5. 檢查服務帳戶權限是否包含Firestore權限")
        return False
def ensure_indexes():
    """Firestore 不需要手動創建索引，由 Google 自動優化"""
    logger.info("Firestore 自動處理索引優化，無需手動創建索引")



async def get_user_by_id(user_id: str):
    """根據使用者ID查找使用者，返回公共資訊"""
    if users_collection is None:
        logger.error("Firestore尚未連接，無法查找使用者")
        return None
    try:
        # Firestore 查詢 - 使用新語法
        query = users_collection.where(filter=FieldFilter("user_id", "==", user_id)).limit(1)
        docs = query.get()
        
        if not docs:
            return None
            
        user_doc = docs[0]
        user_data = user_doc.to_dict()

        result = {
            "id": user_data["user_id"],
            "name": user_data.get("name", ""),
            "email": user_data.get("email", ""),
            "created_at": user_data.get("created_at"),
        }
        # 序列化 Firestore 時間物件，避免 JSON 序列化炸裂
        return _serialize_firestore_data(result)
    except Exception as e:
        logger.error(f"查找使用者時發生錯誤: {e}")
        return None

# 已移除舊的內嵌測試函式 test_connection，避免在生產代碼夾雜測試邏輯

async def save_message(user_id, content, is_bot=False):
    """保存消息到數據庫"""
    if messages_collection is None:
        logger.error("Firestore尚未連接，無法保存消息")
        return False
    try:
        message = {
            "user_id": user_id,  # 使用user_id字段存儲用戶ID
            "content": content,
            "is_bot": is_bot,
            "timestamp": datetime.now(),
        }
        import asyncio as _asyncio
        await _asyncio.to_thread(lambda: messages_collection.add(message))
        logger.debug(f"消息已保存到 Firestore")
        return True
    except Exception as e:
        logger.error(f"保存消息時發生錯誤: {e}")
        return False

async def get_user_history(user_id, limit=20):
    """獲取用戶的歷史對話記錄"""
    if messages_collection is None:
        logger.error("Firestore尚未連接，無法獲取歷史記錄")
        return []
    try:
        import asyncio as _asyncio
        def _fetch_messages():
            docs = messages_collection.where(filter=FieldFilter("user_id", "==", user_id))\
                                    .order_by("timestamp")\
                                    .limit(limit)\
                                    .stream()
            return [doc.to_dict() for doc in docs]
        
        messages = await _asyncio.to_thread(_fetch_messages)
        logger.info(f"已獲取用戶 {user_id} 的 {len(messages)} 條歷史記錄")
        # 序列化 Firestore 時間物件，避免 JSON 序列化炸裂
        return _serialize_firestore_data(messages)
    except Exception as e:
        logger.error(f"獲取歷史記錄時發生錯誤: {e}")
        return []

# Google OAuth 2.0 用戶認證
async def create_or_login_google_user(google_token_info):
    """Google OAuth 唯一登入入口，自動處理首次註冊和後續登入"""
    if users_collection is None or firestore_db is None:
        logger.error("Firestore尚未連接，無法處理用戶認證")
        return {"success": False, "error": "數據庫未連接"}

    # 檢查 Firestore 連接狀態
    try:
        logger.info("🔍 檢查 Firestore 連接狀態...")
        # 快速連接測試
        import asyncio as _asyncio
        def _test_connection():
            test_ref = firestore_db.collection('_connection_test').document('ping')
            test_ref.set({'ping': 'test'}, merge=True)
            test_ref.delete()
            return True

        await _asyncio.wait_for(
            _asyncio.to_thread(_test_connection),
            timeout=5.0  # 5秒連接測試超時
        )
        logger.info("✅ Firestore 連接正常")
    except _asyncio.TimeoutError:
        logger.error("❌ Firestore 連接測試超時")
        return {"success": False, "error": "數據庫連接超時"}
    except Exception as e:
        logger.error(f"❌ Firestore 連接測試失敗: {e}")
        return {"success": False, "error": f"數據庫連接異常: {str(e)}"}

    google_id = google_token_info.get("id") or google_token_info.get("sub")
    if not google_id:
        logger.error(f"Google用戶信息中缺少ID字段，收到的信息: {google_token_info}")
        return {"success": False, "error": "INVALID_GOOGLE_USER_INFO"}

    email = google_token_info.get("email")
    if not email:
        logger.error(f"Google用戶信息中缺少email字段，收到的信息: {google_token_info}")
        return {"success": False, "error": "INVALID_GOOGLE_USER_INFO"}

    logger.info(f"🔍 處理Google用戶: google_id={google_id}, email={email}")

    try:
        import asyncio as _asyncio

        def _fetch_existing_user():
            try:
                logger.info(f"🔍 查詢現有用戶: google_id={google_id}")
                # 使用新的 filter 語法
                query = users_collection.where(filter=FieldFilter("google_id", "==", google_id)).limit(1)
                docs = list(query.stream())
                logger.info(f"🔍 查詢結果: 找到 {len(docs)} 個用戶")
                return docs[0] if docs else None
            except Exception as e:
                error_msg = str(e).lower()
                if "quota" in error_msg or "exceeded" in error_msg:
                    logger.error(f"❌ Firestore 配額已超出限制: {e}")
                    raise Exception("FIRESTORE_QUOTA_EXCEEDED")
                else:
                    logger.error(f"❌ Firestore 查詢失敗: {e}")
                    raise e

        logger.info(f"📤 開始查詢用戶...")
        # 添加超時機制
        try:
            user_doc = await _asyncio.wait_for(
                _asyncio.to_thread(_fetch_existing_user),
                timeout=10.0  # 10秒超時
            )
            logger.info(f"🔍 用戶查詢完成: {'找到現有用戶' if user_doc else '未找到用戶'}")
        except _asyncio.TimeoutError:
            logger.error("❌ Firestore 查詢超時（10秒）")
            return {"success": False, "error": "數據庫查詢超時"}
        except Exception as e:
            error_str = str(e)
            if "FIRESTORE_QUOTA_EXCEEDED" in error_str:
                logger.error("❌ Firestore 每日配額已用完")
                return {
                    "success": False,
                    "error": "QUOTA_EXCEEDED",
                    "message": "Firestore 每日配額已用完，請稍後再試或聯繫管理員升級服務"
                }
            else:
                logger.error(f"❌ 用戶查詢異常: {e}")
                return {"success": False, "error": f"用戶查詢失敗: {str(e)}"}

        if user_doc:
            user_data = user_doc.to_dict()

            def _update_user():
                users_collection.document(user_doc.id).update({
                    "name": google_token_info.get("name", user_data.get("name")),
                    "picture": google_token_info.get("picture", user_data.get("picture")),
                    "last_login": datetime.now(),
                    "updated_at": datetime.now()
                })

            await _asyncio.to_thread(_update_user)

            logger.info(f"用戶 {email} 登入成功，user_id: {user_data.get('user_id')}")
            return {
                "success": True,
                "user": {
                    "id": user_data.get("user_id", google_id),
                    "name": user_data.get("name", ""),
                    "email": user_data.get("email", email),
                    "picture": user_data.get("picture"),
                    "created_at": user_data.get("created_at")
                },
                "is_new_user": False
            }

        logger.info(f"📤 創建新用戶...")
        now = datetime.now()
        new_user = {
            "user_id": google_id,
            "google_id": google_id,
            "email": email,
            "name": google_token_info.get("name", ""),
            "picture": google_token_info.get("picture"),
            "locale": google_token_info.get("locale", "zh-TW"),
            "first_login": now,
            "last_login": now,
            "created_at": now,
            "updated_at": now
        }

        logger.info(f"🔍 新用戶數據: {new_user}")
        logger.info(f"📤 寫入Firestore...")
        await _asyncio.to_thread(lambda: users_collection.document(google_id).set(new_user))
        logger.info(f"✅ 新用戶 {email} 註冊成功，user_id: {google_id}")

        return {
            "success": True,
            "user": {
                "id": google_id,
                "name": new_user["name"],
                "email": new_user["email"],
                "picture": new_user["picture"],
                "created_at": new_user["created_at"]
            },
            "is_new_user": True
        }

    except Exception as e:
        logger.error(f"❌ Google OAuth 認證時發生錯誤: {e}")
        logger.error(f"❌ 錯誤類型: {type(e).__name__}")
        logger.error(f"❌ 錯誤堆疊:", exc_info=True)
        return {"success": False, "error": str(e)}

# 對話管理
async def create_chat(user_id, title="新對話"):
    if chats_collection is None:
        logger.error("Firestore尚未連接，無法創建對話")
        return {"success": False, "error": "數據庫未連接"}
    try:
        chat = {
            "user_id": user_id,
            "title": title,
            "messages": [],
            "created_at": datetime.now(),
            "updated_at": datetime.now(),
        }
        import asyncio as _asyncio
        doc_ref = await _asyncio.to_thread(lambda: chats_collection.add(chat))
        chat_id = doc_ref[1].id
        logger.info(f"為用戶 {user_id} 創建了新對話，ID: {chat_id}")
        chat_info = {
            "chat_id": chat_id,
            "user_id": user_id,
            "title": title,
            "created_at": chat["created_at"],
            "updated_at": chat["updated_at"],
        }
        # 序列化時間物件，避免 JSON 序列化炸裂
        serialized_chat_info = _serialize_firestore_data(chat_info)
        return {"success": True, "chat": serialized_chat_info}
    except Exception as e:
        logger.error(f"創建對話時發生錯誤: {e}")
        return {"success": False, "error": str(e)}

async def get_user_chats(user_id):
    if chats_collection is None:
        logger.error("Firestore尚未連接，無法獲取對話")
        return {"success": False, "error": "數據庫未連接"}
    try:
        import asyncio as _asyncio
        def _fetch_chats():
            docs = chats_collection.where(filter=FieldFilter("user_id", "==", user_id))\
                                 .order_by("updated_at", direction=firestore.Query.DESCENDING)\
                                 .stream()
            chats = []
            for doc in docs:
                chat = doc.to_dict()
                chat["chat_id"] = doc.id
                if "user_id" in chat:
                    del chat["user_id"]
                if "messages" in chat:
                    del chat["messages"]
                if "created_at" in chat:
                    del chat["created_at"]
                chats.append(chat)
            return chats
        
        chats = await _asyncio.to_thread(_fetch_chats)
        logger.info(f"獲取到用戶 {user_id} 的 {len(chats)} 個對話")
        # 序列化 Firestore 時間物件，避免 JSON 序列化炸裂
        serialized_chats = _serialize_firestore_data(chats)
        return {"success": True, "chats": serialized_chats}
    except Exception as e:
        logger.error(f"獲取用戶對話時發生錯誤: {e}")
        return {"success": False, "error": str(e)}

async def get_chat(chat_id):
    if chats_collection is None:
        logger.error("Firestore尚未連接，無法獲取對話")
        return {"success": False, "error": "數據庫未連接"}
    try:
        import asyncio as _asyncio
        def _get_doc():
            doc = chats_collection.document(chat_id).get()
            return doc if doc.exists else None
        
        doc = await _asyncio.to_thread(_get_doc)
        if not doc:
            logger.warning(f"對話 {chat_id} 不存在")
            return {"success": False, "error": "對話不存在"}
        
        chat = doc.to_dict() or {}
        chat["chat_id"] = doc.id

        # 從 chat 子集合讀取完整對話（按時間升序）
        try:
            def _fetch_msgs():
                ref = _get_chat_messages_collection(chat_id)
                return [
                    {**doc.to_dict(), "id": doc.id}
                    for doc in ref.order_by("timestamp").stream()
                ]

            msgs = await _asyncio.to_thread(_fetch_msgs)
            chat["messages"] = msgs
            logger.info(f"獲取到對話 {chat_id}，包含 {len(msgs)} 條消息（chat 子集合）")
        except Exception as _e:
            # 向後相容：若讀取失敗，退回文件內嵌 messages（若存在）
            msgs_fallback = chat.get('messages', []) or []
            chat["messages"] = msgs_fallback
            logger.warning(f"讀取 chat 子集合失敗，使用內嵌 messages。原因: {_e}")

        return {"success": True, "chat": chat}
    except Exception as e:
        logger.error(f"獲取對話時發生錯誤: {e}")
        return {"success": False, "error": str(e)}

async def save_chat_message(chat_id, sender, content):
    """保存對話消息（chat/{chat_id}/messages 子集合作為主要儲存）"""
    if chats_collection is None:
        logger.error("Firestore尚未連接，無法保存消息")
        return {"success": False, "error": "數據庫未連接"}
    try:
        import asyncio as _asyncio

        now = datetime.now()
        message = {
            "chat_id": chat_id,
            "sender": sender,
            "content": content,
            "timestamp": now,
        }

        def _write_message():
            ref = _get_chat_messages_collection(chat_id)
            ref.add(message)

        def _write_legacy_copy():
            if messages_collection is None:
                return
            try:
                messages_collection.add(message)
            except Exception as legacy_err:  # pragma: no cover
                logger.debug(f"寫入頂層 messages 集合失敗（兼容用途，可忽略）: {legacy_err}")

        def _touch_chat():
            doc_ref = chats_collection.document(chat_id)
            snap = doc_ref.get()
            if not snap.exists:
                return False
            doc_ref.update({"updated_at": now})
            return True

        await _asyncio.to_thread(_write_message)
        # 兼容舊資料模型：非阻塞地寫入頂層 messages 集合，供舊功能查詢使用
        await _asyncio.to_thread(_write_legacy_copy)
        touched = await _asyncio.to_thread(_touch_chat)
        if not touched:
            logger.warning(f"對話 {chat_id} 不存在，但消息已寫入 chat 子集合")

        logger.info(f"消息已保存到 chat 子集合（chat_id={chat_id}）")
        return {"success": True, "message": message}
    except Exception as e:
        logger.error(f"保存消息時發生錯誤: {e}")
        return {"success": False, "error": str(e)}


async def get_chat_messages(chat_id: str, limit: int | None = None, ascending: bool = True):
    """讀取指定對話的消息（優先使用 chat 子集合）"""
    if chats_collection is None:
        logger.error("Firestore尚未連接，無法讀取消息")
        return []
    try:
        import asyncio as _asyncio
        from google.cloud import firestore as _fs

        def _query():
            ref = _get_chat_messages_collection(chat_id)
            direction = _fs.Query.ASCENDING if ascending else _fs.Query.DESCENDING
            q = ref.order_by("timestamp", direction=direction)
            if limit and limit > 0:
                q = q.limit(limit)
            docs = q.stream()
            records = []
            for doc in docs:
                data = doc.to_dict()
                data["id"] = doc.id
                records.append(data)
            if not ascending:
                records = list(reversed(records))
            return records

        messages = await _asyncio.to_thread(_query)
        if messages:
            # 序列化 Firestore 時間物件，避免 JSON 序列化炸裂
            return _serialize_firestore_data(messages)

        # 向後相容：若子集合無資料，嘗試讀取舊頂層 messages 集合
        if messages_collection is None:
            return []

        def _legacy_query():
            docs = messages_collection.where(filter=FieldFilter("chat_id", "==", chat_id)).stream()
            legacy = [d.to_dict() for d in docs]
            legacy.sort(key=lambda item: item.get("timestamp"))
            if limit and limit > 0:
                legacy = legacy[:limit]
            return legacy

        legacy_sorted = await _asyncio.to_thread(_legacy_query)
        view_messages = list(legacy_sorted)
        if not ascending:
            view_messages.reverse()
        if legacy_sorted:
            def _backfill():
                try:
                    ref = _get_chat_messages_collection(chat_id)
                    # 若子集合仍為空，將舊資料搬遷過去
                    has_existing = any(True for _ in ref.limit(1).stream())
                    if has_existing:
                        return
                    for legacy_msg in legacy_sorted:
                        ref.add(legacy_msg)
                    logger.info(f"已將 legacy messages 回填至 chat 子集合（chat_id={chat_id}）")
                except Exception as backfill_err:
                    logger.warning(f"回填 legacy messages 失敗（可忽略）: {backfill_err}")

            await _asyncio.to_thread(_backfill)
        # 序列化 Firestore 時間物件，避免 JSON 序列化炸裂
        return _serialize_firestore_data(view_messages)
    except Exception as e:
        logger.error(f"讀取對話消息失敗: {e}")
        return []

async def update_chat_title(chat_id, title):
    if chats_collection is None:
        logger.error("Firestore尚未連接，無法更新對話標題")
        return {"success": False, "error": "數據庫未連接"}
    try:
        import asyncio as _asyncio

        def _update_doc():
            doc_ref = chats_collection.document(chat_id)
            doc = doc_ref.get()
            if not doc.exists:
                return False
            doc_ref.update({
                "title": title,
                "updated_at": datetime.now(),
            })
            return True

        updated = await _asyncio.to_thread(_update_doc)
        if not updated:
            logger.warning(f"對話 {chat_id} 不存在，無法更新標題")
            return {"success": False, "error": "對話不存在"}

        logger.info(f"對話 {chat_id} 標題已更新為 '{title}'")
        return {"success": True}
    except Exception as e:
        logger.error(f"更新對話標題時發生錯誤: {e}")
        return {"success": False, "error": str(e)}

async def delete_chat(chat_id):
    if chats_collection is None:
        logger.error("Firestore尚未連接，無法刪除對話")
        return {"success": False, "error": "數據庫未連接"}
    try:
        import asyncio as _asyncio

        def _delete_doc():
            doc_ref = chats_collection.document(chat_id)
            doc = doc_ref.get()
            if not doc.exists:
                return False
            # 先刪除子集合中的消息，避免孤兒資料
            try:
                messages_ref = _get_chat_messages_collection(chat_id)
                for msg_snapshot in messages_ref.stream():
                    msg_snapshot.reference.delete()
            except Exception as msg_err:
                logger.warning(f"刪除對話 {chat_id} 的子消息時發生錯誤：{msg_err}")
            doc_ref.delete()
            return True

        deleted = await _asyncio.to_thread(_delete_doc)
        if not deleted:
            logger.warning(f"對話 {chat_id} 不存在，無法刪除")
            return {"success": False, "error": "對話不存在"}

        logger.info(f"對話 {chat_id} 已刪除")
        return {"success": True}
    except Exception as e:
        logger.error(f"刪除對話時發生錯誤: {e}")
        return {"success": False, "error": str(e)}


# ===== 對話情緒記憶 =====
async def set_chat_emotion(chat_id: str, emotion: dict):
    """為指定對話記錄最近的情緒狀態（label, confidence, timestamp）。"""
    if chats_collection is None:
        logger.error("Firestore尚未連接，無法設定對話情緒")
        return {"success": False, "error": "數據庫未連接"}
    try:
        import asyncio as _asyncio

        def _update_doc():
            doc_ref = chats_collection.document(chat_id)
            doc = doc_ref.get()
            if not doc.exists:
                return False
            payload = {
                "label": emotion.get("label"),
                "confidence": emotion.get("confidence"),
                "timestamp": datetime.now(),
            }
            doc_ref.update({
                "context.emotion": payload,
                "updated_at": datetime.now(),
            })
            return True

        updated = await _asyncio.to_thread(_update_doc)
        if not updated:
            return {"success": False, "error": "對話不存在"}
        return {"success": True}
    except Exception as e:
        logger.error(f"設定對話情緒時發生錯誤: {e}")
        return {"success": False, "error": str(e)}

async def get_chat_emotion(chat_id: str):
    """取得對話記錄的最近情緒狀態。"""
    if chats_collection is None:
        logger.error("Firestore尚未連接，無法讀取對話情緒")
        return {"success": False, "error": "數據庫未連接"}
    try:
        import asyncio as _asyncio

        def _get_doc():
            doc = chats_collection.document(chat_id).get()
            return doc if doc.exists else None

        doc = await _asyncio.to_thread(_get_doc)
        if not doc:
            return {"success": False, "error": "對話不存在"}
        data = doc.to_dict() or {}
        emotion = (data.get("context") or {}).get("emotion")
        return {"success": True, "emotion": emotion}
    except Exception as e:
        logger.error(f"讀取對話情緒時發生錯誤: {e}")
        return {"success": False, "error": str(e)}

# ===== 語音登入：使用者與說話者標籤關聯 =====
async def set_user_speaker_label(user_id: str, speaker_label: str):
    if users_collection is None:
        logger.error("Firestore尚未連接，無法設定語音標籤")
        return {"success": False, "error": "數據庫未連接"}
    try:
        import asyncio as _asyncio

        def _is_label_taken():
            docs = list(users_collection.where(filter=FieldFilter("speaker_label", "==", speaker_label)).limit(1).stream())
            return docs[0] if docs else None

        existing_label = await _asyncio.to_thread(_is_label_taken)
        if existing_label and existing_label.to_dict().get("user_id") != user_id:
            return {"success": False, "error": "SPEAKER_LABEL_TAKEN"}

        def _update_user():
            doc_ref = users_collection.document(user_id)
            doc = doc_ref.get()
            if not doc.exists:
                return False
            doc_ref.update({"speaker_label": speaker_label})
            return True

        updated = await _asyncio.to_thread(_update_user)
        if not updated:
            return {"success": False, "error": "USER_NOT_FOUND"}

        return {"success": True}
    except Exception as e:
        logger.error(f"設定語音標籤時發生錯誤: {e}")
        return {"success": False, "error": str(e)}


async def get_user_by_speaker_label(speaker_label: str):
    if users_collection is None:
        logger.error("Firestore尚未連接，無法查詢語音標籤")
        return None
    try:
        import asyncio as _asyncio

        def _fetch_user():
            docs = list(users_collection.where(filter=FieldFilter("speaker_label", "==", speaker_label)).limit(1).stream())
            return docs[0] if docs else None

        doc = await _asyncio.to_thread(_fetch_user)
        if not doc:
            return None

        data = doc.to_dict()
        result = {
            "id": data.get("user_id"),
            "name": data.get("name", ""),
            "email": data.get("email", ""),
            "created_at": data.get("created_at"),
        }
        # 序列化 Firestore 時間物件，避免 JSON 序列化炸裂
        return _serialize_firestore_data(result)
    except Exception as e:
        logger.error(f"查詢語音標籤對應用戶時發生錯誤: {e}")
        return None


# ===== 專門記憶系統 =====

async def save_memory(
    user_id: str,
    memory_type: str,
    content: str,
    importance: float = 1.0,
    metadata: dict | None = None,
) -> Dict[str, Any]:
    """保存重要記憶到 Firestore"""
    if users_collection is None or firestore_db is None:
        logger.error("Firestore尚未連接，無法保存記憶")
        return {"success": False, "error": "數據庫未連接"}

    try:
        import asyncio as _asyncio

        now = datetime.now()
        sanitized_importance = max(0.0, min(1.0, importance))
        metadata_payload = metadata.copy() if metadata else {}
        metadata_payload.setdefault("source", "unknown")
        metadata_payload.setdefault("last_updated_by", "memory_system")
        metadata_payload["updated_at"] = now.isoformat()

        context_tags = metadata_payload.get("context_tags", [])
        if not isinstance(context_tags, list):
            context_tags = list(context_tags) if context_tags else []
        metadata_payload["context_tags"] = context_tags

        triggers = metadata_payload.get("triggers", [])
        if not isinstance(triggers, list):
            triggers = list(triggers) if triggers else []
        metadata_payload["triggers"] = triggers

        col_ref = _get_user_memories_collection(user_id)
        content_hash = hashlib.sha1(content.strip().lower().encode("utf-8")).hexdigest()

        def _ensure_user_stub():
            user_doc = _get_user_doc_ref(user_id)
            snap = user_doc.get()
            if not snap.exists:
                user_doc.set(
                    {
                        "user_id": user_id,
                        "created_at": now,
                        "updated_at": now,
                    },
                    merge=True,
                )

        def _find_existing():
            docs = (
                col_ref.where(filter=FieldFilter("content_hash", "==", content_hash))
                .limit(1)
                .stream()
            )
            for doc in docs:
                return doc
            return None

        await _asyncio.to_thread(_ensure_user_stub)
        existing_doc = await _asyncio.to_thread(_find_existing)

        if existing_doc:
            doc_ref = existing_doc.reference

            def _update_memory():
                doc_ref.update({
                    "content": content,
                    "importance": sanitized_importance,
                    "metadata": metadata_payload,
                    "updated_at": now,
                    "access_count": Increment(1),
                    "last_accessed": now,
                    "content_hash": content_hash,
                })

            await _asyncio.to_thread(_update_memory)
            logger.info(f"更新用戶 {user_id} 的記憶: {memory_type}")
            return {"success": True, "action": "updated", "memory_id": existing_doc.id}

        def _create_memory():
            doc_ref = col_ref.document()
            doc_ref.set({
                "user_id": user_id,
                "type": memory_type,
                "content": content,
                "importance": sanitized_importance,
                "metadata": metadata_payload,
                "access_count": 0,
                "last_accessed": now,
                "updated_at": now,
                "created_at": now,
                "content_hash": content_hash,
            })
            return doc_ref.id

        memory_id = await _asyncio.to_thread(_create_memory)
        await _asyncio.to_thread(_enforce_memory_quota, col_ref)
        logger.info(f"保存用戶 {user_id} 的新記憶: {memory_type}")
        return {"success": True, "action": "created", "memory_id": memory_id}

    except Exception as e:
        logger.error(f"保存記憶時發生錯誤: {e}")
        return {"success": False, "error": str(e)}


# ===== 環境 Context（位置/方位/時序） =====

async def set_user_env_current(user_id: str, ctx: Dict[str, Any]) -> Dict[str, Any]:
    """更新使用者環境現況 users/{uid}/context/current（含 TTL/新鮮度由讀取端判斷）。"""
    if users_collection is None:
        return {"success": False, "error": "數據庫未連接"}
    try:
        import asyncio as _asyncio
        now = datetime.now()

        def _update():
            user_doc = _get_user_doc_ref(user_id)
            ctx_ref = user_doc.collection('context').document('current')
            payload = ctx.copy()
            payload['updated_at'] = now
            ctx_ref.set(payload, merge=True)
            return True

        await _asyncio.to_thread(_update)
        return {"success": True}
    except Exception as e:
        logger.error(f"更新環境現況失敗: {e}")
        return {"success": False, "error": str(e)}


async def add_user_env_snapshot(user_id: str, snapshot: Dict[str, Any]) -> Dict[str, Any]:
    """新增使用者環境快照 users/{uid}/context/snapshots。僅保留短期歷史。"""
    if users_collection is None:
        return {"success": False, "error": "數據庫未連接"}
    try:
        import asyncio as _asyncio
        now = datetime.now()

        def _write():
            user_doc = _get_user_doc_ref(user_id)
            col = user_doc.collection('context').document('meta').collection('snapshots')
            payload = snapshot.copy()
            payload['created_at'] = now
            col.add(payload)
            return True

        await _asyncio.to_thread(_write)
        return {"success": True}
    except Exception as e:
        logger.error(f"寫入環境快照失敗: {e}")
        return {"success": False, "error": str(e)}


async def get_user_env_current(user_id: str) -> Dict[str, Any]:
    """讀取使用者環境現況。"""
    if users_collection is None:
        return {"success": False, "error": "數據庫未連接"}
    try:
        import asyncio as _asyncio

        def _read():
            user_doc = _get_user_doc_ref(user_id)
            ctx_ref = user_doc.collection('context').document('current')
            snap = ctx_ref.get()
            return snap.to_dict() if snap.exists else None

        data = await _asyncio.to_thread(_read)
        if not data:
            return {"success": False, "error": "NOT_FOUND"}
        # 序列化 Firestore 時間物件，避免 JSON 序列化炸裂
        serialized_data = _serialize_firestore_data(data)
        return {"success": True, "context": serialized_data}
    except Exception as e:
        logger.error(f"讀取環境現況失敗: {e}")
        return {"success": False, "error": str(e)}


# ===== 反地理/路線 全域快取集合 =====

async def get_geo_cache(geohash7: str) -> Optional[Dict[str, Any]]:
    if geo_cache_collection is None:
        return None
    try:
        import asyncio as _asyncio
        def _read():
            doc = geo_cache_collection.document(geohash7).get()
            return doc.to_dict() if doc.exists else None
        result = await _asyncio.to_thread(_read)
        # 序列化 Firestore 時間物件，避免 JSON 序列化炸裂
        return _serialize_firestore_data(result) if result else None
    except Exception as e:
        logger.warning(f"讀取 geo_cache 失敗: {e}")
        return None


async def set_geo_cache(geohash7: str, payload: Dict[str, Any]) -> bool:
    if geo_cache_collection is None:
        return False
    try:
        import asyncio as _asyncio
        now = datetime.now()
        def _write():
            data = payload.copy()
            data['cached_at'] = now
            geo_cache_collection.document(geohash7).set(data, merge=True)
            return True
        return await _asyncio.to_thread(_write)
    except Exception as e:
        logger.warning(f"寫入 geo_cache 失敗: {e}")
        return False


async def get_route_cache(key: str) -> Optional[Dict[str, Any]]:
    if route_cache_collection is None:
        return None
    try:
        import asyncio as _asyncio
        def _read():
            doc = route_cache_collection.document(key).get()
            return doc.to_dict() if doc.exists else None
        result = await _asyncio.to_thread(_read)
        # 序列化 Firestore 時間物件，避免 JSON 序列化炸裂
        return _serialize_firestore_data(result) if result else None
    except Exception as e:
        logger.warning(f"讀取 route_cache 失敗: {e}")
        return None


async def set_route_cache(key: str, payload: Dict[str, Any]) -> bool:
    if route_cache_collection is None:
        return False
    try:
        import asyncio as _asyncio
        now = datetime.now()
        def _write():
            data = payload.copy()
            data['cached_at'] = now
            route_cache_collection.document(key).set(data, merge=True)
            return True
        return await _asyncio.to_thread(_write)
    except Exception as e:
        logger.warning(f"寫入 route_cache 失敗: {e}")
        return False


async def get_user_memories(
    user_id: str,
    memory_type: str | None = None,
    limit: int = 10,
    min_importance: float = 0.0,
) -> Dict[str, Any]:
    """獲取用戶的記憶"""
    if users_collection is None:
        logger.error("Firestore尚未連接，無法獲取記憶")
        return {"success": False, "error": "數據庫未連接"}

    try:
        import asyncio as _asyncio

        def _fetch_memories():
            col_ref = _get_user_memories_collection(user_id)
            query = col_ref.where(filter=FieldFilter("importance", ">=", min_importance))
            if memory_type:
                query = query.where(filter=FieldFilter("type", "==", memory_type))
            docs = (
                query.order_by("importance", direction=firestore.Query.DESCENDING)
                .order_by("updated_at", direction=firestore.Query.DESCENDING)
                .limit(limit)
                .stream()
            )
            return [doc.to_dict() | {"memory_id": doc.id} for doc in docs]

        memories = await _asyncio.to_thread(_fetch_memories)

        def _mark_accessed(mem_ids):
            col_ref = _get_user_memories_collection(user_id)
            now_inner = datetime.now()
            for mid in mem_ids:
                col_ref.document(mid).update({
                    "access_count": Increment(1),
                    "last_accessed": now_inner,
                })

        if memories:
            await _asyncio.to_thread(_mark_accessed, [m["memory_id"] for m in memories])

        logger.info(f"獲取到用戶 {user_id} 的 {len(memories)} 條記憶")
        # 序列化 Firestore 時間物件，避免 JSON 序列化炸裂
        serialized_memories = _serialize_firestore_data(memories)
        return {"success": True, "memories": serialized_memories}

    except Exception as e:
        logger.error(f"獲取記憶時發生錯誤: {e}")
        return {"success": False, "error": str(e)}


async def search_memories(user_id: str, query_text: str, limit: int = 5) -> Dict[str, Any]:
    """基於簡易文本匹配的記憶搜索"""
    if users_collection is None:
        logger.error("Firestore尚未連接，無法搜索記憶")
        return {"success": False, "error": "數據庫未連接"}

    try:
        import asyncio as _asyncio

        normalized_query = query_text.lower()

        def _candidate_memories():
            col_ref = _get_user_memories_collection(user_id)
            docs = (
                col_ref.order_by("updated_at", direction=firestore.Query.DESCENDING)
                .limit(80)
                .stream()
            )
            results = []
            for doc in docs:
                data = doc.to_dict() or {}
                haystack = "{} {}".format(
                    data.get("content", ""),
                    " ".join(data.get("metadata", {}).get("context_tags", [])),
                ).lower()
                if normalized_query in haystack:
                    data["memory_id"] = doc.id
                    results.append(data)
                    if len(results) >= limit:
                        break
            return results

        memories = await _asyncio.to_thread(_candidate_memories)

        def _mark_accessed(mem_ids):
            col_ref = _get_user_memories_collection(user_id)
            now_inner = datetime.now()
            for mid in mem_ids:
                col_ref.document(mid).update({
                    "access_count": Increment(1),
                    "last_accessed": now_inner,
                })

        if memories:
            await _asyncio.to_thread(_mark_accessed, [m["memory_id"] for m in memories])

        logger.info(f"搜索到用戶 {user_id} 的 {len(memories)} 條相關記憶")
        return {"success": True, "memories": memories}

    except Exception as e:
        logger.error(f"搜索記憶時發生錯誤: {e}")
        return {"success": False, "error": str(e)}


async def update_memory_importance(memory_id: str, new_importance: float):
    """更新記憶的重要性分數"""
    if users_collection is None:
        logger.error("Firestore尚未連接，無法更新記憶")
        return {"success": False, "error": "數據庫未連接"}

    try:
        import asyncio as _asyncio

        sanitized_importance = max(0.0, min(1.0, new_importance))
        now = datetime.now()

        def _update_doc():
            users = users_collection.stream()
            for user_doc in users:
                mem_ref = user_doc.reference.collection("memories").document(memory_id)
                snapshot = mem_ref.get()
                if snapshot.exists:
                    mem_ref.update({
                        "importance": sanitized_importance,
                        "updated_at": now,
                    })
                    return True
            return False

        updated = await _asyncio.to_thread(_update_doc)
        if not updated:
            return {"success": False, "error": "記憶不存在"}

        logger.info(f"更新記憶 {memory_id} 的重要性為 {sanitized_importance}")
        return {"success": True}

    except Exception as e:
        logger.error(f"更新記憶重要性時發生錯誤: {e}")
        return {"success": False, "error": str(e)}


async def delete_memory(memory_id: str):
    """刪除記憶"""
    if users_collection is None:
        logger.error("Firestore尚未連接，無法刪除記憶")
        return {"success": False, "error": "數據庫未連接"}

    try:
        import asyncio as _asyncio

        def _delete_doc():
            users = users_collection.stream()
            for user_doc in users:
                mem_ref = user_doc.reference.collection("memories").document(memory_id)
                snapshot = mem_ref.get()
                if snapshot.exists:
                    mem_ref.delete()
                    return True
            return False

        deleted = await _asyncio.to_thread(_delete_doc)
        if not deleted:
            return {"success": False, "error": "記憶不存在"}

        logger.info(f"刪除記憶 {memory_id}")
        return {"success": True}

    except Exception as e:
        logger.error(f"刪除記憶時發生錯誤: {e}")
        return {"success": False, "error": str(e)}


async def cleanup_old_memories(user_id: str, days_old: int = 90, min_importance: float = 0.3):
    """清理舊的、低重要性的記憶

    Args:
        user_id: 用戶ID
        days_old: 刪除多少天前的記憶
        min_importance: 保留的最小重要性分數
    """
    if users_collection is None:
        logger.error("Firestore尚未連接，無法清理記憶")
        return {"success": False, "error": "數據庫未連接"}

    try:
        import asyncio as _asyncio
        from datetime import timedelta

        cutoff_date = datetime.now() - timedelta(days=days_old)

        def _delete_old():
            col_ref = _get_user_memories_collection(user_id)
            docs = (
                col_ref.where(filter=FieldFilter("importance", "<", min_importance))
                .where(filter=FieldFilter("updated_at", "<", cutoff_date))
                .stream()
            )
            deleted_count = 0
            for doc in docs:
                doc.reference.delete()
                deleted_count += 1
            return deleted_count

        deleted = await _asyncio.to_thread(_delete_old)

        logger.info(f"為用戶 {user_id} 清理 {deleted} 條舊記憶")
        return {"success": True, "deleted": deleted}

    except Exception as e:
        logger.error(f"清理記憶時發生錯誤: {e}")
        return {"success": False, "error": str(e)}


def _enforce_memory_quota(col_ref: CollectionReference) -> None:
    docs = list(
        col_ref.order_by("importance", direction=firestore.Query.ASCENDING)
        .order_by("updated_at", direction=firestore.Query.ASCENDING)
        .stream()
    )
    if len(docs) <= MAX_MEMORIES_PER_USER:
        return
    excess = len(docs) - MAX_MEMORIES_PER_USER
    for doc in docs[:excess]:
        doc.reference.delete()