1"""Anthropic — official docs (served as Markdown by the docs platform) + newsroom listing.23Sources (tier 1):4 * models overview → comparison table transposed into one model per column: API ids, context, max output, cutoffs, retirement5 * pricing → per-model price table → PriceObs for the Anthropic API provider (+ status hints such as "retired")6 * model deprecations → status table keyed by API model name → status / deprecation / retirement claims7 * model pages → per-model spec tables (discovered from the overview)8 * newsroom → ANNOUNCEMENT events, release articles queued for LLM extraction9"""10from __future__ import annotations1112import re13from datetime import UTC, datetime1415from aiatlas.registry import org_ref, provider_ref16from aiatlas.sdk.connector import BaseConnector, Parsed, RunContext17from aiatlas.sdk.extract.dates import parse_datetime18from aiatlas.sdk.extract.feeds import FeedItem19from aiatlas.sdk.facts import Facts, Target20from aiatlas.sdk.fetch import FetchResult2122from ._common import announcement_events, clean_cell, kv_tables, link_in_cell, model_ref, money, month_year, parse_retirement, tokens, transpose_feature_table2324DOCS = "https://docs.claude.com/en/docs/about-claude"25NEWS = "https://www.anthropic.com/news"26PROVIDER_KEY = "anthropic"2728CLAUDE_NAME = re.compile(r"^(Claude [A-Za-z]+(?: [0-9.]+)?(?: [A-Za-z]+)?)")293031class AnthropicConnector(BaseConnector):32 name = "anthropic"33 label = "Anthropic — models, pricing, deprecations, news"34 description = "Official Claude docs (models overview, pricing, deprecations, model pages) and the Anthropic newsroom."35 source_key = "docs.claude.com"36 version = "1"37 parser_version = "1"38 interval_seconds = 360039 min_interval_seconds = 180040 rate_per_min = 1541 tier = 142 priority = 043 expected_min_records = 444 concurrency = 24546 async def discover(self, ctx: RunContext) -> list[Target]:47 return [48 Target(url=f"{DOCS}/models/overview.md", doc_type="model_docs", key="models", min_bytes=2000),49 Target(url=f"{DOCS}/pricing.md", doc_type="pricing", key="pricing", min_bytes=2000),50 Target(url=f"{DOCS}/model-deprecations.md", doc_type="model_docs", key="deprecations", min_bytes=1000),51 Target(url=NEWS, doc_type="listing", key="news", min_bytes=5000),52 ]5354 async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts:55 facts = Facts()56 org = org_ref("anthropic")57 facts.entities.append(org)58 key = target.key or target.meta.get("kind")59 if key == "models" and parsed.markdown:60 self._models_overview(facts, org, parsed)61 elif key == "pricing" and parsed.markdown:62 self._pricing(facts, org, parsed)63 elif key == "deprecations" and parsed.markdown:64 self._deprecations(facts, org, parsed)65 elif key == "news" and parsed.html:66 self._news(facts, org, parsed, res)67 elif target.doc_type == "model_page" and parsed.markdown:68 self._model_page(facts, org, target, parsed)69 return facts7071 # ------------------------------------------------------------------------------------------ models overview72 def _models_overview(self, facts: Facts, org, parsed: Parsed) -> None: # type: ignore[no-untyped-def]73 md = parsed.markdown74 assert md75 facts.document_title = md.front_matter.get("title") or "Models overview"76 table = next((t for t in md.tables if t["headers"] and clean_cell(t["headers"][0]).lower() == "feature"), None)77 if not table:78 return79 raw_headers = table["headers"]80 per_model = transpose_feature_table(table)81 for i, (model_name, feats) in enumerate(per_model.items()):82 api_id = feats.get("claude api id") or feats.get("anthropic api id")83 alias = feats.get("claude api alias") or feats.get("anthropic api alias")84 aliases = [a for a in (api_id, alias) if a]85 ref = model_ref(facts, model_name, org, api_id=api_id, provider_key=PROVIDER_KEY, family="Claude", aliases=aliases + _name_variants(model_name))86 facts.claim(ref, "openness", "proprietary")87 facts.claim(ref, "status", "active")88 facts.claim(ref, "description", feats.get("description"))89 facts.claim(ref, "api_model_id", api_id)90 facts.claim(ref, "api_alias", alias)91 facts.claim(ref, "context_length", tokens(feats.get("context window", "")), unit="tokens")92 facts.claim(ref, "max_output_tokens", tokens(feats.get("max output", "")), unit="tokens")93 facts.claim(ref, "knowledge_cutoff", month_year(feats.get("reliable knowledge cutoff", "")))94 facts.claim(ref, "training_data_cutoff", month_year(feats.get("training data cutoff", "")))95 facts.claim(ref, "latency_tier", feats.get("comparative latency"))96 facts.claim(ref, "thinking", feats.get("thinking") or feats.get("extended thinking"))97 facts.claim(ref, "default_effort", feats.get("default effort"))98 facts.claim(ref, "modalities", ["text", "image"])99 facts.claim(ref, "modalities_input", ["text", "image"])100 facts.claim(ref, "modalities_output", ["text"])101 facts.claim(ref, "tool_calling", True)102 facts.claim(ref, "vision", True)103 retire, tentative = parse_retirement(feats.get("retirement", ""))104 if retire:105 facts.claim(ref, "retirement_date", retire)106 facts.claim(ref, "retirement_tentative", tentative)107 for platform, scheme in (("amazon bedrock id", "bedrock_model_id"), ("google cloud id", "vertex_model_id"), ("microsoft foundry id", "foundry_model_id"),108 ("claude platform on aws id", "claude_aws_model_id")):109 v = feats.get(platform)110 if v and v not in ("—", "-"):111 facts.claim(ref, scheme, v)112 page_url = link_in_cell(feats.get("model page", "")) or (link_in_cell(raw_headers[i + 1]) if i + 1 < len(raw_headers) else None)113 if not page_url:114 # header cells are plain names; the "Model page" row holds the links in the raw table115 for row in parsed.markdown.tables[0]["rows"] if parsed.markdown else []:116 pass117 if page_url:118 if not page_url.endswith(".md"):119 page_url = page_url.rstrip("/") + ".md"120 facts.follow(page_url, doc_type="model_page", entity=ref, key=f"model:{api_id or model_name}", meta={"model": model_name, "api_id": api_id}, min_bytes=500)121 facts.document_entity = org122 # raw table rows keep the links: recover "Model page" links precisely123 for t in md.tables:124 if t["headers"] and clean_cell(t["headers"][0]).lower() == "feature":125 for row in t["rows"]:126 if row and clean_cell(row[0]).lower() == "model page":127 for name, cell in zip([clean_cell(h) for h in t["headers"][1:]], row[1:], strict=False):128 url = link_in_cell(cell)129 if url and not any(f.url.startswith(url.rstrip("/")) for f in facts.targets):130 ref = next((e for e in facts.entities if e.entity_type == "model" and e.name == name), None)131 facts.follow(url.rstrip("/") + ".md", doc_type="model_page", entity=ref, key=f"model:{name}", meta={"model": name}, min_bytes=500)132133 # ------------------------------------------------------------------------------------------ model page134 def _model_page(self, facts: Facts, org, target: Target, parsed: Parsed) -> None: # type: ignore[no-untyped-def]135 md = parsed.markdown136 assert md137 name = target.meta.get("model") or md.front_matter.get("title") or ""138 if not name:139 return140 ref = target.entity or model_ref(facts, name, org, api_id=target.meta.get("api_id"), provider_key=PROVIDER_KEY, family="Claude")141 if ref not in facts.entities:142 facts.entities.append(ref)143 kv = kv_tables(md.tables)144 facts.claim(ref, "official_url", md.front_matter.get("url"))145 facts.claim(ref, "description", md.front_matter.get("description"))146 for k, prop in (("context window", "context_length"), ("max output", "max_output_tokens")):147 if kv.get(k):148 facts.claim(ref, prop, tokens(kv[k]), unit="tokens")149 for k, prop in (("reliable knowledge cutoff", "knowledge_cutoff"), ("training data cutoff", "training_data_cutoff")):150 if kv.get(k):151 facts.claim(ref, prop, month_year(kv[k]))152 if kv.get("release date") or kv.get("released"):153 dt = parse_datetime(kv.get("release date") or kv.get("released"))154 if dt:155 facts.claim(ref, "release_date", dt.date().isoformat())156 facts.document_entity = ref157 facts.document_title = md.front_matter.get("title")158159 # ------------------------------------------------------------------------------------------ pricing160 def _pricing(self, facts: Facts, org, parsed: Parsed) -> None: # type: ignore[no-untyped-def]161 md = parsed.markdown162 assert md163 facts.document_title = md.front_matter.get("title") or "Pricing"164 provider = provider_ref(PROVIDER_KEY)165 facts.entities.append(provider)166 table = next((t for t in md.tables if t["headers"] and clean_cell(t["headers"][0]).lower() == "model" and any("output" in clean_cell(h).lower() for h in t["headers"])), None)167 if not table:168 return169 headers = [clean_cell(h).lower() for h in table["headers"]]170171 def col(*needles: str) -> int | None:172 for i, h in enumerate(headers):173 if all(n in h for n in needles):174 return i175 return None176177 c_in, c_out = col("input"), col("output")178 c_w5, c_w1, c_hit = col("5m", "cache"), col("1h", "cache"), col("cache hit")179 for row in table["rows"]:180 if not row or c_in is None or c_out is None:181 continue182 raw_name = row[0]183 name_clean = clean_cell(raw_name)184 m = CLAUDE_NAME.match(name_clean)185 model_name = m.group(1).strip() if m else name_clean.split("(")[0].strip()186 note = name_clean[len(model_name):].strip(" ()")187 status = None188 low = note.lower()189 if "retired" in low:190 status = "retired"191 elif "deprecated" in low:192 status = "deprecated"193 elif "limited availability" in low:194 status = "limited-availability"195 ref = model_ref(facts, model_name, org, family="Claude", aliases=_name_variants(model_name))196 facts.claim(ref, "openness", "proprietary")197 if status:198 facts.claim(ref, "status", status)199 facts.claim(ref, "availability_note", note)200 facts.price(model=ref, provider=provider, input_per_mtok=money(row[c_in]), output_per_mtok=money(row[c_out]),201 cached_input_per_mtok=money(row[c_hit]) if c_hit is not None and c_hit < len(row) else None,202 cache_write_per_mtok=money(row[c_w5]) if c_w5 is not None and c_w5 < len(row) else None,203 features={"cache_write_1h_per_mtok": money(row[c_w1]) if c_w1 is not None and c_w1 < len(row) else None},204 meta={"from": "pricing page", "note": note or None})205 # batch discount, if stated as a multiplier/percent206 for t in md.tables:207 hs = [clean_cell(h).lower() for h in t["headers"]]208 if hs and "cache operation" in hs[0]:209 facts.claim(provider, "prompt_caching", {clean_cell(r[0]): clean_cell(r[1]) for r in t["rows"] if len(r) >= 2})210 facts.document_entity = provider211212 # ------------------------------------------------------------------------------------------ deprecations213 def _deprecations(self, facts: Facts, org, parsed: Parsed) -> None: # type: ignore[no-untyped-def]214 md = parsed.markdown215 assert md216 facts.document_title = md.front_matter.get("title") or "Model deprecations"217 table = next((t for t in md.tables if t["headers"] and "api model name" in clean_cell(t["headers"][0]).lower()), None)218 if not table:219 return220 headers = [clean_cell(h).lower() for h in table["headers"]]221 i_state = next((i for i, h in enumerate(headers) if "state" in h or "status" in h), 1)222 i_dep = next((i for i, h in enumerate(headers) if "deprecated" in h), 2)223 i_ret = next((i for i, h in enumerate(headers) if "retirement" in h), 3)224 for row in table["rows"]:225 if len(row) <= max(i_state, i_dep, i_ret):226 continue227 api_id = clean_cell(row[0])228 if not api_id.startswith("claude"):229 continue230 display = _display_name(api_id)231 ref = model_ref(facts, display, org, api_id=api_id, provider_key=PROVIDER_KEY, family="Claude", aliases=[api_id] + _name_variants(display))232 state = clean_cell(row[i_state]).lower()233 facts.claim(ref, "status", {"active": "active", "retired": "retired", "deprecated": "deprecated"}.get(state, state))234 dep = clean_cell(row[i_dep])235 if dep and dep.upper() != "N/A":236 d = parse_datetime(dep)237 if d:238 facts.claim(ref, "deprecation_date", d.date().isoformat())239 ret, tentative = parse_retirement(clean_cell(row[i_ret]))240 if ret:241 facts.claim(ref, "retirement_date", ret)242 facts.claim(ref, "retirement_tentative", tentative)243 facts.document_entity = org244245 # ------------------------------------------------------------------------------------------ news246 def _news(self, facts: Facts, org, parsed: Parsed, res: FetchResult) -> None: # type: ignore[no-untyped-def]247 html = parsed.html248 assert html249 items: list[FeedItem] = []250 seen: set[str] = set()251 for node in html.css("a[href^='/news/']"):252 href = node.attributes.get("href") or ""253 if href in seen or href.count("/") != 2:254 continue255 text = re.sub(r"\s+", " ", node.text(separator=" | ", strip=True))256 parts = [p.strip() for p in text.split("|") if p.strip()]257 if len(parts) < 2:258 continue259 date = None260 category = None261 title = None262 summary = None263 for p in parts:264 if date is None and re.fullmatch(r"[A-Z][a-z]{2} \d{1,2}, \d{4}", p):265 date = parse_datetime(p)266 elif category is None and p in ("Announcements", "Product", "Policy", "Research", "Interpretability", "Alignment", "Societal Impacts", "Economic Research", "Education"):267 category = p268 elif title is None:269 title = p270 elif summary is None:271 summary = p272 if not title:273 continue274 seen.add(href)275 items.append(FeedItem(id=href, url=f"https://www.anthropic.com{href}", title=title[:300], summary=summary, published_at=date, updated_at=None,276 categories=[category] if category else []))277 announcement_events(facts, org, items, source_name="anthropic.com/news", follow=True, max_follow=25)278 facts.document_title = "Anthropic newsroom"279 facts.document_entity = org280281282def _name_variants(name: str) -> list[str]:283 """Anthropic has used both 'Claude 3.5 Haiku' and 'Claude Haiku 3.5' — register both orders as aliases."""284 m = re.fullmatch(r"Claude (\d[\d.]*) ([A-Za-z]+)", name)285 if m:286 return [f"Claude {m.group(2)} {m.group(1)}"]287 m = re.fullmatch(r"Claude ([A-Za-z]+) (\d[\d.]*)", name)288 if m:289 return [f"Claude {m.group(2)} {m.group(1)}"]290 return []291292293def _display_name(api_id: str) -> str:294 """claude-opus-4-5-20251101 → Claude Opus 4.5 ; claude-3-7-sonnet-20250219 → Claude 3.7 Sonnet ; claude-fable-5-1 → Claude Fable 5.1"""295 parts = api_id.split("-")296 parts = [p for p in parts if not re.fullmatch(r"\d{8}", p)]297 words: list[str] = []298 nums: list[str] = []299 for p in parts:300 if p.isdigit():301 nums.append(p)302 else:303 if nums:304 words.append(".".join(nums))305 nums = []306 words.append(p.capitalize())307 if nums:308 words.append(".".join(nums))309 return " ".join(words)310311312CONNECTORS = [AnthropicConnector]313