ontocast.tool.facts_validation.literal_repair

LLM-free parse-time repairs on rendered facts graphs.

Every rewrite here either retypes a literal the schema grounds, resolves an unambiguous near-miss or code, or collapses a degenerate encoding — no repair invents a value.

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

Functions:

dedupe_literal_variants(graph, fact_namespaces=None)

Collapse duplicate literals differing only in language tag or datatype.

The renderer emits the same value inconsistently across chunks — "X"@en in one unit, "X"^^xsd:string in another, a plain "X" in a third — and after aggregation one (subject, predicate) carries all three as distinct RDF terms. One survives per lexical form: the language-tagged form (each distinct language kept — those are distinct assertions), else the plain form, else the xsd:string form. Reified provenance follows the survivor.

Parameters:

Name	Type	Description	Default
graph	RDFGraph	Aggregated facts graph, mutated in place.	required
fact_namespaces	Sequence[str] | None	When set, only subjects under these namespaces are touched.	None

Returns:

Name	Type	Description
One	list[GraphRepairRecord]

Source code in ontocast/tool/facts_validation/literal_repair.py

def dedupe_literal_variants(
    graph: RDFGraph,
    fact_namespaces: Sequence[str] | None = None,
) -> list[GraphRepairRecord]:
    """Collapse duplicate literals differing only in language tag or datatype.

    The renderer emits the same value inconsistently across chunks —
    ``"X"@en`` in one unit, ``"X"^^xsd:string`` in another, a plain ``"X"``
    in a third — and after aggregation one ``(subject, predicate)`` carries
    all three as distinct RDF terms. One survives per lexical form: the
    language-tagged form (each distinct language kept — those are distinct
    assertions), else the plain form, else the ``xsd:string`` form. Reified
    provenance follows the survivor.

    Args:
        graph: Aggregated facts graph, mutated in place.
        fact_namespaces: When set, only subjects under these namespaces are
            touched.

    Returns:
        One :class:`GraphRepairRecord` per removed literal variant.
    """
    from ontocast.onto.rdfgraph import retarget_reifiers
    from ontocast.tool.facts_validation.terms import _in_fact_scope

    namespaces = [ns for ns in (fact_namespaces or []) if ns]
    groups: dict[tuple[URIRef, URIRef, str], list[Literal]] = {}
    for subject, predicate, obj in graph:
        if not isinstance(subject, URIRef) or not isinstance(predicate, URIRef):
            continue
        if not isinstance(obj, Literal):
            continue
        if namespaces and not _in_fact_scope(subject, namespaces):
            continue
        groups.setdefault((subject, predicate, str(obj)), []).append(obj)

    records: list[GraphRepairRecord] = []
    replacements: dict[tuple[Node, Node, Node], tuple[Node, Node, Node]] = {}
    for (subject, predicate, _lexical), literals in sorted(
        groups.items(), key=lambda item: (str(item[0][0]), str(item[0][1]), item[0][2])
    ):
        if len(literals) < 2:
            continue
        tagged = sorted(
            (lit for lit in literals if lit.language is not None),
            key=lambda lit: str(lit.language),
        )
        plain = [
            lit for lit in literals if lit.language is None and lit.datatype is None
        ]
        typed = [
            lit
            for lit in literals
            if lit.language is None and lit.datatype == XSD.string
        ]
        if tagged:
            keep = tagged[0]
            drop = plain + typed
        elif plain:
            keep = plain[0]
            drop = typed
        else:
            # Distinct datatypes beyond xsd:string are not variants of one
            # value; leave them to the datatype-aware repairs.
            continue
        for variant in drop:
            graph.remove((subject, predicate, variant))
            replacements[(subject, predicate, variant)] = (subject, predicate, keep)
            records.append(
                GraphRepairRecord(
                    kind=FactsGateRepairKind.LITERAL_VARIANT_PRUNED,
                    source=f"{subject} {predicate} {variant.n3()}",
                    target=keep.n3(),
                )
            )
    if replacements:
        retarget_reifiers(graph, replacements)
        logger.info(
            "Collapsed %d literal variant(s) differing only in language tag "
            "or datatype",
            len(records),
        )
    return records

normalize_literals_against_schema(graph, ontology_graph)

Retype literals whose predicate declares a compatible rdfs:range.

