"""One suggestion, opened up: everything a person needs to answer it without leaving the page. Sources of truth: this module, `proposals.py` (the sheet and the approval), `routers/review.py` (`GET /api/review/rows/{uid}`), `tests/test_review_detail.py`. The sheet's row is the headline: what is being asked, about what, and how sure. That is the right size for a list of twenty questions and far too small to answer a hard one. Rian, 18 Sep: *"What product variants are involved? What attributes and properties are being merged? What conflicts are there and how are we proposing to handle those? Eg: different barcodes, which one are we taking?"* So this assembles, for whichever shape the row is: the things being folded with their variants, the listings behind them with the shop's own text and a link to the source, the attributes that would be carried, the conflicts a merge would have to resolve with what we propose to do about each, and the decision ledger for the rows involved. It writes nothing. A conflict is never resolved here. It is reported so a person can see it and decide; the writer's own rules (a stated attribute that disagrees vetoes a merge) still hold at approval. """ from __future__ import annotations import uuid as _uuid from typing import Any from sqlalchemy import func, select from sqlalchemy.orm import Session from app.models import Brand, Listing, ProductLine, ProductVariant, Proposal, ProposalPass, Retailer, Shop from app.services.decisions import natural_keys #: How many listings one row shows before it says how many more there are. A brand fold can name #: several hundred; a person scanning for the odd one out does not need them all at once. LISTING_CAP = 60 #: The kinds of suggestion a pass makes, as a person reads them. Rian, 18 Sep: an eyebrow on the #: card, and "a growing list of types of decisions that we can maintain in the AI guidelines". #: Keep this list and `docs/AI-REVIEW-GUIDELINES.md` saying the same thing. CATEGORIES: dict[str, str] = { "combine-brands": "Combine brands", "combine-product-lines": "Combine product lines", "combine-product-variants": "Combine product variants", "keep-separate": "Confirm separate", "name-a-brand": "Name a brand", "name-a-product-line": "Name a product line", "move-a-product-variant": "Move a product variant", "set-an-attribute": "Set an attribute", "other": "Waiting for an answer", } def category_of(row: Proposal) -> str: """Which kind of suggestion this is. The card shows the words; this is the key.""" value = row.corrected_value if row.corrected_value is not None else row.value if row.entity_type == "suggestion" and row.field == "decision": if isinstance(value, dict) and value.get("decision") == "separate": return "keep-separate" level = row.natural_key.split(":")[1] if row.natural_key.startswith("pair:") else "" return {"brand": "combine-brands", "line": "combine-product-lines", "product": "combine-product-variants"}.get(level, "other") if row.field == "merged_into": return "combine-product-variants" if row.field == "alias_of": return "combine-brands" if row.entity_type == "brand" else "combine-product-lines" if row.field == "product_line": return "move-a-product-variant" if row.field == "name": if row.entity_type == "brand": return "name-a-brand" # A header that absorbs other lines is a fold; one that only names a line is a naming. absorbs = ((row.natural_key_detail or {}).get("absorbs") or []) + ((row.detail or {}).get("absorbs_derived") or []) return "combine-product-lines" if absorbs else "name-a-product-line" if row.field.startswith("attribute:"): return "set-an-attribute" return "other" def question_of(row: Proposal) -> str: """The question in a person's words, computed once so the compact row and the expanded view cannot disagree about what is being asked.""" if row.entity_type == "suggestion" and row.field == "decision": level = row.natural_key.split(":")[1] if row.natural_key.startswith("pair:") else "" return {"brand": "Same brand?", "line": "Same product line?", "product": "Same product variant?"}.get(level, "Same, or separate?") if row.field == "merged_into": return "Same product variant?" if row.field == "product_line": return "Does this belong in this product line?" if row.field == "name" and row.entity_type == "product_line": return "Is this the product line's name?" if row.field == "name" and row.entity_type == "brand": return "Is this the brand's name?" if row.field == "alias_of": return "Is this the same thing?" if row.field.startswith("attribute:"): return "Is this attribute right?" return "Waiting for an answer" def _shop_of(db: Session, listing: Listing, shops: dict, retailers: dict, variant_names: dict[int, str] | None = None) -> dict: """One listing as a person reads it: which shop, the shop's own words, and the link. `variant_name` is the product variant this listing was filed under, and it is the column that answers the question a person actually opens a row to ask. Rian, 18 Sep: *"Obviously we're getting the word 'Phantom' from somewhere ... what raw data resulted in us deciding this is a product line called Phantom."* `listed_*` is the shop's own wording, and it is empty on every listing that has not been collected since the run that began keeping each shop's fragment: the correlation is exact, and two pages of the same retailer differ only in when each was last seen. So an empty row means stale, not unreadable, and `last_seen` is carried here to say so. Calling it "this shop does not send its own wording" was wrong, and wrong in the direction that makes a person stop asking. """ shop = shops.get(listing.shop_id) retailer = retailers.get(shop.retailer_id) if shop is not None else None return { "key": natural_keys.listing_key(listing), "variant_uid": None, # filled by the caller, which knows the variant it asked for "variant_name": (variant_names or {}).get(listing.variant_id), "source_sku": listing.source_sku, "shop": shop.code if shop else None, "shop_name": shop.name if shop else None, "airport": shop.iata if shop else None, "retailer": retailer.name if retailer else None, "listed_brand": listing.listed_brand, "listed_name": listing.listed_name, "listed_variant": listing.listed_variant, "listed_quantity_text": listing.listed_quantity_text, "gtin": listing.listed_gtin, #: When this listing was last collected. An empty `listed_*` is explained by this date. "last_seen": listing.last_seen_at, # The shop's own page. It is the only way to check a reading against the source, so a # listing without one says so rather than rendering a dead link. "url": listing.url, } def _variant_out(db: Session, v: ProductVariant, lines: dict, counts: dict[int, int] | None = None) -> dict: line = lines.get(v.product_line_id) return { "uid": str(v.uid), "name": v.name, "gtin": v.gtin, "quantity_ml": v.quantity_ml, "quantity_state": v.quantity_state, "attributes": v.attributes or {}, "line_name": line.name if line is not None else None, "product_line_key": f"line:{line.uid}" if line is not None else None, "listings": int((counts or {}).get(v.id, 0)), } def listings_under(db: Session, *, line_uid: str | None = None, variant_uid: str | None = None, limit: int = 5, offset: int = 0) -> dict[str, Any]: """The listings under one product line or one product variant, read when the row is opened. Rian, 18 Sep: *"when I'm eyeballing the list ... and I see something that I may want to look more closely at, I can expand out the raw data from the listings and find the link to the actual source."* So this is per row and paged, top few first. It also answers *"why do some of them have no listings?"*, which was the card's cap and not the catalogue: the detail view fetched one capped array of listings for the whole question and the nested rows filtered it, so once a brand fold's 60-listing budget ran out every remaining variant read as having none, while the count beside its name said otherwise. Reading per row cannot disagree with itself that way. A row that really has none has had its listings merged into another variant's, or is a line whose variants have all been merged away. """ q = select(Listing).where(Listing.ignored_at.is_(None)) if variant_uid: v = db.scalar(select(ProductVariant).where(ProductVariant.uid == _uuid.UUID(str(variant_uid)))) if v is None: return {"listings": [], "total": 0, "shown": 0} q = q.where(Listing.variant_id == v.id) elif line_uid: line = db.scalar(select(ProductLine).where(ProductLine.uid == _uuid.UUID(str(line_uid)))) if line is None: return {"listings": [], "total": 0, "shown": 0} q = q.where(Listing.variant_id.in_( select(ProductVariant.id).where(ProductVariant.product_line_id == line.id, ProductVariant.merged_into_id.is_(None)))) else: return {"listings": [], "total": 0, "shown": 0} total = db.scalar(select(func.count()).select_from(q.subquery())) or 0 shops = {s.id: s for s in db.scalars(select(Shop))} retailers = {r.id: r for r in db.scalars(select(Retailer))} rows = list(db.scalars(q.order_by(Listing.id).offset(max(0, offset)).limit(max(1, min(limit, 200))))) held = list(db.scalars(select(ProductVariant).where(ProductVariant.id.in_({l.variant_id for l in rows} or {0})))) uid_of = {v.id: str(v.uid) for v in held} named = {v.id: v.name for v in held} return {"listings": [{**_shop_of(db, l, shops, retailers, named), "variant_uid": uid_of.get(l.variant_id)} for l in rows], "total": int(total), "shown": len(rows)} def _conflicts(variants: list[dict]) -> list[dict]: """What a merge would have to resolve, and what we propose. Reported, never applied. Rian's own example is the first one: two variants with different barcodes cannot both be right, and a person needs to see WHICH before the merge happens, not after.""" out: list[dict] = [] if len(variants) < 2: return out def spread(field: str, label: str, what: str) -> None: seen: dict[Any, list[str]] = {} for v in variants: value = v.get(field) if value in (None, "", {}): continue seen.setdefault(value, []).append(v["name"]) if len(seen) > 1: out.append({"field": field, "label": label, "values": [{"value": str(k), "on": names} for k, names in seen.items()], "proposed": None, "what": what}) spread("gtin", "Barcode", "Two barcodes name two products. Nothing is chosen here: a merge " "that would drop one is refused until a person says which is right.") spread("quantity_ml", "Quantity", "The quantity is the unit a price is compared on, so two " "values would compare different things.") kinds: dict[str, dict[str, list[str]]] = {} for v in variants: for kind, value in (v.get("attributes") or {}).items(): if value in (None, "", {}): continue kinds.setdefault(kind, {}).setdefault(str(value), []).append(v["name"]) for kind, values in kinds.items(): if len(values) > 1: out.append({"field": f"attribute:{kind}", "label": kind.replace("_", " ").title(), "values": [{"value": k, "on": names} for k, names in values.items()], "proposed": None, "what": "Two stated values for one attribute. A merge is vetoed while they " "disagree; keeping them apart is the safe answer."}) return out def _fold_key(text: str | None) -> str: """Case and accents folded, for spotting two product lines that carry the same name.""" import unicodedata if not text: return "" return unicodedata.normalize("NFKD", text).encode("ascii", "ignore").decode().lower().strip() def consequences(db: Session, row: Proposal, subjects: list[dict]) -> dict[str, Any]: """What answering unlocks, as a SHAPE and never as the questions themselves. Rian, 18 Sep: *"I'm nervous to click confirm because I dont want to create a mess, but I do want to see what comes next."* Showing the queued decisions would bias the answer he has not made yet; showing how many there will be and what kind does not. So this counts and characterises, and names nothing it would be proposing. Two things have to stay true here, because a wrong promise is worse than no promise. Same-named product lines on a brand fold are NOT questions that follow: the applier points each at the line of the same key on the survivor as a consequence of the confirm, so they happen at the moment of the click. Saying "about 15 product line questions follow, they do not exist yet" described neither -- rian caught it by asking whether they were queued. And it counts them on `ProductLine.key`, the key the applier itself matches on, so the number shown is the number that happens rather than a near-enough guess from the names. """ blocked = db.scalar(select(func.count(Proposal.id)).where( Proposal.brand_slug.in_([s.get("slug") for s in subjects if s.get("slug")] or [""]), Proposal.status == "open", Proposal.id != row.id)) or 0 out: dict[str, Any] = {"if_confirmed": [], "if_separate": [], "blocked_now": int(blocked)} if row.natural_key.startswith("pair:brand:") and len(subjects) == 2: value = row.corrected_value if row.corrected_value is not None else row.value keep = (value or {}).get("survivor") if isinstance(value, dict) else None winner = next((s for s in subjects if s["key"] == keep), subjects[0]) loser = next((s for s in subjects if s["key"] != winner["key"]), subjects[-1]) w = db.scalar(select(Brand).where(Brand.slug == winner.get("slug"))) l = db.scalar(select(Brand).where(Brand.slug == loser.get("slug"))) merging = arriving = 0 if w is not None and l is not None: theirs = {r.key for r in db.scalars( select(ProductLine).where(ProductLine.brand_id == w.id, ProductLine.alias_of_id.is_(None)))} moving = list(db.scalars( select(ProductLine).where(ProductLine.brand_id == l.id, ProductLine.alias_of_id.is_(None)))) merging = sum(1 for r in moving if r.key in theirs) arriving = len(moving) - merging # Which row survives and what it is called are two decisions on one confirm (the 18 Sep # simulation): the direction decides how many product variants move and how many product # line addresses are kept, and the card said neither. Both directions, counted the way # the applier counts them, so the person can see what each way round does. out["directions"] = [] for keep, fold in ((w, l), (l, w)): kept_keys = {r.key for r in db.scalars(select(ProductLine).where(ProductLine.brand_id == keep.id, ProductLine.alias_of_id.is_(None)))} folding = list(db.scalars(select(ProductLine).where(ProductLine.brand_id == fold.id, ProductLine.alias_of_id.is_(None)))) n_variants = db.scalar(select(func.count(ProductVariant.id)).where(ProductVariant.brand_id == fold.id, ProductVariant.merged_into_id.is_(None))) or 0 out["directions"].append({ "survivor": f"brand:{keep.slug}", "survivor_name": keep.name, "folded_name": fold.name, "variants_move": int(n_variants), "lines_fold": sum(1 for r in folding if r.key in kept_keys), "lines_arrive": sum(1 for r in folding if r.key not in kept_keys), "addresses_kept": len(kept_keys), "chosen": keep.id == w.id}) out["name_note"] = (f"The display name is a second decision: the surviving row keeps its address /brands/{w.slug}/ " f"whichever name you give it, and a name typed at confirm is recorded on it.") out["if_confirmed"] = [ {"what": f"{loser['lines']:,} product lines and {loser['variants']:,} product variants move to " f"{winner['name']}. Touches {loser['listings']:,} listings.", "count": None}, {"what": f"{merging:,} of those product lines fold straight into a product line of the same " f"name, and {arriving:,} arrive as product lines of their own. That happens on confirm, not later.", "count": merging} if merging or arriving else None, ] out["if_separate"] = [ {"what": "nothing moves, and the two stay two brands.", "count": None}, {"what": "the pair is never proposed as one again.", "count": None}, {"what": "another pass is still needed, on each brand separately, to look at product lines " "and what sits under them.", "count": None}, ] elif row.field == "merged_into" or row.natural_key.startswith("pair:product:"): # A product variant merge is the thinnest card and the most valuable change (the 18 Sep # simulation): it puts retailers who cannot see each other on one page. Say which. uids = [] for s in subjects: try: kind, parts = natural_keys.parse(s["key"]) except ValueError: continue if kind == "product_variant" and parts.get("uid"): uids.append(_uuid.UUID(parts["uid"])) by_side: dict[str, set[str]] = {} if uids: for uid, retailer in db.execute(select(ProductVariant.uid, Retailer.slug).join(Listing, Listing.variant_id == ProductVariant.id) .join(Shop, Listing.shop_id == Shop.id).join(Retailer, Shop.retailer_id == Retailer.id) .where(ProductVariant.uid.in_(uids), Listing.ignored_at.is_(None))): by_side.setdefault(str(uid), set()).add(retailer) joined = sorted(set().union(*by_side.values())) if by_side else [] only_here = any(len(v) and not (v & set().union(*(o for k, o in by_side.items() if k != key))) for key, v in by_side.items()) out["joins_retailers"] = joined out["if_confirmed"] = [ {"what": (f"one product variant stocked at {len(joined)} retailer{'s' if len(joined) != 1 else ''} ({', '.join(joined)}) " f"on one page, where each row held its own" + (": a price comparison that did not exist." if only_here and len(joined) > 1 else ".")) if joined else "the two rows become one product variant.", "count": len(joined) or None}, {"what": "the barcode is kept, the fuller name carried, every listing and award moves to the survivor.", "count": None}, ] out["if_separate"] = [{"what": "they stay two product variants on two pages, and this pair is not proposed again.", "count": None}] else: moved = sum(s["variants"] for s in subjects[1:]) if len(subjects) > 1 else 0 out["if_confirmed"] = [{"what": f"{moved:,} product variants move." if moved else "the value is set.", "count": moved or None}] out["if_separate"] = [{"what": "nothing changes, and this is not proposed again.", "count": None}] if blocked: note = {"what": f"{blocked:,} suggestion{'' if blocked == 1 else 's'} already waiting on this brand " f"become{'s' if blocked == 1 else ''} answerable.", "count": int(blocked)} out["if_confirmed"].append(note) out["if_separate"].append(note) if row.natural_key.startswith("pair:brand:"): out["if_confirmed"].append( {"what": "another pass then looks at what is left, mostly product line combinations. Most come back " "as confident confirmations; a few set a precedent and need you.", "count": None}) out["if_confirmed"] = [x for x in out["if_confirmed"] if x] out["if_separate"] = [x for x in out["if_separate"] if x] return out def read(db: Session, uid: str) -> dict[str, Any]: """Everything behind one suggestion. Returns `{}` when no row answers the uid.""" try: row = db.scalar(select(Proposal).where(Proposal.uid == _uuid.UUID(str(uid)))) except (ValueError, AttributeError, TypeError): row = None if row is None: return {} shops = {s.id: s for s in db.scalars(select(Shop))} retailers = {r.id: r for r in db.scalars(select(Retailer))} pass_row = db.get(ProposalPass, row.pass_id) # What the question is ABOUT: a pair's two sides, a header's absorbed lines, or the row's own # entity. Each side carries its own shape so a person sees the size of what they are joining. subject_keys: list[str] = [] detail = row.natural_key_detail or {} if row.entity_type == "suggestion": subject_keys = [k for k in (detail.get("left"), detail.get("right")) if k] if not subject_keys: try: _, parts = natural_keys.parse(row.natural_key) subject_keys = [parts["left"], parts["right"]] except (ValueError, KeyError): subject_keys = [] else: # Deduped: the loader DERIVES the lines a header absorbs, so a file that also named them # produced the same line twice and the card listed "One Million Gold" under itself. subject_keys = [row.natural_key] for key in (detail.get("absorbs") or []) + ((row.detail or {}).get("absorbs_derived") or []): if key and key not in subject_keys: subject_keys.append(key) # What happens to each side, which is the thing a person is actually agreeing to. Listing a # name and a count said nothing: "Million Gold 4 variants / One Million Gold 2 variants" does # not say which survives (rian, 18 Sep). category = category_of(row) keeps = row.entity_type != "suggestion" # a header keeps its own row; a pair names a survivor survivor_key = None if row.entity_type == "suggestion": value = row.corrected_value if row.corrected_value is not None else row.value survivor_key = (value or {}).get("survivor") if isinstance(value, dict) else None if survivor_key is None and subject_keys: survivor_key = subject_keys[0] winner_key = survivor_key if row.entity_type == "suggestion" else (subject_keys[0] if subject_keys else None) survivor_name: list[str] = [] if winner_key: try: kind, parts = natural_keys.parse(winner_key) if kind == "brand": found = db.scalar(select(Brand).where(Brand.slug == parts.get("slug"))) elif kind == "product_line" and parts.get("uid"): found = db.scalar(select(ProductLine).where(ProductLine.uid == _uuid.UUID(parts["uid"]))) elif kind == "product_variant" and parts.get("uid"): found = db.scalar(select(ProductVariant).where(ProductVariant.uid == _uuid.UUID(parts["uid"]))) else: found = None if found is not None: survivor_name.append(found.name) except (ValueError, KeyError): pass subjects: list[dict] = [] variant_rows: list[ProductVariant] = [] def becomes(key: str, first: bool) -> str: """What happens to this side, said plainly. It NAMES the survivor rather than pointing at it: "becomes the one above" was wrong the first time a pair listed its loser first, which is most of them, because the sides are sorted by key and not by which one wins. """ if category == "keep-separate": return "stays separate" winner = survivor_key if row.entity_type == "suggestion" else (subject_keys[0] if subject_keys else None) if key == winner: return "stays as it is" name = survivor_name[0] if survivor_name else None return f"becomes {name}" if name else "is folded into the other" for key in subject_keys: try: kind, parts = natural_keys.parse(key) except ValueError: continue if kind == "brand": b = db.scalar(select(Brand).where(Brand.slug == parts.get("slug"))) if b is None: subjects.append({"key": key, "kind": kind, "name": None, "variants": 0, "listings": 0, "becomes": "not on this host"}) continue held = list(db.scalars(select(ProductVariant).where(ProductVariant.brand_id == b.id, ProductVariant.merged_into_id.is_(None)))) n_listings = db.scalar(select(func.count(Listing.id)).join( ProductVariant, ProductVariant.id == Listing.variant_id).where( ProductVariant.brand_id == b.id, Listing.ignored_at.is_(None))) or 0 subjects.append({"key": key, "kind": kind, "name": b.name, "slug": b.slug, "becomes": becomes(key, not subjects), "variants": len(held), "listings": int(n_listings), "lines": int(db.scalar(select(func.count(func.distinct(ProductVariant.product_line_id))) .where(ProductVariant.brand_id == b.id, ProductVariant.merged_into_id.is_(None))) or 0)}) variant_rows.extend(held[:LISTING_CAP]) elif kind == "product_line": line = db.scalar(select(ProductLine).where(ProductLine.uid == _uuid.UUID(parts["uid"]))) \ if parts.get("uid") else None if line is None: subjects.append({"key": key, "kind": kind, "name": parts.get("slug"), "variants": 0, "listings": 0, "becomes": "would be created"}) continue held = list(db.scalars(select(ProductVariant).where(ProductVariant.product_line_id == line.id, ProductVariant.merged_into_id.is_(None)))) subjects.append({"key": key, "kind": kind, "name": line.name, "slug": line.slug, "becomes": becomes(key, not subjects), "variants": len(held), "listings": 0}) variant_rows.extend(held) elif kind == "product_variant": v = db.scalar(select(ProductVariant).where(ProductVariant.uid == _uuid.UUID(parts["uid"]))) \ if parts.get("uid") else None if v is not None: subjects.append({"key": key, "kind": kind, "name": v.name, "variants": 1, "listings": 0, "becomes": becomes(key, not subjects)}) variant_rows.append(v) lines = {l.id: l for l in db.scalars(select(ProductLine))} if variant_rows else {} # Every variant's true listing count, so the table can say "3 listings" on a row whose listings # are fetched only when it is opened. Counting from the capped array instead is what made rows # read as empty. counts: dict[int, int] = dict(db.execute( select(Listing.variant_id, func.count(Listing.id)) .where(Listing.variant_id.in_([v.id for v in variant_rows]), Listing.ignored_at.is_(None)) .group_by(Listing.variant_id)).all()) if variant_rows else {} variants = [_variant_out(db, v, lines, counts) for v in variant_rows] listings: list[dict] = [] total = 0 if variant_rows: ids = [v.id for v in variant_rows] total = db.scalar(select(func.count(Listing.id)).where(Listing.variant_id.in_(ids), Listing.ignored_at.is_(None))) or 0 uid_of = {v.id: str(v.uid) for v in variant_rows} named = {v.id: v.name for v in variant_rows} for l in db.scalars(select(Listing).where(Listing.variant_id.in_(ids), Listing.ignored_at.is_(None)) .order_by(Listing.listed_brand, Listing.listed_name).limit(LISTING_CAP)): listings.append({**_shop_of(db, l, shops, retailers, named), "variant_uid": uid_of.get(l.variant_id)}) merging_variants = row.field == "merged_into" or ( row.entity_type == "suggestion" and row.natural_key.startswith("pair:product:")) # The product lines involved, which is the level a brand fold is really about: 54 variants are # hard to read, 24 product lines are not. Each carries its variants so it can be opened. lines_out: list[dict] = [] if variant_rows: seen_line: dict[int, dict] = {} for v in variant_rows: line = lines.get(v.product_line_id) if v.product_line_id else None key = v.product_line_id or 0 slot = seen_line.get(key) if slot is None: slot = {"key": f"line:{line.uid}" if line is not None else "line:none", "name": line.name if line is not None else "no product line", "slug": line.slug if line is not None else None, "brand": (db.get(Brand, line.brand_id).name if line is not None and line.brand_id else None), "variant_uids": [], "listings": 0} seen_line[key] = slot lines_out.append(slot) slot["variant_uids"].append(str(v.uid)) by_uid = {str(v.uid): v.id for v in variant_rows} for slot in lines_out: slot["listings"] = sum(int(counts.get(by_uid.get(u, 0), 0)) for u in slot["variant_uids"]) lines_out.sort(key=lambda r: (-len(r["variant_uids"]), r["name"] or "")) ledger: list[dict] = [] if row.entity_id is not None: from app.services import proposals as review ledger = review.decisions_for(db, row.entity_type, row.entity_id, 50) return { "uid": str(row.uid), "question": question_of(row), "field": row.field, "category": category, "category_label": CATEGORIES.get(category, CATEGORIES["other"]), "entity_type": row.entity_type, "natural_key": row.natural_key, "status": row.status, "confidence": float(row.confidence) if row.confidence is not None else None, "reason": row.reason, "pass_name": pass_row.name if pass_row else None, "is_brand_fold": row.entity_type == "suggestion" and row.natural_key.startswith("pair:brand:"), "subjects": subjects, "variants": variants, "listings": listings, "listings_total": int(total), "listings_shown": len(listings), # Only where variants actually BECOME ONE. A brand or a product line fold moves variants # and keeps them distinct, so their different barcodes and sizes are the catalogue working, # not a conflict; listing them there buried the real ones in noise. "lines": lines_out, "consequences": consequences(db, row, subjects), "conflicts": _conflicts(variants) if merging_variants else [], "merges_variants": merging_variants, "ledger": ledger, }