Utilities¶

def path_or_url(s: str) -> Path | str:
    """Converts a string into a Path, preserving http(s)://..."""
    if s.startswith("http") and "://" in str(s):
        return s
    else:
        return Path(s)

def get_files_from_artifact(
    manifest: Path | tuple[Path, Path, Graph], artifact: Node
) -> list[Path | str] | Generator[Path]:
    """Returns an iterable (list or generator) of Path objects for files within an artifact literal.

    This function will correctly interpret artifacts such as 'file.ttl', '*.ttl', '**/*.trig' etc.
    """
    manifest_path, manifest_root, manifest_graph = get_manifest_paths_and_graph(
        manifest
    )

    if str(artifact).startswith("http") and "://" in str(artifact):
        return [str(artifact)]
    elif isinstance(artifact, Literal):
        if "*" not in str(artifact):
            return [manifest_root / path_or_url(str(artifact))]
        else:
            artifact_str = str(artifact)
            glob_marker_location = artifact_str.find("*")
            glob_parts = [
                artifact_str[:glob_marker_location],
                artifact_str[glob_marker_location:],
            ]
            return Path(manifest_root / path_or_url(glob_parts[0])).rglob(glob_parts[1])
    elif isinstance(artifact, BNode):
        contentLocation = manifest_graph.value(
            subject=artifact, predicate=SDO.contentLocation
        )
        if str(contentLocation).startswith("http") and "://" in str(contentLocation):
            return [str(contentLocation)]
        else:
            return [manifest_root / str(contentLocation)]
    else:
        raise TypeError(f"Unsupported artifact type: {type(artifact)}")

def get_identifier_from_file(file: Path) -> list[URIRef]:
    """Returns a list if RDFLib graph identifier (URIRefs) from a triples or quads file
    for all KNOWN_ENTITY_CLASSES objects"""
    if file.name.endswith(".ttl"):
        g = Graph().parse(file)
        for entity_class in KNOWN_ENTITY_CLASSES:
            v = g.value(predicate=RDF.type, object=entity_class)
            if v is not None:
                return [v]
    elif file.name.endswith(".trig"):
        gs = []
        d = Dataset()
        d.parse(file, format="trig")
        for g in d.graphs():
            gs.append(g.identifier)
        return gs
    else:
        return []

def get_manifest_paths_and_graph(
    manifest: Path | tuple[Path, Path, Graph],
) -> (Path, Graph):
    """Reads either a Manifest file from a Path, or a Manifest file from a Path and its root directory,
    a Path, and the Manifest as a deserialized Graph and returns the Manifest Path, its root dir as a Path
    and its content as a Graph"""

    if isinstance(manifest, Path):
        manifest_path = manifest
        manifest_root = Path(manifest).parent.resolve()
        manifest_graph = kgm.validate(manifest)
    else:  # (Path, Path, Graph)
        manifest_path = manifest[0]
        manifest_root = manifest[1]
        manifest_graph = manifest[2]

    return manifest_path, manifest_root, manifest_graph

