Still Charger API

Overview

gkc.still_charger is the packet assembly and source charging module.

It owns:

• Building curation packet scaffolds from JSON Entity Profiles.
• Charging packet entities from source systems (currently Wikidata via mash).
• Producing packet-level conformance payloads by orchestrating fermenter primitives.
• Maintaining packet metadata integrity sealing after charge-time metadata mutation.

It does not own validation semantics. Fermenter owns validation outcomes and notice semantics.

Contract Shape

Curation packets follow a strict three-section top-level shape:

• metadata
• data
• conformance

Source metadata for charged packets is recorded in metadata.profiles[*] and includes:

• source_qid
• lastrevid
• pulled_at

After source metadata injection, still_charger reseals metadata.integrity.metadata_digest.

Quick Start

from gkc.still_charger import create_curation_packet, charge_packet_from_wikidata_items
from gkc.spirit_safe import set_spirit_safe_source

set_spirit_safe_source(mode="local", local_root="/path/to/SpiritSafe")

packet = create_curation_packet("Q4", operation_mode="single")

qid_map = {entity["id"]: "Q195562" for entity in packet["data"]["entities"]}
charged_packet, notices = charge_packet_from_wikidata_items(packet, qid_map)

print(charged_packet["metadata"]["profiles"][0].get("source_qid"))
print(charged_packet["metadata"]["integrity"]["metadata_digest_algorithm"])
print(len(notices))

Public API

create_curation_packet

Create a packet scaffold from SpiritSafe profile JSON.

from gkc.still_charger import create_curation_packet

packet = create_curation_packet("Q4", operation_mode="single")

Arguments:

Argument	Type	Meaning
profile_id	str	Profile QID or full profile URI
operation_mode	str	single for primary profile only, bulk for profile-graph expansion

Returns:

• Packet scaffold with metadata, data, and empty conformance target surface.

build_curation_packet_from_json_profile

Assemble a packet scaffold from a loaded JSON profile document.

from pathlib import Path
from gkc.still_charger import build_curation_packet_from_json_profile

packet = build_curation_packet_from_json_profile(
    profile_entity="https://datadistillery.wikibase.cloud/entity/Q4",
    json_profile_doc=json_profile_doc,
    source_root=Path("/path/to/SpiritSafe"),
)

Arguments:

Argument	Type	Meaning
profile_entity	str	Profile URI or QID
json_profile_doc	dict	Parsed JSON profile document
source_root	Path | None	Optional SpiritSafe local root for value-list route resolution
source_config	dict | None	Optional source descriptor stored under metadata.source

Returns:

• Uncharged packet scaffold with sealed metadata integrity digest.

charge_packet_from_wikidata_items

Charge a packet from Wikidata entities and emit conformance payloads.

from gkc.still_charger import charge_packet_from_wikidata_items

charged_packet, notices = charge_packet_from_wikidata_items(packet, qid_map)

Arguments:

Argument	Type	Meaning
packet	dict	Packet assembled by build_curation_packet_from_json_profile
qid_map	dict[str, str]	Maps profile URI or profile name_identifier to Wikidata QID
mash_client	Any | None	Optional WikibaseLoader-compatible client

Returns:

Position	Type	Meaning
0	dict	Charged packet
1	list[ConformanceNotice]	Packet notices (currently empty placeholder list for this path)

Charge behavior:

• Resolves linked entities via profile linkage routes.
• Loads primary and linked entity JSON from Wikidata.
• Populates packet data entities with source payloads.
• Injects source provenance fields into metadata.profiles[*].
• Builds conformance payload from fermenter statement evaluators.
• Reseals metadata digest after metadata mutation.

Current runtime note:

• data.entities remains a transitional hybrid surface in current implementation (scaffold slots plus embedded raw entity payload).
• Contract direction in #200 is to eliminate hybrid slot decoration from packet data and keep evaluation semantics in conformance only.

charge_curation_packet (Legacy)

Legacy direct-charge path from caller-provided source values.

from gkc.still_charger import charge_curation_packet

charged_packet, report = charge_curation_packet(packet, source_values)

New workflows should prefer charge_packet_from_wikidata_items.

ChargeIssue and ChargeReport

ChargeIssue captures a non-fatal charging issue.

ChargeReport summarizes charge results:

• entities_charged
• entities_skipped
• issues

Metadata Integrity and Provenance

Metadata digest behavior:

1. Packet scaffolds are sealed at build time.
2. Charged packets inject source provenance into metadata.profiles[*].
3. Metadata is resealed after provenance injection.

This supports packet re-presentation checks where metadata integrity and source revision context must be evaluated together.

Conformance Output Interface

still_charger orchestrates conformance payload construction and delegates atomic statement evaluation to fermenter:

• evaluate_statement_instance
• statement_evaluation_to_record

Ownership split:

• still_charger: packet orchestration, source loading, packet mutation order.
• fermenter: statement-level evaluation semantics and record serialization.

API Reference (mkdocstrings)

ChargeIssue

Represents a non-fatal charging issue.

Source code in gkc/still_charger.py

@dataclass
class ChargeIssue:
    """Represents a non-fatal charging issue."""

    severity: str
    entity_id: str
    field: str
    message: str

ChargeReport

Summary of a charge operation.

Source code in gkc/still_charger.py

@dataclass
class ChargeReport:
    """Summary of a charge operation."""

    entities_charged: int = 0
    entities_skipped: int = 0
    issues: list[ChargeIssue] = field(default_factory=list)

create_curation_packet

Create a curation packet scaffold from SpiritSafe JSON profiles.

This is the canonical packet-assembly entrypoint. It loads the primary profile from SpiritSafe, applies operation-mode expansion policy, and delegates scaffold construction to build_curation_packet_from_json_profile.

Source code in gkc/still_charger.py

def create_curation_packet(
    profile_id: str,
    operation_mode: str = "single",
    load_wikidata_qids: bool = False,
    depth: int = 1,
    manifest: Optional[Any] = None,
) -> dict[str, Any]:
    """Create a curation packet scaffold from SpiritSafe JSON profiles.

    This is the canonical packet-assembly entrypoint. It loads the primary
    profile from SpiritSafe, applies operation-mode expansion policy, and
    delegates scaffold construction to ``build_curation_packet_from_json_profile``.
    """

    del load_wikidata_qids
    del depth
    del manifest

    if operation_mode not in {"single", "bulk"}:
        raise ValueError(
            f"Unsupported operation_mode '{operation_mode}'. Expected 'single' or 'bulk'."
        )

    from gkc.spirit_safe import get_spirit_safe_source, load_profile

    profile_doc = load_profile(profile_id)
    profile_uri, _ = _normalize_entity_uri(str(profile_doc.get("id") or profile_id))

    if operation_mode == "single":
        profile_doc = deepcopy(profile_doc)
        profile_doc.setdefault("metadata", {})["profile_graph"] = []

    source = get_spirit_safe_source()
    source_config: dict[str, Any] = {"mode": source.mode}
    if source.mode == "local" and source.local_root is not None:
        source_config["local_root"] = str(source.local_root)
    else:
        source_config["github_repo"] = source.github_repo
        source_config["github_ref"] = source.github_ref
    packet = build_curation_packet_from_json_profile(
        profile_entity=profile_uri,
        json_profile_doc=profile_doc,
        source_root=source.local_root if source.mode == "local" else None,
        source_config=source_config,
    )
    packet["operation_mode"] = operation_mode
    return packet

build_curation_packet_from_json_profile

Build a curation packet from a JSON Entity Profile document.

This is the first stage of the still_charger pipeline: assembling a blank curation packet scaffold from a pre-loaded JSON profile, ready to be charged with concrete values.

Parameters:

Name	Type	Description	Default
profile_entity	str	Full entity URI (e.g., "https://datadistillery.wikibase.cloud/entity/Q4") or QID (e.g., "Q4")	required
json_profile_doc	dict	The loaded JSON Entity Profile document	required
source_root	Optional[Path]	Optional path root for resolving value list cache files; if provided, item_count will be hydrated from cache JSON	None

Returns:

Type	Description
dict	A dictionary representing the curation packet matching the frozen contract:
dict	{ "packet_id": "pkt-", "operation_mode": "new", "profile_entity": "", "entities": [...], "cross_references": {...}, "value_list_routes": {...}
dict	}

Source code in gkc/still_charger.py

