# CSV Import Implementation Plan


**Goal:** Let a merchant load a catalogue by uploading a Shopify-format CSV, with row-level error reporting and a dry run.

**Architecture:** One pure parser that turns CSV rows into the same source-product shape the Shopify and HTTP connectors already produce, one thin importer that normalises and persists through the existing pipeline, and two endpoints. No new normalisation, no new persistence, no new table.

**Tech Stack:** FastAPI, Python's `csv` module, psycopg2, pytest.

**Spec:** `docs/specs/2026-08-11-csv-import-design.md`

**Depends on:** the ingestion pipeline from Phase 0 — `catalog/normalise.py` and `catalog/sync.py`'s `persist` and `delete_stale`.

## Global Constraints

- **Never commit, never push, never stage.** The user commits their own work. Every task ends with `git status --short`.
- **No assistant attribution** anywhere.
- **Comments explain WHY, not WHAT.**
- **Never guess a currency.** It is a required upload field. A file without one is rejected before any row is parsed. This rule exists because a currency fallback once mislabelled 194 USD products as INR.
- **Reuse the existing pipeline.** `normalise()`, `persist()` and `delete_stale()` already work and are tested — the CSV path produces the same source-product dict the other connectors do and hands it over. Do not write a second normaliser or a second insert.
- **Every write is scoped by `(source_kind, source_ref)`.** `source_kind` is `"csv"`. Getting this wrong makes a CSV upload delete the Shopify catalogue — the exact hazard `save_products` was hardened against.
- **A bad row never fails the file.** Row-level problems are collected and reported; file-level problems reject before writing.
- **Row numbers are as the merchant sees them in a spreadsheet** — the header is row 1, so the first data row is row 2.
- Match conventions: module-level `logger`, snake_case, double quotes, 4-space indent, imports stdlib → third-party → `app.*`.
- Tests: `.venv/bin/python -m pytest`. Baselines: `tests/unit/` **611**, `tests/integration/` **133**.
- **Run the two suites in the FOREGROUND and SEPARATELY.** Never in one pytest process. Never in the background.

---

## File Structure

| File | Responsibility |
|---|---|
| `app/services/catalog/csv_source.py` (new) | Parse CSV text into source products. Pure — no DB, no network, no FastAPI. |
| `app/services/catalog/csv_import.py` (new) | Normalise, persist, replace the previous batch, build the report. |
| `app/api/catalog.py` (modify) | `POST /catalog/import/csv`, `GET /catalog/sample.csv` |
| `tests/unit/test_csv_source.py` (new) | Grouping, variants, every row-level error |
| `tests/integration/test_csv_import.py` (new) | Scoping, replacement, dry run |
| `tests/integration/test_catalog_routes.py` (modify) | The two endpoints |

---

### Task 1: The parser

**Files:**
- Create: `app/services/catalog/csv_source.py`
- Test: `tests/unit/test_csv_source.py`

**Interfaces:**
- Consumes: nothing. **Pure** — stdlib `csv` and `io` only.
- Produces:
  - `REQUIRED_COLUMNS = ("Handle", "Title")`
  - `CsvFormatError` — raised for file-level problems only
  - `parse_products(text: str, source_ref: str, currency: str) -> tuple[list[dict], list[dict]]` returning `(source_products, errors)`

Each source product must match the shape the other connectors emit, so
`normalise()` accepts it unchanged: `external_id`, `title`, `description`,
`brand`, `raw_category`, `product_type`, `tags`, `image_url`, `status`,
`variants` (each with `price`, `compare_at_price`, `available`, `options`),
`source_kind`, `source_ref`, `product_url`, `attributes_raw`,
`tenant_relations`, `rating`.

Read `app/services/catalog/shopify.py`'s `to_source_product` first and mirror its
keys exactly — a missing key surfaces as a `KeyError` deep inside normalisation.

- [ ] **Step 1: Write the failing test**

