"""DeepSeek — API docs (Docusaurus): models & pricing table, change log, news posts. Sources (tier 1): * pricing → api-docs.deepseek.com/quick_start/pricing : one transposed table (models as columns): API ids, model versions, context length, max output, feature matrix (JSON output, tool calls, vision…), peak / off-peak prices (cache hit, cache miss, output) * change log → api-docs.deepseek.com/updates : "Date: YYYY-MM-DD" sections → ANNOUNCEMENT events (release posts queued for LLM) * news → the docs top-nav "News" link (points to the latest post; /news/ itself redirects to the quick start); its sidebar lists every news post with its date → events + follow-ups www.deepseek.com is a client-rendered Next.js site (Chinese landing page, no model list in the HTML) — not crawled. """ from __future__ import annotations import re from typing import Any from aiatlas.registry import org_ref, provider_ref from aiatlas.sdk.connector import BaseConnector, Parsed, RunContext from aiatlas.sdk.extract.dates import parse_datetime from aiatlas.sdk.extract.feeds import FeedItem from aiatlas.sdk.extract.html import node_text from aiatlas.sdk.facts import EntityRef, Facts, Target from aiatlas.sdk.fetch import FetchResult from ._common import MODEL_WORDS, RELEASE_WORDS, announcement_events, clean_cell, model_ref, money, tokens DOCS = "https://api-docs.deepseek.com" PROVIDER_KEY = "deepseek" ID_SCHEME = "deepseek_model_id" FOOTNOTE = re.compile(r"\s*\(\d+\)\s*$") NEWS_LINK = re.compile(r"/news/news\d+/?$") NEWS_DATE = re.compile(r"(\d{4})/(\d{2})/(\d{2})\s*$") class DeepSeekConnector(BaseConnector): name = "deepseek" label = "DeepSeek — API models & pricing, change log, news" description = "DeepSeek API docs: models & pricing table (ids, context, max output, features, peak/off-peak prices), change log and news posts." source_key = "deepseek.com" version = "1" parser_version = "1" interval_seconds = 3600 min_interval_seconds = 1800 rate_per_min = 12 tier = 1 priority = 0 expected_min_records = 3 concurrency = 2 async def discover(self, ctx: RunContext) -> list[Target]: return [ Target(url=f"{DOCS}/quick_start/pricing", doc_type="pricing", key="pricing", min_bytes=5000), Target(url=f"{DOCS}/updates", doc_type="listing", key="changelog", min_bytes=5000), ] async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts: facts = Facts() org = org_ref("deepseek") facts.entities.append(org) key = target.key or "" if key == "pricing" and parsed.html: self._pricing(facts, org, parsed, res.final_url or res.url) elif key == "changelog" and parsed.html: self._changelog(facts, org, parsed, res.final_url or res.url) elif target.doc_type == "news_index" and parsed.html: self._news_index(facts, org, parsed) return facts # ------------------------------------------------------------------------------------------ models & pricing def _pricing(self, facts: Facts, org: EntityRef, parsed: Parsed, url: str) -> None: html = parsed.html assert html provider = provider_ref(PROVIDER_KEY) facts.entities.append(provider) facts.document_title, facts.document_entity = "DeepSeek models & pricing", provider table = next((t for t in html.tables if any(clean_cell(r[0]).upper().startswith("MODEL") for r in t["rows"] if r)), None) if not table: return rows = [[clean_cell(c) for c in r] for r in table["rows"]] ids = next(([FOOTNOTE.sub("", c) for c in r[1:]] for r in rows if r and r[0].upper().startswith("MODEL") and not r[0].upper().startswith("MODEL VERSION")), []) ids = [i for i in ids if re.fullmatch(r"[a-z0-9][a-z0-9.\-]*", i)] if not ids: return n = len(ids) versions = next(([c for c in r[1:]] for r in rows if r and r[0].upper().startswith("MODEL VERSION")), []) refs: list[EntityRef] = [] for i, api_id in enumerate(ids): display = versions[i] if i < len(versions) and versions[i] else api_id ref = model_ref(facts, display, org, api_id=api_id, provider_key=PROVIDER_KEY, family="DeepSeek", aliases=[api_id]) facts.claim(ref, "api_model_id", api_id) facts.claim(ref, "official_url", url) refs.append(ref) prices = [facts.price(model=ref, provider=provider, provider_model_id=api_id, meta={"from": "pricing page", "note": "peak-hour rates; off-peak in features"}) for ref, api_id in zip(refs, ids, strict=True)] def spread(values: list[str]) -> list[str]: return values * n if len(values) == 1 else values kind: str | None = None for r in rows: if not r: continue label = r[0].upper() if label.startswith("CONTEXT LENGTH"): for ref, v in zip(refs, spread(r[1:]), strict=False): facts.claim(ref, "context_length", tokens(v), unit="tokens") elif label.startswith("MAX OUTPUT"): for ref, v in zip(refs, spread(r[1:]), strict=False): facts.claim(ref, "max_output_tokens", tokens(re.sub(r"(?i)maximum:?\s*|default:?\s*\d+\w*", "", v)), unit="tokens") elif label.startswith("THINKING"): for ref in refs: facts.claim(ref, "reasoning", True) cells = r[1:] if not label.startswith("FEATURES") else r[2:] feat_label = r[1] if label.startswith("FEATURES") and len(r) > 1 else r[0] if feat_label in ("Json Output", "JSON Output", "Tool Calls", "Vision", "FIM Completion（Beta）", "Chat Prefix Completion（Beta）", "Responses API", "Anthropic API"): vals = spread(cells) for ref, v in zip(refs, vals, strict=False): supported = "✓" in v or v.lower().startswith(("yes", "support")) if feat_label == "Tool Calls": facts.claim(ref, "tool_calling", supported) elif feat_label.lower() == "json output": facts.claim(ref, "structured_output", supported) elif feat_label == "Vision": facts.claim(ref, "vision", supported) if supported: facts.claim(ref, "modalities_input", ["text", "image"]) # pricing rows: a "kind" cell then PEAK / OFF-PEAK rows with one money cell per model joined = " ".join(r).upper() if "INPUT TOKENS" in joined and "CACHE HIT" in joined: kind = "cached_input_per_mtok" elif "INPUT TOKENS" in joined and "CACHE MISS" in joined: kind = "input_per_mtok" elif "OUTPUT TOKENS" in joined: kind = "output_per_mtok" monies = [money(c) for c in r if c.startswith("$")] if kind and len(monies) == n: period = next((c.upper() for c in r if c.upper() in ("PEAK", "OFF-PEAK", "STANDARD")), "PEAK") for p, v in zip(prices, monies, strict=True): if period == "OFF-PEAK": p.features[f"off_peak_{kind}"] = v else: setattr(p, kind, v) for ref, api_id in zip(refs, ids, strict=True): for p in html.css("p"): t = node_text(p) if t.startswith("(") and api_id in t and "legacy" in t.lower(): for legacy in re.findall(r"\b(deepseek-[a-z0-9.\-]+)\b", t): if legacy != api_id and legacy not in ref.aliases: ref.aliases.append(legacy) facts.prices = [p for p in facts.prices if any(v is not None for v in p.price_tuple()[:-1])] # ------------------------------------------------------------------------------------------ change log def _changelog(self, facts: Facts, org: EntityRef, parsed: Parsed, url: str) -> None: html = parsed.html assert html facts.document_title, facts.document_entity = "DeepSeek API change log", org for h2 in html.css("h2"): m = re.search(r"Date:\s*(\d{4}-\d{2}-\d{2})", node_text(h2)) if not m: continue date = parse_datetime(m.group(1)) anchor = h2.attributes.get("id") or m.group(1) sib = h2.next title: str | None = None body: list[str] = [] n = 0 while sib is not None and sib.tag != "h2": if sib.tag == "h3": if title: n += 1 self._changelog_event(facts, org, url, anchor, n, date, title, " ".join(body)) body = [] title = node_text(sib).replace("\u200b", "").strip() elif sib.tag in ("p", "ul", "ol", "div"): body.append(node_text(sib)) sib = sib.next if title: n += 1 self._changelog_event(facts, org, url, anchor, n, date, title, " ".join(body)) news = next((href for href, text in html.links if text.strip() == "News" and NEWS_LINK.search(href)), None) if news: facts.follow(news, doc_type="news_index", key="news_index", priority=1, min_bytes=5000) def _changelog_event(self, facts: Facts, org: EntityRef, url: str, anchor: str, n: int, date: Any, title: str, body: str) -> None: is_release = bool(MODEL_WORDS.search(title) or RELEASE_WORDS.search(title)) facts.event("ANNOUNCEMENT", "release" if is_release else "company", f"DeepSeek: {title}", entity=org, importance=2 if is_release else 1, effective_at=date, dedupe_key=f"ANNOUNCEMENT:{url}#{anchor}:{n}", source_url=f"{url}#{anchor}", meta={"source": "api-docs.deepseek.com/updates", "summary": body[:300], "is_release": is_release}) # ------------------------------------------------------------------------------------------ news index (sidebar) def _news_index(self, facts: Facts, org: EntityRef, parsed: Parsed) -> None: html = parsed.html assert html items: list[FeedItem] = [] seen: set[str] = set() for href, text in html.links: if not NEWS_LINK.search(href) or "/zh-cn/" in href or href in seen: continue m = NEWS_DATE.search(text) title = NEWS_DATE.sub("", text).strip() if not title or title in ("News", "English"): continue seen.add(href) date = parse_datetime(f"{m.group(1)}-{m.group(2)}-{m.group(3)}") if m else None items.append(FeedItem(id=href, url=href, title=title, summary=None, published_at=date, updated_at=None)) announcement_events(facts, org, items, source_name="api-docs.deepseek.com/news", max_follow=10) facts.document_title, facts.document_entity = "DeepSeek news", org CONNECTORS = [DeepSeekConnector]