openclaw-workspace/habit_bot.py

#!/usr/bin/env python3
"""
Unified Telegram Bot - Habit, Diet, URL Summarizer
Features:
- YouTube/Blog/News summarization (EN/KO)
- Habit logging
- Diet/food logging with photo analysis
- Morning briefing
- Night debrief + motivation
"""

import os
import sys
import json
import re
import datetime
from typing import Optional, Dict, List
from dataclasses import dataclass, field
from enum import Enum

# Try to import telegram, handle if not available
try:
    from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
    from telegram.ext import Application, CommandHandler, MessageHandler, CallbackQueryHandler, ContextTypes
    TELEGRAM_AVAILABLE = True
except ImportError:
    TELEGRAM_AVAILABLE = False

# Configuration
TELEGRAM_BOT_TOKEN = os.environ.get('TELEGRAM_BOT_TOKEN', '8325588419:AAGghb0nosWG8g6QtYeghqUs0RHug06uG74')
OBSIDIAN_PATH = os.environ.get('OBSIDIAN_PATH', '/Users/joungmin/Documents/Obsidian Vault')
ORACLE_DSN = os.environ.get('ORACLE_DSN', 'h8i4i0g8cxtd2lpf_high')
ORACLE_USER = os.environ.get('ORACLE_USER', 'admin')
ORACLE_PASSWORD = os.environ.get('ORACLE_PASSWORD', 'Carter55@26@1')
ORACLE_WALLET = os.environ.get('ORACLE_WALLET', '/Users/joungmin/devkit/db_conn/Wallet_H8I4I0G8CXTD2LPF')

# In-memory storage (replace with Oracle later)
DATA_DIR = '/tmp/habit_bot'
os.makedirs(DATA_DIR, exist_ok=True)

def load_json(f):
    if os.path.exists(f):
        with open(f, 'r') as file:
            return json.load(file)
    return {}

def save_json(f, data):
    with open(f, 'w') as file:
        json.dump(data, file, indent=2, default=str)

HABITS_FILE = os.path.join(DATA_DIR, 'habits.json')
HABIT_LOGS_FILE = os.path.join(DATA_DIR, 'habit_logs.json')
FOOD_LOGS_FILE = os.path.join(DATA_DIR, 'food_logs.json')
USER_DATA_FILE = os.path.join(DATA_DIR, 'users.json')

# Motivational quotes
MOTIVATIONAL_QUOTES = [
    "The only bad workout is the one that didn't happen. 💪",
    "Every expert was once a beginner. Keep going! 🌟",
    "Success is the sum of small efforts repeated day in and day out. 📈",
    "You don't have to be great to start, but you have to start to be great. 🚀",
    "The body achieves what the mind believes. 🧠",
    "Discipline is doing what needs to be done, even if you don't want to do it. 🔥",
    "Your future is created by what you do today, not tomorrow. ⏰",
    "Small steps add up to big changes. Keep walking! 👣",
]

class UserData:
    def __init__(self):
        self.habits = load_json(HABITS_FILE)
        self.habit_logs = load_json(HABIT_LOGS_FILE)
        self.food_logs = load_json(FOOD_LOGS_FILE)
        self.users = load_json(USER_DATA_FILE)
    
    def save(self):
        save_json(HABITS_FILE, self.habits)
        save_json(HABIT_LOGS_FILE, self.habit_logs)
        save_json(FOOD_LOGS_FILE, self.food_logs)
        save_json(USER_DATA_FILE, self.users)

data = UserData()

# URL Patterns
URL_PATTERNS = {
    'youtube': r'(?:youtube\.com|youtu\.be)',
    'blog': r'blog\.|medium\.com|substack\.com',
    'news': r'news\.|cnn\.com|bbc\.com|nytimes\.com|reuters\.com',
}

@dataclass
class Habit:
    name: str
    description: str = ''
    frequency: str = 'daily'
    streak: int = 0
    is_active: bool = True

@dataclass  
class HabitLog:
    habit_name: str
    date: str
    status: str  # completed, skipped
    notes: str = ''
    timestamp: str = ''

@dataclass
class FoodLog:
    date: str
    meal_type: str
    food_name: str
    photo_url: str = ''
    calories: int = 0
    carbs: float = 0
    protein: float = 0
    fat: float = 0
    analysis: str = ''

