sparcur developer guide

function sparc-time-friendly () {

    local UTC_OFFSET_START
    local TIME_START_NO_OFFSET

    # gnu coreutils gdate needed for osx support/freebsd
    # gdate on darwin only has millisecond resolution?
    # this also won't work on freebsd without gnu coreutils
    iso8601millis="+%FT%T,%6N"  # FIXME do we _really_ need millis!? yemaybe? concurrent startups?
    utcoffset="+%:z"
    # I really hope the utc offset doesn't change between start & end
    # but laptops and airplains do exist, so it could
    # also how utterly annoying that the date separator and the
    # negative utc offset share the same symbol ... talk about
    # an annoying design flaw that is going to haunt humanity
    # with double the number of calls to date for # ... as
    # long as anoyone is writing code to deal with time
    TIME_START_NO_OFFSET=$(date ${iso8601millis} || gdate ${iso8601millis})
    UTC_OFFSET_START=$(date ${utcoffset} || gdate ${utcoffset})
    local TIME_START="${TIME_START_NO_OFFSET}${UTC_OFFSET_START}"  # XXX unused

    local TIME_START_NO_OFFSET_FS_OK=${TIME_START_NO_OFFSET//:/}
    local UTC_OFFSET_START_FS_OK=${UTC_OFFSET_START//:/}
    local TIME_START_FRIENDLY=${TIME_START_NO_OFFSET_FS_OK}${UTC_OFFSET_START_FS_OK}
    # So. iso8601 guidance on what to do about subsecond time and the utc offset in the compact
    # representation is not entirely clear, however I _think_ that %FT%T%H%M%S,%6N%z is ok but
    # the -/+ must stay between the timezone and the rest, so we will have to grab tz by itself
    local TIME_START_SAFE=${TIME_START_NO_OFFSET_FS_OK//-/}${UTC_OFFSET_START_FS_OK}  # XXX unused
    mv "$(mktemp --directory sparcur-all-XXXXXX)" "${TIME_START_FRIENDLY}" || \
        { CODE=$?; echo 'mv failed'; return $CODE; }
    echo "${TIME_START_FRIENDLY}"

}

function sparc-get-all-remote-data () {
    # NOTE not quite all the remote data, the google sheets
    # don't have caching functionality yet

    # parse args
    local POSITIONAL=()
    while [[ $# -gt 0 ]]
    do
    key="$1"
    case $key in # (ref:(((((((sigh)
        --project-id)         local PROJECT_ID="${2}"; shift; shift ;;
        --symlink-objects-to) local SYMLINK_OBJECTS_TO="${2}"; shift; shift ;;
        --log-path)           local LOG_PATH="${2}"; shift; shift ;;
        --parent-path)        local PARENT_PATH="${2}"; shift; shift ;;
        --only-filesystem)    local ONLY_FILESYSTEM="ONLY_FS"; shift ;;
        -h|--help)            echo "${HELP}"; return ;;
        *)                    POSITIONAL+=("$1"); shift ;;
    esac
    done

    # Why, you might be asking, are we declaring a local project path here without assignment?
    # Well. Let me tell you. Because local is a command with an exist status. So it _always_
    # returns zero. So if you need to check the output of the command running in a subshell
    # that you are assigning to a local variable _ALWAYS_ set local separately first.
    # Yes, shellcheck does warn about this. See also https://superuser.com/a/1103711
    local PROJECT_PATH

    if [[ -z "${PARENT_PATH}" ]]; then
        local PARENT_PATH
        set -o pipefail
        PARENT_PATH=$(sparc-time-friendly) || {
        CODE=$?;
        echo "Creating "'${PARENT_PATH}'" failed!"
        set +o pipefail
        return $CODE;
        }
        set +o pipefail
    fi

    local LOG_PATH=${LOG_PATH:-"${PARENT_PATH}/logs"}

    #local LOG_PATH=$(python -c "from sparcur.config import auth; print(auth.get_path('log-path'))")
    local PROJECT_ID=${PROJECT_ID:-$(python -c "from sparcur.config import auth; print(auth.get('remote-organization'))")}

    local maybe_slot=()
    if [[ -n "${SYMLINK_OBJECTS_TO}" ]]; then
        # MUST use arrays to capture optional arguments like this otherwise
        # arg values with spaces in them will destroy your sanity
        maybe_slot+=(--symlink-objects-to "${SYMLINK_OBJECTS_TO}")
    fi

    echo "${PARENT_PATH}"  # needed to be able to follow logs

    if [ ! -d "${LOG_PATH}" ]; then
        mkdir "${LOG_PATH}" || { CODE=$?; echo 'mkdir of ${LOG_PATH} failed'; return $CODE; }
    fi

    if [[ -z "${ONLY_FILESYSTEM}" ]]; then
        # fetch annotations (ref:bash-pipeline-fetch-annotations)
        echo "Fetching annotations metadata"
        python -m sparcur.simple.fetch_annotations > "${LOG_PATH}/fetch-annotations.log" 2>&1 &
        local pids_final[0]=$!

        # fetch remote metadata (ref:bash-pipeline-fetch-remote-metadata-all)
        # if this fails with 503 errors, check the
        # remote-backoff-factor config variable
        echo "Fetching remote metadata"
        python -m sparcur.simple.fetch_remote_metadata_all \
            --project-id "${PROJECT_ID}" \
            > "${LOG_PATH}/fetch-remote-metadata.log" 2>&1 &
        local pids[0]=$!
    fi

    local FAIL=0

    # clone aka fetch top level

    # we do not background this assignment because it runs quickly
    # and everything that follows depends on it finishing, plus we
    # need it to finish to set the PROJECT_PATH variable here
    echo python -m sparcur.simple.clone --project-id "${PROJECT_ID}" --parent-path "${PARENT_PATH}" "${maybe_slot[@]}"
    echo "Cloning top level"
    set -o pipefail
    PROJECT_PATH=$(python -m sparcur.simple.clone \
                          --project-id "${PROJECT_ID}" \
                          --parent-path "${PARENT_PATH}" \
                          "${maybe_slot[@]}" \
                          2>&1 | tee "${LOG_PATH}/clone.log" | tail -n 1) || {
        # TODO tee the output when verbose is passed
        CODE=$?;
        tail -n 100 "${LOG_PATH}/clone.log";
        echo "Clone failed! The last 100 lines of ${LOG_PATH}/clone.log are listed above.";
        apids=( "${pids[@]}" "${pids_final[@]}" );
        for pid in "${apids[@]}"; do
            kill $pid;
        done;
        set +o pipefail
        return $CODE;
    }
    set +o pipefail

    # explicit export of the current project path for pipelines
    # ideally we wouldn't need this, and when this pipeline
    # finished the export pipeline would kick off, or the export
    # pipeline would search for ... an existing project path ...
    # by ... oh right, looking for an environment variable or
    # checksing some other persistent state ... so this is the one
    # unless some controlling process sets it top down from the start
    # but we can't assume that
    export SPARCUR_PROJECT_PATH="${PROJECT_PATH}"

    for pid in "${pids[@]}"; do
        wait $pid || { FAIL=$((FAIL+1)); echo "${pid} failed!"; }
    done
    if [[ $FAIL -ne 0 || -z "${PROJECT_PATH}" ]]; then
        echo "${FAIL} commands failed. Cannot continue."
        echo "${PROJECT_PATH}"
        return 1
    fi

    # pull aka fetch file system metadata
    echo "Fetching file system metadata"
    echo python -m sparcur.simple.pull --project-path "${PROJECT_PATH}"
    python -m sparcur.simple.pull \
           --project-path "${PROJECT_PATH}" \
           > "${LOG_PATH}/pull.log" 2>&1 || {
        CODE=$?;
        tail -n 100 "${LOG_PATH}/pull.log";
        echo "Pull failed! The last 100 lines of ${LOG_PATH}/pull.log are listed above.";
        echo "${PROJECT_PATH}";
        return $CODE; }

    # fetch metadata files
    echo "Fetching metadata files"
    # have to pass project path as a position argument here so that it
    # does not try to pull aka fetch the file system metadata again
    echo python -m sparcur.simple.fetch_metadata_files --project-path "${PROJECT_PATH}"
    python -m sparcur.simple.fetch_metadata_files \
           --project-path "${PROJECT_PATH}" \
           > "${LOG_PATH}/fetch-metadata-files.log" 2>&1 &

    pids_final[1]=$!

    # fetch files
    echo "Fetching files"
    # XXX at some point this will probably also depend on the manifests
    # so we don't fetch everything with a matching extension
    # TODO derive --extension from manifests or all it to be passed in
    echo python -m sparcur.simple.fetch_metadata_files --project-path "${PROJECT_PATH}" --extension xml

    # FIXME fetch_files fails silently here :/
    python -m sparcur.simple.fetch_files \
           --project-path "${PROJECT_PATH}" \
           --extension xml \
           > "${LOG_PATH}/fetch-files.log" 2>&1 &

    pids_final[2]=$!

    local FAIL=0
    for pid in "${pids_final[@]}"; do
        wait $pid || { FAIL=$((FAIL+1)); echo "${pid} failed!"; }
    done

    # FIXME HACK
    #find -type f -size 0 -exec getfattr -d {} \;
    #find -type f -size 0 -exec spc fetch --limit=-1 {} \;

    if [[ $FAIL -ne 0 ]]; then
        echo "${FAIL} commands failed. Cannot continue."
        echo "${PROJECT_PATH}"
        return 1
    fi
    echo "All fetching completed successfully."

}