Fixes the qudt:numericValue 230 vs "230"^^xsd:decimal drift at parse time, and the same drift for the date-like datatypes: when the schema declares a range in :data:_RETYPABLE_RANGE_DATATYPES and the lexical form parses as that datatype, the literal is rewritten with it.

A literal is only retyped from an untyped, xsd:string, or numeric source -- a string range must never be able to clobber a correctly typed value -- and language-tagged literals are left alone, since they are rdf:langString and retyping would discard the tag.

Returns:

Type	Description
int	Number of retyped literals.

Source code in ontocast/tool/facts_validation/literal_repair.py

def normalize_literals_against_schema(
    graph: RDFGraph, ontology_graph: RDFGraph | None
) -> int:
    """Retype literals whose predicate declares a compatible ``rdfs:range``.

    Fixes the ``qudt:numericValue 230`` vs ``"230"^^xsd:decimal`` drift at parse
    time, and the same drift for the date-like datatypes: when the schema
    declares a range in :data:`_RETYPABLE_RANGE_DATATYPES` and the lexical form
    parses as that datatype, the literal is rewritten with it.

    A literal is only retyped from an untyped, ``xsd:string``, or numeric source
    -- a string range must never be able to clobber a correctly typed value --
    and language-tagged literals are left alone, since they are
    ``rdf:langString`` and retyping would discard the tag.

    Returns:
        Number of retyped literals.
    """
    if ontology_graph is None:
        return 0
    declared_ranges: dict[URIRef, URIRef] = {}
    for predicate, range_iri in ontology_graph.subject_objects(RDFS.range):
        if (
            isinstance(predicate, URIRef)
            and isinstance(range_iri, URIRef)
            and range_iri in _RETYPABLE_RANGE_DATATYPES
        ):
            declared_ranges[predicate] = range_iri

    if not declared_ranges:
        return 0

    replacements: list[tuple[tuple, tuple]] = []
    for subject, predicate, obj in graph:
        if not isinstance(obj, Literal) or not isinstance(predicate, URIRef):
            continue
        target_datatype = declared_ranges.get(predicate)
        if target_datatype is None or obj.datatype == target_datatype:
            continue
        if obj.language is not None:
            continue
        numeric_target = target_datatype in _NUMERIC_RANGE_DATATYPES
        source_admissible = obj.datatype is None or obj.datatype == XSD.string
        if numeric_target:
            # Keep the pre-existing numeric->numeric promotion (integer to
            # decimal, say), which a source-side "untyped or string" rule alone
            # would silently drop.
            source_admissible = (
                source_admissible or obj.datatype in _NUMERIC_RANGE_DATATYPES
            )
        if not source_admissible:
            continue
        lexical = str(obj).strip()
        gregorian = _GREGORIAN_RANGE_PATTERNS.get(target_datatype)
        if numeric_target:
            parses = canonical_number(lexical) is not None
        elif gregorian is not None:
            parses = gregorian.match(lexical) is not None
        else:
            parses = _literal_parses_as(lexical, target_datatype)
        if not parses:
            continue
        replacements.append(
            (
                (subject, predicate, obj),
                (subject, predicate, Literal(lexical, datatype=target_datatype)),
            )
        )

    for old, new in replacements:
        graph.remove(old)
        graph.add(new)
    return len(replacements)

promote_degenerate_bounds(graph, *, numeric_value_property, lower_bound_property, upper_bound_property, inclusive_flag_properties=())

Rewrite equal lower/upper bounds into a single scalar value, in place.

A node whose lower and upper bounds carry the same canonical numeric value encodes an exact scalar as a fake range. The rewrite fires only when the encoding is unambiguous: exactly one literal per bound property, equal canonical values, no existing scalar on the node, and no exclusive-bound flag (an exclusive equal bound denotes an empty interval — malformed, and left for findings). Property IRIs are injected by the caller from configuration; nothing is hardcoded.

Returns:

Type	Description
int	Number of nodes rewritten.

Source code in ontocast/tool/facts_validation/literal_repair.py

