from flask import Blueprint, render_template, redirect, url_for, flash, request, abort from flask_login import current_user, login_required from app import db from models import Category, Topic, Post, Tag, Reaction, Report from forms import CreateTopicForm, CreatePostForm, ReportForm from datetime import datetime import logging from sqlalchemy import desc, func # Set up logger logger = logging.getLogger(__name__) # Create blueprint forum_bp = Blueprint('forum', __name__) # Helper functions def get_or_create_tags(tag_names): """Gets existing tags or creates new ones from a comma-separated string""" if not tag_names: return [] tags = [] for name in [t.strip() for t in tag_names.split(',') if t.strip()]: tag = Tag.query.filter_by(name=name.lower()).first() if not tag: tag = Tag(name=name.lower()) db.session.add(tag) tags.append(tag) return tags # Routes @forum_bp.route('/') def index(): categories = Category.query.order_by(Category.order).all() return render_template('home.html', categories=categories) @forum_bp.route('/categories') def category_list(): categories = Category.query.order_by(Category.order).all() return render_template('forum/category_list.html', categories=categories) @forum_bp.route('/category/') def category_view(id): category = Category.query.get_or_404(id) page = request.args.get('page', 1, type=int) per_page = 20 # Topics per page # Get topics with pagination topics = Topic.query.filter_by(category_id=id)\ .order_by(Topic.is_pinned.desc(), Topic.updated_at.desc())\ .paginate(page=page, per_page=per_page, error_out=False) return render_template('forum/topic_list.html', category=category, topics=topics) @forum_bp.route('/topic/', methods=['GET', 'POST']) def topic_view(id): topic = Topic.query.get_or_404(id) # Increment view counter if not current_user.is_authenticated or current_user.id != topic.author_id: topic.increment_view() # Handle pagination page = request.args.get('page', 1, type=int) per_page = 10 # Posts per page posts = Post.query.filter_by(topic_id=id)\ .order_by(Post.created_at)\ .paginate(page=page, per_page=per_page, error_out=False) # Create forms post_form = CreatePostForm() if current_user.is_authenticated else None report_form = ReportForm() if current_user.is_authenticated else None # Handle post creation if current_user.is_authenticated and post_form and post_form.validate_on_submit(): if topic.is_locked and not current_user.is_moderator(): flash('This topic is locked. You cannot reply to it.', 'danger') return redirect(url_for('forum.topic_view', id=id)) post = Post( content=post_form.content.data, topic_id=id, author_id=current_user.id ) db.session.add(post) topic.updated_at = datetime.utcnow() db.session.commit() flash('Your reply has been posted!', 'success') return redirect(url_for('forum.topic_view', id=id, page=posts.pages or 1)) return render_template('forum/topic_view.html', topic=topic, posts=posts, post_form=post_form, report_form=report_form) @forum_bp.route('/create_topic', methods=['GET', 'POST']) @login_required def create_topic(): form = CreateTopicForm() # Populate category choices form.category_id.choices = [(c.id, c.name) for c in Category.query.order_by(Category.name).all()] if form.validate_on_submit(): # Create new topic topic = Topic( title=form.title.data, category_id=form.category_id.data, author_id=current_user.id ) # Add tags if provided if form.tags.data: topic.tags = get_or_create_tags(form.tags.data) db.session.add(topic) db.session.flush() # Get the topic ID # Create the first post post = Post( content=form.content.data, topic_id=topic.id, author_id=current_user.id ) db.session.add(post) db.session.commit() flash('Your topic has been created!', 'success') return redirect(url_for('forum.topic_view', id=topic.id)) return render_template('forum/create_topic.html', form=form) @forum_bp.route('/post//quote') @login_required def quote_post(id): post = Post.query.get_or_404(id) topic_id = post.topic_id if post.topic.is_locked and not current_user.is_moderator(): flash('This topic is locked. You cannot reply to it.', 'danger') return redirect(url_for('forum.topic_view', id=topic_id)) quoted_content = f'

{post.content}

Posted by {post.author.username}

