Enhancement/kitsu to ayon name conversions

server/kitsu/addon_helpers.py

@@ -0,0 +1,95 @@
+import unicodedata
+import re
+from nxtools import slugify
+
+"""
+A collection of helper functions for Ayon Addons
+minimal dependencies, pytest unit tests
+"""
+
+
+## ========== KITSU -> AYON NAME CONVERSIONS =====================
+
+
+def create_short_name(name: str) -> str:
+    """create a shortname from the full name when a shortname is not present"""
+    code = name.lower()
+
+    if "_" in code:
+        subwords = code.split("_")
+        code = "".join([subword[0] for subword in subwords])[:4]
+    elif len(name) > 4:
+        vowels = ["a", "e", "i", "o", "u"]
+        filtered_word = "".join([char for char in code if char not in vowels])
+        code = filtered_word[:4]
+
+    # if there is a number at the end of the code, add it to the code
+    last_char = code[-1]
+    if last_char.isdigit():
+        code += last_char
+
+    return code
+
+
+def to_username(first_name: str, last_name: str | None = None) -> str:
+    """converts usernames from kitsu - converts accents"""
+
+    name = (
+        f"{first_name.strip()}.{last_name.strip()}" if last_name else first_name.strip()
+    )
+
+    name = name.lower()
+    name = remove_accents(name)
+    return to_entity_name(name)
+
+
+def remove_accents(input_str: str) -> str:
+    """swap accented characters for a-z equivilants ž => z"""
+
+    nfkd_form = unicodedata.normalize("NFKD", input_str)
+    result = "".join([c for c in nfkd_form if not unicodedata.combining(c)])
+
+    # manually replace exceptions
+    # @see https://stackoverflow.com/questions/3194516/replace-special-characters-with-ascii-equivalent
+    replacement_map = {
+        "Æ": "AE",
+        "Ð": "D",
+        "Ø": "O",
+        "Þ": "TH",
+        "ß": "ss",
+        "æ": "ae",
+        "ð": "d",
+        "ø": "o",
+        "þ": "th",
+        "Œ": "OE",
+        "œ": "oe",
+        "ƒ": "f",
+    }
+    for k, v in replacement_map.items():
+        if k in result:
+            result = result.replace(k, v)
+
+    # remove any unsupported characters
+    result = re.sub(r"[^a-zA-Z0-9_\.\-]", "", result)
+
+    return result
+
+
+def to_entity_name(name) -> str:
+    """convert names so they will pass Ayon Entity name validation
+    @see ayon_server.types.NAME_REGEX = r"^[a-zA-Z0-9_]([a-zA-Z0-9_\.\-]*[a-zA-Z0-9_])?$"
+    """
+
+    assert name, "Entity name cannot be empty"
+
+    name = name.strip()
+
+    # replace whitespace
+    name = re.sub(r"\s+", "_", name)
+    # remove any invalid characters
+    name = re.sub(r"[^a-zA-Z0-9_\.\-]", "", name)
+
+    # first and last characters cannot be . or -
+    name = re.sub(r"^[^a-zA-Z0-9_]+", "", name)
+    name = re.sub(r"[^a-zA-Z0-9_]+$", "", name)
+    return name

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 .addon_helpers import create_short_name, remove_accents
 
 if TYPE_CHECKING:
     from .. import KitsuAddon

server/kitsu/push.py

@@ -23,11 +23,13 @@
     get_folder_by_kitsu_id,
     get_task_by_kitsu_id,
     get_user_by_kitsu_id,
-    remove_accents,
     update_folder,
     update_task,
 )
 
+from .addon_helpers import to_username
+
+
 if TYPE_CHECKING:
     from .. import KitsuAddon
 