def promote_degenerate_bounds(
    graph: RDFGraph,
    *,
    numeric_value_property: str,
    lower_bound_property: str,
    upper_bound_property: str,
    inclusive_flag_properties: Sequence[str] = (),
) -> int:
    """Rewrite equal lower/upper bounds into a single scalar value, in place.

    A node whose lower and upper bounds carry the same canonical numeric value
    encodes an exact scalar as a fake range. The rewrite fires only when the
    encoding is unambiguous: exactly one literal per bound property, equal
    canonical values, no existing scalar on the node, and no exclusive-bound
    flag (an exclusive equal bound denotes an empty interval — malformed, and
    left for findings). Property IRIs are injected by the caller from
    configuration; nothing is hardcoded.

    Returns:
        Number of nodes rewritten.
    """
    lower_ref = URIRef(lower_bound_property)
    upper_ref = URIRef(upper_bound_property)
    value_ref = URIRef(numeric_value_property)
    flag_refs = [URIRef(term) for term in inclusive_flag_properties]
    promoted = 0
    for subject in sorted(set(graph.subjects(lower_ref, None)), key=str):
        lowers = [obj for obj in graph.objects(subject, lower_ref)]
        uppers = [obj for obj in graph.objects(subject, upper_ref)]
        if len(lowers) != 1 or len(uppers) != 1:
            continue
        if not isinstance(lowers[0], Literal) or not isinstance(uppers[0], Literal):
            continue
        if (subject, value_ref, None) in graph:
            continue
        low = canonical_number(str(lowers[0]).strip())
        high = canonical_number(str(uppers[0]).strip())
        if low is None or low != high:
            continue
        if any(
            str(flag_value).strip().lower() == "false"
            for flag_ref in flag_refs
            for flag_value in graph.objects(subject, flag_ref)
        ):
            continue
        graph.remove((subject, lower_ref, lowers[0]))
        graph.remove((subject, upper_ref, uppers[0]))
        for flag_ref in flag_refs:
            for flag_value in list(graph.objects(subject, flag_ref)):
                graph.remove((subject, flag_ref, flag_value))
        graph.add((subject, value_ref, Literal(low, datatype=XSD.decimal)))
        promoted += 1
    if promoted:
        logger.info(
            "Promoted %d degenerate bound pair(s) to <%s>",
            promoted,
            numeric_value_property,
        )
    return promoted

promote_degenerate_bounds_from_vocabulary(graph, ontology_graph, vocabulary)

Run :func:promote_degenerate_bounds with properties from configuration.

Active only when the quantity vocabulary names all three roles — numeric_value, lower_bound, upper_bound (roles containing inclusive supply the optional bound flags). The default vocabulary carries no bound roles, so this is off unless a deployment configures its range encoding.

Source code in ontocast/tool/facts_validation/literal_repair.py

def promote_degenerate_bounds_from_vocabulary(
    graph: RDFGraph,
    ontology_graph: RDFGraph | None,
    vocabulary: dict[str, str] | None,
) -> int:
    """Run :func:`promote_degenerate_bounds` with properties from configuration.

    Active only when the quantity vocabulary names all three roles —
    ``numeric_value``, ``lower_bound``, ``upper_bound`` (roles containing
    ``inclusive`` supply the optional bound flags). The default vocabulary
    carries no bound roles, so this is off unless a deployment configures its
    range encoding.
    """
    vocabulary = vocabulary or {}
    numeric_terms = expand_vocabulary_terms(
        {"numeric_value": vocabulary.get("numeric_value", "")}, graph, ontology_graph
    )
    lower_terms = expand_vocabulary_terms(
        {"lower_bound": vocabulary.get("lower_bound", "")}, graph, ontology_graph
    )
    upper_terms = expand_vocabulary_terms(
        {"upper_bound": vocabulary.get("upper_bound", "")}, graph, ontology_graph
    )
    inclusive_terms = expand_vocabulary_terms(
        _vocabulary_role_subset(vocabulary, "inclusive"), graph, ontology_graph
    )
    if len(numeric_terms) != 1 or len(lower_terms) != 1 or len(upper_terms) != 1:
        return 0
    return promote_degenerate_bounds(
        graph,
        numeric_value_property=next(iter(numeric_terms)),
        lower_bound_property=next(iter(lower_terms)),
        upper_bound_property=next(iter(upper_terms)),
        inclusive_flag_properties=sorted(inclusive_terms),
    )

repair_compact_iri_literals(graph, ontology_context_graph, known_terms)

Coerce plain-string compact IRIs on IRI-valued positions into IRIs.

A JSON-LD bare string ("qudt:unit": "unit:NanoM") parses as a literal, so the node loses its link and reads as missing the property. A literal is rewritten only when every condition holds: it is plain (no language tag, no datatype or xsd:string); the predicate is not rdf:type (that is :func:repair_literal_type_objects); the lexical form is a compact IRI whose prefix the graph or the known-prefix context binds; and either the schema says the predicate takes an IRI or the expanded IRI is a known term. Prose such as "time: 10 minutes" never matches the compact-IRI form.

