feat(static): adds new static origins feature

app/brm/static.py

@@ -0,0 +1,77 @@
+from typing import Optional, Union
+
+from werkzeug.datastructures import FileStorage
+
+from app.extensions import db
+from app.models.base import Group
+from app.models.cloud import CloudAccount
+from app.models.mirrors import StaticOrigin
+
+
+def create_static_origin(
+        description: str,
+        group_id: int,
+        storage_cloud_account_id: int,
+        source_cloud_account_id: int,
+        source_project: str,
+        auto_rotate: bool,
+        matrix_homeserver: Optional[str],
+        keanu_convene_path: Optional[str],
+        keanu_convene_logo: Optional[FileStorage],
+        keanu_convene_color: Optional[str],
+        clean_insights_backend: Optional[Union[str, bool]],
+        db_session_commit: bool = False,
+) -> StaticOrigin:
+    """
+    Create a new static origin.
+
+    :param description: The description of the new static origin.
+    :param group_id: The ID for the origin group to add this new static origin to.
+    :param storage_cloud_account_id: The ID for the cloud account to deploy this new static origin to.
+    :param source_cloud_account_id: The ID for the cloud account used to interact with the web sources.
+    :param source_project: The path for the source project, e.g. the GitLab project path.
+    :param auto_rotate: Whether to automatically rotate this domain when it is detected to be blocked.
+    :param matrix_homeserver: The domain name for the Matrix homeserver to proxy to.
+    :param keanu_convene_path: The path to serve the Keanu Convene application from.
+    :param clean_insights_backend: The domain name for the Clean Insights backend to proxy to.
+    :param db_session_commit: Whether to add the new StaticOrigin to the database session and commit it.
+    :returns: StaticOrigin -- the newly created StaticOrigin
+    :raises: ValueError, sqlalchemy.exc.SQLAlchemyError
+    """
+    static_origin = StaticOrigin()
+    if isinstance(group_id, int):
+        group = Group.query.filter(Group.id == group_id).first()
+        if group is None:
+            raise ValueError("group_id must match an existing group")
+        static_origin.group_id = group_id
+    else:
+        raise ValueError("group_id must be an int")
+    if isinstance(storage_cloud_account_id, int):
+        cloud_account = CloudAccount.query.filter(CloudAccount.id == storage_cloud_account_id).first()
+        if cloud_account is None:
+            raise ValueError("storage_cloud_account_id must match an existing provider")
+        static_origin.storage_cloud_account_id = storage_cloud_account_id
+    else:
+        raise ValueError("storage_cloud_account_id must be an int")
+    if isinstance(source_cloud_account_id, int):
+        cloud_account = CloudAccount.query.filter(CloudAccount.id == source_cloud_account_id).first()
+        if cloud_account is None:
+            raise ValueError("source_cloud_account_id must match an existing provider")
+        static_origin.source_cloud_account_id = source_cloud_account_id
+    else:
+        raise ValueError("source_cloud_account_id must be an int")
+    static_origin.update(
+        source_project,
+        description,
+        auto_rotate,
+        matrix_homeserver,
+        keanu_convene_path,
+        keanu_convene_logo,
+        keanu_convene_color,
+        clean_insights_backend,
+        False
+    )
+    if db_session_commit:
+        db.session.add(static_origin)
+        db.session.commit()
+    return static_origin

app/brm/utils.py

@@ -1,6 +1,12 @@
 from __future__ import annotations
 
-from typing import Any
+import base64
+from io import BytesIO
+from typing import Any, Tuple
+
+import webcolors
+from PIL import Image
+from werkzeug.datastructures import FileStorage
 
 
 def is_integer(contender: Any) -> bool:
@@ -18,3 +24,68 @@ def is_integer(contender: Any) -> bool:
         return False
     else:
         return float(contender).is_integer()
+
+
+def thumbnail_uploaded_image(file: FileStorage, max_size: Tuple[int, int] = (256, 256)) -> bytes:
+    """
+    Process an uploaded image file into a resized image of a specific size.
+
+    :param file: An uploaded image file.
+    :param max_size: A tuple containing the maximum width and height for the thumbnail image. Default is (256, 256).
+    :return: The byte data of the thumbnail image.
+    """
+    if file.filename is None:
+        raise ValueError("No file was uploaded")
+    img = Image.open(file)
+    img.thumbnail(max_size)
+    byte_arr = BytesIO()
+    img.save(byte_arr, format='PNG' if file.filename.lower().endswith('.png') else 'JPEG')
+    return byte_arr.getvalue()
+
+
+def create_data_uri(bytes_data: bytes, file_extension: str) -> str:
+    """
+    Create a data URI from binary data and a file extension.
+
+    :param bytes_data: The binary data of an image.
+    :param file_extension: The file extension of the image.
+    :return: A data URI representing the image.
+    """
+    # base64 encode
+    encoded = base64.b64encode(bytes_data).decode('ascii')
+    # create data URI
+    data_uri = "data:image/{};base64,{}".format('jpeg' if file_extension == 'jpg' else file_extension, encoded)
+    return data_uri
+
+
+def normalize_color(color: str) -> str:
+    """
+    Normalize a string representing a color to its hexadecimal representation.
+
+    This function accepts a string representing a color in one of the following formats:
+
+    - A CSS color name, such as 'red', 'green', or 'blue'.
+    - A 6-digit hexadecimal color code, such as '#FF0000' for red.
+    - A 3-digit hexadecimal color code, such as '#F00' for red.
+
+    The function returns a string with the color in 6-digit or 3-digit hexadecimal format,
+    with lowercase letters. If the color is given as a CSS color name, the function
+    returns its equivalent 6-digit hexadecimal format.
+
+    :param color: A string representing a color.
+    :return: The color in 6-digit or 3-digit hexadecimal format.
+    :raises: ValueError: If the input string does not represent a valid color in any of the accepted formats.
+    """
+    try:
+        return webcolors.name_to_hex(color)  # type: ignore[no-any-return]
+    except ValueError:
+        pass
+    if color.startswith('#'):
+        color = color[1:].lower()
+    if len(color) in [3, 6]:
+        try:
+            _ = int(color, 16)
+            return f"#{color}"
+        except ValueError:
+            pass
+    raise ValueError(f"color must be a valid HTML color, got: {repr(color)}")