# pushd to the project location
pushd "${PROJECT_PATH:-SPARCUR_PROJECT_PATH}"
spc export
popd
# pass the project location as a positional argument
spc export "${PROJECT_PATH:-SPARCUR_PROJECT_PATH}"
# pass the project location as an option
spc export --project-path "${PROJECT_PATH:-SPARCUR_PROJECT_PATH}"

function sparc-export-all () {
    # parse args
    local POSITIONAL=()
    while [[ $# -gt 0 ]]
    do
    key="$1"
    case $key in # (ref:(((sigh)
        --project-path) local PROJECT_PATH="${2}"; shift; shift ;;
        -h|--help)      echo "${HELP}"; return ;;
        *)              POSITIONAL+=("$1"); shift ;;
    esac
    done

    local PROJECT_PATH="${PROJECT_PATH:-$SPARCUR_PROJECT_PATH}"
    spc export --project-path "${PROJECT_PATH}"
}

sparc-fexport ${UUID}
sparc-fexport dataset:${UUID}
sparc-fexport N:dataset:${UUID}
sparc-fexport https://api.pennsieve.io/datasets/N:dataset:${UUID}
sparc-fexport https://app.pennsieve.io/N:organization:618e8dd9-f8d2-4dc4-9abb-c6aaab2e78a0/datasets/N:dataset:${UUID}

function sparc-export () {
    echo TODO not ready yet
    return 1
}

function sparc-fexport () {

    local DATASET_ID="${1}"
    local DATASET_UUID
    local DATASET_PATH
    local EXPORT_PATH

    DATASET_UUID="$(python -m sparcur.simple.utils --dataset-id ${DATASET_ID})"

    python -m sparcur.simple.retrieve --dataset-id ${DATASET_UUID} &&

    EXPORT_PATH="$(realpath "${DATASET_UUID}/exports")" &&
    DATASET_PATH="$(realpath "${DATASET_UUID}/dataset")" &&
    pushd "${DATASET_PATH}" &&

    # FIXME we shouldn't need this refetch so I think that retrieve is
    # broken if files/folders already exist
    python -m sparcur.cli find \
        --name '*.xlsx' \
        --name '*.xml' \
        --name 'submission*' \
        --name 'code_description*' \
        --name 'dataset_description*' \
        --name 'subjects*' \
        --name 'samples*' \
        --name 'manifest*' \
        --name 'resources*' \
        --name 'README*' \
        --no-network \
        --limit -1 \
        --fetch
    wait $!
    python -m sparcur.cli export --export-path "${EXPORT_PATH}" &  # FIXME TODO this conflates phases
    local pids[0]=$!
    # FIXME TODO for now export_single_dataset produces this so we don't run it independently
    # FIXME there is also a difference in the export folder because the path metadata targets
    # the last updated data and thus overwrites if the data has not changed but the code has
    #python -m sparcur.simple.path_metadata_validate --export-path "${EXPORT_PATH}" &
    #local pids[1]=$!
    local FAIL=0
    # TODO log/notify export failure

    for pid in "${pids[@]}"; do
        wait $pid || { FAIL=$((FAIL+1)); echo "${pid} failed!"; }
    done
    if [[ $FAIL -ne 0 ]]; then
        echo "${FAIL} commands failed. Cannot continue."
        echo "${DATASET_UUID}"
        echo "${DATASET_PATH}"
        return 1
    fi
    popd # or do it yourself because we might need to explore??
}