Create `tests/unit/test_csv_source.py`:

```python
import pytest

from app.services.catalog.csv_source import CsvFormatError, parse_products

HEADER = ("Handle,Title,Body (HTML),Vendor,Type,Tags,Published,"
          "Option1 Name,Option1 Value,Option2 Name,Option2 Value,"
          "Variant SKU,Variant Inventory Qty,Variant Price,"
          "Variant Compare At Price,Image Src,Status")

SHIRT = ("oxford-shirt,Oxford Shirt,A white shirt.,Brooks,Shirts,\"formal,cotton\","
         "TRUE,Size,M,Colour,White,SKU-M,12,59.00,79.00,"
         "https://x.example/s.jpg,active")
SHIRT_L = ("oxford-shirt,Oxford Shirt,A white shirt.,Brooks,Shirts,\"formal,cotton\","
           "TRUE,Size,L,Colour,White,SKU-L,0,65.00,79.00,"
           "https://x.example/s.jpg,active")


def parse(*rows, currency="USD"):
    text = "\n".join((HEADER,) + rows)
    return parse_products(text, "test.csv", currency)


def test_one_row_is_one_product():
    products, errors = parse(SHIRT)
    assert errors == []
    assert len(products) == 1
    assert products[0]["title"] == "Oxford Shirt"
    assert products[0]["brand"] == "Brooks"
    assert products[0]["product_type"] == "Shirts"


def test_rows_sharing_a_handle_are_one_product():
    products, _ = parse(SHIRT, SHIRT_L)
    assert len(products) == 1
    assert len(products[0]["variants"]) == 2


def test_the_lowest_variant_price_wins():
    # Matches how the Shopify connector already treats variants.
    products, _ = parse(SHIRT, SHIRT_L)
    assert min(v["price"] for v in products[0]["variants"]) == 59.00


def test_a_product_is_in_stock_if_any_variant_is():
    products, _ = parse(SHIRT, SHIRT_L)
    assert any(v["available"] for v in products[0]["variants"])


def test_a_product_is_out_of_stock_when_every_variant_is():
    zero = SHIRT.replace(",12,", ",0,")
    products, _ = parse(zero)
    assert not any(v["available"] for v in products[0]["variants"])


def test_options_are_collected():
    products, _ = parse(SHIRT)
    assert products[0]["variants"][0]["options"] == {"Size": "M", "Colour": "White"}


def test_a_single_option_product_has_no_empty_second_option():
    row = ("belt,Leather Belt,A belt.,Brooks,Belts,leather,TRUE,"
           "Size,M,,,SKU-B,5,29.00,,https://x.example/b.jpg,active")
    products, _ = parse(row)
    assert products[0]["variants"][0]["options"] == {"Size": "M"}


def test_tags_become_a_list():
    products, _ = parse(SHIRT)
    assert products[0]["tags"] == ["formal", "cotton"]


def test_product_fields_come_from_the_first_row_of_a_handle():
    # A real Shopify export leaves them blank on later rows.
    blank = "oxford-shirt,,,,,,,Size,L,Colour,Blue,SKU-B,4,59.00,,,active"
    products, errors = parse(SHIRT, blank)
    assert errors == []
    assert products[0]["title"] == "Oxford Shirt"
    assert products[0]["brand"] == "Brooks"
    assert len(products[0]["variants"]) == 2


def test_every_product_carries_the_declared_currency():
    products, _ = parse(SHIRT, currency="INR")
    assert products[0]["currency"] == "INR"


def test_the_source_is_recorded():
    products, _ = parse(SHIRT)
    assert products[0]["source_kind"] == "csv"
    assert products[0]["source_ref"] == "test.csv"


def test_the_handle_is_the_external_id():
    products, _ = parse(SHIRT)
    assert products[0]["external_id"] == "oxford-shirt"


# --- row-level errors, all survivable ---------------------------------------

def test_a_row_with_no_handle_is_skipped_and_reported():
    bad = ",No Handle,desc,Brooks,Shirts,,TRUE,Size,M,,,SKU,1,9.00,,,active"
    products, errors = parse(SHIRT, bad)
    assert len(products) == 1
    assert errors[0]["row"] == 3
    assert "Handle" in errors[0]["reason"]


def test_a_first_row_with_no_title_is_skipped_and_reported():
    bad = "ghost,,desc,Brooks,Shirts,,TRUE,Size,M,,,SKU,1,9.00,,,active"
    products, errors = parse(bad)
    assert products == []
    assert errors[0]["row"] == 2
    assert "Title" in errors[0]["reason"]


def test_an_unparseable_price_is_skipped_and_reported():
    bad = SHIRT.replace(",59.00,", ",ask,")
    products, errors = parse(bad)
    assert products == []
    assert errors[0]["row"] == 2
    assert "Price" in errors[0]["reason"]


def test_a_negative_price_is_rejected():
    bad = SHIRT.replace(",59.00,", ",-5.00,")
    _, errors = parse(bad)
    assert errors and errors[0]["row"] == 2


def test_a_negative_quantity_is_rejected():
    bad = SHIRT.replace(",12,", ",-3,")
    _, errors = parse(bad)
    assert errors and errors[0]["row"] == 2


def test_row_numbers_match_the_spreadsheet():
    # Header is row 1, so the first data row is row 2. An off-by-one here sends
    # a merchant to the wrong line in their file.
    bad = ",No Handle,d,B,T,,TRUE,Size,M,,,S,1,9.00,,,active"
    _, errors = parse(SHIRT, SHIRT_L, bad)
    assert errors[0]["row"] == 4


def test_a_file_where_every_row_fails_is_not_an_exception():
    bad = ",,,,,,,,,,,,,,,,"
    products, errors = parse(bad)
    assert products == []
    assert errors


# --- file-level errors, fatal ------------------------------------------------

def test_a_missing_required_column_is_fatal():
    with pytest.raises(CsvFormatError):
        parse_products("Title,Vendor\nA,B", "x.csv", "USD")


def test_a_file_with_no_data_rows_is_fatal():
    with pytest.raises(CsvFormatError):
        parse_products(HEADER, "x.csv", "USD")


def test_empty_input_is_fatal():
    with pytest.raises(CsvFormatError):
        parse_products("", "x.csv", "USD")


def test_a_missing_currency_is_fatal():
    with pytest.raises(CsvFormatError):
        parse_products(f"{HEADER}\n{SHIRT}", "x.csv", "")


def test_unknown_columns_are_ignored():
    # A real Shopify export has ~44 columns; we read 17. Rejecting a file for
    # carrying "Variant Weight Unit" would break "export and upload".
    header = HEADER + ",Variant Weight Unit,Gift Card"
    row = SHIRT + ",g,FALSE"
    products, errors = parse_products(f"{header}\n{row}", "x.csv", "USD")
    assert errors == []
    assert len(products) == 1
```

