Continued:

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

diff --git a/fba/models/blocks.py b/fba/models/blocks.py
index ec4fed2e03a20f1b6361cbae3520b91d4376eaab..1a861c5ea3f7981e29bfa502bc41f8a623934f1c 100644 (file)
--- a/fba/models/blocks.py
+++ b/fba/models/blocks.py
@@ -30,24 +30,24 @@ logger = logging.getLogger(__name__)
 def update_reason(reason: str, blocker: str, blocked: str, block_level: str):
     logger.debug("reason='%s',blocker='%s',blocked='%s',block_level='%s' - CALLED!", reason, blocker, blocked, block_level)
     if not isinstance(reason, str) and reason is not None:
-        raise ValueError(f"Parameter reason[]='{type(reason)}' is not 'str'")
+        raise ValueError(f"Parameter reason[]='{type(reason)}' is not of type 'str'")
     elif not isinstance(blocker, str):
-        raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not 'str'")
+        raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not of type 'str'")
     elif blocker == "":
         raise ValueError("Parameter 'blocker' is empty")
     elif blocker.lower() != blocker:
         raise ValueError(f"Parameter blocker='{blocker}' must be all lower-case")
     elif not isinstance(blocked, str):
-        raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not 'str'")
+        raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not of type 'str'")
     elif blocked == "":
         raise ValueError("Parameter 'blocked' is empty")
     elif blocked.lower() != blocked:
         raise ValueError(f"Parameter blocked='{blocked}' must be all lower-case")
     elif not isinstance(block_level, str):
-        raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not 'str'")
+        raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
     elif block_level == "":
         raise ValueError("Parameter 'block_level' is empty")
-    elif block_level in ["accept", "suspend", "silence"]:
+    elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
         raise ValueError(f"block_level='{block_level}' is not wanted.")
 
     logger.debug("Updating block reason='%s',blocker='%s',blocked='%s',block_level='%s'", reason, blocker, blocked, block_level)
@@ -66,22 +66,22 @@ def update_reason(reason: str, blocker: str, blocked: str, block_level: str):
 def update_last_seen(blocker: str, blocked: str, block_level: str):
     logger.debug("blocker='%s',blocked='%s',block_level='%s' - CALLED!", blocker, blocked, block_level)
     if not isinstance(blocker, str):
-        raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not 'str'")
+        raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not of type 'str'")
     elif blocker == "":
         raise ValueError("Parameter 'blocker' is empty")
     elif blocker.lower() != blocker:
         raise ValueError(f"Parameter blocker='{blocker}' must be all lower-case")
     elif not isinstance(blocked, str):
-        raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not 'str'")
+        raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not of type 'str'")
     elif blocked == "":
         raise ValueError("Parameter 'blocked' is empty")
     elif blocked.lower() != blocked:
         raise ValueError(f"Parameter blocked='{blocked}' must be all lower-case")
     elif not isinstance(block_level, str):
-        raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not 'str'")
+        raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
     elif block_level == "":
         raise ValueError("Parameter 'block_level' is empty")
-    elif block_level in ["accept", "suspend", "silence"]:
+    elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
         raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
 
     database.cursor.execute(
@@ -113,7 +113,7 @@ def is_instance_blocked(blocker: str, blocked: str, block_level: str = None) ->
         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
     elif block_level == "":
         raise ValueError("Parameter 'block_level' is empty")
-    elif block_level in ["accept", "suspend", "silence"]:
+    elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
         raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
 
     if block_level is None:
@@ -152,7 +152,7 @@ def add_instance(blocker: str, blocked: str, reason: str, block_level: str):
         raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
     elif blacklist.is_blacklisted(blocked):
         raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
-    elif block_level in ["accept", "suspend", "silence"]:
+    elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
         raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
 
     if reason is not None:
@@ -174,19 +174,75 @@ def add_instance(blocker: str, blocked: str, reason: str, block_level: str):
 
     logger.debug("EXIT!")
 
-def is_valid_level(block_level: str) -> bool:
-    logger.debug("block_level='%s' - CALLED!", block_level)
-    if not isinstance(block_level, str):
-        raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
-    elif block_level == "":
-        raise ValueError("Parameter 'block_level' is empty")
+def valid(value: str, column: str) -> bool:
+    logger.debug("value='%s' - CALLED!", value)
+    if not isinstance(value, str):
+        raise ValueError(f"Parameter value[]='{type(value)}' is not of type 'str'")
+    elif value == "":
+        raise ValueError("Parameter 'value' is empty")
+    elif not isinstance(column, str):
+        raise ValueError(f"Parameter column[]='{type(column)}' is not of type 'str'")
+    elif column == "":
+        raise ValueError("Parameter 'column' is empty")
 
     # Query database
     database.cursor.execute(
-        "SELECT block_level FROM blocks WHERE block_level = ? LIMIT 1", [block_level]
+        f"SELECT {column} FROM blocks WHERE {column} = ? LIMIT 1", [value]
     )
 
-    valid = database.cursor.fetchone() is not None
+    is_valid = database.cursor.fetchone() is not None
+
+    logger.debug("is_valid='%s' - EXIT!", is_valid)
+    return is_valid
+
+def translate_idnas(rows: list, column: str):
+    logger.debug("rows[]='%s' - CALLED!", type(rows))
+    if not isinstance(rows, list):
+        raise ValueError(f"rows[]='{type(rows)}' is not of type 'list'")
+    elif len(rows) == 0:
+        raise ValueError("Parameter 'rows' is an empty list")
+    elif not isinstance(column, str):
+        raise ValueError(f"column='{type(column)}' is not of type 'str'")
+    elif column == "":
+        raise ValueError("Parameter 'column' is empty")
+    elif column not in ["blocker", "blocked"]:
+        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)
+            database.cursor.execute(f"UPDATE blocks SET {column} = ? WHERE {column} = ?", [translated, row[column]])
+
+            logger.debug("Invoking commit() ...")
+            database.connection.commit()
+
+    logger.debug("EXIT!")
+
+def alias_block_level(block_level: str) -> str:
+    logger.debug("block_level='%s' - CALLED!", block_level)
+    if not isinstance(block_level, str):
+        raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
+    elif block_level == "":
+        raise ValueError("Parameter 'block_level' is empty")
 
-    logger.debug("valid='%s' - EXIT!", valid)
-    return valid
+    if block_level == "silence":
+        logger.debug("Block level 'silence' has been changed to 'silenced'")
+        block_level = "silenced"
+    elif block_level == "suspend":
+        logger.debug("Block level 'suspend' has been changed to 'suspended'")
+        block_level = "suspended"
+    elif block_level == "nsfw":
+        logger.debug("Block level 'nsfw' has been changed to 'media_nsfw'")
+        block_level = "media_nsfw"
+    elif block_level == "quarantined_instances":
+        logger.debug("Block level 'quarantined_instances' has been changed to 'quarantined'")
+        block_level = "quarantined"
+
+    logger.debug("block_level='%s' - EXIT!", block_level)
+    return block_level