diff --git a/Dockerfile b/Dockerfile index 7ebffe6..2caf9d6 100644 --- a/Dockerfile +++ b/Dockerfile @@ -8,41 +8,34 @@ ENV PYTHONUNBUFFERED=1 \ WORKDIR /app -# Install essential system +# Install essential system dependencies (minimal Debian packages) RUN apt-get update && apt-get install -y --no-install-recommends \ ca-certificates \ curl \ - libmagic1 \ libjpeg62-turbo \ - git \ && apt-get clean \ - && rm -rf /var/lib/apt/lists/* \ - && rm -rf /tmp/* /var/tmp/* + && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/* # Create non-root user for security -RUN groupadd -r alya && useradd -r -g alya -d /app -s /bin/bash alya +RUN groupadd -r alya && useradd -r -g alya -d /app -s /usr/sbin/nologin alya # Copy and install Python dependencies COPY requirements.txt . RUN pip install --upgrade pip setuptools wheel && \ - pip install -r requirements.txt && \ - pip cache purge + pip install --no-cache-dir -r requirements.txt # Copy application code COPY --chown=alya:alya . . -# Create necessary directories and set permissions +# Create necessary directories with proper permissions RUN mkdir -p /app/data /app/logs /app/cache && \ - chown -R alya:alya /app && \ - chmod 755 /app/data /app/logs + chmod 755 /app/data /app/logs /app/cache # Switch to non-root user USER alya -# Simple health check tanpa external dependencies +# Health check HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \ - CMD python -c "print('Alya Bot is healthy')" || exit 1 - -EXPOSE 8080 + CMD python -c "import sys; sys.exit(0)" || exit 1 CMD ["python", "main.py"] \ No newline at end of file diff --git a/core/gemini_client.py b/core/gemini_client.py index 4bdc25d..5467005 100644 --- a/core/gemini_client.py +++ b/core/gemini_client.py @@ -243,9 +243,15 @@ async def generate_response( success = self._rotate_key() if not success: logger.critical("All API keys exhausted. Unable to generate content.") + if lang == "en": + return "Sorry, I'm having some internal issues right now. Please try again later. 😓" return "Maaf, sepertinya Alya lagi ada masalah internal. Coba lagi nanti ya. 😓" else: logger.critical(f"Failed to generate content after trying all API keys: {e}") + if lang == "en": + return "I'm so sorry, all my connections to the data center are failing. Maybe try again in a few moments? 😥" return "Aduh, maaf banget, semua koneksi Alya ke pusat data lagi gagal. Mungkin bisa coba beberapa saat lagi? 😥" - + + if lang == "en": + return "I tried multiple times but it still failed. Something's not right. Please try again later. 😔" return "Duh, Alya coba berkali-kali tapi tetep gagal. Kayaknya ada yang gak beres. Coba lagi nanti ya. 😔" diff --git a/core/mood_manager.py b/core/mood_manager.py new file mode 100644 index 0000000..8b3fa3e --- /dev/null +++ b/core/mood_manager.py @@ -0,0 +1,251 @@ +"""Dynamic Mood System for Alya Bot.""" + +import logging +from datetime import datetime +from typing import Dict, List, Any, Tuple +from dataclasses import dataclass + +logger = logging.getLogger(__name__) + +VALID_MOODS = ["happy", "tsundere", "affectionate", "neutral", "annoyed", "sad"] + +MOOD_DECAY_RATES = { + "happy": 0.15, + "tsundere": 0.10, + "affectective": 0.08, + "neutral": 0.0, + "annoyed": 0.20, + "sad": 0.12 +} + +MOOD_AFFECTION_MODIFIERS = { + "happy": 1.2, + "tsundere": 1.0, + "affectionate": 1.3, + "neutral": 1.0, + "annoyed": 1.5, + "sad": 0.8 +} + +MOOD_TRANSITION_THRESHOLD = 8 +MOOD_INTENSITY_MIN = 20 +MOOD_INTENSITY_MAX = 100 + +@dataclass +class MoodState: + mood: str + intensity: int + last_change: datetime + trigger_reason: str = "" + + def to_dict(self) -> Dict[str, Any]: + return { + "mood": self.mood, + "intensity": self.intensity, + "last_change": self.last_change.isoformat(), + "trigger_reason": self.trigger_reason + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'MoodState': + return cls( + mood=data.get("mood", "neutral"), + intensity=data.get("intensity", 50), + last_change=datetime.fromisoformat(data.get("last_change", datetime.now().isoformat())), + trigger_reason=data.get("trigger_reason", "") + ) + +class MoodManager: + def __init__(self): + self.mood_history_limit = 10 + + def calculate_mood( + self, + current_mood: str, + current_intensity: int, + affection_delta: int, + emotion_context: Dict[str, Any], + relationship_level: int, + last_mood_change: datetime + ) -> MoodState: + decayed_mood, decayed_intensity = self._apply_mood_decay( + current_mood, + current_intensity, + last_mood_change + ) + + new_mood, new_intensity, trigger = self._determine_mood_transition( + current_mood=decayed_mood, + current_intensity=decayed_intensity, + affection_delta=affection_delta, + emotion_context=emotion_context, + relationship_level=relationship_level + ) + + new_intensity = max(MOOD_INTENSITY_MIN, min(MOOD_INTENSITY_MAX, new_intensity)) + + return MoodState( + mood=new_mood, + intensity=new_intensity, + last_change=datetime.now() if new_mood != current_mood else last_mood_change, + trigger_reason=trigger + ) + + def _apply_mood_decay( + self, + mood: str, + intensity: int, + last_change: datetime + ) -> Tuple[str, int]: + if mood == "neutral": + return mood, 50 + + hours_elapsed = (datetime.now() - last_change).total_seconds() / 3600 + decay_rate = MOOD_DECAY_RATES.get(mood, 0.1) + decay_amount = int(decay_rate * hours_elapsed * 10) + new_intensity = intensity - decay_amount + + if new_intensity < MOOD_INTENSITY_MIN: + return "neutral", 50 + + return mood, new_intensity + + def _determine_mood_transition( + self, + current_mood: str, + current_intensity: int, + affection_delta: int, + emotion_context: Dict[str, Any], + relationship_level: int + ) -> Tuple[str, int, str]: + emotion = emotion_context.get("emotion", "neutral") + intent = emotion_context.get("intent", "") + signals = emotion_context.get("relationship_signals", {}) + + new_mood = current_mood + new_intensity = current_intensity + trigger = "" + + if affection_delta >= 10: + if relationship_level >= 3: + new_mood = "affectionate" + new_intensity = min(80, current_intensity + 20) + trigger = "strong_positive_interaction" + elif relationship_level >= 1: + new_mood = "happy" + new_intensity = min(75, current_intensity + 15) + trigger = "positive_interaction" + else: + if intent in ["compliment", "affection", "romantic_interest"]: + new_mood = "tsundere" + new_intensity = min(70, current_intensity + 15) + trigger = "embarrassed_by_compliment" + else: + new_mood = "happy" + new_intensity = min(70, current_intensity + 10) + trigger = "pleasant_interaction" + + elif affection_delta >= 5: + if signals.get("romantic_interest", 0) > 0.5 and relationship_level < 2: + new_mood = "tsundere" + new_intensity = min(65, current_intensity + 10) + trigger = "romantic_signal_detected" + else: + new_mood = "happy" + new_intensity = min(70, current_intensity + 10) + trigger = "good_interaction" + + # New block: Handle lighter positive emotions (amusement, optimism, etc.) even with low affection delta + elif affection_delta > 0 and emotion in ["amusement", "optimism", "relief", "pride", "caring", "approval", "happy", "joy"]: + new_mood = "happy" + new_intensity = min(65, current_intensity + 5) + trigger = f"positive_emotion_{emotion}" + + elif affection_delta <= -8: + if intent in ["insult", "toxic_behavior", "rudeness"]: + new_mood = "annoyed" + new_intensity = min(85, current_intensity + 25) + trigger = "insulted_or_toxic" + else: + new_mood = "sad" + new_intensity = min(70, current_intensity + 15) + trigger = "hurt_feelings" + + elif affection_delta <= -3: + new_mood = "annoyed" + new_intensity = min(65, current_intensity + 10) + trigger = "negative_interaction" + + if emotion in ["sad", "worried", "disappointed"] and affection_delta >= 0: + new_mood = "affectionate" + new_intensity = min(70, current_intensity + 10) + trigger = "user_needs_comfort" + + if current_mood == "annoyed" and intent in ["gratitude", "apology"]: + new_mood = "neutral" + new_intensity = 50 + trigger = "user_apologized" + + return new_mood, new_intensity, trigger + + def get_affection_modifier(self, mood: str, affection_delta: int) -> float: + base_modifier = MOOD_AFFECTION_MODIFIERS.get(mood, 1.0) + + if mood == "affectionate" and affection_delta < 0: + return 0.5 + + if mood == "annoyed" and affection_delta < 0: + return 1.8 + + return base_modifier + + def get_mood_prompt_modifier(self, mood: str, intensity: int, lang: str = "en") -> str: + if intensity >= 80: + intensity_desc = "very strong" + elif intensity >= 60: + intensity_desc = "strong" + elif intensity >= 40: + intensity_desc = "moderate" + else: + intensity_desc = "mild" + + mood_prompts = { + "happy": f"You're in a {intensity_desc} cheerful mood. Be playful, energetic, and use cute Russian expressions. Show your happiness through your words and tone.", + "tsundere": f"You're feeling {intensity_desc} tsundere (defensive and embarrassed). Act defensive when complimented, deny your feelings while showing you care. Use expressions like 'бака' (baka) when flustered. Be contradictory - say one thing but mean another.", + "affectionate": f"You're in a {intensity_desc} warm and caring mood. Show genuine affection and tenderness. Use romantic Russian expressions. Be supportive and loving, but still maintain your personality.", + "neutral": "You're in a balanced, normal mood. Respond naturally without strong emotional coloring.", + "annoyed": f"You're {intensity_desc} irritated. Be sharp, sarcastic, and use harsh Russian expressions like 'дурак' (durak) or 'сука' (suka). Show your annoyance but don't be cruel.", + "sad": f"You're feeling {intensity_desc} down or melancholic. Be vulnerable, seek comfort, and show your softer side. Use gentle Russian expressions. It's okay to show weakness." + } + + base_prompt = mood_prompts.get(mood, mood_prompts["neutral"]) + + return f"\n\n**CURRENT MOOD: {mood.upper()} (Intensity: {intensity}/100)**\n{base_prompt}" + + def get_mood_russian_expressions(self, mood: str) -> List[str]: + mood_expressions = { + "happy": ["рада", "хорошо", "милый", "спасибо", "да", "конечно"], + "tsundere": ["бака", "дурак", "что", "ну", "не", "ладно"], + "affectionate": ["люблю", "милый", "красивый", "спасибо", "моя", "мой"], + "neutral": ["да", "нет", "хорошо", "может", "ладно", "понимаешь"], + "annoyed": ["сука", "дурак", "гадость", "нет", "что ты делаешь", "ненавижу"], + "sad": ["боюсь", "извини", "плохо", "грустный", "боже", "может"] + } + return mood_expressions.get(mood, mood_expressions["neutral"]) + + def add_to_mood_history( + self, + mood_history: List[Dict[str, Any]], + new_mood_state: MoodState + ) -> List[Dict[str, Any]]: + mood_entry = new_mood_state.to_dict() + updated_history = mood_history.copy() if mood_history else [] + updated_history.append(mood_entry) + + if len(updated_history) > self.mood_history_limit: + updated_history = updated_history[-self.mood_history_limit:] + + return updated_history + + def validate_mood(self, mood: str) -> bool: + return mood in VALID_MOODS diff --git a/database/database_manager.py b/database/database_manager.py index 23cf200..dc16b73 100755 --- a/database/database_manager.py +++ b/database/database_manager.py @@ -1084,5 +1084,92 @@ def get_rag_texts(self, user_id: int, limit: int = 10) -> list: except Exception as e: logger.error(f"Error in get_rag_texts for user {user_id}: {e}", exc_info=True) return [] + + # ======================================================================== + # MOOD SYSTEM METHODS + # ======================================================================== + + def update_user_mood( + self, + user_id: int, + mood: str, + intensity: int, + mood_history: List[Dict[str, Any]] = None + ) -> bool: + """ + Update user's current mood state. + + Args: + user_id: Telegram user ID + mood: New mood state (happy, tsundere, affectionate, neutral, annoyed, sad) + intensity: Mood intensity (0-100) + mood_history: Optional updated mood history + + Returns: + bool: True if updated successfully + """ + try: + with db_session_context() as session: + user = session.query(User).filter(User.id == user_id).first() + if not user: + logger.warning(f"Cannot update mood for non-existent user {user_id}") + return False + + old_mood = user.current_mood + user.current_mood = mood + user.mood_intensity = intensity + user.last_mood_change = datetime.now() + + if mood_history is not None: + user.mood_history = mood_history + + session.commit() + + logger.info( + f"Updated mood for user {user_id}: {old_mood} → {mood} " + f"(intensity: {intensity})" + ) + return True + + except Exception as e: + logger.error(f"Error updating mood for user {user_id}: {e}", exc_info=True) + return False + + def get_user_mood(self, user_id: int) -> Dict[str, Any]: + """ + Get user's current mood state. + + Args: + user_id: Telegram user ID + + Returns: + Dict with mood, intensity, last_change, and history + """ + try: + with db_session_context() as session: + user = session.query(User).filter(User.id == user_id).first() + if not user: + return { + "mood": "neutral", + "intensity": 50, + "last_change": datetime.now(), + "history": [] + } + + return { + "mood": user.current_mood or "neutral", + "intensity": user.mood_intensity or 50, + "last_change": user.last_mood_change or datetime.now(), + "history": user.mood_history or [] + } + + except Exception as e: + logger.error(f"Error getting mood for user {user_id}: {e}", exc_info=True) + return { + "mood": "neutral", + "intensity": 50, + "last_change": datetime.now(), + "history": [] + } db_manager = DatabaseManager() \ No newline at end of file diff --git a/database/migrate_add_mood.py b/database/migrate_add_mood.py new file mode 100644 index 0000000..3145b37 --- /dev/null +++ b/database/migrate_add_mood.py @@ -0,0 +1,67 @@ +"""Database migration script for adding mood system columns.""" + +import sys +import os +import logging +from sqlalchemy import text, inspect + +# Add project root to path +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from database.session import engine + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +def migrate_add_mood_columns(): + """Add mood columns to users table safely.""" + try: + inspector = inspect(engine) + existing_columns = [c['name'] for c in inspector.get_columns('users')] + + with engine.connect() as conn: + # defined columns to add: name, definition + new_columns = [ + ("current_mood", "VARCHAR(20) DEFAULT 'neutral'"), + ("mood_intensity", "SMALLINT DEFAULT 50"), + ("last_mood_change", "DATETIME DEFAULT CURRENT_TIMESTAMP"), + ("mood_history", "JSON") + ] + + for col_name, col_def in new_columns: + if col_name not in existing_columns: + logger.info(f"Adding column '{col_name}'...") + conn.execute(text(f"ALTER TABLE users ADD COLUMN {col_name} {col_def}")) + else: + logger.info(f"Column '{col_name}' already exists.") + + conn.commit() + + # Create index if it doesn't exist + # Note: Checking for index existence is harder across versions, + # safe to try/catch or just let it fail if exists if using raw sql without checks + # But duplicate index usually doesn't break things as hard as syntax error, + # actually duplicate index name is an error. + + indexes = inspector.get_indexes('users') + index_names = [i['name'] for i in indexes] + + if 'idx_user_mood' not in index_names: + logger.info("Creating mood index...") + conn.execute(text("CREATE INDEX idx_user_mood ON users(current_mood, mood_intensity)")) + conn.commit() + else: + logger.info("Index 'idx_user_mood' already exists.") + + logger.info("Migration completed successfully!") + + except Exception as e: + logger.error(f"Migration failed: {e}") + raise + +if __name__ == "__main__": + print("MOOD SYSTEM DATABASE MIGRATION") + if input("Continue with migration? (yes/no): ").lower() in ['yes', 'y']: + migrate_add_mood_columns() + else: + print("Migration cancelled.") diff --git a/database/models.py b/database/models.py index adc3f28..db5be86 100755 --- a/database/models.py +++ b/database/models.py @@ -42,6 +42,12 @@ class User(Base): interaction_count = Column(Integer, default=0, index=True) topics_discussed = Column(JSON, default=lambda: []) + # Mood System - tracks Alya's current emotional state + current_mood = Column(String(20), default='neutral', index=True) + mood_intensity = Column(SmallInteger, default=50) # 0-100 scale + last_mood_change = Column(DateTime, default=datetime.now) + mood_history = Column(JSON, default=lambda: []) # Recent mood transitions + conversations = relationship("Conversation", back_populates="user", cascade="all, delete-orphan") summaries = relationship("ConversationSummary", back_populates="user", cascade="all, delete-orphan") @@ -49,6 +55,7 @@ class User(Base): Index('idx_user_active_interaction', 'is_active', 'last_interaction'), Index('idx_user_relationship_level', 'relationship_level', 'affection_points'), Index('idx_user_stats', 'interaction_count', 'relationship_level'), + Index('idx_user_mood', 'current_mood', 'mood_intensity'), {'mysql_charset': 'utf8mb4', 'mysql_collate': 'utf8mb4_unicode_ci'} ) diff --git a/handlers/conversation.py b/handlers/conversation.py index 22ce807..21083a4 100644 --- a/handlers/conversation.py +++ b/handlers/conversation.py @@ -163,15 +163,63 @@ async def chat_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) message_context = self.nlp.get_message_context(query, user.id) logger.debug(f"Message context for user {user.id}: {message_context}") - # === STEP 2: Update affection based on analysis === + # === STEP 2: Calculate affection delta (before applying) === + affection_delta = 0 if message_context: - self._update_affection_from_context(user.id, message_context) + affection_delta = self._calculate_affection_delta(user.id, message_context) - # === STEP 3: Increment interaction count (will recalculate level) === + # === STEP 3: Get current mood and calculate new mood === + from core.mood_manager import MoodManager + mood_manager = MoodManager() + + current_mood_data = self.db.get_user_mood(user.id) + user_info = self.db.get_user_relationship_info(user.id) + + new_mood_state = mood_manager.calculate_mood( + current_mood=current_mood_data["mood"], + current_intensity=current_mood_data["intensity"], + affection_delta=affection_delta, + emotion_context=message_context, + relationship_level=user_info["relationship_level"], + last_mood_change=current_mood_data["last_change"] + ) + + # === STEP 4: Apply mood modifier to affection delta === + mood_modifier = mood_manager.get_affection_modifier( + new_mood_state.mood, + affection_delta + ) + modified_affection_delta = int(affection_delta * mood_modifier) + + logger.info( + f"[MOOD] User {user.id}: {current_mood_data['mood']} → {new_mood_state.mood} " + f"(intensity: {new_mood_state.intensity}) | " + f"Affection: {affection_delta} → {modified_affection_delta} (×{mood_modifier:.1f})" + ) + + # === STEP 5: Update affection with mood-modified delta === + if modified_affection_delta != 0: + self.db.update_affection(user.id, modified_affection_delta) + + # === STEP 6: Update mood in database === + updated_history = mood_manager.add_to_mood_history( + current_mood_data["history"], + new_mood_state + ) + self.db.update_user_mood( + user.id, + new_mood_state.mood, + new_mood_state.intensity, + updated_history + ) + + # === STEP 7: Increment interaction count (will recalculate level) === self.db.increment_interaction_count(user.id) - # === STEP 4: Prepare context with LATEST relationship level === - user_context = await self._prepare_conversation_context(user, query, lang, message_context) + # === STEP 8: Prepare context with LATEST relationship level and MOOD === + user_context = await self._prepare_conversation_context( + user, query, lang, message_context, new_mood_state, mood_manager + ) history = self.context_manager.get_context_window(user.id) response = await self.gemini.generate_response( user_id=user.id, @@ -218,9 +266,11 @@ async def _prepare_conversation_context( user, query: str, lang: str, - message_context: Dict[str, Any] + message_context: Dict[str, Any], + mood_state = None, + mood_manager = None ) -> Dict[str, Any]: - """Build conversation context with persona prompt, history, and relationship level.""" + """Build conversation context with persona prompt, history, relationship level, and mood.""" user_task = asyncio.create_task(self._get_user_info(user)) user_info = self.db.get_user_relationship_info(user.id) @@ -240,6 +290,21 @@ async def _prepare_conversation_context( lang=lang ) + # Add mood-specific personality modifier + if mood_state and mood_manager: + mood_prompt = mood_manager.get_mood_prompt_modifier( + mood_state.mood, + mood_state.intensity, + lang + ) + persona_prompt += mood_prompt + + # Add mood-appropriate Russian expressions hint + mood_expressions = mood_manager.get_mood_russian_expressions(mood_state.mood) + if mood_expressions: + expressions_hint = f"\n\nSuggested Russian expressions for current mood: {', '.join(mood_expressions[:4])}" + persona_prompt += expressions_hint + # Extract semantic topics from provided message_context semantic_topics = message_context.get("semantic_topics", []) if message_context else [] @@ -444,7 +509,9 @@ async def _process_and_send_response( self.db.save_message(user.id, "assistant", response) self.memory.save_bot_response(user.id, response) if message_context: - self._update_affection_from_context(user.id, message_context) + # self._update_affection_from_context(user.id, message_context) + # Affection update handled in chat_command before response generation + pass # Step 1: Split mixed quote-narration paragraphs response = self._split_mixed_quote_paragraphs(response) @@ -531,17 +598,16 @@ def _get_relationship_context(self, user: Any, relationship_level: int, is_admin lang=lang ) - def _update_affection_from_context(self, user_id: int, message_context: Dict[str, Any]) -> None: - """Update Alya's affection based on user's emotion, intent, and relationship signals.""" + def _calculate_affection_delta(self, user_id: int, message_context: Dict[str, Any]) -> int: + """Calculate affection delta based on user's emotion, intent, and relationship signals.""" if not message_context: - self.db.update_affection(user_id, AFFECTION_POINTS.get("conversation", 1)) - return + return AFFECTION_POINTS.get("conversation", 1) affection_delta = 0 # User's emotions towards Alya emotion = message_context.get("emotion", "") - if emotion in ["happy", "excited", "grateful", "joy", "love", "admiration"]: + if emotion in ["happy", "excited", "grateful", "joy", "love", "admiration", "amusement", "optimism", "relief", "pride", "caring", "approval"]: affection_delta += AFFECTION_POINTS.get("positive_emotion", 2) logger.debug(f"User {user_id} shows positive emotion '{emotion}': +{AFFECTION_POINTS.get('positive_emotion', 2)}") elif emotion in ["sad", "worried", "disappointed"]: @@ -607,6 +673,7 @@ def _update_affection_from_context(self, user_id: int, message_context: Dict[str if abs(affection_delta) >= 1: affection_delta = round(affection_delta) logger.info(f"[AFFECTION] User {user_id}: {affection_delta:+d} | emotion={emotion}, intent={intent}") - self.db.update_affection(user_id, affection_delta) + return affection_delta else: - logger.debug(f"[AFFECTION] User {user_id}: no change (delta={affection_delta:.1f})") \ No newline at end of file + logger.debug(f"[AFFECTION] User {user_id}: no change (delta={affection_delta:.1f})") + return 0 \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 305358f..ba37a4a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,24 +8,20 @@ PyYAML>=6.0 # AI & NLP google-generativeai>=0.3.0 transformers>=4.35.0 -torch>=2.0.0 -sentence-transformers>=2.2.2 +torch>=2.0.0,<3.0 ; sys_platform != "win32" numpy>=1.24.0 # Database (MySQL) sqlalchemy>=2.0.41 pymysql>=1.1.0 mysql-connector-python>=8.0.32 -cryptography>=3.4.8 # Web & API requests>=2.31.0 aiohttp>=3.8.0 -beautifulsoup4>=4.12.0 # Utilities emoji>=2.8.0 -pydantic>=2.4.0 psutil>=5.9.0 pillow>=10.0.0 qrcode>=7.4.2 diff --git a/utils/affection_helper.py b/utils/affection_helper.py new file mode 100644 index 0000000..5b059d4 --- /dev/null +++ b/utils/affection_helper.py @@ -0,0 +1,50 @@ +"""Helper methods for affection calculation.""" + +from config.settings import AFFECTION_POINTS + +def calculate_affection_delta_from_context(message_context: dict) -> int: + if not message_context: + return AFFECTION_POINTS.get("conversation", 1) + + affection_delta = 0 + emotion = message_context.get("emotion", "") + intent = message_context.get("intent", "") + signals = message_context.get("relationship_signals", {}) + + if emotion in ["happy", "excited", "grateful", "joy", "love", "admiration"]: + affection_delta += AFFECTION_POINTS.get("positive_emotion", 2) + elif emotion in ["sad", "worried", "disappointed"]: + affection_delta += AFFECTION_POINTS.get("mild_positive_emotion", 1) + elif emotion in ["angry", "frustrated", "annoyed"]: + if message_context.get("directed_at_alya", True): + affection_delta += AFFECTION_POINTS.get("anger", -3) + else: + affection_delta += AFFECTION_POINTS.get("mild_positive_emotion", 1) + + intent_map = { + "gratitude": "gratitude", "apology": "apology", "affection": "affection", + "greeting": "greeting", "compliment": "compliment", "question": "question", + "meaningful_conversation": "meaningful_conversation", + "asking_about_alya": "asking_about_alya", + "remembering_details": "remembering_details", + "insult": "insult", "abuse": "insult", + "toxic": "toxic_behavior", "toxic_behavior": "toxic_behavior", + "bullying": "toxic_behavior", "rudeness": "rudeness", + "ignoring": "ignoring", "inappropriate": "inappropriate", + "command": "command", "departure": "command" + } + + if intent in intent_map: + affection_delta += AFFECTION_POINTS.get(intent_map[intent], 0) + + signal_delta = ( + signals.get("friendliness", 0) * AFFECTION_POINTS.get("friendliness", 6) + + signals.get("romantic_interest", 0) * AFFECTION_POINTS.get("romantic_interest", 10) + + signals.get("conflict", 0) * AFFECTION_POINTS.get("conflict", -3) + ) + affection_delta += signal_delta + + if affection_delta < 0: + affection_delta = max(affection_delta, AFFECTION_POINTS.get("min_penalty", -4)) + + return affection_delta if affection_delta != 0 else AFFECTION_POINTS.get("conversation", 1)