def build_curation_packet_from_json_profile(
    profile_entity: str,
    json_profile_doc: dict,
    *,
    source_root: Optional[Path] = None,
    source_config: Optional[dict[str, Any]] = None,
) -> dict:
    """Build a curation packet from a JSON Entity Profile document.

    This is the first stage of the still_charger pipeline: assembling a blank
    curation packet scaffold from a pre-loaded JSON profile, ready to be charged
    with concrete values.

    Args:
        profile_entity: Full entity URI (e.g., "https://datadistillery.wikibase.cloud/entity/Q4")
                       or QID (e.g., "Q4")
        json_profile_doc: The loaded JSON Entity Profile document
        source_root: Optional path root for resolving value list cache files;
                    if provided, item_count will be hydrated from cache JSON

    Returns:
        A dictionary representing the curation packet matching the frozen contract:
        {
            "packet_id": "pkt-<uuid>",
            "operation_mode": "new",
            "profile_entity": "<full_uri>",
            "entities": [...],
            "cross_references": {...},
            "value_list_routes": {...}
        }
    """
    # Step 1: Normalize profile_entity to full URI
    full_profile_uri, _ = _normalize_entity_uri(profile_entity)

    # Step 2: Load primary + linked profile documents from metadata.profile_graph
    profile_docs = _load_related_profile_documents(full_profile_uri, json_profile_doc)

    # Step 3: Build ordered profiles (primary first, then deterministic linked order)
    ordered_profile_uris = [full_profile_uri] + sorted(
        uri for uri in profile_docs.keys() if uri != full_profile_uri
    )

    data_entities: list[dict[str, Any]] = []
    metadata_profiles: list[dict[str, Any]] = []
    profile_name_by_uri: dict[str, str] = {}
    value_list_routes = _build_value_list_routes(profile_docs, source_root)

    for profile_uri in ordered_profile_uris:
        profile_doc = profile_docs[profile_uri]
        profile_uri = _profile_uri_from_doc(profile_doc, profile_uri)
        profile_name = _profile_name_identifier(profile_doc, profile_uri)
        profile_name_by_uri[profile_uri] = profile_name

        statements = profile_doc.get("statements", [])
        if not isinstance(statements, list):
            statements = []

        normalized_statements = [
            _normalize_statement_scaffold(statement)
            for statement in statements
            if isinstance(statement, dict)
        ]

        statement_slots: dict[str, dict[str, Any]] = {}
        for statement in normalized_statements:
            key_base = _statement_name_identifier(statement)
            key = key_base
            suffix = 2
            while key in statement_slots:
                key = f"{key_base}_{suffix}"
                suffix += 1
            statement_slots[key] = _statement_data_slot(
                statement,
                value_list_routes,
                include_children=True,
            )

        identification = deepcopy(profile_doc.get("identification", {}))
        if not isinstance(identification, dict):
            identification = {}

        labels_slot, descriptions_slot, aliases_slot = _identification_language_slots(
            identification
        )

        metadata_profiles.append(
            {
                "id": profile_uri,
                "name_identifier": profile_name,
                "identification": identification,
                "statements": normalized_statements,
                "metadata": deepcopy(profile_doc.get("metadata", {})),
            }
        )

        # Build Wikibase JSON entity shell for profile-only packets
        wikibase_entity_shell = _build_entity_wikibase_json_from_profile(
            profile_doc, profile_uri
        )

        data_entities.append(
            {
                "profile": profile_name,
                "id": profile_uri,
                "labels": labels_slot,
                "descriptions": descriptions_slot,
                "aliases": aliases_slot,
                "statements": statement_slots,
                "entity": wikibase_entity_shell,
            }
        )

    primary_profile_name = profile_name_by_uri.get(
        full_profile_uri, full_profile_uri.rstrip("/").split("/")[-1]
    )

    # Step 4: Build packet metadata graph and mint metadata
    unified_graph = _build_unified_graph(profile_docs, profile_name_by_uri)
    minted_at = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")

    # Step 5: Generate packet_id
    packet_id = f"pkt-{uuid.uuid4()}"

    metadata = {
        "primary_profile": {
            "name_identifier": primary_profile_name,
            "id": full_profile_uri,
        },
        "profiles": metadata_profiles,
        "graph": unified_graph,
        "mint": {
            "minted_at": minted_at,
            "generator": "gkc.still_charger.build_curation_packet_from_json_profile",
            "gkc_version": gkc.__version__,
        },
        "source": source_config or {},
    }

    metadata_digest = _canonical_json_digest(metadata)
    metadata["integrity"] = {
        "metadata_canonicalization": "json-sort-keys-v1",
        "metadata_digest_algorithm": "sha256",
        "metadata_digest": metadata_digest,
    }

    packet = {
        "packet_id": packet_id,
        "operation_mode": "new",
        "metadata": metadata,
        "data": {
            "entities": data_entities,
        },
    }

    return packet

