"""Companies: list, compare (literal path registered before the `/{slug_or_id}` catch-all), detail and every sub-resource.""" from __future__ import annotations from collections import defaultdict from typing import Any from fastapi import APIRouter, HTTPException, Query, Response from companyatlas.api import queries as q from companyatlas.api import serializers as ser from companyatlas.api.common import PageDep, page_payload, public_cache_value from companyatlas.db import connection, fetch_all, fetch_one from companyatlas.services.enrichment import profile_facts from companyatlas.taxonomy import Metric ORDER = 40 router = APIRouter(prefix="/api/v1", tags=["companies"]) TIMELINE_FILTERS: dict[str, list[str]] = { "all": [], "products": ["PRODUCT"], "jobs": ["HIRING"], "pricing": ["PRICING"], "leadership": ["LEADERSHIP"], "locations": ["LOCATION"], "legal": ["LEGAL"], "news": ["COMMUNICATION", "INVESTOR_RELATIONS", "MARKETING"], "developer": ["DEVELOPER", "TECHNOLOGY"], "corporate": ["FINANCING", "M&A", "PARTNERSHIP", "STRATEGY"], } COMPARE_METRICS = [Metric.ACTIVITY_SCORE.value, Metric.HIRING_MOMENTUM_30D.value, Metric.PRODUCT_VELOCITY.value, Metric.AI_ADOPTION.value, Metric.GEO_EXPANSION.value, Metric.DEVELOPER_MOMENTUM.value, Metric.CORPORATE_CHANGE_INDEX.value, Metric.OPEN_JOBS.value, Metric.ANOMALY_SCORE.value] def _pub(response: Response, s: int = 60) -> None: response.headers["cache-control"] = public_cache_value(s) # ------------------------------------------------------------------------------------------------ list / compare @router.get("/companies", summary="Company directory") async def list_companies(response: Response, p: PageDep, q_: str | None = Query(None, alias="q", max_length=200), country: str | None = None, industry: str | None = None, tier: int | None = Query(None, ge=1, le=4), public: str | None = None, status: str | None = None, has_events: str | None = None, sort: str = Query("activity", pattern="^(activity|events|hiring|name|importance|recent|relevance)$"), sparkline: str | None = None) -> dict[str, Any]: _pub(response, 60) where, params = q.company_filters(q=q_, country=country, industry=industry, tier=tier, public=q.parse_bool(public), status=status, has_events=q.parse_bool(has_events)) async with connection() as conn: ids, total = await q.company_page_ids(conn, where, params, sort=sort, limit=p.per_page, offset=p.offset) rows = await q.fetch_cards_by_ids(conn, ids, sparkline=bool(q.parse_bool(sparkline))) return page_payload([ser.company_card(r) for r in rows], total, p) @router.get("/companies/compare", summary="Side-by-side comparison of 2–6 companies") async def compare(response: Response, companies: str = Query(..., description="comma-separated slugs or ids"), days: int = Query(90, ge=7, le=365)) -> dict[str, Any]: _pub(response, 120) keys = list(dict.fromkeys(q.csv_list(companies, maxlen=6))) if len(keys) < 2: raise HTTPException(status_code=422, detail="companies: provide 2 to 6 comma-separated slugs") async with connection() as conn: found = [] for k in keys: c = await q.get_company(conn, k) if c is None: raise HTTPException(status_code=404, detail=f"company not found: {k}") found.append(c) ids = [c["id"] for c in found] slug_of = {c["id"]: c["slug"] for c in found} cards = [ser.company_card(r) for r in await q.fetch_cards_by_ids(conn, ids, sparkline=True)] series_rows = await fetch_all(conn, "select company_id, day, value, confidence from metric_series where metric = 'activity_score' and " "company_id = any(cast(:ids as text[])) and day >= :d order by day", ids=ids, d=q.days_ago(days).date()) ev_rows = await fetch_all(conn, "select company_id, event_type, count(*) as n from events where status = 'active' and detected_at >= :d and " "company_id = any(cast(:ids as text[])) group by 1, 2", ids=ids, d=q.days_ago(30)) job_rows = await fetch_all(conn, "select company_id, count(*) filter (where status = 'open') as open, " "count(*) filter (where status = 'open' and is_ai) as ai_open, " "count(*) filter (where first_seen_at >= :d and not baseline) as new_30d from jobs where company_id = any(cast(:ids as text[])) " "group by company_id", ids=ids, d=q.days_ago(30)) 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[])) " "group by company_id", ids=ids) metrics: dict[str, dict[str, Any]] = {m: {} for m in COMPARE_METRICS} for card in cards: for m, v in card["metrics"].items(): metrics.setdefault(m, {})[card["slug"]] = v series: dict[str, list[dict[str, Any]]] = {s: [] for s in slug_of.values()} for r in series_rows: series[slug_of[r["company_id"]]].append(ser.metric_point(r)) events_30d: dict[str, dict[str, int]] = {s: {} for s in slug_of.values()} for r in ev_rows: events_30d[slug_of[r["company_id"]]][r["event_type"]] = int(r["n"]) 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} for s in slug_of.values(): jobs.setdefault(s, {"open": 0, "ai_open": 0, "new_30d": 0}) locations = {slug_of[r["company_id"]]: int(r["n"]) for r in loc_rows} for s in slug_of.values(): locations.setdefault(s, 0) return {"companies": cards, "metrics": metrics, "series": series, "events_30d": events_30d, "jobs": jobs, "locations": locations} # ------------------------------------------------------------------------------------------------ detail @router.get("/companies/{key}", summary="Company profile") async def company_detail(key: str, response: Response) -> dict[str, Any]: _pub(response, 60) async with connection() as conn: c = await q.require_company(conn, key) cid = c["id"] rows = await q.fetch_cards_by_ids(conn, [cid], sparkline=True) out = ser.company_card(rows[0]) out["company_type"] = c.get("company_type") out["employees"] = c.get("employees") out["wikidata_id"] = c.get("wikidata_id") out["indexed"] = bool(c.get("indexed")) out["discovered_at"] = c.get("discovered_at") out["first_observed_at"] = c.get("first_observed_at") out["last_change_at"] = c.get("last_change_at") 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)] 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) 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, " "r.last_seen_at, o.slug, o.display_name from company_relationships r left join companies o on o.id = r.to_company_id " "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) out["relationships"] = [ser.relationship(r) for r in rel] out["profile"] = ser._dict(c.get("source_meta")).get("profile") or None out["facts"] = profile_facts(out["profile"]) 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) out["metrics_detail"] = [{"metric": r["metric"], "value": ser.metric_value(r["metric"], r["value"]), "confidence": ser._float(r["confidence"], 3), "computed_at": r["computed_at"], "formula_version": r["formula_version"], "inputs": ser._dict(r["inputs"])} for r in md] sbs = await fetch_all(conn, "select surface, count(*) as n, count(*) filter (where status = 'active') as active from sensors where company_id = :id " "and status <> 'retired' group by surface order by n desc", id=cid) out["sensors_by_surface"] = {r["surface"]: int(r["n"]) for r in sbs} total_sensors = sum(int(r["n"]) for r in sbs) active_sensors = sum(int(r["active"]) for r in sbs) days_observed = await fetch_one(conn, "select count(*) as n from company_daily where company_id = :id and observations > 0", id=cid) hist = next((m["value"] for m in out["metrics_detail"] if m["metric"] == Metric.HISTORICAL_COVERAGE.value), None) out["coverage"] = {"historical_coverage": hist, "first_observed_at": c.get("first_observed_at"), "days_observed": int(days_observed["n"]) if days_observed else 0, "sensor_uptime": round(active_sensors / total_sensors, 3) if total_sensors else None} sig = await fetch_all(conn, "select * from signals where company_id = :id and status = 'active' order by detected_at desc limit 10", id=cid) out["signals"] = [ser.signal(r) for r in sig] act = await q.metric_series(conn, cid, Metric.ACTIVITY_SCORE.value, 30) hir = await q.metric_series(conn, cid, Metric.OPEN_JOBS.value, 90) out["sparklines"] = {"activity_30d": [ser._float(r["value"], 1) for r in act], "hiring_90d": [ser._float(r["value"], 1) for r in hir]} out["recent_events"] = [ser.event(r) for r in await q.fetch_events(conn, *q.event_filters(company_id=cid), limit=10)] return out # ------------------------------------------------------------------------------------------------ sub-resources @router.get("/companies/{key}/events", summary="Company events") async def company_events(key: str, response: Response, p: PageDep, event_type: str | None = None, event_subtype: str | None = None, since: str | None = None, until: str | None = None, min_importance: float | None = Query(None, ge=0, le=1), surface: str | None = None, sort: str = Query("recent", pattern="^(recent|importance)$")) -> dict[str, Any]: _pub(response, 30) async with connection() as conn: c = await q.require_company(conn, key) where, params = q.event_filters(company_id=c["id"], event_type=event_type, event_subtype=event_subtype, since=q.parse_iso(since), until=q.parse_iso(until, "until"), min_importance=min_importance, surface=surface) rows = await q.fetch_events(conn, where, params, sort=sort, limit=p.per_page, offset=p.offset) total = await q.count_events(conn, where, params) return page_payload([ser.event(r) for r in rows], total, p) @router.get("/companies/{key}/timeline", summary="Timeline grouped by day") async def company_timeline(key: str, response: Response, filter: str = Query("all", pattern="^(all|products|jobs|pricing|leadership|locations|legal|news|developer|corporate)$"), limit: int = Query(200, ge=1, le=500), before: str | None = None) -> dict[str, Any]: _pub(response, 60) types = TIMELINE_FILTERS[filter] async with connection() as conn: c = await q.require_company(conn, key) where, params = q.event_filters(company_id=c["id"], event_types=types or None, until=q.parse_iso(before, "before")) rows = await q.fetch_events(conn, where, params, sort="recent", limit=limit) where_sql = " where " + " and ".join(where) 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" f"{where_sql} group by 1 order by 1 desc limit 730", **params) items = [] for r in rows: ev = ser.event(r) ev["day"] = r["detected_at"].date() items.append(ev) return {"items": items, "days": [{"day": d["day"], "count": int(d["count"])} for d in days], "filter": filter} @router.get("/companies/{key}/metrics", summary="Current metrics and time series") async def company_metrics(key: str, response: Response, metric: str | None = None, days: int = Query(90, ge=1, le=730)) -> dict[str, Any]: _pub(response, 300) async with connection() as conn: c = await q.require_company(conn, key) 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"]) params: dict[str, Any] = {"id": c["id"], "d": q.days_ago(days).date()} extra = "" if metric: extra = " and metric = cast(:m as text)" params["m"] = metric[:60] rows = await fetch_all(conn, f"select metric, day, value, confidence from metric_series where company_id = :id and day >= :d{extra} " "order by metric, day limit 20000", **params) series: dict[str, list[dict[str, Any]]] = defaultdict(list) for r in rows: series[r["metric"]].append(ser.metric_point(r)) return {"current": [{"metric": r["metric"], "value": ser.metric_value(r["metric"], r["value"]), "confidence": ser._float(r["confidence"], 3), "computed_at": r["computed_at"], "formula_version": r["formula_version"], "inputs": ser._dict(r["inputs"])} for r in cur], "series": dict(series), "days": days} @router.get("/companies/{key}/jobs", summary="Publicly listed jobs with summary") async def company_jobs(key: str, response: Response, p: PageDep, status: str = Query("open", pattern="^(open|removed|all)$"), q_: str | None = Query(None, alias="q", max_length=120), country: str | None = None, ai: str | None = None, department: str | None = None, remote: str | None = None, sort: str = Query("recent", pattern="^(recent|title|posted)$")) -> dict[str, Any]: _pub(response, 120) 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] where = ["j.company_id = :id"] async with connection() as conn: c = await q.require_company(conn, key) params: dict[str, Any] = {"id": c["id"]} if status == "open": where.append("j.status = 'open'") elif status == "removed": where.append("j.status <> 'open'") if q_: where.append("j.title ilike :jq") params["jq"] = f"%{q_.strip()}%" if country: where.append("j.country = cast(:jc as char(2))") params["jc"] = country.upper()[:2] if q.parse_bool(ai): where.append("j.is_ai") if department: where.append("j.department = cast(:jd as text)") params["jd"] = department[:80] if q.parse_bool(remote) is not None: where.append("j.remote = :jr") params["jr"] = q.parse_bool(remote) wsql = " and ".join(where) 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) total = await q.bounded_count(conn, f"from jobs j where {wsql}", params) d7 = q.days_ago(7) 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, " "count(*) filter (where removed_at >= :d7) as removed_7d, count(*) filter (where status = 'open' and is_ai) as ai_open, " "count(*) filter (where status = 'open' and remote is true) as remote_open, " "count(*) filter (where status = 'open' and remote is not null) as remote_known from jobs where company_id = :id", id=c["id"], d7=d7) or {} 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 " "group by country order by n desc limit 15", id=c["id"]) 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 " "group by department order by n desc limit 15", id=c["id"]) remote_known = int(s.get("remote_known") or 0) 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), "ai_open": int(s.get("ai_open") or 0), "by_country": [{"country": r["country"], "n": int(r["n"])} for r in by_country], "by_department": [{"department": r["department"], "n": int(r["n"])} for r in by_dept], "remote_ratio": round(int(s.get("remote_open") or 0) / remote_known, 3) if remote_known else None} payload = page_payload([ser.job(r) for r in rows], total, p) payload["meta"] = {"summary": summary} return payload @router.get("/companies/{key}/people", summary="Leadership listed on monitored pages and on Wikidata") async def company_people(key: str, response: Response) -> dict[str, Any]: _pub(response, 300) async with connection() as conn: c = await q.require_company(conn, key) 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"]) people = [ser.person(r) for r in rows] return {"listed": [p for p in people if p["status"] == "listed"], "no_longer_listed": [p for p in people if p["status"] != "listed"], "sources": sorted({p["source"] for p in people})} @router.get("/companies/{key}/products", summary="Products in the public catalog") async def company_products(key: str, response: Response) -> dict[str, Any]: _pub(response, 300) async with connection() as conn: c = await q.require_company(conn, key) rows = await fetch_all(conn, "select * from products where company_id = :id order by status, last_seen_at desc limit 500", id=c["id"]) 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"]} @router.get("/companies/{key}/pricing", summary="Current plans and every previous version") async def company_pricing(key: str, response: Response, history_limit: int = Query(200, ge=1, le=1000)) -> dict[str, Any]: _pub(response, 300) async with connection() as conn: c = await q.require_company(conn, key) 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"]) hist = await fetch_all(conn, "select * from pricing_plans where company_id = :id and status <> 'current' order by valid_from desc limit :lim", id=c["id"], lim=history_limit) return {"current": [ser.plan(r) for r in cur], "history": [ser.plan(r) for r in hist]} @router.get("/companies/{key}/locations", summary="Locations listed on monitored pages") async def company_locations(key: str, response: Response, status: str = Query("listed", pattern="^(listed|all)$")) -> dict[str, Any]: _pub(response, 300) async with connection() as conn: c = await q.require_company(conn, key) extra = " and status = 'listed'" if status == "listed" else "" rows = await fetch_all(conn, f"select * from locations where company_id = :id{extra} order by kind, country, city limit 1000", id=c["id"]) items = [ser.location(r) for r in rows] return {"items": items, "countries": sorted({r["country"] for r in rows if r["country"] and r["status"] == "listed"})} @router.get("/companies/{key}/news", summary="First-party news, blog and changelog items") async def company_news(key: str, response: Response, limit: int = Query(50, ge=1, le=500), category: str | None = None) -> dict[str, Any]: _pub(response, 120) async with connection() as conn: c = await q.require_company(conn, key) params: dict[str, Any] = {"id": c["id"], "lim": limit} extra = "" if category: extra = " and category = cast(:cat as text)" params["cat"] = category[:40] 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) return {"items": [ser.news_item(r) for r in rows]} @router.get("/companies/{key}/sensors", summary="Sensors attached to the company") async def company_sensors(key: str, response: Response, include_retired: str | None = None) -> dict[str, Any]: _pub(response, 120) async with connection() as conn: c = await q.require_company(conn, key) extra = "" if q.parse_bool(include_retired) else " and status <> 'retired'" 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"]) return {"items": [ser.sensor(r) for r in rows]} @router.get("/companies/{key}/history", summary="Historical page viewer index (≤ 20 versions per sensor)") async def company_history(key: str, response: Response, versions: int = Query(20, ge=1, le=20)) -> dict[str, Any]: _pub(response, 300) async with connection() as conn: c = await q.require_company(conn, key) sensors = await fetch_all(conn, "select * from sensors where company_id = :id and status <> 'retired' and snapshot_count > 0 " "order by quality_score desc, surface limit 200", id=c["id"]) if not sensors: 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"]) snaps = await fetch_all(conn, "select s.* from sensors sn cross join lateral (select * from snapshots x where x.sensor_id = sn.id " "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", id=c["id"], v=versions) by_sensor: dict[str, list[dict[str, Any]]] = defaultdict(list) for s in snaps: by_sensor[s["sensor_id"]].append(ser.snapshot(s)) out = [] for s in sensors: item = ser.sensor(s) item["versions"] = by_sensor.get(s["id"], []) out.append(item) return {"sensors": out} @router.get("/companies/{key}/similar", summary="Similar companies (industry, country, importance)") async def company_similar(key: str, response: Response, limit: int = Query(8, ge=1, le=50)) -> dict[str, Any]: _pub(response, 600) async with connection() as conn: c = await q.require_company(conn, key) rows = await fetch_all(conn, "select c.id from companies c where c.id <> :id and c.status = 'ACTIVE' and " "(c.industry_primary = cast(:ip as text) or c.industries && cast(:inds as text[]) or c.country = cast(:country as char(2))) " "order by (c.industry_primary is not distinct from cast(:ip as text) and :ip is not null) desc, " "(c.industries && cast(:inds as text[])) desc, (c.country is not distinct from cast(:country as char(2))) desc, " "abs(c.importance - :imp) asc, c.importance desc limit :lim", id=c["id"], ip=c.get("industry_primary"), inds=list(c.get("industries") or []), country=c.get("country"), imp=float(c.get("importance") or 0), lim=limit) cards = await q.fetch_cards_by_ids(conn, [r["id"] for r in rows]) return {"items": [ser.company_card(r) for r in cards]}