Parameters:

Name	Type	Description	Default
graph	RDFGraph	Facts graph, repaired in place.	required
ontology_context_graph	RDFGraph	Schema that says which predicates take IRIs.	required
known_terms	set[str]	IRIs the catalog and fallback vocabulary declare; covers predicates whose range the vendored vocabulary leaves undeclared.	required

Returns:

Type	Description
tuple[int, list[GraphRepairRecord]]	Tuple of (number of rewritten triples, applied-repair records).

Source code in ontocast/tool/facts_validation/literal_repair.py

def repair_compact_iri_literals(
    graph: RDFGraph,
    ontology_context_graph: RDFGraph,
    known_terms: set[str],
) -> tuple[int, list[GraphRepairRecord]]:
    """Coerce plain-string compact IRIs on IRI-valued positions into IRIs.

    A JSON-LD bare string (``"qudt:unit": "unit:NanoM"``) parses as a literal,
    so the node loses its link and reads as missing the property. A literal is
    rewritten only when every condition holds: it is plain (no language tag,
    no datatype or ``xsd:string``); the predicate is not ``rdf:type`` (that is
    :func:`repair_literal_type_objects`); the lexical form is a compact IRI
    whose prefix the graph or the known-prefix context binds; and either the
    schema says the predicate takes an IRI or the expanded IRI is a known term.
    Prose such as ``"time: 10 minutes"`` never matches the compact-IRI form.

    Args:
        graph: Facts graph, repaired in place.
        ontology_context_graph: Schema that says which predicates take IRIs.
        known_terms: IRIs the catalog and fallback vocabulary declare; covers
            predicates whose range the vendored vocabulary leaves undeclared.

    Returns:
        Tuple of (number of rewritten triples, applied-repair records).
    """
    prefix_map = dict(RDFGraph.get_known_prefixes() or {})
    prefix_map.update(
        {prefix: str(namespace) for prefix, namespace in graph.namespaces() if prefix}
    )
    replacements: list[tuple[tuple[Node, Node, Node], URIRef, str]] = []
    expects_iri: dict[URIRef, bool] = {}
    for subject, predicate, obj in graph:
        if predicate == RDF.type or not isinstance(obj, Literal):
            continue
        if obj.language or obj.datatype not in (None, XSD.string):
            continue
        lexical = str(obj).strip()
        match = _COMPACT_IRI.match(lexical)
        if match is None:
            continue
        namespace = prefix_map.get(match.group(1))
        if not namespace:
            continue
        expanded = f"{namespace}{match.group(2)}"
        if expanded not in known_terms:
            if not isinstance(predicate, URIRef):
                continue
            if predicate not in expects_iri:
                expects_iri[predicate] = _predicate_expects_iri_object(
                    predicate, ontology_context_graph
                )[0]
            if not expects_iri[predicate]:
                continue
        replacements.append(((subject, predicate, obj), URIRef(expanded), lexical))

    applied: list[GraphRepairRecord] = []
    for (subject, predicate, obj), iri, lexical in replacements:
        graph.remove((subject, predicate, obj))
        graph.add((subject, predicate, iri))
        applied.append(
            GraphRepairRecord(
                kind=FactsUnitFindingKind.COMPACT_IRI_LITERAL,
                source=lexical,
                target=str(iri),
            )
        )
        logger.info("Repaired compact-IRI literal %r -> <%s>", lexical, iri)
    if applied:
        record_active_count("repair/compact_iri_literal", len(applied))
    return len(applied), applied

repair_literal_type_objects(graph)

Coerce literal rdf:type objects into IRIs.

The renderer sometimes emits a "prefix:Class"^^xsd:string instead of a prefix:Class (JSON-LD bare-string type values parse the same way). A literal-typed node is invisible to SPARQL class queries, reasoning, and the aggregator's URI minting/entity matching, all of which guard on isinstance(obj, URIRef). Absolute IRIs and compact IRIs bound in the graph are rewritten deterministically; unresolvable forms become MANDATORY findings.

Returns:

Type	Description
int	Tuple of (number of rewritten triples, unresolved findings,
list[FactsUnitFinding]	applied-repair records).

Source code in ontocast/tool/facts_validation/literal_repair.py

