Develop

affiche-backend/affiche/alembic/versions/e1a3c5d7f9b2_center_gravity_uses_the_offset.py

@@ -0,0 +1,71 @@
+import json
+import logging
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = 'e1a3c5d7f9b2'
+down_revision = 'd5a9c3e7b1f4'
+branch_labels = None
+depends_on = None
+
+logger = logging.getLogger(__name__)
+
+CENTER_OFFSET = 0.5
+
+TABLES = (('library_settings', 'library_id'), ('style_profile', 'id'))
+
+def _repin(raw):
+    if not raw:
+        return None
+    try:
+        options = json.loads(raw) if isinstance(raw, str) else raw
+    except (TypeError, ValueError):
+        return None
+    if not isinstance(options, dict) or options.get('gravity') != 'center':
+        return None
+    if options.get('text_offset_ratio') == CENTER_OFFSET:
+        return None
+    options['text_offset_ratio'] = CENTER_OFFSET
+    return json.dumps(options)
+
+def _patch_tables(connection):
+    for table, key in TABLES:
+        rows = connection.execute(
+            sa.text(f"SELECT {key}, text_options FROM {table} WHERE text_options IS NOT NULL")
+        ).fetchall()
+        for row_key, raw in rows:
+            patched = _repin(raw)
+            if patched is not None:
+                connection.execute(
+                    sa.text(f"UPDATE {table} SET text_options = :opts WHERE {key} = :key"),
+                    {"opts": patched, "key": row_key},
+                )
+
+def _patch_defaults_file():
+    from affiche.config.env_config import POSTER_CONFIG_FILE
+    from pathlib import Path
+
+    path = Path(POSTER_CONFIG_FILE)
+    if not path.exists():
+        return
+    data = json.loads(path.read_text(encoding='utf-8'))
+    text_options = data.get('text_options')
+    if not isinstance(text_options, dict) or text_options.get('gravity') != 'center':
+        return
+    if text_options.get('text_offset_ratio') == CENTER_OFFSET:
+        return
+    text_options['text_offset_ratio'] = CENTER_OFFSET
+    path.write_text(json.dumps(data, indent=2), encoding='utf-8')
+
+def upgrade() -> None:
+    connection = op.get_bind()
+    _patch_tables(connection)
+    try:
+        _patch_defaults_file()
+    except Exception:
+        logger.warning("Could not pin the global style's center offset; set it by hand in "
+                       "Settings -> Style Options if the title has moved", exc_info=True)
+
+def downgrade() -> None:
+    pass

affiche-backend/affiche/api/schemas/settings_schema.py

@@ -10,6 +10,7 @@ class OverlayOptionsResponse(BaseModel):
     matte_height_ratio: float
     fade_height_ratio: float
     gradient_color: str
+    gradient_direction: Literal["bottom", "top", "left", "right"] = "bottom"
     vignette_strength: float
     vignette_color: str
     inner_glow_strength: float

affiche-backend/affiche/app/image/model/overlay_options.py

@@ -36,6 +36,7 @@ class OverlayOptions:
     matte_height_ratio: float = 0.0
     fade_height_ratio: float = 0.0
     gradient_color: str = "#000000"
+    gradient_direction: Literal["bottom", "top", "left", "right"] = "bottom"
 
     vignette_strength: float = 0.0
     vignette_color: str = "#000000"
@@ -58,6 +59,12 @@ def __post_init__(self):
         if self.overlay_type not in ("poster", "background"):
             raise ValueError(f"overlay_type must be 'poster' or 'background', got {self.overlay_type}")
 
+        if self.gradient_direction not in ("bottom", "top", "left", "right"):
+            raise ValueError(
+                f"gradient_direction must be bottom, top, left or right, "
+                f"got {self.gradient_direction}"
+            )
+
         self.border_color = validate_hex_color(self.border_color)
         self.gradient_color = validate_hex_color(self.gradient_color)
         self.vignette_color = validate_hex_color(self.vignette_color)

affiche-backend/affiche/app/image/overlay_generator.py

