backend/app/domains/products/repo.py

@@ -331,16 +331,29 @@ def _run_dict(row: tuple) -> dict:
     }
 
 
-def list_runs(
-    conn: psycopg.Connection, storefront_id: int, limit: int, offset: int
-) -> list[dict]:
+def list_runs(conn: psycopg.Connection, storefront_id: int, limit: int, offset: int) -> list[dict]:
     rows = conn.execute(
-        _RUN_SELECT
-        + " WHERE r.storefront_id = %s ORDER BY r.created_at DESC, r.id DESC"
+        _RUN_SELECT + " WHERE r.storefront_id = %s ORDER BY r.created_at DESC, r.id DESC"
         " LIMIT %s OFFSET %s",
         (storefront_id, limit, offset),
     ).fetchall()
-    return [_run_dict(row) for row in rows]
+    runs = [_run_dict(row) for row in rows]
+    if not runs:
+        return runs
+    ids = [r["id"] for r in runs]
+    by_run = {rid: {"fetched": 0, "rejected": 0, "failed": 0, "total": 0} for rid in ids}
+    for rid, fetched, rejected, failed, total in conn.execute(
+        "SELECT import_run_id,"
+        " count(*) FILTER (WHERE status='fetched'),"
+        " count(*) FILTER (WHERE status IN ('rejected_low_res','rejected_not_image')),"
+        " count(*) FILTER (WHERE status='failed'), count(*)"
+        " FROM product_image WHERE import_run_id = ANY(%s) GROUP BY import_run_id",
+        (ids,),
+    ):
+        by_run[rid] = {"fetched": fetched, "rejected": rejected, "failed": failed, "total": total}
+    for r in runs:
+        r["image_counts"] = by_run[r["id"]]
+    return runs
 
 
 def get_run(conn: psycopg.Connection, storefront_id: int, run_id: int) -> dict | None:
@@ -359,12 +372,105 @@ def get_run(conn: psycopg.Connection, storefront_id: int, run_id: int) -> dict |
             (run_id,),
         )
     ]
-    # SLICE-7 fills these; the §6.4 payload shape is stable from SLICE-5 on.
-    run["image_progress"] = {"done": 0, "total": 0}
-    run["image_outcomes"] = []
+    counts = run_image_counts(conn, run_id)
+    run["image_progress"] = {"done": counts["total"] - counts["pending"], "total": counts["total"]}
+    run["image_counts"] = {"fetched": counts["fetched"], "rejected": counts["rejected"],
+                           "failed": counts["failed"]}
+    run["image_outcomes"] = run_image_outcomes(conn, run_id)
     return run
 
 