charge_packet_from_wikidata_items

Charge a curation packet with raw Wikibase entity JSON and conformance evaluation.

This is the primary charging entry point. It first resolves the linked entity graph from primary entity claims, then loads linked entities and populates the data section with raw Wikibase JSON.

Raw entity JSON is stored unmodified in data.entities[].entity. Conformance evaluation (statement-level alignment vs. profile) is stored separately in conformance.statement_evaluations using JSON paths. Labels, descriptions, and aliases are passed through without conformance evaluation.

Parameters:

Name	Type	Description	Default
packet	dict[str, Any]	Curation packet assembled by build_curation_packet_from_json_profile()	required
qid_map	dict[str, str]	Mapping from profile entity URI or profile name_identifier to Wikidata QID (e.g., {"https://datadistillery.wikibase.cloud/entity/Q4": "Q195562"})	required
mash_client	Optional[Any]	Optional WikibaseLoader; if None, uses default	None

Returns:

Type	Description
dict[str, Any]	Tuple of (charged_packet, notices) with:
list[ConformanceNotice]	charged_packet: packet with data.entities populated with raw Wikibase JSON, conformance section with entity_profile_map and statement_evaluations
tuple[dict[str, Any], list[ConformanceNotice]]	notices: list of ConformanceNotice (integration point for fermenter)

Source code in gkc/still_charger.py

