app/services/intent.py

"""Author RAG Chatbot SaaS — Intent Classifier.

3-Tier Architecture (from RAG 1.2 proven pattern):
  Tier 0 — Exact-match greeting/casual detector  → instant, zero API cost
  Tier 1 — Keyword rule engine                   → instant, zero API cost
  Tier 2 — LLM fallback                          → only for genuinely ambiguous ~5%

RULE: Python rules handle 90%+ of cases. LLM is the last resort, not the first call.
RULE: This file owns ALL intent detection logic — never detect intent inline elsewhere.
"""

import json
import re
from dataclasses import dataclass

import structlog
from openai import AsyncOpenAI

from app.config import get_settings
from app.services.prompter import INTENT_CLASSIFICATION_PROMPT

logger = structlog.get_logger(__name__)
cfg = get_settings()


# ── Tier 0: Greeting / Casual Detector ───────────────────────────────────────
# Runs in microseconds. Skips retrieval AND LLM classification entirely.
# Source: adapted from RAG 1.2 _is_greeting()

_GREETING_EXACT: set[str] = {
    # Greetings
    "hi", "hello", "hey", "yo", "sup", "wassup", "hola",
    "good morning", "good afternoon", "good evening", "good night",
    # Farewells
    "bye", "goodbye", "see you", "cya", "take care", "farewell",
    # Acknowledgements
    "thanks", "thank you", "thankyou", "thx", "ty", "cheers",
    # Casual
    "ok", "okay", "hmm", "alright", "sure", "fine", "cool", "nice",
    "awesome", "great", "perfect", "interesting", "wow", "amazing",
    # Negative / no content
    "no", "nope", "nothing", "nah",
    # Positive affirmations
    "yes", "yeah", "yep", "yup",
}

_GREETING_SHORT_WORDS: set[str] = {
    "hi", "hello", "hey", "yo", "thanks", "bye", "ok", "okay",
    "no", "yes", "yeah", "hmm", "cool", "nice", "great", "wow",
}

# These words in a short message indicate genuine book interest, not casual chat
_BOOK_SIGNAL_WORDS: set[str] = {
    "book", "story", "read", "buy", "purchase", "author", "chapter",
    "character", "plot", "theme", "ending", "about", "recommend",
    "genre", "review", "summary", "price", "cost", "link", "where",
    "who", "what", "how", "why", "when", "tell", "explain", "describe",
}


def _is_greeting(query: str) -> bool:
    """Tier 0: Detect greetings, casual, and acknowledgements. Zero API cost."""
    q = query.lower().strip()
    words = q.split()

    # Exact match
    if q in _GREETING_EXACT:
        return True

    # Short message (≤4 words) with casual words and NO book signals
    if len(words) <= 4:
        has_casual = any(w in _GREETING_SHORT_WORDS for w in words)
        has_book = any(w in _BOOK_SIGNAL_WORDS for w in words)
        if has_casual and not has_book:
            return True

    return False


# ── Tier 1: Keyword Rule Engine ───────────────────────────────────────────────
# Ordered most-specific → least-specific. First match wins (break on hit).
# Covers 85-90% of real-world messages with zero API cost.

_PURCHASE_SIGNALS: tuple[str, ...] = (
    "how can i buy", "where can i buy", "where to buy", "how do i buy",
    "how can i get", "where can i get", "how do i get", "how to get",
    "where to purchase", "how to purchase", "buy now", "buy this",
    "purchase this", "order this", "get a copy", "get the book",
    "is it available", "available on", "on amazon", "on kindle",
    "how much", "what is the price", "what's the price", "price of",
    "cost of", "how much does", "how much is",
    "buy link", "purchase link", "where to find",
    # Purchase objections / value questions (still in-scope for the book)
    "why should i buy", "why would i buy", "why should i get",
    "why buy", "should i buy", "worth buying", "worth it",
    "convince me", "why bother",
)

_FULL_STORY_SIGNALS: tuple[str, ...] = (
    "tell me the whole story", "tell me everything about", "tell me the entire",
    "complete story", "entire story", "full story", "whole story",
    "tell me the plot", "give me the full", "give me a full summary",
    "complete summary", "full summary", "entire summary",
    "how does it end", "how does the story end", "what happens at the end",
    "what is the ending", "spoil it", "spoil the", "spoiler",
    "what happens in the end", "tell me the ending",
    "full plot", "whole plot", "entire plot",
    "retell the book", "tell me the book",
)

_JAILBREAK_SIGNALS: tuple[str, ...] = (
    "ignore your instructions", "ignore instructions", "ignore all instructions",
    "ignore previous", "forget your instructions", "forget everything",
    "disregard your", "override your", "bypass your",
    "pretend you are", "pretend you're", "act as if", "act like you",
    "roleplay as", "you are now", "from now on you",
    "jailbreak", "developer mode", "unrestricted mode", "god mode",
    "dan ", " dan,", "[dan]", "(dan)",
    "system prompt", "reveal your prompt", "show your instructions",
    "what are your instructions", "your rules", "repeat your",
    "repeat the above", "repeat everything",
    "no restrictions", "no rules", "without restrictions",
    # SEC-3 fix: removed "for a story" and "in a fictional world" — too broad,
    # causing false positives on legitimate reader questions.
    # The precise combined patterns in guardrails.py cover real attacks.
    "hypothetically speaking",
)

