feat(artist): move a source into another artist (#130 step 4)

Operator ask: a surface to merge new sources into existing artists (consolidate
the singleton artist a fresh add spins up). Enabled by the #130 slug decoupling —
the storage path is immutable, so re-attribution moves NO files. SourceService
.reassign moves the source, re-points its posts (Post.source_id==S) and the
images it contributed (ImageProvenance via S, scoped to the old artist so shared
images aren't stolen), and deletes the old artist if it's left fully empty (else
clears its subscription flag). POST /api/sources/<id>/reassign. Frontend: a
'Move…' action per source on the artist Management tab → artist-autocomplete
picker → confirm → routes to the target (whose slug is stable).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM

backend/app/api/sources.py

@@ -136,6 +136,25 @@ async def delete_source(source_id: int):
     return "", 204
 
 
+@sources_bp.route("/<int:source_id>/reassign", methods=["POST"])
+async def reassign_source(source_id: int):
+    """Move this source (and the content it brought in) to another artist
+    (#130). Files don't move — the slug is immutable — so this just re-attributes
+    the source, its posts, and its images. Body: {target_artist_id}."""
+    body = await request.get_json(silent=True) or {}
+    target = body.get("target_artist_id")
+    if not isinstance(target, int):
+        return _bad("invalid_body", detail="target_artist_id (int) required")
+    async with get_session() as session:
+        try:
+            record = await SourceService(session).reassign(source_id, target)
+        except LookupError:
+            return _bad("not_found", status=404)
+        except ArtistNotFoundError:
+            return _bad("artist_not_found", detail="target artist not found", status=404)
+    return jsonify(record.to_dict())
+
+
 @sources_bp.route("/<int:source_id>/backfill", methods=["POST"])
 async def set_backfill(source_id: int):
     """Plan #693/#697 + #830: start/stop a backfill, or start a recovery /

backend/app/services/source_service.py

@@ -4,11 +4,18 @@ is_subscription auto-flip on first add / last delete.
 
 from dataclasses import dataclass
 
-from sqlalchemy import func, select
+from sqlalchemy import func, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from ..models import Artist, ImportSettings, Source
+from ..models import (
+    Artist,
+    ImageProvenance,
+    ImageRecord,
+    ImportSettings,
+    Post,
+    Source,
+)
 from .platforms import known_platform_keys
 from .scheduler_service import compute_next_check_at
 
@@ -420,6 +427,70 @@ class SourceService:
         await self.session.commit()
         return await self._row_to_record(source)
 
+    async def reassign(self, source_id: int, target_artist_id: int) -> SourceRecord:
+        """Move a Source — and the content it brought in — to another artist
+        (#130). The slug/storage path is IMMUTABLE, so NO files move: only the
+        artist attribution changes (reads use ImageRecord.path). Re-points the
+        source's Posts and the ImageRecords it contributed (those still
+        attributed to the old artist — images shared with another artist are
+        left alone). If the old artist is left fully empty (no sources, images,
+        or posts) it's deleted (ArtistVisit cascades); if it just lost its last
+        source, its is_subscription flag clears. No-op when already on target."""
+        source = (await self.session.execute(
+            select(Source).where(Source.id == source_id)
+        )).scalar_one_or_none()
+        if source is None:
+            raise LookupError(f"source id={source_id} not found")
+        target = await self._artist_or_raise(target_artist_id)
+        old_artist_id = source.artist_id
+        if old_artist_id == target_artist_id:
+            return await self._row_to_record(source)
+
+        source.artist_id = target_artist_id
+        target.is_subscription = True
+        # Re-attribute this source's posts (Post.artist_id is denormalized).
+        await self.session.execute(
+            update(Post).where(Post.source_id == source_id)
+            .values(artist_id=target_artist_id)
+        )
+        # Re-attribute the images this source contributed that are still on the
+        # OLD artist. Scoping to artist_id == old avoids stealing an image that
+        # a different artist's source also contributed.
+        contributed = select(ImageProvenance.image_record_id).where(
+            ImageProvenance.source_id == source_id
+        )
+        await self.session.execute(
+            update(ImageRecord)
+            .where(
+                ImageRecord.artist_id == old_artist_id,
+                ImageRecord.id.in_(contributed),
+            )
+            .values(artist_id=target_artist_id)
+        )
+        await self.session.flush()
+
+        old = (await self.session.execute(
+            select(Artist).where(Artist.id == old_artist_id)
+        )).scalar_one_or_none()
+        if old is not None:
+            n_src = (await self.session.execute(
+                select(func.count(Source.id)).where(Source.artist_id == old_artist_id)
+            )).scalar_one()
+            n_img = (await self.session.execute(
+                select(func.count(ImageRecord.id)).where(
+                    ImageRecord.artist_id == old_artist_id
+                )
+            )).scalar_one()
+            n_post = (await self.session.execute(
+                select(func.count(Post.id)).where(Post.artist_id == old_artist_id)
+            )).scalar_one()
+            if n_src == 0 and n_img == 0 and n_post == 0:
+                await self.session.delete(old)      # ArtistVisit cascades
+            elif n_src == 0:
+                old.is_subscription = False
+        await self.session.commit()
+        return await self._row_to_record(source)
+
     async def delete(self, source_id: int) -> None:
         source = (await self.session.execute(
             select(Source).where(Source.id == source_id)