Brady/feat/scan schedules

.gitignore

@@ -0,0 +1,13 @@
+.env
+.idea
+.python-version
+CLAUDE.md
+__pycache__/
+app/__pycache__/
+app/controllers/__pycache__/
+app/models/__pycache__/
+app/tasks/__pycache__/
+app/utils/__pycache__/
+app/utils/cron_utils.py
+instance/
+migrations/

app/controllers/tasks.py

@@ -155,15 +155,78 @@ def create():
         db.session.add(scan_task)
         db.session.commit()
 
-        flash('Scan task created successfully!', 'success')
+        # Handle scheduling if enabled
+        if form.enable_schedule.data and form.schedule_type.data:
+            schedule_type = form.schedule_type.data
+            schedule_data = {}
+
+            if schedule_type == 'daily':
+                schedule_data = {
+                    'hour': form.hour.data,
+                    'minute': form.minute.data
+                }
+            elif schedule_type == 'weekly':
+                days_of_week = form.days_of_week.data
+                if days_of_week and len(days_of_week) > 0:
+                    schedule_data = {
+                        'days': [int(d) for d in days_of_week],
+                        'hour': form.hour.data,
+                        'minute': form.minute.data
+                    }
+            elif schedule_type == 'monthly':
+                schedule_data = {
+                    'day': form.day.data,
+                    'hour': form.hour.data,
+                    'minute': form.minute.data
+                }
+            elif schedule_type == 'interval':
+                schedule_data = {
+                    'hours': form.hours.data
+                }
+            elif schedule_type == 'one-time':
+                run_datetime_str = f"{form.run_date.data} {form.run_time_hour.data:02d}:{form.run_time_minute.data:02d}:00"
+                schedule_data = {
+                    'run_datetime_str': run_datetime_str
+                }
+            elif schedule_type == 'cron':
+                schedule_data = {
+                    'cron_expression': form.cron_expression.data,
+                    'description': form.cron_description.data or ''
+                }
+
+            # Update task with schedule information
+            scan_task.is_scheduled = True
+            scan_task.schedule_type = schedule_type
+            scan_task.schedule_data = json.dumps(schedule_data)
+            db.session.commit()
+
+            # Schedule the task with APScheduler
+            try:
+                result = schedule_task(scan_task)
+                if result:
+                    current_app.logger.info(f"Successfully scheduled task {scan_task.id} ({scan_task.name}) with type {schedule_type}")
+                    flash('Scan task created and scheduled successfully!', 'success')
+                else:
+                    current_app.logger.error(f"Failed to schedule task {scan_task.id} ({scan_task.name}) - schedule_task returned False")
+                    flash('Scan task created but scheduling failed. Please edit the schedule.', 'warning')
+            except Exception as e:
+                current_app.logger.error(f"Error scheduling task {scan_task.id} ({scan_task.name}): {str(e)}", exc_info=True)
+                flash('Scan task created but scheduling encountered an error. Please edit the schedule.', 'warning')
+        else:
+            flash('Scan task created successfully!', 'success')
 
         # Run the scan immediately if requested
         if form.run_now.data:
             return redirect(url_for('tasks.run', id=scan_task.id))
 
         return redirect(url_for('tasks.index'))
 
-    return render_template('tasks/create.html', title='Create Scan Task', form=form)
+    # Get user timezone for schedule display
+    from app.models.user import User
+    user = User.query.get(current_user.id)
+    timezone_display = get_timezone_display_name(user.timezone if user else 'UTC')
+
+    return render_template('tasks/create.html', title='Create Scan Task', form=form, timezone_display=timezone_display)
 
 @tasks_bp.route('/<int:id>/edit', methods=['GET', 'POST'])
 @login_required
@@ -391,15 +454,35 @@ def schedule(id):
             form.hour.data = schedule_data.get('hour', 0)
             form.minute.data = schedule_data.get('minute', 0)
         elif scan_task.schedule_type == 'weekly':
