Syncer

def sync(
    manifest: Path | tuple[Path, Path, Graph],
    sparql_endpoint: str = None,
    http_client: httpx2.Client = httpx2.Client(),
    update_remote: bool = True,
    update_local: bool = True,
    add_remote: bool = True,
    add_local: bool = True,
    add_to_system_graph: bool = True,
) -> dict:
    """Synchronises the resources described by a Manifest with a SPARQL Endpoint.

    Each resource is compared with its copy in the SPARQL Endpoint if existing using timestamps and version numbers. Based on selected flags, newer local resources are uploaded, newer remote ones are downloaded, and artifacts that only exist on one side are added to the other. The Catalogue and Resources can be registered in the System Graph.

    To get the content sync() would retrieve locally without contacting a SPARQL Endpoint, use get_sync_dataset().

    Args:
        manifest: the KGM manifest describing the local resources
        sparql_endpoint: a SPARQL endpoint URL to sync resources to
        http_client: an httpx2 client to use for making requests
        update_remote: whether to update the remote artifacts with newer local ones
        update_local: whether to update the local artifacts with newer remote ones
        add_remote: whether to add local artifacts to the remote location if missing
        add_local: whether to add remote artifacts locally  from the remote location if missing
        add_to_system_graph: whether to register synchronized resources and their
            catalogue in the target's system graph

    Returns:
        a dictionary of the state of synchronisation, per artifact
    """

    # list all from resources
    # find all matching to resources
    sync_status = {}

    manifest_path, manifest_root, manifest_graph = get_manifest_paths_and_graph(
        manifest
    )

    # For each Artifact in the Manifest
    artifacts = denormalise_artifacts((manifest_path, manifest_root, manifest_graph))
    local_entities = [v["main_entity"] for k, v in artifacts.items()]
    # An ignored artifact identified in the manifest must not be rediscovered
    # remotely and downloaded as a supposedly new local resource.
    for resource in manifest_graph.objects(None, PROF.hasResource):
        if resource_is_ignored(manifest_graph, resource):
            local_entities.extend(manifest_graph.objects(resource, SDO.mainEntity))
            for artifact in manifest_graph.objects(resource, PROF.hasArtifact):
                local_entities.extend(manifest_graph.objects(artifact, SDO.mainEntity))

    cat_iri = None
    cat_artifact_path = None
    for k, v in artifacts.items():
        if v["role"] in [MRR.ResourceData, MRR.CatalogueData]:
            # save cat_iri for later
            if v["role"] in MRR.CatalogueData:
                cat_iri = v["main_entity"]
                cat_artifact_path = absolutise_path(k, manifest_root)

            # See if each is known remotely (via Main Entity Graph IRI)
            known = query(
                sparql_endpoint,
                "ASK {GRAPH <xxx> {?s ?p ?o}}".replace("xxx", str(v["main_entity"])),
                http_client=http_client,
                return_format="python",
                return_bindings_only=True,
            )
            # If not known by graph IRI, just check if it's the catalogue (+ "-catalogue" to IRI)
            if not known and v["role"] == MRR.CatalogueData:
                known = query(
                    sparql_endpoint,
                    "ASK {GRAPH <xxx> {?s ?p ?o}}".replace(
                        "xxx", str(v["main_entity"] + "-catalogue")
                    ),
                    http_client=http_client,
                    return_format="python",
                    return_bindings_only=True,
                )

            # If known, compare it
            if known:
                replace = which_is_more_recent(
                    v,
                    sparql_endpoint,
                    http_client,
                )
                if replace == VersionIndicatorComparison.First:
                    direction = "upload"
                elif replace == VersionIndicatorComparison.Second:
                    direction = "download"
                elif replace == VersionIndicatorComparison.Neither:
                    direction = "same"
                elif VersionIndicatorComparison.CantCalculate:
                    direction = "upload"
            else:  # not known at remote location so forward sync - upload
                direction = "add-remotely"

            sync_status[str(k)] = {
                "main_entity": v["main_entity"],
                "direction": direction,
                "sync": v["sync"],
            }
        if v["role"] in [
            MRR.IncompleteCatalogueAndResourceLabels,
            MRR.CompleteCatalogueAndResourceLabels,
        ]:
            clear(sparql_endpoint, "http://background", http_client)
            upload(
                sparql_endpoint,
                Path(k),
                "http://background",
                False,
                http_client=http_client,
            )

    # Check for things at remote not known in local
    q = """
        PREFIX dcterms: <http://purl.org/dc/terms/>
        PREFIX schema: <https://schema.org/>

        SELECT ?p
        WHERE {
            GRAPH ?g {
                <xxx> schema:hasPart|dcterms:hasPart ?p
            }
        }
        """.replace("xxx", str(cat_iri))

    for x in query(
        sparql_endpoint,
        q,
        http_client=http_client,
        return_format="python",
        return_bindings_only=True,
    ):
        remote_entity = x["p"]
        if URIRef(remote_entity) not in local_entities:
            sync_status[str(remote_entity)] = {
                "main_entity": URIRef(remote_entity),
                "direction": "add-locally",
                "sync": True,
            }

    update_remote_catalogue = False
    for k, v in sync_status.items():
        if v["sync"]:
            if update_remote and v["direction"] == "upload":
                clear(sparql_endpoint, v["main_entity"], http_client)
                upload(
                    sparql_endpoint,
                    Path(k),
                    v["main_entity"],
                    False,
                    http_client=http_client,
                )

            if add_remote and v["direction"] == "add-remotely":
                # no need to clear() as this asset doesn't exist remotely
                upload(
                    sparql_endpoint,
                    Path(k),
                    v["main_entity"],
                    False,
                    http_client=http_client,
                )
                update_remote_catalogue = True

            if add_local and v["direction"] == "add-locally":
                updated_local_manifest = store_remote_artifact_locally(
                    (manifest_path, manifest_root, manifest_graph),
                    sparql_endpoint,
                    v["main_entity"],
                    http_client,
                )

                updated_local_manifest.bind(
                    "mrr", "https://prez.dev/ManifestResourceRoles"
                )
                updated_local_manifest.serialize(
                    destination=manifest_path, format="longturtle"
                )
                cat = load_graph(cat_artifact_path)
                cat.add((cat_iri, SDO.hasPart, URIRef(v["main_entity"])))
                cat.serialize(destination=cat_artifact_path, format="longturtle")

            if update_local and v["direction"] == "download":
                update_local_artifact(
                    (manifest_path, manifest_root, manifest_graph),
                    Path(k),
                    sparql_endpoint,
                    v["main_entity"],
                    http_client,
                )

    if update_remote_catalogue:
        delete(sparql_endpoint, cat_iri, http_client=http_client)
        upload(
            sparql_endpoint,
            cat_artifact_path,
            cat_iri,
            False,
            http_client=http_client,
        )

    if add_to_system_graph:
        upload(
            sparql_endpoint,
            _make_system_graph(artifacts, manifest_root),
            SYSTEM_GRAPH_IRI,
            True,
            http_client=http_client,
        )

    return sync_status

