- Fixed headers to be properly bilingual (EN/KO) - Added Korean descriptions for all features Files: - habit_bot.py - stock_tracker.py
853 lines
29 KiB
Python
853 lines
29 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
#
|
||
# Habit & Diet Telegram Bot with URL Summarization / URL 요약을 포함한 습관 및 식단 Telegram 봇
|
||
#
|
||
# 기능 / Features:
|
||
# - 습관 추적 (/habit add, log, list) / Habit tracking
|
||
# - 음식 기록 및 영양 분석 / Food logging with nutrition analysis
|
||
# - 아침 브리핑 및 밤 디브리프 / Morning briefings and night debrief
|
||
# - URL 요약 (YouTube/Blog/News) / URL summarization
|
||
# - 케토 다이어트 가이드라인 / Keto diet guidelines
|
||
"""
|
||
|
||
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)
|
||
|
||
def get_daily_totals(self, user_id: str, date: str = None) -> Dict:
|
||
"""Get daily nutrition totals for a user"""
|
||
if date is None:
|
||
date = datetime.datetime.now().strftime('%Y-%m-%d')
|
||
|
||
totals = {'calories': 0, 'carbs': 0, 'protein': 0, 'fat': 0}
|
||
|
||
if user_id in self.food_logs and date in self.food_logs[user_id]:
|
||
for log in self.food_logs[user_id][date]:
|
||
totals['calories'] += log.get('calories', 0)
|
||
totals['carbs'] += log.get('carbs', 0)
|
||
totals['protein'] += log.get('protein', 0)
|
||
totals['fat'] += log.get('fat', 0)
|
||
|
||
return totals
|
||
|
||
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}
|
||
|
||
# ============== MiniMax Vision API ==============
|
||
|
||
MINIMAX_API_URL = "https://api.minimax.chat/v1/text/chatcompletion_v2"
|
||
MINIMAX_API_KEY = os.environ.get('MINIMAX_API_KEY', '')
|
||
|
||
async def analyze_food_photo(file_path: str) -> Dict:
|
||
"""
|
||
Analyze food photo using MiniMax Vision API
|
||
|
||
Returns: Dict with calories, carbs, protein, fat estimation
|
||
"""
|
||
if not MINIMAX_API_KEY:
|
||
# Fallback to placeholder if no API key
|
||
return {
|
||
'calories': 400,
|
||
'carbs': 25,
|
||
'protein': 30,
|
||
'fat': 20,
|
||
'detected_foods': ['food (placeholder - add MiniMax API key)'],
|
||
'confidence': 0.5
|
||
}
|
||
|
||
try:
|
||
import base64
|
||
|
||
# Read and encode image
|
||
with open(file_path, 'rb') as f:
|
||
image_b64 = base64.b64encode(f.read()).decode('utf-8')
|
||
|
||
# Prepare vision prompt
|
||
prompt = """Analyze this food image and estimate nutrition:
|
||
1. What foods are in the image?
|
||
2. Estimate: calories, carbs (g), protein (g), fat (g)
|
||
3. Keto-friendly? (yes/no)
|
||
|
||
Return JSON format:
|
||
{
|
||
"foods": ["item1", "item2"],
|
||
"calories": number,
|
||
"carbs": number,
|
||
"protein": number,
|
||
"fat": number,
|
||
"keto_friendly": boolean,
|
||
"confidence": 0.0-1.0
|
||
}"""
|
||
|
||
# Call MiniMax API
|
||
headers = {
|
||
"Authorization": f"Bearer {MINIMAX_API_KEY}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
payload = {
|
||
"model": "MiniMax-Vision-01",
|
||
"messages": [
|
||
{
|
||
"role": "user",
|
||
"content": [
|
||
{"type": "text", "text": prompt},
|
||
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_b64}"}}
|
||
]
|
||
}
|
||
],
|
||
"max_tokens": 500,
|
||
"temperature": 0.3
|
||
}
|
||
|
||
import httpx
|
||
async with httpx.AsyncClient() as client:
|
||
response = await client.post(
|
||
MINIMAX_API_URL,
|
||
headers=headers,
|
||
json=payload,
|
||
timeout=30.0
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
# Parse JSON from response
|
||
content = result.get('choices', [{}])[0].get('message', {}).get('content', '{}')
|
||
|
||
# Extract JSON
|
||
import json as json_module
|
||
try:
|
||
# Try to parse the response as JSON
|
||
nutrition = json_module.loads(content)
|
||
return {
|
||
'calories': nutrition.get('calories', 400),
|
||
'carbs': nutrition.get('carbs', 25),
|
||
'protein': nutrition.get('protein', 30),
|
||
'fat': nutrition.get('fat', 20),
|
||
'detected_foods': nutrition.get('foods', ['unknown']),
|
||
'confidence': nutrition.get('confidence', 0.8),
|
||
'keto_friendly': nutrition.get('keto_friendly', True)
|
||
}
|
||
except json_module.JSONDecodeError:
|
||
# Fallback if JSON parsing fails
|
||
return {
|
||
'calories': 400,
|
||
'carbs': 25,
|
||
'protein': 30,
|
||
'fat': 20,
|
||
'detected_foods': ['analyzed via MiniMax'],
|
||
'confidence': 0.7
|
||
}
|
||
else:
|
||
print(f"MiniMax API error: {response.status_code}")
|
||
return {
|
||
'calories': 400,
|
||
'carbs': 25,
|
||
'protein': 30,
|
||
'fat': 20,
|
||
'detected_foods': ['analysis failed - using defaults'],
|
||
'confidence': 0.5
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"Photo analysis error: {e}")
|
||
return {
|
||
'calories': 400,
|
||
'carbs': 25,
|
||
'protein': 30,
|
||
'fat': 20,
|
||
'detected_foods': ['error - using defaults'],
|
||
'confidence': 0.5
|
||
}
|
||
|
||
async def food_photo(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
"""Handle food photo upload and analysis"""
|
||
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'
|
||
|
||
# Get photo
|
||
photo = update.message.photo[-1] if update.message.photo else None
|
||
if not photo:
|
||
await update.message.reply_text("❌ No photo found! Please send a food photo.")
|
||
return
|
||
|
||
await update.message.reply_text("📸 Analyzing food photo...")
|
||
|
||
try:
|
||
# Download photo
|
||
file = await context.bot.get_file(photo.file_id)
|
||
file_path = f"/tmp/food_{user_id}_{today}.jpg"
|
||
await file.download_to_drive(file_path)
|
||
|
||
# Analyze with MiniMax Vision API
|
||
nutrition = await analyze_food_photo(file_path)
|
||
|
||
# Log the food
|
||
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': ', '.join(nutrition.get('detected_foods', ['food'])),
|
||
'time': now,
|
||
'calories': nutrition['calories'],
|
||
'carbs': nutrition['carbs'],
|
||
'protein': nutrition['protein'],
|
||
'fat': nutrition['fat'],
|
||
'source': 'photo',
|
||
'confidence': nutrition.get('confidence', 0.8),
|
||
'timestamp': datetime.datetime.now().isoformat()
|
||
})
|
||
data.save()
|
||
|
||
# Build response
|
||
emoji = "✅" if nutrition.get('keto_friendly', True) else "⚠️"
|
||
confidence_pct = int(nutrition.get('confidence', 0.8) * 100)
|
||
|
||
text = f"🍽️ **Food Analyzed**\n\n"
|
||
text += f"Detected: {', '.join(nutrition.get('detected_foods', ['food']))}\n"
|
||
text += f"Confidence: {confidence_pct}%\n\n"
|
||
text += f"📊 **Nutrition:**\n"
|
||
text += f"🔥 Calories: {nutrition['calories']}kcal\n"
|
||
text += f"🥦 Carbs: {nutrition['carbs']}g\n"
|
||
text += f"💪 Protein: {nutrition['protein']}g\n"
|
||
text += f"🥑 Fat: {nutrition['fat']}g\n\n"
|
||
text += f"{emoji} Keto-friendly: {'Yes' if nutrition.get('keto_friendly', True) else 'No'}\n"
|
||
|
||
# Keto check
|
||
if nutrition['carbs'] > 25:
|
||
text += "\n⚠️ Carbs exceed keto limit (25g)!"
|
||
|
||
# Daily total
|
||
total = data.get_daily_totals(user_id, today)
|
||
text += f"\n📈 **Today's Total:** {total['calories']}kcal"
|
||
text += f"\n💪 {2000 - total['calories']}kcal remaining"
|
||
|
||
await update.message.reply_text(text, parse_mode='Markdown')
|
||
|
||
# Clean up
|
||
import os
|
||
if os.path.exists(file_path):
|
||
os.remove(file_path)
|
||
|
||
except Exception as e:
|
||
await update.message.reply_text(f"❌ Error analyzing photo: {str(e)}")
|
||
|
||
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('food_photo', food_photo))
|
||
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
|
||
|
||
# Photo handler (for food photos)
|
||
from telegram.ext import.filters
|
||
app.add_handler(MessageHandler(filters.PHOTO, food_photo))
|
||
|
||
# URL handler
|
||
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_url))
|
||
|
||
print("🔮 Starting Habit & Diet Bot...")
|
||
app.run_polling()
|
||
|
||
if __name__ == '__main__':
|
||
main()
|