answers 403 to crawlers: events come from the feed only, articles are not followed announcement_events(facts, org, parsed.feed_items, source_name="openai.com/news", follow=False) facts.document_title = "OpenAI News" facts.document_entity = org elif target.doc_type == "model_page" and parsed.markdown: self._model_page(facts, org, target, parsed) return facts # ------------------------------------------------------------------------------------------ catalog def _catalog(self, facts: Facts, org: EntityRef, parsed: Parsed) -> None: md = parsed.markdown assert md facts.document_title = "OpenAI models" facts.document_entity = org featured = md.section(r"^Featured models") entries: list[tuple[str, str, str, int]] = [] # (name, id, description, rank) seen: set[str] = set() for m in CATALOG_ITEM.finditer(md.body): name, api_id, desc = m.group(1).strip(), m.group(2).strip(), clean_cell(m.group(3)) if not API_ID.match(api_id) or api_id in seen: continue seen.add(api_id) rank = 0 if m.group(0) in featured else (2 if LEGACY_WORDS.search(desc) else 1) entries.append((name, api_id, desc, rank)) for name, api_id, desc, _rank in entries: ref = _model(facts, org, name, api_id) facts.claim(ref, "description", desc) facts.claim(ref, "official_url", f"{DOCS}/models/{api_id}") low = desc.lower() # API-only catalog: weights are only "open" when the docs say so (gpt-oss) facts.claim(ref, "openness", "open-weights" if ("open-weight" in low or "open weight" in low) else "proprietary") for name, api_id, _desc, _rank in sorted(entries, key=lambda e: e[3])[:MAX_MODEL_PAGES]: facts.follow(f"{DOCS}/models/{api_id}.md", doc_type="model_page", key=f"model:{api_id}", meta={"model": name, "api_id": api_id, "content_type": "text/markdown"}, min_bytes=300, priority=1) # ------------------------------------------------------------------------------------------ model page def _model_page(self, facts: Facts, org: EntityRef, target: Target, parsed: Parsed) -> None: md = parsed.markdown assert md title = next((t for lvl, t in md.headings if lvl == 1), None) or target.meta.get("model") or "" api_id = target.meta.get("api_id") m = re.search(r"Model ID:\s*`([^`]+)`", md.body) if m: api_id = m.group(1).strip() if not title or not api_id: return ref = _model(facts, org, title, api_id) facts.document_title = title facts.document_entity = ref quotes = [q.strip() for q in re.findall(r"^>\s*(.+)$", md.body, re.MULTILINE) if "llms.txt" not in q] if quotes: facts.claim(ref, "description", clean_cell(quotes[0])) facts.claim(ref, "official_url", f"{DOCS}/models/{api_id}") details = md.section(r"^Model details") for line in details.split("\n"): line = line.strip().lstrip("-").strip() low = line.lower() if not line: continue if low.startswith("default snapshot:"): snap = clean_cell(line.split(":", 1)[1]) if snap: ref.aliases.append(snap) facts.claim(ref, "default_snapshot", snap) elif low.startswith("input modalities:"): facts.claim(ref, "modalities_input", _modalities(line.split(":", 1)[1])) elif low.startswith("output modalities:"): facts.claim(ref, "modalities_output", _modalities(line.split(":", 1)[1])) elif low.endswith("context window"): facts.claim(ref, "context_length", tokens(line), unit="tokens") elif low.endswith("max output tokens"): facts.claim(ref, "max_output_tokens", tokens(line), unit="tokens") elif low.endswith("knowledge cutoff"): raw = re.sub(r"knowledge cutoff", "", line, flags=re.IGNORECASE).strip() dt = parse_datetime(raw) facts.claim(ref, "knowledge_cutoff", month_year(raw) or (f"{dt.year:04d}-{dt.month:02d}" if dt else None)) mods_in = next((c.value for c in facts.claims if c.entity is ref and c.property == "modalities_input"), None) mods_out = next((c.value for c in facts.claims if c.entity is ref and c.property == "modalities_output"), None) if mods_in or mods_out: facts.claim(ref, "modalities", sorted(set(mods_in or []) | set(mods_out or []))) if mods_in and "image" in mods_in: facts.claim(ref, "vision", True) features = [ln.strip().lstrip("-").strip() for ln in md.section(r"^Supported features").split("\n") if ln.strip().startswith("-")] if features: facts.claim(ref, "tool_calling", "function_calling" in features) facts.claim(ref, "structured_output", "structured_outputs" in features) facts.claim(ref, "fine_tuning_available", "fine_tuning" in features) if "reasoning" in features: facts.claim(ref, "reasoning", True) facts.claim(ref, "capabilities", normalize_capabilities(features)) # OpenAI slugs → canonical (structured_outputs → structured_output…) facts.claim(ref, "capabilities_raw", features) snapshots = [clean_cell(ln.strip().lstrip("-").strip()) for ln in md.section(r"^Snapshots").split("\n") if ln.strip().startswith("-")] for s in snapshots: if s and API_ID.match(s) and s not in ref.aliases and s != api_id: ref.aliases.append(s) if snapshots: facts.claim(ref, "snapshots", snapshots) facts.claim(ref, "openness", "open-weights" if "open-weight" in md.body.lower() else None) # provider prices are taken from the pricing page only (complete tuple: batch, cache writes, long context) # ------------------------------------------------------------------------------------------ pricing def _pricing(self, facts: Facts, org: EntityRef, parsed: Parsed) -> None: md = parsed.markdown assert md facts.document_title = "OpenAI API pricing" provider = provider_ref(PROVIDER_KEY) facts.entities.append(provider) facts.document_entity = provider labels = _table_labels(md.body) prices: dict[str, Any] = {} # api id -> PriceObs def obs(api_id: str, display: str | None = None) -> Any: """Snapshot rows (gpt-4o-2024-05-13) price the base model (gpt-4o) under their own provider_model_id.""" if api_id not in prices: base = SNAPSHOT_SUFFIX.sub("", api_id) ref = _model(facts, org, display or base, base) if api_id != base and api_id not in ref.aliases: ref.aliases.append(api_id) prices[api_id] = facts.price(model=ref, provider=provider, provider_model_id=api_id, meta={"from": "pricing page"}) return prices[api_id] for i, t in enumerate(md.tables): hs = [clean_cell(h).lower() for h in t["headers"]] tier = labels[i] if i < len(labels) else "standard" if not hs or "training" in hs or hs[0] == "tool": continue if "short context input" in hs: self._flagship_table(t, hs, tier, obs) elif hs[:2] == ["model", "modality"]: self._modality_table(t, hs, tier, obs) elif hs[:2] == ["category", "model"] and tier == "standard": for r in t["rows"]: api_id, _note = _split_model_cell(r[1]) if not api_id or len(r) < 5: continue p = obs(api_id) p.input_per_mtok, p.cached_input_per_mtok, p.output_per_mtok = money(r[2]), money(r[3]), money(r[4]) p.features["category"] = clean_cell(r[0]) if clean_cell(r[2]).lower() == "free": p.features["free"] = True elif hs[:2] == ["model", "size"] and "price per second" in hs: for r in t["rows"]: api_id, _ = _split_model_cell(r[0]) if api_id and len(r) >= 5: p = obs(api_id) p.features.setdefault(f"{tier}_per_second_by_size", {})[clean_cell(r[1])] = money(r[4]) elif hs[:2] == ["model", "use case"]: for r in t["rows"]: api_id, _ = _split_model_cell(r[0]) if not api_id or len(r) < 5: continue p = obs(api_id) p.input_per_mtok = p.input_per_mtok if p.input_per_mtok is not None else money(r[2]) p.output_per_mtok = p.output_per_mtok if p.output_per_mtok is not None else money(r[3]) p.features["estimated_cost_per_minute"] = money(r[4]) p.features["use_case"] = clean_cell(r[1]) elif hs == ["model", "price per minute"]: for r in t["rows"]: api_id, _ = _split_model_cell(r[0]) if api_id and len(r) >= 2: obs(api_id).features["per_minute"] = money(r[1]) for p in prices.values(): p.features = {k: v for k, v in p.features.items() if v not in (None, {}, [])} # rows made only of dashes (e.g. a model announced without prices) carry nothing worth storing facts.prices = [p for p in facts.prices if any(v is not None for v in p.price_tuple()[:-1]) or p.features] def _flagship_table(self, t: dict[str, Any], hs: list[str], tier: str, obs: Any) -> None: col = {h: i for i, h in enumerate(hs)} for r in t["rows"]: api_id, note = _split_model_cell(r[0]) if not api_id or len(r) < len(hs): continue p = obs(api_id) def cell(h: str, row: list[str] = r) -> float | None: return money(row[col[h]]) if h in col else None if tier == "standard": p.input_per_mtok, p.cached_input_per_mtok, p.output_per_mtok = cell("short context input"), cell("short context cached input"), cell("short context output") p.cache_write_per_mtok = cell("short context cache writes") for h in ("long context input", "long context cached input", "long context cache writes", "long context output"): v = cell(h) if v is not None: p.features[h.replace(" ", "_") + "_per_mtok"] = v if note: p.features["context_note"] = note elif tier == "batch": p.batch_input_per_mtok, p.batch_output_per_mtok = cell("short context input"), cell("short context output") else: for h in ("short context input", "short context output"): v = cell(h) if v is not None: p.features[f"{tier.replace(' ', '_')}_{h.split()[-1]}_per_mtok"] = v def _modality_table(self, t: dict[str, Any], hs: list[str], tier: str, obs: Any) -> None: i_in = next((i for i, h in enumerate(hs) if h == "input"), None) i_cached = next((i for i, h in enumerate(hs) if h.startswith("cached")), None) i_out = next((i for i, h in enumerate(hs) if h.startswith("output")), None) for r in t["rows"]: api_id, _ = _split_model_cell(r[0]) if not api_id or i_in is None or i_out is None or len(r) <= max(i_in, i_out): continue modality = clean_cell(r[1]).lower() p = obs(api_id) v_in, v_out = money(r[i_in]), money(r[i_out]) v_cached = money(r[i_cached]) if i_cached is not None and i_cached < len(r) else None if tier == "standard" and modality == "text": p.input_per_mtok, p.output_per_mtok, p.cached_input_per_mtok = v_in, v_out, v_cached elif tier == "standard": p.features[f"{modality}_input_per_mtok"] = v_in p.features[f"{modality}_output_per_mtok"] = v_out if v_cached is not None: p.features[f"{modality}_cached_input_per_mtok"] = v_cached if modality == "image" and p.input_per_mtok is None and p.output_per_mtok is None: pass elif tier == "batch" and modality == "text": p.batch_input_per_mtok, p.batch_output_per_mtok = v_in, v_out elif tier == "batch": p.features[f"batch_{modality}_input_per_mtok"] = v_in p.features[f"batch_{modality}_output_per_mtok"] = v_out # ------------------------------------------------------------------------------------------ deprecations def _deprecations(self, facts: Facts, org: EntityRef, parsed: Parsed, observed: datetime) -> None: """A model can appear in several dated sections (one per retired snapshot): the claims describe its *latest* deprecation round, every retired snapshot is listed once, and the API id is left to the catalog (this page names families and snapshots).""" md = parsed.markdown assert md facts.document_title = "OpenAI deprecations" facts.document_entity = org rounds: dict[str, dict[str, Any]] = {} for level, heading in md.headings: m = re.match(r"^(\d{4}-\d{2}-\d{2}):\s*(.+)$", heading) if level != 3 or not m: continue announced = m.group(1) section = parse_markdown(md.section(re.escape(heading))) for t in section.tables: hs = [clean_cell(h).lower() for h in t["headers"]] i_date = next((i for i, h in enumerate(hs) if "shutdown" in h), None) i_model = next((i for i, h in enumerate(hs) if "model" in h), None) i_repl = next((i for i, h in enumerate(hs) if "replacement" in h or "substitute" in h), None) if i_date is None or i_model is None: continue for r in t["rows"]: if len(r) <= max(i_date, i_model): continue ids = _ids_in_cell(r[i_model]) if not ids: continue base = SNAPSHOT_SUFFIX.sub("", ids[0]) shutdown, tentative = _shutdown_date(r[i_date]) repl = _ids_in_cell(r[i_repl]) if i_repl is not None and i_repl < len(r) else [] entry = rounds.setdefault(base, {"snapshots": set(), "aliases": set(), "rounds": [], "dated": False}) entry["dated"] = entry["dated"] or ids[0] != base entry["snapshots"].update(i for i in ids[:1] if i != base) entry["aliases"].update(i for i in ids if i != base) entry["rounds"].append({"announced": announced, "shutdown": shutdown, "tentative": tentative, "replacement": SNAPSHOT_SUFFIX.sub("", repl[0]) if repl else None, "replacement_raw": repl[0] if repl else None}) # `gpt-4o-audio` (a bare family label, never a dated snapshot) and `gpt-4o-audio-preview-` name one lineage whose catalog id # is the -preview one; `o1` / `o1-preview` or `gpt-4-turbo` / `gpt-4-turbo-preview` are distinct models (each has its own snapshots) for plain in [k for k in rounds if f"{k}-preview" in rounds and not rounds[k]["dated"]]: target_entry = rounds[f"{plain}-preview"] target_entry["rounds"] += rounds[plain]["rounds"] target_entry["snapshots"] |= rounds[plain]["snapshots"] target_entry["aliases"] |= rounds[plain]["aliases"] | {plain} del rounds[plain] today = observed.date().isoformat() for base, entry in rounds.items(): ref = _model(facts, org, base, base, claim_id=False) for extra in sorted(entry["aliases"]): if extra not in ref.aliases: ref.aliases.append(extra) latest = max(entry["rounds"], key=lambda x: (x["shutdown"] or "", x["announced"])) facts.claim(ref, "deprecation_date", latest["announced"]) if latest["shutdown"]: facts.claim(ref, "retirement_date", latest["shutdown"]) facts.claim(ref, "retirement_tentative", latest["tentative"]) claim_status(facts, ref, "retired" if latest["shutdown"] < today else "deprecated") else: claim_status(facts, ref, "deprecated") if entry["snapshots"]: facts.claim(ref, "retired_snapshots", sorted(entry["snapshots"])) if latest["replacement"] and latest["replacement"] != base: new = _model(facts, org, latest["replacement"], latest["replacement"], claim_id=False) if latest["replacement_raw"] != latest["replacement"] and latest["replacement_raw"] not in new.aliases: new.aliases.append(latest["replacement_raw"]) facts.relate(ref, "superseded_by", new, attributes={"announced": latest["announced"], "shutdown": latest["shutdown"]}) # ---------------------------------------------------------------------------------------------- helpers def _model(facts: Facts, org: EntityRef, name: str, api_id: str, *, claim_id: bool = True) -> EntityRef: """One EntityRef per API id inside a Facts (so claims from several tables land on the same ref).""" for e in facts.entities: if e.entity_type == "model" and e.identifiers.get(ID_SCHEME) == api_id: return e aliases = [api_id] if api_id != name else [] ref = model_ref(facts, name, org, api_id=api_id, provider_key=PROVIDER_KEY, aliases=aliases) if claim_id: facts.claim(ref, "api_model_id", api_id) return ref def _modalities(cell: str) -> list[str]: return normalize_modalities(cell) def _split_model_cell(cell: str) -> tuple[str | None, str | None]: """'gpt-5.5 (<272K context length)' → ('gpt-5.5', '<272K context length'); 'Whisper' → (None, None).""" c = clean_cell(cell) m = re.match(r"^([^\s(]+)\s*(?:$(.*)$)?$", c) if not m: return None, None api_id = m.group(1).strip() if not API_ID.match(api_id): return None, None return api_id, (m.group(2).strip() if m.group(2) else None) def _ids_in_cell(cell: str) -> list[str]: ids: list[str] = [] for tok in re.split(r"[|,]| or ", clean_cell(cell)): tok = tok.strip().strip("`*").split(" ")[0].strip("`*") if tok and API_ID.match(tok) and not tok.startswith("ft-") and tok not in ids and tok != "---": ids.append(tok) return ids def _shutdown_date(cell: str) -> tuple[str | None, bool]: c = clean_cell(cell).replace("‑", "-").replace("‐", "-") tentative = "earliest" in c.lower() or "not sooner" in c.lower() m = re.search(r"(\d{4}-\d{2}-\d{2}|[A-Z][a-z]+ \d{1,2}, \d{4})", c) dt = parse_datetime(m.group(1)) if m else None return (dt.date().isoformat() if dt else None), tentative def _table_labels(body: str) -> list[str]: """Pricing tier ('standard', 'batch', 'flex', 'fast mode') preceding each Markdown table, in table order.""" labels: list[str] = [] lines = body.split("\n") current = "standard" i = 0 while i < len(lines): line = lines[i].strip() low = line.lower() if low in TIER_LABELS: current = "fast mode" if low == "fast" else low elif re.match(r"^[A-Z][a-z]+ models?$", line) or re.match(r"^(Tools|Finetuning|Transcription models)$", line): current = "standard" if line.startswith("|") and i + 1 < len(lines) and re.match(r"^\s*\|?\s*:?-{2,}", lines[i + 1]): labels.append(current) i += 2 while i < len(lines) and lines[i].strip().startswith("|"): i += 1 continue i += 1 return labels CONNECTORS = [OpenAIConnector]