external_integrations.py

import json
import re
import uuid
from datetime import datetime, time, timezone
from typing import Any, Dict, List, Optional, cast
import pytz
from langchain_core.prompts import ChatPromptTemplate
from pydantic import ValidationError
import database.action_items as action_items_db
import database.daily_summaries as daily_summaries_db
import database.memories as memories_db
import database.users as users_db
from models.conversation import Conversation
from models.daily_summary_payload import DailySummaryDayStatsPayload, DailySummaryPayload
from models.structured import Structured
from models.structured_extraction import StructuredExtraction
from models.other import Person
from utils.conversations.location import get_google_maps_location
from utils.conversations.render import conversations_to_string
from utils.llm.clients import get_llm, parser
from utils.llm.usage_tracker import track_usage, Features
from utils.llms.memory import get_prompt_memories
from utils.log_sanitizer import sanitize, sanitize_validation_error
import logging

logger = logging.getLogger(__name__)

# Read-time address fills per generated summary (see the pins loop below):
# bounds geocode attempts — and therefore worst-case wall-clock — per summary
# generation; pins past the cap keep an empty address ("Unknown" in the app).
_DAILY_SUMMARY_GEOCODE_ATTEMPT_CAP = 10


def _content_str(response: Any) -> str:
    content = response.content
    return content if isinstance(content, str) else str(content)


def _coerce_structured(response: Any) -> Structured:
    if isinstance(response, StructuredExtraction):
        return response.to_structured()
    return response