def get_sync_dataset(
    manifest: Path | tuple[Path, Path, Graph],
    system_graph_iri: URIRef = SYSTEM_GRAPH_IRI,
) -> Dataset:
    """Gets the content sync() would upload for a Manifest as an in-memory Dataset. Unlike sync(), nothing is compared with or sent to a SPARQL Endpoint, and no local files are changed.

    Args:
        manifest: the KGM manifest describing the local resources
        system_graph_iri: the IRI of the Named Graph to put the System Graph entries in

    Returns:
        a Dataset containing one Named Graph per synchronised Catalogue and Resource, the background graph and the
        System Graph
    """
    manifest_path, manifest_root, manifest_graph = get_manifest_paths_and_graph(
        manifest
    )
    artifacts = denormalise_artifacts((manifest_path, manifest_root, manifest_graph))

    ds = Dataset()
    for k, v in artifacts.items():
        if v["role"] in [MRR.ResourceData, MRR.CatalogueData] and v["sync"]:
            g = ds.graph(v["main_entity"])
        elif v["role"] in [
            MRR.IncompleteCatalogueAndResourceLabels,
            MRR.CompleteCatalogueAndResourceLabels,
        ]:
            g = ds.graph(URIRef("http://background"))
        else:
            continue
        g += load_graph(absolutise_path(k, manifest_root))

    system_graph = ds.graph(system_graph_iri)
    system_graph += _make_system_graph(artifacts, manifest_root)

    return ds

