Business Logic

loglife.app.logic.text.processor

Message processing logic for inbound WhatsApp text commands.

process_text(user, message)

Route incoming text commands to the appropriate goal or rating handler.

Handle commands such as adding goals, submitting ratings, configuring reminder times, and generating look-back summaries. Maintain temporary state for multi-step flows (e.g., goal reminder setup).

Parameters:

Name	Type	Description	Default
user	User	The user record for the message sender	required
message	Message	The incoming message object	required

Returns:

Type	Description
str | None	The WhatsApp response text to send back to the user.

Source code in src/loglife/app/logic/text/processor.py

def process_text(user: User, message: Message) -> str | None:
    """Route incoming text commands to the appropriate goal or rating handler.

    Handle commands such as adding goals, submitting ratings, configuring
    reminder times, and generating look-back summaries. Maintain temporary
    state for multi-step flows (e.g., goal reminder setup).

    Arguments:
        user: The user record for the message sender
        message: The incoming message object

    Returns:
        The WhatsApp response text to send back to the user.

    """
    try:
        text_content: str = message.raw_payload.strip().lower()

        # Check for blocking states
        if user.state == "awaiting_reminder_time":
            return _handle_reminder_time_state(user, text_content)

        # Add aliases (with word boundaries to avoid replacing partial words)
        text_content = _apply_command_aliases(text_content)

        # Execute the first matching command handler
        for handler in HANDLERS:
            result, matched = _try_handler(handler, user, text_content)
            if matched:
                # Handler matched - if it returned a message, return it
                # If it returned None, it means it already sent the message, so return None
                return result
            # If handler didn't match, continue to next handler

    except Exception as exc:
        logger.exception("Error in text processor")
        return messages.ERROR_TEXT_PROCESSOR.format(exc=exc)

    # No handler matched the message
    return messages.ERROR_WRONG_COMMAND

loglife.app.logic.audio.processor

Audio processing workflow for inbound WhatsApp messages.

Orchestrates transcription (Whisper), summarization (GPT), and database storage of voice notes.

process_audio(user, message)

Process an incoming audio message from a user.

Parameters:

Name	Type	Description	Default
user	User	The user record dictionary	required
message	Message	The incoming message object	required

Returns:

Type	Description
str | tuple[str, str]	The summarized text generated from the audio, or a tuple of
str | tuple[str, str]	(transcript_file_base64, summarized_text).

Source code in src/loglife/app/logic/audio/processor.py

def process_audio(
    user: User,
    message: Message,
) -> str | tuple[str, str]:
    """Process an incoming audio message from a user.

    Arguments:
        user: The user record dictionary
        message: The incoming message object

    Returns:
        The summarized text generated from the audio, or a tuple of
        (transcript_file_base64, summarized_text).

    """
    client_type = message.client_type or "whatsapp"
    sender = message.sender
    audio_data = message.raw_payload

    queue_async_message(sender, "Audio received. Transcribing...", client_type=client_type)

    try:
        try:
            transcript: str = transcribe_audio(audio_data)
        except RuntimeError:
            logger.exception("Error transcribing audio")
            return "Transcription failed!"

        if not transcript.strip():
            return "Transcription was empty."

        queue_async_message(
            sender,
            "Audio transcribed. Summarizing...",
            client_type=client_type,
        )

        try:
            summary: str = summarize_transcript(transcript)
        except RuntimeError:
            logger.exception("Error summarizing transcript")
            return "Summarization failed!"

        db.audio_journals.create(
            user_id=user.id,
            transcription_text=transcript,
            summary_text=summary,
        )
        queue_async_message(
            sender,
            "Summary stored in Database.",
            client_type=client_type,
        )

        if user.send_transcript_file:
            transcript_file: str = transcript_to_base64(transcript)
            return transcript_file, summary

    except Exception as exc:
        logger.exception("Error in audio processor")
        return f"Error in audio processor: {exc}"

    return summary

loglife.app.logic.vcard.processor

Processing logic for referral VCARD payloads.

process_vcard(referrer_user, message)

Create referral users from VCARD attachments.

Parse the incoming VCARD JSON payload, ensure each contact exists as a user, link referrals, and send a welcome message to each referred number.

Parameters:

Name	Type	Description	Default
referrer_user	User	The user dict of the person sharing the VCARDs	required
message	Message	The incoming message object	required

Returns:

Type	Description
str	The referral success message constant.

Source code in src/loglife/app/logic/vcard/processor.py

def process_vcard(referrer_user: User, message: Message) -> str:
    """Create referral users from VCARD attachments.

    Parse the incoming VCARD JSON payload, ensure each contact exists as a
    user, link referrals, and send a welcome message to each referred number.

    Arguments:
        referrer_user: The user dict of the person sharing the VCARDs
        message: The incoming message object

    Returns:
        The referral success message constant.

    """
    try:
        vcards: list[str] = json.loads(message.raw_payload)
        referrer_user_id: int = referrer_user.id

        for vcard in vcards:
            referred_phone_number = _extract_phone_number(vcard)
            if not referred_phone_number:
                continue

            referred_user: User | None = db.users.get_by_phone(referred_phone_number)
            if not referred_user:
                # Create new user with referrer
                db.users.create(
                    referred_phone_number,
                    "Asia/Karachi",
                    referred_by_id=referrer_user_id,
                )
                queue_async_message(referred_phone_number, WELCOME_MESSAGE, client_type="whatsapp")
            elif referred_user.referred_by_id is None:
                # Update existing user if they don't have a referrer
                db.users.update(referred_user.id, referred_by_id=referrer_user_id)

    except Exception as exc:
        logger.exception("Error in vcard processor")
        return f"Error in vcard processor: {exc}"

    return REFERRAL_SUCCESS