for d in $(ls *-* -d); do refetch ${d}; done

for d in $(ls *-* -d); do
    find ~/".local/share/sparcur/export/datasets/${d}/LATEST/curation-export.json" -name 'curation-export.json';
done

import json
import rdflib
from pathlib import Path
from dateutil import parser as dateparser
from pyontutils.core import OntResPath
from pyontutils.utils import TZLOCAL, timeformat_friendly
from pyontutils.namespaces import TEMP, rdf, sparc
#from sparcur.utils import fromJson, register_all_types  # FIXME this should also go in sparcron
from sparcur.export.triples import TriplesExportSummary
from sparcur.export.published import _merge_graphs
from sparcur.simple.export import add_prov

tu = 'timestamp_updated'
tuc = 'timestamp_updated_contents'
ip = 'inputs'
rmk = 'remote_dataset_metadata'


class TriplesExportSummaryPublished(TriplesExportSummary):

    @property
    def ontid(self):
        return rdflib.URIRef(super().ontid.replace('ontologies/', 'ontologies/published/'))

    @property
    def header_label(self):
        return rdflib.Literal(f'{self.folder_name} curation export published graph')


def max_dataset_or_contents_updated(datasets_list):
    return max(set.union(
        set([a['meta'][tuc] for a in datasets_list
            if tuc in a['meta'] and a['meta'][tuc]]),
        set([a['meta'][tu] for a in datasets_list
            if tu in a['meta'] and a['meta'][tu]])))


def from_dataset_export_path_snapshot(dataset_export_path, snapshots_path, time_now):
    derefs = [l.resolve() for l in [c / 'LATEST' for c in dataset_export_path.children] if l.exists()]
    snapshot_path = snapshots_path / time_now.START_TIMESTAMP_LOCAL_FRIENDLY
    snapshot_path.mkdir(parents=True)
    [(snapshot_path / d.parts[-2]).symlink_to((snapshot_path / 'asdf').relative_path_to(d)) for d in derefs]
    return snapshot_path


def from_snapshot_path_datasets_lists(snapshot_path):
    alld = []
    pubd = []
    for uuid_path in snapshot_path.children:
        with open(uuid_path / 'curation-export.json', 'rt') as f:
            j = json.load(f)
            # TODO validate the load XXX this should probably
            # happen as a final roundtrip check during export
            # TODO filter by organization
            alld.append(j)
            if ip in j and rmk in j[ip] and 'id_published' in j[ip][rmk]:
                pubd.append(j)

    return alld, pubd


def from_snapshot_path_summary_json(snapshot_path, project_id, time_now):
    l_all, l_pub = from_snapshot_path_datasets_lists(snapshot_path)
    sum_all = from_datasets_list_summary_json(l_all, project_id, time_now)
    sum_pub = from_datasets_list_summary_json(l_pub, project_id, time_now)
    return sum_all, sum_pub


def from_snapshot_path_summary_ttl(snapshot_path, project_id, time_now, blob):
    tes = TriplesExportSummary(blob)
    tesp = TriplesExportSummaryPublished(blob)

    orps = [OntResPath(uuid_path / 'curation-export.ttl')
            for uuid_path in sorted(snapshot_path.children, key=lambda p: p.name)]

    graphs_all = [o.graph for o in orps]
    graphs_pub = [
        g for g, doi in [(g, list(g[ds:TEMP.hasDoi]))
                         for g in graphs_all
                         for ds in g[:rdf.type:sparc.Dataset]]
        if doi]
    merged_all = _merge_graphs(graphs_all)
    merged_all.populate_from_triples(tes.triples_header)
    merged_pub = _merge_graphs(graphs_pub)
    merged_pub.populate_from_triples(tesp.triples_header)
    return merged_all, merged_pub, graphs_all, graphs_pub

def from_snapshot_path_summary_ttl_BAD(snapshot_path, project_id, time_now, blob):
    # this variant is too complex, trying to reuse the published graph as the all graph
    # and the implementation of the OntConjunctiveGraph is not far enough along to do it
    tes = TriplesExportSummary(blob)  # FIXME nasty annoying dep
    graph_meta = OntGraph()
    graph_meta.populate_from_triples(tes._triples_header(tes.ontid, time_now._start_time))
    rev_replace_pairs = _fix_for_pub(graph_meta, graph_meta)
    replace_pairs = tuple([(b, a) for a, b in rev_replace_pairs])

    orps = [OntResPath(uuid_path / 'curation_export.ttl')
            for uuid_path in snapshot_path.children]

    graphs = [o.graph for o in orps]
    # FIXME should use id_published here as well, but that isn't being
    # propagated at the moment
    graphs_pub = []
    graphs_nop = []
    for g, doi in [(g, list(g[ds:TEMP.hasDoi]))
                   for g in graphs for ds in g[:rdf.type:sparc.Dataset]]:
        if doi:
            graphs_pub.append(g)
        else:
            graphs_nop.add(g)
    graph_pub = _merge_graphs(published_graphs)
    graph_pub.populate_from(graph_meta)
    # FIXME this is manually aligned with TriplesExport.triples_header
    graph_pub.asdf
    for g in graphs_nop:
        graph_pub.namespace_manager.populate_from(
            {k:v for k, v in dict(g.namespace_manager).items()
            if k not in ('contributor', 'sample', 'subject')})

    ttl_all = None
    ttl_pub = _populate_published(curation_export, graph)


def from_dataset_export_path_datasets_lists(dataset_export_path):
    dep = dataset_export_path
    alld = []
    pubd = []
    derefs = [l.resolve() for l in [c / 'LATEST' for c in dep.children] if l.exists()]
    # TODO consider creating a folder that is just symlinks before this
    for lp in sorted(derefs, key=lambda p: p.name):
        with open(lp / 'curation-export.json', 'rt') as f:
            j = json.load(f)
            # TODO validate the load XXX this should probably
            # happen as a final roundtrip check during export
            # TODO filter by organization
            alld.append(j)
            if ip in j and rmk in j[ip] and 'id_published' in j[ip][rmk]:
                pubd.append(j)

    return alld, pubd