def repair_literal_type_objects(
    graph: RDFGraph,
) -> tuple[int, list[FactsUnitFinding], list[GraphRepairRecord]]:
    """Coerce literal ``rdf:type`` objects into IRIs.

    The renderer sometimes emits ``a "prefix:Class"^^xsd:string`` instead of
    ``a prefix:Class`` (JSON-LD bare-string type values parse the same way).
    A literal-typed node is invisible to SPARQL class queries, reasoning, and
    the aggregator's URI minting/entity matching, all of which guard on
    ``isinstance(obj, URIRef)``. Absolute IRIs and compact IRIs bound in the
    graph are rewritten deterministically; unresolvable forms become MANDATORY
    findings.

    Returns:
        Tuple of (number of rewritten triples, unresolved findings,
        applied-repair records).
    """
    prefix_map = {
        prefix: str(namespace) for prefix, namespace in graph.namespaces() if prefix
    }
    rewritten = 0
    findings: list[FactsUnitFinding] = []
    applied: list[GraphRepairRecord] = []
    for subject, predicate, obj in list(graph.triples((None, RDF.type, None))):
        if not isinstance(obj, Literal):
            continue
        lexical = str(obj).strip()
        resolved = _resolve_type_literal(lexical, prefix_map)
        if resolved is not None:
            graph.remove((subject, predicate, obj))
            graph.add((subject, RDF.type, URIRef(resolved)))
            rewritten += 1
            applied.append(
                GraphRepairRecord(
                    kind=FactsUnitFindingKind.LITERAL_TYPE_OBJECT,
                    source=lexical,
                    target=resolved,
                )
            )
            logger.info(
                "Repaired literal rdf:type object %r -> <%s>", lexical, resolved
            )
            continue
        findings.append(
            FactsUnitFinding(
                kind=FactsUnitFindingKind.LITERAL_TYPE_OBJECT,
                message=(
                    f"rdf:type object '{lexical}' is a string literal, not an "
                    "IRI; assert the type as a catalog class IRI "
                    "(`a prefix:Class`), never as a quoted string."
                ),
                subject=str(subject),
                value=lexical,
            )
        )
    return rewritten, findings, applied

repair_property_aliases(graph, ontology_graph, *, min_ratio=0.95, exempt_terms=None, full_catalog_terms=None)

Rewrite near-miss predicates in catalog namespaces; report ambiguity.

A predicate whose namespace belongs to the ontology context but which is not itself a catalog term is a near-miss (qqval:lowerBound for qqval:hasLowerBound). The rewrite is deterministic and fires only when exactly one candidate qualifies: its name tokens contain the alias's, the alias's contain its, or the two spell the same once case and separators are folded. Several qualifying candidates are a tie that the SequenceMatcher ratio may break -- the best-scoring one wins when it clears min_ratio and nothing ties it. Similarity alone never licenses a rewrite: a high ratio also joins names that differ by exactly one token, and rewriting across that gap substitutes one property for another. Anything else becomes a mandatory finding carrying the top suggestions.

Only namespaces the catalog declares terms in are eligible (see :func:collect_declared_namespaces); exempt_terms (expanded fallback vocabulary) are never treated as near-misses, and neither is a predicate in full_catalog_terms: a term the whole catalog declares is real even when this unit's snapshot did not retrieve it, and the snapshot's look-alike is then a different property, not the intended one. Such a predicate is left untouched here; whether it is reported is the term checks' business.

Returns:

Type	Description
int	Tuple of (number of rewritten triples, unresolved findings,
list[FactsUnitFinding]	applied-repair records).

Source code in ontocast/tool/facts_validation/literal_repair.py

