admin panel

Author: reckt32

.env.example

@@ -44,3 +44,7 @@ REPORT_PRICE_PAISE=99900
 # - Get Firebase service account from: Firebase Console > Project Settings > Service Accounts > Generate New Private Key
 # - Get Razorpay keys from: Dashboard > Settings > API Keys (generate separately in Test and Live mode)
 # - Set RAZORPAY_MODE=live when ready for production payments
+
+# 6) Admin Panel
+# Comma-separated list of email addresses that have admin access
+ADMIN_EMAILS=admin@example.com

app.py

@@ -41,10 +41,15 @@
     update_questionnaire_upload_metadata,
     get_user_by_firebase_uid,
     create_or_update_user,
+    list_all_users,
+    delete_user,
+    set_user_credits,
+    get_user_count,
+    get_payment_history,
 )
 
 # Auth and Payment modules
-from auth import require_auth, require_payment, consume_credit, verify_firebase_token, optional_auth
+from auth import require_auth, require_payment, consume_credit, verify_firebase_token, optional_auth, require_admin
 from payment import (
     create_razorpay_order,
     verify_razorpay_signature,
@@ -6157,8 +6162,111 @@ def payment_reconcile():
             "message": "Could not check payment status with Razorpay"
         }), 500
 
+# =============================================================================
+# ADMIN ROUTES
+# =============================================================================
+
+@app.route("/api/admin/stats", methods=["GET"])
+@require_auth
+@require_admin
+def admin_stats():
+    """Get dashboard statistics: total users, paid users, users with credits."""
+    counts = get_user_count()
+    return jsonify({"success": True, "stats": counts}), 200
+
+
+@app.route("/api/admin/users", methods=["GET"])
+@require_auth
+@require_admin
+def admin_list_users():
+    """
+    List all users with pagination and optional search.
+    Query params: page (default 1), per_page (default 25), search (optional)
+    """
+    page = request.args.get("page", 1, type=int)
+    per_page = request.args.get("per_page", 25, type=int)
+    search = request.args.get("search", "").strip() or None
+
+    # Clamp per_page to prevent abuse
+    per_page = min(max(per_page, 1), 100)
+
+    users, total = list_all_users(page=page, per_page=per_page, search=search)
+    return jsonify({
+        "success": True,
+        "users": users,
+        "pagination": {
+            "page": page,
+            "per_page": per_page,
+            "total": total,
+            "pages": (total + per_page - 1) // per_page if per_page else 1,
+        },
+    }), 200
+
+
+@app.route("/api/admin/users/<firebase_uid>", methods=["GET"])
+@require_auth
+@require_admin
+def admin_get_user(firebase_uid):
+    """Get single user details including payment history."""
+    user = get_user_by_firebase_uid(firebase_uid)
+    if not user:
+        return jsonify({"error": "User not found"}), 404
+
+    payments = get_payment_history(firebase_uid)
+    return jsonify({
+        "success": True,
+        "user": user,
+        "payments": payments,
+    }), 200
+
+
+@app.route("/api/admin/users/<firebase_uid>", methods=["DELETE"])
+@require_auth
+@require_admin
+def admin_delete_user(firebase_uid):
+    """Delete a user and their payment history."""
+    deleted = delete_user(firebase_uid)
+    if not deleted:
+        return jsonify({"error": "User not found"}), 404
+    return jsonify({"success": True, "message": "User deleted"}), 200
+
+
+@app.route("/api/admin/users/<firebase_uid>/credits", methods=["POST"])
+@require_auth
+@require_admin
+def admin_set_credits(firebase_uid):
+    """
+    Set or add credits for a user.
+    Request body: { "credits": int, "mode": "set" | "add" }
+    """
+    data = request.get_json(force=True) or {}
+    credits = data.get("credits")
+    mode = data.get("mode", "set")
+
+    if credits is None or not isinstance(credits, int) or credits < 0:
+        return jsonify({"error": "Valid non-negative integer 'credits' required"}), 400
+
+    user = get_user_by_firebase_uid(firebase_uid)
+    if not user:
+        return jsonify({"error": "User not found"}), 404
+
+    if mode == "add":
+        from db import add_user_credits
+        success = add_user_credits(firebase_uid, credits)
+    else:
+        success = set_user_credits(firebase_uid, credits)
+
+    if not success:
+        return jsonify({"error": "Failed to update credits"}), 500
+
+    updated_user = get_user_by_firebase_uid(firebase_uid)
+    return jsonify({
+        "success": True,
+        "message": f"Credits {'added' if mode == 'add' else 'set'} successfully",
+        "report_credits": updated_user.get("report_credits", 0),
+    }), 200
+
 
 if __name__ == '__main__':
     port = int(os.environ.get('PORT', 5000))
     app.run(host='0.0.0.0', port=port, debug=False)