def from_datasets_list_summary_json(datasets_list, project_id, time_now):
    # XXX FIXME issue with datasets from multiple projects
    fn = Path(datasets_list[0]['prov']['export_project_path']).name
    out = {
        'id': project_id.id,
        'meta': {
            'count': len(datasets_list),
            'folder_name': fn,  # WHAT A RELIEF we don't need network
            'uri_api': project_id.uri_api,
            'uri_human': project_id.uri_human(),
        },
        'datasets': datasets_list,
    }
    add_prov(out, time_now)
    # some potential prov summaries, but lets not do that here
    # most prov stats should be on the single dataset level
    #'export_timestamp_start_min': min(tes),
    #'export_timestamp_start_max': max(tes),
    return out


def from_dataset_export_path_summary_json(dataset_export_path, project_id, time_now):
    l_all, l_pub = from_dataset_export_path_datasets_lists(dataset_export_path)
    #[a['meta']['timestamp_updated'] < a['meta']['timestamp_updated_contents']
    #for a in l_all if a['meta']['timestamp_updated_contents']]
    sum_all = from_datasets_list_summary_json(l_all, project_id, time_now)
    sum_pub = from_datasets_list_summary_json(l_pub, project_id, time_now)
    return sum_all, sum_pub


def main(project_id=None, export_path=None, time_now=None,
         project_id_auth_var='remote-organization',  # FIXME move to clifun
         disco=False, **kwargs):
    from sparcur.paths import Path
    from sparcur.config import auth

    if project_id is None:
        from sparcur.config import auth
        from sparcur.utils import PennsieveId
        project_id = auth.get(project_id_auth_var)
        project_id = PennsieveId(project_id)  # FIXME abstract the hardcoded backend

    if export_path is None:  # XXX see issues mentioned above
        export_path = Path(auth.get_path('export-path'))

    dataset_export_path = export_path / 'datasets'
    snapshots_path = export_path / 'snapshots'
    snapshot_path = from_dataset_export_path_snapshot(
        dataset_export_path, snapshots_path, time_now)
    sum_all, sum_pub = from_snapshot_path_summary_json(
        snapshot_path, project_id, time_now)
    # write symlink LATEST_PARTIAL
    ttl_all, ttl_pub, graphs_all, graphs_pub = from_snapshot_path_summary_ttl(
        snapshot_path, project_id, time_now, sum_all)
    # write symlink LATEST
    maxdt = max_dataset_or_contents_updated(sum_all['datasets'])
    dt_maxdt = dateparser.parse(maxdt)
    dt_maxdt_local = dt_maxdt.astimezone(TZLOCAL())
    friendly_maxdt_local = timeformat_friendly(dt_maxdt_local)
    # FIXME there are some bad assumptions in here that should be refactored out
    # at some point, but for now we implicitly assume that all datasets come from
    # the same organization, which can easily be violated because we don't check 
    # however the existing internal schema requires an id for the summary which is
    # currenty the organization id
    # FIXME summary is a hardcoded path
    # XXX WARNING it is possible to overwrite since maxdt might not change between runs
    # this is desirable behavior for development, but could cause issues in other cases
    pexpath = export_path / 'summary' / project_id.uuid
    latest = pexpath / 'LATEST'
    npath = pexpath / friendly_maxdt_local
    snapshot_link = npath / 'snapshot'
    if not npath.exists():
        npath.mkdir(parents=True)
    else:
        # FIXME not sure if this makes sense?
        if snapshot_link.is_symlink():
            snapshot_link.unlink()

    snapshot_link.symlink_to(snapshot_link.relative_path_to(snapshot_path))

    npath_ce = npath / 'curation-export.json'
    npath_cep = npath / 'curation-export-published.json'
    for path, blob in ((npath_ce, sum_all),
                       (npath_cep, sum_pub)):
        with open(path, 'wt') as f:
            json.dump(blob, f, indent=2)

    npath_ttl = npath / 'curation-export.ttl'
    npath_ttlp = npath / 'curation-export-published.ttl'
    ttl_all.write(npath_ttl)
    ttl_pub.write(npath_ttlp)

    if disco:
        # export xml and tsv for disco
        from sparcur.simple.disco import from_curation_export_json_path_datasets_json_xml_and_disco
        from_curation_export_json_path_datasets_json_xml_and_disco(
            npath_ce, sum_all['datasets'], graphs_all)

    if latest.is_symlink():
        latest.unlink()

    latest.symlink_to(friendly_maxdt_local)

    return npath_ce, sum_all, sum_pub, graphs_all, graphs_pub


if __name__ == '__main__':
    #import pprint
    from sparcur.simple.utils import pipe_main
    # these are really big, don't print them
    # pipe_main(main, after=pprint.pprint)
    pipe_main(main)

import json
from pyontutils.core import OntGraph
from sparcur import export as ex
from sparcur.utils import PennsieveId
from sparcur.utils import fromJson, register_all_types  # FIXME this should also go in sparcron


def from_curation_export_json_path_xml_and_disco(curation_export_json_path):
    with open(curation_export_json_path, 'rt') as f:
        summary = json.load(f)

    datasets_json = summary['datasets']
    from_curation_export_json_path_datasets_json_xml_and_disco(
        curation_export_json_path, datasets_json)


def from_curation_export_json_path_datasets_json_xml_and_disco(
        curation_export_json_path, datasets_json, graphs=None):
    # FIXME need the snapshot linked somehow, export time started if we are using summary
    # or summary prov timestamp_export_start will give us the snapshot path as well if we
    # parse it back to a date
    if not graphs:
        snapshot_path = curation_export_json_path.parent / 'snapshot'
        paths = [(snapshot_path /
                  PennsieveId(d['id']).uuid /
                  'curation-export.ttl')
                 for d in datasets_json]
        graphs = [OntGraph().parse(path) for path in paths]

    datasets_ir = fromJson(datasets_json)
    ex.export_xml(curation_export_json_path, datasets_ir)
    ex.export_disco(curation_export_json_path, datasets_ir, graphs)
    # XXX not doing jsonld here, it will be combined
    # from single dataset jsonld or similar