def _basic_daily_summary(
    date_str: str,
    total_conversations: int,
    total_duration_minutes: float,
    actual_action_items: List[Dict[str, Any]],
    locations: List[Dict[str, Any]],
    stats: DailySummaryDayStatsPayload,
    memories_learned: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    return {
        "id": str(uuid.uuid4()),
        "date": date_str,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "headline": "Your Day in Review",
        "overview": f"You had {total_conversations} conversations today.",
        "day_emoji": "📅",
        "stats": stats.model_dump(),
        "highlights": [],
        "action_items": actual_action_items,
        "unresolved_questions": [],
        "decisions_made": [],
        "knowledge_nuggets": [],
        "memories_learned": list(memories_learned or []),
        "locations": locations,
    }


def get_message_structure(
    text: str,
    started_at: datetime,
    language_code: str,
    tz: str,
    text_source_spec: Optional[str] = None,
    output_language_code: Optional[str] = None,
) -> Structured:
    response_language = output_language_code or language_code
    prompt_text = '''
    You are an expert message analyzer. Your task is to analyze the message content and provide structure and clarity.
    The message language is {language_code}. You MUST respond entirely in {response_language}.

    For the title, create a concise title that captures the main topic of the message.
    For the overview, summarize the message with the main points discussed, make sure to capture the key information and important details.
    For the action items, include any tasks or actions that need to be taken based on the message.
    For the category, classify the message into one of the available categories.
    For Calendar Events, include any events or meetings mentioned in the message. For date context, this message was sent on {started_at}. {tz} is the user's timezone, respond in user local timezone.

    Message Content: ```{text}```
    Message Source: {text_source_spec}

    {format_instructions}'''.replace('    ', '').strip()

    prompt = cast(Any, ChatPromptTemplate).from_messages([('system', prompt_text)])
    chain = prompt | get_llm('external_structure') | parser

    response = _coerce_structured(
        chain.invoke(
            {
                'language_code': language_code,
                'response_language': response_language,
                'started_at': started_at.isoformat(),
                'tz': tz,
                'text': text,
                'text_source_spec': text_source_spec if text_source_spec else 'Messaging App',
                'format_instructions': parser.get_format_instructions(),
            }
        )
    )

    for event in response.events or []:
        if event.duration > 180:
            event.duration = 180
        event.created = False

    # Set created_at for action items if not already set
    for action_item in response.action_items or []:
        if action_item.created_at is None:
            action_item.created_at = datetime.now(timezone.utc)

    return response


def summarize_experience_text(
    text: str, text_source_spec: Optional[str] = None, tz: Optional[str] = None
) -> Structured:
    source_context = f"Source: {text_source_spec}" if text_source_spec else "their own experiences or thoughts"
    tz = tz or 'UTC'
    try:
        current_date = datetime.now(pytz.timezone(tz)).strftime('%Y-%m-%d')
    except Exception:  # unknown/invalid timezone -> anchor to UTC
        tz = 'UTC'
        current_date = datetime.now(timezone.utc).strftime('%Y-%m-%d')
    prompt = f'''The user sent a text of {source_context}, and wants to create a memory from it.
      For the title, use the main topic of the experience or thought.
      For the overview, condense the descriptions into a brief summary with the main topics discussed, make sure to capture the key points and important details.
      For the category, classify the scenes into one of the available categories.
      For the action items, include any tasks or actions that need to be taken based on the content.
      For Calendar Events, include any events or meetings mentioned in the content. For date context, today is {current_date} in the user's timezone ({tz}); resolve any relative dates like "tomorrow" or "next week" against it.

      Text: ```{text}```
      '''.replace('    ', '').strip()

    response = _coerce_structured(
        get_llm('external_structure').with_structured_output(StructuredExtraction).invoke(prompt)
    )

    # Set created_at for action items if not already set
    for action_item in response.action_items or []:
        if action_item.created_at is None:
            action_item.created_at = datetime.now(timezone.utc)

    return response


def get_conversation_summary(uid: str, memories: List[Conversation]) -> str:
    user_name, memories_str = get_prompt_memories(uid)
    user_language = users_db.get_user_language_preference(uid)

    all_person_ids: List[str] = []
    for m in memories:
        all_person_ids.extend(m.get_person_ids())

    people: List[Person] = []
    if all_person_ids:
        people_data = users_db.get_people_by_ids(uid, list(set(all_person_ids)))
        people = [Person(**p) for p in people_data]

    conversation_history = conversations_to_string(memories, people=people)

    language_instruction = ''
    if user_language and user_language != 'en':
        language_instruction = f'You MUST respond entirely in {user_language}. Do NOT respond in English.'

    prompt = f"""
    You are an experienced mentor, that helps people achieve their goals and improve their lives.
    You are advising {user_name} right now, {memories_str}

    The following are a list of {user_name}'s conversations from today, with the transcripts and a slight summary of each, that {user_name} had during his day.
    {user_name} wants to get a summary of the key action items {user_name} has to take based on today's conversations.

    Remember {user_name} is busy so this has to be very efficient and concise.
    Respond in at most 50 words.
    {language_instruction}

    Output your response in plain text, without markdown. No newline character and only use numbers for the action items.
    ```
    ${conversation_history}
    ```
    """.replace('    ', '').strip()
    # print(prompt)
    with track_usage(uid, Features.DAILY_SUMMARY):
        return _content_str(get_llm('daily_summary_simple').invoke(prompt))


def generate_comprehensive_daily_summary(
    uid: str,
    conversations: List[Conversation],
    date_str: str,
    start_date_utc: Optional[datetime] = None,
    end_date_utc: Optional[datetime] = None,
    memories_learned: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    """
    Generate a comprehensive daily summary with structured data for storage.

    ``memories_learned`` is the already-selected review contract from
    ``utils.memory.learned_today``. It is passed in rather than read here: this
    module is the LLM summary builder, and making it reach into the memory stack
    would put a Firestore read behind every caller and every test of it.

    Returns a dictionary matching the DailySummary model structure.
    """
    learned_refs: List[Dict[str, Any]] = list(memories_learned or [])
    # Get user's timezone
    user_profile = users_db.get_user_profile(uid)
    user_tz_str = user_profile.get('time_zone', 'UTC')
    try:
        user_tz = pytz.timezone(user_tz_str)
    except Exception:
        user_tz = pytz.UTC

    user_name, memories_str = get_prompt_memories(uid)

    # Get user's language preference for generating summary in their language
    output_language = user_profile.get('language', '') or 'en'

    all_person_ids: List[str] = []
    for m in conversations:
        all_person_ids.extend(m.get_person_ids())

    people: List[Person] = []
    if all_person_ids:
        people_data = users_db.get_people_by_ids(uid, list(set(all_person_ids)))
        people = [Person(**p) for p in people_data]

    conversation_history = conversations_to_string(conversations, people=people)

    # Calculate stats - exclude discarded conversations
    non_discarded = [c for c in conversations if not c.discarded]
    total_conversations = len(non_discarded)
    total_duration_minutes = sum(
        (c.finished_at - c.started_at).total_seconds() / 60 for c in non_discarded if c.finished_at and c.started_at
    )

    stats_start_date_utc = start_date_utc
    stats_end_date_utc = end_date_utc
    if stats_start_date_utc is None or stats_end_date_utc is None:
        target_date = datetime.strptime(date_str, '%Y-%m-%d').date()
        start_of_day = user_tz.localize(datetime.combine(target_date, time.min))
        end_of_day = user_tz.localize(datetime.combine(target_date, time.max))
        stats_start_date_utc = stats_start_date_utc or start_of_day.astimezone(pytz.UTC)
        stats_end_date_utc = stats_end_date_utc or end_of_day.astimezone(pytz.UTC)
    assert stats_start_date_utc is not None and stats_end_date_utc is not None

    memories_created = memories_db.count_memories_created(uid, stats_start_date_utc, stats_end_date_utc)
    action_items_created = len(
        action_items_db.get_action_items(
            uid,
            start_date=stats_start_date_utc,
            end_date=stats_end_date_utc,
        )
    )
    desktop_usage = daily_summaries_db.get_desktop_daily_usage(uid, date_str)
    watching_minutes = int(round(desktop_usage.get('watching_seconds', 0) / 60))
    proactive_moments = desktop_usage.get('proactive_cards_shown', 0)

    # Extract ALL locations from non-discarded conversations.
    # latitude/longitude are required floats on the Geolocation model, so guarding on
    # their truthiness wrongly drops a valid coordinate of exactly 0.0 (for example
    # longitude 0.0 on the prime meridian). Guard on the geolocation's presence instead.
    locations: List[Dict[str, Any]] = []
    geocode_attempts = 0
    for c in non_discarded:
        if c.geolocation:
            address = c.geolocation.address
            if not address and geocode_attempts < _DAILY_SUMMARY_GEOCODE_ATTEMPT_CAP:
                # Read-time fill for conversations created before write-time
                # enrichment existed (notably the sync path): look the address up
                # through the shared ~100m-rounded geocode cache, so a day whose
                # conversations were already enriched costs no extra upstream call.
                # A geocode miss or error leaves the address empty — the pin stays
                # and the app labels it "Unknown"; a pin is never dropped.
                # The attempt cap bounds wall-clock: cache hits are cheap, but
                # attempts (hits and misses alike) are the deterministic bound —
                # 10 attempts x the geocoder's 10s worst case stays far inside
                # the summary job's budget. Pins past the cap keep an empty
                # address and fall back to the app's "Unknown" label.
                geocode_attempts += 1
                try:
                    geocoded = get_google_maps_location(c.geolocation.latitude, c.geolocation.longitude)
                except Exception as error:
                    logger.warning('daily summary address geocode failed error_type=%s', type(error).__name__)
                    geocoded = None
                if geocoded is not None:
                    address = geocoded.address
            # Convert UTC time to user's local timezone
            local_time = None
            if c.started_at:
                utc_time = c.started_at
                if utc_time.tzinfo is None:
                    utc_time = pytz.UTC.localize(utc_time)
                local_time = utc_time.astimezone(user_tz).strftime("%H:%M")
            locations.append(
                {
                    "latitude": c.geolocation.latitude,
                    "longitude": c.geolocation.longitude,
                    "address": address,
                    "conversation_id": c.id,
                    "time": local_time,
                }
            )

    # Fetch action items for the specific conversations being summarised.
    # Querying by conversation_id (not date range) prevents pulling in items whose
    # async processing happened to land on the same UTC day as an unrelated conversation.
    actual_action_items: List[Dict[str, Any]] = []
    for c in non_discarded:
        for item in action_items_db.get_action_items(uid, conversation_id=c.id):
            actual_action_items.append(
                {
                    "description": item.get("description", ""),
                    "priority": "high" if item.get("completed") == False else "medium",
                    "completed": item.get("completed", False),
                    "source_conversation_id": item.get("conversation_id"),
                }
            )

    stats = DailySummaryDayStatsPayload(
        total_conversations=total_conversations,
        total_duration_minutes=int(total_duration_minutes),
        action_items_count=len(actual_action_items),
        memories_created=memories_created,
        action_items_created=action_items_created,
        watching_minutes=watching_minutes,
        proactive_moments=proactive_moments,
    )

    # Build conversation ID mapping for the LLM
    convo_id_map = {i + 1: c.id for i, c in enumerate(non_discarded)}

    prompt = f"""You are creating a daily summary for {user_name}. {memories_str}
OUTPUT LANGUAGE: {output_language}. You MUST write every word of this summary in {output_language}, regardless of the language the conversations are in.

Today's date: {date_str}
Conversations: {total_conversations}
Daily stats: {memories_created} memories created, {action_items_created} action items created, {watching_minutes} minutes watched, {proactive_moments} proactive moments.

Here are {user_name}'s conversations from today (numbered 1-{total_conversations}):
```
{conversation_history}
```

Generate a JSON response. ONLY include sections with genuinely useful content - skip sections entirely if data is thin or low quality.

{{
    "headline": "Catchy one-liner (max 8 words)",
    "overview": "2-3 snappy lines. Crisp, insightful, no fluff.",
    "day_emoji": "Single emoji",
    "highlights": [
        {{
            "topic": "Short topic name",
            "emoji": "🎯",
            "summary": "One crisp sentence.",
            "conversation_numbers": [1, 2]
        }}
    ],
    "unresolved_questions": [
        {{
            "question": "Short question that wasn't answered",
            "conversation_number": 1
        }}
    ],
    "decisions_made": [
        {{
            "decision": "Short decision or conclusion",
            "conversation_number": 1
        }}
    ],
    "knowledge_nuggets": [
        {{
            "insight": "Short interesting fact or tip learned",
            "conversation_number": 1
        }}
    ]
}}

RULES:
- highlights: Max 4. One sentence each.
- unresolved_questions: Max 3. Short, punchy questions only. Keep each question short and snappy, less than 15 words.
- decisions_made: Max 3. Concrete decisions only. Only add here if it is something that the user has decided on. Tasks or action items don't belong here. Keep each decision short and snappy, less than 15 words.
- knowledge_nuggets: Max 3. Genuinely interesting learnings. Learnings are new learnings for the user, not something they might have already known. Shouldn't be very generic, should be a very specific learning. Keep each learning short and snappy, less than 15 words.
- conversation_number: Reference which conversation (1-{total_conversations}) it came from.
- SKIP sections entirely if no quality content.
- Be snappy. No fluff. No corporate speak. Only include sections that are genuinely useful and relevant.
- OUTPUT LANGUAGE: Every word — headline, overview, highlights, questions, decisions, knowledge nuggets — MUST be in {output_language}. Do not use any other language.

Respond with ONLY valid JSON. Do not include any other text or comments."""

    try:
        with track_usage(uid, Features.DAILY_SUMMARY):
            response = _content_str(get_llm('daily_summary', cache_key='omi-daily-summary').invoke(prompt))
        # Clean up response - remove markdown if present
        response = response.strip()
        if response.startswith('```'):
            response = response.split('```')[1]
            if response.startswith('json'):
                response = response[4:]
        response = response.strip()

        # Try to repair common JSON issues from LLM
        response = re.sub(r':\s*\\"([^"]*)\\"', r': "\1"', response)
        response = response.replace('\\"', '"')

        summary_data = DailySummaryPayload.model_validate(json.loads(response))

        # Helper to map conversation number to ID
        def get_convo_id(num: Any):
            if num and isinstance(num, int) and num in convo_id_map:
                return convo_id_map[num]
            return None

        # Process highlights - map conversation_numbers to conversation_ids
        highlights: List[Dict[str, Any]] = []
        for h in summary_data.highlights:
            convo_nums = h.conversation_numbers
            convo_ids = [get_convo_id(n) for n in convo_nums if get_convo_id(n)]
            highlights.append(
                {
                    "topic": h.topic,
                    "emoji": h.emoji or "💡",
                    "summary": h.summary,
                    "conversation_ids": convo_ids,
                }
            )

        # Process unresolved questions
        unresolved_questions: List[Dict[str, Any]] = []
        for q in summary_data.unresolved_questions:
            unresolved_questions.append(
                {"question": q.question, "conversation_id": get_convo_id(q.conversation_number)}
            )

        # Process decisions made
        decisions_made: List[Dict[str, Any]] = []
        for d in summary_data.decisions_made:
            decisions_made.append({"decision": d.decision, "conversation_id": get_convo_id(d.conversation_number)})

        # Process knowledge nuggets
        knowledge_nuggets: List[Dict[str, Any]] = []
        for k in summary_data.knowledge_nuggets:
            knowledge_nuggets.append({"insight": k.insight, "conversation_id": get_convo_id(k.conversation_number)})

        # Build the complete summary object
        summary_id = str(uuid.uuid4())
        return {
            "id": summary_id,
            "date": date_str,
            "created_at": datetime.now(timezone.utc).isoformat(),
            "headline": summary_data.headline,
            "overview": summary_data.overview,
            "day_emoji": summary_data.day_emoji,
            "stats": stats.model_dump(),
            "highlights": highlights,
            "action_items": actual_action_items,
            "unresolved_questions": unresolved_questions,
            "decisions_made": decisions_made,
            "knowledge_nuggets": knowledge_nuggets,
            "memories_learned": learned_refs,
            "locations": locations,
        }
    except json.JSONDecodeError as e:
        logger.error("Failed to decode daily summary payload JSON: %s", sanitize(str(e)))
        return _basic_daily_summary(
            date_str,
            total_conversations,
            total_duration_minutes,
            actual_action_items,
            locations,
            stats,
            memories_learned=learned_refs,
        )
    except ValidationError as e:
        logger.error("Failed to validate daily summary payload: %s", sanitize_validation_error(cast(Any, e)))
        return _basic_daily_summary(
            date_str,
            total_conversations,
            total_duration_minutes,
            actual_action_items,
            locations,
            stats,
            memories_learned=learned_refs,
        )