spb/company-atlas

1"""Atlassian Statuspage summary — `GET https://status.<company>/api/v2/summary.json` → {page:{name, updated_at}, status:{indicator, description},2components:[{name, status}], incidents:[{name, status, impact, shortlink, created_at}], scheduled_maintenances:[…]}. One block per3component and per active incident; the page-level `updated_at` is noise-normalised away. Verified 2026-09-12 against githubstatus.com."""4from __future__ import annotations56import hashlib7from collections.abc import Mapping8from typing import Any910from companyatlas.connectors._util import load_json, parse_date, text_of11from companyatlas.fetch import FetchResult12from companyatlas.sdk.connector import Connector, ConnectorMeta, register13from companyatlas.sdk.models import Block, ExtractedNewsItem, Extraction14from companyatlas.sdk.normalize import normalized_text, simhash15from companyatlas.taxonomy import FetchMode, Surface161718@register19class StatuspageConnector(Connector):20    meta = ConnectorMeta(connector_id="statuspage-v1", name="Statuspage summary", version="1", category=Surface.STATUS, fetch_mode=FetchMode.JSON,21                         default_interval_s=6 * 3600, url_pattern=r"/api/v2/summary\.json$", pattern_required=True, priority=40, accept="application/json",22                         description="Atlassian Statuspage public summary API")2324    def extract(self, sensor: Mapping[str, Any], result: FetchResult) -> Extraction:25        data = load_json(result)26        if not isinstance(data, dict):27            raise TypeError("statuspage summary is not an object")28        status = data.get("status") or {}29        components = [c for c in (data.get("components") or []) if isinstance(c, dict) and c.get("name")]30        incidents = [i for i in (data.get("incidents") or []) if isinstance(i, dict) and i.get("name")]31        maint = [m for m in (data.get("scheduled_maintenances") or []) if isinstance(m, dict) and m.get("name")]32        blocks: list[Block] = []33        lines = [f"Status: {status.get('description') or status.get('indicator') or 'unknown'}"]3435        def add(kind: str, key: str, text: str, weight: float, path: str) -> None:36            blocks.append(Block(key=key, kind=kind, text=text, path=path, hash=hashlib.sha256(normalized_text(text).encode()).hexdigest()[:16],37                                simhash=simhash(text), weight=weight, order=len(blocks)))3839        add("hero", "status:overall", lines[0], 1.5, "")40        for c in components:41            txt = f"{c['name']}: {c.get('status') or 'unknown'}"42            lines.append(txt)43            add("section", f"component:{hashlib.blake2b(str(c.get('id') or c['name']).encode(), digest_size=6).hexdigest()}", txt, 1.0, "Components")44        news: list[ExtractedNewsItem] = []45        for i in incidents + maint:46            txt = f"{i['name']} — {i.get('status') or ''} ({i.get('impact') or 'n/a'})"47            lines.append(txt)48            add("news_item", f"incident:{i.get('id') or i['name']}", txt, 1.3, "Incidents")49            link = i.get("shortlink") or result.final_url50            news.append(ExtractedNewsItem(title=str(i["name"])[:300], url=str(link), published_at=parse_date(i.get("created_at")), category="other",51                                          summary=text_of(i.get("impact"))))52        meta = {"indicator": status.get("indicator"), "description": status.get("description"), "component_count": len(components),53                "incident_count": len(incidents), "maintenance_count": len(maint), "page_name": (data.get("page") or {}).get("name"), "structured": True}54        return Extraction(text="\n".join(lines), blocks=blocks, title=f"{meta['page_name'] or 'Status'} — {lines[0]}", meta=meta, news=news)55