def get_artifact_main_entity_iri(
    artifact: Path,
    manifest: Path | tuple[Path, Path, Graph],
    artifact_graph: Graph = None,
    cc: URIRef = None,
    atype: URIRef = None,
) -> URIRef:
    """Gets the IRI of the instance of the Main Entity in an artifact where the class of the Main Entity is either
    one of KNOWN_ENTITY_CLASSES or supplied in the Manifest either as a class to search for - by using schema:additionalType
     - or by directly indicating the IRI of the main Entity with schema:mainEntity"""

    # load the manifest
    # get Main Entity directly from Manifest mainEntity indicated
    # load the artifact graph
    # check artifact graph load
    # get Main Entity class using specified atype
    # get Main Entity class via Manifest additionalType indicated
    # get Main Entity class via known profiles' classes
    # get Main Entity classes from KNOWN_ENTITY_CLASSES static list
    # build the query to look for the Main Entity in the artifact
    # check artifact query return
    # query the artifact for the Main Entity
    # check the Main Entity query return
    # return

    known_entity_classes = []

    # load the manifest
    manifest_path, manifest_root, manifest_graph = get_manifest_paths_and_graph(
        manifest
    )

    # get Main Entity directly from Manifest mainEntity indicated
    import os

    artifact_path_abs = absolutise_path(artifact, manifest_root)
    artifact_path_rel = os.path.relpath(
        artifact,
        start=os.path.dirname(manifest_path)
    )
    artifact_file = artifact.name
    # artifact_path_rel = Path(str(artifact_path_abs).replace(os.path.commonpath([artifact.parent, manifest_root]), "")) / artifact_file
    q = ("""
        PREFIX prof: <http://www.w3.org/ns/dx/prof/>
        PREFIX schema: <https://schema.org/>

        SELECT ?iri
        WHERE {
            VALUES ?cl {
                "{artifact_path_abs}"
                "{artifact_path_rel}"
                "{artifact_file}"
            }
            ?resource prof:hasArtifact ?bn .
            ?bn schema:contentLocation ?cl .            
            ?bn schema:mainEntity ?iri .
        }
        """.replace("{artifact_path_abs}", str(artifact_path_abs))
         .replace("{artifact_path_rel}", str(artifact_path_rel))
         .replace("{artifact_file}", str(artifact_file)))

    for r in manifest_graph.query(q):
        return URIRef(r["iri"])

    # load the artifact graph
    g = artifact_graph if artifact_graph is not None else load_graph(artifact_path_abs)

    # check artifact graph load
    if not isinstance(g, Graph):
        raise ValueError(f"Could not load a graph of the artifact at {artifact_path_abs}")

    # get Main Entity class using specified atype
    if atype is not None:
        known_entity_classes.append(str(atype))

    # get Main Entity class via Manifest additionalType indicated
    if len(known_entity_classes) < 1:
        q = ("""
            PREFIX prof: <http://www.w3.org/ns/dx/prof/>
            PREFIX schema: <https://schema.org/>

            SELECT DISTINCT ?at
            WHERE {
                VALUES ?art {
                    "{artifact_path_abs}"
                    "{artifact_path_rel}"
                    "{artifact_file}"
                }
                ?resource 
                    prof:hasArtifact ?art ;        
                    schema:additionalType ?at ;
                .
            }
            """.replace("{artifact_path_abs}", str(artifact_path_abs))
             .replace("{artifact_path_rel}", str(artifact_path_rel))
             .replace("{artifact_file}", str(artifact_file)))

        for r in manifest_graph.query(q):
            known_entity_classes.append(r["at"])

    # get Main Entity class via known profiles' classes
    if len(known_entity_classes) < 1:
        if cc is not None:
            if cc in KNOWN_PROFILES.keys():
                for m_e_c in KNOWN_PROFILES[cc]["main_entity_classes"]:
                    known_entity_classes.append(str(m_e_c))

    # get Main Entity classes from KNOWN_ENTITY_CLASSES static list
    if len(known_entity_classes) < 1:
        known_entity_classes = [str(x) for x in KNOWN_ENTITY_CLASSES]

    # build the query to look for the Main Entity in the artifact
    known_entity_classes_str = f"<{'>\n                <'.join(known_entity_classes)}>"
    q = f"""
        SELECT ?me
        WHERE {{
            VALUES ?t {{
                {known_entity_classes_str.strip()}
            }}
            ?me a ?t .
        }}
        """
    mes = []

    # query the artifact for the Main Entity
    for r in query(g, q, return_format="python", return_bindings_only=True):
        if r.get("me"):
            mes.append(r["me"])

    # check the Main Entity query return
    if len(mes) != 1:
        if len(mes) > 1:
            raise ValueError(
                f"The artifact at {artifact_path_abs} has more than one Main Entity: {', '.join(mes)} "
                f"based on the class {cc if cc is not None else '(none given)'}. There must only be one."
            )
        else:
            raise ValueError(
                f"The artifact at {artifact_path_abs} has no recognizable Main Entity, "
                f"based on the classes {cc if cc is not None else '(none given)'}. There must be one."
            )

    # return
    return URIRef(mes[0])

