{code}\1', text)
+
+ # Bold (**...**)
+ text = re.sub(r"\*\*([^*]+)\*\*", r"\1", text)
+
+ # Italic (*...*)
+ text = re.sub(r"\*([^*]+)\*", r"\1", text)
+
+ # Links [text](url) - only allow http/https URLs
+ def replace_link(match):
+ link_text = match.group(1)
+ url = match.group(2)
+ if url.startswith(("http://", "https://", "/")):
+ return f'{link_text}'
+ return match.group(0)
+
+ text = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", replace_link, text)
+
+ # Blockquotes (> ...)
+ lines = text.split("\n")
+ result_lines = []
+ in_blockquote = False
+ blockquote_lines = []
+
+ for line in lines:
+ if line.startswith("> "): # Escaped >
+ if not in_blockquote:
+ in_blockquote = True
+ blockquote_lines = []
+ blockquote_lines.append(line[5:])
+ else:
+ if in_blockquote:
+ result_lines.append(f'{" ".join(blockquote_lines)}') + in_blockquote = False + blockquote_lines = [] + result_lines.append(line) + + if in_blockquote: + result_lines.append(f'
{" ".join(blockquote_lines)}') + + text = "\n".join(result_lines) + + # Unordered lists (- item) + text = re.sub(r"^- (.+)$", r'
{p}
') + + return Markup("\n".join(formatted)) + + def register_filters(app: Flask) -> None: """Register all custom filters with a Flask app.""" app.jinja_env.filters["timeago"] = timeago app.jinja_env.filters["format_number"] = format_number app.jinja_env.filters["date_format"] = date_format app.jinja_env.filters["truncate_words"] = truncate_words + app.jinja_env.filters["markdown"] = render_markdown # Global functions app.jinja_env.globals["now"] = datetime.utcnow diff --git a/src/smarttools/web/forum/__init__.py b/src/smarttools/web/forum/__init__.py new file mode 100644 index 0000000..01dffd8 --- /dev/null +++ b/src/smarttools/web/forum/__init__.py @@ -0,0 +1,5 @@ +"""Forum blueprint for SmartTools community discussions.""" + +from .routes import forum_bp + +__all__ = ["forum_bp"] diff --git a/src/smarttools/web/forum/models.py b/src/smarttools/web/forum/models.py new file mode 100644 index 0000000..6638d36 --- /dev/null +++ b/src/smarttools/web/forum/models.py @@ -0,0 +1,245 @@ +"""Forum database models and schema.""" + +from __future__ import annotations + +import re +import sqlite3 +from typing import Any + +from smarttools.registry.db import connect_db, query_one, query_all + + +FORUM_SCHEMA = """ +CREATE TABLE IF NOT EXISTS forum_categories ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + slug TEXT UNIQUE NOT NULL, + name TEXT NOT NULL, + description TEXT, + icon TEXT, + sort_order INTEGER DEFAULT 0, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP +); + +CREATE TABLE IF NOT EXISTS forum_topics ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + category_id INTEGER NOT NULL REFERENCES forum_categories(id), + title TEXT NOT NULL, + slug TEXT NOT NULL, + author_name TEXT NOT NULL, + author_email TEXT, + content TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + is_pinned INTEGER DEFAULT 0, + is_locked INTEGER DEFAULT 0, + view_count INTEGER DEFAULT 0 +); + +CREATE TABLE IF NOT EXISTS forum_replies ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + topic_id INTEGER NOT NULL REFERENCES forum_topics(id) ON DELETE CASCADE, + author_name TEXT NOT NULL, + author_email TEXT, + content TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + is_solution INTEGER DEFAULT 0 +); + +CREATE INDEX IF NOT EXISTS idx_forum_topics_category ON forum_topics(category_id); +CREATE INDEX IF NOT EXISTS idx_forum_topics_updated ON forum_topics(updated_at DESC); +CREATE INDEX IF NOT EXISTS idx_forum_replies_topic ON forum_replies(topic_id); +""" + +DEFAULT_CATEGORIES = [ + ("general", "General Discussion", "Chat about anything SmartTools related", "chat", 1), + ("help", "Help & Support", "Get help with installation, configuration, or usage", "question", 2), + ("showcase", "Showcase", "Share tools and projects you've built with SmartTools", "star", 3), + ("ideas", "Ideas & Feedback", "Suggest features and improvements", "lightbulb", 4), + ("tutorials", "Tutorials & Guides", "Community-written guides and how-tos", "book", 5), +] + + +def init_forum_schema(conn: sqlite3.Connection) -> None: + """Initialize forum tables.""" + conn.executescript(FORUM_SCHEMA) + conn.commit() + + +def seed_categories(conn: sqlite3.Connection) -> None: + """Seed default forum categories if they don't exist.""" + for slug, name, description, icon, sort_order in DEFAULT_CATEGORIES: + existing = query_one( + conn, "SELECT id FROM forum_categories WHERE slug = ?", [slug] + ) + if not existing: + conn.execute( + """INSERT INTO forum_categories (slug, name, description, icon, sort_order) + VALUES (?, ?, ?, ?, ?)""", + [slug, name, description, icon, sort_order], + ) + conn.commit() + + +def slugify(text: str) -> str: + """Convert text to URL-friendly slug.""" + text = text.lower().strip() + text = re.sub(r"[^\w\s-]", "", text) + text = re.sub(r"[-\s]+", "-", text) + return text[:100] # Limit length + + +def get_categories_with_stats(conn: sqlite3.Connection) -> list[dict[str, Any]]: + """Get all categories with topic counts and latest activity.""" + rows = query_all( + conn, + """ + SELECT + c.*, + COUNT(t.id) as topic_count, + COALESCE(SUM((SELECT COUNT(*) FROM forum_replies r WHERE r.topic_id = t.id)), 0) as reply_count, + MAX(COALESCE(t.updated_at, c.created_at)) as latest_activity + FROM forum_categories c + LEFT JOIN forum_topics t ON t.category_id = c.id + GROUP BY c.id + ORDER BY c.sort_order + """, + ) + return [dict(row) for row in rows] + + +def get_category_by_slug(conn: sqlite3.Connection, slug: str) -> dict[str, Any] | None: + """Get a category by slug.""" + row = query_one(conn, "SELECT * FROM forum_categories WHERE slug = ?", [slug]) + return dict(row) if row else None + + +def get_topics_for_category( + conn: sqlite3.Connection, category_id: int, page: int = 1, per_page: int = 20 +) -> tuple[list[dict[str, Any]], int]: + """Get topics for a category with pagination.""" + offset = (page - 1) * per_page + + # Get total count + count_row = query_one( + conn, + "SELECT COUNT(*) as total FROM forum_topics WHERE category_id = ?", + [category_id], + ) + total = count_row["total"] if count_row else 0 + + # Get topics with reply counts + rows = query_all( + conn, + """ + SELECT + t.*, + (SELECT COUNT(*) FROM forum_replies WHERE topic_id = t.id) as reply_count, + (SELECT MAX(created_at) FROM forum_replies WHERE topic_id = t.id) as last_reply_at + FROM forum_topics t + WHERE t.category_id = ? + ORDER BY t.is_pinned DESC, COALESCE( + (SELECT MAX(created_at) FROM forum_replies WHERE topic_id = t.id), + t.updated_at + ) DESC + LIMIT ? OFFSET ? + """, + [category_id, per_page, offset], + ) + return [dict(row) for row in rows], total + + +def get_topic_by_id(conn: sqlite3.Connection, topic_id: int) -> dict[str, Any] | None: + """Get a topic by ID.""" + row = query_one(conn, "SELECT * FROM forum_topics WHERE id = ?", [topic_id]) + return dict(row) if row else None + + +def get_replies_for_topic(conn: sqlite3.Connection, topic_id: int) -> list[dict[str, Any]]: + """Get all replies for a topic.""" + rows = query_all( + conn, + "SELECT * FROM forum_replies WHERE topic_id = ? ORDER BY created_at", + [topic_id], + ) + return [dict(row) for row in rows] + + +def create_topic( + conn: sqlite3.Connection, + category_id: int, + title: str, + content: str, + author_name: str, + author_email: str | None = None, +) -> int: + """Create a new topic and return its ID.""" + slug = slugify(title) + cursor = conn.execute( + """INSERT INTO forum_topics + (category_id, title, slug, author_name, author_email, content) + VALUES (?, ?, ?, ?, ?, ?)""", + [category_id, title, slug, author_name, author_email, content], + ) + conn.commit() + return cursor.lastrowid + + +def create_reply( + conn: sqlite3.Connection, + topic_id: int, + content: str, + author_name: str, + author_email: str | None = None, +) -> int: + """Create a new reply and return its ID.""" + cursor = conn.execute( + """INSERT INTO forum_replies (topic_id, author_name, author_email, content) + VALUES (?, ?, ?, ?)""", + [topic_id, author_name, author_email, content], + ) + # Update topic's updated_at timestamp + conn.execute( + "UPDATE forum_topics SET updated_at = CURRENT_TIMESTAMP WHERE id = ?", + [topic_id], + ) + conn.commit() + return cursor.lastrowid + + +def increment_view_count(conn: sqlite3.Connection, topic_id: int) -> None: + """Increment a topic's view count.""" + conn.execute( + "UPDATE forum_topics SET view_count = view_count + 1 WHERE id = ?", + [topic_id], + ) + conn.commit() + + +def get_recent_topics(conn: sqlite3.Connection, limit: int = 5) -> list[dict[str, Any]]: + """Get the most recently active topics across all categories.""" + rows = query_all( + conn, + """ + SELECT + t.*, + c.name as category_name, + c.slug as category_slug, + (SELECT COUNT(*) FROM forum_replies WHERE topic_id = t.id) as reply_count + FROM forum_topics t + JOIN forum_categories c ON c.id = t.category_id + ORDER BY t.updated_at DESC + LIMIT ? + """, + [limit], + ) + return [dict(row) for row in rows] + + +def get_forum_stats(conn: sqlite3.Connection) -> dict[str, int]: + """Get overall forum statistics.""" + topics = query_one(conn, "SELECT COUNT(*) as count FROM forum_topics") + replies = query_one(conn, "SELECT COUNT(*) as count FROM forum_replies") + return { + "topics": topics["count"] if topics else 0, + "replies": replies["count"] if replies else 0, + } diff --git a/src/smarttools/web/forum/routes.py b/src/smarttools/web/forum/routes.py new file mode 100644 index 0000000..e1a5954 --- /dev/null +++ b/src/smarttools/web/forum/routes.py @@ -0,0 +1,291 @@ +"""Forum blueprint routes.""" + +from __future__ import annotations + +import html +import re +from functools import wraps +from typing import Callable + +from flask import ( + Blueprint, + abort, + flash, + g, + redirect, + render_template, + request, + session, + url_for, +) + +from smarttools.registry.db import connect_db + +from .models import ( + create_reply, + create_topic, + get_categories_with_stats, + get_category_by_slug, + get_forum_stats, + get_recent_topics, + get_replies_for_topic, + get_topic_by_id, + get_topics_for_category, + increment_view_count, + init_forum_schema, + seed_categories, +) + +forum_bp = Blueprint( + "forum", + __name__, + url_prefix="/forum", + template_folder="../templates/forum", +) + + +def get_db(): + """Get database connection, creating forum tables if needed.""" + if "forum_db" not in g: + g.forum_db = connect_db() + init_forum_schema(g.forum_db) + seed_categories(g.forum_db) + return g.forum_db + + +@forum_bp.before_request +def before_request(): + """Set up database connection before each request.""" + g.forum_db = get_db() + + +@forum_bp.teardown_request +def teardown_request(exception): + """Close database connection after each request.""" + db = g.pop("forum_db", None) + if db is not None: + db.close() + + +def validate_content(content: str, min_length: int = 10, max_length: int = 50000) -> str | None: + """Validate content and return error message if invalid.""" + if not content or len(content.strip()) < min_length: + return f"Content must be at least {min_length} characters." + if len(content) > max_length: + return f"Content must be less than {max_length} characters." + return None + + +def validate_name(name: str) -> str | None: + """Validate author name and return error message if invalid.""" + if not name or len(name.strip()) < 2: + return "Name must be at least 2 characters." + if len(name) > 100: + return "Name must be less than 100 characters." + # Basic sanitization - allow letters, numbers, spaces, and common punctuation + if not re.match(r"^[\w\s\-'.]+$", name, re.UNICODE): + return "Name contains invalid characters." + return None + + +def validate_email(email: str | None) -> str | None: + """Validate email if provided.""" + if not email: + return None # Email is optional + if len(email) > 254: + return "Email is too long." + if not re.match(r"^[^@]+@[^@]+\.[^@]+$", email): + return "Invalid email format." + return None + + +def validate_title(title: str) -> str | None: + """Validate topic title.""" + if not title or len(title.strip()) < 5: + return "Title must be at least 5 characters." + if len(title) > 200: + return "Title must be less than 200 characters." + return None + + +def honeypot_check(func: Callable) -> Callable: + """Decorator to check honeypot field for spam prevention.""" + @wraps(func) + def wrapper(*args, **kwargs): + # If the honeypot field is filled, it's likely a bot + if request.form.get("website_url"): # Hidden field bots often fill + # Silently redirect without processing + return redirect(url_for("forum.index")) + return func(*args, **kwargs) + return wrapper + + +# ───────────────────────────────────────────────────────────────────────────── +# Forum Index - List all categories +# ───────────────────────────────────────────────────────────────────────────── +@forum_bp.route("/") +def index(): + """Forum home - list all categories with stats.""" + db = get_db() + categories = get_categories_with_stats(db) + recent_topics = get_recent_topics(db, limit=5) + stats = get_forum_stats(db) + + return render_template( + "forum/index.html", + categories=categories, + recent_topics=recent_topics, + stats=stats, + ) + + +# ───────────────────────────────────────────────────────────────────────────── +# Category View - List topics in a category +# ───────────────────────────────────────────────────────────────────────────── +@forum_bp.route("/c/{{ category.description }}
+Be the first to start a discussion in this category.
+ + Create the first topic + ++ Ask questions, share your projects, and connect with other SmartTools users. +
+{{ cat.description }}
+No topics yet. Be the first to start a discussion!
++ {{ topic.view_count }} views · {{ replies|length }} replies +
+This topic is locked and cannot receive new replies.
+