Spaces:
Running
Running
File size: 4,419 Bytes
b7d08cf a7c1db4 b7d08cf 529a8bd b7d08cf 529a8bd b7d08cf 529a8bd b7d08cf 529a8bd b7d08cf 529a8bd b7d08cf 529a8bd b7d08cf 529a8bd b7d08cf 529a8bd b7d08cf 529a8bd b7d08cf 529a8bd b7d08cf 529a8bd |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 |
import sqlite3
import json
import uuid
import os
from typing import Any, Dict, Optional
from src.infra.context import get_session_id
from src.infra.logger import get_logger
logger = get_logger(__name__)
class PoiRepository:
# =========================================================================
# 1. 設定絕對路徑 (Class Attributes) - 確保在 __init__ 之前就存在
# =========================================================================
# 抓出這個檔案 (poi_repository.py) 的絕對路徑: .../LifeFlow-AI/src/infra
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
# 往上推兩層回到專案根目錄: .../LifeFlow-AI
# (src/infra -> src -> LifeFlow-AI)
PROJECT_ROOT = os.path.dirname(os.path.dirname(BASE_DIR))
# 設定 DB 存放目錄: .../LifeFlow-AI/storage
DB_DIR = os.path.join(PROJECT_ROOT, "storage")
# 設定 DB 完整路徑: .../LifeFlow-AI/storage/lifeflow_payloads.db
DB_FILE = "lifeflow_payloads.db"
DB_PATH = os.path.join(DB_DIR, DB_FILE)
# =========================================================================
# 2. 初始化邏輯
# =========================================================================
def __init__(self):
# 1. 先確保資料夾存在 (這時候 self.DB_DIR 已經讀得到了)
os.makedirs(self.DB_DIR, exist_ok=True)
# 2. 再初始化 DB (這時候 self.DB_PATH 已經讀得到了)
self._init_db()
def _init_db(self):
# 使用 self.DB_PATH
with sqlite3.connect(self.DB_PATH) as conn:
# 1. 既有的資料表
conn.execute("""
CREATE TABLE IF NOT EXISTS offloaded_data (
ref_id TEXT PRIMARY KEY,
data_type TEXT,
payload JSON,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# 2. Session 狀態表
conn.execute("""
CREATE TABLE IF NOT EXISTS session_state (
session_id TEXT PRIMARY KEY,
last_ref_id TEXT,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# =========================================================================
# 3. 讀寫操作
# =========================================================================
def save(self, data: Any, data_type: str = "generic") -> str:
ref_id = f"{data_type}_{uuid.uuid4().hex[:8]}"
with sqlite3.connect(self.DB_PATH) as conn:
# 1. 寫入 Payload
conn.execute(
"INSERT INTO offloaded_data (ref_id, data_type, payload) VALUES (?, ?, ?)",
(ref_id, data_type, json.dumps(data, default=str))
)
# 2. 更新 Session 的最後 ID
current_session = get_session_id()
if current_session:
conn.execute("""
INSERT OR REPLACE INTO session_state (session_id, last_ref_id)
VALUES (?, ?)
""", (current_session, ref_id))
logger.info(f"💾 [Repo] Saved {ref_id} for Session: {current_session}")
else:
# 這裡改用 debug level，以免 log 太多雜訊
logger.debug(f"⚠️ [Repo] No session context found, 'last_id' not tracked.")
return ref_id
def load(self, ref_id: str) -> Optional[Any]:
if not ref_id: return None
# 容錯：如果路徑還沒建立 (理論上 init 已建立，但防呆)
if not os.path.exists(self.DB_PATH):
return None
with sqlite3.connect(self.DB_PATH) as conn:
cursor = conn.execute("SELECT payload FROM offloaded_data WHERE ref_id = ?", (ref_id,))
row = cursor.fetchone()
if row:
return json.loads(row[0])
return None
def get_last_id_by_session(self, session_id: str) -> Optional[str]:
if not os.path.exists(self.DB_PATH):
return None
with sqlite3.connect(self.DB_PATH) as conn:
cursor = conn.execute("SELECT last_ref_id FROM session_state WHERE session_id = ?", (session_id,))
row = cursor.fetchone()
if row:
return row[0]
return None
# 全域單例
poi_repo = PoiRepository()
|