def repair_property_aliases(
    graph: RDFGraph,
    ontology_graph: RDFGraph | None,
    *,
    min_ratio: float = 0.95,
    exempt_terms: set[str] | None = None,
    full_catalog_terms: set[str] | None = None,
) -> tuple[int, list[FactsUnitFinding], list[GraphRepairRecord]]:
    """Rewrite near-miss predicates in catalog namespaces; report ambiguity.

    A predicate whose namespace belongs to the ontology context but which is
    not itself a catalog term is a near-miss (``qqval:lowerBound`` for
    ``qqval:hasLowerBound``). The rewrite is deterministic and fires only when
    exactly one candidate *qualifies*: its name tokens contain the alias's,
    the alias's contain its, or the two spell the same once case and
    separators are folded. Several qualifying candidates are a tie that the
    ``SequenceMatcher`` ratio may break -- the best-scoring one wins when it
    clears ``min_ratio`` and nothing ties it. Similarity alone never licenses
    a rewrite: a high ratio also joins names that differ by exactly one token,
    and rewriting across that gap substitutes one property for another.
    Anything else becomes a mandatory finding carrying the top suggestions.

    Only namespaces the catalog *declares* terms in are eligible (see
    :func:`collect_declared_namespaces`); ``exempt_terms`` (expanded fallback
    vocabulary) are never treated as near-misses, and neither is a predicate
    in ``full_catalog_terms``: a term the whole catalog declares is real even
    when this unit's snapshot did not retrieve it, and the snapshot's
    look-alike is then a different property, not the intended one. Such a
    predicate is left untouched here; whether it is reported is the term
    checks' business.

    Returns:
        Tuple of (number of rewritten triples, unresolved findings,
        applied-repair records).
    """
    catalog_terms = collect_catalog_terms(ontology_graph)
    if not catalog_terms:
        return 0, [], []
    declared_namespaces = collect_declared_namespaces(ontology_graph)
    exempt = exempt_terms or set()
    known_elsewhere = full_catalog_terms or set()

    findings: list[FactsUnitFinding] = []
    applied: list[GraphRepairRecord] = []
    rewritten = 0
    predicates = {
        predicate
        for predicate in graph.predicates()
        if isinstance(predicate, URIRef)
        and str(predicate) not in catalog_terms
        and str(predicate) not in exempt
        and str(predicate) not in known_elsewhere
        and _namespace_of(str(predicate)) in declared_namespaces
    }
    for alias in sorted(predicates, key=str):
        candidates = _alias_candidates(
            alias, graph, catalog_terms, ontology_graph=ontology_graph
        )
        target = _select_alias_target(str(alias), candidates, min_ratio)
        if target is not None:
            replacement = URIRef(target)
            alias_triples = 0
            for subject, predicate, obj in list(graph.triples((None, alias, None))):
                graph.remove((subject, predicate, obj))
                graph.add((subject, replacement, obj))
                rewritten += 1
                alias_triples += 1
            applied.append(
                GraphRepairRecord(
                    kind=FactsUnitFindingKind.PROPERTY_ALIAS,
                    source=str(alias),
                    target=str(replacement),
                    triple_count=alias_triples,
                )
            )
            logger.info("Repaired property alias %s -> %s", alias, replacement)
            continue
        findings.append(
            FactsUnitFinding(
                kind=FactsUnitFindingKind.PROPERTY_ALIAS,
                message=(
                    f"Predicate <{alias}> is not defined in its ontology; "
                    "replace it with the correct catalog property."
                ),
                predicate=str(alias),
                suggestions=candidates,
            )
        )
    return rewritten, findings, applied

resolve_code_literals(graph, ontology_graph, code_predicates=())

Link nodes to the catalog individual whose code they already carry.

A renderer that reads 4-15 days often annotates the value node with the code it saw — qudt:ucumCode "d" — instead of the object property that points at the individual — qudt:unit unit:DAY. The graph is well-formed, so no range check fires, but every query reading the object property gets an unbound result. The code came from the text and the individual is in the catalog, so the link is recoverable without asking the model again.

Fully schema-driven, no vocabulary compiled in: the connecting property is whichever object property the ontology context declares with a range the resolved individual is typed as, and a domain the subject satisfies. If the schema offers several such properties, or none, nothing is added.

Parameters:

Name	Type	Description	Default
graph	RDFGraph	Rendered facts graph, repaired in place.	required
ontology_graph	RDFGraph | None	Merged ontology context, read-only.	required
code_predicates	Sequence[str]	Predicates carrying machine-resolvable codes.	()

Returns:

Type	Description
tuple[int, list[GraphRepairRecord]]	Tuple of (number of added triples, applied-repair records).

Source code in ontocast/tool/facts_validation/literal_repair.py

