# Attribute Extraction Implementation Plan **Goal:** Give every product the structured attributes the recommendation matching specification reads from, by extracting them with one language-model call per batch of twenty products. **Architecture:** Four focused modules. `price_tiers.py` computes a tenant's own price bands, because "premium" only means expensive *for this catalogue*. `attribute_extractor.py` owns the single model call and validates its output against closed sets. `attribute_merge.py` merges inferred values into existing ones without ever overwriting merchant data. `enrichment.py` orchestrates: select, batch, extract, merge, persist, report. One endpoint triggers it. **Tech Stack:** FastAPI, psycopg2, Azure OpenAI via the existing client, pytest. **Spec:** `docs/specs/2026-08-10-attribute-extraction-design.md` **Depends on:** Phase 0 (sub-projects A, B and C1), built and live-verified. The live catalogue holds 218 products across three sources. ## Global Constraints - **Never commit, never push, never stage.** The user commits their own work. Every task ends with `git status --short` and a report. Do not run `git commit`, `git push`, or `git add`. - **No assistant attribution** in any file, message, or commit. - **Comments explain WHY, not WHAT.** A short comment only where the reason is not evident from the code. A reviewer who knows Python must find zero noise comments. - **Inferred data never overwrites merchant data.** Extraction fills gaps only. Every inferred value carries `source: "inferred"` and a confidence, so the matching layer can weight a merchant's own value above a guess. - **A model authors a durable artifact; the artifact is what downstream reads.** Extraction results are stored and reused, never recomputed per request. - **Cached on `content_hash`.** A product is re-extracted only when its content actually changed. - **Tests assert shape, not model values.** Asserting the model returns `"navy"` breaks on any prompt or model change, and a suite that breaks for non-reasons stops being trusted. - **Never log a full product payload, an API key, or a raw model response body.** - Match conventions: module-level `logger = logging.getLogger(__name__)`, snake_case, double quotes, 4-space indent, type hints on public signatures, imports stdlib → third-party → `app.*`, `psycopg2` with `sql.SQL`/`sql.Identifier` for interpolated identifiers. - Existing LLM call convention in this repo: `client.chat.completions.create(model=settings.LLM_MODEL, messages=[...], response_format={"type": "json_object"})`, then `json.loads(response.choices[0].message.content)`. Follow it. - Tests: `.venv/bin/python -m pytest`. Baselines: `tests/unit/` **502 passing**, `tests/integration/` **54 passing**. `.venv` is a symlink to the main checkout's virtualenv. - **Run test suites in the FOREGROUND, one at a time.** Do not start a background run. - Do **not** modify `app/services/products.py`. Recommendation behaviour is owned by separate work. --- ## File Structure | File | Responsibility | |---|---| | `app/services/price_tiers.py` (new) | Compute a tenant's price percentiles and classify one price. Pure apart from one query. | | `app/services/attribute_extractor.py` (new) | The single model call: build the batch prompt, parse, validate against closed sets. No DB. | | `app/services/attribute_merge.py` (new) | Merge inferred attributes into existing ones without overwriting. Pure. | | `app/services/enrichment.py` (new) | Orchestration: select products, batch, extract, merge, persist, report. | | `app/services/database.py` (modify) | Two new columns: `is_accessory`, `price_tier`. | | `app/api/endpoints.py` (modify) | `POST /catalog/enrich`. | --- ### Task 1: Two new columns **Files:** - Modify: `app/services/database.py` — `_PRODUCT_COLUMNS` and `bootstrap_tenant` - Modify: `tests/integration/test_products_table.py` — `EXPECTED_COLUMNS` - Test: `tests/integration/test_products_migration.py` **Interfaces:** - Consumes: `migrate_products_table(tenant_id)`. - Produces: columns `is_accessory BOOLEAN`, `price_tier TEXT`. **Why columns rather than JSONB:** both are *filtered on* rather than read — `is_accessory` by the later complement materialisation, `price_tier` by the online ranker. A JSONB lookup for a hot filter is the wrong shape. **This task breaks an existing passing test unless you update it.** `test_products_table.py` asserts `set(columns) == EXPECTED_COLUMNS` exactly. Add both names. **Do not weaken the `==` to a subset check** — its exactness is what catches an accidental column. - [ ] **Step 1: Write the failing test** Append to `tests/integration/test_products_migration.py`. That file already has an `old_table` fixture and helpers for reading columns and running SQL — read it first and use whatever they are actually called rather than the `_columns` / `_exec` / `_query` names used below. ```python EXTRACTION_COLUMNS = {"is_accessory", "price_tier"} def test_extraction_columns_are_added(old_table): assert not (EXTRACTION_COLUMNS & _columns(old_table)) migrate_products_table(old_table) assert EXTRACTION_COLUMNS <= _columns(old_table) def test_is_accessory_defaults_to_null_not_false(old_table): # Null means "not yet extracted"; false means "extracted, and it is not an # accessory". Defaulting to false would make an unextracted catalog look # fully processed. migrate_products_table(old_table) _exec(f"""INSERT INTO "{old_table}".strategist_products (product_key, name, product_url) VALUES ('c:z','Z','https://x/z')""") rows = _query(f"SELECT is_accessory FROM \"{old_table}\".strategist_products " "WHERE product_key = 'c:z'") assert rows[0][0] is None ``` - [ ] **Step 2: Run test to verify it fails** Run: `.venv/bin/python -m pytest tests/integration/test_products_migration.py -v` Expected: FAIL — the two columns do not exist. - [ ] **Step 3: Add the columns** In `app/services/database.py`, append to `_PRODUCT_COLUMNS`: ```python ("is_accessory", "BOOLEAN"), ("price_tier", "TEXT"), ``` Add the same two to `bootstrap_tenant`'s `strategist_products` CREATE TABLE, in the same style as the columns already there. Note the brace-escaping convention in that statement: a literal `{}` is written `{{}}` because it goes through `.format()`. - [ ] **Step 4: Update the exact-columns test** Add `"is_accessory"` and `"price_tier"` to `EXPECTED_COLUMNS` in `tests/integration/test_products_table.py`, with a comment recording that they exist for attribute extraction. - [ ] **Step 5: Run the live migration, then the suite** ```bash PYTHONPATH=. .venv/bin/python -c " from app.services.database import migrate_products_table print(migrate_products_table('org_8c32bf3e-6a18-4739-9b1c-94c0cf11125f'))" .venv/bin/python -m pytest tests/integration/ -q ``` Run the migration BEFORE the suite: `bootstrap_tenant` indexes columns the live tenant may not have yet, and running the suite first fails with `UndefinedColumn`. Run it twice and confirm the second reports zero added. - [ ] **Step 6: Report changed files** ```bash git status --short ``` Do not commit and do not stage. --- ### Task 2: Tenant-relative price tiers **Files:** - Create: `app/services/price_tiers.py` - Test: `tests/unit/test_price_tiers.py` - Test: `tests/integration/test_price_tiers_db.py` **Interfaces:** - Consumes: `get_db_connection` from `app.services.database`. - Produces: - `compute_bands(tenant_id: str) -> tuple[int, int] | None` — the 33rd and 67th percentile in minor units, or `None` when there are too few priced products - `classify(price_cents: int | None, bands: tuple[int, int] | None) -> str | None` — `"budget"`, `"mid"`, `"premium"`, or `None` **Why this exists:** "premium" has no global meaning — it means expensive *for this catalogue*. The live catalogue spans 79 to 3,699,999 minor units, so any fixed threshold would put nearly everything in one bucket. - [ ] **Step 1: Write the failing test** Create `tests/unit/test_price_tiers.py`: ```python import pytest from app.services.price_tiers import classify BANDS = (5000, 20000) # 33rd and 67th percentile, in minor units @pytest.mark.parametrize("price,expected", [ (100, "budget"), (4999, "budget"), (5000, "mid"), # boundary belongs to the higher band (12000, "mid"), (20000, "premium"), # boundary belongs to the higher band (500000, "premium"), ]) def test_classify(price, expected): assert classify(price, BANDS) == expected def test_no_price_is_unclassified(): assert classify(None, BANDS) is None def test_no_bands_means_unclassified(): # A catalog too small to have a distribution must not be given fake tiers. assert classify(1000, None) is None def test_boundaries_are_documented_and_stable(): # Pinned because a later phase filters on these and a silent shift would # move products between tiers without any code appearing to change. assert classify(4999, BANDS) == "budget" assert classify(5000, BANDS) == "mid" assert classify(19999, BANDS) == "mid" assert classify(20000, BANDS) == "premium" ``` - [ ] **Step 2: Run test to verify it fails** Run: `.venv/bin/python -m pytest tests/unit/test_price_tiers.py -v` Expected: FAIL with `ModuleNotFoundError: No module named 'app.services.price_tiers'` - [ ] **Step 3: Write the implementation** Create `app/services/price_tiers.py`: ```python """Bands a tenant's own prices into budget, mid and premium. "Premium" has no global meaning -- it means expensive for this catalog. The live catalog spans 79 to 3,699,999 minor units, so a fixed threshold would put nearly everything in one bucket. """ import logging from psycopg2 import sql from app.services.database import get_db_connection logger = logging.getLogger(__name__) MIN_PRICED_PRODUCTS = 10 def compute_bands(tenant_id: str): """The 33rd and 67th percentile of a tenant's prices, or None. Prefers price_reference_cents so a tenant with two currencies is not banded on unconverted numbers. """ conn = get_db_connection() try: with conn.cursor() as cur: cur.execute(sql.SQL(""" SELECT count(*), percentile_disc(0.33) WITHIN GROUP (ORDER BY p), percentile_disc(0.67) WITHIN GROUP (ORDER BY p) FROM ( SELECT COALESCE(price_reference_cents, price_cents) AS p FROM {}.strategist_products WHERE COALESCE(price_reference_cents, price_cents) IS NOT NULL ) prices """).format(sql.Identifier(tenant_id))) count, low, high = cur.fetchone() finally: conn.close() if not count or count < MIN_PRICED_PRODUCTS or low is None or high is None: logger.info("Too few priced products in %s to band prices", tenant_id) return None return int(low), int(high) def classify(price_cents, bands): if price_cents is None or bands is None: return None low, high = bands if price_cents < low: return "budget" if price_cents < high: return "mid" return "premium" ``` - [ ] **Step 4: Run test to verify it passes** Run: `.venv/bin/python -m pytest tests/unit/test_price_tiers.py -v` Expected: PASS, 9 tests - [ ] **Step 5: Write the database test** Create `tests/integration/test_price_tiers_db.py`: ```python import pytest from app.services.database import bootstrap_tenant, get_db_connection from app.services.price_tiers import MIN_PRICED_PRODUCTS, compute_bands def _insert(schema, rows): conn = get_db_connection() try: with conn.cursor() as cur: for i, price in enumerate(rows): cur.execute( f'INSERT INTO "{schema}".strategist_products ' "(product_key, name, product_url, price_cents) " "VALUES (%s,%s,%s,%s)", (f"t:{i}", f"P{i}", f"https://x.example/{i}", price)) conn.commit() finally: conn.close() @pytest.fixture def priced(temp_tenant): bootstrap_tenant(temp_tenant) # 1..30 hundred: a known distribution so the percentiles are checkable. _insert(temp_tenant, [i * 100 for i in range(1, 31)]) return temp_tenant def test_bands_come_from_the_tenants_own_distribution(priced): low, high = compute_bands(priced) assert 900 <= low <= 1100 # around the 33rd percentile of 100..3000 assert 2000 <= high <= 2200 # around the 67th def test_too_few_products_yields_no_bands(temp_tenant): bootstrap_tenant(temp_tenant) _insert(temp_tenant, [100] * (MIN_PRICED_PRODUCTS - 1)) assert compute_bands(temp_tenant) is None def test_reference_price_is_preferred_over_source_price(temp_tenant): # A two-currency tenant must not be banded on unconverted numbers. bootstrap_tenant(temp_tenant) conn = get_db_connection() try: with conn.cursor() as cur: for i in range(MIN_PRICED_PRODUCTS + 5): cur.execute( f'INSERT INTO "{temp_tenant}".strategist_products ' "(product_key, name, product_url, price_cents, " " price_reference_cents) VALUES (%s,%s,%s,%s,%s)", (f"r:{i}", f"P{i}", f"https://x.example/r{i}", 999999, i * 100)) conn.commit() finally: conn.close() low, high = compute_bands(temp_tenant) assert low < 999999 and high < 999999 ``` - [ ] **Step 6: Run the database test** Run: `.venv/bin/python -m pytest tests/integration/test_price_tiers_db.py -v` Expected: PASS, 3 tests - [ ] **Step 7: Report changed files** ```bash git status --short ``` Do not commit and do not stage. --- ### Task 3: The extractor **Files:** - Create: `app/services/attribute_extractor.py` - Test: `tests/unit/test_attribute_extractor.py` **Interfaces:** - Consumes: `app.core.llm_client.client`, `settings.LLM_MODEL`. - Produces: - `BATCH_SIZE: int` = 20 - `GENDERS`, `SIZE_SYSTEMS`, `PRICE_TIERS`: closed sets - `build_prompt(products: list[dict]) -> str` - `validate_one(raw: dict) -> dict` — discards anything outside the schema - `extract_batch(products: list[dict]) -> dict[str, dict]` — keyed by `product_key`; returns `{}` on failure **The rule that makes this safe:** the model's output is never trusted. Anything outside the declared schema, or outside a closed set, is discarded rather than stored — a single bad value must not reach a column a later phase filters on. - [ ] **Step 1: Write the failing test** Create `tests/unit/test_attribute_extractor.py`: ```python import pytest from app.services.attribute_extractor import ( BATCH_SIZE, GENDERS, PRICE_TIERS, SIZE_SYSTEMS, build_prompt, extract_batch, validate_one, ) PRODUCTS = [ {"product_key": "shopify:1", "name": "Oxford Shirt", "description": "A white cotton formal shirt.", "brand": "Brooks", "taxonomy_path": ["Apparel", "Shirts"], "attributes": []}, {"product_key": "http_api:2", "name": "Brown Leather Belt", "description": "Full-grain leather belt.", "brand": None, "taxonomy_path": ["Apparel", "Belts"], "attributes": []}, ] def test_batch_size_is_twenty(): assert BATCH_SIZE == 20 def test_prompt_carries_every_product_key(): prompt = build_prompt(PRODUCTS) for p in PRODUCTS: assert p["product_key"] in prompt def test_prompt_does_not_carry_a_whole_description_when_long(): # A single record can be kilobytes of prose that teaches the model nothing. long_one = dict(PRODUCTS[0], description="x" * 5000) assert len(build_prompt([long_one])) < 3000 def test_validate_keeps_known_fields(): out = validate_one({ "color": "white", "material": "cotton", "style": "formal", "gender": "male", "size_system": "alpha", "use_case": "office", "is_accessory": False, "price_tier": "mid", "key_features": ["button-down", "breathable"], }) assert out["color"] == "white" assert out["is_accessory"] is False assert out["key_features"] == ["button-down", "breathable"] def test_validate_discards_unknown_fields(): out = validate_one({"color": "white", "invented_field": "nonsense"}) assert "invented_field" not in out @pytest.mark.parametrize("field,bad", [ ("gender", "attack helicopter"), ("size_system", "furlongs"), ("price_tier", "luxury"), ]) def test_validate_discards_values_outside_a_closed_set(field, bad): # These reach columns a later phase filters on; one bad value would make a # filter silently miss products. assert field not in validate_one({field: bad}) @pytest.mark.parametrize("field,good", [ ("gender", "female"), ("size_system", "uk"), ("price_tier", "premium"), ]) def test_validate_keeps_values_inside_a_closed_set(field, good): assert validate_one({field: good})[field] == good def test_is_accessory_must_be_a_real_boolean(): assert "is_accessory" not in validate_one({"is_accessory": "yes"}) assert validate_one({"is_accessory": True})["is_accessory"] is True def test_key_features_capped_at_five(): out = validate_one({"key_features": ["a", "b", "c", "d", "e", "f", "g"]}) assert len(out["key_features"]) == 5 def test_key_features_must_be_a_list_of_strings(): assert "key_features" not in validate_one({"key_features": "not a list"}) assert "key_features" not in validate_one({"key_features": [1, 2, 3]}) def test_extract_batch_keys_by_product_key(monkeypatch): import app.services.attribute_extractor as mod monkeypatch.setattr(mod, "_call_llm", lambda prompt: { "products": [ {"product_key": "shopify:1", "color": "white"}, {"product_key": "http_api:2", "color": "brown", "is_accessory": True}, ]}) out = extract_batch(PRODUCTS) assert set(out) == {"shopify:1", "http_api:2"} assert out["http_api:2"]["is_accessory"] is True def test_extract_batch_ignores_a_key_it_did_not_ask_about(monkeypatch): # A model that invents a product must not create a row for it. import app.services.attribute_extractor as mod monkeypatch.setattr(mod, "_call_llm", lambda prompt: { "products": [{"product_key": "invented:99", "color": "red"}]}) assert extract_batch(PRODUCTS) == {} def test_extract_batch_returns_empty_on_failure(monkeypatch): # One bad batch must not abort a whole catalog. import app.services.attribute_extractor as mod def boom(prompt): raise RuntimeError("model unavailable") monkeypatch.setattr(mod, "_call_llm", boom) assert extract_batch(PRODUCTS) == {} def test_extract_batch_survives_a_non_dict_response(monkeypatch): import app.services.attribute_extractor as mod monkeypatch.setattr(mod, "_call_llm", lambda prompt: "not a dict") assert extract_batch(PRODUCTS) == {} ``` - [ ] **Step 2: Run test to verify it fails** Run: `.venv/bin/python -m pytest tests/unit/test_attribute_extractor.py -v` Expected: FAIL with `ModuleNotFoundError` - [ ] **Step 3: Write the implementation** Create `app/services/attribute_extractor.py`: ```python """One model call per batch of products, producing validated attributes. The model's output is never trusted: anything outside the declared schema, or outside a closed set, is discarded rather than stored. Two of these fields reach columns a later phase filters on, and one bad value there makes a filter silently miss products. """ import json import logging from app.core.config import settings from app.core.llm_client import client logger = logging.getLogger(__name__) BATCH_SIZE = 20 DESCRIPTION_CHARS = 400 MAX_KEY_FEATURES = 5 GENDERS = ("male", "female", "unisex", "kids") SIZE_SYSTEMS = ("alpha", "numeric", "uk", "eu", "us", "volume") PRICE_TIERS = ("budget", "mid", "premium") STRING_FIELDS = ("color", "material", "style", "use_case") CLOSED_SETS = {"gender": GENDERS, "size_system": SIZE_SYSTEMS, "price_tier": PRICE_TIERS} _SYSTEM = ( "You extract structured attributes from e-commerce products. Return JSON only.\n" 'Return {"products": [{"product_key": ..., ...fields...}]}\n' "Fields: color, material, style, use_case (short lowercase strings); " f"gender one of {GENDERS}; size_system one of {SIZE_SYSTEMS}; " f"price_tier one of {PRICE_TIERS}; is_accessory a boolean; " f"key_features an array of at most {MAX_KEY_FEATURES} short phrases.\n" "Omit any field you cannot determine. Never guess a product_key." ) def build_prompt(products: list) -> str: payload = [ { "product_key": p["product_key"], "title": p.get("name"), # Truncated: a single record can be kilobytes of prose that teaches # the model nothing about the product's attributes. "description": (p.get("description") or "")[:DESCRIPTION_CHARS], "brand": p.get("brand"), "category": " > ".join(p.get("taxonomy_path") or []), "known_attributes": [ {"key": a["key"], "value": a["value"]} for a in (p.get("attributes") or []) ], } for p in products ] return json.dumps({"products": payload}, default=str) def validate_one(raw: dict) -> dict: out = {} if not isinstance(raw, dict): return out for field in STRING_FIELDS: value = raw.get(field) if isinstance(value, str) and value.strip(): out[field] = value.strip().lower() for field, allowed in CLOSED_SETS.items(): value = raw.get(field) if isinstance(value, str) and value.strip().lower() in allowed: out[field] = value.strip().lower() accessory = raw.get("is_accessory") if isinstance(accessory, bool): out["is_accessory"] = accessory features = raw.get("key_features") if isinstance(features, list) and all(isinstance(f, str) for f in features): cleaned = [f.strip() for f in features if f.strip()] if cleaned: out["key_features"] = cleaned[:MAX_KEY_FEATURES] return out def _call_llm(prompt: str) -> dict: response = client.chat.completions.create( model=settings.LLM_MODEL, response_format={"type": "json_object"}, messages=[{"role": "system", "content": _SYSTEM}, {"role": "user", "content": prompt}], ) return json.loads(response.choices[0].message.content) def extract_batch(products: list) -> dict: """Returns {product_key: attributes}. An empty dict on any failure, so one bad batch degrades to "not extracted" rather than aborting a catalog.""" if not products: return {} try: raw = _call_llm(build_prompt(products)) except Exception: logger.error("Attribute extraction failed for a batch of %d", len(products), exc_info=True) return {} if not isinstance(raw, dict) or not isinstance(raw.get("products"), list): logger.warning("Attribute extraction returned an unexpected shape") return {} asked = {p["product_key"] for p in products} out = {} for item in raw["products"]: if not isinstance(item, dict): continue key = item.get("product_key") # A model that invents a product must not create a row for it. if key not in asked: continue validated = validate_one(item) if validated: out[key] = validated return out ``` - [ ] **Step 4: Run test to verify it passes** Run: `.venv/bin/python -m pytest tests/unit/test_attribute_extractor.py -v` Expected: PASS, 20 tests - [ ] **Step 5: Report changed files** ```bash git status --short ``` Do not commit and do not stage. --- ### Task 4: Merging without overwriting **Files:** - Create: `app/services/attribute_merge.py` - Test: `tests/unit/test_attribute_merge.py` **Interfaces:** - Consumes: nothing. - Produces: - `INFERRED_CONFIDENCE: float` = 0.5 - `merge_attributes(existing: list[dict], inferred: dict) -> tuple[list[dict], int]` returning the merged list and a count of conflicts **The rule:** a merchant's own value always wins. If a Shopify variant option says the colour is white, the model does not get to change it. Extraction fills gaps only. - [ ] **Step 1: Write the failing test** Create `tests/unit/test_attribute_merge.py`: ```python from app.services.attribute_merge import INFERRED_CONFIDENCE, merge_attributes def existing(key, value, source="option", confidence=1.0): return {"key": key, "raw_value": value, "value": value, "unit": None, "source": source, "confidence": confidence} def test_a_gap_is_filled(): merged, conflicts = merge_attributes([], {"color": "white"}) assert len(merged) == 1 assert merged[0]["key"] == "color" assert merged[0]["source"] == "inferred" assert merged[0]["confidence"] == INFERRED_CONFIDENCE assert conflicts == 0 def test_a_merchant_value_is_never_overwritten(): # If a Shopify variant option said white, the model does not get to change it. merged, conflicts = merge_attributes([existing("color", "white")], {"color": "cream"}) colors = [a for a in merged if a["key"] == "color"] assert len(colors) == 1 assert colors[0]["value"] == "white" assert colors[0]["source"] == "option" def test_a_disagreement_is_counted(): # A high conflict rate means the prompt is wrong, which is worth knowing. _, conflicts = merge_attributes([existing("color", "white")], {"color": "cream"}) assert conflicts == 1 def test_agreement_is_not_a_conflict(): _, conflicts = merge_attributes([existing("color", "white")], {"color": "white"}) assert conflicts == 0 def test_inferred_values_are_marked_and_weighted_below_structured(): merged, _ = merge_attributes([], {"material": "cotton"}) assert merged[0]["confidence"] < 1.0 def test_scalar_only_fields_are_not_stored_as_attributes(): # is_accessory and price_tier get columns; duplicating them as attributes # would inflate the attribute count that quality_score depends on. merged, _ = merge_attributes([], {"is_accessory": True, "price_tier": "mid", "color": "white"}) assert {a["key"] for a in merged} == {"color"} def test_key_features_become_one_attribute_each(): merged, _ = merge_attributes([], {"key_features": ["waterproof", "vegan"]}) features = sorted(a["value"] for a in merged if a["key"] == "feature") assert features == ["vegan", "waterproof"] def test_result_is_deterministically_ordered(): # A later phase hashes this list to decide whether to re-embed; an unstable # order would re-embed the whole catalog on every run. a, _ = merge_attributes([], {"material": "cotton", "color": "white"}) b, _ = merge_attributes([], {"color": "white", "material": "cotton"}) assert a == b def test_existing_attributes_are_not_reordered_away(): merged, _ = merge_attributes( [existing("size", "m"), existing("color", "white")], {"material": "cotton"}) assert {a["key"] for a in merged} == {"size", "color", "material"} ``` - [ ] **Step 2: Run test to verify it fails** Run: `.venv/bin/python -m pytest tests/unit/test_attribute_merge.py -v` Expected: FAIL with `ModuleNotFoundError` - [ ] **Step 3: Write the implementation** Create `app/services/attribute_merge.py`: ```python """Merges inferred attributes into existing ones without overwriting. A merchant's own value always wins. Extraction fills gaps only, and everything it produces is marked inferred with a confidence below any structured source, so the matching layer can weight a real value above a guess. """ import logging logger = logging.getLogger(__name__) INFERRED_CONFIDENCE = 0.5 # These get their own columns because they are filtered on. Storing them as # attributes as well would inflate the attribute count quality_score depends on. SCALAR_FIELDS = ("is_accessory", "price_tier") def merge_attributes(existing: list, inferred: dict) -> tuple: merged = list(existing or []) present = {a["key"] for a in merged} conflicts = 0 for key, value in sorted((inferred or {}).items()): if key in SCALAR_FIELDS: continue if key == "key_features": for feature in value: if not any(a["key"] == "feature" and a["value"] == feature for a in merged): merged.append(_inferred("feature", feature)) continue if key in present: prior = next(a for a in merged if a["key"] == key) if prior.get("value") != value: conflicts += 1 continue merged.append(_inferred(key, value)) return sorted(merged, key=lambda a: (a["key"], str(a["value"]))), conflicts def _inferred(key: str, value) -> dict: return {"key": key, "raw_value": value, "value": value, "unit": None, "source": "inferred", "confidence": INFERRED_CONFIDENCE} ``` - [ ] **Step 4: Run test to verify it passes** Run: `.venv/bin/python -m pytest tests/unit/test_attribute_merge.py -v` Expected: PASS, 9 tests - [ ] **Step 5: Report changed files** ```bash git status --short ``` Do not commit and do not stage. --- ### Task 5: Orchestration **Files:** - Create: `app/services/enrichment.py` - Test: `tests/unit/test_enrichment.py` - Test: `tests/integration/test_enrichment_db.py` **Interfaces:** - Consumes: `price_tiers.compute_bands`/`classify`, `attribute_extractor.extract_batch`/`BATCH_SIZE`, `attribute_merge.merge_attributes`, `get_db_connection`, `migrate_products_table`. - Produces: - `MIN_TEXT_WORDS: int` = 3 - `select_products(tenant_id: str, force: bool) -> list[dict]` - `is_extractable(product: dict) -> bool` - `enrich_tenant(tenant_id: str, force: bool = False) -> dict` returning the report - [ ] **Step 1: Write the failing test** Create `tests/unit/test_enrichment.py`: ```python import pytest from app.services.enrichment import is_extractable def test_a_product_with_a_description_is_extractable(): assert is_extractable({"name": "Oxford Shirt", "description": "A white cotton shirt."}) is True def test_a_long_title_alone_is_extractable(): assert is_extractable({"name": "Brown Full Grain Leather Belt", "description": None}) is True def test_a_short_title_with_no_description_is_not(): # No prompt fixes an absent input. Flag it rather than pay for a call that # can only guess. assert is_extractable({"name": "Belt", "description": None}) is False def test_no_name_at_all_is_not_extractable(): assert is_extractable({"name": None, "description": None}) is False def test_batches_are_capped(monkeypatch): import app.services.enrichment as mod from app.services.attribute_extractor import BATCH_SIZE seen = [] monkeypatch.setattr(mod, "extract_batch", lambda batch: seen.append(len(batch)) or {}) monkeypatch.setattr(mod, "select_products", lambda t, f: [ {"product_key": f"k{i}", "name": f"Product Number {i}", "description": "A description.", "attributes": [], "price_cents": 100, "taxonomy_path": [], "brand": None} for i in range(BATCH_SIZE + 5)]) monkeypatch.setattr(mod, "compute_bands", lambda t: (100, 200)) monkeypatch.setattr(mod, "persist_attributes", lambda *a, **kw: None) monkeypatch.setattr(mod, "migrate_products_table", lambda t: None) monkeypatch.setattr(mod, "flag_unextractable", lambda *a, **kw: None) monkeypatch.setattr(mod, "catalog_stats", lambda t: (BATCH_SIZE + 5, 0)) mod.enrich_tenant("org_x") assert max(seen) <= BATCH_SIZE assert sum(seen) == BATCH_SIZE + 5 def test_a_failed_batch_does_not_abort_the_run(monkeypatch): import app.services.enrichment as mod from app.services.attribute_extractor import BATCH_SIZE calls = {"n": 0} def half_fail(batch): calls["n"] += 1 if calls["n"] == 1: return {} # extract_batch already degrades to empty return {p["product_key"]: {"color": "white"} for p in batch} monkeypatch.setattr(mod, "extract_batch", half_fail) monkeypatch.setattr(mod, "select_products", lambda t, f: [ {"product_key": f"k{i}", "name": f"Product Number {i}", "description": "A description.", "attributes": [], "price_cents": 100, "taxonomy_path": [], "brand": None} for i in range(BATCH_SIZE * 2)]) monkeypatch.setattr(mod, "compute_bands", lambda t: (100, 200)) monkeypatch.setattr(mod, "persist_attributes", lambda *a, **kw: None) monkeypatch.setattr(mod, "migrate_products_table", lambda t: None) monkeypatch.setattr(mod, "flag_unextractable", lambda *a, **kw: None) monkeypatch.setattr(mod, "catalog_stats", lambda t: (BATCH_SIZE * 2, 0)) report = mod.enrich_tenant("org_x") assert report["extracted"] == BATCH_SIZE assert report["failed"] == BATCH_SIZE def test_report_shape(monkeypatch): import app.services.enrichment as mod monkeypatch.setattr(mod, "select_products", lambda t, f: []) monkeypatch.setattr(mod, "compute_bands", lambda t: None) monkeypatch.setattr(mod, "migrate_products_table", lambda t: None) monkeypatch.setattr(mod, "catalog_stats", lambda t: (0, 0)) report = mod.enrich_tenant("org_x") for key in ("products", "considered", "extracted", "skipped_unchanged", "skipped_unextractable", "failed", "conflicts", "attribute_coverage"): assert key in report def test_an_empty_catalog_reports_zero_coverage_not_a_crash(monkeypatch): import app.services.enrichment as mod monkeypatch.setattr(mod, "select_products", lambda t, f: []) monkeypatch.setattr(mod, "compute_bands", lambda t: None) monkeypatch.setattr(mod, "migrate_products_table", lambda t: None) monkeypatch.setattr(mod, "catalog_stats", lambda t: (0, 0)) assert mod.enrich_tenant("org_x")["attribute_coverage"] == 0.0 ``` - [ ] **Step 2: Run test to verify it fails** Run: `.venv/bin/python -m pytest tests/unit/test_enrichment.py -v` Expected: FAIL with `ModuleNotFoundError` - [ ] **Step 3: Write the implementation** Create `app/services/enrichment.py`: ```python """Runs attribute extraction across a tenant's catalog. Selection is driven by content_hash: a product is re-extracted only when its content actually changed. That is the real cost control -- the matching specification's quality-score gate was measured at a 0.35 median on live data, so it would catch nearly everything and control nothing. """ import json import logging from psycopg2 import sql from psycopg2.extras import Json, RealDictCursor from app.services.attribute_extractor import BATCH_SIZE, extract_batch from app.services.attribute_merge import merge_attributes from app.services.database import get_db_connection, migrate_products_table from app.services.price_tiers import classify, compute_bands logger = logging.getLogger(__name__) MIN_TEXT_WORDS = 3 def is_extractable(product: dict) -> bool: """No prompt fixes an absent input, so do not pay for a call that can only guess.""" if product.get("description"): return True name = product.get("name") or "" return len(name.split()) >= MIN_TEXT_WORDS def select_products(tenant_id: str, force: bool) -> list: where = "TRUE" if force else "(enriched_hash IS NULL OR enriched_hash <> content_hash)" conn = get_db_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute(sql.SQL( "SELECT product_key, name, description, brand, taxonomy_path, " "attributes, price_cents, price_reference_cents, content_hash " "FROM {}.strategist_products WHERE " + where ).format(sql.Identifier(tenant_id))) return [dict(r) for r in cur.fetchall()] finally: conn.close() def catalog_stats(tenant_id: str) -> tuple: """(total products, products holding at least one attribute).""" conn = get_db_connection() try: with conn.cursor() as cur: cur.execute(sql.SQL( "SELECT count(*), count(*) FILTER " "(WHERE jsonb_array_length(attributes) > 0) " "FROM {}.strategist_products" ).format(sql.Identifier(tenant_id))) return cur.fetchone() finally: conn.close() def persist_attributes(tenant_id: str, product_key: str, attributes: list, is_accessory, price_tier, content_hash) -> None: conn = get_db_connection() try: with conn.cursor() as cur: cur.execute(sql.SQL( "UPDATE {}.strategist_products SET attributes = %s, " "is_accessory = COALESCE(%s, is_accessory), " "price_tier = %s, enriched_hash = %s WHERE product_key = %s" ).format(sql.Identifier(tenant_id)), (Json(attributes), is_accessory, price_tier, content_hash, product_key)) conn.commit() except Exception: conn.rollback() logger.error("Could not persist attributes for %s", product_key, exc_info=True) raise finally: conn.close() def flag_unextractable(tenant_id: str, product_key: str) -> None: """Visible, not silent: a merchant should see a data gap rather than blame the recommendations.""" conn = get_db_connection() try: with conn.cursor() as cur: cur.execute(sql.SQL( "UPDATE {}.strategist_products " "SET missing_fields = array_append(missing_fields, 'attributes') " "WHERE product_key = %s AND NOT ('attributes' = ANY(missing_fields))" ).format(sql.Identifier(tenant_id)), (product_key,)) conn.commit() except Exception: conn.rollback() logger.error("Could not flag %s as unextractable", product_key, exc_info=True) finally: conn.close() def enrich_tenant(tenant_id: str, force: bool = False) -> dict: migrate_products_table(tenant_id) products = select_products(tenant_id, force) bands = compute_bands(tenant_id) extractable = [p for p in products if is_extractable(p)] skipped = [p for p in products if not is_extractable(p)] for product in skipped: flag_unextractable(tenant_id, product["product_key"]) extracted = failed = conflicts = 0 for start in range(0, len(extractable), BATCH_SIZE): batch = extractable[start:start + BATCH_SIZE] results = extract_batch(batch) if not results: failed += len(batch) continue for product in batch: inferred = results.get(product["product_key"]) if not inferred: failed += 1 continue current = product.get("attributes") or [] if isinstance(current, str): current = json.loads(current) merged, product_conflicts = merge_attributes(current, inferred) conflicts += product_conflicts price = product.get("price_reference_cents") or product.get("price_cents") # The computed band wins over the model's guess: the model sees one # batch and cannot know the tenant's price distribution, which is # the only thing "premium" means here. Its guess is the fallback for # a product with no price at all. persist_attributes( tenant_id, product["product_key"], merged, inferred.get("is_accessory"), classify(price, bands) or inferred.get("price_tier"), product.get("content_hash")) extracted += 1 total, with_attributes = catalog_stats(tenant_id) report = {"products": total, "considered": len(products), "extracted": extracted, "skipped_unchanged": total - len(products), "skipped_unextractable": len(skipped), "failed": failed, "conflicts": conflicts, "attribute_coverage": round(with_attributes / total, 3) if total else 0.0} logger.info("Enriched %s: %s", tenant_id, report) return report ``` - [ ] **Step 4: Add the `enriched_hash` column** `select_products` and `persist_attributes` both reference `enriched_hash`, which records the `content_hash` a product was last extracted at. Without a separate column, a re-run after a sync would re-extract everything, because `content_hash` alone cannot say whether extraction has happened. In `app/services/database.py`, append to `_PRODUCT_COLUMNS` and to `bootstrap_tenant`'s CREATE TABLE: ```python ("enriched_hash", "TEXT"), ``` Add `"enriched_hash"` to `EXPECTED_COLUMNS` in `tests/integration/test_products_table.py` and to `EXTRACTION_COLUMNS` in `tests/integration/test_products_migration.py`. - [ ] **Step 5: Run the unit test** Run: `.venv/bin/python -m pytest tests/unit/test_enrichment.py -v` Expected: PASS, 8 tests - [ ] **Step 6: Write the database test** Create `tests/integration/test_enrichment_db.py`: ```python import pytest from app.services.database import bootstrap_tenant, get_db_connection from app.services.enrichment import enrich_tenant, select_products def _insert(schema, key, name, description, content_hash): conn = get_db_connection() try: with conn.cursor() as cur: cur.execute( f'INSERT INTO "{schema}".strategist_products ' "(product_key, name, description, product_url, price_cents, " " content_hash) VALUES (%s,%s,%s,%s,%s,%s)", (key, name, description, f"https://x.example/{key}", 1000, content_hash)) conn.commit() finally: conn.close() @pytest.fixture def catalog(temp_tenant): bootstrap_tenant(temp_tenant) _insert(temp_tenant, "s:1", "Oxford Shirt", "A white cotton shirt.", "hash-a") _insert(temp_tenant, "s:2", "Belt", None, "hash-b") return temp_tenant def test_unextracted_products_are_selected(catalog): keys = {p["product_key"] for p in select_products(catalog, force=False)} assert keys == {"s:1", "s:2"} def test_an_already_extracted_product_is_skipped(catalog): conn = get_db_connection() try: with conn.cursor() as cur: cur.execute(f'UPDATE "{catalog}".strategist_products ' "SET enriched_hash = content_hash WHERE product_key = 's:1'") conn.commit() finally: conn.close() keys = {p["product_key"] for p in select_products(catalog, force=False)} assert keys == {"s:2"} def test_force_reselects_everything(catalog): conn = get_db_connection() try: with conn.cursor() as cur: cur.execute(f'UPDATE "{catalog}".strategist_products ' "SET enriched_hash = content_hash") conn.commit() finally: conn.close() assert len(select_products(catalog, force=True)) == 2 def test_an_unextractable_product_is_flagged(catalog, monkeypatch): import app.services.enrichment as mod monkeypatch.setattr(mod, "extract_batch", lambda batch: {p["product_key"]: {"color": "white"} for p in batch}) report = enrich_tenant(catalog) assert report["skipped_unextractable"] == 1 conn = get_db_connection() try: with conn.cursor() as cur: cur.execute(f'SELECT missing_fields FROM "{catalog}".strategist_products ' "WHERE product_key = 's:2'") assert "attributes" in cur.fetchone()[0] finally: conn.close() def test_extraction_writes_attributes_and_marks_the_hash(catalog, monkeypatch): import app.services.enrichment as mod monkeypatch.setattr(mod, "extract_batch", lambda batch: {p["product_key"]: {"color": "white", "is_accessory": False} for p in batch}) enrich_tenant(catalog) conn = get_db_connection() try: with conn.cursor() as cur: cur.execute(f'SELECT attributes, is_accessory, enriched_hash ' f'FROM "{catalog}".strategist_products ' "WHERE product_key = 's:1'") attributes, is_accessory, enriched_hash = cur.fetchone() finally: conn.close() assert any(a["key"] == "color" for a in attributes) assert is_accessory is False assert enriched_hash == "hash-a" ``` - [ ] **Step 7: Run the database test** Run: `.venv/bin/python -m pytest tests/integration/test_enrichment_db.py -v` Expected: PASS, 5 tests - [ ] **Step 8: Report changed files** ```bash git status --short ``` Do not commit and do not stage. --- ### Task 6: The endpoint **Files:** - Modify: `app/api/endpoints.py` - Test: `tests/integration/test_enrich_route.py` **Interfaces:** - Consumes: `enrichment.enrich_tenant`. - Produces: `POST /catalog/enrich`. - [ ] **Step 1: Write the failing test** Create `tests/integration/test_enrich_route.py`: ```python import pytest from fastapi.testclient import TestClient from app.main import app client = TestClient(app) TENANT = "org_8c32bf3e-6a18-4739-9b1c-94c0cf11125f" def test_tenant_id_is_required(): assert client.post("/catalog/enrich", data={}).status_code == 422 def test_route_is_registered(): assert "/catalog/enrich" in set(app.openapi()["paths"]) def test_returns_the_report(monkeypatch): from app.api import endpoints monkeypatch.setattr(endpoints, "enrich_tenant", lambda t, force=False: { "products": 218, "extracted": 211, "skipped_unextractable": 7, "failed": 0, "conflicts": 3}) resp = client.post("/catalog/enrich", data={"tenant_id": TENANT}) assert resp.status_code == 200 assert resp.json()["extracted"] == 211 def test_force_is_passed_through(monkeypatch): from app.api import endpoints seen = {} monkeypatch.setattr(endpoints, "enrich_tenant", lambda t, force=False: seen.update(force=force) or {}) client.post("/catalog/enrich", data={"tenant_id": TENANT, "force": "true"}) assert seen["force"] is True def test_a_failure_returns_502_not_500(monkeypatch): # Enrichment is a background-ish operation; a model outage is an upstream # fault, not a bug in this service. from app.api import endpoints def boom(t, force=False): raise RuntimeError("model unavailable") monkeypatch.setattr(endpoints, "enrich_tenant", boom) resp = client.post("/catalog/enrich", data={"tenant_id": TENANT}) assert resp.status_code == 502 assert "RuntimeError" in resp.text # class name only, no detail ``` - [ ] **Step 2: Run test to verify it fails** Run: `.venv/bin/python -m pytest tests/integration/test_enrich_route.py -v` Expected: FAIL — the route 404s. - [ ] **Step 3: Write the implementation** Add to the imports in `app/api/endpoints.py`: ```python from app.services.enrichment import enrich_tenant ``` And add the route, following the style of `/catalog/sync` in the same file: ```python @router.post("/catalog/enrich") async def catalog_enrich_endpoint( tenant_id: str = Form(...), force: bool = Form(False), ): """Run attribute extraction over a tenant's catalog. Separate from /catalog/sync because it costs money and must be re-runnable without re-fetching a catalog. """ try: return await anyio.to_thread.run_sync( lambda: enrich_tenant(normalise_tenant_id(tenant_id), force)) except Exception as ex: logger.error("Enrichment failed for %s", tenant_id, exc_info=True) raise HTTPException(status_code=502, detail=type(ex).__name__) ``` Check whether `normalise_tenant_id` exists in that module and is used by the neighbouring routes; if it does not, pass `tenant_id` directly. - [ ] **Step 4: Run the tests** Run: `.venv/bin/python -m pytest tests/integration/test_enrich_route.py -v` Expected: PASS, 5 tests - [ ] **Step 5: Run both suites** Run `.venv/bin/python -m pytest tests/unit/ -q`, then `.venv/bin/python -m pytest tests/integration/ -q`, sequentially in the foreground. Expected: PASS, no regressions. Baselines were 502 unit and 54 integration. - [ ] **Step 6: Report changed files** ```bash git status --short ``` Do not commit and do not stage. --- ### Task 7: Live extraction **Files:** none — verification only. **Prerequisites:** uvicorn running from this worktree on port 8001, Postgres reachable, the live tenant holding 218 products. - [ ] **Step 1: Record the starting point** ```bash PYTHONPATH=. .venv/bin/python -c " from app.services.database import get_db_connection T='org_8c32bf3e-6a18-4739-9b1c-94c0cf11125f' c=get_db_connection(); cur=c.cursor() cur.execute('SELECT count(*) FROM \"'+T+'\".strategist_products') total = cur.fetchone()[0] cur.execute('SELECT count(*) FROM \"'+T+'\".strategist_products WHERE jsonb_array_length(attributes) > 0') print('products:', total, 'with attributes:', cur.fetchone()[0]) c.close()" ``` Expected: 218 products, 7 with attributes. That 7 is the number this phase exists to move. - [ ] **Step 2: Start the server and run extraction** ```bash .venv/bin/uvicorn app.main:app --port 8001 & sleep 8 curl -s -X POST http://localhost:8001/catalog/enrich \ -F "tenant_id=org_8c32bf3e-6a18-4739-9b1c-94c0cf11125f" | python3 -m json.tool ``` Expected: roughly 11 batches, `extracted` near 211, `failed` low, `skipped_unextractable` around 7 — the crawled products, which have no description. A high `conflicts` count means the prompt disagrees with merchant data often, which is worth reporting rather than ignoring. - [ ] **Step 3: Confirm coverage moved** ```bash PYTHONPATH=. .venv/bin/python -c " from app.services.database import get_db_connection T='org_8c32bf3e-6a18-4739-9b1c-94c0cf11125f' c=get_db_connection(); cur=c.cursor() cur.execute('SELECT count(*) FROM \"'+T+'\".strategist_products WHERE jsonb_array_length(attributes) > 0') print('with attributes:', cur.fetchone()[0], 'of 218') cur.execute('SELECT is_accessory, count(*) FROM \"'+T+'\".strategist_products GROUP BY 1') print('is_accessory:', cur.fetchall()) cur.execute('SELECT price_tier, count(*) FROM \"'+T+'\".strategist_products GROUP BY 1 ORDER BY 2 DESC') print('price_tier:', cur.fetchall()) c.close()" ``` Expected: attribute coverage near complete, and a genuine spread across the three price tiers rather than everything in one — a single-bucket result means the bands were computed wrongly. - [ ] **Step 4: Confirm `is_accessory` found the accessory categories** ```bash PYTHONPATH=. .venv/bin/python -c " from app.services.database import get_db_connection T='org_8c32bf3e-6a18-4739-9b1c-94c0cf11125f' c=get_db_connection(); cur=c.cursor() cur.execute(\"SELECT category, count(*) FILTER (WHERE is_accessory) AS acc, count(*) FROM \\\"\"+T+\"\\\".strategist_products WHERE category IN ('mobile-accessories','sports-accessories','kitchen-accessories','smartphones','groceries') GROUP BY 1 ORDER BY 1\") for r in cur.fetchall(): print(' ', r) c.close()" ``` Expected: the three `*-accessories` categories mostly true, `smartphones` and `groceries` mostly false. **This is the single most important result of the phase** — the later complement work uses `is_accessory` to tell a phone case from another phone, and the online ranker exempts complements from the price rule that would otherwise crush a cheap accessory shown against an expensive anchor. - [ ] **Step 5: Confirm a second run is cheap** Re-run step 2. Expected: `extracted: 0`, everything skipped as unchanged, because `enriched_hash` now matches `content_hash`. A non-zero count means the caching is not working and every run would cost money. - [ ] **Step 6: Report** Summarise: changed files, test counts, the enrichment report, the before-and-after attribute coverage, the `is_accessory` breakdown by category, the price-tier spread, and confirmation that a second run extracted nothing. Remind the user to commit. --- ## Follow-on work (not in this plan) - **Phase 2:** `product_neighbors`, `pairing_decisions`, the pairing job with four sources including upsell, and the category → product → pairing browse endpoints. - **Phase 3:** complement inference with compatible attribute values, and the approval queue. - **Phase 4:** online ranking, replacing `decide()`'s internals. - **Phase 5:** measurement counters. - **Phase 6:** the seven known gaps, including per-size stock, which matters for apparel tenants and is where `size_system` extracted here becomes load-bearing.