' # Create a form and pre-fill it with the quoted content form = CreatePostForm(topic_id=topic_id, content=quoted_content) return render_template('forum/create_post.html', form=form, topic=post.topic, is_quote=True) @forum_bp.route('/post//edit', methods=['GET', 'POST']) @login_required def edit_post(id): post = Post.query.get_or_404(id) # Check permissions if current_user.id != post.author_id and not current_user.is_moderator(): abort(403) # Create form and pre-fill with existing content form = CreatePostForm(topic_id=post.topic_id) if request.method == 'GET': form.content.data = post.content if form.validate_on_submit(): post.content = form.content.data post.updated_at = datetime.utcnow() post.edited_by_id = current_user.id db.session.commit() flash('Your post has been updated!', 'success') return redirect(url_for('forum.topic_view', id=post.topic_id)) return render_template('forum/create_post.html', form=form, topic=post.topic, post=post, is_edit=True) @forum_bp.route('/post//delete', methods=['POST']) @login_required def delete_post(id): post = Post.query.get_or_404(id) topic_id = post.topic_id # Check permissions if current_user.id != post.author_id and not current_user.is_moderator(): abort(403) # Check if this is the first post in a topic is_first_post = post.id == Post.query.filter_by(topic_id=topic_id).order_by(Post.created_at).first().id if is_first_post: # If first post, delete the entire topic topic = Topic.query.get(topic_id) db.session.delete(topic) # This should cascade delete all posts db.session.commit() flash('The topic has been deleted!', 'success') return redirect(url_for('forum.category_view', id=topic.category_id)) else: # Delete just this post db.session.delete(post) db.session.commit() flash('The post has been deleted!', 'success') return redirect(url_for('forum.topic_view', id=topic_id)) @forum_bp.route('/topic//lock', methods=['POST']) @login_required def lock_topic(id): if not current_user.is_moderator(): abort(403) topic = Topic.query.get_or_404(id) topic.is_locked = not topic.is_locked # Toggle lock state db.session.commit() status = 'locked' if topic.is_locked else 'unlocked' flash(f'The topic has been {status}!', 'success') return redirect(url_for('forum.topic_view', id=id)) @forum_bp.route('/topic//pin', methods=['POST']) @login_required def pin_topic(id): if not current_user.is_moderator(): abort(403) topic = Topic.query.get_or_404(id) topic.is_pinned = not topic.is_pinned # Toggle pin state db.session.commit() status = 'pinned' if topic.is_pinned else 'unpinned' flash(f'The topic has been {status}!', 'success') return redirect(url_for('forum.topic_view', id=id)) @forum_bp.route('/post//react', methods=['POST']) @login_required def react_to_post(post_id): post = Post.query.get_or_404(post_id) reaction_type = request.form.get('reaction_type', 'like') # Check if user already reacted existing_reaction = Reaction.query.filter_by( user_id=current_user.id, post_id=post_id ).first() if existing_reaction: if existing_reaction.reaction_type == reaction_type: # If same reaction, remove it db.session.delete(existing_reaction) db.session.commit() return {'status': 'removed', 'count': post.get_reaction_count(reaction_type)} else: # If different reaction, update it existing_reaction.reaction_type = reaction_type db.session.commit() return {'status': 'updated', 'count': post.get_reaction_count(reaction_type)} else: # Create new reaction reaction = Reaction( user_id=current_user.id, post_id=post_id, reaction_type=reaction_type ) db.session.add(reaction) db.session.commit() return {'status': 'added', 'count': post.get_reaction_count(reaction_type)} @forum_bp.route('/report', methods=['POST']) @login_required def create_report(): form = ReportForm() if form.validate_on_submit(): report = Report( reason=form.reason.data, reporter_id=current_user.id ) # Set either post_id or topic_id if form.post_id.data: report.post_id = form.post_id.data item = Post.query.get_or_404(form.post_id.data) report.topic_id = item.topic_id elif form.topic_id.data: report.topic_id = form.topic_id.data item = Topic.query.get_or_404(form.topic_id.data) else: flash('Invalid report submission.', 'danger') return redirect(url_for('forum.index')) db.session.add(report) db.session.commit() flash('Your report has been submitted. A moderator will review it soon.', 'success') # Redirect back to the topic return redirect(url_for('forum.topic_view', id=report.topic_id)) flash('There was an error with your report submission.', 'danger') return redirect(url_for('forum.index')) @forum_bp.route('/tags/') def tag_view(tag_name): tag = Tag.query.filter_by(name=tag_name.lower()).first_or_404() page = request.args.get('page', 1, type=int) per_page = 20 topics = tag.topics.order_by(Topic.updated_at.desc())\ .paginate(page=page, per_page=per_page, error_out=False) return render_template('forum/tag_topics.html', tag=tag, topics=topics) @forum_bp.route('/search') def search(): query = request.args.get('q', '') page = request.args.get('page', 1, type=int) per_page = 20 if not query or len(query) < 3: flash('Search query must be at least 3 characters long.', 'warning') return redirect(url_for('forum.index')) # Search in topics topics = Topic.query.filter(Topic.title.ilike(f'%{query}%'))\ .order_by(Topic.updated_at.desc())\ .paginate(page=page, per_page=per_page, error_out=False) return render_template('forum/search_results.html', query=query, topics=topics)