diff --git a/backend/app/domains/products/repo.py b/backend/app/domains/products/repo.py index 2a5158a..520eaf2 100644 --- a/backend/app/domains/products/repo.py +++ b/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. # --------------------------------------------------------------------------- diff --git a/backend/tests/test_products_repo_images.py b/backend/tests/test_products_repo_images.py new file mode 100644 index 0000000..0f28ee0 --- /dev/null +++ b/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