[ENH] add ability to clone a project (#1150)

* codex: add ability to clone project

* style

* add copy to project name

* set snapshot to None

since the snapshot cannot be inferred.

* set snapshot to None

the snapshot cannot be inferred.

* fix syntax

Author: jdkent

compose/backend/neurosynth_compose/openapi

@@ -1,7 +1,9 @@
 from collections import ChainMap
+from copy import deepcopy
 import json
 import pathlib
 from operator import itemgetter
+from urllib.parse import urlencode
 
 import connexion
 from connexion.lifecycle import ConnexionResponse
@@ -50,6 +52,7 @@
     NeurostoreStudySchema,
     ProjectSchema,
 )
+from .neurostore import neurostore_session
 from .singular import singularize
 
 
@@ -697,14 +700,28 @@ def db_validation(self, data):
                     )
 
     def post(self):
+        clone_args = parser.parse(
+            {
+                "source_id": fields.String(missing=None),
+                "copy_annotations": fields.Boolean(missing=True),
+            },
+            request,
+            location="query",
+        )
+
+        source_id = clone_args.get("source_id")
+        if source_id:
+            return self._clone_project(
+                source_id, clone_args.get("copy_annotations", True)
+            )
+
         try:
             data = parser.parse(self.__class__._schema, request)
         except ValidationError as e:
             abort(422, description=f"input does not conform to specification: {str(e)}")
 
         with db.session.no_autoflush:
             record = self.__class__.update_or_create(data)
-            # create neurostore study
             ns_study = NeurostoreStudy(project=record)
             db.session.add(ns_study)
             commit_session()
@@ -714,6 +731,265 @@ def post(self):
         payload = self.__class__._schema().dump(record)
         return _make_json_response(payload)
 