- [ ] **Step 2: Run test to verify it fails**

Run: `.venv/bin/python -m pytest tests/unit/test_csv_source.py -v`
Expected: FAIL with `ModuleNotFoundError`

- [ ] **Step 3: Write the implementation**

Create `app/services/catalog/csv_source.py`. Structure it as: validate the header,
then iterate rows with `enumerate(reader, start=2)` so the row number is the
spreadsheet's, grouping by `Handle` in an ordered dict. Keep it pure.

```python
"""Turns a Shopify-format product CSV into source products.

Pure: no database, no network, no request objects, so every grouping and
validation rule is testable without a fixture. The output is the same shape the
Shopify and HTTP connectors emit, so normalise() takes it unchanged.
"""
import csv
import io
import logging

logger = logging.getLogger(__name__)

REQUIRED_COLUMNS = ("Handle", "Title")
_OUT_OF_STOCK_QTY = 0


class CsvFormatError(Exception):
    """The file cannot be read at all. Row problems are reported, not raised."""


def _clean(value):
    return (value or "").strip()


def _options(row: dict) -> dict:
    options = {}
    for index in (1, 2, 3):
        name = _clean(row.get(f"Option{index} Name"))
        value = _clean(row.get(f"Option{index} Value"))
        if name and value:
            options[name] = value
    return options
```

