progress.
This commit is contained in:
112
admin.py
Normal file
112
admin.py
Normal file
@@ -0,0 +1,112 @@
|
||||
import json
|
||||
import os
|
||||
from functools import wraps
|
||||
from flask import render_template, request, redirect, url_for, session, abort, jsonify
|
||||
from firebase_init import db
|
||||
from firebase_admin import auth as fb_auth
|
||||
from utils import get_user_profile
|
||||
|
||||
def admin_required(view):
|
||||
@wraps(view)
|
||||
def wrapped(*args, **kwargs):
|
||||
uid = session.get("uid")
|
||||
if not uid:
|
||||
return redirect(url_for("login"))
|
||||
profile = get_user_profile(uid)
|
||||
if not profile.get("is_admin"):
|
||||
abort(403, "Access denied. Admin privileges required.")
|
||||
return view(*args, **kwargs)
|
||||
return wrapped
|
||||
|
||||
def register_admin_routes(app):
|
||||
@app.route("/admin/users")
|
||||
@app.route("/admin/users/<int:page>")
|
||||
@admin_required
|
||||
def admin_users(page=1):
|
||||
"""Admin page to manage all users"""
|
||||
# Pagination settings
|
||||
per_page = 25
|
||||
offset = (page - 1) * per_page
|
||||
|
||||
# Get all users from Firestore
|
||||
try:
|
||||
# Get total count of users
|
||||
users_ref = db.collection("users")
|
||||
total_users = int(users_ref.count().get()[0][0].value)
|
||||
|
||||
# Get users for current page
|
||||
users_query = users_ref.order_by("user_email").limit(per_page).offset(offset)
|
||||
docs = users_query.stream()
|
||||
|
||||
users = []
|
||||
for doc in docs:
|
||||
user_data = doc.to_dict()
|
||||
users.append({
|
||||
"uid": doc.id,
|
||||
"user_email": user_data.get("user_email", ""),
|
||||
"case_email": user_data.get("case_email", ""),
|
||||
"enabled": bool(user_data.get("enabled", False)),
|
||||
"is_admin": bool(user_data.get("is_admin", False))
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
print(f"[ERR] Failed to fetch users: {e}")
|
||||
users = []
|
||||
total_users = 0
|
||||
|
||||
# Calculate pagination
|
||||
total_pages = (total_users + per_page - 1) // per_page # Ceiling division
|
||||
|
||||
return render_template("admin_users.html",
|
||||
users=users,
|
||||
current_page=page,
|
||||
total_pages=total_pages,
|
||||
total_users=total_users,
|
||||
per_page=per_page)
|
||||
|
||||
@app.route("/admin/users/<uid>")
|
||||
@admin_required
|
||||
def admin_user_detail(uid):
|
||||
"""Show user details for admin"""
|
||||
# Get user data
|
||||
user_doc = db.collection("users").document(uid).get()
|
||||
if not user_doc.exists:
|
||||
abort(404, "User not found")
|
||||
|
||||
user_data = user_doc.to_dict()
|
||||
user = {
|
||||
"uid": uid,
|
||||
"user_email": user_data.get("user_email", ""),
|
||||
"case_email": user_data.get("case_email", ""),
|
||||
"enabled": bool(user_data.get("enabled", False)),
|
||||
"is_admin": bool(user_data.get("is_admin", False))
|
||||
}
|
||||
|
||||
return render_template("admin_user_edit.html", user=user)
|
||||
|
||||
@app.route("/admin/users/update", methods=["POST"])
|
||||
@admin_required
|
||||
def update_user():
|
||||
"""Update user information"""
|
||||
try:
|
||||
data = request.get_json()
|
||||
if not data:
|
||||
abort(400, "Invalid JSON data")
|
||||
|
||||
target_uid = data.get("uid")
|
||||
if not target_uid:
|
||||
abort(400, "User ID is required")
|
||||
|
||||
# Update user in Firestore
|
||||
user_ref = db.collection("users").document(target_uid)
|
||||
user_ref.update({
|
||||
"enabled": data.get("enabled", False),
|
||||
"is_admin": data.get("is_admin", False),
|
||||
"case_email": data.get("case_email", "")
|
||||
})
|
||||
|
||||
return jsonify({"success": True})
|
||||
|
||||
except Exception as e:
|
||||
print(f"[ERR] Failed to update user: {e}")
|
||||
abort(500, "Failed to update user")
|
||||
169
app.py
169
app.py
@@ -7,28 +7,17 @@ import pytz
|
||||
from flask import Flask, render_template, request, redirect, url_for, session, abort, jsonify
|
||||
from dotenv import load_dotenv
|
||||
load_dotenv()
|
||||
import firebase_admin
|
||||
from firebase_admin import credentials, auth as fb_auth, firestore
|
||||
import requests
|
||||
from filevine_client import FilevineClient
|
||||
from utils import get_user_profile
|
||||
from firebase_init import db
|
||||
from firebase_admin import auth as fb_auth
|
||||
|
||||
|
||||
app = Flask(__name__)
|
||||
app.secret_key = os.environ.get("FLASK_SECRET_KEY", os.urandom(32))
|
||||
|
||||
# --- Firebase Admin init ---
|
||||
_creds = None
|
||||
json_inline = os.environ.get("FIREBASE_SERVICE_ACCOUNT_JSON")
|
||||
file_path = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS")
|
||||
if json_inline:
|
||||
_creds = credentials.Certificate(json.loads(json_inline))
|
||||
elif file_path and os.path.exists(file_path):
|
||||
_creds = credentials.Certificate(file_path)
|
||||
else:
|
||||
raise RuntimeError("Firebase credentials not configured. Set GOOGLE_APPLICATION_CREDENTIALS or FIREBASE_SERVICE_ACCOUNT_JSON.")
|
||||
|
||||
firebase_admin.initialize_app(_creds)
|
||||
db = firestore.client()
|
||||
# --- Firebase Admin init --- (moved to firebase_init.py)
|
||||
|
||||
# --- Filevine env ---
|
||||
FV_CLIENT_ID = os.environ.get("FILEVINE_CLIENT_ID")
|
||||
@@ -40,9 +29,6 @@ FV_USER_ID = os.environ.get("FILEVINE_USER_ID")
|
||||
if not all([FV_CLIENT_ID, FV_CLIENT_SECRET, FV_PAT, FV_ORG_ID, FV_USER_ID]):
|
||||
print("[WARN] Missing one or more Filevine env vars — dashboard will fail until set.")
|
||||
|
||||
# --- Cache ---
|
||||
# No longer using cache - projects are stored in Firestore
|
||||
|
||||
PHASES = {
|
||||
209436: "Nonpayment File Review",
|
||||
209437: "Attorney File Review",
|
||||
@@ -82,38 +68,6 @@ def login_required(view):
|
||||
return wrapped
|
||||
|
||||
|
||||
def get_user_profile(uid: str):
|
||||
"""Fetch user's Firestore profile: users/{uid} => { enabled, case_email, is_admin, user_email }"""
|
||||
doc_ref = db.collection("users").document(uid)
|
||||
snap = doc_ref.get()
|
||||
if not snap.exists:
|
||||
# bootstrap a placeholder doc so admins can fill it in
|
||||
# Get user email from Firebase Auth
|
||||
try:
|
||||
user = fb_auth.get_user(uid)
|
||||
user_email = user.email
|
||||
except Exception:
|
||||
user_email = None
|
||||
|
||||
doc_ref.set({
|
||||
"enabled": False,
|
||||
"is_admin": False,
|
||||
"user_email": user_email,
|
||||
"case_email": user_email
|
||||
}, merge=True)
|
||||
return {
|
||||
"enabled": False,
|
||||
"is_admin": False,
|
||||
"user_email": user_email,
|
||||
"case_email": user_email
|
||||
}
|
||||
data = snap.to_dict() or {}
|
||||
return {
|
||||
"enabled": bool(data.get("enabled", False)),
|
||||
"is_admin": bool(data.get("is_admin", False)),
|
||||
"user_email": data.get("user_email"),
|
||||
"case_email": data.get("case_email")
|
||||
}
|
||||
|
||||
@app.context_processor
|
||||
def inject_user_profile():
|
||||
@@ -278,119 +232,10 @@ def dashboard(page=1):
|
||||
total_projects=total_projects,
|
||||
per_page=per_page)
|
||||
|
||||
import admin
|
||||
|
||||
@app.route("/admin/users")
|
||||
@app.route("/admin/users/<int:page>")
|
||||
@login_required
|
||||
def admin_users(page=1):
|
||||
"""Admin page to manage all users"""
|
||||
uid = session.get("uid")
|
||||
profile = get_user_profile(uid)
|
||||
|
||||
# Only admins can access this page
|
||||
if not profile.get("is_admin"):
|
||||
abort(403, "Access denied. Admin privileges required.")
|
||||
|
||||
# Pagination settings
|
||||
per_page = 25
|
||||
offset = (page - 1) * per_page
|
||||
|
||||
# Get all users from Firestore
|
||||
try:
|
||||
# Get total count of users
|
||||
users_ref = db.collection("users")
|
||||
total_users = int(users_ref.count().get()[0][0].value)
|
||||
|
||||
# Get users for current page
|
||||
users_query = users_ref.order_by("user_email").limit(per_page).offset(offset)
|
||||
docs = users_query.stream()
|
||||
|
||||
users = []
|
||||
for doc in docs:
|
||||
user_data = doc.to_dict()
|
||||
users.append({
|
||||
"uid": doc.id,
|
||||
"user_email": user_data.get("user_email", ""),
|
||||
"case_email": user_data.get("case_email", ""),
|
||||
"enabled": bool(user_data.get("enabled", False)),
|
||||
"is_admin": bool(user_data.get("is_admin", False))
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
print(f"[ERR] Failed to fetch users: {e}")
|
||||
users = []
|
||||
total_users = 0
|
||||
|
||||
# Calculate pagination
|
||||
total_pages = (total_users + per_page - 1) // per_page # Ceiling division
|
||||
|
||||
return render_template("admin_users.html",
|
||||
users=users,
|
||||
current_page=page,
|
||||
total_pages=total_pages,
|
||||
total_users=total_users,
|
||||
per_page=per_page)
|
||||
|
||||
@app.route("/admin/users/<uid>")
|
||||
@login_required
|
||||
def admin_user_detail(uid):
|
||||
"""Show user details for admin"""
|
||||
uid = session.get("uid")
|
||||
profile = get_user_profile(uid)
|
||||
|
||||
# Only admins can access this page
|
||||
if not profile.get("is_admin"):
|
||||
abort(403, "Access denied. Admin privileges required.")
|
||||
|
||||
# Get user data
|
||||
user_doc = db.collection("users").document(uid).get()
|
||||
if not user_doc.exists:
|
||||
abort(404, "User not found")
|
||||
|
||||
user_data = user_doc.to_dict()
|
||||
user = {
|
||||
"uid": uid,
|
||||
"user_email": user_data.get("user_email", ""),
|
||||
"case_email": user_data.get("case_email", ""),
|
||||
"enabled": bool(user_data.get("enabled", False)),
|
||||
"is_admin": bool(user_data.get("is_admin", False))
|
||||
}
|
||||
|
||||
return render_template("admin_user_edit.html", user=user)
|
||||
|
||||
@app.route("/admin/users/update", methods=["POST"])
|
||||
@login_required
|
||||
def update_user():
|
||||
"""Update user information"""
|
||||
uid = session.get("uid")
|
||||
profile = get_user_profile(uid)
|
||||
|
||||
# Only admins can update users
|
||||
if not profile.get("is_admin"):
|
||||
abort(403, "Access denied. Admin privileges required.")
|
||||
|
||||
try:
|
||||
data = request.get_json()
|
||||
if not data:
|
||||
abort(400, "Invalid JSON data")
|
||||
|
||||
target_uid = data.get("uid")
|
||||
if not target_uid:
|
||||
abort(400, "User ID is required")
|
||||
|
||||
# Update user in Firestore
|
||||
user_ref = db.collection("users").document(target_uid)
|
||||
user_ref.update({
|
||||
"enabled": data.get("enabled", False),
|
||||
"is_admin": data.get("is_admin", False),
|
||||
"case_email": data.get("case_email", "")
|
||||
})
|
||||
|
||||
return jsonify({"success": True})
|
||||
|
||||
except Exception as e:
|
||||
print(f"[ERR] Failed to update user: {e}")
|
||||
abort(500, "Failed to update user")
|
||||
# Register admin routes
|
||||
admin.register_admin_routes(app)
|
||||
|
||||
# GAE compatibility
|
||||
if __name__ == "__main__":
|
||||
|
||||
19
firebase_init.py
Normal file
19
firebase_init.py
Normal file
@@ -0,0 +1,19 @@
|
||||
import os
|
||||
from firebase_admin import credentials, initialize_app, firestore
|
||||
|
||||
# Load credentials
|
||||
_creds = None
|
||||
json_inline = os.environ.get("FIREBASE_SERVICE_ACCOUNT_JSON")
|
||||
file_path = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS")
|
||||
if json_inline:
|
||||
_creds = credentials.Certificate(json.loads(json_inline))
|
||||
elif file_path and os.path.exists(file_path):
|
||||
_creds = credentials.Certificate(file_path)
|
||||
else:
|
||||
raise RuntimeError("Firebase credentials not configured. Set GOOGLE_APPLICATION_CREDENTIALS or FIREBASE_SERVICE_ACCOUNT_JSON.")
|
||||
|
||||
# Initialize Firebase Admin SDK
|
||||
firebase_admin_app = initialize_app(_creds)
|
||||
|
||||
# Create Firestore client
|
||||
db = firestore.client()
|
||||
36
utils.py
Normal file
36
utils.py
Normal file
@@ -0,0 +1,36 @@
|
||||
|
||||
from firebase_init import db
|
||||
from firebase_admin import auth as fb_auth
|
||||
|
||||
def get_user_profile(uid: str):
|
||||
"""Fetch user's Firestore profile: users/{uid} => { enabled, case_email, is_admin, user_email }"""
|
||||
doc_ref = db.collection("users").document(uid)
|
||||
snap = doc_ref.get()
|
||||
if not snap.exists:
|
||||
# bootstrap a placeholder doc so admins can fill it in
|
||||
# Get user email from Firebase Auth
|
||||
try:
|
||||
user = fb_auth.get_user(uid)
|
||||
user_email = user.email
|
||||
except Exception:
|
||||
user_email = None
|
||||
|
||||
doc_ref.set({
|
||||
"enabled": False,
|
||||
"is_admin": False,
|
||||
"user_email": user_email,
|
||||
"case_email": user_email
|
||||
}, merge=True)
|
||||
return {
|
||||
"enabled": False,
|
||||
"is_admin": False,
|
||||
"user_email": user_email,
|
||||
"case_email": user_email
|
||||
}
|
||||
data = snap.to_dict() or {}
|
||||
return {
|
||||
"enabled": bool(data.get("enabled", False)),
|
||||
"is_admin": bool(data.get("is_admin", False)),
|
||||
"user_email": data.get("user_email"),
|
||||
"case_email": data.get("case_email")
|
||||
}
|
||||
Reference in New Issue
Block a user