def main(path=None, **kwargs):
    register_all_types()
    # FIXME the problem with this approach is that can cannot run
    # multiple downstream steps from the same upstream step, we would
    # need a compositional way to tell each downstream which upstreams
    # we wanted to run in any given situation, all to save additional
    # reads from disk
    if path is None:  # assume the user wants to run combine first
        from sparcur.simple.combine import main as combine
        curation_export_json_path, summary_all, _, graphs_all, _ = combine(**kwargs)
        datasets_json = summary_all['datasets']
        from_curation_export_json_path_datasets_json_xml_and_disco(
            curation_export_json_path, datasets_json, graphs)
    else:
        curation_export_json_path = path
        from_curation_export_json_path_xml_and_disco(curation_export_json_path)


if __name__ == '__main__':
    #import pprint
    from sparcur.simple.utils import pipe_main
    # these are really big, don't print them
    # pipe_main(main, after=pprint.pprint)
    pipe_main(main)

"""
Examples:
    # look at the second one and then run the first one since it is easier and safer to stop and resume
    # python -m sparcur.simple.compact | xargs -P4 -r -I{} sh -c 'tar -cvJf "{}.tar.xz" "{}" && rm -r "{}"'
    python -m sparcur.simple.compact | xargs -P12 -r -I{} echo tar -cvJf '{}.tar.xz' '{}'
    # python -m sparcur.simple.compact | xargs -P6 -r -I{} echo rm -r '{}'
"""
from sparcur.paths import Path
from sparcur.config import auth


__dep_cache = {}
def latest_snapped(dataset_export_path, snapped):
    if dataset_export_path not in __dep_cache:
        cs = set(c for c in dataset_export_path.children if c.is_dir() and not c.is_symlink())
        csnap = cs.intersection(snapped)
        if not csnap:  # no snap, pretend that latest is snapped
            # this can happen if there is no LATEST because we
            # were e.g. just exporting path metadata and nothing else
            maxsnap = sorted(cs, key=lambda p: p.name)[-1]
        else:
            maxsnap = sorted(csnap, key=lambda p: p.name)[-1]

        __dep_cache[dataset_export_path] = maxsnap.name

    return __dep_cache[dataset_export_path]


def main():
    export_path = Path(auth.get_path('export-path'))
    summary_path = export_path / 'summary'
    snapshots_path = export_path / 'snapshots'
    datasets_path = export_path / 'datasets'

    snap_shotted = [
        dss.resolve()
        for d in summary_path.rchildren_dirs
        for l in d.rchildren if l.is_symlink() and l.name == 'snapshot'
        for dss in l.resolve().children]

    snapped = set(snap_shotted)

    latest_sums = [
        d.resolve()
        for c in summary_path.children
        for d in (c / 'LATEST',) if d.exists()]

    all_builds = [
        build
        for date in datasets_path.children if date.is_dir() and not date.is_symlink()
        for build in date.children if build.is_dir() and not build.is_symlink()]

    older_not_snap = [
        a for a in all_builds
        if a not in snapped and a.name < latest_snapped(a.parent, snapped)]

    assert not snapped.intersection(older_not_snap)

    # newer = set(all_builds) - snapped - set(older_not_snap)

    _ = [print(p) for p in older_not_snap]


if __name__ == '__main__':
    main()

from sparcur.config import auth
__doc__ = f"""Common command line options for all sparcur.simple modules
Usage:
    sparcur-simple manifest  [options]  [<path>...]
    sparcur-simple get-uuid  <remote-id>...
    sparcur-simple datasets
    sparcur-simple for-racket
    sparcur-simple check-names
    sparcur-simple git-repos [update] [options]
    sparcur-simple [options] [<path>...]
    sparcur-simple [options] [--dataset-id=<ID>...]
    sparcur-simple [options] [--extension=<EXT>...] [<path>...]

Commands:
    manifest                        generate manifest files for path
    for-racket                      print data for reading into Racket

Options:
    -h --help                       show this

    --hypothesis-group-name=NAME    the hypotheis group name

    --project-id=ID                 the project id
    --dataset-id=<ID>...            one or more datset ids
    --project-id-auth-var=VAR       name of the auth variable holding the project-id

    --project-path=PATH             the project path, will be path if <path>... is empty
    --parent-path=PATH              the parent path where the project will be cloned to
    --parent-parent-path=PATH       parent in which a random tempdir is generated
                                    or the dataset uuid is used as the parent path, don't use this ...
    --invariant-local-path=PATH     path relative to parent path for dataset [default: dataset]
    --export-local                  set export-parent-path to {{:parent-path}}/exports/
    --export-path=PATH              base export path containing the standard path structure [default: {auth.get_path('export-path')}]
    --cleaned-parent-path=PATH      direct parent path into which cleaned paths will be placed
    --cleaned-path=PATH             base cleaned path containing the standard path structure [default: {auth.get_path('cleaned-path')}]
    --export-parent-path=PATH       direct parent path into which exports will be placed
    --extension=<EXT>...            one or more file extensions to fetch

    -j --jobs=N                     number joblib jobs [default: 12]
    --exclude-uploaded              do not pull files from remote marked as uploaded
    --sparse-limit=N                package count that forces a sparse pull [default: {auth.get('sparse-limit')}]
    --symlink-objects-to=PATH       path to an existing objects directory
    --log-level=LEVEL               log level info [default: INFO]
    --open=PROGRAM                  show file with PROGRAM
    --show                          show file with xopen
    --pretend                       show what would be done for update
"""

import os
import sys
from types import GeneratorType
from pyontutils import clifun as clif
from sparcur import exceptions as exc
from sparcur.utils import _find_command
from sparcur.utils import log, logd, loge, GetTimeNow, PennsieveId
from sparcur.paths import Path, PennsieveCache
from sparcur.backends import PennsieveRemote


def backend_pennsieve(project_id=None, Local=Path, Cache=PennsieveCache):  # (def_babf)
    """return a configured pennsieve backend
        calling this is sufficient to get everything set up correclty

        You must call RemotePath.init(project_id) before using
        RemotePath.  Passing the project_id argument to this function
        will do that for you. It is not required because there are
        cases where the project_id may not be known at the time that
        this function is called. """

    RemotePath = PennsieveRemote._new(Local, Cache)

    if project_id is not None:
        RemotePath.init(project_id)

    return RemotePath


