1"""Google — Gemini API docs (models, pricing, release notes) + Google DeepMind and Google AI blog feeds.23Sources (tier 1):4 * models → ai.google.dev/gemini-api/docs/models : model cards (name, endpoint id, status, description) + "Model | Endpoint" tables;5 per-model pages followed (Property/Description table: model code, data types, token limits, capabilities, versions, cutoff)6 * pricing → …/docs/pricing : one section per model (h2 + endpoint id) with Standard / Batch / Flex / Priority tables7 (paid tier, USD per 1M tokens; long-context tiers and scheduled changes kept in `features`)8 * release notes → …/docs/changelog : dated sections → ANNOUNCEMENT events9 * DeepMind blog → deepmind.google/blog/rss.xml (organization google-deepmind)10 * Google AI blog → blog.google/innovation-and-ai/technology/ai/rss/ (organization google)11"""12from __future__ import annotations1314import re15from html import unescape as html_unescape16from typing import Any1718from selectolax.parser import Node1920from aiatlas.registry import org_ref, provider_ref21from aiatlas.sdk.connector import BaseConnector, Parsed, RunContext22from aiatlas.sdk.extract.dates import parse_datetime23from aiatlas.sdk.extract.html import node_text24from aiatlas.sdk.facts import EntityRef, Facts, Target25from aiatlas.sdk.fetch import FetchResult2627from ._common import (28 MODEL_WORDS,29 RELEASE_WORDS,30 announcement_events,31 clean_cell,32 model_ref,33 money,34 month_year,35 slug_of,36 tokens,37)3839DOCS = "https://ai.google.dev/gemini-api/docs"40DEEPMIND_RSS = "https://deepmind.google/blog/rss.xml"41GOOGLE_AI_RSS = "https://blog.google/innovation-and-ai/technology/ai/rss/"42PROVIDER_KEY = "google-gemini-api"43ID_SCHEME = "gemini_model_id"44MAX_MODEL_PAGES = 454546ENDPOINT_ID = re.compile(r"^[a-z0-9][a-z0-9.\-]*$")47CAPABILITY = re.compile(r"(Audio generation|Caching|Code execution|Computer use|File search|Function calling|Grounding with Google Maps|Grounding with Google Search|"48 r"Image generation|Live API|Search grounding|Structured outputs|Thinking|URL context|Batch API|Flex inference|Priority inference|Tuning|"49 r"Video generation|Audio understanding|Image understanding)\s+(Supported|Not supported)(?:\s*\(([^)]*)\))?")50MODALITY_WORDS = {"text": "text", "image": "image", "images": "image", "video": "video", "audio": "audio", "pdf": "pdf", "code": "code", "embedding": "embedding"}515253class GoogleConnector(BaseConnector):54 name = "google"55 label = "Google — Gemini API models, pricing, release notes; DeepMind & Google AI blogs"56 description = "Gemini API developer docs (model cards, model pages, pricing, release notes) plus the Google DeepMind and Google AI blog feeds."57 source_key = "ai.google.dev"58 version = "1"59 parser_version = "1"60 interval_seconds = 360061 min_interval_seconds = 180062 rate_per_min = 1563 tier = 164 priority = 065 expected_min_records = 3066 concurrency = 26768 async def discover(self, ctx: RunContext) -> list[Target]:69 return [70 Target(url=f"{DOCS}/models", doc_type="model_docs", key="models", min_bytes=5000),71 Target(url=f"{DOCS}/pricing", doc_type="pricing", key="pricing", min_bytes=5000),72 Target(url=f"{DOCS}/changelog", doc_type="listing", key="changelog", min_bytes=3000),73 Target(url=DEEPMIND_RSS, doc_type="feed", key="deepmind_feed", min_bytes=1000),74 Target(url=GOOGLE_AI_RSS, doc_type="feed", key="google_ai_feed", min_bytes=1000),75 ]7677 async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts:78 facts = Facts()79 google = org_ref("google")80 key = target.key or ""81 if key == "deepmind_feed" and parsed.kind == "feed":82 deepmind = org_ref("google-deepmind")83 facts.entities.append(deepmind)84 announcement_events(facts, deepmind, parsed.feed_items, source_name="deepmind.google/blog", max_follow=15)85 facts.document_title, facts.document_entity = "Google DeepMind blog", deepmind86 return facts87 facts.entities.append(google)88 if key == "google_ai_feed" and parsed.kind == "feed":89 announcement_events(facts, google, parsed.feed_items, source_name="blog.google/technology/ai", max_follow=10)90 facts.document_title, facts.document_entity = "Google AI blog", google91 elif key == "models" and parsed.html:92 self._models(facts, google, parsed)93 elif key == "pricing" and parsed.html:94 self._pricing(facts, google, parsed)95 elif key == "changelog" and parsed.html:96 self._changelog(facts, google, parsed, res.final_url or res.url)97 elif target.doc_type == "model_page" and parsed.html:98 self._model_page(facts, google, target, parsed)99 return facts100101 # ------------------------------------------------------------------------------------------ models overview102 def _models(self, facts: Facts, org: EntityRef, parsed: Parsed) -> None:103 html = parsed.html104 assert html105 facts.document_title, facts.document_entity = "Gemini models", org106 ids: list[str] = []107 for card in html.css("a.gemini-card-centered"):108 href = card.attributes.get("href") or ""109 api_id = slug_of(href)110 h3 = card.css_first("h3")111 if not h3 or not ENDPOINT_ID.match(api_id):112 continue113 name = _clean_name(node_text(h3))114 status_node = card.css_first("p.status-subtext")115 status = _status(node_text(status_node)) if status_node else None116 # two cards may share a title (GA endpoint + its preview): the later one takes its status as suffix so alias117 # resolution never fuses separately priced endpoints ("Gemini Omni Flash" / "Gemini Omni Flash Preview")118 if any(e.entity_type == "model" and e.name.lower() == name.lower() and e.identifiers.get(ID_SCHEME) != api_id for e in facts.entities):119 suffix = "Preview" if status == "preview" else api_id120 name = f"{name} {suffix}" if not name.lower().endswith(suffix.lower()) else name121 ref = _model(facts, org, name, api_id)122 desc = card.css_first("p.description-centered")123 facts.claim(ref, "description", node_text(desc) if desc else None)124 facts.claim(ref, "status", status)125 if api_id not in ids:126 ids.append(api_id)127 api_aliases: dict[int, tuple[EntityRef, set[str]]] = {}128 for t in html.tables:129 hs = [clean_cell(h).lower() for h in t["headers"]]130 if not hs or hs[0] != "model" or "endpoint" not in hs:131 continue132 i_end = hs.index("endpoint")133 i_desc = hs.index("description") if "description" in hs else None134 for r in t["rows"]:135 if len(r) <= i_end:136 continue137 raw_name = clean_cell(r[0]).replace("\xa0", " ")138 endpoints = [e for e in re.split(r"\s+", clean_cell(r[i_end])) if ENDPOINT_ID.match(e)]139 if not endpoints:140 continue141 name = _clean_name(raw_name)142 # the same display name may carry several endpoints (GA + preview / live variants); the first one is the identity,143 # the others are recorded as a claim — never as aliases, which would merge separately priced endpoints144 same_name = next((e for e in facts.entities if e.entity_type == "model" and e.name.lower() == name.lower()), None)145 ref = same_name or _model(facts, org, name, endpoints[0])146 extra = {*endpoints} - {ref.identifiers.get(ID_SCHEME, "")}147 if extra:148 api_aliases.setdefault(id(ref), (ref, set()))[1].update(extra)149 note = re.search(r"\(([^)]*)\)", raw_name)150 if note:151 facts.claim(ref, "status", _status(note.group(1)))152 if i_desc is not None and i_desc < len(r):153 facts.claim(ref, "description", clean_cell(r[i_desc]))154 if endpoints[0] not in ids:155 ids.append(endpoints[0])156 for ref, extra in api_aliases.values():157 facts.claim(ref, "api_aliases", sorted(extra))158 for api_id in ids[:MAX_MODEL_PAGES]:159 facts.follow(f"{DOCS}/models/{api_id}", doc_type="model_page", key=f"model:{api_id}", meta={"api_id": api_id}, min_bytes=3000, priority=1)160161 # ------------------------------------------------------------------------------------------ model page162 def _model_page(self, facts: Facts, org: EntityRef, target: Target, parsed: Parsed) -> None:163 html = parsed.html164 assert html165 table = next((t for t in html.tables if [clean_cell(h).lower() for h in t["headers"]][:2] == ["property", "description"]), None)166 if not table:167 return168 props: dict[str, str] = {}169 for r in table["rows"]:170 if len(r) >= 2:171 label = re.sub(r"^[a-z0-9_]+\s+", "", clean_cell(r[0])) # drop the leading material-icon name ("id_card Model code")172 props[label.lower().replace("[*]", "").strip()] = clean_cell(r[1])173 api_id = props.get("model code") or target.meta.get("api_id")174 if not api_id or not ENDPOINT_ID.match(api_id):175 return176 name = _clean_name(next((t for lvl, t in html.headings if lvl == 1), None) or api_id)177 ref = _model(facts, org, name, api_id)178 facts.document_title, facts.document_entity = name, ref179 facts.claim(ref, "official_url", f"{DOCS}/models/{api_id}")180 types = props.get("supported data types", "")181 m = re.search(r"Inputs?\s+(.*?)\s+Outputs?\s+(.*)$", types, re.IGNORECASE)182 if m:183 mods_in, mods_out = _modalities(m.group(1)), _modalities(m.group(2))184 facts.claim(ref, "modalities_input", mods_in)185 facts.claim(ref, "modalities_output", mods_out)186 facts.claim(ref, "modalities", sorted(set(mods_in) | set(mods_out)))187 if "image" in mods_in:188 facts.claim(ref, "vision", True)189 if "audio" in mods_in:190 facts.claim(ref, "audio", True)191 limits = props.get("token limits", "")192 m_in = re.search(r"Input token limit\s+([\d,]+)", limits)193 m_out = re.search(r"Output token limit\s+([\d,]+)", limits)194 if m_in:195 facts.claim(ref, "context_length", tokens(m_in.group(1)), unit="tokens")196 if m_out:197 facts.claim(ref, "max_output_tokens", tokens(m_out.group(1)), unit="tokens")198 caps = props.get("capabilities", "")199 if caps:200 supported: dict[str, Any] = {}201 for cap, state, detail in CAPABILITY.findall(caps):202 supported[cap] = (detail or True) if state == "Supported" else False203 if supported:204 facts.claim(ref, "capabilities", sorted(k for k, v in supported.items() if v))205 if "Function calling" in supported:206 facts.claim(ref, "tool_calling", bool(supported["Function calling"]))207 if "Structured outputs" in supported:208 facts.claim(ref, "structured_output", bool(supported["Structured outputs"]))209 if "Thinking" in supported:210 facts.claim(ref, "reasoning", bool(supported["Thinking"]))211 versions: dict[str, list[str]] = {}212 for label, vid in re.findall(r"(Stable|Preview|Latest|Experimental):\s*([a-z0-9][a-z0-9.\-]*)", props.get("versions", "")):213 versions.setdefault(label.lower(), []).append(vid)214 facts.claim(ref, "versions", versions or None)215 facts.claim(ref, "knowledge_cutoff", month_year(props.get("knowledge cutoff", "")))216 facts.claim(ref, "latest_update", month_year(props.get("latest update", "")))217 for label, prop in (("release date", "release_date"), ("deprecation date", "deprecation_date"), ("shutdown date", "retirement_date"), ("retirement date", "retirement_date")):218 if props.get(label):219 dt = parse_datetime(props[label])220 facts.claim(ref, prop, dt.date().isoformat() if dt else None)221222 # ------------------------------------------------------------------------------------------ pricing223 def _pricing(self, facts: Facts, org: EntityRef, parsed: Parsed) -> None:224 html = parsed.html225 assert html226 provider = provider_ref(PROVIDER_KEY)227 facts.entities.append(provider)228 facts.document_title, facts.document_entity = "Gemini Developer API pricing", provider229 for section in html.css("div.models-section"):230 h2 = section.css_first("h2")231 code = section.css_first("em a code") or section.css_first("em code")232 if not h2 or not code:233 continue234 api_id = node_text(code)235 if not ENDPOINT_ID.match(api_id):236 continue237 name = _clean_name(node_text(h2))238 ref = _model(facts, org, name, api_id)239 tiers = _tier_tables(section)240 if not tiers:241 continue242 obs = facts.price(model=ref, provider=provider, provider_model_id=api_id, meta={"from": "pricing page", "tier": "paid, standard"})243 per_unit = {k: v for k, v in tiers.items() if k.startswith("per_")}244 for unit_key, rows in per_unit.items():245 # per-second (Veo), per-request/song (Lyria), per-image tables: keep every row as stated246 table: dict[str, list[str]] = {label: lines for label, _free, lines in rows if lines and not label.startswith("used to")}247 if table:248 obs.features[unit_key] = table249 if unit_key == "per_request":250 obs.per_request = next((money(lines[0]) for lines in table.values() if money(lines[0]) is not None), None)251 if unit_key == "per_image":252 obs.per_image = next((money(lines[0]) for lines in table.values() if money(lines[0]) is not None), None)253 standard = tiers.get("standard") or next((v for k, v in tiers.items() if not k.startswith("per_")), None)254 if standard:255 _apply_paid_rows(obs, standard, prefix="")256 if "batch" in tiers:257 batch: dict[str, Any] = {}258 _apply_paid_rows(batch, tiers["batch"], prefix="", as_dict=True)259 obs.batch_input_per_mtok = batch.get("input_per_mtok")260 obs.batch_output_per_mtok = batch.get("output_per_mtok")261 for tier in ("flex", "priority"):262 if tier in tiers:263 extra: dict[str, Any] = {}264 _apply_paid_rows(extra, tiers[tier], prefix="", as_dict=True)265 for k in ("input_per_mtok", "output_per_mtok"):266 if extra.get(k) is not None:267 obs.features[f"{tier}_{k}"] = extra[k]268 obs.features = {k: v for k, v in obs.features.items() if v not in (None, {}, [])}269 facts.prices = [p for p in facts.prices if any(v is not None for v in p.price_tuple()[:-1]) or any(k.startswith("per_") for k in p.features)]270271 # ------------------------------------------------------------------------------------------ release notes272 def _changelog(self, facts: Facts, org: EntityRef, parsed: Parsed, url: str) -> None:273 html = parsed.html274 assert html275 facts.document_title, facts.document_entity = "Gemini API release notes", org276 n_dates = 0277 for h2 in html.css("h2"):278 date = parse_datetime(node_text(h2)) if re.fullmatch(r"[A-Z][a-z]+ \d{1,2}, \d{4}", node_text(h2)) else None279 if not date:280 continue281 n_dates += 1282 if n_dates > 80:283 break284 anchor = h2.attributes.get("id") or date.date().isoformat()285 sib = h2.next286 i = 0287 while sib is not None and sib.tag != "h2":288 if sib.tag == "ul":289 for li in sib.iter():290 if li.tag != "li":291 continue292 strong = li.css_first("strong")293 title = node_text(strong).rstrip(":") if strong else node_text(li)[:120]294 text = node_text(li)295 if not title:296 continue297 i += 1298 codes = sorted({node_text(c) for c in li.css("code") if ENDPOINT_ID.match(node_text(c))})299 is_release = bool(RELEASE_WORDS.search(text) and MODEL_WORDS.search(text)) or bool(re.search(r"\b(GA|generally available|released|preview|deprecat)", text, re.IGNORECASE))300 facts.event("ANNOUNCEMENT", "release" if is_release else "company", f"Gemini API: {title}", entity=org, importance=2 if is_release else 1,301 effective_at=date, dedupe_key=f"ANNOUNCEMENT:{url}#{anchor}:{i}", source_url=f"{url}#{anchor}",302 meta={"source": "ai.google.dev/gemini-api/docs/changelog", "summary": text[:300], "models": codes[:10], "is_release": is_release})303 sib = sib.next304305306# ---------------------------------------------------------------------------------------------- helpers307def _model(facts: Facts, org: EntityRef, name: str, api_id: str) -> EntityRef:308 for e in facts.entities:309 if e.entity_type == "model" and e.identifiers.get(ID_SCHEME) == api_id:310 return e311 ref = model_ref(facts, name, org, api_id=api_id, provider_key="gemini", aliases=[api_id] if api_id != name else [])312 facts.claim(ref, "api_model_id", api_id)313 facts.claim(ref, "family", "Gemini" if api_id.startswith("gemini") else None)314 return ref315316317def _clean_name(name: str) -> str:318 name = name.replace("\xa0", " ").replace("🍌", "").strip()319 name = re.sub(r"\((Shut down|Deprecated|Retired)\)", "", name, flags=re.IGNORECASE).strip()320 return re.sub(r"\s+", " ", name)321322323def _status(text: str) -> str | None:324 low = (text or "").lower()325 if "shut down" in low or "retired" in low:326 return "retired"327 if "deprecat" in low:328 return "deprecated"329 if "experimental" in low or "preview" in low:330 return "preview"331 if "stable" in low or "ga" == low.strip():332 return "active"333 return None334335336def _modalities(text: str) -> list[str]:337 out: list[str] = []338 for tok in re.split(r"[,/]|\band\b", text.lower()):339 tok = tok.strip().strip(".")340 mod = MODALITY_WORDS.get(tok)341 if mod and mod not in out:342 out.append(mod)343 return out344345346def _table_rows(table: Node) -> tuple[str, list[tuple[str, str, list[str]]]]:347 """(paid-tier unit: tokens|second|request|image, [(row label, free cell, paid cell lines)])"""348 unit = "tokens"349 for th in table.css("thead th"):350 m = re.search(r"per\s+(1M tokens|second|request|image|minute)", node_text(th), re.IGNORECASE)351 if m:352 unit = m.group(1).lower().replace("1m ", "")353 rows: list[tuple[str, str, list[str]]] = []354 for tr in table.css("tbody tr"):355 tds = tr.css("td")356 if len(tds) < 2:357 continue358 label = node_text(tds[0]).lower()359 free = node_text(tds[1]) if len(tds) >= 3 else ""360 paid_html = tds[-1].html or ""361 lines = [clean_cell(html_unescape(re.sub(r"<[^>]+>", " ", part))) for part in re.split(r"<br\s*/?>", paid_html)]362 rows.append((label, free, [ln for ln in lines if ln]))363 return unit, rows364365366def _tier_tables(section: Node) -> dict[str, list[tuple[str, str, list[str]]]]:367 """Tables that follow one `div.models-section` until the next one: {tier: [(row label, free cell, paid cell lines)]}.368 Token-priced models use a `devsite-selector` with one `section` per tier (Standard / Batch / Flex / Priority); per-second /369 per-request models (Veo, Lyria…) have a single bare table, exposed under the pseudo tier `per_<unit>`."""370 tiers: dict[str, list[tuple[str, str, list[str]]]] = {}371 sib = section.next372 while sib is not None:373 if sib.tag == "div" and "models-section" in (sib.attributes.get("class") or ""):374 break375 if sib.tag in ("h2",):376 break377 if sib.tag != "-text":378 sections = [sib] if sib.tag == "section" else sib.css("section")379 for sec in sections:380 h3 = sec.css_first("h3")381 table = sec.css_first("table")382 if not table:383 continue384 unit, rows = _table_rows(table)385 tier = (node_text(h3).lower() if h3 else "standard").split()[0]386 if unit != "tokens":387 tier = f"per_{unit}"388 if rows:389 tiers.setdefault(tier, rows)390 if not sections:391 for table in ([sib] if sib.tag == "table" else sib.css("table")):392 unit, rows = _table_rows(table)393 if rows:394 tiers.setdefault("standard" if unit == "tokens" else f"per_{unit}", rows)395 sib = sib.next396 return tiers397398399def _apply_paid_rows(target: Any, rows: list[tuple[str, str, list[str]]], *, prefix: str, as_dict: bool = False) -> None:400 """Map 'Input price' / 'Output price' / 'Context caching price' rows of a paid-tier table onto a PriceObs (or a dict)."""401 def put(key: str, value: Any) -> None:402 if value is None:403 return404 if as_dict:405 target[key] = value406 elif hasattr(target, key):407 setattr(target, key, value)408 else:409 target.features[key] = value410411 unit_rx = re.compile(r"per (image|second|video|request|query|minute|song)|/\s*(image|second|sec|video|request|query|min)\b", re.IGNORECASE)412 qualifier_rx = re.compile(r"\(([^)]*)\)")413 for label, free, lines in rows:414 if not lines:415 continue416 first = lines[0]417 bare_first = qualifier_rx.sub("", first) # "(text / image / video)" is a modality list, not a unit418 primary = money(bare_first) if not unit_rx.search(bare_first) else None419 if re.match(r"(text )?(input|output) price", label):420 kind = "input" if "input" in label.split(" price")[0] else "output"421 put(f"{kind}_per_mtok", primary)422 if unit_rx.search(bare_first) and re.search(r"per image|/\s*image", bare_first, re.IGNORECASE):423 put("per_image", money(bare_first))424 for ln in lines[1:]:425 if ln.lower().startswith("equivalent"):426 continue427 v = money(qualifier_rx.sub("", ln)) if not unit_rx.search(qualifier_rx.sub("", ln)) else None428 if v is None:429 continue430 qual = qualifier_rx.search(ln)431 if re.search(r">\s*200k|>\s*128k|long context|prompts >", ln, re.IGNORECASE):432 put(f"long_context_{kind}_per_mtok", v)433 elif re.search(r"\bstarting\b", ln, re.IGNORECASE):434 put(f"scheduled_{kind}_per_mtok", v)435 put(f"scheduled_{kind}_effective", re.sub(r"^\$[\d.,]+\s*", "", qualifier_rx.sub("", ln)).strip())436 elif qual:437 mod = re.sub(r"[^a-z]+", "_", qual.group(1).lower()).strip("_")438 put(f"{mod}_{kind}_per_mtok", v)439 if "through" in bare_first.lower() and not as_dict:440 target.features.setdefault("promotional_until", re.sub(r"^\$[\d.,]+\s*(through)?\s*", "", bare_first).strip().rstrip("."))441 if free.lower().startswith("free of charge") and not as_dict:442 target.features["free_tier"] = True443 elif re.match(r"(image|audio|video) input price", label):444 put(f"{label.split()[0]}_input_per_mtok", primary)445 elif label.startswith("context caching price"):446 for ln in lines:447 v = money(ln)448 if v is None:449 continue450 if "storage" in ln.lower():451 put("cache_storage_per_mtok_hour", v)452 break453 if not (as_dict and "cached_input_per_mtok" in target) and not (not as_dict and target.cached_input_per_mtok is not None):454 put("cached_input_per_mtok", v)455 elif re.search(r">\s*200k|prompts >", ln, re.IGNORECASE):456 put("long_context_cached_input_per_mtok", v)457 elif label.startswith("grounding with google search"):458 m = re.search(r"\$\s*([\d.]+)\s*(?:per|/)\s*1,?000", " ".join(lines))459 if m:460 put("search_grounding_per_1k_requests", float(m.group(1)))461462463CONNECTORS = [GoogleConnector]464