FabledScribe/src/fabledassistant/services/user_profile.py

"""User profile service — structured per-user preferences for LLM context."""
import asyncio
import logging
from datetime import date, datetime, timedelta, timezone

from sqlalchemy import select

from fabledassistant.models import async_session
from fabledassistant.models.user_profile import UserProfile

logger = logging.getLogger(__name__)

VALID_EXPERTISE = {"novice", "intermediate", "expert"}
VALID_STYLES = {"concise", "balanced", "detailed"}
VALID_TONES = {"casual", "professional", "technical"}

# Trigger consolidation when raw observations reach this count
_CONSOLIDATION_THRESHOLD = 14


async def get_profile(user_id: int) -> UserProfile:
    """Get or create the profile row for a user."""
    async with async_session() as session:
        result = await session.execute(
            select(UserProfile).where(UserProfile.user_id == user_id)
        )
        profile = result.scalar_one_or_none()
        if profile is None:
            profile = UserProfile(user_id=user_id)
            session.add(profile)
            await session.commit()
            await session.refresh(profile)
        return profile


async def update_profile(user_id: int, data: dict) -> UserProfile:
    """Upsert structured profile fields from a validated dict."""
    allowed = {
        "display_name", "job_title", "industry", "expertise_level",
        "response_style", "tone", "interests", "work_schedule",
    }
    async with async_session() as session:
        result = await session.execute(
            select(UserProfile).where(UserProfile.user_id == user_id)
        )
        profile = result.scalar_one_or_none()
        if profile is None:
            profile = UserProfile(user_id=user_id)
            session.add(profile)
        for key, value in data.items():
            if key in allowed:
                setattr(profile, key, value)
        await session.commit()
        await session.refresh(profile)
    return profile


async def append_observations(user_id: int, bullets: str) -> None:
    """
    Append a new dated observation entry from the day's briefing closeout.
    Automatically triggers consolidation when the raw list grows large.
    """
    if not bullets.strip():
        return

    async with async_session() as session:
        result = await session.execute(
            select(UserProfile).where(UserProfile.user_id == user_id)
        )
        profile = result.scalar_one_or_none()
        if profile is None:
            profile = UserProfile(user_id=user_id)
            session.add(profile)

        existing: list = list(profile.observations_raw or [])
        existing.append({
            "date": date.today().isoformat(),
            "bullets": bullets.strip(),
        })
        # Keep at most 60 raw entries as a rolling window
        profile.observations_raw = existing[-60:]
        profile.observations_updated_at = datetime.now(timezone.utc)
        await session.commit()
        raw_count = len(profile.observations_raw or [])

    logger.info("Appended observations for user %d (%d raw entries)", user_id, raw_count)

    if raw_count >= _CONSOLIDATION_THRESHOLD:
        asyncio.create_task(_consolidate_observations(user_id))


async def consolidate_observations(user_id: int) -> str:
    """Public entry point to manually trigger observation consolidation."""
    return await _consolidate_observations(user_id)


async def clear_learned_data(user_id: int) -> None:
    """Reset all learned observations and summary for a user."""
    async with async_session() as session:
        result = await session.execute(
            select(UserProfile).where(UserProfile.user_id == user_id)
        )
        profile = result.scalar_one_or_none()
        if profile:
            profile.learned_summary = None
            profile.observations_raw = []
            profile.observations_updated_at = None
            await session.commit()


async def _consolidate_observations(user_id: int) -> str:
    """
    LLM pass: synthesise all raw observation bullets into an updated
    learned_summary paragraph. Prunes raw entries older than 30 days afterwards.
    """
    from fabledassistant.config import Config
    from fabledassistant.services.llm import generate_completion
    from fabledassistant.services.settings import get_setting

    async with async_session() as session:
        result = await session.execute(
            select(UserProfile).where(UserProfile.user_id == user_id)
        )
        profile = result.scalar_one_or_none()
        if not profile or not profile.observations_raw:
            return ""
        observations = list(profile.observations_raw)
        existing_summary = profile.learned_summary or ""

    obs_text = "\n\n".join(
        f"[{entry['date']}]\n{entry['bullets']}"
        for entry in observations
    )

    system = (
        "You are synthesising preference observations into a concise user profile summary. "
        "Consolidate the observations into 3-6 factual sentences describing the user's patterns, "
        "preferences, and habits. Be specific and useful for a personal assistant. "
        "Merge with any existing summary, removing duplicates and outdated information. "
        "Output only the consolidated summary paragraph — no preamble, no bullet points."
    )
    user_prompt = ""
    if existing_summary:
        user_prompt += f"Existing summary:\n{existing_summary}\n\n"
    user_prompt += f"New observations:\n{obs_text}"

    # Profile observation consolidation reasons over multiple documents to
    # produce a coherent summary — closer to curator-shaped work than chat.
    # Route to worker (background_model). Falls back to OLLAMA_MODEL only if
    # neither setting nor BACKGROUND default is available.
    model = (
        await get_setting(user_id, "background_model", "")
        or Config.OLLAMA_BACKGROUND_MODEL
        or Config.OLLAMA_MODEL
    )
    try:
        new_summary = (await generate_completion(
            [
                {"role": "system", "content": system},
                {"role": "user", "content": user_prompt},
            ],
            model,
        )).strip()
    except Exception:
        logger.warning("Observation consolidation failed for user %d", user_id, exc_info=True)
        new_summary = ""

    if new_summary:
        cutoff = (date.today() - timedelta(days=30)).isoformat()
        async with async_session() as session:
            result = await session.execute(
                select(UserProfile).where(UserProfile.user_id == user_id)
            )
            profile = result.scalar_one_or_none()
            if profile:
                profile.learned_summary = new_summary
                profile.observations_raw = [
                    o for o in (profile.observations_raw or [])
                    if o.get("date", "") >= cutoff
                ]
                await session.commit()
        logger.info("Consolidated observations for user %d", user_id)

    return new_summary


async def build_profile_context(user_id: int) -> str:
    """
    Build a formatted context string from the user's structured profile
    for injection into LLM system prompts (briefing and chat).
    Returns an empty string if no meaningful data is set.
    """
    profile = await get_profile(user_id)

    parts: list[str] = []

    if profile.display_name:
        parts.append(f"User's name: {profile.display_name}")
    if profile.job_title or profile.industry:
        job = " in ".join(filter(None, [profile.job_title, profile.industry]))
        parts.append(f"Occupation: {job}")
    if profile.expertise_level and profile.expertise_level != "intermediate":
        parts.append(
            f"Expertise level: {profile.expertise_level} — calibrate explanation depth accordingly"
        )
    if profile.response_style or profile.tone:
        style = profile.response_style or "balanced"
        tone = profile.tone or "casual"
        parts.append(f"Preferred response style: {style}, tone: {tone}")
    if profile.interests:
        parts.append(f"Interests: {', '.join(profile.interests)}")
    if profile.work_schedule:
        sched = profile.work_schedule
        days = ", ".join(sched.get("days") or []) or "weekdays"
        start = sched.get("start", "9:00")
        end = sched.get("end", "17:00")
        parts.append(f"Work schedule: {days}, {start}–{end}")
    if profile.learned_summary:
        parts.append(f"What the assistant has learned about this user: {profile.learned_summary}")

    return "\n".join(parts)