def charge_packet_from_wikidata_items(
    packet: dict[str, Any],
    qid_map: dict[str, str],
    *,
    mash_client: Optional[Any] = None,
) -> tuple[dict[str, Any], list[ConformanceNotice]]:
    """Charge a curation packet with raw Wikibase entity JSON and conformance evaluation.

    This is the primary charging entry point. It first resolves the linked
    entity graph from primary entity claims, then loads linked entities and
    populates the data section with raw Wikibase JSON.

    Raw entity JSON is stored unmodified in data.entities[].entity.
    Conformance evaluation (statement-level alignment vs. profile) is stored separately
    in conformance.statement_evaluations using JSON paths.
    Labels, descriptions, and aliases are passed through without conformance evaluation.

    Args:
        packet: Curation packet assembled by build_curation_packet_from_json_profile()
        qid_map: Mapping from profile entity URI or profile name_identifier to
            Wikidata QID (e.g., {"https://datadistillery.wikibase.cloud/entity/Q4": "Q195562"})
        mash_client: Optional WikibaseLoader; if None, uses default

    Returns:
        Tuple of (charged_packet, notices) with:
        - charged_packet: packet with data.entities populated with raw Wikibase JSON,
                         conformance section with entity_profile_map and statement_evaluations
        - notices: list of ConformanceNotice (integration point for fermenter)
    """
    try:
        from gkc.mash import WikibaseLoader
    except ImportError:
        raise RuntimeError(
            "mash module required for Wikidata charging. "
            "Ensure gkc is installed with full dependencies."
        )

    if mash_client is None:
        mash_client = WikibaseLoader()

    # Determine primary entity QID from qid_map
    primary_profile = packet.get("metadata", {}).get("primary_profile", {})
    primary_profile_uri = primary_profile.get("id")
    primary_profile_name = primary_profile.get("name_identifier")

    primary_qid = None
    if isinstance(primary_profile_uri, str):
        primary_qid = qid_map.get(primary_profile_uri)
    if not primary_qid and isinstance(primary_profile_name, str):
        primary_qid = qid_map.get(primary_profile_name)

    if not primary_qid:
        raise ValueError(
            f"Could not resolve primary entity QID from qid_map. "
            f"Expected key for profile URI or name_identifier: "
            f"{primary_profile_uri}, {primary_profile_name}"
        )

    # Resolve linked entity graph from the primary entity.
    entity_profile_map, primary_entity = _resolve_linked_entity_graph_from_primary(
        packet, primary_qid, mash_client
    )

    # Load linked entities and evaluate conformance.
    data_entities, statement_evaluations = _load_and_evaluate_linked_entities(
        packet, entity_profile_map, primary_entity, primary_qid, mash_client
    )

    # Build charged packet while preserving packet-native scaffold slots.
    charged = deepcopy(packet)

    pulled_at = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")

    def _resolve_profile_qid(
        profile_uri: Optional[str], profile_name: Optional[str]
    ) -> Optional[str]:
        resolved_qid: Optional[str] = None
        if isinstance(profile_uri, str):
            resolved_qid = qid_map.get(profile_uri)
        if not resolved_qid and isinstance(profile_name, str):
            resolved_qid = qid_map.get(profile_name)
        if resolved_qid:
            return resolved_qid

        if not isinstance(profile_uri, str):
            return None
        for map_key, mapped_profile in entity_profile_map.items():
            if mapped_profile == profile_uri and _looks_like_qid(map_key):
                return map_key
        return None

    raw_entity_by_qid: dict[str, dict[str, Any]] = {
        entry.get("id"): entry.get("entity")
        for entry in data_entities
        if isinstance(entry, dict)
        and isinstance(entry.get("id"), str)
        and isinstance(entry.get("entity"), dict)
    }

    scaffold_entities = packet_entities(charged)

    profiles_meta = charged.get("metadata", {}).get("profiles", [])
    if isinstance(profiles_meta, list):
        for profile_meta in profiles_meta:
            if not isinstance(profile_meta, dict):
                continue
            profile_uri = profile_meta.get("id")
            profile_name = profile_meta.get("name_identifier")
            entity_qid = _resolve_profile_qid(
                profile_uri if isinstance(profile_uri, str) else None,
                profile_name if isinstance(profile_name, str) else None,
            )
            if not entity_qid:
                continue
            raw_entity = raw_entity_by_qid.get(entity_qid)
            if not isinstance(raw_entity, dict):
                continue

            profile_meta["source_qid"] = entity_qid
            profile_meta["lastrevid"] = raw_entity.get("lastrevid")
            profile_meta["pulled_at"] = pulled_at

    if not scaffold_entities:
        charged["data"] = {"entities": data_entities}
        charged["conformance"] = {
            "entity_profile_map": entity_profile_map,
            "statement_evaluations": statement_evaluations,
        }
        charged["operation_mode"] = "edit"
        _reseal_packet_metadata(charged)
        notices: list[ConformanceNotice] = []
        return charged, notices

    if not isinstance(profiles_meta, list):
        profiles_meta = []

    def _profile_meta_for_entity(
        entity_slot: dict[str, Any],
    ) -> Optional[dict[str, Any]]:
        profile_uri = entity_slot.get("id")
        profile_name = entity_slot.get("profile")
        for profile_meta in profiles_meta:
            if not isinstance(profile_meta, dict):
                continue
            if profile_uri and profile_meta.get("id") == profile_uri:
                return profile_meta
            if profile_name and profile_meta.get("name_identifier") == profile_name:
                return profile_meta
        return None

    for entity_slot in scaffold_entities:
        profile_uri = entity_slot.get("id")
        profile_name = entity_slot.get("profile")

        entity_qid = None
        if isinstance(profile_uri, str):
            entity_qid = qid_map.get(profile_uri)
        if not entity_qid and isinstance(profile_name, str):
            entity_qid = qid_map.get(profile_name)
        if not entity_qid and isinstance(profile_uri, str):
            for map_key, mapped_profile in entity_profile_map.items():
                if mapped_profile == profile_uri and _looks_like_qid(map_key):
                    entity_qid = map_key
                    break

        if not entity_qid:
            continue

        raw_entity = raw_entity_by_qid.get(entity_qid)
        if not isinstance(raw_entity, dict):
            continue

        entity_slot["entity"] = raw_entity
        entity_slot["wikibase_entity"] = raw_entity

        labels_slot = entity_slot.get("labels")
        if isinstance(labels_slot, dict):
            _copy_language_values(labels_slot, raw_entity.get("labels"))

        descriptions_slot = entity_slot.get("descriptions")
        if isinstance(descriptions_slot, dict):
            _copy_language_values(descriptions_slot, raw_entity.get("descriptions"))

        aliases_slot = entity_slot.get("aliases")
        if isinstance(aliases_slot, dict):
            _copy_language_values(aliases_slot, raw_entity.get("aliases"), aliases=True)

        profile_meta = _profile_meta_for_entity(entity_slot)
        statements_slot = entity_slot.get("statements")
        claims = (
            raw_entity.get("claims")
            if isinstance(raw_entity.get("claims"), dict)
            else {}
        )
        if not isinstance(profile_meta, dict) or not isinstance(statements_slot, dict):
            continue

        statement_by_property, _ = _profile_statement_map(profile_meta)
        uri_to_name = _statement_uri_to_name_identifier_map(profile_meta)

        for property_id, statement_uri in statement_by_property.items():
            property_claims = claims.get(property_id)
            if not isinstance(property_claims, list):
                continue

            values = _claims_to_values(property_claims)
            if not values:
                continue

            candidates: list[str] = []
            statement_name = uri_to_name.get(statement_uri)
            if isinstance(statement_name, str) and statement_name:
                candidates.append(statement_name)
            if isinstance(statement_uri, str) and statement_uri:
                candidates.append(statement_uri.rstrip("/").split("/")[-1])
                candidates.append(statement_uri)

            slot_payload = None
            for candidate in candidates:
                payload = statements_slot.get(candidate)
                if isinstance(payload, dict):
                    slot_payload = payload
                    break

            if isinstance(slot_payload, dict):
                slot_payload["data-value"] = values[0]

    # Populate conformance section
    conformance = {
        "entity_profile_map": entity_profile_map,
        "statement_evaluations": statement_evaluations,
    }
    charged["conformance"] = conformance
    charged["operation_mode"] = "edit"
    _reseal_packet_metadata(charged)

    # For now, return empty notices (will be populated by fermenter integration)
    notices: list[ConformanceNotice] = []

    return charged, notices

