Continued:

[fba.git] / fba / models / instances.py

diff --git a/fba/models/instances.py b/fba/models/instances.py
index 7968ff236007aa4d086266dca30a4b001cd5ed06..1b8450499fbb7594c18ee738fef5fe2279751d68 100644 (file)
--- a/fba/models/instances.py
+++ b/fba/models/instances.py
@@ -270,9 +270,13 @@ def set_success(domain: str):
 
     logger.debug("EXIT!")
 
-def is_registered(domain: str) -> bool:
-    logger.debug("domain='%s' - CALLED!", domain)
-    domain_helper.raise_on(domain)
+def is_registered(domain: str, skip_raise = False) -> bool:
+    logger.debug("domain='%s',skip_raise='%s' - CALLED!", domain, skip_raise)
+    if not isinstance(skip_raise, bool):
+        raise ValueError(f"skip_raise[]='%s' is not type of 'bool'", type(skip_raise))
+
+    if not skip_raise:
+        domain_helper.raise_on(domain)
 
     logger.debug("domain='%s' - CALLED!", domain)
     if not cache.key_exists("is_registered"):
@@ -474,3 +478,36 @@ def valid(value: str, column: str) -> bool:
 
     logger.debug("valid='%s' - EXIT!", valid)
     return valid
+
+def translate_idnas(rows: list, column: str):
+    logger.debug("rows[]='%s' - CALLED!", type(rows))
+    if not isinstance(rows, list):
+        raise ValueError(f"rows[]='%s' is not of type 'list'", type(rows))
+    elif len(rows) == 0:
+        raise ValueError("Parameter 'rows' is an empty list")
+    elif not isinstance(column, str):
+        raise ValueError(f"column='%s' is not of type 'str'", type(column))
+    elif column == "":
+        raise ValueError("Parameter 'column' is empty")
+    elif column not in ["domain", "origin"]:
+        raise ValueError(f"column='{column}' is not supported")
+
+    logger.info("Checking/converting %d domain names ...", len(rows))
+    for row in rows:
+        logger.debug("row[]='%s'", type(row))
+
+        translated = row[column].encode("idna").decode("utf-8")
+        logger.debug("translated='%s',row[%s]='%s'", translated, column, row[column])
+
+        if translated != row[column]:
+            logger.info("Translated row[%s]='%s' to '%s'", column, row[column], translated)
+            if is_registered(translated, True):
+                logger.warning("Deleting row[%s]='%s' as translated='%s' already exist", column, row[column], translated)
+                database.cursor.execute(f"DELETE FROM instances WHERE {column} = ? LIMIT 1", [row[column]])
+            else:
+                database.cursor.execute(f"UPDATE instances SET {column} = ? WHERE {column} = ? LIMIT 1", [translated, row[column]])
+
+            logger.debug("Invoking commit() ...")
+            database.connection.commit()
+
+    logger.debug("EXIT!")