notid/rothbard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

much better

Browse Source
This commit is contained in:
Bryce
2025-11-09 20:44:26 -08:00
parent b3a96cab1e
commit bafa9190e2
7 changed files with 362 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

231
app.py
View File

@@ -83,19 +83,46 @@ def login_required(view):
def get_user_profile(uid: str):
"""Fetch user's Firestore profile: users/{uid} => { enabled, caseEmail, is_admin }"""
"""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
doc_ref.set({"enabled": False}, merge=True)
return {"enabled": False, "caseEmail": None, "is_admin": False}
# 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)),
"caseEmail": data.get("caseEmail"),
"is_admin": bool(data.get("is_admin", 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():
"""Make user profile available in all templates"""
if 'uid' in session:
profile = get_user_profile(session['uid'])
return {'get_user_profile': lambda uid: profile if uid == session['uid'] else get_user_profile(uid)}
return {'get_user_profile': lambda uid: {}}
def get_firestore_document(collection_name: str, document_id: str):
"""
Retrieve a specific document from Firestore.
@@ -114,66 +141,13 @@ def get_firestore_document(collection_name: str, document_id: str):
else:
return None
def convert_to_pacific_time(date_str):
"""Convert UTC date string to Pacific Time and format as YYYY-MM-DD.
Args:
date_str (str): UTC date string in ISO 8601 format (e.g., "2025-10-24T19:20:22.377Z")
Returns:
str: Date formatted as YYYY-MM-DD in Pacific Time, or empty string if input is empty
"""
if not date_str:
return ''
try:
# Parse the UTC datetime
utc_time = datetime.fromisoformat(date_str.replace('Z', '+00:00'))
# Set timezone to UTC
utc_time = utc_time.replace(tzinfo=pytz.UTC)
# Convert to Pacific Time
pacific_time = utc_time.astimezone(pytz.timezone('America/Los_Angeles'))
# Format as YYYY-MM-DD
return pacific_time.strftime('%Y-%m-%d')
except (ValueError, AttributeError) as e:
print(f"[WARN] Date conversion failed for '{date_str}': {e}")
return ''
def fetch_all_projects():
"""Fetch all projects for a user and store them in Firestore"""
# This function is now only used by sync.py
# In production, this should be removed or marked as deprecated
print("Fetching projects....")
# Initialize Filevine client
client = FilevineClient()
bearer = client.get_bearer_token()
# List projects (all pages)
projects = client.list_all_projects()
projects = projects[:]
# Fetch details for each
detailed_rows = []
# This functionality has been moved to sync.py
# The worker_pool module has been removed
# This function is kept for backward compatibility but should not be used in production
print("[DEPRECATED] fetch_all_projects() is deprecated. Use sync.py instead.")
return []
@app.route("/")
def index():
uid = session.get("uid")
if not uid:
return redirect(url_for("login"))
profile = get_user_profile(uid)
if profile.get("enabled") and profile.get("caseEmail"):
if profile.get("enabled") and profile.get("case_email"):
return redirect(url_for("dashboard"))
return redirect(url_for("welcome"))
@@ -236,20 +210,22 @@ def dashboard(page=1):
if not profile.get("enabled"):
return redirect(url_for("welcome"))
# If user is admin and caseEmail query parameter is provided, use that instead
case_email = profile.get("caseEmail")
if profile.get("is_admin") and request.args.get('case_email'):
is_admin = profile.get("is_admin")
case_email = None
if not is_admin:
case_email = profile.get("case_email")
if not case_email:
return redirect(url_for("welcome"))
if is_admin and request.args.get('case_email'):
case_email = request.args.get('case_email').lower()
# Validate email format
if '@' not in case_email:
return abort(400, "Invalid email format")
if not case_email:
return redirect(url_for("welcome"))
# Pagination settings
per_page = 25
offset = (page - 1) * per_page
query = None
# Get total count efficiently using a count aggregation query
try:
@@ -258,7 +234,11 @@ def dashboard(page=1):
start_time = time.time()
projects_ref = db.collection("projects")
# Filter projects where case_email is in viewing_emails array
query = projects_ref.where("viewing_emails", "array_contains", case_email.lower())
if case_email:
query = projects_ref.where("viewing_emails", "array_contains", case_email.lower())
else:
query = projects_ref
total_projects = int(query.count().get()[0][0].value)
end_time = time.time()
print(f"Filtered projects count: {total_projects} (took {end_time - start_time:.2f}s)")
@@ -273,7 +253,11 @@ def dashboard(page=1):
import time
start_time = time.time()
# Filter projects where case_email is in viewing_emails array
projects_ref = db.collection("projects").where("viewing_emails", "array_contains", case_email.lower()).order_by("matter_description").limit(per_page).offset(offset)
if case_email:
projects_ref = db.collection("projects").where("viewing_emails", "array_contains", case_email.lower()).order_by("matter_description").limit(per_page).offset(offset)
else:
projects_ref = db.collection("projects").order_by("matter_description").limit(per_page).offset(offset)
docs = projects_ref.stream()
paginated_rows = []
@@ -295,6 +279,119 @@ def dashboard(page=1):
per_page=per_page)
@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")
# GAE compatibility
if __name__ == "__main__":
app.run(debug=True, host="0.0.0.0", port=int(os.environ.get("PORT", "5004")))