Internal
Public Access
Fixes 500 error when user timezone is invalid or missing. Now gracefully falls back to UTC if timezone conversion fails. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
368 lines
11 KiB
Python
368 lines
11 KiB
Python
"""
|
|
Task models for KeepItGoing.
|
|
"""
|
|
|
|
import uuid
|
|
from django.conf import settings
|
|
from django.db import models
|
|
|
|
|
|
class SoftDeleteManager(models.Manager):
|
|
"""Manager that excludes soft-deleted items by default."""
|
|
|
|
def get_queryset(self):
|
|
return super().get_queryset().filter(is_deleted=False)
|
|
|
|
|
|
class AllObjectsManager(models.Manager):
|
|
"""Manager that includes soft-deleted items."""
|
|
pass
|
|
|
|
|
|
class Tag(models.Model):
|
|
"""
|
|
Tags for organizing and categorizing tasks.
|
|
Supports multiple tags per task.
|
|
"""
|
|
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
|
user = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL,
|
|
on_delete=models.CASCADE,
|
|
related_name='tags'
|
|
)
|
|
name = models.CharField(max_length=255)
|
|
description = models.TextField(blank=True)
|
|
color = models.CharField(max_length=7, default='#3B82F6') # Hex color
|
|
icon = models.CharField(max_length=50, blank=True) # Icon name
|
|
sort_order = models.IntegerField(default=0)
|
|
is_archived = models.BooleanField(default=False)
|
|
|
|
# Sync fields
|
|
sync_id = models.UUIDField(default=uuid.uuid4, unique=True)
|
|
is_deleted = models.BooleanField(default=False)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
# Managers
|
|
objects = SoftDeleteManager()
|
|
all_objects = AllObjectsManager()
|
|
|
|
class Meta:
|
|
db_table = 'tags'
|
|
ordering = ['sort_order', 'name']
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
|
|
class Task(models.Model):
|
|
"""
|
|
Main task model with support for subtasks, recurrence, and time tracking.
|
|
"""
|
|
STATUS_CHOICES = [
|
|
('pending', 'Pending'),
|
|
('in_progress', 'In Progress'),
|
|
('completed', 'Completed'),
|
|
('cancelled', 'Cancelled'),
|
|
]
|
|
|
|
PRIORITY_CHOICES = [
|
|
('low', 'Low'),
|
|
('medium', 'Medium'),
|
|
('high', 'High'),
|
|
('urgent', 'Urgent'),
|
|
]
|
|
|
|
RECURRENCE_CHOICES = [
|
|
('none', 'None'),
|
|
('daily', 'Daily'),
|
|
('weekly', 'Weekly'),
|
|
('biweekly', 'Bi-weekly'),
|
|
('monthly', 'Monthly'),
|
|
('yearly', 'Yearly'),
|
|
('custom', 'Custom'),
|
|
]
|
|
|
|
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
|
user = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL,
|
|
on_delete=models.CASCADE,
|
|
related_name='tasks'
|
|
)
|
|
|
|
# Hierarchy
|
|
parent = models.ForeignKey(
|
|
'self',
|
|
on_delete=models.CASCADE,
|
|
null=True,
|
|
blank=True,
|
|
related_name='subtasks'
|
|
)
|
|
|
|
# Core fields
|
|
title = models.CharField(max_length=500)
|
|
description = models.TextField(blank=True)
|
|
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending')
|
|
priority = models.CharField(max_length=20, choices=PRIORITY_CHOICES, default='medium')
|
|
|
|
# Dates and times
|
|
due_date = models.DateField(null=True, blank=True)
|
|
due_time = models.TimeField(null=True, blank=True)
|
|
reminder_at = models.DateTimeField(null=True, blank=True)
|
|
completed_at = models.DateTimeField(null=True, blank=True)
|
|
|
|
# Recurrence (RRULE format for custom)
|
|
recurrence = models.CharField(max_length=20, choices=RECURRENCE_CHOICES, default='none')
|
|
recurrence_rule = models.CharField(max_length=500, blank=True) # RRULE string
|
|
recurrence_end_date = models.DateField(null=True, blank=True)
|
|
|
|
# Tags (multiple per task)
|
|
tags = models.ManyToManyField(Tag, blank=True, related_name='tasks')
|
|
|
|
# Ordering
|
|
sort_order = models.IntegerField(default=0)
|
|
|
|
# Sync fields
|
|
sync_id = models.UUIDField(default=uuid.uuid4, unique=True)
|
|
is_deleted = models.BooleanField(default=False)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
# Managers
|
|
objects = SoftDeleteManager()
|
|
all_objects = AllObjectsManager()
|
|
|
|
class Meta:
|
|
db_table = 'tasks'
|
|
ordering = ['sort_order', '-priority', 'due_date', 'created_at']
|
|
|
|
def __str__(self):
|
|
return self.title
|
|
|
|
def save(self, *args, **kwargs):
|
|
# Auto-set completed_at when status changes to completed
|
|
if self.status == 'completed' and self.completed_at is None:
|
|
from django.utils import timezone
|
|
self.completed_at = timezone.now()
|
|
elif self.status != 'completed':
|
|
self.completed_at = None
|
|
super().save(*args, **kwargs)
|
|
|
|
@property
|
|
def is_overdue(self):
|
|
"""Check if task is overdue."""
|
|
if self.due_date and self.status not in ('completed', 'cancelled'):
|
|
from django.utils import timezone
|
|
import pytz
|
|
|
|
# Get current date in user's timezone
|
|
try:
|
|
user_tz = pytz.timezone(self.user.timezone)
|
|
user_now = timezone.now().astimezone(user_tz)
|
|
user_today = user_now.date()
|
|
except (pytz.UnknownTimeZoneError, AttributeError):
|
|
# Fall back to UTC if user timezone is invalid or missing
|
|
user_today = timezone.now().date()
|
|
|
|
return self.due_date < user_today
|
|
return False
|
|
|
|
@property
|
|
def total_time_spent(self):
|
|
"""Calculate total time spent on this task in seconds."""
|
|
total_seconds = sum(
|
|
entry.duration_seconds or 0
|
|
for entry in self.time_entries.all()
|
|
)
|
|
return total_seconds
|
|
|
|
@property
|
|
def total_time_formatted(self):
|
|
"""Return total time spent formatted as H:MM:SS."""
|
|
seconds = self.total_time_spent
|
|
hours = seconds // 3600
|
|
minutes = (seconds % 3600) // 60
|
|
secs = seconds % 60
|
|
return f"{hours}:{minutes:02d}:{secs:02d}"
|
|
|
|
def calculate_next_due_date(self, from_date=None):
|
|
"""
|
|
Calculate the next due date based on recurrence pattern.
|
|
Returns None if task doesn't recur or has reached end date.
|
|
"""
|
|
if self.recurrence == 'none':
|
|
return None
|
|
|
|
from datetime import timedelta
|
|
from dateutil.relativedelta import relativedelta
|
|
|
|
base_date = from_date or self.due_date
|
|
if not base_date:
|
|
from django.utils import timezone
|
|
base_date = timezone.now().date()
|
|
|
|
# Calculate next date based on recurrence type
|
|
if self.recurrence == 'daily':
|
|
next_date = base_date + timedelta(days=1)
|
|
elif self.recurrence == 'weekly':
|
|
next_date = base_date + timedelta(weeks=1)
|
|
elif self.recurrence == 'biweekly':
|
|
next_date = base_date + timedelta(weeks=2)
|
|
elif self.recurrence == 'monthly':
|
|
next_date = base_date + relativedelta(months=1)
|
|
elif self.recurrence == 'yearly':
|
|
next_date = base_date + relativedelta(years=1)
|
|
elif self.recurrence == 'custom' and self.recurrence_rule:
|
|
# TODO: Implement RRULE parsing for custom recurrence
|
|
# For now, default to weekly
|
|
next_date = base_date + timedelta(weeks=1)
|
|
else:
|
|
return None
|
|
|
|
# Check if we've passed the end date
|
|
if self.recurrence_end_date and next_date > self.recurrence_end_date:
|
|
return None
|
|
|
|
return next_date
|
|
|
|
def create_next_recurrence(self):
|
|
"""
|
|
Create the next instance of this recurring task.
|
|
Returns the new task or None if no recurrence should be created.
|
|
"""
|
|
if self.recurrence == 'none':
|
|
return None
|
|
|
|
next_due_date = self.calculate_next_due_date()
|
|
if not next_due_date:
|
|
return None
|
|
|
|
# Create new task with same properties
|
|
new_task = Task.objects.create(
|
|
user=self.user,
|
|
parent=self.parent,
|
|
title=self.title,
|
|
description=self.description,
|
|
status='pending',
|
|
priority=self.priority,
|
|
due_date=next_due_date,
|
|
due_time=self.due_time,
|
|
recurrence=self.recurrence,
|
|
recurrence_rule=self.recurrence_rule,
|
|
recurrence_end_date=self.recurrence_end_date,
|
|
sort_order=self.sort_order,
|
|
)
|
|
|
|
# Copy tags
|
|
new_task.tags.set(self.tags.all())
|
|
|
|
return new_task
|
|
|
|
|
|
class TimeEntry(models.Model):
|
|
"""
|
|
Time tracking entries for tasks.
|
|
"""
|
|
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
|
task = models.ForeignKey(Task, on_delete=models.CASCADE, related_name='time_entries')
|
|
user = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL,
|
|
on_delete=models.CASCADE,
|
|
related_name='time_entries'
|
|
)
|
|
|
|
started_at = models.DateTimeField()
|
|
ended_at = models.DateTimeField(null=True, blank=True)
|
|
duration_seconds = models.IntegerField(null=True, blank=True)
|
|
notes = models.TextField(blank=True)
|
|
|
|
# Sync fields
|
|
sync_id = models.UUIDField(default=uuid.uuid4, unique=True)
|
|
is_deleted = models.BooleanField(default=False)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
# Managers
|
|
objects = SoftDeleteManager()
|
|
all_objects = AllObjectsManager()
|
|
|
|
class Meta:
|
|
db_table = 'time_entries'
|
|
ordering = ['-started_at']
|
|
|
|
def __str__(self):
|
|
return f"{self.task.title} - {self.started_at}"
|
|
|
|
def save(self, *args, **kwargs):
|
|
# Calculate duration if ended
|
|
if self.ended_at and self.started_at:
|
|
delta = self.ended_at - self.started_at
|
|
self.duration_seconds = int(delta.total_seconds())
|
|
super().save(*args, **kwargs)
|
|
|
|
@property
|
|
def is_running(self):
|
|
"""Check if this time entry is currently running."""
|
|
return self.ended_at is None
|
|
|
|
|
|
class TaskShare(models.Model):
|
|
"""
|
|
Sharing tasks or tags with other users.
|
|
"""
|
|
PERMISSION_CHOICES = [
|
|
('viewer', 'Viewer'),
|
|
('editor', 'Editor'),
|
|
]
|
|
|
|
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
|
owner = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL,
|
|
on_delete=models.CASCADE,
|
|
related_name='shares_given'
|
|
)
|
|
shared_with = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL,
|
|
on_delete=models.CASCADE,
|
|
related_name='shares_received'
|
|
)
|
|
|
|
# Can share either a task or a tag
|
|
task = models.ForeignKey(
|
|
Task,
|
|
on_delete=models.CASCADE,
|
|
null=True,
|
|
blank=True,
|
|
related_name='shares'
|
|
)
|
|
tag = models.ForeignKey(
|
|
Tag,
|
|
on_delete=models.CASCADE,
|
|
null=True,
|
|
blank=True,
|
|
related_name='shares'
|
|
)
|
|
|
|
permission = models.CharField(max_length=20, choices=PERMISSION_CHOICES, default='viewer')
|
|
|
|
# Sync fields
|
|
sync_id = models.UUIDField(default=uuid.uuid4, unique=True)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
class Meta:
|
|
db_table = 'task_shares'
|
|
constraints = [
|
|
models.CheckConstraint(
|
|
check=(
|
|
models.Q(task__isnull=False, tag__isnull=True) |
|
|
models.Q(task__isnull=True, tag__isnull=False)
|
|
),
|
|
name='share_task_or_tag_not_both'
|
|
)
|
|
]
|
|
|
|
def __str__(self):
|
|
shared_item = self.task or self.tag
|
|
return f"{shared_item} shared with {self.shared_with.email}"
|