def compare_version_indicators(first: dict, second: dict) -> VersionIndicatorComparison:
    """Compares Modified Date, Version IRI & Version info for each and returns latest"""

    """Even weighted aggregate score for each version indicator"""
    first_score = 0
    second_score = 0
    has_modified_date_comparison = first.get("modified_date") and second.get(
        "modified_date"
    )
    has_version_iri_comparison = first.get("version_iri") and second.get("version_iri")
    has_version_info_comparison = first.get("version_info") and second.get(
        "version_info"
    )

    if (
        not has_modified_date_comparison
        and not has_version_iri_comparison
        and not has_version_info_comparison
    ):
        return VersionIndicatorComparison.CantCalculate

    if has_modified_date_comparison:
        first_date = (
            first["modified_date"].date()
            if isinstance(first["modified_date"], datetime.datetime)
            else first["modified_date"]
        )
        if first_date > second["modified_date"]:
            first_score += 1
        elif first_date == second["modified_date"]:
            pass
        else:
            second_score += 1

    if has_version_iri_comparison:
        if str(first["version_iri"]) > str(second["version_iri"]):
            first_score += 1
        elif str(first["version_iri"]) == str(second["version_iri"]):
            pass
        else:
            second_score += 1

    if has_version_info_comparison:
        if str(first["version_info"]) > str(second["version_info"]):
            first_score += 1
        elif str(first["version_info"]) == str(second["version_info"]):
            pass
        else:
            second_score += 1

    # TODO: add test for file_size, Git version etc.

    if first_score > second_score:
        return VersionIndicatorComparison.First
    elif second_score == first_score:
        return VersionIndicatorComparison.Neither
    else:
        return VersionIndicatorComparison.Second

def which_is_more_recent(
    version_indicators: dict,
    sparql_endpoint: str = None,
    http_client: httpx.Client | None = None,
) -> VersionIndicatorComparison:
    """Tests to see if the given artifact is more recent than a previously stored copy of its content"""

    remote = get_version_indicators_sparql(
        version_indicators["main_entity"], sparql_endpoint, http_client
    )

    return compare_version_indicators(version_indicators, remote)

def denormalise_artifacts(manifest: Path | tuple[Path, Path, Graph]) -> dict:
    """Extracts all the artifacts from a Manifest.

    Returns a dict of:

    Artifact path,
    Main Entity,
    Conformance Claims
    Date Modified
    Version IRI
    Version Info
    Role
    Version Indicators"""
    artifacts_info = {}

    manifest_path, manifest_root, manifest_graph = get_manifest_paths_and_graph(
        manifest
    )

    # for each artifact, get what we can directly from the Manifest
    q = """
        PREFIX dcterms: <http://purl.org/dc/terms/>
        PREFIX mrr: <https://prez.dev/ManifestResourceRoles/>
        PREFIX owl: <http://www.w3.org/2002/07/owl#>
        PREFIX prez: <https://prez.dev/>
        PREFIX prof: <http://www.w3.org/ns/dx/prof/>
        PREFIX schema: <https://schema.org/>

        SELECT ?a ?me ?cc ?atype ?sync ?dm ?vi ?v ?r
        WHERE {
            # if the Resource has a Blank Node artifact, it must provide the Main Entity IRI
            {
                ?x 
                    prof:hasArtifact ?bn ;
                    prof:hasRole ?r ;
                .

                ?bn 
                    schema:mainEntity ?me ;
                    schema:contentLocation ?a ;
                .

                OPTIONAL {
                    ?bn dcterms:conformsTo ?cc_local .
                }

                OPTIONAL {
                    ?x dcterms:conformsTo ?cc_resource .
                }

                OPTIONAL {
                    ?bn schema:additionalType ?atype_local .
                }

                OPTIONAL {
                    ?x schema:additionalType ?atype_resource .
                }

                OPTIONAL {
                    ?bn prez:sync ?sync_local .
                }

                OPTIONAL {
                    ?x prez:sync ?sync_resource .
                }                

                OPTIONAL {
                    ?bn schema:dateModified ?dm .
                } 

                OPTIONAL {
                    ?bn owl:versionIRI ?vi .
                }

                OPTIONAL {
                    ?bn owl:versionInfo|schema:version ?v .
                }     

                BIND(COALESCE(?cc_local, ?cc_resource) AS ?cc)

                BIND(COALESCE(?atype_local, ?atype_resource) AS ?atype)

                BIND(COALESCE(?sync_local, ?sync_resource) AS ?sync)

                FILTER isBLANK(?bn)
            }
            UNION 
            {
                ?x 
                    prof:hasArtifact ?a ;
                    prof:hasRole ?r ;
                .

                OPTIONAL {
                    ?x dcterms:conformsTo ?cc .
                }

                OPTIONAL {
                    ?x schema:additionalType ?atype .
                }

                OPTIONAL {
                    ?x prez:sync ?sync .
                } 

                FILTER isLITERAL(?a)
            }
        }
        """

    for r in query(
        manifest_graph, q, return_format="python", return_bindings_only=True
    ):
        artifact = path_or_url(r["a"])
        files = get_files_from_artifact(
            (manifest_path, manifest_root, manifest_graph), Literal(artifact)
        )

        for file in files:
            me = URIRef(r["me"]) if r.get("me") is not None else None
            role = URIRef(r["r"])
            dm = r["dm"] if r.get("dm") is not None else None
            vi = r["vi"] if r.get("vi") is not None else None
            v = r["v"] if r.get("v") is not None else None
            cc = URIRef(r["cc"]) if r.get("cc") is not None else None
            atype = URIRef(r["atype"]) if r.get("atype") is not None else None
            if r.get("sync") is not None:
                sync = False if r["sync"] == "false" else True
            else:
                sync = True

            artifacts_info[file] = {
                "main_entity": me,
                "role": role,
                "date_modified": dm,
                "version_iri": vi,
                "version_info": v,
                "file_size": None,
                "conformance_claim": cc,
                "additional_type": atype,
                "sync": sync,
            }

    # get Version Indicators info only for Resources with certain Roles
    for k, v in artifacts_info.items():
        if v["role"] in [MRR.CatalogueData, MRR.ResourceData]:
            get_version_indicators_local(
                (manifest_path, manifest_root, manifest_graph), k, v
            )

    return artifacts_info

