notid/rothbard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
cfd7020e1b07f57f866bec4a971bb75fd1cf445c
rothbard/app.py
Bryce cfd7020e1b multi
2025-11-07 15:03:20 -08:00

567 lines
22 KiB
Python
Raw Blame History

import json
import os
from functools import wraps
from datetime import datetime, timedelta
import pytz
from flask import Flask, render_template, request, redirect, url_for, session, abort, jsonify
from dotenv import load_dotenv
import firebase_admin
from firebase_admin import credentials, auth as fb_auth, firestore
import requests
load_dotenv()
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()
# --- Filevine env ---
FV_CLIENT_ID = os.environ.get("FILEVINE_CLIENT_ID")
FV_CLIENT_SECRET = os.environ.get("FILEVINE_CLIENT_SECRET")
FV_PAT = os.environ.get("FILEVINE_PERSONAL_ACCESS_TOKEN")
FV_ORG_ID = os.environ.get("FILEVINE_ORG_ID")
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",
209438: "Notice Preparation",
209439: "Notice Pending",
209440: "Notice Expired",
209442: "Preparing and Filing UD",
209443: "Waiting for Answer",
209444: "Archived",
210761: "Service of Process",
211435: "Default",
211436: "Pre-Answer Motion",
211437: "Request for Trial",
211438: "Trial Prep and Trial",
211439: "Writ and Sheriff",
211440: "Lockout Pending",
211441: "Stipulation Preparation",
211442: "Stipulation Pending",
211443: "Stipulation Expired",
211446: "On Hold",
211466: "Request for Monetary Judgment",
211467: "Appeals and Post-Poss. Motions",
211957: "Migrated",
213691: "Close Out/ Invoicing",
213774: "Judgment After Stip & Order",
}
# --- Helpers ---
def login_required(view):
@wraps(view)
def wrapped(*args, **kwargs):
uid = session.get("uid")
if not uid:
return redirect(url_for("login"))
return view(*args, **kwargs)
return wrapped
def get_user_profile(uid: str):
"""Fetch user's Firestore profile: users/{uid} => { enabled, caseEmail }"""
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}
data = snap.to_dict() or {}
return {"enabled": bool(data.get("enabled", False)), "caseEmail": data.get("caseEmail")}
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"""
print("Fetching projects....")
# Get bearer token
bearer = get_filevine_bearer()
# List projects (all pages)
projects = list_all_projects(bearer)
projects = projects[:]
# Fetch details for each
detailed_rows = []
# for p in projects:
# pid = (p.get("projectId") or {}).get("native")
# c = fetch_client(bearer, (p.get("clientId") or {}).get("native"))
# cs = fetch_contacts(bearer, pid)
# if pid is None:
# continue
# try:
# detail = fetch_project_detail(bearer, pid)
# except Exception as e:
# print(f"[WARN] detail fetch failed for {pid}: {e}")
# detail = {}
# from pprint import pprint
# defendant_one = next((c.get('orgContact', {}) for c in cs if "Defendant" in c.get('orgContact', {}).get('personTypes', [])), {})
# new_file_review = fetch_form(bearer, pid, "newFileReview") or {}
# dates_and_deadlines = fetch_form(bearer, pid, "datesAndDeadlines") or {}
# service_info = fetch_collection(bearer, pid, "serviceInfo") or []
# property_info = fetch_form(bearer, pid, "propertyInfo")
# matter_overview = fetch_form(bearer, pid, "matterOverview")
# fees_and_costs = fetch_form(bearer, pid, "feesAndCosts") or {}
# property_contacts = fetch_form(bearer, pid, "propertyContacts") or {}
# pprint(property_contacts)
# lease_info_np = fetch_form(bearer, pid, "leaseInfoNP") or {}
# completed_tasks = [{"description": x.get("body") ,
# "completed": convert_to_pacific_time(x.get("completedDate"))}
# for x in fetch_project_tasks(bearer, pid).get("items")
# if x.get("isCompleted")]
# pending_tasks = [{"description": x.get("body") ,
# "completed": convert_to_pacific_time(x.get("completedDate"))}
# for x in fetch_project_tasks(bearer, pid).get("items")
# if not x.get("isCompleted")]
# team = fetch_project_team(bearer, pid)
# assigned_attorney = next((m.get('fullname')
# for m in team
# if ('Assigned Attorney' in [r.get('name') for r in m.get('teamOrgRoles')])
# ), '')
# primary_contact = next((m.get('fullname')
# for m in team
# if ('Primary' in [r.get('name') for r in m.get('teamOrgRoles')])
# ), '')
# secondary_paralegal = next((m.get('fullname')
# for m in team
# if ('Secondary Paralegal' in [r.get('name') for r in m.get('teamOrgRoles')])
# ), '')
# # Extract notice service and expiration dates
# notice_service_date = convert_to_pacific_time(new_file_review.get("noticeServiceDate")) or ''
# notice_expiration_date = convert_to_pacific_time(new_file_review.get("noticeExpirationDate")) or ''
# # Extract daily rent damages
# daily_rent_damages = lease_info_np.get("dailyRentDamages") or dates_and_deadlines.get("dailyRentDamages") or ''
# # Extract default date
# default_date = convert_to_pacific_time(dates_and_deadlines.get("defaultDate")) or ''
# case_filed_date = convert_to_pacific_time(dates_and_deadlines.get("dateCaseFiled")) or ''
# # Extract motion hearing dates
# demurrer_hearing_date = convert_to_pacific_time(dates_and_deadlines.get("demurrerHearingDate")) or ''
# motion_to_strike_hearing_date = convert_to_pacific_time(dates_and_deadlines.get("mTSHearingDate")) or ''
# motion_to_quash_hearing_date = convert_to_pacific_time(dates_and_deadlines.get("mTQHearingDate")) or ''
# other_motion_hearing_date = convert_to_pacific_time(dates_and_deadlines.get("otherMotion1HearingDate")) or ''
# # Extract MSC details
# msc_date = convert_to_pacific_time(dates_and_deadlines.get("mSCDate")) or ''
# msc_time = dates_and_deadlines.get("mSCTime") or '' # Time field, not converting
# msc_address = dates_and_deadlines.get("mSCAddress") or ''
# msc_div_dept_room = dates_and_deadlines.get("mSCDeptDiv") or ''
# # Extract trial details
# trial_date = convert_to_pacific_time(dates_and_deadlines.get("trialDate")) or ''
# trial_time = dates_and_deadlines.get("trialTime") or '' # Time field, not converting
# trial_address = dates_and_deadlines.get("trialAddress") or ''
# trial_div_dept_room = dates_and_deadlines.get("trialDeptDivRoom") or ''
# # Extract final result of trial/MSC
# final_result = dates_and_deadlines.get("finalResultOfTrialMSCCa") or ''
# # Extract settlement details
# date_of_settlement = convert_to_pacific_time(dates_and_deadlines.get("dateOfStipulation")) or ''
# final_obligation = dates_and_deadlines.get("finalObligationUnderTheStip") or ''
# def_comply_stip = dates_and_deadlines.get("defendantsComplyWithStip") or ''
# # Extract judgment and writ details
# judgment_date = convert_to_pacific_time(dates_and_deadlines.get("dateOfJudgment")) or ''
# writ_issued_date = convert_to_pacific_time(dates_and_deadlines.get("writIssuedDate")) or ''
# # Extract lockout and stay details
# scheduled_lockout = convert_to_pacific_time(dates_and_deadlines.get("sheriffScheduledDate")) or ''
# oppose_stays = dates_and_deadlines.get("opposeStays") or ''
# # Extract premises safety and entry code
# premises_safety = new_file_review.get("lockoutSafetyIssuesOrSpecialCareIssues") or ''
# matter_gate_code = property_info.get("propertyEntryCodeOrInstructions") or ''
# # Extract possession recovered date
# date_possession_recovered = convert_to_pacific_time(dates_and_deadlines.get("datePossessionRecovered")) or ''
# # Extract attorney fees and costs
# attorney_fees = fees_and_costs.get("totalAttorneysFees") or ''
# costs = fees_and_costs.get("totalCosts") or ''
# row = {
# "client": c.get("firstName"),
# "matter_description": p.get("projectName"),
# "defendant_1": defendant_one.get('fullName', 'Unknown'),
# "matter_open": convert_to_pacific_time(dates_and_deadlines.get("dateCaseFiled") or p.get("createdDate")),
# "notice_type": new_file_review.get("noticeType", '') or '',
# "case_number": dates_and_deadlines.get('caseNumber', '') or '',
# "premises_address": property_info.get("premisesAddressWithUnit") or '',
# "premises_city": property_info.get("premisesCity") or '',
# "responsible_attorney": assigned_attorney,
# "staff_person": primary_contact,
# "staff_person_2": secondary_paralegal,
# "phase_name": p.get("phaseName"),
# "completed_tasks": completed_tasks,
# "pending_tasks": pending_tasks,
# "notice_service_date": notice_service_date,
# "notice_expiration_date": notice_expiration_date,
# "case_field_date": case_filed_date,
# "daily_rent_damages": daily_rent_damages,
# "default_date": default_date,
# "demurrer_hearing_date": demurrer_hearing_date,
# "motion_to_strike_hearing_date": motion_to_strike_hearing_date,
# "motion_to_quash_hearing_date": motion_to_quash_hearing_date,
# "other_motion_hearing_date": other_motion_hearing_date,
# "msc_date": msc_date,
# "msc_time": msc_time,
# "msc_address": msc_address,
# "msc_div_dept_room": msc_div_dept_room,
# "trial_date": trial_date,
# "trial_time": trial_time,
# "trial_address": trial_address,
# "trial_div_dept_room": trial_div_dept_room,
# "final_result": final_result,
# "date_of_settlement": date_of_settlement,
# "final_obligation": final_obligation,
# "def_comply_stip": def_comply_stip,
# "judgment_date": judgment_date,
# "writ_issued_date": writ_issued_date,
# "scheduled_lockout": scheduled_lockout,
# "oppose_stays": oppose_stays,
# "premises_safety": premises_safety,
# "matter_gate_code": matter_gate_code,
# "date_possession_recovered": date_possession_recovered,
# "attorney_fees": attorney_fees,
# "costs": costs,
# "documents_url": matter_overview.get('documentShareFolderURL') or '',
# "service_attempt_date_1": convert_to_pacific_time(next(iter(service_info), {}).get('serviceDate')),
# "contacts": cs,
# "ProjectEmailAddress": p.get("projectEmailAddress"),
# "Number": p.get("number"),
# "IncidentDate": convert_to_pacific_time(p.get("incidentDate") or detail.get("incidentDate")),
# "ProjectId": pid,
# "ProjectName": p.get("projectName") or detail.get("projectName"),
# "ProjectUrl": p.get("projectUrl") or detail.get("projectUrl"),
# }
# detailed_rows.append(row)
import worker_pool
detailed_rows = worker_pool.process_projects_parallel(projects, bearer, 5)
# Store the results in Firestore
projects_ref = db.collection("projects")
# Clear existing projects
projects_ref.stream()
for doc in projects_ref.stream():
doc.reference.delete()
# Add new projects
for row in detailed_rows:
project_id = str(row.get("ProjectId"))
if project_id:
projects_ref.document(project_id).set(row)
print(f"Stored {len(detailed_rows)} projects in Firestore")
return detailed_rows
# No longer using cache - projects are stored in Firestore
# --- Routes ---
@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"):
return redirect(url_for("dashboard"))
return redirect(url_for("welcome"))
@app.route("/login")
def login():
# Pass public Firebase config to template
fb_public = {
"apiKey": os.environ.get("FIREBASE_API_KEY", ""),
"authDomain": os.environ.get("FIREBASE_AUTH_DOMAIN", ""),
"projectId": os.environ.get("FIREBASE_PROJECT_ID", ""),
"appId": os.environ.get("FIREBASE_APP_ID", ""),
}
return render_template("login.html", firebase_config=fb_public)
@app.route("/session_login", methods=["POST"])
def session_login():
"""Exchanges Firebase ID token for a server session."""
id_token = request.json.get("idToken") if request.is_json else request.form.get("idToken")
if not id_token:
abort(400, "Missing idToken")
try:
decoded = fb_auth.verify_id_token(id_token)
uid = decoded["uid"]
session.clear()
session["uid"] = uid
# Optional: short session
session["expires_at"] = (datetime.utcnow() + timedelta(hours=8)).isoformat()
return jsonify({"ok": True})
except Exception as e:
print("[ERR] session_login:", e)
abort(401, "Invalid ID token")
@app.route("/logout")
def logout():
session.clear()
return redirect(url_for("login"))
@app.route("/welcome")
@login_required
def welcome():
uid = session.get("uid")
profile = get_user_profile(uid)
return render_template("welcome.html", profile=profile)
# --- Filevine API ---
def get_filevine_bearer():
url = "https://identity.filevine.com/connect/token"
data = {
"client_id": FV_CLIENT_ID,
"client_secret": FV_CLIENT_SECRET,
"grant_type": "personal_access_token",
"scope": "fv.api.gateway.access tenant filevine.v2.api.* email openid fv.auth.tenant.read",
"token": FV_PAT,
}
headers = {"Accept": "application/json"}
resp = requests.post(url, data=data, headers=headers, timeout=30)
resp.raise_for_status()
js = resp.json()
token = js.get("access_token")
print(f"Got bearer js", js)
return token
def list_all_projects(bearer: str):
base = "https://api.filevineapp.com/fv-app/v2/Projects"
headers = {
"Accept": "application/json",
"Authorization": f"Bearer {bearer}",
"x-fv-orgid": str(FV_ORG_ID),
"x-fv-userid": str(FV_USER_ID),
}
results = []
last_count = None
tries = 0
offset = 0
# TODO we probably need to sync the data with fierbase
cnt = 0
while len(results) < 200:
cnt = len(results)
print(f"list try {tries}, starting at {offset}, previous count {last_count}, currently at {cnt}")
tries += 1
url = base
params = {}
if last_count is not None:
# Some deployments use LastID/Offset pagination; adapt if needed
offset = offset + last_count
params["offset"] = offset
r = requests.get(url, headers=headers, params=params, timeout=30)
r.raise_for_status()
page = r.json()
from pprint import pprint
print(f"Fetched page. Headers: {r.headers}, Offset: {offset}")
items = page.get("items", [])
results.extend(items)
has_more = page.get("hasMore")
last_count = page.get("count")
if not has_more:
break
# Safety valve
if tries > 200:
break
return results
def fetch_project_detail(bearer: str, project_id_native: int):
url = f"https://api.filevineapp.com/fv-app/v2/Projects/{project_id_native}"
headers = {
"Accept": "application/json",
"Authorization": f"Bearer {bearer}",
"x-fv-orgid": str(FV_ORG_ID),
"x-fv-userid": str(FV_USER_ID),
}
r = requests.get(url, headers=headers, timeout=30)
r.raise_for_status()
return r.json()
def fetch_project_team(bearer: str, project_id_native: int):
url = f"https://api.filevineapp.com/fv-app/v2/Projects/{project_id_native}/team?limit=1000"
headers = {
"Accept": "application/json",
"Authorization": f"Bearer {bearer}",
"x-fv-orgid": str(FV_ORG_ID),
"x-fv-userid": str(FV_USER_ID),
}
r = requests.get(url, headers=headers, timeout=30)
r.raise_for_status()
from pprint import pprint
return r.json().get('items') or []
def fetch_project_tasks(bearer: str, project_id_native: int):
url = f"https://api.filevineapp.com/fv-app/v2/Projects/{project_id_native}/tasks"
headers = {
"Accept": "application/json",
"Authorization": f"Bearer {bearer}",
"x-fv-orgid": str(FV_ORG_ID),
"x-fv-userid": str(FV_USER_ID),
}
r = requests.get(url, headers=headers, timeout=30)
r.raise_for_status()
return r.json()
def fetch_client(bearer: str, client_id_native: int):
url = f"https://api.filevineapp.com/fv-app/v2/contacts/{client_id_native}"
headers = {
"Accept": "application/json",
"Authorization": f"Bearer {bearer}",
"x-fv-orgid": str(FV_ORG_ID),
"x-fv-userid": str(FV_USER_ID),
}
r = requests.get(url, headers=headers, timeout=30)
r.raise_for_status()
return r.json()
def fetch_contacts(bearer: str, project_id_native: int):
url = f"https://api.filevineapp.com/fv-app/v2/projects/{project_id_native}/contacts"
headers = {
"Accept": "application/json",
"Authorization": f"Bearer {bearer}",
"x-fv-orgid": str(FV_ORG_ID),
"x-fv-userid": str(FV_USER_ID),
}
r = requests.get(url, headers=headers, timeout=30)
r.raise_for_status()
return r.json().get("items")
def fetch_form(bearer: str, project_id_native: int, form: str):
try:
url = f"https://api.filevineapp.com/fv-app/v2/Projects/{project_id_native}/Forms/{form}"
headers = {
"Accept": "application/json",
"Authorization": f"Bearer {bearer}",
"x-fv-orgid": str(FV_ORG_ID),
"x-fv-userid": str(FV_USER_ID),
}
r = requests.get(url, headers=headers, timeout=30)
r.raise_for_status()
return r.json()
except Exception as e:
print(e)
return {}
def fetch_collection(bearer: str, project_id_native: int, collection: str):
try:
url = f"https://api.filevineapp.com/fv-app/v2/Projects/{project_id_native}/Collections/{collection}"
headers = {
"Accept": "application/json",
"Authorization": f"Bearer {bearer}",
"x-fv-orgid": str(FV_ORG_ID),
"x-fv-userid": str(FV_USER_ID),
}
r = requests.get(url, headers=headers, timeout=30)
r.raise_for_status()
return [x.get('dataObject') for x in r.json().get("items")]
except Exception as e:
print(e)
return {}
@app.route("/dashboard")
@login_required
def dashboard():
uid = session.get("uid")
profile = get_user_profile(uid)
if not profile.get("enabled"):
return redirect(url_for("welcome"))
case_email = profile.get("caseEmail")
if not case_email:
return redirect(url_for("welcome"))
# Read projects directly from Firestore
projects_ref = db.collection("projects")
docs = projects_ref.stream()
detailed_rows = []
for doc in docs:
detailed_rows.append(doc.to_dict())
print(f"Retrieved {len(detailed_rows)} projects from Firestore")
# Render table
return render_template("dashboard.html", rows=detailed_rows, case_email=case_email)
# GAE compatibility
if __name__ == "__main__":
app.run(debug=True, host="0.0.0.0", port=int(os.environ.get("PORT", "5004")))
Reference in New Issue View Git Blame Copy Permalink