"""Settings and profile route registration.""" import json import logging import os from flask import current_app, flash, redirect, render_template, request, url_for from flask_login import current_user, login_required from werkzeug.utils import secure_filename from config import MAX_FILENAME_LENGTH, UPLOAD_FOLDER from database import db from extensions import get_filter_engine from security import is_allowed_file, is_safe_filterset from services import SettingsService, load_platform_config logger = logging.getLogger(__name__) DEFAULT_AVATARS = [ {"id": "default_1", "name": "Gradient Blue", "bg": "linear-gradient(135deg, #667eea 0%, #764ba2 100%)"}, {"id": "default_2", "name": "Gradient Green", "bg": "linear-gradient(135deg, #4facfe 0%, #00f2fe 100%)"}, {"id": "default_3", "name": "Gradient Orange", "bg": "linear-gradient(135deg, #fa709a 0%, #fee140 100%)"}, {"id": "default_4", "name": "Gradient Purple", "bg": "linear-gradient(135deg, #a8edea 0%, #fed6e3 100%)"}, {"id": "default_5", "name": "Brand Colors", "bg": "linear-gradient(135deg, #4db6ac 0%, #26a69a 100%)"}, {"id": "default_6", "name": "Sunset", "bg": "linear-gradient(135deg, #ff7e5f 0%, #feb47b 100%)"}, ] def register_settings_routes(app, user_service): """Register settings/profile routes while preserving endpoint names.""" @app.route("/settings") @login_required def settings(): """Main settings page.""" user_settings = SettingsService.parse(current_user.settings) try: with open("filtersets.json", "r", encoding="utf-8") as f: filter_sets = json.load(f) except (FileNotFoundError, json.JSONDecodeError, IOError): filter_sets = {} return render_template( "settings.html", user=current_user, user_settings=user_settings, filter_sets=filter_sets, ) @app.route("/settings/profile", methods=["GET", "POST"]) @login_required def settings_profile(): """Profile settings page.""" if request.method == "POST": username = request.form.get("username") email = request.form.get("email") default_avatar = request.form.get("default_avatar") if not username or not email: flash("Username and email are required", "error") return render_template("settings_profile.html", user=current_user) if username != current_user.username and user_service.username_exists(username): flash("Username already taken", "error") return render_template("settings_profile.html", user=current_user) if email != current_user.email and user_service.email_exists(email): flash("Email already registered", "error") return render_template("settings_profile.html", user=current_user) current_user.username = username current_user.email = email if default_avatar and default_avatar.startswith("default_"): current_user.profile_picture_url = f"/static/default-avatars/{default_avatar}.png" db.session.commit() flash("Profile updated successfully", "success") return redirect(url_for("settings")) return render_template( "settings_profile.html", user=current_user, default_avatars=DEFAULT_AVATARS, ) @app.route("/settings/communities", methods=["GET", "POST"]) @login_required def settings_communities(): """Community/source selection settings.""" if request.method == "POST": user_settings = SettingsService.parse(current_user.settings) user_settings["communities"] = request.form.getlist("communities") current_user.settings = json.dumps(user_settings) db.session.commit() flash("Community preferences updated", "success") return redirect(url_for("settings")) user_settings = SettingsService.parse(current_user.settings) selected_communities = user_settings.get("communities", []) available_communities = [] try: platform_config = load_platform_config() or {"platforms": {}, "collection_targets": []} except Exception as e: logger.error(f"Error loading platform config: {e}") platform_config = {"platforms": {}, "collection_targets": []} enabled_communities = set() try: for target in platform_config.get("collection_targets", []): if "platform" in target and "community" in target: enabled_communities.add((target["platform"], target["community"])) except Exception as e: logger.error(f"Error processing collection_targets: {e}") try: for platform_name, platform_info in platform_config.get("platforms", {}).items(): if not isinstance(platform_info, dict): continue communities = platform_info.get("communities", []) if not isinstance(communities, list): continue for community_info in communities: try: if not isinstance(community_info, dict): continue if (platform_name, community_info["id"]) in enabled_communities: available_communities.append( { "id": community_info["id"], "name": community_info["name"], "display_name": community_info.get("display_name", community_info["name"]), "platform": platform_name, "icon": community_info.get("icon", platform_info.get("icon", "\U0001f4c4")), "description": community_info.get("description", ""), } ) except Exception as e: logger.error(f"Error processing community {community_info}: {e}") continue except Exception as e: logger.error(f"Error building community list: {e}") logger.info(f"Found {len(available_communities)} available communities") return render_template( "settings_communities.html", user=current_user, available_communities=available_communities, selected_communities=selected_communities, ) @app.route("/settings/filters", methods=["GET", "POST"]) @login_required def settings_filters(): """Filter settings page.""" if request.method == "POST": selected_filter = request.form.get("filter_set", "no_filter") user_settings = SettingsService.validate(current_user.settings) if is_safe_filterset(selected_filter): user_settings["filter_set"] = selected_filter else: flash("Invalid filter selection", "error") return redirect(url_for("settings")) try: current_user.settings = json.dumps(user_settings) db.session.commit() flash("Filter settings updated successfully", "success") except Exception as e: db.session.rollback() logger.error(f"Error saving filter settings for user {current_user.id}: {e}") flash("Error saving settings", "error") return redirect(url_for("settings")) user_settings = SettingsService.parse(current_user.settings) current_filter = user_settings.get("filter_set", "no_filter") filter_engine = get_filter_engine() filter_sets = { filterset_name: filter_engine.config.get_filterset(filterset_name) for filterset_name in filter_engine.get_available_filtersets() } return render_template( "settings_filters.html", user=current_user, filter_sets=filter_sets, current_filter=current_filter, ) @app.route("/settings/experience", methods=["GET", "POST"]) @login_required def settings_experience(): """Experience and behavioral settings page.""" if request.method == "POST": user_settings = SettingsService.parse(current_user.settings) user_settings["experience"] = { "infinite_scroll": request.form.get("infinite_scroll") == "on", "auto_refresh": request.form.get("auto_refresh") == "on", "push_notifications": request.form.get("push_notifications") == "on", "dark_patterns_opt_in": request.form.get("dark_patterns_opt_in") == "on", "time_filter_enabled": request.form.get("time_filter_enabled") == "on", "time_filter_days": int(request.form.get("time_filter_days", 7)), } current_user.settings = json.dumps(user_settings) db.session.commit() flash("Experience settings updated successfully", "success") return redirect(url_for("settings")) return render_template( "settings_experience.html", user=current_user, experience_settings=SettingsService.experience_settings(current_user.settings), ) @app.route("/upload-avatar", methods=["POST"]) @login_required def upload_avatar(): """Upload profile picture.""" try: logger.info(f"Avatar upload attempt by user {current_user.id} ({current_user.username})") logger.debug(f"Request files: {list(request.files.keys())}") logger.debug(f"Request form: {dict(request.form)}") if not hasattr(current_user, "id") or not current_user.id: logger.error("User missing ID attribute") flash("Authentication error. Please log in again.", "error") return redirect(url_for("login")) if not hasattr(current_user, "username") or not current_user.username: logger.error("User missing username attribute") flash("User profile incomplete. Please update your profile.", "error") return redirect(url_for("settings_profile")) if "avatar" not in request.files: logger.warning("No avatar file in request") flash("No file selected", "error") return redirect(url_for("settings_profile")) file = request.files["avatar"] if file.filename == "": logger.warning("Empty filename provided") flash("No file selected", "error") return redirect(url_for("settings_profile")) logger.info(f"Processing file: {file.filename}") if not is_allowed_file(file.filename): logger.warning(f"Invalid file type: {file.filename}") flash("Invalid file type. Please upload PNG, JPG, or GIF", "error") return redirect(url_for("settings_profile")) max_content_length = current_app.config.get("MAX_CONTENT_LENGTH", 16 * 1024 * 1024) if hasattr(file, "content_length") and file.content_length > max_content_length: logger.warning(f"File too large: {file.content_length}") flash("File too large. Maximum size is 16MB", "error") return redirect(url_for("settings_profile")) filename = secure_filename(file.filename) if not filename or len(filename) > MAX_FILENAME_LENGTH: logger.warning(f"Invalid filename after sanitization: {filename}") flash("Invalid filename", "error") return redirect(url_for("settings_profile")) unique_filename = f"{current_user.id}_{filename}" logger.info(f"Generated unique filename: {unique_filename}") upload_dir = os.path.abspath(UPLOAD_FOLDER) os.makedirs(upload_dir, exist_ok=True) upload_path = os.path.join(upload_dir, unique_filename) if not os.path.abspath(upload_path).startswith(upload_dir): logger.warning(f"Path traversal attempt in file upload: {upload_path}") flash("Invalid file path", "error") return redirect(url_for("settings_profile")) file.save(upload_path) logger.info(f"File saved successfully: {upload_path}") old_avatar_url = current_user.profile_picture_url current_user.profile_picture_url = f"/static/avatars/{unique_filename}" db.session.commit() logger.info(f"User profile updated successfully for {current_user.username}") if old_avatar_url and old_avatar_url.startswith("/static/avatars/") and current_user.id in old_avatar_url: try: old_file_path = os.path.join(upload_dir, os.path.basename(old_avatar_url)) if os.path.exists(old_file_path): os.remove(old_file_path) logger.info(f"Cleaned up old avatar: {old_file_path}") except Exception as e: logger.warning(f"Could not clean up old avatar: {e}") flash("Profile picture updated successfully", "success") return redirect(url_for("settings_profile")) except Exception as e: logger.error(f"Unexpected error in avatar upload: {e}") db.session.rollback() flash("An unexpected error occurred. Please try again.", "error") return redirect(url_for("settings_profile")) @app.route("/profile") @login_required def profile(): """User profile page.""" return render_template("profile.html", user=current_user)