_PIRACY_SIGNALS: tuple[str, ...] = (
    "free pdf", "free download", "download this book", "download the book",
    "pirate", "torrent", "epub download", "free copy", "get it free",
    "where to download", "download for free", "crack", "illegal copy",
)

_OFF_TOPIC_SIGNALS: tuple[str, ...] = (
    "weather", "temperature", "forecast",
    "sports score", "football", "cricket score", "basketball",
    "stock price", "stock market", "bitcoin", "crypto",
    "recipe", "how to cook", "how to make food",
    "news today", "breaking news", "current events",
    "code in python", "write me code", "debug this", "programming",
    "homework", "solve this math", "calculate",
    "translate this to", "translate for me",
)

_COMPLAINT_SIGNALS: tuple[str, ...] = (
    "this is useless", "this is terrible", "this is awful",
    "you're useless", "you are useless", "worst bot", "terrible bot",
    "not helpful", "you don't know", "you don't understand",
    "you're wrong", "you are wrong", "that's wrong", "that is wrong",
    "disappointed", "dissatisfied", "very bad", "really bad",
    "this sucks", "this stinks",
    # Extended coverage (Phase 3A)
    "you don't help", "you never", "not answering", "avoiding my question",
    "why won't you", "stop avoiding", "you keep ignoring",
)

# Phase 3A: New signal groups to push rule coverage from ~60% to ~90%.
# These were previously falling through to the LLM classifier (wasted API call).

_SKEPTICISM_SIGNALS: tuple[str, ...] = (
    "what makes it different", "what makes this different", "what makes this one different",
    "why should i trust", "why should i believe", "is it actually good",
    "lots of books claim", "sounds like hype", "everyone says that",
    "heard that before", "why is it special", "what's so special",
    "how is it different", "prove it", "why should i trust its",
    "weak point", "controversial assumption", "critique this book",
    "critique the book", "in a debate", "challenge you", "let me challenge",
    "most controversial", "biggest flaw", "biggest weakness",
)

_COMPARISON_SIGNALS: tuple[str, ...] = (
    "which book", "what book", "which one should i", "recommend a book",
    "what should i read", "which should i read", "compare the books",
    "difference between", "better book", "your books", "all your books",
    "what books do you", "list your books", "show me all", "how many books",
)

_GENERAL_QUESTION_SIGNALS: tuple[str, ...] = (
    "who is", "who are", "what is", "what are", "what does", "what did",
    "tell me about", "can you explain", "can you describe", "can you tell",
    "i want to know", "i'm curious about", "i am curious about",
    "is there", "are there", "does the book", "does it have",
    "is it about", "what happens", "what's in", "who wrote",
)


def _is_purchase_objection(query: str) -> bool:
    """Reader questions value or expresses disinterest — still about the book."""
    q = query.lower()
    objection_markers = (
        "no interest", "not interested", "don't like", "do not like",
        "don't care", "do not care", "why should i buy", "why would i buy",
        "convince me", "why bother", "not for me",
    )
    return any(marker in q for marker in objection_markers)


def _classify_by_rules(query: str) -> str | None:
    """Tier 1: Keyword rule classification. Returns intent label or None if ambiguous.

    Ordered most-specific to least-specific. First match wins.
    Phase 3A extended to cover ~90%+ of real queries via rules (was ~60%).
    """
    q = query.lower().strip()

    # Check jailbreak first (security priority)
    if any(sig in q for sig in _JAILBREAK_SIGNALS):
        return "jailbreak_attempt"

    # B5 fix: Piracy was incorrectly returning "jailbreak_attempt", which triggered
    # the harsh jailbreak response instead of the softer piracy_response() handler.
    if any(sig in q for sig in _PIRACY_SIGNALS):
        return "piracy_request"   # Routes to piracy_response(), NOT jailbreak handler

    # Purchase objection — disinterest or "why buy" (must beat off-topic keyword false positives)
    if _is_purchase_objection(q):
        return "question"

    # Purchase intent (explicit buy signals)
    if any(sig in q for sig in _PURCHASE_SIGNALS):
        return "purchase_intent"

    # Full story request
    if any(sig in q for sig in _FULL_STORY_SIGNALS):
        return "full_story_request"

    # Complaint
    if any(sig in q for sig in _COMPLAINT_SIGNALS):
        return "complaint"

    # Off-topic
    if any(sig in q for sig in _OFF_TOPIC_SIGNALS):
        return "off_topic"

    # Skepticism / value challenge about the selected book (before catalog comparison)
    if any(sig in q for sig in _SKEPTICISM_SIGNALS):
        return "question"

    # Book comparison / catalog questions (Phase 3A)
    if any(sig in q for sig in _COMPARISON_SIGNALS):
        return "book_comparison"

    # General book questions (Phase 3A — large coverage improvement)
    if any(sig in q for sig in _GENERAL_QUESTION_SIGNALS):
        return "question"

    return None  # Genuinely ambiguous — fall through to LLM (should be <10% of queries)