class Options(clif.Options):

    @property
    def id(self):
        # dataset_id has priority since project_id can occure without a
        # dataset_id, but dataset_id may sometimes come with a project_id
        # in which case the dataset_id needs priority for functions that
        # branch on the most granular identifier type provided
        return (self.dataset_id[0]
                if self.dataset_id else
                (self.project_id
                 if self.project_id else
                 None))

    @property
    def project_id(self):
        if not hasattr(self, '_cache_project_id'):
            id = self._args['--project-id']
            if id is not None:
                identifier = PennsieveId(id, type='organization')
                self._cache_project_id = identifier
            else:
                return

        return self._cache_project_id

    @property
    def dataset_id(self):
        if not hasattr(self, '_cache_dataset_id'):
            ids = self._args['--dataset-id']
            if ids:
                identifiers = [PennsieveId(id, type='dataset') for id in ids]
                self._cache_dataset_id = identifiers
            else:
                return ids

        return self._cache_dataset_id

    @property
    def remote_id(self):
        if not hasattr(self, '_cache_remote_id'):
            ids = self._args['<remote-id>']
            if ids:
                identifiers = [PennsieveId(id) for id in ids]
                self._cache_remote_id = identifiers
            else:
                return ids

        return self._cache_remote_id

    @property
    def jobs(self):
        return int(self._args['--jobs'])

    n_jobs = jobs  # match internal kwargs conventions

    @property
    def paths(self):
        return [Path(p).expanduser().resolve() for p in self._args['<path>']]

    @property
    def path(self):
        paths = self.paths
        if paths:
            return paths[0]
        elif self.project_path:
            return self.project_path
        else:
            # if no paths were listed default to cwd
            # consistent with how the default kwargs
            # are set on a number of mains
            # this is preferable to allow path=None
            # to be overwritten by the conventions of
            # individual pipeline mains
            return Path.cwd()

    @property
    def project_path(self):
        pp = self._args['--project-path']
        if pp:
            return Path(pp).expanduser().resolve()

    @property
    def parent_parent_path(self):
        ppp = self._args['--parent-parent-path']
        if ppp:
            return Path(ppp).expanduser().resolve()
        else:
            return Path.cwd()

    @property
    def parent_path(self):
        pap = self._args['--parent-path']
        did = self.dataset_id
        if pap:
            return Path(pap).expanduser().resolve()
        elif did:
            id = self.id  # dataset_id is a list so use id which handles that
            uuid = id.uuid
            return (self.parent_parent_path / uuid).expanduser().resolve()

    @property
    def export_path(self):
        ep = self._args['--export-path']
        epp = self.export_parent_path
        if ep and epp:
            raise TypeError('Only one of --export-path and --export-parent-path may be set.')
        elif ep:
            return Path(ep).expanduser().resolve()
        else:
            raise Exception('should not happen')

    @property
    def export_parent_path(self):
        epp = self._args['--export-parent-path']
        el = self.export_local
        pap = self.parent_path
        if epp and el:
            raise TypeError('Only one of --export-local and --export-parent-path may be set.')
        elif epp:
            return Path(epp).expanduser().resolve()
        elif el and pap:
            # it is ok to do this here becuase the TypeError above prevents
            # the case where both epp and el are set, so even though epp
            # is no longer always what was set on the command line, it is
            # it is the case that there won't be conflicting sources
            return pap / 'exports'

    @property
    def cleaned_path(self):
        cp = self._args['--cleaned-path']
        cpp = self.export_parent_path
        if cp and cpp:
            raise TypeError('Only one of --cleaned-path and --cleaned-parent-path may be set.')
        elif cp:
            return Path(cp).expanduser().resolve()
        else:
            raise Exception('should not happen')

    @property
    def cleaned_parent_path(self):
        cpp = self._args['--cleaned-parent-path']
        if cpp:
            return Path(cpp).expanduser().resolve()

    @property
    def extensions(self):
        return self.extension

    @property
    def symlink_objects_to(self):
        slot = self._args['--symlink-objects-to']
        if slot:
            return Path(slot).expanduser()

    @property
    def sparse_limit(self):  # FIXME not being pulled in by asKwargs ??
        return int(self._args['--sparse-limit'])

    @property
    def time_now(self):  # FIXME make it possible to pass this in?
        if not hasattr(self, '_time_now'):
            self._time_now = GetTimeNow()

        return self._time_now

    @property
    def log_level(self):  # FIXME not being pulled in by asKwargs ??
        ll = self._args['--log-level']
        if ll.isdigit() or ll[0] == '-' and ll[1:].isdigit():
            return int(ll)
        else:
            return ll


def pipe_main(main, after=None, argv=None):
    options, args, defaults = Options.setup(__doc__, argv=argv)
    # _entry_point is used as a way to switch behavior when a
    # pipeline main is run directly or actually in a pipeline
    try:
        log.setLevel(options.log_level)
        logd.setLevel(options.log_level)
        loge.setLevel(options.log_level)
        out = main(_entry_point=True, **options.asKwargs())
    except Exception as e:
        log.exception(e)
        log.error(options.path)
        raise e

    if after:
        after(out)

    return out


def rglob(path, pattern):
    """ Hack around the absurd slowness of python's rglob """

    if sys.platform == 'win32':
        log.warning('no findutils on windows, watch out for unexpected files')
        return list(path.rglob(pattern))

    doig = (hasattr(path, 'cache') and
            path.cache and
            path.cache.cache_ignore)
    exclude = ' '.join([f"-not -path './{p}*'" for p in path.cache.cache_ignore]) if doig else ''
    command = f"""{_find_command} {exclude} -name {pattern!r}"""

    with path:
        with os.popen(command) as p:
            string = p.read()

    path_strings = string.split('\n')  # XXX posix path names can contain newlines
    paths = [path / s for s in path_strings if s]
    return paths


def _fetch(cache):  # sigh joblib multiprocessing pickling
    # lambda functions are great right up until you have to handle an
    # error function inside of them ... thanks python for yet another
    # failure to be homogenous >_<
    meta = cache.meta
    try:
        size_mb = meta.size.mb
    except AttributeError as e:
        if meta.errors:
            logd.debug(f'remote errors {meta.errors} for {cache!r}')
            return
        else:
            raise exc.StopTheWorld(cache) from e

    return cache.fetch(size_limit_mb=size_mb + 1)  # FIXME somehow this is not working !?


