spb/company-atlas
Public
Python 66.3%
TypeScript 22.7%
JavaScript 8.6%
HTML 1.4%
CSS 0.7%
1"""Companies: list, compare (literal path registered before the `/{slug_or_id}` catch-all), detail and every sub-resource."""2from __future__ import annotations34from collections import defaultdict5from typing import Any67from fastapi import APIRouter, HTTPException, Query, Response89from companyatlas.api import queries as q10from companyatlas.api import serializers as ser11from companyatlas.api.common import PageDep, page_payload, public_cache_value12from companyatlas.db import connection, fetch_all, fetch_one13from companyatlas.services.enrichment import profile_facts14from companyatlas.taxonomy import Metric1516ORDER = 4017router = APIRouter(prefix="/api/v1", tags=["companies"])1819TIMELINE_FILTERS: dict[str, list[str]] = {20 "all": [], "products": ["PRODUCT"], "jobs": ["HIRING"], "pricing": ["PRICING"], "leadership": ["LEADERSHIP"], "locations": ["LOCATION"],21 "legal": ["LEGAL"], "news": ["COMMUNICATION", "INVESTOR_RELATIONS", "MARKETING"], "developer": ["DEVELOPER", "TECHNOLOGY"],22 "corporate": ["FINANCING", "M&A", "PARTNERSHIP", "STRATEGY"],23}24COMPARE_METRICS = [Metric.ACTIVITY_SCORE.value, Metric.HIRING_MOMENTUM_30D.value, Metric.PRODUCT_VELOCITY.value, Metric.AI_ADOPTION.value,25 Metric.GEO_EXPANSION.value, Metric.DEVELOPER_MOMENTUM.value, Metric.CORPORATE_CHANGE_INDEX.value, Metric.OPEN_JOBS.value,26 Metric.ANOMALY_SCORE.value]272829def _pub(response: Response, s: int = 60) -> None:30 response.headers["cache-control"] = public_cache_value(s)313233# ------------------------------------------------------------------------------------------------ list / compare343536@router.get("/companies", summary="Company directory")37async def list_companies(response: Response, p: PageDep, q_: str | None = Query(None, alias="q", max_length=200),38 country: str | None = None, industry: str | None = None, tier: int | None = Query(None, ge=1, le=4),39 public: str | None = None, status: str | None = None, has_events: str | None = None,40 sort: str = Query("activity", pattern="^(activity|events|hiring|name|importance|recent|relevance)$"),41 sparkline: str | None = None) -> dict[str, Any]:42 _pub(response, 60)43 where, params = q.company_filters(q=q_, country=country, industry=industry, tier=tier, public=q.parse_bool(public), status=status,44 has_events=q.parse_bool(has_events))45 async with connection() as conn:46 ids, total = await q.company_page_ids(conn, where, params, sort=sort, limit=p.per_page, offset=p.offset)47 rows = await q.fetch_cards_by_ids(conn, ids, sparkline=bool(q.parse_bool(sparkline)))48 return page_payload([ser.company_card(r) for r in rows], total, p)495051@router.get("/companies/compare", summary="Side-by-side comparison of 2–6 companies")52async def compare(response: Response, companies: str = Query(..., description="comma-separated slugs or ids"), days: int = Query(90, ge=7, le=365)) -> dict[str, Any]:53 _pub(response, 120)54 keys = list(dict.fromkeys(q.csv_list(companies, maxlen=6)))55 if len(keys) < 2:56 raise HTTPException(status_code=422, detail="companies: provide 2 to 6 comma-separated slugs")57 async with connection() as conn:58 found = []59 for k in keys:60 c = await q.get_company(conn, k)61 if c is None:62 raise HTTPException(status_code=404, detail=f"company not found: {k}")63 found.append(c)64 ids = [c["id"] for c in found]65 slug_of = {c["id"]: c["slug"] for c in found}66 cards = [ser.company_card(r) for r in await q.fetch_cards_by_ids(conn, ids, sparkline=True)]67 series_rows = await fetch_all(conn, "select company_id, day, value, confidence from metric_series where metric = 'activity_score' and "68 "company_id = any(cast(:ids as text[])) and day >= :d order by day", ids=ids, d=q.days_ago(days).date())69 ev_rows = await fetch_all(conn, "select company_id, event_type, count(*) as n from events where status = 'active' and detected_at >= :d and "70 "company_id = any(cast(:ids as text[])) group by 1, 2", ids=ids, d=q.days_ago(30))71 job_rows = await fetch_all(conn, "select company_id, count(*) filter (where status = 'open') as open, "72 "count(*) filter (where status = 'open' and is_ai) as ai_open, "73 "count(*) filter (where first_seen_at >= :d and not baseline) as new_30d from jobs where company_id = any(cast(:ids as text[])) "74 "group by company_id", ids=ids, d=q.days_ago(30))75 loc_rows = await fetch_all(conn, "select company_id, count(*) as n from locations where status = 'listed' and company_id = any(cast(:ids as text[])) "76 "group by company_id", ids=ids)77 metrics: dict[str, dict[str, Any]] = {m: {} for m in COMPARE_METRICS}78 for card in cards:79 for m, v in card["metrics"].items():80 metrics.setdefault(m, {})[card["slug"]] = v81 series: dict[str, list[dict[str, Any]]] = {s: [] for s in slug_of.values()}82 for r in series_rows:83 series[slug_of[r["company_id"]]].append(ser.metric_point(r))84 events_30d: dict[str, dict[str, int]] = {s: {} for s in slug_of.values()}85 for r in ev_rows:86 events_30d[slug_of[r["company_id"]]][r["event_type"]] = int(r["n"])87 jobs = {slug_of[r["company_id"]]: {"open": int(r["open"]), "ai_open": int(r["ai_open"]), "new_30d": int(r["new_30d"])} for r in job_rows}88 for s in slug_of.values():89 jobs.setdefault(s, {"open": 0, "ai_open": 0, "new_30d": 0})90 locations = {slug_of[r["company_id"]]: int(r["n"]) for r in loc_rows}91 for s in slug_of.values():92 locations.setdefault(s, 0)93 return {"companies": cards, "metrics": metrics, "series": series, "events_30d": events_30d, "jobs": jobs, "locations": locations}949596# ------------------------------------------------------------------------------------------------ detail979899@router.get("/companies/{key}", summary="Company profile")100async def company_detail(key: str, response: Response) -> dict[str, Any]:101 _pub(response, 60)102 async with connection() as conn:103 c = await q.require_company(conn, key)104 cid = c["id"]105 rows = await q.fetch_cards_by_ids(conn, [cid], sparkline=True)106 out = ser.company_card(rows[0])107 out["company_type"] = c.get("company_type")108 out["employees"] = c.get("employees")109 out["wikidata_id"] = c.get("wikidata_id")110 out["indexed"] = bool(c.get("indexed"))111 out["discovered_at"] = c.get("discovered_at")112 out["first_observed_at"] = c.get("first_observed_at")113 out["last_change_at"] = c.get("last_change_at")114 out["aliases"] = [r["alias"] for r in await fetch_all(conn, "select alias from company_aliases where company_id = :id order by kind, alias limit 50", id=cid)]115 out["domains"] = await fetch_all(conn, "select domain, kind, status, first_seen_at, last_seen_at from domains where company_id = :id order by kind, domain limit 100", id=cid)116 rel = await fetch_all(conn, "select r.kind, r.to_name, r.valid_from, r.valid_to, r.confidence, r.source_url, r.provenance, r.first_seen_at, "117 "r.last_seen_at, o.slug, o.display_name from company_relationships r left join companies o on o.id = r.to_company_id "118 "where r.from_company_id = :id order by (r.valid_to is not null), r.kind, o.importance desc nulls last, r.last_seen_at desc limit 200", id=cid)119 out["relationships"] = [ser.relationship(r) for r in rel]120 out["profile"] = ser._dict(c.get("source_meta")).get("profile") or None121 out["facts"] = profile_facts(out["profile"])122 md = await fetch_all(conn, "select metric, value, confidence, computed_at, formula_version, inputs from metrics_current where company_id = :id order by metric", id=cid)123 out["metrics_detail"] = [{"metric": r["metric"], "value": ser.metric_value(r["metric"], r["value"]), "confidence": ser._float(r["confidence"], 3),124 "computed_at": r["computed_at"], "formula_version": r["formula_version"], "inputs": ser._dict(r["inputs"])} for r in md]125 sbs = await fetch_all(conn, "select surface, count(*) as n, count(*) filter (where status = 'active') as active from sensors where company_id = :id "126 "and status <> 'retired' group by surface order by n desc", id=cid)127 out["sensors_by_surface"] = {r["surface"]: int(r["n"]) for r in sbs}128 total_sensors = sum(int(r["n"]) for r in sbs)129 active_sensors = sum(int(r["active"]) for r in sbs)130 days_observed = await fetch_one(conn, "select count(*) as n from company_daily where company_id = :id and observations > 0", id=cid)131 hist = next((m["value"] for m in out["metrics_detail"] if m["metric"] == Metric.HISTORICAL_COVERAGE.value), None)132 out["coverage"] = {"historical_coverage": hist, "first_observed_at": c.get("first_observed_at"),133 "days_observed": int(days_observed["n"]) if days_observed else 0,134 "sensor_uptime": round(active_sensors / total_sensors, 3) if total_sensors else None}135 sig = await fetch_all(conn, "select * from signals where company_id = :id and status = 'active' order by detected_at desc limit 10", id=cid)136 out["signals"] = [ser.signal(r) for r in sig]137 act = await q.metric_series(conn, cid, Metric.ACTIVITY_SCORE.value, 30)138 hir = await q.metric_series(conn, cid, Metric.OPEN_JOBS.value, 90)139 out["sparklines"] = {"activity_30d": [ser._float(r["value"], 1) for r in act], "hiring_90d": [ser._float(r["value"], 1) for r in hir]}140 out["recent_events"] = [ser.event(r) for r in await q.fetch_events(conn, *q.event_filters(company_id=cid), limit=10)]141 return out142143144# ------------------------------------------------------------------------------------------------ sub-resources145146147@router.get("/companies/{key}/events", summary="Company events")148async def company_events(key: str, response: Response, p: PageDep, event_type: str | None = None, event_subtype: str | None = None,149 since: str | None = None, until: str | None = None, min_importance: float | None = Query(None, ge=0, le=1),150 surface: str | None = None, sort: str = Query("recent", pattern="^(recent|importance)$")) -> dict[str, Any]:151 _pub(response, 30)152 async with connection() as conn:153 c = await q.require_company(conn, key)154 where, params = q.event_filters(company_id=c["id"], event_type=event_type, event_subtype=event_subtype, since=q.parse_iso(since),155 until=q.parse_iso(until, "until"), min_importance=min_importance, surface=surface)156 rows = await q.fetch_events(conn, where, params, sort=sort, limit=p.per_page, offset=p.offset)157 total = await q.count_events(conn, where, params)158 return page_payload([ser.event(r) for r in rows], total, p)159160161@router.get("/companies/{key}/timeline", summary="Timeline grouped by day")162async def company_timeline(key: str, response: Response, filter: str = Query("all", pattern="^(all|products|jobs|pricing|leadership|locations|legal|news|developer|corporate)$"),163 limit: int = Query(200, ge=1, le=500), before: str | None = None) -> dict[str, Any]:164 _pub(response, 60)165 types = TIMELINE_FILTERS[filter]166 async with connection() as conn:167 c = await q.require_company(conn, key)168 where, params = q.event_filters(company_id=c["id"], event_types=types or None, until=q.parse_iso(before, "before"))169 rows = await q.fetch_events(conn, where, params, sort="recent", limit=limit)170 where_sql = " where " + " and ".join(where)171 days = await fetch_all(conn, f"select date(e.detected_at at time zone 'UTC') as day, count(*) as count from events e join companies c on c.id = e.company_id"172 f"{where_sql} group by 1 order by 1 desc limit 730", **params)173 items = []174 for r in rows:175 ev = ser.event(r)176 ev["day"] = r["detected_at"].date()177 items.append(ev)178 return {"items": items, "days": [{"day": d["day"], "count": int(d["count"])} for d in days], "filter": filter}179180181@router.get("/companies/{key}/metrics", summary="Current metrics and time series")182async def company_metrics(key: str, response: Response, metric: str | None = None, days: int = Query(90, ge=1, le=730)) -> dict[str, Any]:183 _pub(response, 300)184 async with connection() as conn:185 c = await q.require_company(conn, key)186 cur = await fetch_all(conn, "select metric, value, confidence, computed_at, formula_version, inputs from metrics_current where company_id = :id order by metric", id=c["id"])187 params: dict[str, Any] = {"id": c["id"], "d": q.days_ago(days).date()}188 extra = ""189 if metric:190 extra = " and metric = cast(:m as text)"191 params["m"] = metric[:60]192 rows = await fetch_all(conn, f"select metric, day, value, confidence from metric_series where company_id = :id and day >= :d{extra} "193 "order by metric, day limit 20000", **params)194 series: dict[str, list[dict[str, Any]]] = defaultdict(list)195 for r in rows:196 series[r["metric"]].append(ser.metric_point(r))197 return {"current": [{"metric": r["metric"], "value": ser.metric_value(r["metric"], r["value"]), "confidence": ser._float(r["confidence"], 3),198 "computed_at": r["computed_at"], "formula_version": r["formula_version"], "inputs": ser._dict(r["inputs"])} for r in cur],199 "series": dict(series), "days": days}200201202@router.get("/companies/{key}/jobs", summary="Publicly listed jobs with summary")203async def company_jobs(key: str, response: Response, p: PageDep, status: str = Query("open", pattern="^(open|removed|all)$"),204 q_: str | None = Query(None, alias="q", max_length=120), country: str | None = None, ai: str | None = None,205 department: str | None = None, remote: str | None = None,206 sort: str = Query("recent", pattern="^(recent|title|posted)$")) -> dict[str, Any]:207 _pub(response, 120)208 order = {"recent": "j.first_seen_at desc, j.id", "title": "j.title asc, j.id", "posted": "j.posted_at desc nulls last, j.id"}[sort]209 where = ["j.company_id = :id"]210 async with connection() as conn:211 c = await q.require_company(conn, key)212 params: dict[str, Any] = {"id": c["id"]}213 if status == "open":214 where.append("j.status = 'open'")215 elif status == "removed":216 where.append("j.status <> 'open'")217 if q_:218 where.append("j.title ilike :jq")219 params["jq"] = f"%{q_.strip()}%"220 if country:221 where.append("j.country = cast(:jc as char(2))")222 params["jc"] = country.upper()[:2]223 if q.parse_bool(ai):224 where.append("j.is_ai")225 if department:226 where.append("j.department = cast(:jd as text)")227 params["jd"] = department[:80]228 if q.parse_bool(remote) is not None:229 where.append("j.remote = :jr")230 params["jr"] = q.parse_bool(remote)231 wsql = " and ".join(where)232 rows = await fetch_all(conn, f"select j.* from jobs j where {wsql} order by {order} limit :limit offset :offset", **params, limit=p.per_page, offset=p.offset)233 total = await q.bounded_count(conn, f"from jobs j where {wsql}", params)234 d7 = q.days_ago(7)235 s = await fetch_one(conn, "select count(*) filter (where status = 'open') as open, count(*) filter (where first_seen_at >= :d7 and status = 'open' and not baseline) as new_7d, "236 "count(*) filter (where removed_at >= :d7) as removed_7d, count(*) filter (where status = 'open' and is_ai) as ai_open, "237 "count(*) filter (where status = 'open' and remote is true) as remote_open, "238 "count(*) filter (where status = 'open' and remote is not null) as remote_known from jobs where company_id = :id",239 id=c["id"], d7=d7) or {}240 by_country = await fetch_all(conn, "select country, count(*) as n from jobs where company_id = :id and status = 'open' and country is not null "241 "group by country order by n desc limit 15", id=c["id"])242 by_dept = await fetch_all(conn, "select department, count(*) as n from jobs where company_id = :id and status = 'open' and department is not null "243 "group by department order by n desc limit 15", id=c["id"])244 remote_known = int(s.get("remote_known") or 0)245 summary = {"open": int(s.get("open") or 0), "new_7d": int(s.get("new_7d") or 0), "removed_7d": int(s.get("removed_7d") or 0),246 "ai_open": int(s.get("ai_open") or 0), "by_country": [{"country": r["country"], "n": int(r["n"])} for r in by_country],247 "by_department": [{"department": r["department"], "n": int(r["n"])} for r in by_dept],248 "remote_ratio": round(int(s.get("remote_open") or 0) / remote_known, 3) if remote_known else None}249 payload = page_payload([ser.job(r) for r in rows], total, p)250 payload["meta"] = {"summary": summary}251 return payload252253254@router.get("/companies/{key}/people", summary="Leadership listed on monitored pages and on Wikidata")255async def company_people(key: str, response: Response) -> dict[str, Any]:256 _pub(response, 300)257 async with connection() as conn:258 c = await q.require_company(conn, key)259 rows = await fetch_all(conn, "select * from people where company_id = :id order by is_executive desc, status, last_seen_at desc limit 500", id=c["id"])260 people = [ser.person(r) for r in rows]261 return {"listed": [p for p in people if p["status"] == "listed"], "no_longer_listed": [p for p in people if p["status"] != "listed"],262 "sources": sorted({p["source"] for p in people})}263264265@router.get("/companies/{key}/products", summary="Products in the public catalog")266async def company_products(key: str, response: Response) -> dict[str, Any]:267 _pub(response, 300)268 async with connection() as conn:269 c = await q.require_company(conn, key)270 rows = await fetch_all(conn, "select * from products where company_id = :id order by status, last_seen_at desc limit 500", id=c["id"])271 return {"listed": [ser.product(r) for r in rows if r["status"] == "listed"], "removed": [ser.product(r) for r in rows if r["status"] != "listed"]}272273274@router.get("/companies/{key}/pricing", summary="Current plans and every previous version")275async def company_pricing(key: str, response: Response, history_limit: int = Query(200, ge=1, le=1000)) -> dict[str, Any]:276 _pub(response, 300)277 async with connection() as conn:278 c = await q.require_company(conn, key)279 cur = await fetch_all(conn, "select * from pricing_plans where company_id = :id and status = 'current' order by price nulls last, plan_name limit 100", id=c["id"])280 hist = await fetch_all(conn, "select * from pricing_plans where company_id = :id and status <> 'current' order by valid_from desc limit :lim",281 id=c["id"], lim=history_limit)282 return {"current": [ser.plan(r) for r in cur], "history": [ser.plan(r) for r in hist]}283284285@router.get("/companies/{key}/locations", summary="Locations listed on monitored pages")286async def company_locations(key: str, response: Response, status: str = Query("listed", pattern="^(listed|all)$")) -> dict[str, Any]:287 _pub(response, 300)288 async with connection() as conn:289 c = await q.require_company(conn, key)290 extra = " and status = 'listed'" if status == "listed" else ""291 rows = await fetch_all(conn, f"select * from locations where company_id = :id{extra} order by kind, country, city limit 1000", id=c["id"])292 items = [ser.location(r) for r in rows]293 return {"items": items, "countries": sorted({r["country"] for r in rows if r["country"] and r["status"] == "listed"})}294295296@router.get("/companies/{key}/news", summary="First-party news, blog and changelog items")297async def company_news(key: str, response: Response, limit: int = Query(50, ge=1, le=500), category: str | None = None) -> dict[str, Any]:298 _pub(response, 120)299 async with connection() as conn:300 c = await q.require_company(conn, key)301 params: dict[str, Any] = {"id": c["id"], "lim": limit}302 extra = ""303 if category:304 extra = " and category = cast(:cat as text)"305 params["cat"] = category[:40]306 rows = await fetch_all(conn, f"select * from news_items where company_id = :id{extra} order by coalesce(published_at, first_seen_at) desc limit :lim", **params)307 return {"items": [ser.news_item(r) for r in rows]}308309310@router.get("/companies/{key}/sensors", summary="Sensors attached to the company")311async def company_sensors(key: str, response: Response, include_retired: str | None = None) -> dict[str, Any]:312 _pub(response, 120)313 async with connection() as conn:314 c = await q.require_company(conn, key)315 extra = "" if q.parse_bool(include_retired) else " and status <> 'retired'"316 rows = await fetch_all(conn, f"select * from sensors where company_id = :id{extra} order by quality_score desc, surface limit 500", id=c["id"])317 return {"items": [ser.sensor(r) for r in rows]}318319320@router.get("/companies/{key}/history", summary="Historical page viewer index (≤ 20 versions per sensor)")321async def company_history(key: str, response: Response, versions: int = Query(20, ge=1, le=20)) -> dict[str, Any]:322 _pub(response, 300)323 async with connection() as conn:324 c = await q.require_company(conn, key)325 sensors = await fetch_all(conn, "select * from sensors where company_id = :id and status <> 'retired' and snapshot_count > 0 "326 "order by quality_score desc, surface limit 200", id=c["id"])327 if not sensors:328 sensors = await fetch_all(conn, "select * from sensors where company_id = :id and status <> 'retired' order by quality_score desc, surface limit 200", id=c["id"])329 snaps = await fetch_all(conn, "select s.* from sensors sn cross join lateral (select * from snapshots x where x.sensor_id = sn.id "330 "order by x.fetched_at desc limit :v) s where sn.company_id = :id and sn.status <> 'retired' order by s.sensor_id, s.fetched_at desc",331 id=c["id"], v=versions)332 by_sensor: dict[str, list[dict[str, Any]]] = defaultdict(list)333 for s in snaps:334 by_sensor[s["sensor_id"]].append(ser.snapshot(s))335 out = []336 for s in sensors:337 item = ser.sensor(s)338 item["versions"] = by_sensor.get(s["id"], [])339 out.append(item)340 return {"sensors": out}341342343@router.get("/companies/{key}/similar", summary="Similar companies (industry, country, importance)")344async def company_similar(key: str, response: Response, limit: int = Query(8, ge=1, le=50)) -> dict[str, Any]:345 _pub(response, 600)346 async with connection() as conn:347 c = await q.require_company(conn, key)348 rows = await fetch_all(conn, "select c.id from companies c where c.id <> :id and c.status = 'ACTIVE' and "349 "(c.industry_primary = cast(:ip as text) or c.industries && cast(:inds as text[]) or c.country = cast(:country as char(2))) "350 "order by (c.industry_primary is not distinct from cast(:ip as text) and :ip is not null) desc, "351 "(c.industries && cast(:inds as text[])) desc, (c.country is not distinct from cast(:country as char(2))) desc, "352 "abs(c.importance - :imp) asc, c.importance desc limit :lim",353 id=c["id"], ip=c.get("industry_primary"), inds=list(c.get("industries") or []), country=c.get("country"),354 imp=float(c.get("importance") or 0), lim=limit)355 cards = await q.fetch_cards_by_ids(conn, [r["id"] for r in rows])356 return {"items": [ser.company_card(r) for r in cards]}357