`parse_products` then:

1. raises `CsvFormatError` when `currency` is blank, the text is empty, a
   required column is absent, or there are no data rows
2. for each row, records an error and continues on: no handle; no title on a
   handle's first row; a non-numeric, negative or missing price; a negative
   quantity
3. builds one dict per handle with the Shopify-connector keys, taking product
   fields from the handle's first row and appending a variant per row
4. sets `product_url` to `None` — a CSV carries no product URL, and normalise()
   already flags that in `missing_fields`
5. returns `(products, errors)` where each error is `{"row": int, "reason": str}`

- [ ] **Step 4: Run test to verify it passes**

Run: `.venv/bin/python -m pytest tests/unit/test_csv_source.py -v`
Expected: PASS, 22 tests

- [ ] **Step 5: Report changed files**

```bash
git status --short
```

Do not commit and do not stage.

---

### Task 2: The importer

**Files:**
- Create: `app/services/catalog/csv_import.py`
- Test: `tests/integration/test_csv_import.py`

**Interfaces:**
- Consumes: `parse_products`, `normalise` from `catalog/normalise.py`, `persist` and `delete_stale` from `catalog/sync.py`, `migrate_products_table`.
- Produces: `import_csv(tenant_id, text, source_ref, currency, dry_run=False) -> dict`

Read `sync_source` in `catalog/sync.py` first and follow exactly what it does
after it has source products: build the `context`, call `normalise` per product,
`persist`, then `delete_stale` with the run timestamp. The CSV path is that same
tail, with no fetch in front of it.

- [ ] **Step 1: Write the failing test**

Create `tests/integration/test_csv_import.py`:

```python
import pytest

from app.services.catalog.csv_import import import_csv
from app.services.infra.database import bootstrap_tenant, get_db_connection

HEADER = ("Handle,Title,Body (HTML),Vendor,Type,Tags,Published,"
          "Option1 Name,Option1 Value,Option2 Name,Option2 Value,"
          "Variant SKU,Variant Inventory Qty,Variant Price,"
          "Variant Compare At Price,Image Src,Status")


def csv_text(*handles):
    rows = [f"{h},{h.title()},A product.,Brand,Shirts,tag,TRUE,Size,M,,,"
            f"SKU-{h},5,19.00,,https://x.example/{h}.jpg,active"
            for h in handles]
    return "\n".join([HEADER, *rows])


def _rows(schema, where="TRUE"):
    conn = get_db_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(f'SELECT product_key, name, source_kind, source_ref '
                        f'FROM "{schema}".strategist_products WHERE {where}')
            return cur.fetchall()
    finally:
        conn.close()


@pytest.fixture
def tenant(temp_tenant):
    bootstrap_tenant(temp_tenant)
    return temp_tenant


def test_products_are_imported(tenant):
    report = import_csv(tenant, csv_text("shirt", "belt"), "spring.csv", "USD")
    assert report["imported"] == 2
    assert len(_rows(tenant)) == 2


def test_imported_rows_are_scoped_to_csv(tenant):
    # Without this a CSV upload and a Shopify sync delete each other.
    import_csv(tenant, csv_text("shirt"), "spring.csv", "USD")
    kinds = {r[2] for r in _rows(tenant)}
    refs = {r[3] for r in _rows(tenant)}
    assert kinds == {"csv"}
    assert refs == {"spring.csv"}


def test_re_uploading_the_same_file_replaces_that_batch(tenant):
    import_csv(tenant, csv_text("shirt", "belt"), "spring.csv", "USD")
    report = import_csv(tenant, csv_text("shirt"), "spring.csv", "USD")
    names = {r[1] for r in _rows(tenant)}
    assert names == {"Shirt"}
    assert report["replaced"] == 1


def test_a_second_file_adds_rather_than_replaces(tenant):
    import_csv(tenant, csv_text("shirt"), "spring.csv", "USD")
    import_csv(tenant, csv_text("coat"), "winter.csv", "USD")
    assert len(_rows(tenant)) == 2


def test_another_sources_products_are_never_touched(tenant):
    # The hazard this scoping exists for.
    conn = get_db_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                f'INSERT INTO "{tenant}".strategist_products '
                "(product_key, name, product_url, source_kind, source_ref) "
                "VALUES ('shopify:1','Shopify Product','https://x/1',"
                "'shopify','shop.myshopify.com')")
        conn.commit()
    finally:
        conn.close()

    import_csv(tenant, csv_text("shirt"), "spring.csv", "USD")
    import_csv(tenant, csv_text("belt"), "spring.csv", "USD")
    assert len(_rows(tenant, "source_kind = 'shopify'")) == 1


def test_a_dry_run_writes_nothing(tenant):
    report = import_csv(tenant, csv_text("shirt"), "spring.csv", "USD",
                        dry_run=True)
    assert report["dry_run"] is True
    assert report["products"] == 1
    assert _rows(tenant) == []


def test_bad_rows_are_reported_but_good_ones_land(tenant):
    text = csv_text("shirt") + "\n,,,,,,,,,,,,,,,,"
    report = import_csv(tenant, text, "spring.csv", "USD")
    assert report["imported"] == 1
    assert report["errors"]
    assert len(_rows(tenant)) == 1


def test_the_declared_currency_is_stored(tenant):
    import_csv(tenant, csv_text("shirt"), "spring.csv", "INR")
    conn = get_db_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(f'SELECT currency FROM "{tenant}".strategist_products')
            assert cur.fetchone()[0] == "INR"
    finally:
        conn.close()


def test_imported_products_are_flagged_for_having_no_url(tenant):
    # A CSV carries no product URL, so they are visible in the catalogue but
    # must not be served. This is the most likely surprise in the feature.
    import_csv(tenant, csv_text("shirt"), "spring.csv", "USD")
    conn = get_db_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(f'SELECT missing_fields FROM "{tenant}".strategist_products')
            assert "product_url" in cur.fetchone()[0]
    finally:
        conn.close()
```

- [ ] **Step 2: Run test to verify it fails**

Run: `.venv/bin/python -m pytest tests/integration/test_csv_import.py -v`
Expected: FAIL with `ModuleNotFoundError`

- [ ] **Step 3: Write the implementation**

Create `app/services/catalog/csv_import.py`. It calls `migrate_products_table`,
parses, and on `dry_run` returns the report before touching the database.
Otherwise it normalises each source product with a context of
`{"shop_name": None, "primary_domain": None, "url_template": None,
"currency": currency, "known_brands": []}`, persists with
`source_kind="csv"`, then calls `delete_stale` with the run's start timestamp so
products from a previous upload of the same `source_ref` are removed.

`replaced` is the count `delete_stale` returns. Report the shape from spec §8.

- [ ] **Step 4: Run test to verify it passes**

Run: `.venv/bin/python -m pytest tests/integration/test_csv_import.py -v`
Expected: PASS, 9 tests

- [ ] **Step 5: Report changed files**

```bash
git status --short
```

Do not commit and do not stage.

---

### Task 3: The endpoints

**Files:**
- Modify: `app/api/catalog.py`
- Test: `tests/integration/test_catalog_routes.py`

**Interfaces:**
- Produces: `POST /catalog/import/csv` (multipart) and `GET /catalog/sample.csv`.

