implement task notification system with scheduled emails for upcoming tasks

Author: yassinezorgui

__pycache__/app.cpython-312.pyc

@@ -3,6 +3,8 @@
 from models import db, User
 from werkzeug.security import generate_password_hash
 from config import Config
+from notification import setup_mail
+from scheduler import init_scheduler
 
 def create_app():
     """Application factory function"""
@@ -12,6 +14,8 @@ def create_app():
 
     # Initialize extensions
     db.init_app(app)
+    init_scheduler(app)
+    setup_mail(app)
 
     # Setup Flask-Login
     login_manager = LoginManager()
@@ -36,7 +40,8 @@ def load_user(user_id):
             admin_user = User(
                 username='admin',
                 password=generate_password_hash('admin_password'),
-                role='admin'
+                role='admin',
+                email='yassine16kata@gmail.com',
             )
             db.session.add(admin_user)
             db.session.commit()

__pycache__/models.cpython-312.pyc

@@ -1,6 +1,8 @@
 from flask_login import UserMixin
 from flask_sqlalchemy import SQLAlchemy
 from datetime import datetime, timedelta
+from dateutil.relativedelta import relativedelta
+from datetime import datetime, timedelta
 
 # Initialize SQLAlchemy instance
 db = SQLAlchemy()
@@ -11,6 +13,7 @@ class User(UserMixin, db.Model):
     id = db.Column(db.Integer, primary_key=True)
     username = db.Column(db.String(80), unique=True, nullable=False)
     password = db.Column(db.String(200), nullable=False)
+    email = db.Column(db.String(120), unique=True, nullable=False)  # Add email
     role = db.Column(db.String(10), nullable=False)
 
 class Task(db.Model):
@@ -25,6 +28,30 @@ class Task(db.Model):
     acteurs_externes = db.Column(db.String(200), nullable=True)
     last_completed = db.Column(db.DateTime, nullable=True)
 
+    def calculate_next_due_date(self):
+        """Calculate next due date based on periodicite"""
+        amount, unit = self.periodicite.lower().split()
+        amount = int(amount)
+        
+        today = datetime.now()
+        
+        if 'day' in unit:
+            return today + timedelta(days=amount)
+        elif 'week' in unit:
+            return today + timedelta(weeks=amount)
+        elif 'month' in unit:
+            return today + relativedelta(months=amount)
+        elif 'year' in unit or 'annual' in unit:
+            return today + relativedelta(years=amount)
+        
+        return today
+    
+    def is_due_soon(self, days=7):
+        """Check if task is due within specified days"""
+        if not self.echeance_prochaine:
+            return False
+        return datetime.now() <= self.echeance_prochaine <= datetime.now() + timedelta(days=days)
+
     def complete_task(self):
         """Mark task as complete and calculate next due date"""
         self.last_completed = datetime.now()

__pycache__/notification.cpython-312.pyc

@@ -0,0 +1,49 @@
+from flask_mail import Mail, Message
+from models import User, Task
+from datetime import datetime, timedelta
+
+mail = Mail()
+
+def setup_mail(app):
+    app.config['MAIL_SERVER'] = 'smtp.gmail.com'
+    app.config['MAIL_PORT'] = 587
+    app.config['MAIL_USE_TLS'] = True
+    app.config['MAIL_USERNAME'] = 'zorguimohamedyassine@gmail.com'  
+    app.config['MAIL_PASSWORD'] = 'ngxj cwbe vkff nkkk' 
+    mail.init_app(app)
+
+def send_task_notifications():
+    """Send notifications for tasks due in the next 7 days"""
+    upcoming_tasks = Task.query.filter(
+        Task.echeance_prochaine.between(
+            datetime.now(),
+            datetime.now() + timedelta(days=7)
+        )
+    ).all()
+    
+    if upcoming_tasks:
+        users = User.query.all()
+        for user in users:
+            msg = Message(
+                'Upcoming Tasks Notification',
+                sender='zorguimohamedyassine@gmail.com',
+                recipients=[user.email]
+            )
+            
+            task_list = "\n".join([
+                f"- {task.action_programmee} (Due: {task.echeance_prochaine.strftime('%Y-%m-%d')})"
+                for task in upcoming_tasks
+            ])
+            
+            msg.body = f"""
+            Hello {user.username},
+
+            The following tasks are due in the next 7 days:
+
+            {task_list}
+
+            Best regards,
+            Task Management System
+            """
+            
+            mail.send(msg)

__pycache__/routes.cpython-312.pyc

@@ -1,9 +1,10 @@
-from flask import render_template, redirect, url_for, flash
+from flask import render_template, redirect, url_for, flash,request
 from flask_login import login_user, login_required, current_user
 from werkzeug.security import check_password_hash
 from models import User, Task, db
 from forms import LoginForm, TaskForm
-from datetime import datetime
+from datetime import datetime, timedelta
+from sqlalchemy import or_
 
 def register_routes(app):
     """Register all routes with the Flask application"""
@@ -46,11 +47,54 @@ def add_task():
     @app.route('/calendar')
     @login_required
     def tasks_calendar():
-        tasks = Task.query.order_by(Task.echeance_prochaine).all()
-        return render_template('tasks_calendar.html', 
-                                tasks=tasks,
-                                today=datetime.now(),
-                                datetime=datetime)
+        #filter parameters
+        status = request.args.get('status', 'all')
+        sort_by = request.args.get('sort', 'due_date')
+        search = request.args.get('search', '')
+        time_frame = request.args.get('time_frame', 'all')
+
+        #base query
+        query = Task.query
+
+        #apply filters
+        if status == 'completed':
+            query = query.filter(Task.last_completed.isnot(None))
+        elif status == 'pending':
+            query = query.filter(Task.last_completed.is_(None))
+
+        if search:
+            query = query.filter(or_(
+                Task.volet.ilike(f'%{search}%'),
+                Task.action_programmee.ilike(f'%{search}%'),
+                Task.responsable.ilike(f'%{search}%'),
+            ))
+        if time_frame == 'week':
+            query.filter(Task.echeance_prochaine <= datetime.now() + timedelta(days=7))
+        elif time_frame == 'month':
+            query.filter(Task.echeance_prochaine <= datetime.now() + timedelta(days=30))
+
+        #apply sorting
+        if sort_by == 'volet':
+            query = query.order_by(Task.volet)
+        elif sort_by == 'responsable':
+            query = query.order_by(Task.responsable)
+        else:
+            query = query.order_by(Task.echeance_prochaine)
+
+        tasks = query.all()
+
+        return render_template(
+            'tasks_calendar.html',
+            tasks=tasks,
+            today=datetime.now(),
+            datetime=datetime,
+            current_filters={
+                'status': status,
+                'sort_by': sort_by,
+                'search': search,
+                'time_frame': time_frame
+            }
+        )
 
     @app.route('/complete-task/<int:task_id>', methods=['POST'])
     @login_required