""" Database Layer — PostgreSQL (asyncpg) ────────────────────────────────────── เก็บ conversation history และ sentiment results """ import logging import asyncpg from datetime import datetime, timezone from app.config import settings logger = logging.getLogger(__name__) CREATE_TABLE_SQL = """ CREATE TABLE IF NOT EXISTS conversations ( id SERIAL PRIMARY KEY, conv_id TEXT NOT NULL UNIQUE, user_id TEXT NOT NULL, message TEXT NOT NULL, source_type TEXT DEFAULT 'user', timestamp BIGINT NOT NULL, -- LINE timestamp (ms) created_at TIMESTAMPTZ DEFAULT NOW(), -- Sentiment fields (nullable ก่อน analyze เสร็จ) sentiment_label TEXT, -- positive | neutral | negative sentiment_score FLOAT, -- confidence score 0–1 raw_scores JSONB -- {"positive": 0.8, ...} ); CREATE INDEX IF NOT EXISTS idx_conv_user ON conversations(user_id); CREATE INDEX IF NOT EXISTS idx_conv_created ON conversations(created_at DESC); CREATE INDEX IF NOT EXISTS idx_conv_sentiment ON conversations(sentiment_label); """ class Database: def __init__(self): self._pool: asyncpg.Pool | None = None async def connect(self): self._pool = await asyncpg.create_pool( dsn=settings.DATABASE_URL, min_size=2, max_size=10, ) logger.info("✅ Database pool created") async def disconnect(self): if self._pool: await self._pool.close() async def _pool_ok(self) -> asyncpg.Pool: if not self._pool: await self.connect() return self._pool # ── Write ────────────────────────────────────────────────────────────────── async def save_conversation( self, conv_id: str, user_id: str, message: str, timestamp: int, source_type: str = "user", ) -> dict: pool = await self._pool_ok() async with pool.acquire() as conn: row = await conn.fetchrow( """ INSERT INTO conversations (conv_id, user_id, message, source_type, timestamp) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (conv_id) DO NOTHING RETURNING * """, conv_id, user_id, message, source_type, timestamp, ) return dict(row) if row else {} async def update_sentiment( self, conv_id: str, label: str, score: float, raw_scores: dict, ) -> None: import json pool = await self._pool_ok() async with pool.acquire() as conn: await conn.execute( """ UPDATE conversations SET sentiment_label = $1, sentiment_score = $2, raw_scores = $3 WHERE conv_id = $4 """, label, score, json.dumps(raw_scores), conv_id, ) # ── Read ─────────────────────────────────────────────────────────────────── async def get_conversations(self, limit: int = 50, offset: int = 0) -> list[dict]: pool = await self._pool_ok() async with pool.acquire() as conn: rows = await conn.fetch( """ SELECT conv_id, user_id, message, sentiment_label, sentiment_score, raw_scores, created_at FROM conversations ORDER BY created_at DESC LIMIT $1 OFFSET $2 """, limit, offset, ) return [dict(r) for r in rows] async def get_sentiment_summary(self) -> dict: """สรุปยอดรวม sentiment สำหรับ stat cards""" pool = await self._pool_ok() async with pool.acquire() as conn: rows = await conn.fetch( """ SELECT COUNT(*) FILTER (WHERE sentiment_label = 'positive') AS positive, COUNT(*) FILTER (WHERE sentiment_label = 'neutral') AS neutral, COUNT(*) FILTER (WHERE sentiment_label = 'negative') AS negative, COUNT(*) AS total FROM conversations WHERE created_at >= NOW() - INTERVAL '7 days' """ ) row = dict(rows[0]) total = row["total"] or 1 # ป้องกัน div/0 return { "positive": {"count": row["positive"], "pct": round(row["positive"] / total * 100, 1)}, "neutral": {"count": row["neutral"], "pct": round(row["neutral"] / total * 100, 1)}, "negative": {"count": row["negative"], "pct": round(row["negative"] / total * 100, 1)}, "total": row["total"], "period": "last 7 days", } async def init_db(): """สร้าง table ถ้ายังไม่มี (รันตอน startup)""" pool = await asyncpg.create_pool(dsn=settings.DATABASE_URL) async with pool.acquire() as conn: await conn.execute(CREATE_TABLE_SQL) await pool.close() logger.info("✅ Database initialized") # Singleton db = Database()