@@ -221,14 +223,15 @@ async def generate_user_settings(
 async def sync_person(
     addon: "KitsuAddon",
     user: "UserEntity",
+    existing_users: dict[str, Any],
     entity_dict: "EntityDict",
 ):
-    logging.info("sync_person")
-
-    username = remove_accents(
-        f"{entity_dict['first_name']}.{entity_dict['last_name']}".lower().strip()
+    logging.info(
+        f"sync_person: {entity_dict.get('first_name')} {entity_dict.get('last_name')}"
     )
 
+    username = to_username(entity_dict["first_name"], entity_dict["last_name"])
+
     payload = {
         "name": username,
         "attrib": {
@@ -265,8 +268,8 @@ async def sync_person(
                 )
             # Rename the user
             payload = {
-                "newName": remove_accents(
-                    f"{entity_dict['first_name']}.{entity_dict['last_name']}".lower().strip()
+                "newName": to_username(
+                    entity_dict["first_name"], entity_dict["last_name"]
                 )
             }
             async with httpx.AsyncClient() as client:
@@ -283,6 +286,9 @@ async def sync_person(
         user.set_password(settings.sync_settings.sync_users.default_password)
         await user.save()
 
+    # update the id map
+    existing_users[entity_dict["id"]] = username
+
 
 async def update_project(
     addon: "KitsuAddon",
@@ -549,6 +555,7 @@ async def push_entities(
 
     folders = {}
     tasks = {}
+    users = {}
 
     settings = await addon.get_studio_settings()
     for entity_dict in payload.entities:
@@ -577,6 +584,7 @@ async def push_entities(
                 await sync_person(
                     addon,
                     user,
+                    users,
                     entity_dict,
                 )
         elif entity_dict["type"] != "Task":
@@ -602,7 +610,7 @@ async def push_entities(
     )
 
     # pass back the map of kitsu to ayon ids
-    return {"folders": folders, "tasks": tasks}
+    return {"folders": folders, "tasks": tasks, "users": users}
 
 
 async def remove_entities(

server/kitsu/utils.py

@@ -37,30 +37,6 @@ def calculate_end_frame(
         return frame_start + nb_frames - 1
 
 
-def remove_accents(input_str: str) -> str:
-    nfkd_form = unicodedata.normalize("NFKD", input_str)
-    return "".join([c for c in nfkd_form if not unicodedata.combining(c)])
-
-
-def create_short_name(name: str) -> str:
-    code = name.lower()
-
-    if "_" in code:
-        subwords = code.split("_")
-        code = "".join([subword[0] for subword in subwords])[:4]
-    elif len(name) > 4:
-        vowels = ["a", "e", "i", "o", "u"]
-        filtered_word = "".join([char for char in code if char not in vowels])
-        code = filtered_word[:4]
-
-    # if there is a number at the end of the code, add it to the code
-    last_char = code[-1]
-    if last_char.isdigit():
-        code += last_char
-
-    return code
-
-
 def create_name_and_label(kitsu_name: str) -> dict[str, str]:
     """From a name coming from kitsu, create a name and label"""
     name_slug = slugify(kitsu_name, separator="_")

tests/pytest.ini

@@ -1,2 +1,2 @@
 [pytest]  
-pythonpath = . ../services/processor
+pythonpath = ../server/kitsu ../services/processor

tests/tests/fixtures.py

@@ -16,6 +16,9 @@
 PROJECT_CODE = "TK"
 PAIR_PROJECT_NAME = "another_test_kitsu_project"
 PAIR_PROJECT_CODE = "ATK"
+USER1_NAME = "testkitsu.user1"
+USER2_NAME = "testkitsu.user2"
+USER3_NAME = "testkitsu.user3"
 
 PROJECT_META = {
     "code": PROJECT_CODE,
@@ -103,6 +106,29 @@ def init_data(api, kitsu_url):
     assert res.status_code == 200
 
 
+@pytest.fixture(scope="module")
+def users(api, kitsu_url):
+    """create ayon users"""
+    api.delete(f"/users/{USER1_NAME}")
+    api.delete(f"/users/{USER2_NAME}")
+    api.delete(f"/users/{USER3_NAME}")
+
+    # only create 2 users so the other one can be created if missing
+    api.put(f"/users/{USER1_NAME}")
+    api.put(f"/users/{USER2_NAME}")
+    print(f"created user: {USER1_NAME}")
+    print(f"created user: {USER2_NAME}")
+
+    yield
+
+    api.delete(f"/users/{USER1_NAME}")
+    api.delete(f"/users/{USER2_NAME}")
+    api.delete(f"/users/{USER3_NAME}")
+
+    # ensure renamed user is deleted
+    api.delete("/users/testkitsu.newusername")
+
+
 @pytest.fixture(scope="module")
 def gazu():
     # host = os.environ.get('KITSU_API_URL', 'http://localhost/api')
@@ -126,3 +152,80 @@ def get_paired_ayon_project(self, kitsu_project_id):
             return PROJECT_NAME
 
     return MockProcessor()
+
+
+# ======= Studio Settings Fixtures ==========
+
+
+@pytest.fixture()
+def users_enabled(api, kitsu_url):
+    """update kitsu addon settings.sync_settings.sync_users.enabled"""
+    # lets get the settings for the addon
+    res = api.get(f"{kitsu_url}/settings")
+    assert res.status_code == 200
+    settings = res.data
+
+    # get original values
+    users_enabled = settings["sync_settings"]["sync_users"]["enabled"]
+
+    # set settings for tests
+    if not users_enabled:
+        settings["sync_settings"]["sync_users"]["enabled"] = True
+        res = api.post(f"{kitsu_url}/settings", **settings)
+
+    yield
+
+    # set settings back to orginal values
+    if not users_enabled:
+        settings["sync_settings"]["sync_users"]["enabled"] = users_enabled
+        res = api.post(f"{kitsu_url}/settings", **settings)
+
+
+@pytest.fixture()
+def users_disabled(api, kitsu_url):
+    """update kitsu addon settings.sync_settings.sync_users.enabled"""
+    # lets get the settings for the addon
+    res = api.get(f"{kitsu_url}/settings")
+    assert res.status_code == 200
+    settings = res.data
+
+    # get original values
+    value = settings["sync_settings"]["sync_users"]["enabled"]
+    print(f"users_disabled: {value}")
+
+    # set settings for tests
+    if value:
+        settings["sync_settings"]["sync_users"]["enabled"] = False
+        res = api.post(f"{kitsu_url}/settings", **settings)
+
+    yield
+
+    # set settings back to orginal values
+    if value:
+        settings["sync_settings"]["sync_users"]["enabled"] = value
+        res = api.post(f"{kitsu_url}/settings", **settings)
+
+
+@pytest.fixture()
+def access_group(api, kitsu_url):
+    """update kitsu addon settings.sync_settings.sync_users.access_group"""
+
+    # lets get the settings for the addon
+    res = api.get(f"{kitsu_url}/settings")
+    assert res.status_code == 200
+    settings = res.data
+
+    # get original values
+    value = settings["sync_settings"]["sync_users"]["access_group"]
+
+    # set settings for tests
+    if value != "test_kitsu_group":
+        settings["sync_settings"]["sync_users"]["access_group"] = "test_kitsu_group"
+        res = api.post(f"{kitsu_url}/settings", **settings)
+
+    yield
+
+    # set settings back to orginal values
+    if value != "test_kitsu_group":
+        settings["sync_settings"]["sync_users"]["access_group"] = value
+        res = api.post(f"{kitsu_url}/settings", **settings)