def _fetch_path(path):  # sigh joblib multiprocessing pickling
    path = Path(path)
    cache = path.cache
    if cache is None:
        raise exc.NoCachedMetadataError(path)

    # do not return to avoid cost of serialization back to the control process
    _fetch(cache)


def fetch_paths_parallel(paths, n_jobs=12, use_async=True):
    if n_jobs <= 1:
        [_fetch_path(path) for path in paths]
    elif use_async:
        from pyontutils.utils import Async, deferred
        Async()(deferred(_fetch_path)(path) for path in paths)
    else:
        import pathlib
        from joblib import Parallel, delayed
        backend = 'multiprocessing' if hasattr(sys, 'pypy_version_info') else None
        # FIXME FIXME FIXME somehow this can result in samples.xlsx being fetched to subjects.xlsx !?!??!!
        # XXX either a race condition on our end or something insane from the remote
        Parallel(n_jobs=n_jobs, backend=backend)(delayed(_fetch_path)(pathlib.Path(path)) for path in paths)
        #Parallel(n_jobs=n_jobs)(delayed(fetch_path)(path) for path in paths)


def combinate(*functions):
    def combinator(*args, **kwargs):
        for f in functions:
            # NOTE no error handling is needed here
            # in no cases should the construction of
            # the python object version of a path fail
            # all failures should happen _after_ construction
            # the way we have implemented this they fail when
            # the data attribute is accessed
            obj = f(*args, **kwargs)
            if isinstance(obj, GeneratorType):
                yield from obj
                # FIXME last one wins, vs yield tuple vs ...?
                # FIXME manifests are completely broken for this
            else:
                yield obj

    return combinator


def multiple(func, merge=None):
    """ combine multiple results """
    def express(*args, **kwargs):
        vs = tuple(func(*args, **kwargs))
        if merge is not None:
            yield merge(vs)
        else:
            yield vs

    return express


def early_failure(path, error, dataset_blob=None):
    # these are the 9999 5555 and 4444 errors
    # TODO match the minimal schema reqs as
    # we did in pipelines
    if dataset_blob is None:
        cache = path.cache
        return {'id': cache.id,
                'meta': {'uri_api': cache.uri_api,
                         'uri_human': cache.uri_human,},
                #'status': 'early_failure',  # XXX note that status is not requried
                # if we cannot compute it, but if there are inputs there should be
                # a status
                'errors': [error],  # FIXME format errro
                }

    else:
        if 'errors' not in datset_blob:
            dataset_blob['errors'] = []

        datset_blob['errors'].append(error)
        return dataset_blob


class DataWrapper:
    # sigh patterns are stupid, move this to elsewhere so it doesn't taint everything
    def __init__(self, data):
        self.data = data


def main(id=None, **kwargs):
    def ik(key):
        return key in kwargs and kwargs[key]

    if id is not None:
        print(id.uuid)

    if ik('get_uuid'):
        for id in kwargs['remote_id']:
            print(id.uuid)

        return

    if (ik('datasets') or ik('for_racket') or ik('check_names')):
        log.setLevel(60)  # logging.CRITICAL = 50
        from sparcur.config import auth
        from sparcur.simple.utils import backend_pennsieve
        if ik('project_id'):
            pass  # project_id from command line
        else:
            project_id = auth.get('remote-organization')

        PennsieveRemote = backend_pennsieve(project_id)
        root = PennsieveRemote(project_id)
        datasets = list(root.children)

    if ik('datasets'):
        print('\n'.join([d.id for d in datasets]))

    if ik('for_racket'):
        import json
        _dsmeta = '\n'.join([f"({json.dumps(d.id)} {json.dumps(d.name)})"
                             for d in datasets])
        dsmeta = f"({_dsmeta})"
        # lab pi last name should be cached in some other way
        print(dsmeta)

    if ik('check_names'):
        # you will want to run sparcur.simple.fetch_remote_metadata_all
        from pathlib import PurePosixPath
        def report(pid, exp, pub):
            pname = pub['name']
            name_mismatch = (
                False if exp['basename'] == pname
                else (exp['basename'], pname))
            # [6:] to strip files/
            ppname = PurePosixPath(pub['path']).name
            pathname_mismatch = (
                False if exp['basename'] == ppname
                else (exp['basename'], ppname))

            eppp = PurePosixPath(exp['dataset_relative_path']).parent.as_posix()
            pppp = PurePosixPath(pub['path'][6:]).parent.as_posix()
            parent_path_mismatch = (
                False if eppp == pppp
                else (eppp, pppp))

            # once we fix things on our end names should match
            # parent paths should match
            # name and pathname might match but do not have to match
            pid = f'  {pid}'
            nsp = '\n        '
            if name_mismatch:
                if pathname_mismatch and pname != ppname:
                    return (f'{pid}    name mismatch and pathname mismatch                '
                            f'{nsp}{nsp.join(name_mismatch)}{nsp}{nsp.join(pathname_mismatch)}')
                else:
                    return (f'{pid}    name mismatch                                      '
                            f'{nsp}{nsp.join(name_mismatch)}')

            if parent_path_mismatch:
                if True:
                    return (f'{pid}                                        parent mismatch'
                            f'{nsp}{nsp.join(parent_path_mismatch)}')

            if True:
                if True:
                    return ''
                    #return (f'{pid} ok                                                    ')

        import json
        from sparcur.backends import PennsieveDatasetData
        export_path = kwargs['export_path']
        epd = export_path / 'datasets'
        for dataset in datasets:
            latest = epd / dataset.identifier.uuid / 'LATEST'
            export_path_metadata =  latest / 'path-metadata.json'
            exported = export_path_metadata.exists()

            # pass the remote not just the id so that bfobject is
            # accessible to the RemoteDatasetData class
            pdd = PennsieveDatasetData(dataset)
            rmeta = pdd.fromCache()
            published = 'id_published' in rmeta
            if exported and published:
                with open(export_path_metadata, 'rt') as f:
                    j = json.load(f)
                epni = {'N:' + o['remote_id']:o for o in j['data']
                        if o['remote_id'].startswith('package:')}
                ppni = pdd._published_package_name_index()
                se, sp = set(epni), set(ppni)
                e_missing = sp - se
                p_missing = se - sp
                s_common = sp & se
                rep = [report(c, epni[c], ppni[c]) for c in s_common]
                repstr = '\n'.join([r for r in rep if r])
                if repstr:
                    print(dataset.id, 'bad')
                    print(repstr)
                else:
                    print(dataset.id, 'ok')
            elif exported:
                print(dataset.id, 'unpublished')
            elif published:
                print(dataset.id, 'unexported')
            else:
                print(dataset.id, 'unpublished and unexported')

    if ik('git_repos'):
        import augpathlib as aug
        import sparcur
        from sparcur.config import auth
        from importlib import metadata
        d = metadata.distribution(sparcur.__package__)
        rps = [p for p in [aug.RepoPath(d._path) for d in d.discover()] if p.working_dir]
        setups = [p for p in [p.parent / 'setup.py' for p in rps] if p.exists()]
        wds = sorted(set([p.working_dir for p in rps]))
        never_update = auth.get('never-update')
        pretend=ik('pretend') or never_update
        if pretend:
            if never_update:
                print(f'never-update: true is set in {auth.user_config._path}')
            print('These are the commands that would be run.')
        def doupdate(rp):
            if pretend:
                print(f'git -C {rp.as_posix()} stash')
                print(f'git -C {rp.as_posix()} pull')
                return
            print(f'Pulling {rp.as_posix()}')
            print(rp.repo.git.stash())
            # TODO checkout to a safety branch and tag for rollback
            print(rp.repo.git.pull())
        if ik('update'):
            for wd in wds:
                doupdate(wd)
            # indescriminately run setup.py with --release set to tangle
            from importlib import util as imu
            oldargv = sys.argv
            try:
                for s in setups:
                    if pretend:
                        print(f'pushd {s.parent.as_posix()}; python setup.py --release; popd')
                        continue
                    sys.argv = ['setup.py', '--release']  # reset every time since most will mutate
                    print(f'Maybe tangling via {s.as_posix()}')
                    spec = imu.spec_from_file_location(f'setup_{s.parent.name}', s)
                    mod = imu.module_from_spec(spec)
                    try:
                        with s.parent:  # ah relative paths
                            spec.loader.exec_module(mod)
                    except SystemExit:
                        pass
                    except Exception as e:
                        log.exception(e)
            finally:
                sys.argv = oldargv

    if ik('manifest'):
        from sparcur.utils import write_manifests
        paths = kwargs['paths']
        if not paths:
            paths = [Path.cwd()]

        manifests_rendered = write_manifests(parents=paths)
        manifests, rendered = zip(*manifests_rendered)
        nl = '\n'
        print(f'generated manifests at:\n{nl.join([m.as_posix() for m in manifests])}')
        if ik('open'):
            cmd = kwargs['open']
            [m.xopen(cmd) for m in manifests]
        elif ik('show'):
            [m.xopen() for m in manifests]