# ============== Telegram Handlers ==============

async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Welcome message"""
    welcome = """
🔮 **Welcome to Your Life Assistant Bot!**

I can help you with:

📚 **Content Summarization**
- Send me a YouTube/Blog/News URL
- I'll summarize in English & Korean

✅ **Habit Tracking**
- `/habit add <name>` - Add new habit
- `/habit log <name> [notes]` - Log completion
- `/habit list` - Show all habits
- `/habit streak <name>` - Show streak

🍽️ **Diet Logging**
- Send a photo of your meal
- Or text: "had chicken breast 200g"
- I'll analyze nutrition

📊 **Daily Status**
- `/morning` - Morning briefing
- `/debrief` - Night summary + motivation
- `/status` - Today's progress

What would you like to do?
    """
    await update.message.reply_text(welcome, parse_mode='Markdown')

async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Help message"""
    help_text = """
🔮 **Available Commands**

**Habit Management**
- `/habit add <name>` - Add new habit
- `/habit log <name> [notes]` - Log completion
- `/habit list` - Show all habits
- `/habit streak <name>` - Show streak
- `/habit delete <name>` - Remove habit

**Food/Diet Logging**
- Send meal photo - AI nutrition analysis
- Text: "breakfast eggs 2" - Quick log
- `/food today` - Today's meals
- `/food stats` - Nutrition summary

**Daily Briefings**
- `/morning` - Morning briefing
- `/debrief` - Night summary + motivation
- `/status` - Current progress

**Content**
- Send URL - Summarize (EN/KO)
    """
    await update.message.reply_text(help_text, parse_mode='Markdown')

# ============== Habit Commands ==============

