feat: add Base64 encoding/decoding for media data (#102)

Author: leodiegues

src/mosaico/media.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
+import base64
+import binascii
 import contextlib
 import io
 import mimetypes
+import re
 import uuid
 from collections.abc import Generator
 from typing import IO, Any, cast
@@ -11,14 +14,18 @@
 from pydantic import BaseModel
 from pydantic.config import ConfigDict
 from pydantic.fields import Field
-from pydantic.functional_validators import model_validator
+from pydantic.functional_serializers import field_serializer
+from pydantic.functional_validators import field_validator, model_validator
 from typing_extensions import Self
 
 from mosaico.config import settings
 from mosaico.integrations.base.adapters import Adapter
 from mosaico.types import FilePath
 
 
+_BASE64_BYTE_PATTERN = re.compile(rb"^[A-Za-z0-9+/]+={0,2}$")
+
+
 class Media(BaseModel):
     """
     Represents a media object.
@@ -72,6 +79,34 @@ def validate_media(cls, values: dict[str, Any]) -> Any:
 
         return values
 
+    @field_validator("data")
+    @classmethod
+    def decode_base64_data(cls, v: bytes | str | None) -> bytes | str | None:
+        """
+        Decode field data from Base64 only if it looks like Base64.
+        """
+        if isinstance(v, str):
+            try:
+                raw = v.encode("ascii")
+            except UnicodeEncodeError:
+                return v
+
+            if len(raw) % 4 == 0 and _BASE64_BYTE_PATTERN.match(raw) is not None and raw.endswith(b"="):
+                try:
+                    return base64.b64decode(raw, validate=True)
+                except binascii.Error:
+                    pass
+        return v
+
+    @field_serializer("data", when_used="json")
+    def encode_base64_data(self, v) -> str:
+        """
+        Codifica campo data em base64.
+        """
+        if isinstance(v, bytes):
+            return base64.b64encode(v).decode(self.encoding)
+        return v
+
     @classmethod
     def from_path(
         cls,

src/mosaico/transcription_aligners/sequence_matcher.py

@@ -47,7 +47,7 @@ def align(self, transcription: Transcription, original_text: str) -> Transcripti
                 # No change needed, keep original words and time ranges
                 logger.debug("No change needed, keeping original words and time ranges.")
                 for word in words_with_time_ranges[i1:i2]:
-                    fixed_words.append(word.copy())
+                    fixed_words.append(word.model_copy())
 
             elif tag == "replace":
                 # Replace words but adapt their time ranges

tests/test_media.py

@@ -1,3 +1,5 @@
+import base64
+
 import pytest
 from pydantic import ValidationError
 
@@ -27,6 +29,69 @@ def test_creation_with_path():
     assert media.metadata == {}
 
 
+def test_decode_base64_str_to_bytes():
+    original = b"hello world \xf0\x9f\x8c\x8d"
+    b64 = base64.b64encode(original).decode("ascii")
+    m = Media(data=b64, path=None)
+    assert isinstance(m.data, bytes)
+    assert m.data == original
+
+
+def test_encode_bytes_to_base64_json():
+    original = b"foo bar baz"
+    m = Media(data=original, path=None)
+    m_dict = m.model_dump(mode="json")
+    assert isinstance(m_dict["data"], str)
+    decoded = base64.b64decode(m_dict["data"])
+    assert decoded == original
+
+
+def test_round_trip_via_model_dump_and_parse():
+    original = b"\x00\x01\x02binary\xff"
+    b64 = base64.b64encode(original).decode("ascii")
+    m1 = Media.from_data(b64)
+    json_str = m1.model_dump_json()
+    m2 = Media.model_validate_json(json_str)
+    assert isinstance(m2.data, bytes)
+    assert m2.data == original
+
+
+def test_non_base64_string_pass_through():
+    raw = "not base64!!!"
+    m = Media(data=raw)
+    assert isinstance(m.data, str)
+    assert m.data == raw
+
+
+def test_non_ascii_string_pass_through():
+    raw = "hello 😊"
+    m = Media(data=raw)
+    assert isinstance(m.data, str)
+    assert m.data == raw
+
+
+def test_length_not_multiple_of_four_pass_through():
+    raw = "YWxsb3c"
+    assert len(raw) % 4 != 0
+    m = Media(data=raw)
+    assert isinstance(m.data, str)
+    assert m.data == raw
+
+
+def test_empty_string_pass_through():
+    raw = ""
+    m = Media(data=raw)
+    assert isinstance(m.data, str)
+    assert m.data == raw
+
+
+def test_bytes_input_unchanged():
+    original = b"abcd"
+    m = Media(data=original)
+    assert isinstance(m.data, bytes)
+    assert m.data == original
+
+
 def test_creation_with_mime_type():
     media = Media(data="test content", mime_type="text/plain")
     assert media.data == "test content"