if __name__ == '__main__':
    pipe_main(main)

from sparcur.simple.utils import pipe_main

def test_pipe_main():
    def main(id=None, project_path=None, **kwargs):
        print(id, project_path, kwargs)

    pipe_main(main, argv=['sparcur-simple'])

from sparcur.paths import Path
from sparcur import datasets as dat
from sparcur.utils import symlink_latest
from sparcur.config import auth
from sparcur.simple.utils import rglob


def prepare_dataset_cleaned(dataset_path, cleaned_path=None, time_now=None):
    if cleaned_path is None:  # FIXME confusing and breaks w/ convention -> Options maybe?
        cleaned_path = Path(auth.get_path('cleaned-path'))

    from sparcur.utils import PennsieveId
    identifier = PennsieveId(dataset_path.cache.id)
    uuid = identifier.uuid
    cleaned_dataset_folder = cleaned_path / uuid
    cleaned_parent = cleaned_dataset_folder / time_now.START_TIMESTAMP_LOCAL_FRIENDLY
    if not cleaned_parent.exists():
        cleaned_parent.mkdir(parents=True)

    cleaned_latest = cleaned_dataset_folder / 'LATEST'
    # FIXME do we symlink before we know things have succeeded ???
    symlink_latest(cleaned_parent, cleaned_latest)
    return cleaned_parent


def from_dataset_path_metadata_file_paths(dataset_path):
    matches = []
    for candidate in rglob(dataset_path, '*.xlsx'):
        rp = candidate.relative_path_from(dataset_path)
        if not rp.parent.name or 'anifest' in rp.name:
            matches.append(candidate)

    return matches


def from_path_cleaned_object(path):
    t = dat.Tabular(path)
    sheet, wb, sparse_empty_rows = t._openpyxl_fixes()
    return wb


def from_file_paths_objects(paths):
    for path in paths:
        if path.suffix == '.xlsx':
            cleaned = from_path_cleaned_object(path)
            yield cleaned
        else:
            yield None


def from_dataset_path_cleaned_files(dataset_path, cleaned_parent):
    "NOTE this actually does the cleaning"

    paths = from_dataset_path_metadata_file_paths(dataset_path)
    for path, obj in zip(paths, from_file_paths_objects(paths)):
        if obj is not None:
            drp = path.dataset_relative_path
            target = cleaned_parent / drp
            if not target.parent.exists():
                target.parent.mkdir(parents=True)

            obj.save(target)


def main(path=Path.cwd(), id=None, time_now=None,
         parent_path=None, invariant_local_path='dataset',
         parent_parent_path=Path.cwd(),
         cleaned_path=None, cleaned_parent_path=None, **kwargs):
    # setup taken from path_metadata.py::main so check the notes there
    if path == Path.cwd() and (id is not None or parent_path is not None):
        if parent_path is None:
            uuid = id.uuid
            parent_path = parent_parent_path / uuid

        invariant_path = parent_path / invariant_local_path
        path = invariant_path.expanduser().resolve()
    else:
        parent_parent_path = None

    path = Path(path)
    cache = path.cache
    if not cache.is_dataset():
        raise TypeError('can only run on a single dataset')

    cleaned_parent = prepare_dataset_cleaned(path, cleaned_path, time_now)
    from_dataset_path_cleaned_files(path, cleaned_parent)


if __name__ == '__main__':
    import pprint
    from sparcur.simple.utils import pipe_main
    pipe_main(main, after=pprint.pprint)