-            form.day_of_week.data = schedule_data.get('day_of_week', 0)
             form.hour.data = schedule_data.get('hour', 0)
             form.minute.data = schedule_data.get('minute', 0)
+            # Check for multi-day format first
+            if 'days' in schedule_data:
+                form.days_of_week.data = [str(d) for d in schedule_data.get('days', [])]
+            else:
+                # Backward compatible single day format
+                form.day_of_week.data = schedule_data.get('day_of_week', 0)
         elif scan_task.schedule_type == 'monthly':
             form.day.data = schedule_data.get('day', 1)
             form.hour.data = schedule_data.get('hour', 0)
             form.minute.data = schedule_data.get('minute', 0)
         elif scan_task.schedule_type == 'interval':
             form.hours.data = schedule_data.get('hours', 24)
+        elif scan_task.schedule_type == 'one-time':
+            run_datetime_str = schedule_data.get('run_datetime_str', '')
+            if run_datetime_str:
+                # Parse datetime string to pre-populate fields
+                try:
+                    from datetime import datetime as dt
+                    run_dt = dt.strptime(run_datetime_str, '%Y-%m-%d %H:%M:%S')
+                    form.run_date.data = run_dt.strftime('%Y-%m-%d')
+                    form.run_time_hour.data = run_dt.hour
+                    form.run_time_minute.data = run_dt.minute
+                except Exception:
+                    pass
+        elif scan_task.schedule_type == 'cron':
+            form.cron_expression.data = schedule_data.get('cron_expression', '')
+            form.cron_description.data = schedule_data.get('description', '')
 
     if form.validate_on_submit():
         schedule_type = form.schedule_type.data
@@ -413,11 +496,19 @@ def schedule(id):
                 'minute': form.minute.data
             }
         elif schedule_type == 'weekly':
-            schedule_data = {
-                'day_of_week': form.day_of_week.data,
-                'hour': form.hour.data,
-                'minute': form.minute.data
-            }
+            # Handle multi-day selection (new format)
+            days_of_week = form.days_of_week.data  # List of strings
+            if days_of_week and len(days_of_week) > 0:
+                schedule_data = {
+                    'days': [int(d) for d in days_of_week],
+                    'hour': form.hour.data,
+                    'minute': form.minute.data
+                }
+            else:
+                # No days selected - this should have been caught by validation
+                flash('Please select at least one day for weekly schedule', 'danger')
+                return render_template('tasks/schedule.html', form=form, scan_task=scan_task,
+                                     title=f'Schedule Task: {scan_task.name}')
         elif schedule_type == 'monthly':
             schedule_data = {
                 'day': form.day.data,
@@ -428,6 +519,17 @@ def schedule(id):
             schedule_data = {
                 'hours': form.hours.data
             }
+        elif schedule_type == 'one-time':
+            # Build datetime string from date and time inputs
+            run_datetime_str = f"{form.run_date.data} {form.run_time_hour.data:02d}:{form.run_time_minute.data:02d}:00"
+            schedule_data = {
+                'run_datetime_str': run_datetime_str
+            }
+        elif schedule_type == 'cron':
+            schedule_data = {
+                'cron_expression': form.cron_expression.data,
+                'description': form.cron_description.data or ''
+            }
 
         # Update task with schedule information
         scan_task.is_scheduled = True
@@ -485,6 +587,29 @@ def api_status(run_id):
         'completed_at': scan_run.completed_at.isoformat() if scan_run.completed_at else None
     })
 
+@tasks_bp.route('/api/cron-preview', methods=['POST'])
+@login_required
+def api_cron_preview():
+    """Preview next execution times for a cron expression"""
+    from app.utils.cron_utils import validate_cron_expression, get_next_run_times
+    from app.models.user import User
+
+    cron_expr = request.json.get('cron_expression', '')
+
+    # Get user's timezone
+    user = User.query.get(current_user.id)
+    user_timezone = user.timezone if user else 'UTC'
+
+    # Validate the cron expression
+    is_valid, error = validate_cron_expression(cron_expr)
+    if not is_valid:
+        return jsonify({'error': error}), 400
+
+    # Get next run times
+    next_runs = get_next_run_times(cron_expr, count=5, timezone=user_timezone)
+
+    return jsonify({'next_runs': next_runs})
+
 @tasks_bp.route('/<int:run_id>/kill', methods=['POST'])
 @login_required
 def kill_task(run_id):

