spb/company-atlas

1"""Digest data (spec §142): weekly company digest and industry / country market digests as JSON. No e-mail sending here — the API or2a future mailer renders these. Everything comes from measured tables (events, metrics, jobs, signals); empty sections stay empty.3"""4from __future__ import annotations56from datetime import UTC, datetime, timedelta7from typing import Any89from companyatlas.db import fetch_all, fetch_one, transaction1011EVENT_COLS = "id, event_type, event_subtype, importance, confidence, confidence_label, title, summary, detected_at, source_url, surface, origin"121314async def company_digest(slug_or_id: str, *, days: int = 7, now: datetime | None = None) -> dict[str, Any] | None:15    now = now or datetime.now(UTC)16    since = now - timedelta(days=days)17    async with transaction() as conn:18        co = await fetch_one(conn, "select id, slug, display_name, canonical_domain, country, industries, last_event_at from companies where slug = :s or id = :s", s=slug_or_id)19        if co is None:20            return None21        events = await fetch_all(conn, f"select {EVENT_COLS} from events where company_id = :c and detected_at >= :since and status = 'active' order by importance desc, detected_at desc limit 50",22                                 c=co["id"], since=since)23        by_type: dict[str, int] = {}24        for e in events:25            by_type[e["event_type"]] = by_type.get(e["event_type"], 0) + 126        metrics = {r["metric"]: {"value": r["value"], "confidence": r["confidence"], "computed_at": r["computed_at"]} for r in27                   await fetch_all(conn, "select metric, value, confidence, computed_at from metrics_current where company_id = :c", c=co["id"])}28        series = await fetch_all(conn, "select metric, day, value from metric_series where company_id = :c and day >= :d and metric in ('activity_score', 'open_jobs') order by day",29                                 c=co["id"], d=(now - timedelta(days=days * 2)).date())30        jobs = await fetch_one(conn, """select count(*) filter (where status = 'open') as open, count(*) filter (where first_seen_at >= :since) as new,31                                              count(*) filter (where removed_at >= :since) as no_longer_listed, count(*) filter (where status = 'open' and is_ai) as ai_open32                                        from jobs where company_id = :c""", c=co["id"], since=since)33        signals = await fetch_all(conn, "select kind, strength, confidence, title, explanation, detected_at from signals where company_id = :c and status = 'active' order by strength desc", c=co["id"])34        sensors = await fetch_one(conn, "select count(*) filter (where status = 'active') as active, count(*) as total, max(last_success_at) as last_checked from sensors where company_id = :c", c=co["id"])35    return {"company": co, "window": {"days": days, "since": since.isoformat(), "until": now.isoformat()}, "highlights": events[:8], "events": events,36            "events_by_type": by_type, "metrics": metrics, "series": {m: [{"day": r["day"].isoformat(), "value": r["value"]} for r in series if r["metric"] == m] for m in ("activity_score", "open_jobs")},37            "jobs": jobs, "signals": signals, "coverage": sensors, "generated_at": now.isoformat()}383940async def scope_digest(scope: str, key: str, *, days: int = 7, now: datetime | None = None, limit: int = 10) -> dict[str, Any]:41    """scope ∈ industry | country. Movers = highest activity, top events = most important, hiring = aggregate momentum."""42    now = now or datetime.now(UTC)43    since = now - timedelta(days=days)44    if scope == "country":45        where = "co.country = :key"46    elif scope == "industry":47        where = "(:key = any(co.industries) or co.industry_primary = :key)"48    else:49        raise ValueError("scope must be industry or country")50    async with transaction() as conn:51        companies_n = await fetch_val_int(conn, f"select count(*) from companies co where {where}", key=key)52        events = await fetch_all(conn, f"""select e.{EVENT_COLS.replace(', ', ', e.')}, co.slug, co.display_name from events e join companies co on co.id = e.company_id53                                           where {where} and e.detected_at >= :since and e.status = 'active' order by e.importance desc, e.detected_at desc limit :limit""",54                                 key=key, since=since, limit=limit * 2)55        by_type = await fetch_all(conn, f"""select e.event_type, count(*) as n from events e join companies co on co.id = e.company_id56                                            where {where} and e.detected_at >= :since and e.status = 'active' group by 1 order by n desc""", key=key, since=since)57        movers = await fetch_all(conn, f"""select co.slug, co.display_name, m.value as activity_score from metrics_current m join companies co on co.id = m.company_id58                                           where {where} and m.metric = 'activity_score' order by m.value desc limit :limit""", key=key, limit=limit)59        hiring = await fetch_one(conn, f"""select count(*) filter (where j.status = 'open') as open, count(*) filter (where j.first_seen_at >= :since) as new,60                                                 count(*) filter (where j.removed_at >= :since) as no_longer_listed61                                           from jobs j join companies co on co.id = j.company_id where {where}""", key=key, since=since)62        momentum = await fetch_one(conn, f"""select avg(m.value) as avg_momentum_30d, count(*) as companies from metrics_current m join companies co on co.id = m.company_id63                                             where {where} and m.metric = 'hiring_momentum_30d'""", key=key)64        signals = await fetch_all(conn, "select kind, strength, title, explanation, evidence, detected_at from signals where scope = :s and scope_key = :k and status = 'active' order by strength desc",65                                  s=scope, k=key)66    return {"scope": scope, "key": key, "window": {"days": days, "since": since.isoformat(), "until": now.isoformat()}, "companies": companies_n,67            "top_events": events, "events_by_type": {r["event_type"]: r["n"] for r in by_type}, "movers": movers, "hiring": {**(hiring or {}), **(momentum or {})},68            "signals": signals, "generated_at": now.isoformat()}697071async def fetch_val_int(conn, sql: str, **params: Any) -> int:  # type: ignore[no-untyped-def]72    row = await fetch_one(conn, sql, **params)73    return int(next(iter(row.values()))) if row else 0747576__all__ = ["company_digest", "scope_digest"]77