Additional tests and fixes

client/ayon_kitsu/addon.py

@@ -94,7 +94,7 @@ def get_plugin_paths(self):
         return {
             "publish": self.get_publish_plugin_paths(),
             # The laucher action is not working since AYON conversion
-            # "actions": [os.path.join(KITSU_ROOT, "plugins", "launcher")],
+            "actions": [os.path.join(KITSU_ROOT, "plugins", "launcher")],
         }
 
     def get_publish_plugin_paths(self, host_name=None):

client/ayon_kitsu/plugins/launcher/launcher_show_in_kitsu.py

@@ -1,86 +1,107 @@
+try:
+    from openpype.pipeline import LauncherAction
+    from openpype.modules import ModulesManager
+
+except ImportError:
+    from ayon_core.pipeline import LauncherAction
+    from ayon_core.modules import ModulesManager
+
 import webbrowser
 import ayon_api
-
-from openpype.pipeline import LauncherAction
-from openpype.modules import ModulesManager
+import re
 
 
 class ShowInKitsu(LauncherAction):
     name = "showinkitsu"
     label = "Show in Kitsu"
     icon = "external-link-square"
-    color = "#e0e1e1"
+    # color = "#e0e1e1"
     order = 10
 
     @staticmethod
     def get_kitsu_module():
         return ModulesManager().modules_by_name.get("kitsu")
 
     def is_compatible(self, session):
-        if not session.get("AVALON_PROJECT"):
-            return False
-        return True
+        """Return whether the action is compatible with the session"""
+        return bool(self.get_kitsu_data(session))
 
-    def process(self, session, **kwargs):
-        # Context inputs
-        project_name = session["AVALON_PROJECT"]
-        asset_name = session.get("AVALON_ASSET", None)
-        task_name = session.get("AVALON_TASK", None)
+    def get_kitsu_data(self, session):
+        ## example session
+        # {'AVALON_PROJECT': 'Test_Project', 'AVALON_ASSET': '/episodes/test_ep/010/001', 'AVALON_TASK': 'mytask'}
+
+        # support for ayon_core
+        project_name = session.get("AYON_PROJECT_NAME", None)
+        folder_path = session.get("AYON_FOLDER_PATH", None)
+        task_name = session.get("AYON_TASK_NAME", None)
+
+        # support for OpenPype
+        if not project_name:
+            project_name = session.get("AVALON_PROJECT", None)
+            folder_path = session.get("AVALON_ASSET", None)
+            task_name = session.get("AVALON_TASK", None)
+
+        if not project_name or not folder_path:
+            return
 
         project = ayon_api.get_project(project_name)
         if not project:
-            raise RuntimeError("Project {} not found.".format(project_name))
-
-        project_zou_id = project["data"].get("zou_id")
-        if not project_zou_id:
-            raise RuntimeError(
-                "Project {} has no connected kitsu id.".format(project_name)
-            )
-
-        asset_zou_name = None
-        asset_zou_id = None
-        asset_zou_type = "Assets"
-        task_zou_id = None
-        zou_sub_type = ["AssetType", "Sequence"]
-        if asset_name:
-            asset_zou_name = asset_name
-            asset_fields = ["data.zou.id", "data.zou.type"]
-            if task_name:
-                asset_fields.append("data.tasks.{}.zou.id".format(task_name))
-
-            asset = get_asset_by_name(
-                project_name, asset_name=asset_name, fields=asset_fields
-            )
-
-            asset_zou_data = asset["data"].get("zou")
-
-            if asset_zou_data:
-                asset_zou_type = asset_zou_data["type"]
-                if asset_zou_type not in zou_sub_type:
-                    asset_zou_id = asset_zou_data["id"]
-            else:
-                asset_zou_type = asset_name
-
-            if task_name:
-                task_data = asset["data"]["tasks"][task_name]
-                task_zou_data = task_data.get("zou", {})
-                if not task_zou_data:
-                    self.log.debug(
-                        "No zou task data for task: {}".format(task_name)
-                    )
-                task_zou_id = task_zou_data["id"]
-
-        # Define URL
+            return
+
+        data = project.get("data")
+        kitsu_project_id = data.get("kitsuProjectId") if data else None
+
+        if not kitsu_project_id:
+            return None
+
+        kitsu_entity_id = None
+        kitsu_entity_type = None
+        kitsu_task_id = None
+
+        folder = ayon_api.get_folder_by_path(project_name, folder_path)
+        if not folder:
+            return
+        data = folder.get("data")
+        kitsu_entity_id = data.get("kitsuId") if data else None
+        kitsu_entity_type = data.get("kitsuType") if data else None
+
+        # required data
+        if not kitsu_entity_id or not kitsu_entity_type:
+            return
+
+        if task_name:
+            task = ayon_api.get_task_by_name(project_name, folder.get("id"), task_name)
+            if not task:
+                return
+            data = task.get("data")
+            kitsu_task_id = data.get("kitsuId") if data else None
+
+        # print(f"kitsu_project_id: {kitsu_project_id}")
+        # print(f"kitsu_entity_type: {kitsu_entity_type}")
+        # print(f"kitsu_entity_id: {kitsu_entity_id}")
+        # print(f"kitsu_task_id: {kitsu_task_id}")
+
+        return (kitsu_project_id, kitsu_entity_type, kitsu_entity_id, kitsu_task_id)
+
+    def process(self, session, **kwargs):
+        result = self.get_kitsu_data(session)
+        if not result:
+            return
+
+        kitsu_project_id, kitsu_entity_type, kitsu_entity_id, kitsu_task_id = result
+
+        # # Define URL
         url = self.get_url(
-            project_id=project_zou_id,
-            asset_name=asset_zou_name,
-            asset_id=asset_zou_id,
-            asset_type=asset_zou_type,
-            task_id=task_zou_id,
+            project_id=kitsu_project_id,
+            entity_id=kitsu_entity_id,
+            entity_type=kitsu_entity_type,
+            task_id=kitsu_task_id,
         )