- [ ] **Step 1: Write the failing test**

Append to `tests/integration/test_catalog_routes.py`:

```python
CSV_BODY = (
    "Handle,Title,Body (HTML),Vendor,Type,Tags,Published,"
    "Option1 Name,Option1 Value,Option2 Name,Option2 Value,"
    "Variant SKU,Variant Inventory Qty,Variant Price,"
    "Variant Compare At Price,Image Src,Status\n"
    "shirt,Shirt,A shirt.,Brand,Shirts,tag,TRUE,Size,M,,,SKU,5,19.00,,"
    "https://x.example/s.jpg,active\n")


def test_the_import_route_is_registered():
    assert "/catalog/import/csv" in set(app.openapi()["paths"])


def test_import_requires_a_currency():
    resp = client.post("/catalog/import/csv",
                       data={"tenant_id": TENANT},
                       files={"file": ("p.csv", CSV_BODY, "text/csv")})
    assert resp.status_code == 422


def test_import_returns_the_report(monkeypatch):
    from app.api import catalog
    monkeypatch.setattr(catalog, "import_csv",
                        lambda *a, **kw: {"imported": 1, "products": 1,
                                          "errors": [], "dry_run": False})
    resp = client.post("/catalog/import/csv",
                       data={"tenant_id": TENANT, "currency": "USD"},
                       files={"file": ("p.csv", CSV_BODY, "text/csv")})
    assert resp.status_code == 200
    assert resp.json()["imported"] == 1


def test_the_uploaded_filename_becomes_the_source_ref(monkeypatch):
    from app.api import catalog
    seen = {}
    monkeypatch.setattr(catalog, "import_csv",
                        lambda t, text, ref, cur, dry_run=False:
                        seen.update(ref=ref) or {"imported": 0})
    client.post("/catalog/import/csv",
                data={"tenant_id": TENANT, "currency": "USD"},
                files={"file": ("spring-2026.csv", CSV_BODY, "text/csv")})
    assert seen["ref"] == "spring-2026.csv"


def test_dry_run_is_passed_through(monkeypatch):
    from app.api import catalog
    seen = {}
    monkeypatch.setattr(catalog, "import_csv",
                        lambda t, text, ref, cur, dry_run=False:
                        seen.update(dry=dry_run) or {"imported": 0})
    client.post("/catalog/import/csv",
                data={"tenant_id": TENANT, "currency": "USD", "dry_run": "true"},
                files={"file": ("p.csv", CSV_BODY, "text/csv")})
    assert seen["dry"] is True


def test_a_malformed_file_is_a_422_not_a_500(monkeypatch):
    # A bad spreadsheet is the merchant's problem to fix, not a server fault.
    from app.api import catalog
    from app.services.catalog.csv_source import CsvFormatError

    def boom(*a, **kw):
        raise CsvFormatError("missing required column: Handle")

    monkeypatch.setattr(catalog, "import_csv", boom)
    resp = client.post("/catalog/import/csv",
                       data={"tenant_id": TENANT, "currency": "USD"},
                       files={"file": ("p.csv", "nonsense", "text/csv")})
    assert resp.status_code == 422
    assert "Handle" in resp.text


def test_the_sample_csv_is_downloadable():
    resp = client.get("/catalog/sample.csv")
    assert resp.status_code == 200
    assert "Handle,Title" in resp.text
    assert "attachment" in resp.headers.get("content-disposition", "")
```

- [ ] **Step 2: Run test to verify it fails**

Run: `.venv/bin/python -m pytest tests/integration/test_catalog_routes.py -v`
Expected: FAIL — both routes 404.

- [ ] **Step 3: Write the implementation**

In `app/api/catalog.py`, mirror the existing routes' style — `Form(...)` fields,
`anyio.to_thread.run_sync` for the blocking work, and a `502` with only the
exception class name for unexpected failures. The file comes in as
`UploadFile = File(...)`; decode with `errors="replace"` so an odd byte does not
500. A `CsvFormatError` becomes a `422` carrying its message, because a bad
spreadsheet is the merchant's to fix.