app/models/task.py

@@ -63,10 +63,30 @@ def get_schedule_display(self, timezone=None):
             hour = schedule_data.get('hour', 0)
             minute = schedule_data.get('minute', 0)
             time_str = f"{int(hour):02d}:{int(minute):02d}"
-            day_of_week = schedule_data.get('day_of_week', 0)  # 0 = Monday in most systems
-            days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
-            day_name = days[int(day_of_week) % 7]  # Ensure it's within range
-            return f"Weekly on {day_name} at {time_str}"
+            days_names = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
+
+            # Check for multi-day format first (new format)
+            if 'days' in schedule_data:
+                days_list = schedule_data.get('days', [])
+                if days_list:
+                    # Convert day indices to names
+                    selected_days = [days_names[int(d) % 7] for d in days_list]
+                    # Format nicely
+                    if len(selected_days) == 1:
+                        return f"Weekly on {selected_days[0]} at {time_str}"
+                    elif len(selected_days) == 7:
+                        return f"Daily at {time_str}"  # All days selected
+                    else:
+                        # Shorten day names for multi-day display
+                        short_days = [d[:3] for d in selected_days]
+                        return f"Weekly on {', '.join(short_days)} at {time_str}"
+                else:
+                    return f"Weekly (no days selected) at {time_str}"
+            else:
+                # Backward compatible single day format (old format)
+                day_of_week = schedule_data.get('day_of_week', 0)
+                day_name = days_names[int(day_of_week) % 7]
+                return f"Weekly on {day_name} at {time_str}"
 
         elif self.schedule_type == 'monthly':
             hour = schedule_data.get('hour', 0)
@@ -84,7 +104,53 @@ def get_schedule_display(self, timezone=None):
             else:
                 suffix = 'th'
             return f"Monthly on the {day_of_month}{suffix} at {time_str}"
-
+
+        elif self.schedule_type == 'interval':
+            hours = schedule_data.get('hours', 24)
+            if hours == 1:
+                return "Every hour"
+            elif hours < 24:
+                return f"Every {hours} hours"
+            else:
+                days = hours / 24
+                if days == int(days):
+                    return f"Every {int(days)} day(s)"
+                else:
+                    return f"Every {hours} hours"
+
+        elif self.schedule_type == 'one-time':
+            run_datetime_str = schedule_data.get('run_datetime_str', '')
+            if run_datetime_str:
+                # Parse and format the datetime string
+                try:
+                    from datetime import datetime as dt
+                    run_dt = dt.strptime(run_datetime_str, '%Y-%m-%d %H:%M:%S')
+                    return f"Once on {run_dt.strftime('%Y-%m-%d at %H:%M')}"
+                except Exception:
+                    return f"Once at {run_datetime_str}"
+            else:
+                return "One-time schedule"
+
+        elif self.schedule_type == 'cron':
+            cron_expression = schedule_data.get('cron_expression', '')
+            description = schedule_data.get('description', '')
+
+            if description:
+                return f"Cron: {cron_expression} ({description})"
+            elif cron_expression:
+                # Try to make it more readable using the cron_utils
+                try:
+                    from app.utils.cron_utils import cron_to_human_readable
+                    human_readable = cron_to_human_readable(cron_expression)
+                    if human_readable != cron_expression:
+                        return f"Cron: {cron_expression} ({human_readable})"
+                    else:
+                        return f"Cron: {cron_expression}"
+                except Exception:
+                    return f"Cron: {cron_expression}"
+            else:
+                return "Cron schedule (not configured)"
+
         else:
             # For any custom or unknown schedule types
             return f"{self.schedule_type.capitalize()} schedule"