@@ -89,23 +89,31 @@ def _apply_gradient_matte(self, canvas: Image.Image, options: OverlayOptions) ->
         mask = Image.new("L", (width, height), 0)
         draw = ImageDraw.Draw(mask)
 
-        matte_h_px = int(height * options.matte_height_ratio)
-        fade_h_px = int(height * options.fade_height_ratio)
-
-        if matte_h_px > 0:
-            draw.rectangle(
-                [0, height - matte_h_px, width, height],
-                fill=255
-            )
-
-        if fade_h_px > 0:
-            start_y = height - matte_h_px - fade_h_px
-            for y in range(fade_h_px):
-                alpha = int(255 * (y / fade_h_px))
-                draw.line(
-                    [(0, start_y + y), (width, start_y + y)],
-                    fill=alpha
-                )
+        vertical = options.gradient_direction in ("bottom", "top")
+        extent = height if vertical else width
+
+        matte_px = int(extent * options.matte_height_ratio)
+        fade_px = int(extent * options.fade_height_ratio)
+
+        profile = [255] * min(matte_px, extent)
+        for step in range(fade_px):
+            if len(profile) >= extent:
+                break
+            profile.append(int(255 * (fade_px - 1 - step) / fade_px))
+
+        for offset, alpha in enumerate(profile):
+            if alpha <= 0:
+                continue
+            if options.gradient_direction == "bottom":
+                y = height - 1 - offset
+                draw.line([(0, y), (width, y)], fill=alpha)
+            elif options.gradient_direction == "top":
+                draw.line([(0, offset), (width, offset)], fill=alpha)
+            elif options.gradient_direction == "right":
+                x = width - 1 - offset
+                draw.line([(x, 0), (x, height)], fill=alpha)
+            else:
+                draw.line([(offset, 0), (offset, height)], fill=alpha)
 
         gradient_layer.putalpha(mask)
         return Image.alpha_composite(canvas, gradient_layer)

affiche-backend/affiche/app/image/text_renderer.py

