""" Bug Reports API endpoints. Routes for submitting, listing, and managing bug reports. """ from flask import Blueprint, request, session, jsonify, render_template, current_app from ...core.extensions import limiter from ...models.bug_reports import ( create_report, get_reports, get_report_by_id, add_admin_reply, add_broadcast_reply, add_user_reply, update_report_status, get_stats, ) from ...models.user import get_user_count, get_recent_users from ...models.announcements import ( create_announcement, get_announcements, get_announcement_by_id, update_announcement, delete_announcement, toggle_announcement, get_active_announcements, ) from ...models.user_activity import ping_user, get_active_users from ...core.config import Config import time # Track server start time for uptime display _SERVER_START_TIME = time.time() # API blueprint - registered under /api/ bug_reports_api_bp = Blueprint("bug_reports_api", __name__) # Page blueprint - registered at root level bug_reports_pages_bp = Blueprint("bug_reports_pages", __name__) def _require_auth(): """Return (user_id, username, avatar) if logged in, else None.""" if "username" in session and "_id" in session: return str(session["_id"]), session["username"], session.get("avatar") return None def _is_admin(): """Check if the current user is an admin.""" if "username" not in session: return False admin_usernames = getattr(Config, "ADMIN_USERNAMES", []) if isinstance(admin_usernames, str): admin_usernames = [u.strip().lower() for u in admin_usernames.split(",") if u.strip()] return session["username"].lower() in admin_usernames # ═══════════════════════════════════════════════════════════════ # PAGE ROUTES # ═══════════════════════════════════════════════════════════════ @bug_reports_pages_bp.route("/bug-report", methods=["GET"]) def bug_report_page(): """GET /bug-report — render the bug report submission page.""" return render_template("shared/bug_report.html") @bug_reports_pages_bp.route("/admin/bug-reports", methods=["GET"]) def admin_bug_reports_page(): """GET /admin/bug-reports — render the admin bug report management page.""" if not _is_admin(): return render_template("shared/404.html", error_message="Page not found"), 404 return render_template("shared/admin_bug_reports.html") @bug_reports_pages_bp.route("/admin", methods=["GET"]) def admin_dashboard_page(): """GET /admin — render the admin dashboard with stats.""" if not _is_admin(): return render_template("shared/404.html", error_message="Page not found"), 404 return render_template("shared/admin_dashboard.html") @bug_reports_pages_bp.route("/admin/announcements", methods=["GET"]) def admin_announcements_page(): """GET /admin/announcements — render the admin announcement management page.""" if not _is_admin(): return render_template("shared/404.html", error_message="Page not found"), 404 return render_template("shared/admin_announcements.html") # ═══════════════════════════════════════════════════════════════ # API: Submit a report # ═══════════════════════════════════════════════════════════════ @bug_reports_api_bp.route("/bug-reports", methods=["POST"]) @limiter.limit("3 per minute") def submit_report(): """POST /api/bug-reports — submit a new bug report.""" auth = _require_auth() if not auth: return jsonify({"success": False, "message": "Login required"}), 401 user_id, username, avatar = auth data = request.get_json() or {} title = data.get("title", "").strip() body = data.get("body", "").strip() category = data.get("category", "bug") page_url = data.get("page_url", "").strip() if not title or len(title) < 3: return jsonify({"success": False, "message": "Title must be at least 3 characters"}), 400 if not body or len(body) < 10: return jsonify({"success": False, "message": "Description must be at least 10 characters"}), 400 if len(title) > 200: return jsonify({"success": False, "message": "Title too long (max 200 chars)"}), 400 if len(body) > 5000: return jsonify({"success": False, "message": "Description too long (max 5000 chars)"}), 400 if category not in ("bug", "feature", "other"): category = "bug" report = create_report( title=title, body=body, category=category, author=username, author_id=user_id, author_avatar=avatar, page_url=page_url, ) if not report: return jsonify({"success": False, "message": "Failed to submit report"}), 500 return jsonify({"success": True, "report": report}), 201 # ═══════════════════════════════════════════════════════════════ # API: Get user's own reports # ═══════════════════════════════════════════════════════════════ @bug_reports_api_bp.route("/bug-reports/my", methods=["GET"]) def my_reports(): """GET /api/bug-reports/my — get the current user's reports.""" auth = _require_auth() if not auth: return jsonify({"success": False, "message": "Login required"}), 401 user_id, _, _ = auth page = request.args.get("page", 1, type=int) reports, total = get_reports(page=page, per_page=20, author_id=user_id) return jsonify({ "success": True, "reports": reports, "total": total, "page": page, }) # ═══════════════════════════════════════════════════════════════ # API: Admin list all reports with filtering # ═══════════════════════════════════════════════════════════════ @bug_reports_api_bp.route("/bug-reports/admin/list", methods=["GET"]) def admin_list_reports(): """GET /api/bug-reports/admin/list — get all reports (admin only).""" if not _is_admin(): return jsonify({"success": False, "message": "Unauthorized"}), 403 page = request.args.get("page", 1, type=int) status = request.args.get("status", "") if status not in ("open", "in_progress", "resolved", "closed", ""): status = "" per_page = request.args.get("per_page", 30, type=int) reports, total = get_reports(page=page, per_page=per_page, status=status if status else None) stats = get_stats() return jsonify({ "success": True, "reports": reports, "total": total, "page": page, "stats": stats, }) @bug_reports_api_bp.route("/bug-reports/admin/stats", methods=["GET"]) def admin_stats(): """GET /api/bug-reports/admin/stats — get report stats (admin only).""" if not _is_admin(): return jsonify({"success": False, "message": "Unauthorized"}), 403 return jsonify({"success": True, "stats": get_stats()}) @bug_reports_api_bp.route("/admin/dashboard", methods=["GET"]) def admin_dashboard_data(): """GET /api/admin/dashboard — get dashboard stats (admin only).""" if not _is_admin(): return jsonify({"success": False, "message": "Unauthorized"}), 403 user_count = get_user_count() recent_users = get_recent_users(limit=10) bug_stats = get_stats() # Sanitize recent users (remove sensitive fields) safe_users = [] for u in recent_users: safe_users.append({ "_id": str(u.get("_id")), "username": u.get("username"), "avatar": u.get("avatar"), "created_at": u.get("created_at").isoformat() if u.get("created_at") else None, "auth_method": u.get("auth_method", "local"), }) # Get active users active_users = get_active_users(minutes=5) active_count = len(active_users) return jsonify({ "success": True, "server_start_time": _SERVER_START_TIME, "users": { "total": user_count, "recent": safe_users, "active": active_users, "active_count": active_count, }, "bugs": bug_stats, }) # ═══════════════════════════════════════════════════════════════ # ANNOUNCEMENTS API (admin only) # ═══════════════════════════════════════════════════════════════ @bug_reports_api_bp.route("/announcements", methods=["GET"]) def list_announcements(): """GET /api/announcements — list all announcements (admin).""" if not _is_admin(): return jsonify({"success": False, "message": "Unauthorized"}), 403 page = request.args.get("page", 1, type=int) announcements, total = get_announcements(page=page, per_page=50) return jsonify({"success": True, "announcements": announcements, "total": total, "page": page}) @bug_reports_api_bp.route("/announcements/active", methods=["GET"]) def list_active_announcements(): """GET /api/announcements/active — get all active announcements (public).""" announcements = get_active_announcements() return jsonify({"success": True, "announcements": announcements}) @bug_reports_api_bp.route("/announcements", methods=["POST"]) @limiter.limit("10 per minute") def add_announcement(): """POST /api/announcements — create a new announcement (admin only).""" if not _is_admin(): return jsonify({"success": False, "message": "Unauthorized"}), 403 data = request.get_json() or {} title = data.get("title", "").strip() message = data.get("message", "").strip() a_type = data.get("type", "info") active = data.get("active", False) priority = data.get("priority", 0) if not title or len(title) < 2: return jsonify({"success": False, "message": "Title must be at least 2 characters"}), 400 if not message or len(message) < 2: return jsonify({"success": False, "message": "Message must be at least 2 characters"}), 400 if len(title) > 200: return jsonify({"success": False, "message": "Title too long (max 200 chars)"}), 400 if len(message) > 2000: return jsonify({"success": False, "message": "Message too long (max 2000 chars)"}), 400 ann = create_announcement(title, message, a_type, active, priority) if not ann: return jsonify({"success": False, "message": "Failed to create announcement"}), 500 return jsonify({"success": True, "announcement": ann}), 201 @bug_reports_api_bp.route("/announcements/", methods=["PATCH"]) @limiter.limit("20 per minute") def edit_announcement(announcement_id): """PATCH /api/announcements/ — update an announcement (admin only).""" if not _is_admin(): return jsonify({"success": False, "message": "Unauthorized"}), 403 data = request.get_json() or {} success = update_announcement(announcement_id, data) if not success: return jsonify({"success": False, "message": "Announcement not found or no changes"}), 404 ann = get_announcement_by_id(announcement_id) return jsonify({"success": True, "announcement": ann}) @bug_reports_api_bp.route("/announcements/", methods=["DELETE"]) def remove_announcement(announcement_id): """DELETE /api/announcements/ — delete an announcement (admin only).""" if not _is_admin(): return jsonify({"success": False, "message": "Unauthorized"}), 403 success = delete_announcement(announcement_id) if not success: return jsonify({"success": False, "message": "Announcement not found"}), 404 return jsonify({"success": True}) # ═══════════════════════════════════════════════════════════════ # USER ACTIVITY API # ═══════════════════════════════════════════════════════════════ @bug_reports_api_bp.route("/activity/ping", methods=["POST"]) def user_heartbeat(): """POST /api/activity/ping — record user activity. Works for both logged-in and anonymous visitors. """ data = request.get_json() or {} page_url = data.get("page_url", request.referrer or "/") page_title = data.get("page_title", "") anonymous_id = data.get("anonymous_id") ip_address = request.remote_addr or "" if "username" in session and "_id" in session: # Logged-in user ping_user( user_id=str(session["_id"]), username=session["username"], avatar=session.get("avatar", ""), page_url=page_url, page_title=page_title, ip_address=ip_address, ) elif anonymous_id: # Anonymous visitor with client ID ping_user( user_id=None, username=None, avatar=None, page_url=page_url, page_title=page_title, anonymous_id=anonymous_id, ip_address=ip_address, ) else: # Anonymous without ID — skip silently return jsonify({"success": False, "message": "No identifier"}), 200 return jsonify({"success": True}) @bug_reports_api_bp.route("/activity/active", methods=["GET"]) def get_active_users_endpoint(): """GET /api/activity/active — get active users (admin only).""" if not _is_admin(): return jsonify({"success": False, "message": "Unauthorized"}), 403 minutes = request.args.get("minutes", 5, type=int) active = get_active_users(minutes=minutes) count = len(active) return jsonify({"success": True, "active_users": active, "count": count}) @bug_reports_api_bp.route("/announcements//toggle", methods=["POST"]) def toggle_announcement_active(announcement_id): """POST /api/announcements//toggle — toggle active status (admin only).""" if not _is_admin(): return jsonify({"success": False, "message": "Unauthorized"}), 403 data = request.get_json() or {} active = data.get("active", False) success = toggle_announcement(announcement_id, active) if not success: return jsonify({"success": False, "message": "Announcement not found"}), 404 ann = get_announcement_by_id(announcement_id) return jsonify({"success": True, "announcement": ann}) # ═══════════════════════════════════════════════════════════════ # API: View a single report # ═══════════════════════════════════════════════════════════════ @bug_reports_api_bp.route("/bug-reports/", methods=["GET"]) def get_report(report_id): """GET /api/bug-reports/ — get a single report.""" report = get_report_by_id(report_id) if not report: return jsonify({"success": False, "message": "Report not found"}), 404 return jsonify({"success": True, "report": report}) # ═══════════════════════════════════════════════════════════════ # API: Reply to a report (individual) # ═══════════════════════════════════════════════════════════════ @bug_reports_api_bp.route("/bug-reports//reply", methods=["POST"]) @limiter.limit("10 per minute") def reply_to_report(report_id): """POST /api/bug-reports//reply — admin replies to a report.""" if not _is_admin(): return jsonify({"success": False, "message": "Unauthorized"}), 403 data = request.get_json() or {} body = data.get("body", "").strip() if not body: return jsonify({"success": False, "message": "Reply cannot be empty"}), 400 if len(body) > 5000: return jsonify({"success": False, "message": "Reply too long (max 5000 chars)"}), 400 success = add_admin_reply(report_id, body, is_broadcast=False) if not success: return jsonify({"success": False, "message": "Report not found"}), 404 return jsonify({"success": True}) # ═══════════════════════════════════════════════════════════════ # API: Group reply (broadcast to multiple reports) # ═══════════════════════════════════════════════════════════════ @bug_reports_api_bp.route("/bug-reports/broadcast-reply", methods=["POST"]) @limiter.limit("5 per minute") def broadcast_reply(): """POST /api/bug-reports/broadcast-reply — admin replies to multiple reports at once.""" if not _is_admin(): return jsonify({"success": False, "message": "Unauthorized"}), 403 data = request.get_json() or {} report_ids = data.get("report_ids", []) body = data.get("body", "").strip() if not report_ids or not isinstance(report_ids, list): return jsonify({"success": False, "message": "No reports selected"}), 400 if not body: return jsonify({"success": False, "message": "Reply cannot be empty"}), 400 if len(body) > 5000: return jsonify({"success": False, "message": "Reply too long (max 5000 chars)"}), 400 count = add_broadcast_reply(report_ids, body) return jsonify({"success": True, "replied_to": count}) # ═══════════════════════════════════════════════════════════════ # API: User replies to their own report # ═══════════════════════════════════════════════════════════════ @bug_reports_api_bp.route("/bug-reports//user-reply", methods=["POST"]) @limiter.limit("5 per minute") def user_reply_to_report(report_id): """POST /api/bug-reports//user-reply — the report author replies to admin.""" auth = _require_auth() if not auth: return jsonify({"success": False, "message": "Login required"}), 401 user_id, username, _ = auth # Verify the report exists and belongs to this user report = get_report_by_id(report_id) if not report: return jsonify({"success": False, "message": "Report not found"}), 404 if report["author_id"] != user_id: return jsonify({"success": False, "message": "Unauthorized"}), 403 data = request.get_json() or {} body = data.get("body", "").strip() if not body: return jsonify({"success": False, "message": "Reply cannot be empty"}), 400 if len(body) > 5000: return jsonify({"success": False, "message": "Reply too long (max 5000 chars)"}), 400 success = add_user_reply(report_id, username, user_id, body) if not success: return jsonify({"success": False, "message": "Failed to add reply"}), 500 return jsonify({"success": True}) # ═══════════════════════════════════════════════════════════════ # API: Update report status # ═══════════════════════════════════════════════════════════════ @bug_reports_api_bp.route("/bug-reports//status", methods=["PATCH"]) @limiter.limit("20 per minute") def change_report_status(report_id): """PATCH /api/bug-reports//status — update report status (admin only).""" if not _is_admin(): return jsonify({"success": False, "message": "Unauthorized"}), 403 data = request.get_json() or {} new_status = data.get("status", "") success = update_report_status(report_id, new_status) if not success: return jsonify({"success": False, "message": "Failed to update status"}), 400 return jsonify({"success": True})