+# ---------------------------------------------------------------------------
+# Image fetch phase (SLICE-7, SD-0002 §6.5.4) — claim/mark/count/outcomes.
+# ---------------------------------------------------------------------------
+
+
+def pending_images_for_run(conn: psycopg.Connection, run_id: int) -> list[dict]:
+    rows = conn.execute(
+        "SELECT i.id, i.source_url, p.storefront_id, i.product_id"
+        " FROM product_image i JOIN product p ON p.id = i.product_id"
+        " WHERE i.import_run_id = %s AND i.status = 'pending' ORDER BY i.id",
+        (run_id,),
+    ).fetchall()
+    return [{"id": r[0], "source_url": r[1], "storefront_id": r[2], "product_id": r[3]} for r in rows]
+
+
+def claim_image_for_fetch(conn: psycopg.Connection, image_id: int) -> bool:
+    row = conn.execute(
+        "UPDATE product_image SET status = 'pending' WHERE id = %s AND status = 'pending' RETURNING id",
+        (image_id,),
+    ).fetchone()
+    return row is not None
+
+
+def mark_image_fetched(conn: psycopg.Connection, image_id: int, keys: dict[str, str]) -> None:
+    conn.execute(
+        "UPDATE product_image SET status = 'fetched', failure_reason = NULL,"
+        " key_original = %s, key_thumb = %s, key_card = %s, key_detail = %s,"
+        " fetched_at = now() WHERE id = %s",
+        (keys["original"], keys["thumb"], keys["card"], keys["detail"], image_id),
+    )
+
+
+def mark_image_rejected(conn: psycopg.Connection, image_id: int, status: str, reason: str) -> None:
+    conn.execute(
+        "UPDATE product_image SET status = %s, failure_reason = %s, fetched_at = now() WHERE id = %s",
+        (status, reason, image_id),
+    )
+
+
+def mark_image_failed(conn: psycopg.Connection, image_id: int, reason: str) -> None:
+    conn.execute(
+        "UPDATE product_image SET status = 'failed', failure_reason = %s, fetched_at = now() WHERE id = %s",
+        (reason, image_id),
+    )
+
+
+def run_image_counts(conn: psycopg.Connection, run_id: int) -> dict:
+    row = conn.execute(
+        "SELECT count(*) FILTER (WHERE status = 'fetched'),"
+        " count(*) FILTER (WHERE status IN ('rejected_low_res', 'rejected_not_image')),"
+        " count(*) FILTER (WHERE status = 'failed'),"
+        " count(*) FILTER (WHERE status = 'pending'), count(*)"
+        " FROM product_image WHERE import_run_id = %s",
+        (run_id,),
+    ).fetchone()
+    return {"fetched": row[0], "rejected": row[1], "failed": row[2], "pending": row[3], "total": row[4]}
+
+
+def set_run_status(conn: psycopg.Connection, run_id: int, status: str) -> None:
+    conn.execute(
+        "UPDATE import_run SET status = %s,"
+        " completed_at = CASE WHEN %s THEN now() ELSE completed_at END WHERE id = %s",
+        (status, status in _TERMINAL_RUN_STATUSES, run_id),
+    )
+
+
+def incomplete_runs(conn: psycopg.Connection) -> list[dict]:
+    rows = conn.execute(
+        "SELECT id, storefront_id FROM import_run WHERE status = 'fetching_images' ORDER BY id",
+    ).fetchall()
+    return [{"id": r[0], "storefront_id": r[1]} for r in rows]
+
+
+def run_image_outcomes(conn: psycopg.Connection, run_id: int) -> list[dict]:
+    rows = conn.execute(
+        "SELECT p.handle, i.source_url, i.status, i.failure_reason,"
+        " v.option1_value, v.option2_value, v.option3_value"
+        " FROM product_image i JOIN product p ON p.id = i.product_id"
+        " LEFT JOIN variant v ON v.image_id = i.id"
+        " WHERE i.import_run_id = %s"
+        " AND i.status IN ('rejected_low_res', 'rejected_not_image', 'failed')"
+        " ORDER BY p.handle, i.position, i.id",
+        (run_id,),
+    ).fetchall()
+    return [
+        {"handle": r[0], "url": r[1], "outcome": r[2], "reason": r[3],
+         "variant": " / ".join(x for x in (r[4], r[5], r[6]) if x) or None}
+        for r in rows
+    ]
+
+
 # ---------------------------------------------------------------------------
 # Apply primitives — called inside the confirm transaction (Task 8); no commits.
 # ---------------------------------------------------------------------------

backend/tests/test_products_repo_images.py