charge_curation_packet

Fill packet entities with real source values.

Parameters:

Name	Type	Description	Default
packet	dict[str, Any]	Curation packet generated from profile scaffolds.	required
source_values	dict[str, dict[str, Any]]	Mapping keyed by entity ID, profile_entity URI, QID, or profile name. Each value is expected to include a statements mapping and optional metadata (labels/descriptions/aliases).	required
specificationless	bool	When True, allows unknown statements and records warnings instead of blocking charging.	True

Returns:

Type	Description
tuple[dict[str, Any], ChargeReport]	Tuple of (charged_packet, report).

Source code in gkc/still_charger.py

def charge_curation_packet(
    packet: dict[str, Any],
    source_values: dict[str, dict[str, Any]],
    *,
    specificationless: bool = True,
) -> tuple[dict[str, Any], ChargeReport]:
    """Fill packet entities with real source values.

    Args:
        packet: Curation packet generated from profile scaffolds.
        source_values: Mapping keyed by entity ID, profile_entity URI, QID, or profile name.
            Each value is expected to include a ``statements`` mapping and optional metadata
            (labels/descriptions/aliases).
        specificationless: When ``True``, allows unknown statements and records
            warnings instead of blocking charging.

    Returns:
        Tuple of (charged_packet, report).
    """
    charged = deepcopy(packet)
    report = ChargeReport()

    entities = charged.get("entities", [])
    for entity in entities:
        entity_id = str(entity.get("id", ""))
        payload = _entity_payload_for(entity, source_values)
        if not payload:
            report.entities_skipped += 1
            continue

        entity.setdefault("data", {})
        entity_data = entity["data"]
        allowed_statement_ids = _extract_statement_ids(entity)

        payload_statements = payload.get("statements", {})
        if not isinstance(payload_statements, dict):
            report.issues.append(
                ChargeIssue(
                    severity="error",
                    entity_id=entity_id,
                    field="statements",
                    message="Expected 'statements' payload to be a mapping.",
                )
            )
            report.entities_skipped += 1
            continue

        unknown_statement_ids = [
            statement_id
            for statement_id in payload_statements
            if allowed_statement_ids and statement_id not in allowed_statement_ids
        ]

        if unknown_statement_ids and not specificationless:
            report.issues.append(
                ChargeIssue(
                    severity="error",
                    entity_id=entity_id,
                    field="statements",
                    message=(
                        "Unknown statements for profile scaffold: "
                        + ", ".join(sorted(unknown_statement_ids))
                    ),
                )
            )
            report.entities_skipped += 1
            continue

        if unknown_statement_ids and specificationless:
            report.issues.append(
                ChargeIssue(
                    severity="warning",
                    entity_id=entity_id,
                    field="statements",
                    message=(
                        "Specificationless charging accepted unknown statements: "
                        + ", ".join(sorted(unknown_statement_ids))
                    ),
                )
            )

        for key in ("labels", "descriptions", "aliases", "sitelinks"):
            if key in payload:
                entity_data[key] = payload[key]

        entity_data.setdefault("statements", {})
        entity_data["statements"].update(payload_statements)
        report.entities_charged += 1

    return charged, report

See Also

• Fermenter API
• SpiritSafe API
• Curation Packet Contract