def resolve_code_literals(
    graph: RDFGraph,
    ontology_graph: RDFGraph | None,
    code_predicates: Sequence[str] = (),
) -> tuple[int, list[GraphRepairRecord]]:
    """Link nodes to the catalog individual whose code they already carry.

    A renderer that reads ``4-15 days`` often annotates the value node with the
    code it saw — ``qudt:ucumCode "d"`` — instead of the object property that
    points at the individual — ``qudt:unit unit:DAY``. The graph is well-formed,
    so no range check fires, but every query reading the object property gets
    an unbound result. The code came from the text and the individual is in the
    catalog, so the link is recoverable without asking the model again.

    Fully schema-driven, no vocabulary compiled in: the connecting property is
    whichever object property the ontology context declares with a range the
    resolved individual is typed as, and a domain the subject satisfies. If the
    schema offers several such properties, or none, nothing is added.

    Args:
        graph: Rendered facts graph, repaired in place.
        ontology_graph: Merged ontology context, read-only.
        code_predicates: Predicates carrying machine-resolvable codes.

    Returns:
        Tuple of (number of added triples, applied-repair records).
    """
    if ontology_graph is None or not code_predicates:
        return 0, []
    code_terms = [URIRef(predicate) for predicate in code_predicates]
    # Only the code predicates themselves resolve here: a label match is a
    # different, much weaker signal and belongs to the shapes-driven pass.
    code_index: dict[str, set[str]] = {}
    for predicate in code_terms:
        for subject, value in ontology_graph.subject_objects(predicate):
            if isinstance(subject, URIRef) and isinstance(value, Literal):
                text = str(value).strip()
                if text:
                    code_index.setdefault(text, set()).add(str(subject))
    if not code_index:
        return 0, []

    domains = _declared_domains(ontology_graph)
    ranges: dict[URIRef, set[URIRef]] = {}
    for predicate, _, range_iri in ontology_graph.triples((None, RDFS.range, None)):
        if isinstance(predicate, URIRef) and isinstance(range_iri, URIRef):
            ranges.setdefault(predicate, set()).add(range_iri)

    # Superclass closures repeat heavily across literals; memoise per call.
    closures: dict[URIRef, set[URIRef]] = {}

    def closure(class_iri: URIRef) -> set[URIRef]:
        if class_iri not in closures:
            closures[class_iri] = _superclass_closure(class_iri, ontology_graph)
        return closures[class_iri]

    # The usage-evidence scan is a full graph walk; build it lazily, once,
    # only if some literal actually needs the no-declared-range fallback.
    linking_evidence: list[tuple[set[URIRef], set[URIRef], URIRef]] | None = None

    added = 0
    records: list[GraphRepairRecord] = []
    for code_predicate in code_terms:
        for subject, value in list(graph.subject_objects(code_predicate)):
            if not isinstance(subject, URIRef) or not isinstance(value, Literal):
                continue
            resolved = resolve_unique_surface(code_index, str(value))
            if resolved is None:
                continue
            resolved_types: set[URIRef] = set()
            for type_iri in ontology_graph.objects(resolved, RDF.type):
                if isinstance(type_iri, URIRef):
                    resolved_types |= closure(type_iri)
            if not resolved_types:
                continue
            subject_types: set[URIRef] = set()
            for type_iri in graph.objects(subject, RDF.type):
                if isinstance(type_iri, URIRef):
                    subject_types |= closure(type_iri)

            candidates = [
                predicate
                for predicate, range_set in ranges.items()
                if range_set & resolved_types
                and (
                    predicate not in domains
                    or not domains[predicate]
                    or domains[predicate] & subject_types
                )
            ]
            if not candidates:
                # Vendored vocabulary projections often declare individuals and
                # their codes but no rdfs:range (the shipped QUDT unit subset is
                # one). Fall back to how the graph already links this kind of
                # subject to this kind of individual -- the same
                # induce-from-usage move the functional-predicate harvest makes.
                if linking_evidence is None:
                    linking_evidence = _collect_linking_evidence(
                        graph, ontology_graph, closure
                    )
                candidates = _observed_linking_predicates(
                    linking_evidence, subject_types, resolved_types
                )
            # Already linked, ambiguous, or unsupported by the schema.
            candidates = [
                predicate
                for predicate in candidates
                if (subject, predicate, None) not in graph
            ]
            if len(candidates) != 1:
                continue
            predicate = candidates[0]
            graph.add((subject, predicate, resolved))
            added += 1
            records.append(
                GraphRepairRecord(
                    kind=FactsGateRepairKind.CODE_RESOLVED,
                    source=f"{code_predicate} {value.n3()}",
                    target=f"{predicate} {resolved}",
                )
            )
            logger.info(
                "Resolved code %s on <%s> to <%s %s>",
                value.n3(),
                subject,
                predicate,
                resolved,
            )
    return added, records