File size: 5,814 Bytes
c5f49b9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 | import argparse
import json
import random
import time
import urllib.parse
import urllib.request
from dataclasses import dataclass
from typing import Dict, Iterable, List, Optional, Tuple
WIKIDATA_SPARQL_URL = "https://query.wikidata.org/sparql"
USER_AGENT = "LocalJarvisDatasetBuilder/1.0 (offline training; contact: none)"
@dataclass
class Item:
qid: str
label: str
description: str
type_label: Optional[str] = None
country_label: Optional[str] = None
inception: Optional[str] = None
def sparql(query: str, timeout_s: int = 45) -> Dict:
params = {"format": "json", "query": query}
url = f"{WIKIDATA_SPARQL_URL}?{urllib.parse.urlencode(params)}"
req = urllib.request.Request(
url,
headers={
"Accept": "application/sparql-results+json",
"User-Agent": USER_AGENT,
},
method="GET",
)
with urllib.request.urlopen(req, timeout=timeout_s) as resp:
raw = resp.read().decode("utf-8", errors="ignore")
return json.loads(raw)
def take_value(row: Dict, key: str) -> Optional[str]:
v = row.get(key)
if not v:
return None
if isinstance(v, dict):
return v.get("value")
return None
def query_items_for_instance_of(instance_qid: str, limit: int, seed_offset: int = 0) -> List[Item]:
# We use a randomized ORDER BY with a stable seed-ish offset to get variety
# while staying simple and robust.
query = f"""
SELECT ?item ?itemLabel ?itemDescription ?typeLabel ?countryLabel ?inception WHERE {{
?item wdt:P31/wdt:P279* wd:{instance_qid} .
OPTIONAL {{ ?item wdt:P31 ?type . }}
OPTIONAL {{ ?item wdt:P17 ?country . }}
OPTIONAL {{ ?item wdt:P571 ?inception . }}
SERVICE wikibase:label {{ bd:serviceParam wikibase:language "en". }}
}}
LIMIT {int(limit)}
"""
data = sparql(query)
out: List[Item] = []
for row in data.get("results", {}).get("bindings", []):
item_url = take_value(row, "item") or ""
qid = item_url.rsplit("/", 1)[-1] if "/" in item_url else item_url
label = take_value(row, "itemLabel") or ""
desc = take_value(row, "itemDescription") or ""
if not qid or not label or not desc:
continue
out.append(
Item(
qid=qid,
label=label.strip(),
description=desc.strip(),
type_label=(take_value(row, "typeLabel") or "").strip() or None,
country_label=(take_value(row, "countryLabel") or "").strip() or None,
inception=(take_value(row, "inception") or "").strip() or None,
)
)
# Shuffle locally so different runs mix categories.
rnd = random.Random(1337 + int(seed_offset))
rnd.shuffle(out)
return out
def clean_text(s: str) -> str:
s = " ".join((s or "").replace("\r", " ").replace("\n", " ").split())
return s.strip()
def make_qa(item: Item) -> List[Tuple[str, str]]:
label = clean_text(item.label)
desc = clean_text(item.description)
if not label or not desc:
return []
type_hint = clean_text(item.type_label or "")
country = clean_text(item.country_label or "")
inception = clean_text(item.inception or "")
base = f"{label} is {desc}."
if type_hint and type_hint.lower() not in base.lower():
base = f"{label} is {desc} (type: {type_hint})."
if country:
base = f"{base} It is associated with {country}."
if inception:
# inception is typically an xsd:dateTime; keep it short.
base = f"{base} Notable date: {inception[:10]}."
qa: List[Tuple[str, str]] = []
qa.append((f"What is {label}?", base))
qa.append((f"Give me a short explanation of {label}.", base))
qa.append((f"Be concise: what is {label}?", clean_text(f"{label}: {desc}.")))
qa.append((f"Keep it practical: explain {label}.", base))
return qa
def write_pairs(path: str, pairs: Iterable[Tuple[str, str]]):
with open(path, "w", encoding="utf-8") as f:
for u, a in pairs:
u = clean_text(u)
a = clean_text(a)
if len(u) < 3 or len(a) < 8:
continue
f.write(f"User: {u}\nAssistant: {a}\n\n")
def main():
ap = argparse.ArgumentParser(description="Fetch CC0 Wikidata Q/A pairs for offline Jarvis training.")
ap.add_argument("--out", default="data/web_wikidata_qa.txt")
ap.add_argument("--per-category", type=int, default=250)
ap.add_argument("--sleep", type=float, default=0.7, help="Seconds to sleep between category queries.")
args = ap.parse_args()
# Wikidata is CC0. We generate our own wording, but the underlying facts come from Wikidata.
categories = [
("programming languages", "Q9143"),
("operating systems", "Q9135"),
("countries", "Q6256"),
("cities", "Q515"),
("planets", "Q634"),
("chemical elements", "Q11344"),
("inventions", "Q11426"),
("scientific disciplines", "Q11862829"),
]
all_pairs: List[Tuple[str, str]] = []
for idx, (name, qid) in enumerate(categories):
print(f"Querying {name} ({qid}) ...")
try:
items = query_items_for_instance_of(qid, limit=args.per_category, seed_offset=idx)
except Exception as exc:
print(f"Skipping {name}: {exc}")
continue
added = 0
for it in items:
qa = make_qa(it)
if qa:
all_pairs.extend(qa)
added += len(qa)
print(f" got items={len(items)} pairs_added={added}")
time.sleep(max(0.0, float(args.sleep)))
random.Random(1337).shuffle(all_pairs)
write_pairs(args.out, all_pairs)
print(f"Wrote {len(all_pairs)} pairs to {args.out}")
if __name__ == "__main__":
main()
|