316 lines
14 KiB
Python
316 lines
14 KiB
Python
from flask import Blueprint, render_template, request, jsonify, make_response
|
|
from flask_login import login_required, current_user
|
|
from app import db
|
|
from app.models import Folder
|
|
from app.imap_service import IMAPService
|
|
from app.processed_emails_service import ProcessedEmailsService
|
|
import logging
|
|
|
|
imap_bp = Blueprint('imap', __name__)
|
|
|
|
@imap_bp.route('/api/imap/config', methods=['GET'])
|
|
@login_required
|
|
def imap_config_modal():
|
|
"""Return the IMAP configuration modal."""
|
|
# Pass existing IMAP config to the template if it exists
|
|
response = make_response(render_template('partials/imap_config_modal.html',
|
|
server=current_user.imap_config.get('server') if current_user.imap_config else None,
|
|
port=current_user.imap_config.get('port') if current_user.imap_config else None,
|
|
username=current_user.imap_config.get('username') if current_user.imap_config else None,
|
|
password=current_user.imap_config.get('password') if current_user.imap_config else None,
|
|
use_ssl=current_user.imap_config.get('use_ssl', True) if current_user.imap_config else True))
|
|
response.headers['HX-Trigger'] = 'open-modal'
|
|
return response
|
|
|
|
@imap_bp.route('/api/imap/folders/modal', methods=['GET'])
|
|
@login_required
|
|
def imap_folders_modal():
|
|
"""Return the folder selection modal after successful IMAP connection."""
|
|
try:
|
|
if not current_user.imap_config:
|
|
return jsonify({'error': 'No IMAP configuration found. Please configure IMAP first.'}), 400
|
|
|
|
# Test connection first
|
|
imap_service = IMAPService(current_user)
|
|
|
|
# Get folders from IMAP server
|
|
imap_folders = imap_service.get_folders()
|
|
|
|
if not imap_folders:
|
|
return jsonify({'error': 'No folders found on IMAP server'}), 400
|
|
|
|
# Get all existing folders in a single query
|
|
existing_folders = Folder.query.filter(
|
|
Folder.user_id == current_user.id
|
|
).all()
|
|
|
|
# Create a dictionary for quick lookup
|
|
existing_folders_dict = {folder.name: folder for folder in existing_folders}
|
|
|
|
# Process folders
|
|
unique_folders = []
|
|
for imap_folder in imap_folders:
|
|
folder_name = imap_folder['name'].strip()
|
|
|
|
# Skip special folders that might not be needed
|
|
if folder_name.lower() in ['sent', 'drafts', 'spam', 'trash']:
|
|
continue
|
|
|
|
# Check if this folder already exists in the database
|
|
existing_folder = existing_folders_dict.get(folder_name)
|
|
|
|
# Add folder type and subscribed status if it exists
|
|
if existing_folder:
|
|
imap_folder['folder_type'] = existing_folder.folder_type
|
|
imap_folder['subscribed'] = True
|
|
imap_folder['selected'] = True
|
|
else:
|
|
# Set default folder type
|
|
imap_folder['folder_type'] = 'tidy' if folder_name.lower() == 'inbox' else 'destination'
|
|
imap_folder['subscribed'] = False
|
|
imap_folder['selected'] = True
|
|
|
|
unique_folders.append(imap_folder)
|
|
|
|
print(unique_folders, existing_folders_dict)
|
|
# Return the folder selection modal
|
|
response = make_response(render_template('partials/folder_selection_modal.html', folders=unique_folders))
|
|
response.headers['hx-retarget'] = "#modal-holder"
|
|
response.headers['HX-Trigger'] = 'open-modal'
|
|
return response
|
|
|
|
except Exception as e:
|
|
logging.exception("Error getting IMAP folders modal: %s", e)
|
|
print(e)
|
|
return jsonify({'error': 'An unexpected error occurred while fetching folders'}), 500
|
|
|
|
@imap_bp.route('/api/imap/test', methods=['POST'])
|
|
@login_required
|
|
def test_imap_connection():
|
|
"""Test IMAP connection with provided configuration."""
|
|
try:
|
|
# Get form data
|
|
server = request.form.get('server')
|
|
port = request.form.get('port')
|
|
username = request.form.get('username')
|
|
password = request.form.get('password')
|
|
use_ssl = request.form.get('use_ssl') == 'on'
|
|
|
|
# Validate required fields
|
|
errors = {}
|
|
if not server:
|
|
errors['server'] = 'Server is required'
|
|
if not port:
|
|
errors['port'] = 'Port is required'
|
|
elif not port.isdigit():
|
|
errors['port'] = 'Port must be a number'
|
|
if not username:
|
|
errors['username'] = 'Username is required'
|
|
if not password:
|
|
errors['password'] = 'Password is required'
|
|
|
|
if errors:
|
|
response = make_response(render_template('partials/imap_config_modal.html', errors=errors, server=server, port=port, username=username, use_ssl=use_ssl))
|
|
response.headers['HX-Retarget'] = '#imap-modal'
|
|
response.headers['HX-Reswap'] = 'outerHTML'
|
|
return response
|
|
|
|
# Store configuration temporarily for testing
|
|
test_config = {
|
|
'server': server,
|
|
'port': int(port),
|
|
'username': username,
|
|
'password': password,
|
|
'use_ssl': False,
|
|
'use_tls': False,
|
|
'connection_timeout': 30
|
|
}
|
|
|
|
# Test connection
|
|
temp_user = type('User', (), {'imap_config': test_config})()
|
|
imap_service = IMAPService(temp_user)
|
|
print(temp_user, test_config)
|
|
success, message = imap_service.test_connection()
|
|
|
|
if success:
|
|
# Save configuration to user's profile
|
|
current_user.imap_config = test_config
|
|
db.session.commit()
|
|
|
|
# Redirect to folder selection modal after successful connection
|
|
return imap_folders_modal()
|
|
#response.headers['HX-Trigger'] = 'open-modal'
|
|
else:
|
|
response = make_response(render_template('partials/imap_config_modal.html',
|
|
errors={'general': message}, server=server, port=port, username=username, use_ssl=use_ssl))
|
|
response.headers['HX-Retarget'] = '#imap-modal'
|
|
response.headers['HX-Reswap'] = 'outerHTML'
|
|
|
|
return response
|
|
|
|
except Exception as e:
|
|
logging.exception("Error testing IMAP connection: %s", e)
|
|
print(e)
|
|
errors = {'general': 'An unexpected error occurred. Please try again.'}
|
|
response = make_response(render_template('partials/imap_config_modal.html', errors=errors, server=server, port=port, username=username, use_ssl=use_ssl))
|
|
response.headers['HX-Retarget'] = '#imap-modal'
|
|
response.headers['HX-Reswap'] = 'outerHTML'
|
|
return response
|
|
|
|
@imap_bp.route('/api/imap/folders', methods=['GET'])
|
|
@login_required
|
|
def get_imap_folders():
|
|
"""Get folders from IMAP server without creating database records."""
|
|
try:
|
|
if not current_user.imap_config:
|
|
return jsonify({'error': 'No IMAP configuration found. Please configure IMAP first.'}), 400
|
|
|
|
# Test connection first
|
|
imap_service = IMAPService(current_user)
|
|
|
|
# Get folders from IMAP server
|
|
imap_folders = imap_service.get_folders()
|
|
|
|
if not imap_folders:
|
|
return jsonify({'error': 'No folders found on IMAP server'}), 400
|
|
|
|
# Deduplicate folders by name to prevent creating multiple entries for the same folder
|
|
unique_folders = []
|
|
seen_names = set()
|
|
for imap_folder in imap_folders:
|
|
folder_name = imap_folder['name']
|
|
|
|
# Skip special folders that might not be needed
|
|
if folder_name.lower() in ['sent', 'drafts', 'spam', 'trash']:
|
|
continue
|
|
|
|
# Use case-insensitive comparison for deduplication
|
|
folder_name_lower = folder_name.lower()
|
|
if folder_name_lower not in seen_names:
|
|
unique_folders.append(imap_folder)
|
|
seen_names.add(folder_name_lower)
|
|
|
|
return jsonify({'folders': unique_folders})
|
|
|
|
except Exception as e:
|
|
logging.exception("Error getting IMAP folders: %s", e)
|
|
print(e)
|
|
return jsonify({'error': 'An unexpected error occurred while fetching folders'}), 500
|
|
|
|
@imap_bp.route('/api/imap/sync-selected', methods=['POST'])
|
|
@login_required
|
|
def sync_selected_folders():
|
|
"""Sync only the selected folders from IMAP server with processed email tracking."""
|
|
try:
|
|
if not current_user.imap_config:
|
|
return jsonify({'error': 'No IMAP configuration found. Please configure IMAP first.'}), 400
|
|
|
|
# Get selected folder names and types from form data
|
|
selected_folders = {}
|
|
for key, value in request.form.items():
|
|
if key.startswith('folder_'):
|
|
folder_name = value
|
|
# Check if there's a corresponding folder type
|
|
type_key = f'folder_type_{key.split("_")[1]}'
|
|
folder_type = request.form.get(type_key, 'destination') # Default to 'destination'
|
|
selected_folders[folder_name] = folder_type
|
|
|
|
if not selected_folders:
|
|
return jsonify({'error': 'No folders selected'}), 400
|
|
|
|
# Test connection first
|
|
imap_service = IMAPService(current_user)
|
|
|
|
# Get folders from IMAP server
|
|
imap_folders = imap_service.get_folders()
|
|
|
|
if not imap_folders:
|
|
return jsonify({'error': 'No folders found on IMAP server'}), 400
|
|
|
|
# Deduplicate folders by name to prevent creating multiple entries for the same folder
|
|
unique_folders = []
|
|
seen_names = set()
|
|
for imap_folder in imap_folders:
|
|
folder_name = imap_folder['name']
|
|
|
|
# Skip special folders that might not be needed
|
|
if folder_name.lower() in ['sent', 'drafts', 'spam', 'trash']:
|
|
continue
|
|
|
|
# Only include folders that were selected by the user
|
|
if folder_name in selected_folders:
|
|
# Use case-insensitive comparison for deduplication
|
|
folder_name_lower = folder_name.lower()
|
|
if folder_name_lower not in seen_names:
|
|
# Add the selected folder type
|
|
imap_folder['selected_type'] = selected_folders[folder_name]
|
|
unique_folders.append(imap_folder)
|
|
seen_names.add(folder_name_lower)
|
|
|
|
# Process each selected unique folder
|
|
synced_count = 0
|
|
processed_emails_service = ProcessedEmailsService(current_user)
|
|
|
|
# Create a list of folders to process
|
|
folders_to_process = []
|
|
|
|
for imap_folder in unique_folders:
|
|
folder_name = imap_folder['name'].strip()
|
|
folder_type = imap_folder.get('selected_type', 'destination') # Get the selected type
|
|
|
|
# Handle nested folder names (convert slashes to underscores or keep as-is)
|
|
# According to requirements, nested folders should be created with slashes in the name
|
|
display_name = folder_name
|
|
|
|
# Check if folder already exists
|
|
existing_folder = Folder.query.filter_by(
|
|
user_id=current_user.id,
|
|
name=display_name
|
|
).first()
|
|
|
|
if not existing_folder:
|
|
# Create new folder with the selected type
|
|
new_folder = Folder(
|
|
user_id=current_user.id,
|
|
name=display_name,
|
|
rule_text=f"Auto-synced from IMAP folder: {folder_name}",
|
|
priority=0, # Default priority
|
|
folder_type=folder_type
|
|
)
|
|
db.session.add(new_folder)
|
|
synced_count += 1
|
|
folders_to_process.append(new_folder)
|
|
else:
|
|
# Update existing folder with the selected type and email counts
|
|
existing_folder.folder_type = folder_type
|
|
|
|
# Get all email UIDs in this folder
|
|
email_uids = imap_service.get_folder_email_uids(folder_name)
|
|
|
|
# Sync with processed emails service
|
|
new_emails_count = processed_emails_service.sync_folder_emails(display_name, email_uids)
|
|
print("NEW", new_emails_count)
|
|
|
|
# Update counts
|
|
pending_count = processed_emails_service.get_pending_count(display_name)
|
|
existing_folder.pending_count = pending_count
|
|
existing_folder.total_count = len(email_uids)
|
|
|
|
# Get the most recent emails for this folder
|
|
recent_emails = imap_service.get_recent_emails(folder_name, 3)
|
|
existing_folder.recent_emails = recent_emails
|
|
|
|
folders_to_process.append(existing_folder)
|
|
|
|
db.session.commit()
|
|
|
|
# Just trigger the folder list update and close the modal
|
|
response = make_response('')
|
|
response.headers['HX-Trigger'] = 'close-modal, folder-list-invalidated'
|
|
return response
|
|
|
|
except Exception as e:
|
|
logging.exception("Error syncing selected IMAP folders: %s", e)
|
|
print(e)
|
|
db.session.rollback()
|
|
return jsonify({'error': 'An unexpected error occurred'}), 500 |