feat: Integrate Folk CRM API for investor synchronization and compatibility scoring
This commit is contained in:
bolade
@@ -0,0 +1,260 @@
|
||||
import os
|
||||
import sys
|
||||
|
||||
import requests
|
||||
|
||||
|
||||
class FolkAPI:
|
||||
BASE_URL = "https://api.folk.app/v1"
|
||||
|
||||
def __init__(self, api_key: str):
|
||||
self.headers = {"Authorization": f"Bearer {api_key}"}
|
||||
|
||||
def get_groups(self):
|
||||
"""Fetch all groups from Folk."""
|
||||
url = f"{self.BASE_URL}/groups"
|
||||
response = requests.get(url, headers=self.headers)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def create_company(
|
||||
self,
|
||||
name: str,
|
||||
group_id: str = None,
|
||||
website: str = None,
|
||||
linkedin_url: str = None,
|
||||
description: str = None,
|
||||
emails=None,
|
||||
phones=None,
|
||||
addresses=None,
|
||||
urls=None,
|
||||
custom_field_values=None,
|
||||
groups=None,
|
||||
**kwargs,
|
||||
):
|
||||
"""Create a company (investor) in a specific group.
|
||||
|
||||
This method builds a payload matching Folk's Create Company API:
|
||||
https://developer.folk.app/api-reference/companies/create-a-company
|
||||
|
||||
It keeps backward compatibility with the previous `group_id`,
|
||||
`website` and `linkedin_url` arguments.
|
||||
"""
|
||||
url = f"{self.BASE_URL}/companies"
|
||||
|
||||
# Build the top-level payload expected by Folk
|
||||
data = {"name": name}
|
||||
if description:
|
||||
data["description"] = description
|
||||
|
||||
# Groups: prefer explicit `groups`, else fall back to `group_id`
|
||||
if groups:
|
||||
# Accept either list of ids or list of dicts
|
||||
formatted = []
|
||||
for g in groups:
|
||||
if isinstance(g, dict) and g.get("id"):
|
||||
formatted.append({"id": g["id"]})
|
||||
else:
|
||||
formatted.append({"id": str(g)})
|
||||
data["groups"] = formatted
|
||||
elif group_id:
|
||||
data["groups"] = [{"id": group_id}]
|
||||
|
||||
# Helper to normalize single or multiple inputs into lists
|
||||
def _to_list(val):
|
||||
if val is None:
|
||||
return None
|
||||
if isinstance(val, (list, tuple)):
|
||||
return [v for v in val if v is not None]
|
||||
return [val]
|
||||
|
||||
# URLs: include website and linkedin_url if provided and merge with urls
|
||||
urls_list = _to_list(urls) or []
|
||||
if website:
|
||||
urls_list.append(website)
|
||||
if linkedin_url:
|
||||
urls_list.append(linkedin_url)
|
||||
if urls_list:
|
||||
data["urls"] = urls_list
|
||||
|
||||
# Emails/phones/addresses
|
||||
emails_list = _to_list(emails)
|
||||
if emails_list:
|
||||
data["emails"] = emails_list
|
||||
phones_list = _to_list(phones)
|
||||
if phones_list:
|
||||
data["phones"] = phones_list
|
||||
addresses_list = _to_list(addresses)
|
||||
if addresses_list:
|
||||
data["addresses"] = addresses_list
|
||||
|
||||
# Custom field values follow the API's structure
|
||||
if custom_field_values:
|
||||
data["customFieldValues"] = custom_field_values
|
||||
|
||||
# Allow passing any additional top-level fields via kwargs (careful)
|
||||
for k, v in kwargs.items():
|
||||
# don't overwrite keys we explicitly set
|
||||
if k not in data:
|
||||
data[k] = v
|
||||
|
||||
response = requests.post(url, headers=self.headers, json=data)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def create_person(
|
||||
self,
|
||||
first_name: str,
|
||||
last_name: str,
|
||||
email: str = None,
|
||||
company_id: str = None,
|
||||
group_id: str = None,
|
||||
companies=None,
|
||||
emails=None,
|
||||
phones=None,
|
||||
addresses=None,
|
||||
urls=None,
|
||||
custom_field_values=None,
|
||||
groups=None,
|
||||
**kwargs,
|
||||
):
|
||||
"""Create a person in the workspace.
|
||||
|
||||
Builds payload matching Folk's Create Person API: use camelCase
|
||||
keys (firstName, lastName, groups, companies, emails, etc.).
|
||||
Keeps backward compatibility with `company_id` and `group_id`.
|
||||
"""
|
||||
url = f"{self.BASE_URL}/people"
|
||||
|
||||
data = {"firstName": first_name, "lastName": last_name}
|
||||
|
||||
# Groups: explicit `groups` preferred, else fallback to `group_id`
|
||||
if groups:
|
||||
formatted = []
|
||||
for g in groups:
|
||||
if isinstance(g, dict) and g.get("id"):
|
||||
formatted.append({"id": g["id"]})
|
||||
else:
|
||||
formatted.append({"id": str(g)})
|
||||
data["groups"] = formatted
|
||||
elif group_id:
|
||||
data["groups"] = [{"id": group_id}]
|
||||
|
||||
# Companies: keep backward compatibility with company_id
|
||||
if companies:
|
||||
formatted = []
|
||||
for c in companies:
|
||||
if isinstance(c, dict):
|
||||
formatted.append(c)
|
||||
elif isinstance(c, str):
|
||||
# treat as id
|
||||
formatted.append({"id": c})
|
||||
if formatted:
|
||||
data["companies"] = formatted
|
||||
elif company_id:
|
||||
data["companies"] = [{"id": company_id}]
|
||||
|
||||
# Helper to normalize into lists
|
||||
def _to_list(val):
|
||||
if val is None:
|
||||
return None
|
||||
if isinstance(val, (list, tuple)):
|
||||
return [v for v in val if v is not None]
|
||||
return [val]
|
||||
|
||||
emails_list = _to_list(emails) or []
|
||||
if email:
|
||||
emails_list.insert(0, email)
|
||||
if emails_list:
|
||||
data["emails"] = emails_list
|
||||
|
||||
phones_list = _to_list(phones)
|
||||
if phones_list:
|
||||
data["phones"] = phones_list
|
||||
addresses_list = _to_list(addresses)
|
||||
if addresses_list:
|
||||
data["addresses"] = addresses_list
|
||||
urls_list = _to_list(urls)
|
||||
if urls_list:
|
||||
data["urls"] = urls_list
|
||||
|
||||
if custom_field_values:
|
||||
data["customFieldValues"] = custom_field_values
|
||||
|
||||
# Allow passthrough of other top-level fields in kwargs
|
||||
for k, v in kwargs.items():
|
||||
if k not in data:
|
||||
data[k] = v
|
||||
|
||||
response = requests.post(url, headers=self.headers, json=data)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
|
||||
# Prefer getting the API key from the environment. If not set, fall back to the
|
||||
# existing (hard-coded) key so behavior is unchanged for now.
|
||||
DEFAULT_API_KEY = "FOLKfIGXuv74ML9EAajxyiUR39ePaNrZ"
|
||||
api_key = os.environ.get("FOLK_API_KEY", DEFAULT_API_KEY)
|
||||
|
||||
folk = FolkAPI(api_key=api_key)
|
||||
|
||||
|
||||
def example_flow():
|
||||
# Step 1: Get groups
|
||||
groups = folk.get_groups()
|
||||
print(groups)
|
||||
|
||||
# Safely dig into the returned structure. The API returns groups under
|
||||
# groups['data']['items'] (not groups['data'][0]). Handle missing/empty.
|
||||
items = groups.get("data", {}).get("items", [])
|
||||
if not items:
|
||||
print("No groups returned by Folk API.")
|
||||
sys.exit(1)
|
||||
|
||||
# Choose the first group as an example
|
||||
group_id = items[0].get("id")
|
||||
if not group_id:
|
||||
print("No id found for the first group item.")
|
||||
sys.exit(1)
|
||||
|
||||
# Step 2: Choose a group_id and create a company
|
||||
company = folk.create_company(
|
||||
name="2050 Investment Partners",
|
||||
group_id=group_id,
|
||||
website="https://2050.com",
|
||||
linkedin_url="https://linkedin.com/company/2050-investments",
|
||||
)
|
||||
|
||||
# Step 3: Add a person to the same group or company
|
||||
person = folk.create_person(
|
||||
first_name="John",
|
||||
last_name="Doe",
|
||||
email="john@2050.com",
|
||||
company_id=company.get("data", {}).get("id"),
|
||||
group_id=group_id,
|
||||
)
|
||||
|
||||
print("Created company:", company)
|
||||
print("Created person:", person)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
example_flow()
|
||||
except requests.HTTPError as e:
|
||||
# Try to include response body for easier debugging if available
|
||||
resp = getattr(e, "response", None)
|
||||
if resp is not None:
|
||||
try:
|
||||
body = resp.text
|
||||
except Exception:
|
||||
body = "<unreadable response body>"
|
||||
print("HTTP error while talking to Folk API:", e)
|
||||
print("Response status:", resp.status_code)
|
||||
print("Response body:", body)
|
||||
else:
|
||||
print("HTTP error while talking to Folk API:", e)
|
||||
sys.exit(1)
|
||||
except Exception as e: # pragma: no cover - top-level safety
|
||||
print("Unexpected error:", e)
|
||||
sys.exit(1)
|
||||
|
||||
Reference in New Issue
Block a user