saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
214 lines
8.5 KiB
Python
214 lines
8.5 KiB
Python
"""NinjaPear (nubela.co) company-enrichment provider. A fixed, trusted,
|
|
first-party integration endpoint (like Brave/Twilio) - calls httpx
|
|
directly rather than through `safe_fetch`, which exists specifically to
|
|
guard arbitrary/user-supplied collector targets, not our own known-safe
|
|
API integrations (see app/search/brave.py for the same reasoning).
|
|
|
|
Endpoint paths, parameters, and response field names below are taken from
|
|
nubela.co/llms-full.txt (a plain-text API reference, unlike the JS-rendered
|
|
docs site) and verified live against a real account. Every company-level
|
|
endpoint identifies the company by `website` only (NinjaPear has no
|
|
name-based lookup for these calls) - a company with no `official_website`
|
|
on file cannot be enriched at all, see `_require_website`. Response
|
|
parsing stays defensive (`.get()` throughout) since a live vendor API can
|
|
still change shape without notice.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
|
|
from app.core.config import Settings
|
|
from app.enrichment.base import (
|
|
CompanyDetails,
|
|
CompanyFunding,
|
|
CompetitorWithReason,
|
|
Customer,
|
|
FundingRound,
|
|
LeadershipMember,
|
|
Product,
|
|
RecentUpdate,
|
|
)
|
|
|
|
_BASE_URL = "https://nubela.co/api/v1"
|
|
_DEFAULT_TIMEOUT = 100
|
|
_FUNDING_TIMEOUT = 300 # documented by NinjaPear as long-running (up to 5 min)
|
|
|
|
|
|
def _require_website(website: str | None) -> str:
|
|
if not website:
|
|
raise ValueError(
|
|
"NinjaPear identifies a company by website only - this company has none on file"
|
|
)
|
|
return website
|
|
|
|
|
|
def _domain_from_website(website: str) -> str:
|
|
domain = website.split("//", 1)[-1].split("/", 1)[0]
|
|
return domain[4:] if domain.startswith("www.") else domain
|
|
|
|
|
|
def _split_name(person_name: str) -> tuple[str, str | None]:
|
|
parts = person_name.split(maxsplit=1)
|
|
return (parts[0], parts[1] if len(parts) > 1 else None)
|
|
|
|
|
|
class NinjaPearProvider:
|
|
provider_name = "ninjapear"
|
|
|
|
def __init__(self, settings: Settings) -> None:
|
|
self._api_key = settings.ninjapear_api_key
|
|
|
|
def _headers(self) -> dict[str, str]:
|
|
return {"Authorization": f"Bearer {self._api_key}"}
|
|
|
|
async def _get(self, url: str, params: dict[str, str], *, timeout: float) -> dict:
|
|
async with httpx.AsyncClient(timeout=timeout) as client:
|
|
response = await client.get(url, params=params, headers=self._headers())
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def get_company_details(self, name: str, website: str | None) -> CompanyDetails:
|
|
data = await self._get(
|
|
f"{_BASE_URL}/company/details",
|
|
{"website": _require_website(website)},
|
|
timeout=_DEFAULT_TIMEOUT,
|
|
)
|
|
leadership = [
|
|
LeadershipMember(name=exec_["name"], title=exec_.get("title") or exec_.get("role"))
|
|
for exec_ in data.get("executives") or []
|
|
if exec_.get("name")
|
|
]
|
|
# employee_count comes back as a raw int (e.g. 9030) and industry as
|
|
# a numeric taxonomy code, not human-readable strings - stringified
|
|
# defensively rather than left as-is, since our internal shape
|
|
# types both as `str | None`.
|
|
employee_count = data.get("employee_count")
|
|
industry = data.get("industry")
|
|
return CompanyDetails(
|
|
description=data.get("description"),
|
|
industry=str(industry) if industry is not None else None,
|
|
founded_year=data.get("founded_year"),
|
|
specialties=data.get("specialties") or [],
|
|
leadership_team=leadership,
|
|
employee_count_range=str(employee_count) if employee_count is not None else None,
|
|
)
|
|
|
|
async def get_funding(self, name: str, website: str | None) -> CompanyFunding:
|
|
data = await self._get(
|
|
f"{_BASE_URL}/company/funding",
|
|
{"website": _require_website(website)},
|
|
timeout=_FUNDING_TIMEOUT,
|
|
)
|
|
# amount/amount_usd/total_funds_raised are raw numbers, and each
|
|
# investor is an object (name/type/website/...), not a plain string
|
|
# - stringified/extracted defensively, same reasoning as
|
|
# get_company_details's employee_count/industry coercion.
|
|
rounds = []
|
|
for r in data.get("funding_rounds") or []:
|
|
amount = r.get("amount_usd") or r.get("amount")
|
|
investors = [inv.get("name") for inv in (r.get("investors") or []) if inv.get("name")]
|
|
rounds.append(
|
|
FundingRound(
|
|
round_name=r.get("round_type"),
|
|
amount=str(amount) if amount is not None else None,
|
|
date=r.get("date"),
|
|
investors=investors,
|
|
)
|
|
)
|
|
total_raised = data.get("total_funds_raised_usd") or data.get("total_funds_raised")
|
|
return CompanyFunding(
|
|
total_raised=str(total_raised) if total_raised is not None else None, rounds=rounds
|
|
)
|
|
|
|
async def get_updates(self, name: str, website: str | None) -> list[RecentUpdate]:
|
|
data = await self._get(
|
|
f"{_BASE_URL}/company/updates",
|
|
{"website": _require_website(website)},
|
|
timeout=_DEFAULT_TIMEOUT,
|
|
)
|
|
return [
|
|
RecentUpdate(
|
|
type=u.get("source", "update"),
|
|
text=u.get("title") or u.get("description") or "",
|
|
url=u.get("url"),
|
|
date=u.get("timestamp"),
|
|
)
|
|
for u in data.get("updates") or []
|
|
if u.get("title") or u.get("description")
|
|
]
|
|
|
|
async def get_competitors(self, name: str, website: str | None) -> list[CompetitorWithReason]:
|
|
data = await self._get(
|
|
f"{_BASE_URL}/competitor/listing",
|
|
{"website": _require_website(website)},
|
|
timeout=_DEFAULT_TIMEOUT,
|
|
)
|
|
return [
|
|
CompetitorWithReason(
|
|
name=c.get("name") or c.get("website", ""), reason=c.get("competition_reason")
|
|
)
|
|
for c in data.get("competitors") or []
|
|
if c.get("website") or c.get("name")
|
|
]
|
|
|
|
async def get_products(self, name: str, website: str | None) -> list[Product]:
|
|
data = await self._get(
|
|
f"{_BASE_URL}/product/listing",
|
|
{"website": _require_website(website)},
|
|
timeout=_DEFAULT_TIMEOUT,
|
|
)
|
|
products = []
|
|
for p in data.get("products") or []:
|
|
if not p.get("name"):
|
|
continue
|
|
categories = p.get("categories") or []
|
|
products.append(
|
|
Product(
|
|
name=p["name"],
|
|
description=p.get("description"),
|
|
category=", ".join(categories) if categories else None,
|
|
)
|
|
)
|
|
return products
|
|
|
|
async def get_customers(self, name: str, website: str | None) -> list[Customer]:
|
|
data = await self._get(
|
|
f"{_BASE_URL}/customer/listing",
|
|
{"website": _require_website(website)},
|
|
timeout=_DEFAULT_TIMEOUT,
|
|
)
|
|
# NinjaPear returns three separately-categorized arrays rather than
|
|
# one flat list - merged here with a relationship tag per group.
|
|
customers = []
|
|
for relationship, key in (
|
|
("customer", "customers"),
|
|
("investor", "investors"),
|
|
("partner", "partner_platforms"),
|
|
):
|
|
for entry in data.get(key) or []:
|
|
if entry.get("name"):
|
|
customers.append(Customer(name=entry["name"], relationship=relationship))
|
|
return customers
|
|
|
|
async def get_work_email(self, person_name: str, company_website: str) -> str | None:
|
|
first_name, last_name = _split_name(person_name)
|
|
params = {"first_name": first_name, "domain": _domain_from_website(company_website)}
|
|
if last_name:
|
|
params["last_name"] = last_name
|
|
data = await self._get(f"{_BASE_URL}/employee/work-email", params, timeout=_DEFAULT_TIMEOUT)
|
|
return data.get("work_email")
|
|
|
|
async def get_person_profile(
|
|
self, person_name: str, company_website: str
|
|
) -> tuple[str | None, str | None]:
|
|
first_name, _ = _split_name(person_name)
|
|
# v2 endpoint per NinjaPear's docs - the only one of the endpoints
|
|
# used here that isn't under /api/v1.
|
|
data = await self._get(
|
|
"https://nubela.co/api/v2/employee/profile",
|
|
{"first_name": first_name, "employer_website": company_website},
|
|
timeout=_DEFAULT_TIMEOUT,
|
|
)
|
|
return data.get("x_profile_url"), data.get("bio")
|