`GET /catalog/sample.csv` returns `app/static/sample_products.csv` as a
`FileResponse` with `media_type="text/csv"` and
`filename="sample_products.csv"` so a browser downloads rather than renders it.
Register both above `/products/{product_key}`, as the existing route-ordering
comment in that file requires.

- [ ] **Step 4: Run the tests**

Run: `.venv/bin/python -m pytest tests/integration/test_catalog_routes.py -v`
Expected: PASS

- [ ] **Step 5: Run both suites**

`.venv/bin/python -m pytest tests/unit/ -q`, then
`.venv/bin/python -m pytest tests/integration/ -q`, sequentially, in the
foreground. Expected: no regressions against 611 / 133.

- [ ] **Step 6: Report changed files**

```bash
git status --short
```

Do not commit and do not stage.

---

### Task 4: Live verification

**Files:** none — verification only.

- [ ] **Step 1: Import the sample into a scratch tenant**

Start uvicorn on 8001. Import `app/static/sample_products.csv` into a throwaway
tenant — **not** the live catalogue — with `currency=GBP`:

```bash
curl -s -X POST http://localhost:8001/catalog/import/csv \
  -F "tenant_id=org_csv_probe" -F "currency=GBP" \
  -F "file=@app/static/sample_products.csv" | python3 -m json.tool
```

Expected: 12 rows, 8 products, 8 imported, 0 skipped. If the numbers differ from
the parse check already done on this file, say so — the file is known good.

- [ ] **Step 2: Confirm the variant collapsing**

Query the scratch tenant and confirm the Oxford Shirt landed once, at its lowest
price, in stock (one variant has 12), and that the Rain Jacket is present but
carries a non-serving status.

- [ ] **Step 3: Prove the scoping**

Import the file a second time and confirm the product count stays 8, not 16.
Then check the live tenant's counts are untouched:

```bash
PYTHONPATH=. .venv/bin/python -c "
from app.services.infra.database import get_db_connection
T='org_8c32bf3e-6a18-4739-9b1c-94c0cf11125f'
c=get_db_connection(); cur=c.cursor()
cur.execute('SELECT source_kind, count(*) FROM \"'+T+'\".strategist_products GROUP BY 1 ORDER BY 1')
print('live tenant unchanged:', cur.fetchall()); c.close()"
```

Expected: crawl 7, http_api 194, shopify 17 — exactly as before.

- [ ] **Step 4: Run the pipeline over the imported products**

Call `/catalog/enrich` and `/catalog/pair` for the scratch tenant and confirm the
imported products flow through unchanged — attributes extracted, pairs built.
This is the point of matching the existing source-product shape.

Report whether the clothing products pair sensibly. This is the first genuinely
apparel-shaped catalogue this system has seen, so it is also the first real test
of the colour-compatibility rule that could never be exercised on electronics.

- [ ] **Step 5: Drop the scratch tenant and report**

```bash
PYTHONPATH=. .venv/bin/python -c "
from app.services.infra.database import get_db_connection
c=get_db_connection(); cur=c.cursor()
cur.execute('DROP SCHEMA IF EXISTS \"org_csv_probe\" CASCADE'); c.commit(); c.close()
print('scratch tenant dropped')"
```

Summarise: changed files, test counts, the import report, proof the live tenant
was untouched, and what the clothing pairings looked like. Remind the user to
commit.

---

## Follow-on work (not in this plan)

- **A store domain for CSV products**, so they get product URLs and can actually be served. Today they import flagged and unserved.
- **Excel (.xlsx) upload**, which is what merchants will actually try first.
- **An upload size limit**, since the file is read into memory.
- **Per-variant rows**, the same gap Shopify import has.