@@ -234,8 +234,7 @@ def _calculate_text_position(
         elif options.gravity == "north":
             y = padding + text_offset
         else:
-            safe_height = img_height - (2 * padding)
-            y = padding + (safe_height - text_height) // 2
+            y = (img_height - text_offset) - text_height // 2
 
         x = max(padding, min(x, img_width - padding - text_width))
 

affiche-backend/affiche/app/mediaserver/library/seasons/connector/alchemy_library_season_connector.py

@@ -25,14 +25,18 @@ def update_seasons(self, seasons: List[LibrarySeason], state: SeasonPosterState)
 
     def rekey_seasons(self, adoptions: Mapping[int, str]) -> int:
         rekeyed = 0
-        for season_id, external_id in adoptions.items():
-            entity = self._session.get(LibrarySeasonEntity, season_id)
-            if entity is None:
-                continue
-            entity.external_id = external_id
-            entity.poster_hash = None
-            rekeyed += 1
-        self._session.commit()
+        try:
+            for season_id, external_id in adoptions.items():
+                entity = self._session.get(LibrarySeasonEntity, season_id)
+                if entity is None:
+                    continue
+                entity.external_id = external_id
+                entity.poster_hash = None
+                rekeyed += 1
+            self._session.commit()
+        except Exception:
+            self._session.rollback()
+            raise
         return rekeyed
 
     def create_or_update_seasons_batch(self, seasons: List[LibrarySeason]) -> None:

affiche-backend/affiche/app/mediaserver/library/sync/reidentification.py

@@ -63,13 +63,16 @@ def match_readded_items(existing: Sequence,
 
 def match_readded_seasons(existing: Sequence,
                           incoming: Mapping[int, str]) -> Dict[int, str]:
-    known = set(incoming.values())
+    held = {season.external_id for season in existing}
+    candidates: Dict[int, list] = {}
+    for season in existing:
+        target = incoming.get(season.season_number)
+        if target is not None and target not in held and season.external_id not in incoming.values():
+            candidates.setdefault(season.season_number, []).append(season)
     return {
-        season.id: incoming[season.season_number]
-        for season in existing
-        if season.season_number in incoming
-        and season.external_id != incoming[season.season_number]
-        and season.external_id not in known
+        rows[0].id: incoming[number]
+        for number, rows in candidates.items()
+        if len(rows) == 1
     }
 
 def match_split_items(existing: Sequence,

affiche-backend/requirements.txt

@@ -1,16 +1,24 @@
 fastapi>=0.115.0
 uvicorn[standard]>=0.27.0
 python-multipart>=0.0.6
+
 sqlalchemy>=2.0.25
 alembic>=1.13.1
+
 plexapi>=4.15.10
+
 requests>=2.31.0
+
 Pillow>=10.3.0
 numpy
+
 python-jose[cryptography]>=3.3.0
 bcrypt>=4.0.1
 python-dotenv>=1.0.1
+
 pydantic>=2.5.3
+
 tvdb-v4-official
+
 pytest
 httpx

affiche-backend/tests/test_item_reidentification.py

@@ -1,10 +1,13 @@
 from datetime import datetime, timedelta, timezone
 
 import pytest
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from affiche.app.mediaserver.connector.media_server_connector import MediaServerPersistenceConnector
-from affiche.app.mediaserver.library.model import Library, LibraryItem, LibraryItemSearch, LibrarySearch
+from affiche.app.mediaserver.library.model import Library, LibraryItem, LibraryItemSearch, LibrarySearch, LibrarySeason
+from affiche.app.mediaserver.library.seasons.library_season_repository import LibrarySeasonRepository
+from affiche.app.mediaserver.library.seasons.library_season_service import LibrarySeasonService
 from affiche.app.mediaserver.library.service.library_service import LibraryService
 from affiche.app.mediaserver.library.sync.reidentification import (
     RemoteIdentity,
@@ -99,8 +102,21 @@ def test_seasons_already_holding_the_new_id_are_left_alone(self):
 
         assert match_readded_seasons(existing, {1: "900"}) == {}
 
+    def test_seasons_whose_new_id_already_has_a_row_are_a_split_not_a_move(self):
+        existing = [_Season(10, "200", 1), _Season(20, "900", 1), _Season(11, "201", 2)]
+
+        assert match_readded_seasons(existing, {1: "900", 2: "901"}) == {11: "901"}
+
+    def test_two_stale_rows_for_one_season_number_match_neither(self):
+        existing = [_Season(10, "200", 1), _Season(11, "300", 1)]
+
+        assert match_readded_seasons(existing, {1: "900"}) == {}
+
 @pytest.fixture
 def library(session: Session):
+    return _create_library(session)
+
+def _create_library(session: Session) -> Library:
     server = MediaServerPersistenceConnector(session).create(MediaServer(
         name="Plex", type=MediaServerType.PLEX, url="http://x", token="t",
     ))
@@ -181,6 +197,42 @@ def test_a_row_already_in_the_trash_is_rescued(self, session: Session, library):
         assert restored == 1
         assert _only_item(service, library.id).deleted_at is None
 
+class TestSeasonAdoption:
+
+    def _show_with_split_seasons(self, session, library):
+        library_service = LibraryService(session)
+        library_service.create_or_update_items_batch([LibraryItem(
+            library_id=library.id, external_id="500", title="Severance", type="show",
+            tvdb_id="371980", last_seen_at=T0,
+        )])
+        session.flush()
+        show = _only_item(library_service, library.id)
+        seasons = LibrarySeasonService(session)
+        seasons.create_or_update([
+            LibrarySeason(show_id=show.id, library_id=library.id, external_id=external_id,
+                          season_number=number, title=f"Season {number}")
+            for external_id, number in (("200", 1), ("900", 1), ("201", 2), ("901", 2))
+        ])
+        return show, seasons
+
+    def test_a_split_show_is_left_as_it_is(self, session: Session, library):
+        show, seasons = self._show_with_split_seasons(session, library)
+
+        assert seasons.adopt_readded_seasons(library.id, show.id, {1: "900", 2: "901"}) == []
+        stored = seasons.get_item_seasons(library.id, show.id)
+        assert sorted(s.external_id for s in stored) == ["200", "201", "900", "901"]
+
+    def test_a_failed_rekey_leaves_the_session_usable(self, clean_session: Session):
+        session = clean_session
+        library = _create_library(session)
+        show, seasons = self._show_with_split_seasons(session, library)
+        stale = next(s for s in seasons.get_item_seasons(library.id, show.id) if s.external_id == "200")
+
+        with pytest.raises(IntegrityError):
+            LibrarySeasonRepository(session).rekey_seasons({stale.id: "900"})
+
+        seasons.get_item_seasons(library.id, show.id)
+
 class TestMerging:
 
     def _split(self, session, library, **fresh_state):

affiche-backend/tests/test_overlay_generator.py

@@ -103,6 +103,49 @@ def test_gradient_matte_applied(self):
         top_pixel = overlay.getpixel((1000, 100))
         assert top_pixel[3] == 0
 
+    def test_gradient_direction_defaults_to_bottom(self):
+        generator = OverlayGenerator()
+        options = OverlayOptions(matte_height_ratio=0.1, fade_height_ratio=0.2)
+
+        assert options.gradient_direction == "bottom"
+
+        overlay = generator.generate_overlay(options)
+
+        assert overlay.getpixel((1000, 2900))[3] == 255
+        assert overlay.getpixel((1000, 100))[3] == 0
+
+    @pytest.mark.parametrize("direction,solid,clear", [
+        ("bottom", (1000, 2950), (1000, 50)),
+        ("top", (1000, 50), (1000, 2950)),
+        ("left", (50, 1500), (1950, 1500)),
+        ("right", (1950, 1500), (50, 1500)),
+    ])
+    def test_gradient_grows_from_the_chosen_edge(self, direction, solid, clear):
+        generator = OverlayGenerator()
+        options = OverlayOptions(
+            matte_height_ratio=0.1,
+            fade_height_ratio=0.2,
+            gradient_direction=direction,
+        )
+
+        overlay = generator.generate_overlay(options)
+
+        assert overlay.getpixel(solid)[3] == 255
+        assert overlay.getpixel(clear)[3] == 0
+
+    def test_horizontal_gradient_measures_against_the_width(self):
+        generator = OverlayGenerator()
+        options = OverlayOptions(matte_height_ratio=0.1, gradient_direction="left")
+
+        overlay = generator.generate_overlay(options)
+
+        assert overlay.getpixel((199, 1500))[3] == 255
+        assert overlay.getpixel((201, 1500))[3] == 0
+
+    def test_rejects_an_unknown_direction(self):
+        with pytest.raises(ValueError, match="gradient_direction"):
+            OverlayOptions(gradient_direction="sideways")
+
     def test_inner_glow_applied(self):
         generator = OverlayGenerator()
         options = OverlayOptions(

affiche-backend/tests/test_text_renderer.py

@@ -213,6 +213,7 @@ def test_position_north_gravity(self):
     def test_position_center_gravity(self):
         options = TextOptions(
             gravity="center",
+            text_offset_ratio=0.5,
             border_padding_ratio=0,
         )
 
@@ -224,6 +225,17 @@ def test_position_center_gravity(self):
 
         assert y == 1450
 
+    def test_center_offset_measures_from_the_bottom(self):
+        high = TextOptions(gravity="center", text_offset_ratio=0.75, border_padding_ratio=0)
+        low = TextOptions(gravity="center", text_offset_ratio=0.25, border_padding_ratio=0)
+
+        _, y_high = self.renderer._calculate_text_position((2000, 3000), (500, 100), high)
+        _, y_low = self.renderer._calculate_text_position((2000, 3000), (500, 100), low)
+
+        assert y_high == 3000 - 2250 - 50
+        assert y_low == 3000 - 750 - 50
+        assert y_low > y_high
+
     def test_position_horizontal_centering(self):
         options = TextOptions(
             gravity="south",

affiche-frontend/index.html

@@ -6,6 +6,7 @@
     <meta name="viewport" content="width=device-width, initial-scale=1.0" />
     <title>Affiche</title>
     <script>
+
       (function () {
         try {
           var stored = localStorage.getItem('affiche.theme');