@@ -0,0 +1,93 @@
+"""products repo — image-phase helpers (SD-0002 §6.5.4)."""
+import psycopg
+import pytest
+
+from app.domains.products import repo
+from app.platform import db
+
+
+@pytest.fixture()
+def migrated_conn(fresh_db_url):
+    with psycopg.connect(fresh_db_url) as conn:
+        db.migrate(conn)
+        yield conn
+
+
+@pytest.fixture()
+def merchant(migrated_conn):
+    acct = migrated_conn.execute(
+        "INSERT INTO account (email) VALUES ('m@example.com') RETURNING id").fetchone()[0]
+    sf = migrated_conn.execute(
+        "INSERT INTO storefront (name) VALUES ('Shop') RETURNING id").fetchone()[0]
+    migrated_conn.execute(
+        "INSERT INTO storefront_membership (account_id, storefront_id) VALUES (%s,%s)", (acct, sf))
+    migrated_conn.commit()
+    return {"account_id": acct, "storefront_id": sf}
+
+
+def _run(conn, m, status="fetching_images"):
+    return conn.execute(
+        "INSERT INTO import_run (storefront_id, account_id, file_name, dialect,"
+        " products_added, products_updated, rows_errored, status)"
+        " VALUES (%s,%s,'c.csv','canonical',0,0,0,%s) RETURNING id",
+        (m["storefront_id"], m["account_id"], status)).fetchone()[0]
+
+
+def _product(conn, m, handle):
+    return conn.execute(
+        "INSERT INTO product (storefront_id, handle, title) VALUES (%s,%s,%s) RETURNING id",
+        (m["storefront_id"], handle, handle.title())).fetchone()[0]
+
+
+def _image(conn, product_id, url, run_id, status="pending", position=1):
+    return conn.execute(
+        "INSERT INTO product_image (product_id, source_url, position, status, import_run_id)"
+        " VALUES (%s,%s,%s,%s,%s) RETURNING id",
+        (product_id, url, position, status, run_id)).fetchone()[0]
+
+
+def test_pending_images_for_run_returns_only_pending(migrated_conn, merchant):
+    rid = _run(migrated_conn, merchant)
+    p = _product(migrated_conn, merchant, "lamp")
+    a = _image(migrated_conn, p, "https://m/a.png", rid, status="pending", position=1)
+    _image(migrated_conn, p, "https://m/b.png", rid, status="fetched", position=2)
+    pend = repo.pending_images_for_run(migrated_conn, rid)
+    assert [img["id"] for img in pend] == [a]
+    assert pend[0]["source_url"] == "https://m/a.png"
+    assert pend[0]["storefront_id"] == merchant["storefront_id"]
+
+
+def test_claim_image_for_fetch_is_idempotent(migrated_conn, merchant):
+    rid = _run(migrated_conn, merchant)
+    p = _product(migrated_conn, merchant, "lamp")
+    img = _image(migrated_conn, p, "https://m/a.png", rid)
+    assert repo.claim_image_for_fetch(migrated_conn, img) is True
+    repo.mark_image_fetched(migrated_conn, img,
+                            {"original": "o", "thumb": "t", "card": "c", "detail": "d"})
+    assert repo.claim_image_for_fetch(migrated_conn, img) is False
+
+
+def test_mark_image_outcomes_and_run_counts(migrated_conn, merchant):
+    rid = _run(migrated_conn, merchant)
+    p = _product(migrated_conn, merchant, "lamp")
+    ok = _image(migrated_conn, p, "https://m/a.png", rid, position=1)
+    bad = _image(migrated_conn, p, "https://m/b.png", rid, position=2)
+    miss = _image(migrated_conn, p, "https://m/c.png", rid, position=3)
+    repo.mark_image_fetched(migrated_conn, ok, {"original": "o", "thumb": "t", "card": "c", "detail": "d"})
+    repo.mark_image_rejected(migrated_conn, bad, "rejected_low_res", "below the resolution bar")
+    repo.mark_image_failed(migrated_conn, miss, "host unreachable")
+    assert repo.run_image_counts(migrated_conn, rid) == {
+        "fetched": 1, "rejected": 1, "failed": 1, "pending": 0, "total": 3}
+    outcomes = repo.run_image_outcomes(migrated_conn, rid)
+    assert {o["handle"] for o in outcomes} == {"lamp"}
+    assert {o["outcome"] for o in outcomes} == {"rejected_low_res", "failed"}  # fetched not listed
+
+
+def test_set_run_status_and_incomplete_runs(migrated_conn, merchant):
+    rid = _run(migrated_conn, merchant, status="fetching_images")
+    _run(migrated_conn, merchant, status="complete")
+    assert rid in [r["id"] for r in repo.incomplete_runs(migrated_conn)]
+    repo.set_run_status(migrated_conn, rid, "complete")
+    assert rid not in [r["id"] for r in repo.incomplete_runs(migrated_conn)]
+    row = migrated_conn.execute("SELECT status, completed_at FROM import_run WHERE id=%s", (rid,)).fetchone()
+    assert row[0] == "complete" and row[1] is not None