def store_remote_artifact_locally(
    manifest: Path | tuple[Path, Path, Graph],
    sparql_endpoint: str,
    graph_id: str,
    http_client: httpx.Client | None = None,
) -> Graph:
    """Writes a remote graph to a local file and registers that file as a Resource in the given Manifest.

    Only the Resource Role ResourceData is supported."""
    manifest_path, manifest_root, manifest_graph = get_manifest_paths_and_graph(
        manifest
    )
    q = """
        CONSTRUCT {
            ?s ?p ?o
        }
        WHERE {
            GRAPH <xxx> {
                ?s ?p ?o
            }
        }
        """.replace("xxx", graph_id)
    r = query(sparql_endpoint, q, http_client=http_client, return_format="python")
    artifact_path = str(artifact_file_name_from_graph_id(graph_id))
    r.serialize(destination=manifest_root / artifact_path, format="longturtle")

    new_manifest_graph = Graph()
    new_manifest_graph += manifest_graph

    for m in new_manifest_graph.subjects(RDF.type, PREZ.Manifest):
        new_r = BNode()
        for r in new_manifest_graph.objects(m, PROF.hasResource):
            if (r, PROF.hasRole, MRR.ResourceData) in new_manifest_graph:
                new_r = r

        a = BNode()
        new_manifest_graph.add(
            (
                a,
                SDO.contentLocation,
                Literal(artifact_path),  # relative to manifest_root
            )
        )
        new_manifest_graph.add((a, SDO.mainEntity, URIRef(graph_id)))
        new_manifest_graph.add((new_r, PROF.hasArtifact, a))
        new_manifest_graph.add(
            (new_r, PROF.hasRole, MRR.ResourceData)  # only one supported for now
        )
        new_manifest_graph.add((m, PROF.hasResource, new_r))

    return new_manifest_graph

def get_background_graph(manifest: Path | tuple[Path, Path, Graph]) -> Graph:
    """Returns the contents of all Manifest resources with role *CatalogueAndResourceLabels as a graph"""
    background_graph = Graph()

    # can't use get_manifest_paths_and_graph here as it uses validate
    manifest_path = manifest
    manifest_root = Path(manifest).parent.resolve()
    manifest_graph = load_graph(manifest)

    for resource in manifest_graph.objects(None, PROF.hasResource):
        for role in manifest_graph.objects(resource, PROF.hasRole):
            # The data files & background - must be processed after Catalogue
            if role in [
                MRR.CompleteCatalogueAndResourceLabels,
                MRR.IncompleteCatalogueAndResourceLabels,
            ]:
                for artifact in manifest_graph.objects(resource, PROF.hasArtifact):
                    for file in get_files_from_artifact(
                        (manifest_path, manifest_root, manifest_graph), artifact
                    ):
                        if not file.is_file():
                            raise ValueError(
                                f"The artifact {file} in Manifest {manifest} is not a file"
                            )

                        if str(file.name).endswith(".ttl"):
                            background_graph += load_graph(file)

    return background_graph