app/services/session_core/context.py

"""Author RAG Chatbot SaaS — User Interest Context Profiler.

Tracks per-session interest signals to drive the upsell engine.
Tags are accumulated across turns and used by the upsell engine to
select the most relevant purchase hook.

RULE: No PII ever stored — only topic tags extracted from message content.
RULE: Tags are stored in Redis alongside session history (see session/manager.py).
"""

import re
import structlog

logger = structlog.get_logger(__name__)

# Topic → tag mapping (simple keyword extraction)
_TAG_PATTERNS: dict[str, list[str]] = {
    "pricing": ["price", "cost", "how much", "buy", "purchase", "discount", "deal"],
    "preview": ["preview", "sample", "excerpt", "read first", "try before"],
    "genre": ["thriller", "romance", "fantasy", "sci-fi", "mystery", "horror", "biography"],
    "characters": ["character", "protagonist", "villain", "hero", "who is"],
    "plot": ["plot", "story", "ending", "spoiler", "what happens"],
    "author": ["author", "writer", "written by", "biography", "about you"],
    "series": ["series", "sequel", "next book", "trilogy", "part 2"],
    "comparison": ["compare", "difference", "better", "recommend", "which book"],
}

_TAG_WEIGHTS: dict[str, float] = {
    "pricing": 0.35,
    "preview": 0.25,
    "plot": 0.20,
    "characters": 0.15,
    "comparison": 0.20,
    "series": 0.25,
    "genre": 0.10,
    "author": 0.10,
}

_INTENT_BOOST: dict[str, float] = {
    "purchase_intent": 0.40,
    "full_story_request": 0.30,
    "book_comparison": 0.15,
    "comparison": 0.15,
}

_MAX_TURN_COMPONENT = 0.25
_MAX_TAG_COMPONENT = 0.75

_COMPILED_TAGS: dict[str, list[re.Pattern]] = {
    tag: [re.compile(kw, re.IGNORECASE) for kw in keywords]
    for tag, keywords in _TAG_PATTERNS.items()
}


def effective_interest_score(
    turn_count: int,
    tags: list[str],
    current_intent: str | None = None,
) -> float:
    """Compute interest score including optional boost from the current turn's intent."""
    return compute_interest_score(turn_count, tags, current_intent=current_intent)


def extract_interest_tags(message: str) -> list[str]:
    """Extract interest tags from a user message.

    Args:
        message: Raw user message text.

    Returns:
        List of interest tag strings matched from the message.
    """
    detected: list[str] = []
    for tag, patterns in _COMPILED_TAGS.items():
        for pattern in patterns:
            if pattern.search(message):
                detected.append(tag)
                break  # One match per tag is enough
    if detected:
        logger.debug("Interest tags extracted", tags=detected)
    return detected


def compute_interest_score(
    turn_count: int,
    tags: list[str],
    current_intent: str | None = None,
) -> float:
    """Compute a 0.0–1.0 interest score from session signals.

    Higher score = more engaged visitor = more assertive (but human) upsell framing.
    Buying-signal tags and high-intent messages weigh more than raw turn count.

    Args:
        turn_count: Number of completed conversation turns.
        tags: Accumulated interest tags for the session.
        current_intent: Optional intent for this turn (adds a one-shot boost).

    Returns:
        Float between 0.0 (cold) and 1.0 (highly engaged).
    """
    turn_score = min(turn_count / 10.0, 1.0) * _MAX_TURN_COMPONENT

    unique_tags = set(tags)
    tag_score = sum(_TAG_WEIGHTS.get(tag, 0.05) for tag in unique_tags)
    tag_score = min(tag_score, _MAX_TAG_COMPONENT)

    score = turn_score + tag_score
    if current_intent:
        score += _INTENT_BOOST.get(current_intent, 0.0)

    score = round(min(score, 1.0), 3)
    logger.debug(
        "Interest score computed",
        turns=turn_count,
        tags=len(unique_tags),
        intent=current_intent,
        score=score,
    )
    return score