async def habit_add(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Add new habit"""
    if not context.args:
        await update.message.reply_text("Usage: `/habit add <habit name>`")
        return
    
    habit_name = ' '.join(context.args).strip().lower()
    user_id = str(update.message.from_user.id)
    
    if user_id not in data.habits:
        data.habits[user_id] = {}
    
    if habit_name in data.habits[user_id]:
        await update.message.reply_text(f"✅ Habit '{habit_name}' already exists!")
        return
    
    data.habits[user_id][habit_name] = {
        'name': habit_name,
        'streak': 0,
        'created_at': datetime.datetime.now().isoformat(),
        'is_active': True
    }
    data.save()
    
    await update.message.reply_text(f"✅ Added habit: *{habit_name}*", parse_mode='Markdown')

async def habit_list(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """List all habits"""
    user_id = str(update.message.from_user.id)
    
    if user_id not in data.habits or not data.habits[user_id]:
        await update.message.reply_text("No habits yet. Add one with `/habit add <name>`")
        return
    
    today = datetime.datetime.now().strftime('%Y-%m-%d')
    completed_today = set()
    
    if user_id in data.habit_logs and today in data.habit_logs[user_id]:
        for log in data.habit_logs[user_id][today]:
            if log.get('status') == 'completed':
                completed_today.add(log.get('habit_name', ''))
    
    text = "📋 **Your Habits:**\n\n"
    for name, info in data.habits[user_id].items():
        if info.get('is_active', True):
            streak = info.get('streak', 0)
            status = "✅" if name in completed_today else "⬜"
            text += f"{status} *{name}* (streak: {streak}🔥)\n"
    
    await update.message.reply_text(text, parse_mode='Markdown')

async def habit_log(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Log habit completion"""
    if not context.args:
        await update.message.reply_text("Usage: `/habit log <habit name> [notes]`")
        return
    
    user_id = str(update.message.from_user.id)
    today = datetime.datetime.now().strftime('%Y-%m-%d')
    
    # Parse args
    args_text = ' '.join(context.args)
    if ' ' in args_text:
        habit_name, notes = args_text.split(' ', 1)
    else:
        habit_name = args_text
        notes = ''
    
    habit_name = habit_name.strip().lower()
    
    # Verify habit exists
    if user_id not in data.habits or habit_name not in data.habits[user_id]:
        await update.message.reply_text(f"❌ Habit '{habit_name}' not found!")
        return
    
    # Log it
    if user_id not in data.habit_logs:
        data.habit_logs[user_id] = {}
    if today not in data.habit_logs[user_id]:
        data.habit_logs[user_id][today] = []
    
    data.habit_logs[user_id][today].append({
        'habit_name': habit_name,
        'status': 'completed',
        'notes': notes,
        'timestamp': datetime.datetime.now().isoformat()
    })
    
    # Update streak
    prev_streak = data.habits[user_id][habit_name].get('streak', 0)
    data.habits[user_id][habit_name]['streak'] = prev_streak + 1
    data.save()
    
    # Motivational response
    quote = MOTIVATIONAL_QUOTES[datetime.datetime.now().second % len(MOTIVATIONAL_QUOTES)]
    await update.message.reply_text(
        f"✅ *{habit_name}* completed! Streak: {prev_streak + 1}🔥\n\n{quote}",
        parse_mode='Markdown'
    )

async def habit_streak(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Show habit streak"""
    if not context.args:
        await update.message.reply_text("Usage: `/habit streak <habit name>`")
        return
    
    user_id = str(update.message.from_user.id)
    habit_name = ' '.join(context.args).strip().lower()
    
    if user_id not in data.habits or habit_name not in data.habits[user_id]:
        await update.message.reply_text(f"❌ Habit '{habit_name}' not found!")
        return
    
    streak = data.habits[user_id][habit_name].get('streak', 0)
    await update.message.reply_text(
        f"🔥 *{habit_name}* streak: {streak} days",
        parse_mode='Markdown'
    )

# ============== Food/Diet Commands ==============

async def food_log(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Log food/meal"""
    user_id = str(update.message.from_user.id)
    today = datetime.datetime.now().strftime('%Y-%m-%d')
    now = datetime.datetime.now().strftime('%H:%M')
    
    # Determine meal type
    hour = datetime.datetime.now().hour
    if 5 <= hour < 11:
        meal_type = 'breakfast'
    elif 11 <= hour < 14:
        meal_type = 'lunch'
    elif 14 <= hour < 17:
        meal_type = 'snack'
    else:
        meal_type = 'dinner'
    
    text = ' '.join(context.args) if context.args else ''
    
    # Simple food analysis (placeholder - would use MiniMax/vision API)
    food_info = analyze_food_text(text)
    
    if user_id not in data.food_logs:
        data.food_logs[user_id] = {}
    if today not in data.food_logs[user_id]:
        data.food_logs[user_id][today] = []
    
    data.food_logs[user_id][today].append({
        'meal_type': meal_type,
        'food_name': text or 'Photo',
        'time': now,
        'calories': food_info['calories'],
        'carbs': food_info['carbs'],
        'protein': food_info['protein'],
        'fat': food_info['fat'],
        'timestamp': datetime.datetime.now().isoformat()
    })
    data.save()
    
    # Keto guidance
    remaining = 2000 - food_info['calories']  # Simplified
    await update.message.reply_text(
        f"🍽️ Logged: *{text or 'Photo'}*\n"
        f"📊 {food_info['calories']}kcal | "
        f" carbs: {food_info['carbs']}g | "
        f"protein: {food_info['protein']}g | "
        f"fat: {food_info['fat']}g\n"
        f"\n💪 Keep going! {remaining}kcal remaining today.",
        parse_mode='Markdown'
    )

def analyze_food_text(text: str) -> Dict:
    """Simple food analysis (placeholder)"""
    # This would use MiniMax/vision API in production
    # For now, return placeholder data
    
    # Simple keyword matching
    calories = 0
    carbs = 0
    protein = 0
    fat = 0
    
    food_database = {
        'chicken': {'cal': 165, 'carb': 0, 'pro': 31, 'fat': 3.6},
        'egg': {'cal': 78, 'carb': 0.6, 'pro': 6, 'fat': 5},
        'rice': {'cal': 130, 'carb': 28, 'pro': 2.7, 'fat': 0.3},
        'beef': {'cal': 250, 'carb': 0, 'pro': 26, 'fat': 15},
        'salad': {'cal': 50, 'carb': 5, 'pro': 2, 'fat': 3},
        'bread': {'cal': 265, 'carb': 49, 'pro': 9, 'fat': 3.2},
        'apple': {'cal': 95, 'carb': 25, 'pro': 0.5, 'fat': 0.3},
        'banana': {'cal': 105, 'carb': 27, 'pro': 1.3, 'fat': 0.4},
    }
    
    text_lower = text.lower()
    for food, info in food_database.items():
        if food in text_lower:
            # Check for quantity
            numbers = re.findall(r'\d+', text)
            qty = int(numbers[0]) if numbers else 1
            calories += info['cal'] * qty
            carbs += info['carb'] * qty
            protein += info['pro'] * qty
            fat += info['fat'] * qty
    
    # Default if no match
    if calories == 0:
        calories, carbs, protein, fat = 300, 20, 15, 12
    
    return {'calories': calories, 'carbs': carbs, 'protein': protein, 'fat': fat}

async def food_today(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Show today's food log"""
    user_id = str(update.message.from_user.id)
    today = datetime.datetime.now().strftime('%Y-%m-%d')
    
    if user_id not in data.food_logs or today not in data.food_logs[user_id]:
        await update.message.reply_text("No food logged today yet!")
        return
    
    logs = data.food_logs[user_id][today]
    total_cal = sum(l.get('calories', 0) for l in logs)
    total_carb = sum(l.get('carbs', 0) for l in logs)
    total_pro = sum(l.get('protein', 0) for l in logs)
    total_fat = sum(l.get('fat', 0) for l in logs)
    
    text = f"🍽️ **Today's Meals:**\n\n"
    for log in logs:
        text += f"- {log.get('meal_type', '')}: {log.get('food_name', '')} ({log.get('calories', 0)}kcal)\n"
    
    text += f"\n📊 **Total:** {total_cal}kcal | {total_carb}g carbs | {total_pro}g protein | {total_fat}g fat"
    
    remaining = 2000 - total_cal
    if remaining > 0:
        text += f"\n💪 {remaining}kcal remaining for today!"
    else:
        text += f"\n⚠️ Over by {abs(remaining)}kcal"
    
    await update.message.reply_text(text, parse_mode='Markdown')

# ============== Daily Briefings ==============

async def morning_briefing(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Morning briefing with habits to do"""
    user_id = str(update.message.from_user.id)
    today = datetime.datetime.now().strftime('%Y-%m-%d')
    
    if user_id not in data.habits or not data.habits[user_id]:
        await update.message.reply_text("☀️ Good morning! No habits set yet. Add some with `/habit add <name>`")
        return
    
    # Check yesterday's uncompleted habits
    yesterday = (datetime.datetime.now() - datetime.timedelta(days=1)).strftime('%Y-%m-%d')
    uncompleted = []
    
    if user_id in data.habit_logs and yesterday in data.habit_logs[user_id]:
        completed = set(l.get('habit_name', '') for l in data.habit_logs[user_id][yesterday] 
                       if l.get('status') == 'completed')
        for name in data.habits[user_id]:
            if name not in completed and data.habits[user_id][name].get('is_active', True):
                uncompleted.append(name)
    
    text = "☀️ **Good Morning!** Here's your plan:\n\n"
    
    # Today's habits
    text += "*Today's Habits:*\n"
    for name, info in data.habits[user_id].items():
        if info.get('is_active', True):
            streak = info.get('streak', 0)
            text += f"⬜ {name} (🔥 {streak})\n"
    
    if uncompleted:
        text += f"\n*Yesterday's unfinished:*\n"
        for name in uncompleted:
            text += f"⚠️ {name}\n"
    
    text += "\n💪 Let's make today count!"
    
    await update.message.reply_text(text, parse_mode='Markdown')

async def debrief(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Night debrief with progress and motivation"""
    user_id = str(update.message.from_user.id)
    today = datetime.datetime.now().strftime('%Y-%m-%d')
    
    # Count today's achievements
    habits_completed = 0
    if user_id in data.habit_logs and today in data.habit_logs[user_id]:
        habits_completed = len([l for l in data.habit_logs[user_id][today] 
                               if l.get('status') == 'completed'])
    
    total_habits = len([h for h in data.habits.get(user_id, {}).values() 
                        if h.get('is_active', True)])
    
    # Food stats
    total_cal = 0
    if user_id in data.food_logs and today in data.food_logs[user_id]:
        total_cal = sum(l.get('calories', 0) for l in data.food_logs[user_id][today])
    
    quote = MOTIVATIONAL_QUOTES[datetime.datetime.now().second % len(MOTIVATIONAL_QUOTES)]
    
    text = f"🌙 **Night Debrief**\n\n"
    text += f"📋 *Habits:* {habits_completed}/{total_habits} completed\n"
    text += f"🍽️ *Calories:* {total_cal} consumed\n"
    
    if habits_completed >= total_habits:
        text += f"\n🎉 Amazing day! You crushed all your habits!"
    elif habits_completed > 0:
        text += f"\n👍 Good effort! {total_habits - habits_completed} habits left for tomorrow."
    else:
        text += f"\n💪 Tomorrow is a new chance. You've got this!"
    
    text += f"\n\n{quote}"
    
    await update.message.reply_text(text, parse_mode='Markdown')

# ============== URL Summarization ==============

async def handle_url(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handle URL messages - summarize content"""
    url = update.message.text.strip()
    
    # Check if it's a URL
    if not url.startswith(('http://', 'https://')):
        return
    
    # Determine type
    url_type = 'general'
    for t, pattern in URL_PATTERNS.items():
        if re.search(pattern, url, re.IGNORECASE):
            url_type = t
            break
    
    await update.message.reply_text(f"🔄 Processing {url_type} URL...")
    
    # TODO: Use MiniMax API to summarize
    # For now, return placeholder
    
    summary_en = f"""
**English Summary ({url_type})**

Title: [Would extract from page]
Key Points:
1. [Main point 1]
2. [Main point 2]
3. [Main point 3]

Tags: #summary
    """.strip()
    
    summary_ko = f"""
**한국어 요약 ({url_type})**

제목: [Would extract from page]
주요 포인트:
1. [메인 포인트 1]
2. [메인 포인트 2]
3. [메인 포인트 3]

태그: #요약
    """.strip()
    
    # Save to Obsidian
    save_to_obsidian(url, summary_en, summary_ko, url_type)
    
    # Send response
    text = f"**📚 Summary saved to Obsidian**\n\n{summary_en}\n\n---\n\n{summary_ko}"
    await update.message.reply_text(text, parse_mode='Markdown')

def save_to_obsidian(url: str, summary_en: str, summary_ko: str, url_type: str):
    """Save summary to Obsidian"""
    date = datetime.datetime.now().strftime('%Y-%m-%d')
    filename = f"URL Summary - {date}.md"
    filepath = os.path.join(OBSIDIAN_PATH, 'URL Summaries', filename)
    
    os.makedirs(os.path.dirname(filepath), exist_ok=True)
    
    content = f"""# URL Summary - {date}

**Source:** {url}
**Type:** {url_type}
**Date:** {date}

---

## English Summary

{summary_en}

---

## 한국어 요약

{summary_ko}

---

*Generated by OpenClaw*
"""
    
    with open(filepath, 'w', encoding='utf-8') as f:
        f.write(content)

# ============== Main ==============

def main():
    """Run the bot"""
    if not TELEGRAM_AVAILABLE:
        print("⚠️ Telegram library not installed. Run: pip install python-telegram-bot")
        print("Bot code is ready but cannot run without the library.")
        return
    
    app = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
    
    # Commands
    app.add_handler(CommandHandler('start', start_command))
    app.add_handler(CommandHandler('help', help_command))
    app.add_handler(CommandHandler('habit', habit_list))  # Default handler
    app.add_handler(CommandHandler('habit_add', habit_add))
    app.add_handler(CommandHandler('habit_list', habit_list))
    app.add_handler(CommandHandler('habit_log', habit_log))
    app.add_handler(CommandHandler('habit_streak', habit_streak))
    app.add_handler(CommandHandler('food', food_log))
    app.add_handler(CommandHandler('food_today', food_today))
    app.add_handler(CommandHandler('morning', morning_briefing))
    app.add_handler(CommandHandler('debrief', debrief))
    app.add_handler(CommandHandler('status', lambda u, c: food_today(u, c)))  # Alias
    
    # URL handler
    app.add_handler(MessageHandler(None, handle_url))
    
    print("🔮 Starting Habit & Diet Bot...")
    app.run_polling()

if __name__ == '__main__':
    main()