"""Map a pinned Wikidata entity export into a canonical snapshot. This is the third connector. It does not use the JSON, NDJSON, or Markdown loaders, and it does not import UI code. Candidates go through ``project_records`` and ``publish_directory``. """ from __future__ import annotations import json from pathlib import Path from typing import Any from content_explorer.ingest import project_records, publish_directory, snapshot_hash CONNECTOR_ID = "wikidata-export" CONNECTOR_VERSION = "0.1.0" SOURCE_NAMESPACE = "https://www.wikidata.org/wiki/" MAPPING_VERSION = "1" PROPERTY_PREDICATES = {"P287": "designed-by"} FORMAT = "wikidata-entity-export" def _digest(payload: bytes) -> str: import hashlib return hashlib.sha256(payload).hexdigest() def _en(block: dict | None) -> str: entry = (block or {}).get("en") or {} if isinstance(entry, dict): return str(entry.get("value") or "").strip() return "" def _target_qid(statement: dict) -> str | None: snak = statement.get("mainsnak") or {} if snak.get("snaktype") != "value": return None value = (snak.get("datavalue") or {}).get("value") if not isinstance(value, dict): return None qid = value.get("id") if isinstance(qid, str) and len(qid) > 1 and qid[0] == "Q" and qid[1:].isdigit(): return qid return None def load_export(path: Path) -> tuple[dict[str, Any], str]: """Read one pinned Wikidata export. The digest is of the file bytes.""" raw = path.read_bytes() document = json.loads(raw.decode("utf-8")) if not isinstance(document, dict) or document.get("format") != FORMAT: raise ValueError(f"{path} is not a {FORMAT}") if not isinstance(document.get("entities"), dict): raise ValueError(f"{path} has no Wikidata entity map") return document, _digest(raw) def map_candidates(document: dict[str, Any]) -> list[dict[str, Any]]: """Turn Wikidata items into candidate records. Native Q-ids become aliases.""" entities = document["entities"] bindings = document.get("kindByQid") or {} if not isinstance(bindings, dict) or not bindings: raise ValueError("kindByQid is required") located = { qid: f"{binding['kind']}:{binding['key']}" for qid, binding in bindings.items() } records: list[dict[str, Any]] = [] for qid, binding in bindings.items(): item = entities.get(qid) if not isinstance(item, dict): raise ValueError(f"{qid} is not in the export entities") title = _en(item.get("labels")) description = _en(item.get("descriptions")) if not title or not description: raise ValueError(f"{qid} needs an English label and description") aliases = [qid, f"{SOURCE_NAMESPACE}{qid}"] for alias in (item.get("aliases") or {}).get("en") or []: value = alias.get("value") if isinstance(alias, dict) else str(alias) value = str(value).strip() if value and value not in aliases: aliases.append(value) relations = [] claims = item.get("claims") or {} for prop, predicate in PROPERTY_PREDICATES.items(): for statement in claims.get(prop) or []: target = _target_qid(statement) if not target: continue relations.append({ "predicate": predicate, "to": located.get(target) or f"missing:{target.lower()}", }) records.append({ "kind": binding["kind"], "key": binding["key"], "title": title, "tldr": description, "summary": description, "tags": [binding["kind"]], "aliases": aliases, "relations": relations, }) return records def _source(document: dict[str, Any]) -> dict[str, str]: raw = document.get("source") or {} source = { "id": str(raw.get("id") or ""), "uri": str(raw.get("uri") or ""), "title": str(raw.get("title") or ""), "license": str(raw.get("license") or ""), "attribution": str(raw.get("attribution") or ""), } retrieved = str(document.get("retrievedAt") or "") if retrieved: source["retrievedAt"] = retrieved return source def _annotate(report: dict[str, Any], digest: str) -> None: report["connectorVersion"] = f"{CONNECTOR_ID}/{CONNECTOR_VERSION}" report["sourceDigest"] = digest report["inputDigest"] = digest report["connector"] = { "id": CONNECTOR_ID, "version": CONNECTOR_VERSION, "sourceNamespace": SOURCE_NAMESPACE, "capabilities": ["map", "validate", "commit"], "licensePolicy": "CC0-1.0, CC-BY-4.0, MIT", "mappingVersion": MAPPING_VERSION, "refreshMode": "pinned-export-replay", "sourcePriority": 1, "lastCheckpoint": digest, } report["counts"] = { "candidates": report.get("candidates", 0), "accepted": report.get("accepted", 0), "tombstoned": len(report.get("tombstoned") or []), "quarantined": len(report.get("quarantined") or []), "dangling": len(report.get("dangling") or []), "conflicts": len(report.get("conflicts") or []), } def _browsable_columns(manifest: dict[str, Any] | None) -> None: if not manifest: return for view in (manifest.get("kindViews") or {}).values(): view["columns"] = ["name", "tldr", "tags"] view["facets"] = ["tags"] def run(export_path: Path, destination: Path | None = None) -> dict[str, Any]: """Map ``export_path`` and publish only when the existing gate allows it.""" document, digest = load_export(export_path) dataspace = document["dataspace"] result = project_records( map_candidates(document), dataspace_id=dataspace["id"], title=dataspace["title"], description=dataspace["description"], source=_source(document), kinds=list(document["kinds"]), observed_at=str(document["retrievedAt"]), source_digest=digest, relation_labels=document.get("relationLabels") or {}, fact_policy=document.get("factPolicy") or "show-all", acyclic_predicates=list(document.get("acyclicPredicates") or []), ) _annotate(result["report"], digest) _browsable_columns(result.get("manifest")) if result["report"]["published"] and result.get("snapshot") is not None: result["report"]["contentHash"] = snapshot_hash(result["snapshot"]) if destination is not None and result["report"]["published"]: publish_directory(destination, result) return result