def make_catalogue(
    manifest: Path | tuple[Path, Path, Graph],
    reuse_cat_iri: bool = False,
    new_cat_iri: URIRef | str = None,
) -> Graph:
    """Makes a catalogue graph for a Manifest

    Args:
        manifest: the KGM manifest to create the catalogue for
        new_cat_iri: the iri of the catalogue, if known
        reuse_cat_iri: whether to reuse the IRI of an existing catalogue if defined in the manifest

    Returns:
        a simple graph of the catalogue
        also writes back to the Manifest and overwrites or creates a catalogue artifact file
    """
    # create the return graph
    # read, add or create the catalogue IRI
    # add in each resource's IRI where role is ResourceData & *CatalogueAndResourceLabels

    if reuse_cat_iri:
        if new_cat_iri is not None:
            raise ValueError(
                "If you select to reuse any existing catalogue 'reuse_cat_iri=True', then new_cat_iri must be None"
            )
    else:
        if new_cat_iri is None:
            raise ValueError(
                "If you select not to reuse any existing catalogue 'reuse_cat_iri=False', then new_cat_iri must be provided"
            )

        if new_cat_iri is not None:
            new_cat_iri = (
                URIRef(new_cat_iri) if isinstance(new_cat_iri, str) else new_cat_iri
            )

    manifest_path, manifest_root, manifest_graph = get_manifest_paths_and_graph(
        manifest
    )
    manifest_graph: Graph

    # create the return graph
    c: Graph = None
    cat_path: Path = None

    # read, add or create the catalogue IRI
    # see if we already have one in the Manifest
    existing_cat = False
    for r in manifest_graph.objects(None, PROF.hasResource):
        if resource_is_ignored(manifest_graph, r):
            continue
        if (r, PROF.hasRole, MRR.CatalogueData) in manifest_graph:
            existing_cat = True
            cat_path = next(manifest_graph.objects(r, PROF.hasArtifact))
            c = load_graph(manifest_root / cat_path)

    if cat_path is None:
        if Path(manifest_root / "catalogue.ttl").is_file():
            cat_path = manifest_root / f"catalogue.{kgm.utils.make_dateModified()}.ttl"
        else:
            cat_path = manifest_root / "catalogue.ttl"

    # see if we can find an IRI for a catalogue, if we have one loaded in c
    if c is not None:
        cat_iri_existing = c.value(predicate=RDF.type, object=SDO.DataCatalog)
        if cat_iri_existing is None:
            c.value(predicate=RDF.type, object=DCAT.Catalog)

    # check we are able to reuse
    if reuse_cat_iri:
        if c is None:
            raise ValueError(
                "You nave selected to reuse an existing catalogue "
                "but no resource with CatalogueData was found in the manifest"
            )

        if not cat_iri_existing:
            raise ValueError(
                f"You nave selected to reuse an existing catalogue but an IRI for one "
                f"could not be found in file {cat_path}"
            )
        cat_iri = cat_iri_existing
    else:  # we are not reusing so scrub existing catalogue IRI and ensure at least cat declaration is present
        if c is not None:
            for p, o in c.predicate_objects(cat_iri_existing):
                c.remove((cat_iri_existing, p, o))
                c.add((new_cat_iri, p, o))
            for s, p in c.subject_predicates(cat_iri_existing):
                c.remove((s, p, cat_iri_existing))
                c.add((s, p, new_cat_iri))

        if c is None:
            c = Graph()
        cat_iri = new_cat_iri

        c.add((cat_iri, RDF.type, SDO.DataCatalog))

    # add/update basic catalogue facts
    c.remove((cat_iri, SDO.dateModified, None))
    c.add((cat_iri, SDO.dateModified, kgm.utils.make_dateModified()))

    # add in each resource's IRI
    artifacts = denormalise_artifacts((manifest_path, manifest_root, manifest_graph))

    for k, v in artifacts.items():
        if v["role"] == MRR.ResourceData:
            c.add((cat_iri, SDO.hasPart, v["main_entity"]))

    # replace catalogue entry in manifest
    if existing_cat:
        pass
    else:
        """
        [
            prof:hasArtifact "catalogue.ttl" ;
            prof:hasRole <https://prez.dev/ManifestResourceRoles/CatalogueData> ;
        ] ;
        """
        import os

        cat_path_rel = os.path.relpath(cat_path, start=os.path.dirname(manifest_path))

        r = BNode()
        manifest_graph.add((r, PROF.hasArtifact, Literal(cat_path_rel)))
        manifest_graph.add((r, PROF.hasRole, MRR.CatalogueData))
        for s in manifest_graph.subjects(PROF.hasResource, None):
            manifest_graph.add((s, PROF.hasResource, r))

    with open(manifest_path, "w") as f:
        f.write(manifest_graph.serialize(format="longturtle"))

    # save catalogue artifact
    c.serialize(format="longturtle", destination=cat_path)

    return c