+        if not url:
+            raise Exception("URL cound not be created")
 
-        # Open URL in webbrowser
-        self.log.info("Opening URL: {}".format(url))
+        # # Open URL in webbrowser
+        self.log.info(f"Opening URL: {url}")
         webbrowser.open(
             url,
             # Try in new tab
@@ -90,41 +111,38 @@ def process(self, session, **kwargs):
     def get_url(
         self,
         project_id,
-        asset_name=None,
-        asset_id=None,
-        asset_type=None,
+        entity_id=None,
+        entity_type=None,
         task_id=None,
     ):
-        shots_url = {"Shots", "Sequence", "Shot"}
-        sub_type = {"AssetType", "Sequence"}
+        # sub_type = {"AssetType", "Sequence"}
         kitsu_module = self.get_kitsu_module()
 
         # Get kitsu url with /api stripped
         kitsu_url = kitsu_module.server_url
-        if kitsu_url.endswith("/api"):
-            kitsu_url = kitsu_url[: -len("/api")]
+        kitsu_url = re.sub(r"\/?(api)?\/?$", "", kitsu_url)
 
         sub_url = f"/productions/{project_id}"
-        asset_type_url = "shots" if asset_type in shots_url else "assets"
+        type_url = f"{entity_type.lower()}s" if entity_type else "shots"
 
         if task_id:
             # Go to task page
-            # /productions/{project-id}/{asset_type}/tasks/{task_id}
-            sub_url += f"/{asset_type_url}/tasks/{task_id}"
+            # /productions/{project-id}/{type}/tasks/{task_id}
+            sub_url += f"/{type_url}/tasks/{task_id}"
 
-        elif asset_id:
+        elif entity_id:
             # Go to asset or shot page
             # /productions/{project-id}/assets/{entity_id}
             # /productions/{project-id}/shots/{entity_id}
-            sub_url += f"/{asset_type_url}/{asset_id}"
-
-        else:
-            # Go to project page
-            # Project page must end with a view
-            # /productions/{project-id}/assets/
-            # Add search method if is a sub_type
-            sub_url += f"/{asset_type_url}"
-            if asset_type in sub_type:
-                sub_url += f"?search={asset_name}"
+            sub_url += f"/{type_url}/{entity_id}"
+
+        # else:
+        #     # Go to project page
+        #     # Project page must end with a view
+        #     # /productions/{project-id}/assets/
+        #     # Add search method if is a sub_type
+        #     sub_url += f"/{asset_type_url}"
+        #     if asset_type in sub_type:
+        #         sub_url += f"?search={asset_name}"
 
         return f"{kitsu_url}{sub_url}"

package.py

@@ -1,3 +1,3 @@
 name = "kitsu"
 title = "Kitsu"
-version = "1.1.0"
+version = "1.1.1"

server/__init__.py

@@ -101,7 +101,7 @@ async def remove(
             payload=payload,
         )
 
-    async def list_pairings(self, mock: bool = False) -> list[PairingItemModel]:
+    async def list_pairings(self, mock: bool | None = None) -> list[PairingItemModel]:
         await self.ensure_kitsu(mock)
         return await get_pairing_list(self)
 
@@ -112,21 +112,21 @@ async def init_pairing(
     ) -> EmptyResponse:
         if not user.is_manager:
             raise ForbiddenException("Only managers can pair Kitsu projects")
-        await self.ensure_kitsu()
+        await self.ensure_kitsu(request.mock)
         await init_pairing(self, user, request)
         return EmptyResponse(status_code=201)
 
     #
     # Helpers
     #
     async def ensure_kitsu(self, mock: bool = False):
-        if self.kitsu is not None:
-            return
-
         if mock is True:
             self.kitsu = KitsuMock()
             return
 
+        if mock is None and self.kitsu is not None:
+            return
+
         settings = await self.get_studio_settings()
         if not settings.server:
             raise InvalidSettingsException("Kitsu server is not set")

server/kitsu/anatomy.py

@@ -7,7 +7,7 @@
 from ayon_server.settings.anatomy.statuses import Status
 from ayon_server.settings.anatomy.task_types import TaskType
 
-from .utils import create_short_name, remove_accents
+from .format_utils import create_short_name, remove_accents
 
 if TYPE_CHECKING:
     from .. import KitsuAddon