database.py

"""
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()