Make upload processing phase-based: MD5 -> visual -> IQDB over the batch

Uploads were doing md5 + local visual matching inside the upload request
(backend create) while the frontend later ran its own e621 MD5 pass, so the
pipeline looked interleaved per file. Now every step is a phase applied to
the whole batch in order:

1. upload (fast: md5 + exact-duplicate check only),
2. e621 MD5 lookup, 75 md5: metatags per posts.json request,
3. local visual similarity, one file at a time via the new
   POST /api/uploads/<id>/visual-match/ action,
4. IQDB through the existing serial queue.

The board shows the active phase with its own progress bar (e621 MD5 in
peach, visual in lavender, IQDB in teal) and every step updates the staged
list as it lands. Verified from a headless run: one batched posts.json
request for 10 files, then 10 visual-match calls, then IQDB.

backend/apps/library/tests/test_uploads.py

@@ -1,7 +1,8 @@
-"""Staged uploads: complete-set listing, bulk rating and IQDB recording."""
+"""Staged uploads: listing, phases (MD5/visual/IQDB) and the bulk tool."""
 
 import base64
 import hashlib
+import io
 import json
 import shutil
 import tempfile
@@ -11,6 +12,8 @@ from django.contrib.auth import get_user_model
 from django.core.files.uploadedfile import SimpleUploadedFile
 from django.test import Client, TestCase, override_settings
 
+from PIL import Image
+
 from rest_framework.authtoken.models import Token
 
 from apps.library.models import MediaItem, TempUpload
@@ -23,6 +26,13 @@ TINY_PNG = base64.b64decode(
 )
 
 
+def similar_png(color=(200, 30, 40)):
+    """A 1x1 PNG with the same visual hashes but a different MD5."""
+    buffer = io.BytesIO()
+    Image.new("RGB", (1, 1), color).save(buffer, format="PNG")
+    return buffer.getvalue()
+
+
 def jpost(client, path, body=None):
     return client.post(path, data=json.dumps(body or {}), content_type="application/json")
 
@@ -85,7 +95,7 @@ class TempUploadListTests(TestCase):
         self.assertEqual(Client().get("/api/uploads/").status_code, 401)
 
 
-class BulkResolveTests(TestCase):
+class StagedUploadWorkflowTests(TestCase):
     @classmethod
     def setUpClass(cls):
         super().setUpClass()
@@ -196,6 +206,61 @@ class BulkResolveTests(TestCase):
         self.assertEqual(body["errors"][0]["error"], "already in the library")
         self.assertEqual(MediaItem.objects.count(), 1)
 
+    def upload_via_api(self, client, label="phase.png"):
+        return client.post(
+            "/api/uploads/",
+            {
+                "file": SimpleUploadedFile(
+                    label, similar_png(), content_type="image/png"
+                )
+            },
+        )
+
+    def seed_library_item(self, client, label):
+        seed = self.make_temp(self.uploader, label)
+        response = self.resolve_bulk(client, [seed.id], "s")
+        self.assertEqual(response.status_code, 200)
+        return seed
+
+    def test_upload_defers_visual_similarity_to_its_phase(self):
+        client = self.api_client(self.uploader)
+        self.seed_library_item(client, "seed-defer")
+        response = self.upload_via_api(client)
+        self.assertEqual(response.status_code, 201)
+        body = response.json()
+        # Uploading must not run the expensive library scan.
+        self.assertEqual(body["status"], TempUpload.STATUS_PENDING)
+        self.assertFalse(body["visual_matches"])
+
+    def test_visual_match_phase_flags_similar_library_items(self):
+        client = self.api_client(self.uploader)
+        self.seed_library_item(client, "seed-visual")
+        temp = self.make_temp(self.uploader, "check-visual")
+        response = client.post(f"/api/uploads/{temp.id}/visual-match/")
+        self.assertEqual(response.status_code, 200)
+        body = response.json()
+        self.assertEqual(body["status"], TempUpload.STATUS_VISUAL_MATCH)
+        self.assertGreaterEqual(len(body["visual_matches"]), 1)
+        temp.refresh_from_db()
+        self.assertEqual(temp.status, TempUpload.STATUS_VISUAL_MATCH)
+
+    def test_visual_match_phase_with_empty_library_stays_pending(self):
+        client = self.api_client(self.uploader)
+        temp = self.make_temp(self.uploader, "no-matches")
+        response = client.post(f"/api/uploads/{temp.id}/visual-match/")
+        self.assertEqual(response.status_code, 200)
+        body = response.json()
+        self.assertEqual(body["visual_matches"], [])
+        self.assertEqual(body["status"], TempUpload.STATUS_PENDING)
+
+    def test_visual_match_phase_rejects_completed_uploads(self):
+        client = self.api_client(self.uploader)
+        temp = self.make_temp(
+            self.uploader, "done-visual", status=TempUpload.STATUS_COMPLETED
+        )
+        response = client.post(f"/api/uploads/{temp.id}/visual-match/")
+        self.assertEqual(response.status_code, 400)
+
 
 class IqdbRecordingTests(TestCase):
     """The modal needs to tell "checked, no match" from "never checked"."""