memory.py

"""
memory.py
SQLite-backed long-term memory for Otaku Concierge Pro.
Stores:
- user preferences
- watch history
- recommendation logs
"""

import os
import sqlite3
from datetime import datetime
from typing import Optional, Dict, Any, List

DB_PATH = os.environ.get("OTAKU_DB_PATH", "otaku_memory.db")


def get_conn() -> sqlite3.Connection:
    """Create and return a SQLite connection with row factory."""
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


def init_db() -> Dict[str, Any]:
    """Initialise the SQLite database with all required tables."""
    conn = get_conn()
    cur = conn.cursor()

    # User preferences
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS user_preferences (
            user_id TEXT PRIMARY KEY,
            fav_genres TEXT,
            disliked_genres TEXT,
            preferred_runtime TEXT,
            last_updated TEXT
        )
        """
    )

    # Watch history
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS watch_history (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id TEXT,
            title TEXT,
            kind TEXT,
            genres TEXT,
            imdb_rating REAL,
            user_rating REAL,
            watched_at TEXT
        )
        """
    )

    # Recommendation log
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS recommendation_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id TEXT,
            title TEXT,
            kind TEXT,
            imdb_rating REAL,
            reason TEXT,
            recommended_at TEXT
        )
        """
    )

    conn.commit()
    conn.close()

    return {"status": "ok", "message": "Database initialised.", "db_path": DB_PATH}


def save_user_preferences(
    user_id: str,
    fav_genres: Optional[str] = None,
    disliked_genres: Optional[str] = None,
    preferred_runtime: Optional[str] = None,
) -> Dict[str, Any]:
    """
    Insert/update user's taste preferences.
    Genres and runtime are stored as simple strings (e.g. comma-separated).
    """
    conn = get_conn()
    cur = conn.cursor()

    now = datetime.utcnow().isoformat()

    cur.execute(
        """
        INSERT INTO user_preferences (user_id, fav_genres, disliked_genres, preferred_runtime, last_updated)
        VALUES (?, ?, ?, ?, ?)
        ON CONFLICT(user_id) DO UPDATE SET
            fav_genres = COALESCE(EXCLUDED.fav_genres, user_preferences.fav_genres),
            disliked_genres = COALESCE(EXCLUDED.disliked_genres, user_preferences.disliked_genres),
            preferred_runtime = COALESCE(EXCLUDED.preferred_runtime, user_preferences.preferred_runtime),
            last_updated = EXCLUDED.last_updated
        """,
        (user_id, fav_genres, disliked_genres, preferred_runtime, now),
    )

    conn.commit()
    conn.close()

    return {
        "status": "ok",
        "user_id": user_id,
        "fav_genres": fav_genres,
        "disliked_genres": disliked_genres,
        "preferred_runtime": preferred_runtime,
    }


def load_user_preferences(user_id: str) -> Dict[str, Any]:
    """Return user preferences row if present."""
    conn = get_conn()
    cur = conn.cursor()

    cur.execute(
        """
        SELECT user_id, fav_genres, disliked_genres, preferred_runtime, last_updated
        FROM user_preferences
        WHERE user_id = ?
        """,
        (user_id,),
    )
    row = cur.fetchone()
    conn.close()

    if not row:
        return {
            "found": False,
            "user_id": user_id,
            "fav_genres": "",
            "disliked_genres": "",
            "preferred_runtime": "",
        }

    return {
        "found": True,
        "user_id": row["user_id"],
        "fav_genres": row["fav_genres"] or "",
        "disliked_genres": row["disliked_genres"] or "",
        "preferred_runtime": row["preferred_runtime"] or "",
        "last_updated": row["last_updated"],
    }


def log_watch_event(
    user_id: str,
    title: str,
    kind: str,
    genres: Optional[str] = None,
    imdb_rating: Optional[float] = None,
    user_rating: Optional[float] = None,
) -> Dict[str, Any]:
    """Insert a watch event into history."""
    conn = get_conn()
    cur = conn.cursor()

    cur.execute(
        """
        INSERT INTO watch_history (user_id, title, kind, genres, imdb_rating, user_rating, watched_at)
        VALUES (?, ?, ?, ?, ?, ?, ?)
        """,
        (
            user_id,
            title,
            kind,
            genres,
            imdb_rating,
            user_rating,
            datetime.utcnow().isoformat(),
        ),
    )

    conn.commit()
    conn.close()

    return {"status": "ok", "user_id": user_id, "title": title}


def log_recommendation(
    user_id: str,
    title: str,
    kind: str,
    imdb_rating: Optional[float],
    reason: str,
) -> Dict[str, Any]:
    """Store a recommended title & reason."""
    conn = get_conn()
    cur = conn.cursor()

    cur.execute(
        """
        INSERT INTO recommendation_log (user_id, title, kind, imdb_rating, reason, recommended_at)
        VALUES (?, ?, ?, ?, ?, ?)
        """,
        (
            user_id,
            title,
            kind,
            imdb_rating,
            reason,
            datetime.utcnow().isoformat(),
        ),
    )

    conn.commit()
    conn.close()

    return {
        "status": "ok",
        "user_id": user_id,
        "title": title,
        "imdb_rating": imdb_rating,
    }


def get_recent_watch_history(user_id: str, limit: int = 10) -> List[Dict[str, Any]]:
    """Return up to `limit` recent watch events."""
    conn = get_conn()
    cur = conn.cursor()

    cur.execute(
        """
        SELECT title, kind, genres, imdb_rating, user_rating, watched_at
        FROM watch_history
        WHERE user_id = ?
        ORDER BY watched_at DESC
        LIMIT ?
        """,
        (user_id, limit),
    )
    rows = cur.fetchall()
    conn.close()

    result: List[Dict[str, Any]] = []
    for r in rows:
        result.append(
            {
                "title": r["title"],
                "kind": r["kind"],
                "genres": r["genres"],
                "imdb_rating": r["imdb_rating"],
                "user_rating": r["user_rating"],
                "watched_at": r["watched_at"],
            }
        )
    return result


def get_taste_summary(user_id: str) -> Dict[str, Any]:
    """Produce a compact natural language summary of taste."""
    prefs = load_user_preferences(user_id)
    history = get_recent_watch_history(user_id, limit=8)

    lines = []
    lines.append(f"User ID: {user_id}")
    lines.append(f"Favourite genres: {prefs.get('fav_genres') or 'not specified'}")
    lines.append(f"Disliked genres: {prefs.get('disliked_genres') or 'not specified'}")
    lines.append(
        f"Preferred runtime: {prefs.get('preferred_runtime') or 'not specified'}"
    )
    lines.append("")
    lines.append("Recent watch history:")

    if not history:
        lines.append("- None yet")
    else:
        for h in history:
            lines.append(
                f"- {h['title']} ({h['kind']}) | genres={h['genres']} | imdb={h['imdb_rating']} | user_rating={h['user_rating']}"
            )

    return {"status": "ok", "summary": "\n".join(lines)}