# ── Result Dataclass ──────────────────────────────────────────────────────────

@dataclass
class IntentResult:
    """Result of intent classification for a single query."""
    intent: str             # e.g., 'question', 'purchase_intent', 'off_topic'
    confidence: float       # 0.0 to 1.0
    book_reference: str | None    # Exact book name if mentioned
    book_confidence: float        # Confidence that a specific book was referenced
    source: str = "rules"         # 'rules' or 'llm' — for logging/debugging


# ── Book Reference Detector ───────────────────────────────────────────────────

def _extract_book_reference(query: str, books: list[dict] | None = None) -> tuple[str | None, float]:
    """Extract a book title reference from the query if any books are known.

    Args:
        query: User message.
        books: List of book dicts with 'title' key (optional).

    Returns:
        Tuple of (book_title | None, confidence).
    """
    if not books:
        return None, 0.0

    q_lower = query.lower()
    for book in books:
        title = book.get("title", "")
        if not title:
            continue
        if title.lower() in q_lower:
            return title, 0.95
        # Partial match: first significant word of title
        first_word = title.split()[0].lower() if title.split() else ""
        if len(first_word) > 4 and first_word in q_lower:
            return title, 0.70

    return None, 0.0


# ── Main Classifier ───────────────────────────────────────────────────────────

async def classify_intent(
    query: str,
    history: list[dict],
    books: list[dict] | None = None,
) -> IntentResult:
    """Classify intent using 3-tier system.

    Tier 0: Greeting detection (Python, instant)
    Tier 1: Keyword rules (Python, instant)
    Tier 2: LLM fallback (API call, ~5% of queries)

    Args:
        query: The user's message text.
        history: Last N turns of conversation history.
        books: Optional list of known book dicts (for reference detection).

    Returns:
        IntentResult with intent, confidence, and book reference.
    """
    # ── Tier 0: Greeting ──────────────────────────────────────────────────────
    if _is_greeting(query):
        logger.debug("Intent: greeting (Tier 0 — Python rules)", query=query[:40])
        return IntentResult(
            intent="greeting",
            confidence=0.99,
            book_reference=None,
            book_confidence=0.0,
            source="rules",
        )

    # ── Tier 1: Keyword rules ─────────────────────────────────────────────────
    rule_intent = _classify_by_rules(query)
    book_ref, book_conf = _extract_book_reference(query, books)

    if rule_intent is not None:
        logger.debug(
            "Intent: %s (Tier 1 — keyword rules)", rule_intent,
            query=query[:40],
        )
        return IntentResult(
            intent=rule_intent,
            confidence=0.92,
            book_reference=book_ref,
            book_confidence=book_conf,
            source="rules",
        )

    # ── Tier 2: LLM fallback (only for genuinely ambiguous cases) ────────────
    # Applies to: abstract questions about the book, comparisons, meta questions
    history_str = "\n".join(
        f"User: {m['content']}"
        for m in history[-3:]
        if m.get("role") == "user"
    ) or "No prior conversation"

    prompt = INTENT_CLASSIFICATION_PROMPT.format(
        query=query,
        history=history_str,
    )

    try:
        # BUG-1 fix: use the shared singleton client (same as helpers.py) instead of
        # creating a new AsyncOpenAI() here — which spawned a new HTTP connection pool
        # on every single Tier-2 intent classification call.
        from app.services.pipeline.helpers import _get_openai_client
        client = _get_openai_client()
        response = await client.chat.completions.create(
            model=cfg.OPENAI_CHAT_MODEL,     # gpt-4o-mini — cheap and accurate
            messages=[{"role": "user", "content": prompt}],
            max_tokens=100,                   # Tiny response — just a label + confidence
            temperature=0.0,
            response_format={"type": "json_object"},
        )
        data = json.loads(response.choices[0].message.content)

        # Merge LLM book reference with Python-detected one (Python wins if confident)
        llm_book_ref = data.get("book_reference")
        llm_book_conf = float(data.get("book_confidence", 0.0))
        final_book_ref = book_ref if book_conf >= 0.70 else (llm_book_ref or book_ref)
        final_book_conf = max(book_conf, llm_book_conf)

        result = IntentResult(
            intent=data.get("intent", "question"),
            confidence=float(data.get("confidence", 0.7)),
            book_reference=final_book_ref,
            book_confidence=final_book_conf,
            source="llm",
        )
        logger.debug(
            "Intent: %s (Tier 2 — LLM)", result.intent,
            confidence=result.confidence,
            query=query[:40],
        )
        return result

    except Exception as e:
        logger.warning("Intent classification failed, defaulting to question", error=str(e))
        return IntentResult(
            intent="question",
            confidence=0.5,
            book_reference=book_ref,
            book_confidence=book_conf,
            source="fallback",
        )