227 lines
No EOL
9 KiB
Python
227 lines
No EOL
9 KiB
Python
from flask import Flask, render_template, request, redirect, url_for, jsonify
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from flask_socketio import SocketIO, emit
|
|
from datetime import datetime, timedelta
|
|
|
|
app = Flask(__name__)
|
|
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///tasks.db'
|
|
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
|
|
db = SQLAlchemy(app)
|
|
socketio = SocketIO(app)
|
|
|
|
# Database models
|
|
class Task(db.Model):
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(100), nullable=False)
|
|
priority = db.Column(db.String(1), nullable=False) # A, B, C, or D
|
|
duration = db.Column(db.Integer, nullable=False) # Duration in minutes
|
|
status = db.Column(db.String(10), default='pending')
|
|
due_date = db.Column(db.Date, nullable=True) # Due date for the task
|
|
completed_at = db.Column(db.DateTime, nullable=True) # Timestamp for when the task was completed
|
|
start_time = db.Column(db.Time, nullable=True) # Start time for the task
|
|
|
|
def format_duration(minutes):
|
|
"""Convert duration in minutes to 'H M' format."""
|
|
hours = minutes // 60
|
|
remaining_minutes = minutes % 60
|
|
if hours > 0:
|
|
return f"{hours}H {remaining_minutes}M" if remaining_minutes > 0 else f"{hours}H"
|
|
return f"{remaining_minutes}M"
|
|
|
|
@app.context_processor
|
|
def utility_processor():
|
|
return dict(format_duration=format_duration)
|
|
|
|
def cleanup_completed_tasks():
|
|
"""Remove tasks that have been completed for more than 1 hour."""
|
|
expiration_time = datetime.utcnow() - timedelta(hours=1)
|
|
old_tasks = Task.query.filter(Task.status == 'completed', Task.completed_at < expiration_time).all()
|
|
for task in old_tasks:
|
|
db.session.delete(task)
|
|
db.session.commit()
|
|
|
|
# Routes
|
|
@app.route('/')
|
|
def index():
|
|
cleanup_completed_tasks() # Remove old completed tasks
|
|
|
|
# Get the sorting parameter from the query string (default is "priority")
|
|
sort_by = request.args.get('sort_by', 'priority')
|
|
|
|
# Dynamically sort tasks based on the parameter
|
|
if sort_by == 'priority':
|
|
in_progress_tasks = Task.query.filter(Task.status == 'in_progress').order_by(Task.priority).all()
|
|
pending_tasks = Task.query.filter(Task.status == 'pending').order_by(Task.priority).all()
|
|
elif sort_by == 'due_date':
|
|
in_progress_tasks = Task.query.filter(Task.status == 'in_progress').order_by(Task.due_date).all()
|
|
pending_tasks = Task.query.filter(Task.status == 'pending').order_by(Task.due_date).all()
|
|
elif sort_by == 'duration':
|
|
in_progress_tasks = Task.query.filter(Task.status == 'in_progress').order_by(Task.duration).all()
|
|
pending_tasks = Task.query.filter(Task.status == 'pending').order_by(Task.duration).all()
|
|
else:
|
|
in_progress_tasks = Task.query.filter(Task.status == 'in_progress').all() # Default: no sorting
|
|
pending_tasks = Task.query.filter(Task.status == 'pending').all()
|
|
|
|
return render_template('index.html', in_progress_tasks=in_progress_tasks, pending_tasks=pending_tasks, sort_by=sort_by)
|
|
|
|
@app.route('/add_task', methods=['POST'])
|
|
def add_task():
|
|
name = request.form['name']
|
|
priority = request.form['priority']
|
|
duration_input = request.form['duration']
|
|
due_date_input = request.form['due_date']
|
|
|
|
# Parse duration (e.g., "10M" -> 10 minutes, "2H" -> 120 minutes)
|
|
if duration_input.endswith('M'):
|
|
duration = int(duration_input[:-1]) # Extract the number before 'M'
|
|
elif duration_input.endswith('H'):
|
|
duration = int(duration_input[:-1]) * 60 # Convert hours to minutes
|
|
else:
|
|
return "Invalid duration format. Use 'M' for minutes or 'H' for hours.", 400
|
|
|
|
# Parse due date (e.g., "2025-03-30")
|
|
try:
|
|
due_date = datetime.strptime(due_date_input, '%Y-%m-%d').date() if due_date_input else None
|
|
except ValueError:
|
|
return "Invalid due date format. Use 'YYYY-MM-DD'.", 400
|
|
|
|
# Create and save the new task
|
|
new_task = Task(name=name, priority=priority, duration=duration, due_date=due_date)
|
|
db.session.add(new_task)
|
|
db.session.commit()
|
|
|
|
# Emit the new task to connected clients
|
|
socketio.emit('task_update', {
|
|
'action': 'add',
|
|
'task': {
|
|
'id': new_task.id,
|
|
'name': name,
|
|
'priority': priority,
|
|
'duration': duration,
|
|
'due_date': due_date_input,
|
|
'status': 'pending'
|
|
}
|
|
})
|
|
return redirect(url_for('index'))
|
|
|
|
@app.route('/complete_task/<int:task_id>')
|
|
def complete_task(task_id):
|
|
task = Task.query.get(task_id)
|
|
if task:
|
|
task.status = 'completed'
|
|
task.completed_at = datetime.utcnow() # Store the timestamp in UTC
|
|
db.session.commit()
|
|
# Emit a real-time update to notify all clients
|
|
socketio.emit('task_update', {
|
|
'action': 'complete',
|
|
'task_id': task_id,
|
|
'status': 'completed'
|
|
})
|
|
return redirect(url_for('index'))
|
|
|
|
@app.route('/start_task/<int:task_id>')
|
|
def start_task(task_id):
|
|
task = Task.query.get(task_id)
|
|
if task:
|
|
task.status = 'in_progress'
|
|
db.session.commit()
|
|
socketio.emit('task_update', {'action': 'start', 'task_id': task_id})
|
|
return redirect(url_for('index'))
|
|
|
|
@app.route('/move_to_pending/<int:task_id>')
|
|
def move_to_pending(task_id):
|
|
task = Task.query.get(task_id)
|
|
if task and task.status == 'in_progress': # Ensure the task is currently "in_progress"
|
|
task.status = 'pending'
|
|
db.session.commit()
|
|
socketio.emit('task_update', {'action': 'move_to_pending', 'task_id': task_id})
|
|
return redirect(url_for('index'))
|
|
|
|
@app.route('/move_to_pending_from_completed/<int:task_id>')
|
|
def move_to_pending_from_completed(task_id):
|
|
task = Task.query.get(task_id)
|
|
if task and task.status == 'completed': # Ensure the task is currently "completed"
|
|
task.status = 'pending'
|
|
task.completed_at = None # Clear the completed timestamp
|
|
db.session.commit()
|
|
# Emit a real-time update to all connected clients
|
|
socketio.emit('task_update', {
|
|
'action': 'move_to_pending',
|
|
'task_id': task_id,
|
|
'status': 'pending'
|
|
})
|
|
return redirect(url_for('completed_tasks'))
|
|
|
|
@app.route('/completed_tasks')
|
|
def completed_tasks():
|
|
cleanup_completed_tasks() # Remove old completed tasks
|
|
|
|
# Get the sorting parameter from the query string (default is "completed_at")
|
|
sort_by = request.args.get('sort_by', 'completed_at')
|
|
|
|
# Dynamically sort tasks based on the parameter
|
|
if sort_by == 'priority':
|
|
tasks = Task.query.filter(Task.status == 'completed').order_by(Task.priority).all()
|
|
elif sort_by == 'due_date':
|
|
tasks = Task.query.filter(Task.status == 'completed').order_by(Task.due_date).all()
|
|
elif sort_by == 'duration':
|
|
tasks = Task.query.filter(Task.status == 'completed').order_by(Task.duration).all()
|
|
elif sort_by == 'completed_at':
|
|
tasks = Task.query.filter(Task.status == 'completed').order_by(Task.completed_at.desc()).all()
|
|
else:
|
|
tasks = Task.query.filter(Task.status == 'completed').all() # Default: no sorting
|
|
|
|
return render_template('completed_tasks.html', tasks=tasks, sort_by=sort_by)
|
|
|
|
@app.route('/edit_task/<int:task_id>', methods=['GET', 'POST'])
|
|
def edit_task(task_id):
|
|
task = Task.query.get(task_id)
|
|
if not task:
|
|
return "Task not found", 404
|
|
|
|
if request.method == 'POST':
|
|
# Update task details from the form
|
|
task.name = request.form['name']
|
|
task.priority = request.form['priority']
|
|
duration_input = request.form['duration']
|
|
due_date_input = request.form['due_date']
|
|
|
|
# Parse duration (e.g., "10M" -> 10 minutes, "2H" -> 120 minutes)
|
|
if duration_input.endswith('M'):
|
|
task.duration = int(duration_input[:-1]) # Extract the number before 'M'
|
|
elif duration_input.endswith('H'):
|
|
task.duration = int(duration_input[:-1]) * 60 # Convert hours to minutes
|
|
else:
|
|
return "Invalid duration format. Use 'M' for minutes or 'H' for hours.", 400
|
|
|
|
# Parse due date (e.g., "2025-03-30")
|
|
try:
|
|
task.due_date = datetime.strptime(due_date_input, '%Y-%m-%d').date() if due_date_input else None
|
|
except ValueError:
|
|
return "Invalid due date format. Use 'YYYY-MM-DD'.", 400
|
|
|
|
db.session.commit()
|
|
return redirect(url_for('index'))
|
|
|
|
return render_template('edit_task.html', task=task)
|
|
|
|
@app.route('/update_task_order', methods=['POST'])
|
|
def update_task_order():
|
|
data = request.get_json()
|
|
if not data or 'order' not in data:
|
|
return jsonify({'error': 'Invalid data'}), 400
|
|
|
|
# Update the order of tasks in the database (optional: add an "order" field to the Task model)
|
|
for index, task_id in enumerate(data['order']):
|
|
task = Task.query.get(task_id)
|
|
if task:
|
|
task.order = index # Assuming you add an "order" column to the Task model
|
|
db.session.commit()
|
|
|
|
return jsonify({'success': True})
|
|
|
|
if __name__ == '__main__':
|
|
with app.app_context():
|
|
# Create database tables within the app context
|
|
db.create_all()
|
|
socketio.run(app, debug=True) |