+    def _clone_project(self, source_id, copy_annotations):
+        current_user = self._ensure_current_user()
+
+        source_project = db.session.execute(
+            select(Project)
+            .options(
+                selectinload(Project.studyset).options(
+                    selectinload(Studyset.studyset_reference)
+                ),
+                selectinload(Project.annotation).options(
+                    selectinload(Annotation.annotation_reference)
+                ),
+                selectinload(Project.meta_analyses).options(
+                    selectinload(MetaAnalysis.specification).options(
+                        selectinload(Specification.specification_conditions)
+                    ),
+                    selectinload(MetaAnalysis.results),
+                ),
+            )
+            .where(Project.id == source_id)
+        ).scalar_one_or_none()
+
+        if source_project is None:
+            abort(404)
+
+        if (
+            not source_project.public
+            and source_project.user_id != current_user.external_id
+        ):
+            abort(403, description="project is not public")
+
+        access_token = request.headers.get("Authorization")
+        if not access_token:
+            from auth0.v3.authentication.get_token import GetToken
+
+            domain = current_app.config["AUTH0_BASE_URL"].lstrip("https://")
+            g_token = GetToken(domain)
+            token_resp = g_token.client_credentials(
+                client_id=current_app.config["AUTH0_CLIENT_ID"],
+                client_secret=current_app.config["AUTH0_CLIENT_SECRET"],
+                audience=current_app.config["AUTH0_API_AUDIENCE"],
+            )
+            access_token = " ".join(
+                [token_resp["token_type"], token_resp["access_token"]]
+            )
+
+        ns_session = neurostore_session(access_token)
+
+        with db.session.no_autoflush:
+            new_studyset, new_annotation = self._clone_studyset_and_annotation(
+                ns_session, source_project, current_user, copy_annotations
+            )
+
+            cloned_project = Project(
+                name=source_project.name + "Copy",
+                description=source_project.description,
+                provenance=self._clone_provenance(
+                    source_project.provenance,
+                    new_studyset.studyset_reference.id if new_studyset else None,
+                    new_annotation.annotation_reference.id if new_annotation else None,
+                ),
+                user=current_user,
+                public=False,
+                draft=True,
+                studyset=new_studyset,
+                annotation=new_annotation,
+            )
+
+            cloned_metas = []
+            for meta in source_project.meta_analyses:
+                cloned_meta = self._clone_meta_analysis(
+                    meta,
+                    current_user,
+                    cloned_project,
+                    new_studyset,
+                    new_annotation,
+                )
+                cloned_metas.append(cloned_meta)
+
+            cloned_project.meta_analyses = cloned_metas
+
+            db.session.add(cloned_project)
+            for meta in cloned_metas:
+                db.session.add(meta)
+
+            commit_session()
+
+            ns_study = NeurostoreStudy(project=cloned_project)
+            db.session.add(ns_study)
+            commit_session()
+            create_or_update_neurostore_study(ns_study)
+            db.session.add(ns_study)
+            commit_session()
+
+        payload = self.__class__._schema().dump(cloned_project)
+        return _make_json_response(payload)
+
+    def _ensure_current_user(self):
+        current_user = get_current_user()
+        if current_user:
+            return current_user
+        current_user = create_user()
+        if current_user:
+            db.session.add(current_user)
+            commit_session()
+            return current_user
+        abort(401, description="user authentication required")
+
+    def _clone_studyset_and_annotation(
+        self, ns_session, source_project, current_user, copy_annotations
+    ):
+        source_studyset = getattr(source_project, "studyset", None)
+        new_studyset = None
+        new_annotation = None
+
+        if source_studyset and source_studyset.studyset_reference:
+            query_params = {
+                "source_id": source_studyset.studyset_reference.id,
+            }
+            if copy_annotations is not None:
+                query_params["copy_annotations"] = str(bool(copy_annotations)).lower()
+
+            path = "/api/studysets/"
+            if query_params:
+                path = f"{path}?{urlencode(query_params)}"
+
+            ns_response = ns_session.post(path, json={})
+            ns_payload = ns_response.json()
+
+            ss_ref = self._get_or_create_reference(
+                StudysetReference, ns_payload.get("id")
+            )
+            new_studyset = Studyset(
+                user=current_user,
+                snapshot=None,
+                version=source_studyset.version,
+                studyset_reference=ss_ref,
+            )
+            db.session.add(new_studyset)
+
+            source_annotation = getattr(source_project, "annotation", None)
+            annotations_payload = ns_payload.get("annotations") or []
+            annotation_id = None
+            if annotations_payload:
+                annotation_id = annotations_payload[0].get("id")
+            if source_annotation and copy_annotations and annotation_id:
+                annot_ref = self._get_or_create_reference(
+                    AnnotationReference, annotation_id
+                )
+                new_annotation = Annotation(
+                    user=current_user,
+                    snapshot=None,
+                    annotation_reference=annot_ref,
+                    studyset=new_studyset,
+                )
+                db.session.add(new_annotation)
+
+        return new_studyset, new_annotation
+
+    def _clone_meta_analysis(self, meta, user, project, new_studyset, new_annotation):
+        cloned_spec = self._clone_specification(meta.specification, user)
+        cloned_meta = MetaAnalysis(
+            name=meta.name,
+            description=meta.description,
+            specification=cloned_spec,
+            studyset=new_studyset,
+            annotation=new_annotation,
+            user=user,
+            project=project,
+            provenance=self._clone_meta_provenance(
+                meta.provenance,
+                new_studyset.studyset_reference.id if new_studyset else None,
+                new_annotation.annotation_reference.id if new_annotation else None,
+            ),
+        )
+
+        if new_studyset and new_studyset.studyset_reference:
+            cloned_meta.neurostore_studyset_id = new_studyset.studyset_reference.id
+            cloned_meta.cached_studyset = new_studyset
+
+        if new_annotation and new_annotation.annotation_reference:
+            cloned_meta.neurostore_annotation_id = (
+                new_annotation.annotation_reference.id
+            )
+            cloned_meta.cached_annotation = new_annotation
+
+        return cloned_meta
+
+    @staticmethod
+    def _clone_specification(specification, user):
+        if specification is None:
+            return None
+        cloned_spec = Specification(
+            type=specification.type,
+            estimator=deepcopy(specification.estimator),
+            database_studyset=specification.database_studyset,
+            filter=specification.filter,
+            corrector=deepcopy(specification.corrector),
+            user=user,
+        )
+        for spec_cond in specification.specification_conditions:
+            cloned_cond = SpecificationCondition(
+                weight=spec_cond.weight,
+                condition=spec_cond.condition,
+                user=user,
+            )
+            cloned_spec.specification_conditions.append(cloned_cond)
+        return cloned_spec
+
+    @staticmethod
+    def _clone_provenance(provenance, studyset_id, annotation_id):
+        if provenance is None:
+            return None
+        cloned = deepcopy(provenance)
+        extraction = cloned.get("extractionMetadata", {})
+        if studyset_id:
+            extraction["studysetId"] = studyset_id
+        if annotation_id:
+            extraction["annotationId"] = annotation_id
+        cloned["extractionMetadata"] = extraction
+
+        meta_meta = cloned.get("metaAnalysisMetadata")
+        if isinstance(meta_meta, dict):
+            meta_meta["canEditMetaAnalyses"] = True
+            cloned["metaAnalysisMetadata"] = meta_meta
+
+        return cloned
+
+    @staticmethod
+    def _clone_meta_provenance(provenance, studyset_id, annotation_id):
+        if provenance is None:
+            return None
+        cloned = deepcopy(provenance)
+        if studyset_id:
+            for key in ("studysetId", "studyset_id"):
+                if key in cloned:
+                    cloned[key] = studyset_id
+        if annotation_id:
+            for key in ("annotationId", "annotation_id"):
+                if key in cloned:
+                    cloned[key] = annotation_id
+        for key in ("hasResults", "has_results"):
+            if key in cloned:
+                cloned[key] = False
+        return cloned
+
+    @staticmethod
+    def _get_or_create_reference(model_cls, identifier):
+        if identifier is None:
+            return None
+        existing = db.session.execute(
+            select(model_cls).where(model_cls.id == identifier)
+        ).scalar_one_or_none()
+        if existing:
+            return existing
+        reference = model_cls(id=identifier)
+        db.session.add(reference)
+        return reference
+
 
 def create_neurovault_collection(nv_collection):
     import flask