392 lines
15 KiB
Python
392 lines
15 KiB
Python
"""Habit tracking endpoints (CRUD + check / skip / uncheck)."""
|
|
import json
|
|
import re
|
|
import uuid
|
|
from datetime import datetime
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
import constants
|
|
import habits_helpers
|
|
|
|
|
|
def _enrich(habit):
|
|
"""Return habit with calculated stats added."""
|
|
enriched = habit.copy()
|
|
enriched['current_streak'] = habits_helpers.calculate_streak(habit)
|
|
enriched['best_streak'] = habit.get('streak', {}).get('best', 0)
|
|
enriched['completion_rate_30d'] = habits_helpers.get_completion_rate(habit, days=30)
|
|
enriched['weekly_summary'] = habits_helpers.get_weekly_summary(habit)
|
|
enriched['should_check_today'] = habits_helpers.should_check_today(habit)
|
|
return enriched
|
|
|
|
|
|
class HabitsHandlers:
|
|
"""Mixin providing /api/habits endpoints."""
|
|
|
|
def handle_habits_get(self):
|
|
"""Return all habits with enriched stats."""
|
|
try:
|
|
if not constants.HABITS_FILE.exists():
|
|
self.send_json([])
|
|
return
|
|
|
|
with open(constants.HABITS_FILE, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
|
|
enriched = [_enrich(h) for h in data.get('habits', [])]
|
|
enriched.sort(key=lambda h: h.get('priority', 999))
|
|
self.send_json(enriched)
|
|
except Exception as e:
|
|
self.send_json({'error': str(e)}, 500)
|
|
|
|
def handle_habits_post(self):
|
|
"""Create a new habit."""
|
|
try:
|
|
content_length = int(self.headers['Content-Length'])
|
|
post_data = self.rfile.read(content_length).decode('utf-8')
|
|
data = json.loads(post_data)
|
|
|
|
name = data.get('name', '').strip()
|
|
if not name:
|
|
self.send_json({'error': 'name is required'}, 400)
|
|
return
|
|
if len(name) > 100:
|
|
self.send_json({'error': 'name must be max 100 characters'}, 400)
|
|
return
|
|
|
|
color = data.get('color', '#3b82f6')
|
|
if color and not re.match(r'^#[0-9A-Fa-f]{6}$', color):
|
|
self.send_json({'error': 'color must be valid hex format (#RRGGBB)'}, 400)
|
|
return
|
|
|
|
frequency_type = data.get('frequency', {}).get('type', 'daily')
|
|
valid_types = ['daily', 'specific_days', 'x_per_week', 'weekly', 'monthly', 'custom']
|
|
if frequency_type not in valid_types:
|
|
self.send_json({'error': f'frequency.type must be one of: {", ".join(valid_types)}'}, 400)
|
|
return
|
|
|
|
habit_id = str(uuid.uuid4())
|
|
now = datetime.now().isoformat()
|
|
|
|
new_habit = {
|
|
'id': habit_id,
|
|
'name': name,
|
|
'category': data.get('category', 'other'),
|
|
'color': color,
|
|
'icon': data.get('icon', 'check-circle'),
|
|
'priority': data.get('priority', 5),
|
|
'notes': data.get('notes', ''),
|
|
'reminderTime': data.get('reminderTime', ''),
|
|
'frequency': data.get('frequency', {'type': 'daily'}),
|
|
'streak': {'current': 0, 'best': 0, 'lastCheckIn': None},
|
|
'lives': 3,
|
|
'completions': [],
|
|
'createdAt': now,
|
|
'updatedAt': now,
|
|
}
|
|
|
|
if constants.HABITS_FILE.exists():
|
|
with open(constants.HABITS_FILE, 'r', encoding='utf-8') as f:
|
|
habits_data = json.load(f)
|
|
else:
|
|
habits_data = {'lastUpdated': '', 'habits': []}
|
|
|
|
habits_data['habits'].append(new_habit)
|
|
habits_data['lastUpdated'] = now
|
|
|
|
with open(constants.HABITS_FILE, 'w', encoding='utf-8') as f:
|
|
json.dump(habits_data, f, indent=2)
|
|
|
|
self.send_json(new_habit, 201)
|
|
except json.JSONDecodeError:
|
|
self.send_json({'error': 'Invalid JSON'}, 400)
|
|
except Exception as e:
|
|
self.send_json({'error': str(e)}, 500)
|
|
|
|
def handle_habits_put(self):
|
|
"""Update an existing habit."""
|
|
try:
|
|
path_parts = self.path.split('/')
|
|
if len(path_parts) < 4:
|
|
self.send_json({'error': 'Invalid path'}, 400)
|
|
return
|
|
habit_id = path_parts[3]
|
|
|
|
content_length = int(self.headers['Content-Length'])
|
|
post_data = self.rfile.read(content_length).decode('utf-8')
|
|
data = json.loads(post_data)
|
|
|
|
if not constants.HABITS_FILE.exists():
|
|
self.send_json({'error': 'Habit not found'}, 404)
|
|
return
|
|
|
|
with open(constants.HABITS_FILE, 'r', encoding='utf-8') as f:
|
|
habits_data = json.load(f)
|
|
|
|
habits = habits_data.get('habits', [])
|
|
habit_index = next((i for i, h in enumerate(habits) if h['id'] == habit_id), None)
|
|
if habit_index is None:
|
|
self.send_json({'error': 'Habit not found'}, 404)
|
|
return
|
|
|
|
if 'name' in data:
|
|
name = data['name'].strip()
|
|
if not name:
|
|
self.send_json({'error': 'name cannot be empty'}, 400)
|
|
return
|
|
if len(name) > 100:
|
|
self.send_json({'error': 'name must be max 100 characters'}, 400)
|
|
return
|
|
if 'color' in data:
|
|
color = data['color']
|
|
if color and not re.match(r'^#[0-9A-Fa-f]{6}$', color):
|
|
self.send_json({'error': 'color must be valid hex format (#RRGGBB)'}, 400)
|
|
return
|
|
if 'frequency' in data:
|
|
frequency_type = data.get('frequency', {}).get('type', 'daily')
|
|
valid_types = ['daily', 'specific_days', 'x_per_week', 'weekly', 'monthly', 'custom']
|
|
if frequency_type not in valid_types:
|
|
self.send_json({'error': f'frequency.type must be one of: {", ".join(valid_types)}'}, 400)
|
|
return
|
|
|
|
allowed_fields = ['name', 'category', 'color', 'icon', 'priority', 'notes', 'frequency', 'reminderTime']
|
|
habit = habits[habit_index]
|
|
for field in allowed_fields:
|
|
if field in data:
|
|
habit[field] = data[field]
|
|
|
|
habit['updatedAt'] = datetime.now().isoformat()
|
|
habits_data['lastUpdated'] = habit['updatedAt']
|
|
with open(constants.HABITS_FILE, 'w', encoding='utf-8') as f:
|
|
json.dump(habits_data, f, indent=2)
|
|
|
|
self.send_json(habit)
|
|
except json.JSONDecodeError:
|
|
self.send_json({'error': 'Invalid JSON'}, 400)
|
|
except Exception as e:
|
|
self.send_json({'error': str(e)}, 500)
|
|
|
|
def handle_habits_delete(self):
|
|
"""Delete a habit."""
|
|
try:
|
|
path_parts = self.path.split('/')
|
|
if len(path_parts) < 4:
|
|
self.send_json({'error': 'Invalid path'}, 400)
|
|
return
|
|
habit_id = path_parts[3]
|
|
|
|
if not constants.HABITS_FILE.exists():
|
|
self.send_json({'error': 'Habit not found'}, 404)
|
|
return
|
|
|
|
with open(constants.HABITS_FILE, 'r', encoding='utf-8') as f:
|
|
habits_data = json.load(f)
|
|
|
|
habits = habits_data.get('habits', [])
|
|
habit_found = False
|
|
for i, habit in enumerate(habits):
|
|
if habit['id'] == habit_id:
|
|
habits.pop(i)
|
|
habit_found = True
|
|
break
|
|
|
|
if not habit_found:
|
|
self.send_json({'error': 'Habit not found'}, 404)
|
|
return
|
|
|
|
habits_data['lastUpdated'] = datetime.now().isoformat()
|
|
with open(constants.HABITS_FILE, 'w', encoding='utf-8') as f:
|
|
json.dump(habits_data, f, indent=2)
|
|
|
|
self.send_response(204)
|
|
self.send_header('Access-Control-Allow-Origin', '*')
|
|
self.end_headers()
|
|
except Exception as e:
|
|
self.send_json({'error': str(e)}, 500)
|
|
|
|
def handle_habits_check(self):
|
|
"""Check in on a habit for today."""
|
|
try:
|
|
path_parts = self.path.split('/')
|
|
if len(path_parts) < 5:
|
|
self.send_json({'error': 'Invalid path'}, 400)
|
|
return
|
|
habit_id = path_parts[3]
|
|
|
|
body_data = {}
|
|
content_length = self.headers.get('Content-Length')
|
|
if content_length:
|
|
post_data = self.rfile.read(int(content_length)).decode('utf-8')
|
|
if post_data.strip():
|
|
try:
|
|
body_data = json.loads(post_data)
|
|
except json.JSONDecodeError:
|
|
self.send_json({'error': 'Invalid JSON'}, 400)
|
|
return
|
|
|
|
if not constants.HABITS_FILE.exists():
|
|
self.send_json({'error': 'Habit not found'}, 404)
|
|
return
|
|
|
|
with open(constants.HABITS_FILE, 'r', encoding='utf-8') as f:
|
|
habits_data = json.load(f)
|
|
|
|
habit = next((h for h in habits_data.get('habits', []) if h['id'] == habit_id), None)
|
|
if not habit:
|
|
self.send_json({'error': 'Habit not found'}, 404)
|
|
return
|
|
|
|
if not habits_helpers.should_check_today(habit):
|
|
self.send_json({'error': 'Habit is not relevant for today based on its frequency'}, 400)
|
|
return
|
|
|
|
today = datetime.now().date().isoformat()
|
|
for completion in habit.get('completions', []):
|
|
if completion.get('date') == today:
|
|
self.send_json({'error': 'Habit already checked in today'}, 409)
|
|
return
|
|
|
|
completion_entry = {'date': today, 'type': 'check'}
|
|
if 'note' in body_data:
|
|
completion_entry['note'] = body_data['note']
|
|
if 'rating' in body_data:
|
|
rating = body_data['rating']
|
|
if not isinstance(rating, int) or rating < 1 or rating > 5:
|
|
self.send_json({'error': 'rating must be an integer between 1 and 5'}, 400)
|
|
return
|
|
completion_entry['rating'] = rating
|
|
if 'mood' in body_data:
|
|
mood = body_data['mood']
|
|
if mood not in ['happy', 'neutral', 'sad']:
|
|
self.send_json({'error': 'mood must be one of: happy, neutral, sad'}, 400)
|
|
return
|
|
completion_entry['mood'] = mood
|
|
|
|
habit['completions'].append(completion_entry)
|
|
|
|
current_streak = habits_helpers.calculate_streak(habit)
|
|
habit['streak']['current'] = current_streak
|
|
if current_streak > habit['streak']['best']:
|
|
habit['streak']['best'] = current_streak
|
|
habit['streak']['lastCheckIn'] = today
|
|
|
|
new_lives, was_awarded = habits_helpers.check_and_award_weekly_lives(habit)
|
|
lives_awarded_this_checkin = False
|
|
if was_awarded:
|
|
habit['lives'] = new_lives
|
|
habit['lastLivesAward'] = today
|
|
lives_awarded_this_checkin = True
|
|
|
|
habit['updatedAt'] = datetime.now().isoformat()
|
|
habits_data['lastUpdated'] = habit['updatedAt']
|
|
|
|
with open(constants.HABITS_FILE, 'w', encoding='utf-8') as f:
|
|
json.dump(habits_data, f, indent=2)
|
|
|
|
enriched = _enrich(habit)
|
|
enriched['livesAwarded'] = lives_awarded_this_checkin
|
|
self.send_json(enriched, 200)
|
|
except Exception as e:
|
|
self.send_json({'error': str(e)}, 500)
|
|
|
|
def handle_habits_uncheck(self):
|
|
"""Remove a habit completion for a specific date."""
|
|
try:
|
|
path_parts = self.path.split('?')[0].split('/')
|
|
if len(path_parts) < 5:
|
|
self.send_json({'error': 'Invalid path'}, 400)
|
|
return
|
|
habit_id = path_parts[3]
|
|
|
|
query_params = parse_qs(urlparse(self.path).query)
|
|
if 'date' not in query_params:
|
|
self.send_json({'error': 'date parameter is required (format: YYYY-MM-DD)'}, 400)
|
|
return
|
|
|
|
target_date = query_params['date'][0]
|
|
try:
|
|
datetime.fromisoformat(target_date)
|
|
except ValueError:
|
|
self.send_json({'error': 'Invalid date format. Use YYYY-MM-DD'}, 400)
|
|
return
|
|
|
|
if not constants.HABITS_FILE.exists():
|
|
self.send_json({'error': 'Habit not found'}, 404)
|
|
return
|
|
|
|
with open(constants.HABITS_FILE, 'r', encoding='utf-8') as f:
|
|
habits_data = json.load(f)
|
|
|
|
habit = next((h for h in habits_data.get('habits', []) if h['id'] == habit_id), None)
|
|
if not habit:
|
|
self.send_json({'error': 'Habit not found'}, 404)
|
|
return
|
|
|
|
completions = habit.get('completions', [])
|
|
completion_found = False
|
|
for i, completion in enumerate(completions):
|
|
if completion.get('date') == target_date:
|
|
completions.pop(i)
|
|
completion_found = True
|
|
break
|
|
|
|
if not completion_found:
|
|
self.send_json({'error': 'No completion found for the specified date'}, 404)
|
|
return
|
|
|
|
current_streak = habits_helpers.calculate_streak(habit)
|
|
habit['streak']['current'] = current_streak
|
|
if current_streak > habit['streak']['best']:
|
|
habit['streak']['best'] = current_streak
|
|
|
|
habit['updatedAt'] = datetime.now().isoformat()
|
|
habits_data['lastUpdated'] = habit['updatedAt']
|
|
|
|
with open(constants.HABITS_FILE, 'w', encoding='utf-8') as f:
|
|
json.dump(habits_data, f, indent=2)
|
|
|
|
self.send_json(_enrich(habit), 200)
|
|
except Exception as e:
|
|
self.send_json({'error': str(e)}, 500)
|
|
|
|
def handle_habits_skip(self):
|
|
"""Skip a day using a life to preserve streak."""
|
|
try:
|
|
path_parts = self.path.split('/')
|
|
if len(path_parts) < 5:
|
|
self.send_json({'error': 'Invalid path'}, 400)
|
|
return
|
|
habit_id = path_parts[3]
|
|
|
|
if not constants.HABITS_FILE.exists():
|
|
self.send_json({'error': 'Habit not found'}, 404)
|
|
return
|
|
|
|
with open(constants.HABITS_FILE, 'r', encoding='utf-8') as f:
|
|
habits_data = json.load(f)
|
|
|
|
habit = next((h for h in habits_data.get('habits', []) if h['id'] == habit_id), None)
|
|
if not habit:
|
|
self.send_json({'error': 'Habit not found'}, 404)
|
|
return
|
|
|
|
current_lives = habit.get('lives', 3)
|
|
if current_lives <= 0:
|
|
self.send_json({'error': 'No lives remaining'}, 400)
|
|
return
|
|
|
|
habit['lives'] = current_lives - 1
|
|
|
|
today = datetime.now().date().isoformat()
|
|
habit['completions'].append({'date': today, 'type': 'skip'})
|
|
|
|
habit['updatedAt'] = datetime.now().isoformat()
|
|
habits_data['lastUpdated'] = habit['updatedAt']
|
|
|
|
with open(constants.HABITS_FILE, 'w', encoding='utf-8') as f:
|
|
json.dump(habits_data, f, indent=2)
|
|
|
|
self.send_json(_enrich(habit), 200)
|
|
except Exception as e:
|
|
self.send_json({'error': str(e)}, 500)
|