-

auth.py

@@ -253,3 +253,43 @@ def decorated_function(*args, **kwargs):
 
         return f(*args, **kwargs)
     return decorated_function
+
+
+def require_admin(f):
+    """
+    Decorator that requires the authenticated user to be an admin.
+
+    Must be used AFTER @require_auth decorator.
+    Checks g.firebase_token["email"] against the ADMIN_EMAILS environment variable
+    (comma-separated list of admin email addresses).
+    Returns 403 Forbidden if not an admin.
+    """
+    @functools.wraps(f)
+    def decorated_function(*args, **kwargs):
+        # Ensure require_auth was called first
+        if not hasattr(g, 'firebase_token') or g.firebase_token is None:
+            return jsonify({
+                "error": "Authentication required",
+                "message": "This endpoint requires authentication"
+            }), 401
+
+        user_email = (g.firebase_token.get("email") or "").lower().strip()
+        admin_emails_raw = os.getenv("ADMIN_EMAILS", "")
+        admin_emails = [
+            e.strip().lower() for e in admin_emails_raw.split(",") if e.strip()
+        ]
+
+        if not admin_emails:
+            return jsonify({
+                "error": "Admin not configured",
+                "message": "No admin emails have been configured on the server"
+            }), 403
+
+        if user_email not in admin_emails:
+            return jsonify({
+                "error": "Forbidden",
+                "message": "You do not have admin access"
+            }), 403
+
+        return f(*args, **kwargs)
+    return decorated_function

db.py

@@ -625,6 +625,82 @@ def migrate_existing_paid_users(credits: int = 3) -> int:
     return result.rowcount
 
 
+# --- Admin helpers ---
+
+def list_all_users(page: int = 1, per_page: int = 25, search: Optional[str] = None):
+    """List all users with pagination and optional email search. Returns (users, total)."""
+    offset = (page - 1) * per_page
+
+    if search:
+        like = f"%{search}%"
+        total_rows = _query(
+            "SELECT COUNT(*) AS cnt FROM users WHERE email LIKE ? OR display_name LIKE ?",
+            (like, like),
+        )
+        total = total_rows[0]["cnt"] if total_rows else 0
+        rows = _query(
+            """
+            SELECT id, firebase_uid, email, display_name, has_paid,
+                   report_credits, payment_date, created_at, updated_at
+            FROM users
+            WHERE email LIKE ? OR display_name LIKE ?
+            ORDER BY created_at DESC
+            LIMIT ? OFFSET ?
+            """,
+            (like, like, per_page, offset),
+        )
+    else:
+        total_rows = _query("SELECT COUNT(*) AS cnt FROM users")
+        total = total_rows[0]["cnt"] if total_rows else 0
+        rows = _query(
+            """
+            SELECT id, firebase_uid, email, display_name, has_paid,
+                   report_credits, payment_date, created_at, updated_at
+            FROM users
+            ORDER BY created_at DESC
+            LIMIT ? OFFSET ?
+            """,
+            (per_page, offset),
+        )
+
+    users = [dict(r) for r in rows]
+    return users, total
+
+
+def delete_user(firebase_uid: str) -> bool:
+    """Delete a user and their payment history. Returns True if user existed."""
+    user = get_user_by_firebase_uid(firebase_uid)
+    if not user:
+        return False
+    _exec("DELETE FROM payments WHERE firebase_uid = ?", (firebase_uid,))
+    _exec("DELETE FROM users WHERE firebase_uid = ?", (firebase_uid,))
+    return True
+
+
+def set_user_credits(firebase_uid: str, credits: int) -> bool:
+    """Set credits to an exact value. Returns True if updated."""
+    result = _exec(
+        """
+        UPDATE users SET report_credits = ?,
+        updated_at = CURRENT_TIMESTAMP WHERE firebase_uid = ?
+        """,
+        (credits, firebase_uid),
+    )
+    return result.rowcount > 0
+
+
+def get_user_count() -> Dict[str, int]:
+    """Get total and paid user counts for dashboard stats."""
+    total_rows = _query("SELECT COUNT(*) AS cnt FROM users")
+    paid_rows = _query("SELECT COUNT(*) AS cnt FROM users WHERE has_paid = 1")
+    credits_rows = _query("SELECT COUNT(*) AS cnt FROM users WHERE report_credits > 0")
+    return {
+        "total": total_rows[0]["cnt"] if total_rows else 0,
+        "paid": paid_rows[0]["cnt"] if paid_rows else 0,
+        "with_credits": credits_rows[0]["cnt"] if credits_rows else 0,
+    }
+
+
 # Initialize schema on import
 init_db()