scripts/fetch_wikipedia.py
scripts/fetch_wikipedia.pyBrowse 29 files
2,392 tokens
9,401 bytes
Token encoding: o200k_base
Snapshot 24fd22b
← Back to SKILL.md
1#!/usr/bin/env python32"""Search Wikipedia + Wikidata for an entity (person, company, place, concept).3 4Two free APIs:5 - Wikipedia OpenSearch + REST summary endpoint for narrative bio6 - Wikidata SPARQL endpoint for structured facts (birth, employer, awards, etc.)7 8Both are anonymous-access. Useful for resolving who-is-this-entity questions9and surfacing cross-references that other sources can join against.10"""11from __future__ import annotations12 13import argparse14import csv15import re16import sys17import urllib.parse18from pathlib import Path19 20sys.path.insert(0, str(Path(__file__).parent))21from _http import get_json # noqa: E40222 23WP_OPENSEARCH = "https://en.wikipedia.org/w/api.php"24WP_SUMMARY = "https://en.wikipedia.org/api/rest_v1/page/summary/"25WD_ACTION = "https://www.wikidata.org/w/api.php"26 27COLUMNS = [28 "source",29 "label",30 "description",31 "qid",32 "wikipedia_title",33 "wikipedia_url",34 "wikidata_url",35 "instance_of",36 "country",37 "occupation",38 "employer",39 "date_of_birth",40 "place_of_birth",41 "summary",42]43 44 45def _wp_search(query: str, limit: int) -> list[dict]:46 params = {47 "action": "opensearch",48 "search": query,49 "limit": str(min(limit, 20)),50 "format": "json",51 }52 url = f"{WP_OPENSEARCH}?{urllib.parse.urlencode(params)}"53 data = get_json(url)54 if not isinstance(data, list) or len(data) < 4:55 return []56 titles, descs, urls = data[1], data[2], data[3]57 out = []58 for i, title in enumerate(titles):59 out.append(60 {61 "title": title,62 "description": descs[i] if i < len(descs) else "",63 "url": urls[i] if i < len(urls) else "",64 }65 )66 return out67 68 69def _wp_summary(title: str) -> dict:70 """Pull the REST summary for a title — short bio, image, type."""71 url = f"{WP_SUMMARY}{urllib.parse.quote(title.replace(' ', '_'))}"72 try:73 return get_json(url) # type: ignore[return-value]74 except Exception as e: # noqa: BLE00175 print(f"Wikipedia summary lookup for {title!r} failed: {e}", file=sys.stderr)76 return {}77 78 79def _wd_lookup_by_qid(qid: str) -> dict:80 """Pull common facts for a QID via Wikidata's Action API (no SPARQL).81 82 The Action API is far more lenient on rate-limits than the SPARQL Query83 Service. We get claims as QIDs and then resolve labels in one batch call.84 """85 # Properties of interest. The Action API returns claims as QIDs or86 # typed literals, so the slot mapping is local-only.87 interesting = {88 "P31": "instance_of",89 "P17": "country", # for orgs / places90 "P27": "country", # for individuals (country of citizenship)91 "P106": "occupation",92 "P108": "employer",93 "P569": "date_of_birth",94 "P19": "place_of_birth",95 }96 params = {97 "action": "wbgetentities",98 "ids": qid,99 "props": "claims",100 "format": "json",101 }102 url = f"{WD_ACTION}?{urllib.parse.urlencode(params)}"103 try:104 data = get_json(url)105 except Exception as e: # noqa: BLE001106 print(f"Wikidata wbgetentities for {qid} failed: {e}", file=sys.stderr)107 return {}108 if not isinstance(data, dict):109 return {}110 claims = (data.get("entities", {}).get(qid, {}) or {}).get("claims", {}) or {}111 112 # Collect raw values (QIDs or literals) and remember which slot each113 # came from. Date literals come back as ISO strings; QIDs need a label114 # resolution pass.115 qid_to_slots: dict[str, list[str]] = {}116 facts: dict[str, list[str]] = {}117 for prop_id, slot in interesting.items():118 for claim in claims.get(prop_id, []) or []:119 v = (claim.get("mainsnak", {}) or {}).get("datavalue", {}) or {}120 vtype = v.get("type")121 value = v.get("value")122 if vtype == "wikibase-entityid" and isinstance(value, dict):123 vqid = value.get("id", "")124 if vqid:125 qid_to_slots.setdefault(vqid, [])126 if slot not in qid_to_slots[vqid]:127 qid_to_slots[vqid].append(slot)128 elif vtype == "time" and isinstance(value, dict):129 raw = value.get("time", "") or ""130 # +1955-10-28T00:00:00Z → 1955-10-28131 m = re.search(r"[+-]?(\d{4})-(\d{2})-(\d{2})", raw)132 if m:133 facts.setdefault(slot, []).append(134 f"{m.group(1)}-{m.group(2)}-{m.group(3)}"135 )136 elif vtype == "string":137 facts.setdefault(slot, []).append(str(value))138 139 # Resolve labels for all referenced QIDs in one batch (up to 50 at a time).140 qids = list(qid_to_slots)141 for i in range(0, len(qids), 50):142 batch = qids[i : i + 50]143 params = {144 "action": "wbgetentities",145 "ids": "|".join(batch),146 "props": "labels",147 "languages": "en",148 "format": "json",149 }150 url = f"{WD_ACTION}?{urllib.parse.urlencode(params)}"151 try:152 data = get_json(url)153 except Exception as e: # noqa: BLE001154 print(f"Wikidata label batch failed: {e}", file=sys.stderr)155 continue156 if not isinstance(data, dict):157 continue158 ents = data.get("entities", {}) or {}159 for vqid, ent in ents.items():160 label = (ent.get("labels", {}).get("en", {}) or {}).get("value", "") or vqid161 for slot in qid_to_slots.get(vqid, []):162 facts.setdefault(slot, []).append(label)163 164 # Deduplicate per slot, preserving order.165 deduped: dict[str, list[str]] = {}166 for slot, vals in facts.items():167 seen = set()168 out = []169 for v in vals:170 if v in seen:171 continue172 seen.add(v)173 out.append(v)174 deduped[slot] = out175 return deduped176 177 178def _wd_qid_for_title(title: str) -> str:179 """Get the Wikidata QID associated with a Wikipedia article title."""180 params = {181 "action": "query",182 "format": "json",183 "prop": "pageprops",184 "ppprop": "wikibase_item",185 "titles": title,186 "redirects": 1,187 }188 url = f"{WP_OPENSEARCH}?{urllib.parse.urlencode(params)}"189 try:190 data = get_json(url)191 except Exception: # noqa: BLE001192 return ""193 if not isinstance(data, dict):194 return ""195 pages = data.get("query", {}).get("pages", {}) or {}196 for page in pages.values():197 qid = (page.get("pageprops") or {}).get("wikibase_item", "")198 if qid:199 return qid200 return ""201 202 203def fetch(query: str, limit: int, no_wikidata: bool, out_path: str) -> int:204 hits = _wp_search(query, limit)205 rows: list[dict[str, str]] = []206 for hit in hits[:limit]:207 title = hit.get("title", "")208 if not title:209 continue210 summary = _wp_summary(title)211 qid = _wd_qid_for_title(title) if not no_wikidata else ""212 facts: dict = {}213 if qid:214 facts = _wd_lookup_by_qid(qid)215 rows.append(216 {217 "source": "wikipedia+wikidata" if qid else "wikipedia",218 "label": title,219 "description": (summary.get("description") or hit.get("description") or "").strip(),220 "qid": qid,221 "wikipedia_title": title,222 "wikipedia_url": hit.get("url", ""),223 "wikidata_url": f"https://www.wikidata.org/wiki/{qid}" if qid else "",224 "instance_of": "; ".join(facts.get("instance_of", [])),225 "country": "; ".join(facts.get("country", [])),226 "occupation": "; ".join(facts.get("occupation", [])),227 "employer": "; ".join(facts.get("employer", [])),228 "date_of_birth": "; ".join(facts.get("date_of_birth", []))[:10] if facts.get("date_of_birth") else "",229 "place_of_birth": "; ".join(facts.get("place_of_birth", [])),230 "summary": (summary.get("extract") or "").replace("\n", " ")[:1000],231 }232 )233 234 Path(out_path).parent.mkdir(parents=True, exist_ok=True)235 with open(out_path, "w", newline="", encoding="utf-8") as fh:236 w = csv.DictWriter(fh, fieldnames=COLUMNS)237 w.writeheader()238 w.writerows(rows)239 if not rows:240 print(241 f"Wikipedia: 0 articles for query={query!r}. "242 "Private individuals not notable enough for a Wikipedia article "243 "won't appear here (the bar is real).",244 file=sys.stderr,245 )246 return len(rows)247 248 249def main() -> int:250 p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)251 p.add_argument("--query", required=True, help="Entity name (person, company, place, concept)")252 p.add_argument("--limit", type=int, default=5)253 p.add_argument(254 "--no-wikidata",255 action="store_true",256 help="Skip the Wikidata SPARQL enrichment (faster, less detail)",257 )258 p.add_argument("--out", required=True)259 a = p.parse_args()260 n = fetch(query=a.query, limit=a.limit, no_wikidata=a.no_wikidata, out_path=a.out)261 print(f"Wrote {n} Wikipedia/Wikidata rows to {a.out}")262 return 0263 264 265if __name__ == "__main__":266 raise SystemExit(main())267