142 lines
5.6 KiB
Python
142 lines
5.6 KiB
Python
import os
|
|
import requests
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
# Load environment variables
|
|
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")
|
|
|
|
class FilevineClient:
|
|
def __init__(self, bearer_token: str = None):
|
|
self.bearer_token = bearer_token
|
|
self.base_url = "https://api.filevineapp.com/fv-app/v2"
|
|
self.headers = {
|
|
"Accept": "application/json",
|
|
"Authorization": f"Bearer {self.bearer_token}",
|
|
"x-fv-orgid": str(FV_ORG_ID),
|
|
"x-fv-userid": str(FV_USER_ID),
|
|
}
|
|
|
|
def get_bearer_token(self) -> str:
|
|
"""Get a new bearer token using Filevine credentials"""
|
|
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"}
|
|
print(data)
|
|
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)
|
|
self.bearer_token = token
|
|
self.headers["Authorization"] = f"Bearer {token}"
|
|
return token
|
|
|
|
def list_all_projects(self, latest_activity_since: Optional[str] = None) -> List[Dict[str, Any]]:
|
|
"""Fetch all projects from Filevine API, optionally filtered by latest activity date.
|
|
|
|
Args:
|
|
latest_activity_since: Optional date string in mm/dd/yyyy, mm-dd-yyyy, or yyyy-mm-dd format.
|
|
Only projects with activity since this date will be returned.
|
|
"""
|
|
base = f"{self.base_url}/Projects?limit=500"
|
|
results = []
|
|
last_count = None
|
|
tries = 0
|
|
offset = 0
|
|
cnt = 0
|
|
|
|
while True:
|
|
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:
|
|
offset = offset + last_count
|
|
params["offset"] = offset
|
|
|
|
# Add latestActivitySince filter if provided
|
|
if latest_activity_since:
|
|
params["latestActivitySince"] = latest_activity_since
|
|
|
|
r = requests.get(url, headers=self.headers, params=params, timeout=30)
|
|
r.raise_for_status()
|
|
page = r.json()
|
|
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(self, project_id_native: int) -> Dict[str, Any]:
|
|
"""Fetch detailed information for a specific project"""
|
|
url = f"{self.base_url}/Projects/{project_id_native}"
|
|
r = requests.get(url, headers=self.headers, timeout=30)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def fetch_project_team(self, project_id_native: int) -> List[Dict[str, Any]]:
|
|
"""Fetch team members for a specific project"""
|
|
url = f"{self.base_url}/Projects/{project_id_native}/team?limit=1000"
|
|
r = requests.get(url, headers=self.headers, timeout=30)
|
|
r.raise_for_status()
|
|
return r.json().get('items') or []
|
|
|
|
def fetch_project_tasks(self, project_id_native: int) -> Dict[str, Any]:
|
|
"""Fetch tasks for a specific project"""
|
|
url = f"{self.base_url}/Projects/{project_id_native}/tasks"
|
|
r = requests.get(url, headers=self.headers, timeout=30)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def fetch_client(self, client_id_native: int) -> Dict[str, Any]:
|
|
"""Fetch client information by client ID"""
|
|
url = f"{self.base_url}/contacts/{client_id_native}"
|
|
r = requests.get(url, headers=self.headers, timeout=30)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def fetch_contacts(self, project_id_native: int) -> Optional[List[Dict[str, Any]]]:
|
|
"""Fetch contacts for a specific project"""
|
|
url = f"{self.base_url}/projects/{project_id_native}/contacts"
|
|
r = requests.get(url, headers=self.headers, timeout=30)
|
|
r.raise_for_status()
|
|
return r.json().get("items")
|
|
|
|
def fetch_form(self, project_id_native: int, form: str) -> Dict[str, Any]:
|
|
"""Fetch a specific form for a project"""
|
|
try:
|
|
url = f"{self.base_url}/Projects/{project_id_native}/Forms/{form}"
|
|
r = requests.get(url, headers=self.headers, timeout=30)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
except Exception as e:
|
|
print(e)
|
|
return {}
|
|
|
|
def fetch_collection(self, project_id_native: int, collection: str) -> List[Dict[str, Any]]:
|
|
"""Fetch a collection for a project"""
|
|
try:
|
|
url = f"{self.base_url}/Projects/{project_id_native}/Collections/{collection}"
|
|
r = requests.get(url, headers=self.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 {} |