"""The giant Listings table on /collectors (Stream L, LT6): every listing with its listed, standard and decided layers side by side, sorted and filtered by the database. Sources of truth: this module (`COLUMNS` is the registry the API and the column picker share), `routers/collectors.py` (`GET /api/collectors/listings`, `format=csv`), `web/src/components/collectors/ListingsTable.tsx`, `tests/test_listings_table.py`. The listings table used to page 50 rows and read every fragment into Python for "differs"; 24k rows could not be sorted. Now it is ONE statement, joined once: listing, listed (the columns `collected.write_listed` filled), standard (the product), house, line, variation, shop, collector, price (the newest observation through a window), provenance and differs. Every column of the first nine groups sorts and filters in SQL (`sort=&dir=asc|desc`); provenance and differs are computed per row from the same statement, and `differs` is also a WHERE clause: `brand` is `listed_brand_key <> house.slug` (both slug-shaped, so "Paco Rabanne" listed under the house `paco-rabanne` does not differ), `name` is `listed_name_key <> products.name_key`, `quantity` compares the listed pair with the product's. Ignored listings are hidden unless asked for (`ignored=only|any`). Listed cells are handed over exactly as stored: never trimmed, cased or unescaped again. The CSV is the same statement streamed in batches of 1,000 rows, the chosen columns flattened. """ from __future__ import annotations import csv import io from collections.abc import Iterator from dataclasses import dataclass from datetime import datetime from typing import Any from sqlalchemy import Select, and_, case, func, literal, or_, select from sqlalchemy.orm import Session, aliased from app.models import ( Brand, CollectionRun, Listing, Location, PriceObservation, Product, ProductLine, ProductMerge, RawRecord, Retailer, Source, ) from app.models.accounts import Override from app.services.collectors.registry import COLLECTORS from app.services.lines import display_variation PER_PAGE_MAX = 500 CSV_BATCH = 1000 @dataclass(frozen=True) class Column: id: str group: str label: str sortable: bool = True GROUPS = ("listing", "listed", "standard", "house", "line", "variation", "shop", "collector", "price", "provenance", "differs") GROUP_LABELS = { "listing": "Listing", "listed": "Listed (the shop's words)", "standard": "Standard (the product)", "house": "House", "line": "Line", "variation": "Variation", "shop": "Shop", "collector": "Collector", "price": "Price", "provenance": "Provenance", "differs": "Differs", } COLUMNS: tuple[Column, ...] = ( Column("listing_id", "listing", "Listing id"), Column("airport", "listing", "Airport"), Column("location", "listing", "Location"), Column("collector", "listing", "Collector"), Column("source_sku", "listing", "SKU"), Column("url", "listing", "URL"), Column("first_seen", "listing", "First seen"), Column("last_seen", "listing", "Last seen"), Column("pinned", "listing", "Pinned to"), Column("ignored", "listing", "Ignored"), Column("ignore_reason", "listing", "Ignore reason"), Column("listed_brand", "listed", "Listed brand"), Column("listed_name", "listed", "Listed name"), Column("listed_variant", "listed", "Listed variant"), Column("listed_quantity_text", "listed", "Listed size words"), Column("listed_category", "listed", "Listed category"), Column("listed_gtin", "listed", "Listed barcode"), Column("listed_quantity_value", "listed", "Listed quantity"), Column("listed_quantity_unit", "listed", "Listed unit"), Column("fragment_seen_at", "listed", "Fragment seen"), Column("fragment_parser", "listed", "Fragment parser"), Column("product_id", "standard", "Product id"), Column("product_name", "standard", "Product name"), Column("quantity_value", "standard", "Quantity"), Column("quantity_unit", "standard", "Unit"), Column("pack_count", "standard", "Pack count"), Column("form", "standard", "Form"), Column("set_contents", "standard", "Set contents"), Column("quantity_state", "standard", "Quantity state"), Column("size_ml", "standard", "Size ml"), Column("gtin", "standard", "Barcode"), Column("gtin_source", "standard", "Barcode source"), Column("category", "standard", "Category"), Column("vertical", "standard", "Vertical"), Column("rules_version", "standard", "Rules version"), Column("brand", "house", "Collected brand"), Column("house_name", "house", "House"), Column("house_slug", "house", "House slug"), Column("house_canonical", "house", "House canonical"), Column("line_name", "line", "Line"), Column("line_slug", "line", "Line slug"), Column("line_canonical", "line", "Line canonical"), Column("line_products", "line", "Products in line"), Column("variation", "variation", "Variation"), Column("variation_display", "variation", "Variation shown", sortable=False), Column("variation_kind", "variation", "Kind"), Column("retailer", "shop", "Retailer"), Column("shop_name", "shop", "Shop"), Column("currency", "shop", "Currency"), Column("country", "shop", "Country"), Column("visible", "shop", "Visible"), Column("collector_slug", "collector", "Collector slug"), Column("platform", "collector", "Platform"), Column("parser_version", "collector", "Parser version"), Column("run_status", "collector", "Last run"), Column("price", "price", "Price"), Column("was_price", "price", "Was"), Column("price_currency", "price", "Price currency"), Column("price_usd", "price", "USD"), Column("fx_rate", "price", "FX rate"), Column("observed_at", "price", "Observed"), Column("in_stock", "price", "In stock"), Column("prov_name", "provenance", "Name from", sortable=False), Column("prov_line", "provenance", "Line from", sortable=False), Column("prov_variation", "provenance", "Variation from", sortable=False), Column("prov_quantity", "provenance", "Quantity from", sortable=False), Column("differs_brand", "differs", "Brand differs", sortable=False), Column("differs_name", "differs", "Name differs", sortable=False), Column("differs_quantity", "differs", "Quantity differs", sortable=False), ) COLUMN_IDS = tuple(c.id for c in COLUMNS) _BY_ID = {c.id: c for c in COLUMNS} DIFFERS = ("any", "brand", "name", "quantity") def registry() -> list[dict[str, Any]]: """The groups and their columns, for the picker; the ids are the row's field names.""" return [ {"group": g, "label": GROUP_LABELS[g], "columns": [{"id": c.id, "label": c.label, "sortable": c.sortable} for c in COLUMNS if c.group == g]} for g in GROUPS ] def chosen_columns(spec: str | None) -> list[str]: """`columns=a,b,c`, `all`, or nothing (the default set).""" if not spec or spec == "all": return list(COLUMN_IDS) wanted = [c.strip() for c in spec.split(",") if c.strip() in _BY_ID] return wanted or list(COLUMN_IDS) def _collector_case(): codes = {spec.code: slug for slug, collector in COLLECTORS.items() for spec in collector.locations()} if not codes: return literal(None) return case(codes, value=Location.code, else_=None) def _platform_case(): from app.services.collector_view import platform_of platforms = {spec.code: platform_of(slug) for slug, collector in COLLECTORS.items() for spec in collector.locations()} if not platforms: return literal(None) return case(platforms, value=Location.code, else_=None) def _parser_case(): versions = {spec.code: getattr(collector, "parser_version", slug) for slug, collector in COLLECTORS.items() for spec in collector.locations()} if not versions: return literal(None) return case(versions, value=Location.code, else_=None) def _latest_observation(): ranked = select( PriceObservation.id.label("obs_id"), PriceObservation.listing_id.label("listing_id"), func.row_number().over(partition_by=PriceObservation.listing_id, order_by=(PriceObservation.observed_at.desc(), PriceObservation.id.desc())).label("rn"), ).subquery() return select(ranked.c.obs_id, ranked.c.listing_id).where(ranked.c.rn == 1).subquery() def _latest_run(): ranked = select( CollectionRun.id.label("run_id"), CollectionRun.source_id.label("source_id"), CollectionRun.status.label("status"), func.row_number().over(partition_by=CollectionRun.source_id, order_by=CollectionRun.id.desc()).label("rn"), ).subquery() return select(ranked.c.source_id, ranked.c.status).where(ranked.c.rn == 1).subquery() def build(db: Session) -> tuple[Select, dict[str, Any]]: """The one statement and the expression per column id (the sortable ones), so a sort or a filter is the same SQL as the selection.""" brand = aliased(Brand, name="brand_row") house = aliased(Brand, name="house_row") line = aliased(ProductLine, name="line_row") line_canonical = aliased(ProductLine, name="line_canonical_row") obs = aliased(PriceObservation, name="obs") latest = _latest_observation() last_run = _latest_run() fragment = aliased(RawRecord, name="fragment") line_products = ( select(Product.line_id.label("line_id"), func.count(Product.id).label("n")) .where(Product.merged_into_id.is_(None)).group_by(Product.line_id).subquery() ) merged_into = select(ProductMerge.to_id).distinct().subquery() collector_expr = _collector_case() house_id = func.coalesce(brand.canonical_id, brand.id) line_id = func.coalesce(line.canonical_id, line.id) expr: dict[str, Any] = { "listing_id": Listing.id, "airport": func.coalesce(Location.iata, Location.code), "location": Location.code, "collector": collector_expr, "source_sku": Listing.source_sku, "url": Listing.url, "first_seen": Listing.created_at, "last_seen": Listing.last_seen_at, "pinned": Listing.pinned_product_id, "ignored": Listing.ignored_at, "ignore_reason": Listing.ignore_reason, "listed_brand": Listing.listed_brand, "listed_name": Listing.listed_name, "listed_variant": Listing.listed_variant, "listed_quantity_text": Listing.listed_quantity_text, "listed_category": Listing.listed_category, "listed_gtin": Listing.listed_gtin, "listed_quantity_value": Listing.listed_quantity_value, "listed_quantity_unit": Listing.listed_quantity_unit, "fragment_seen_at": fragment.created_at, "fragment_parser": fragment.parser_version, "product_id": Product.id, "product_name": Product.name, "quantity_value": Product.quantity_value, "quantity_unit": Product.quantity_unit, "pack_count": Product.pack_count, "form": Product.form, "set_contents": Product.set_contents, "quantity_state": Product.quantity_state, "size_ml": Product.size_ml, "gtin": Product.gtin, "gtin_source": Product.gtin_source, "category": Product.category, "vertical": Product.vertical, "rules_version": Product.identity_rules_version, "brand": Product.brand, "house_name": func.coalesce(house.name, brand.name), "house_slug": func.coalesce(house.slug, brand.slug), "house_canonical": house.slug, "line_name": func.coalesce(line_canonical.name, line.name), "line_slug": func.coalesce(line_canonical.slug, line.slug), "line_canonical": line_canonical.slug, "line_products": line_products.c.n, "variation": Product.attributes["variation"].as_string(), "variation_kind": Product.attributes["variation_kind"].as_string(), "retailer": Retailer.name, "shop_name": Location.name, "currency": Location.currency, "country": Location.country, "visible": Location.visible, "collector_slug": collector_expr, "platform": _platform_case(), "parser_version": _parser_case(), "run_status": last_run.c.status, "price": obs.price, "was_price": obs.was_price, "price_currency": obs.currency, "price_usd": obs.price_usd, "fx_rate": obs.fx_rate, "observed_at": obs.observed_at, "in_stock": obs.in_stock, # For provenance and differs, computed per row from these. "_merged": Product.id.in_(select(merged_into.c.to_id)), "_house_slug_for_differs": func.coalesce(house.slug, brand.slug), "_name_key": Product.name_key, } stmt = ( select(*[e.label(k) for k, e in expr.items()]) .select_from(Listing) .join(Product, Product.id == Listing.product_id) .join(Location, Location.id == Listing.location_id) .join(Retailer, Retailer.id == Location.retailer_id) .outerjoin(brand, brand.id == Product.brand_id) .outerjoin(house, house.id == brand.canonical_id) .outerjoin(line, line.id == Product.line_id) .outerjoin(line_canonical, line_canonical.id == line.canonical_id) .outerjoin(line_products, line_products.c.line_id == line_id) .outerjoin(fragment, fragment.id == Listing.listed_record_id) .outerjoin(latest, latest.c.listing_id == Listing.id) .outerjoin(obs, obs.id == latest.c.obs_id) .outerjoin(Source, Source.slug == collector_expr) .outerjoin(last_run, last_run.c.source_id == Source.id) ) _ = house_id # the house is one alias hop deep in SQL; deeper chains are rare and resolved on the page return stmt, expr def differs_clause(which: str, expr: dict[str, Any]): """The WHERE for `differs=`: both sides known and unequal.""" brand_differs = and_(Listing.listed_brand_key.isnot(None), expr["house_slug"].isnot(None), Listing.listed_brand_key != expr["house_slug"]) name_differs = and_(Listing.listed_name_key.isnot(None), Product.name_key.isnot(None), Listing.listed_name_key != Product.name_key) quantity_differs = and_( Listing.listed_quantity_unit.isnot(None), Product.quantity_unit.isnot(None), or_(Listing.listed_quantity_unit != Product.quantity_unit, and_(Listing.listed_quantity_value.isnot(None), Product.quantity_value.isnot(None), Listing.listed_quantity_value != Product.quantity_value)), ) clauses = {"brand": brand_differs, "name": name_differs, "quantity": quantity_differs, "size": quantity_differs} if which == "any": return or_(*clauses.values()) return clauses[which] def conditions(db: Session, expr: dict[str, Any], *, q=None, brand=None, collector=None, airport=None, listed_brand=None, quantity_unit=None, quantity_state=None, form=None, variation_kind=None, has_fragment=None, ignored=None, pinned=None, differs=None) -> list: from app.services.collector_view import _scope_location_ids where = [] if ignored == "only": where.append(Listing.ignored_at.isnot(None)) elif ignored != "any": where.append(Listing.ignored_at.is_(None)) if q: like = f"%{q.strip().lower()}%" where.append(or_(func.lower(Product.name).like(like), func.lower(Product.brand).like(like), func.lower(Listing.listed_name).like(like))) if brand: where.append(func.lower(Product.brand).like(f"%{brand.strip().lower()}%")) if listed_brand: where.append(func.lower(Listing.listed_brand).like(f"%{listed_brand.strip().lower()}%")) location_ids = _scope_location_ids(db, collector, airport) if location_ids is not None: where.append(Listing.location_id.in_(location_ids)) if quantity_unit: where.append(Product.quantity_unit == quantity_unit) if quantity_state: where.append(Product.quantity_state == quantity_state) if form: where.append(Product.form == form) if variation_kind: where.append(expr["variation_kind"] == variation_kind) if has_fragment is not None: where.append(Listing.listed_record_id.isnot(None) if has_fragment else Listing.listed_record_id.is_(None)) if pinned is not None: where.append(Listing.pinned_product_id.isnot(None) if pinned else Listing.pinned_product_id.is_(None)) if differs in DIFFERS or differs == "size": where.append(differs_clause(differs, expr)) return where def order_for(expr: dict[str, Any], sort: str | None, direction: str | None): column = _BY_ID.get(sort or "") if column is None or not column.sortable: return (Product.brand.asc().nullslast(), Product.name.asc(), Location.code.asc(), Listing.id.asc()) e = expr[column.id] ordered = e.desc().nullslast() if direction == "desc" else e.asc().nullsfirst() return (ordered, Listing.id.asc()) def _iso(value: datetime | None) -> str | None: return value.isoformat(timespec="minutes") if value else None def _plain(value: Any) -> Any: if isinstance(value, datetime): return _iso(value) if hasattr(value, "quantize"): return float(value) return value def shape(row: Any, overrides_of: dict[str, set[str]]) -> dict[str, Any]: """A result row as the page and the CSV see it: every column id, the listed cells exactly as stored, provenance and differs computed here.""" m = row._mapping out = {k: _plain(m[k]) for k in COLUMN_IDS if k in m} out["variation_display"] = display_variation(m["variation"]) if m["variation"] else None decided = overrides_of.get(str(m["product_id"]), set()) merged = bool(m["_merged"]) def prov(field: str, from_merge: bool = True) -> str: if field in decided: return "override" if merged and from_merge: return "merge" return "rule" out["prov_name"] = prov("name") out["prov_line"] = prov("line_id") out["prov_variation"] = prov("variation") out["prov_quantity"] = prov("quantity") listed_brand_key = m.get("listed_brand_key") if "listed_brand_key" in m else None out["differs_brand"] = None out["differs_name"] = bool(m["listed_name"]) and bool(m["_name_key"]) and ( __import__("app.services.normalize", fromlist=["flat_key"]).flat_key(m["listed_name"])[:400] != m["_name_key"]) out["differs_quantity"] = bool(m["listed_quantity_unit"]) and bool(m["quantity_unit"]) and ( m["listed_quantity_unit"] != m["quantity_unit"] or (m["listed_quantity_value"] is not None and m["quantity_value"] is not None and float(m["listed_quantity_value"]) != float(m["quantity_value"]))) _ = listed_brand_key return out def _brand_differs_expr(expr: dict[str, Any]): return case((and_(Listing.listed_brand_key.isnot(None), expr["house_slug"].isnot(None), Listing.listed_brand_key != expr["house_slug"]), True), else_=False) def query(db: Session, *, sort: str | None = None, direction: str | None = None, page: int = 1, per_page: int = 50, columns: str | None = None, **filters) -> dict[str, Any]: """One page, with `total` and `with_fragment` counted by the database on every request.""" stmt, expr = build(db) stmt = stmt.add_columns(_brand_differs_expr(expr).label("_differs_brand")) where = conditions(db, expr, **filters) if where: stmt = stmt.where(*where) per_page = max(1, min(per_page, PER_PAGE_MAX)) page = max(1, page) counted = stmt.with_only_columns(Listing.id, Listing.listed_record_id).subquery() total = int(db.scalar(select(func.count()).select_from(counted)) or 0) with_fragment = int(db.scalar(select(func.count()).select_from(counted).where(counted.c.listed_record_id.isnot(None))) or 0) rows = db.execute(stmt.order_by(*order_for(expr, sort, direction)).offset((page - 1) * per_page).limit(per_page)).all() overrides_of = _overrides_of(db, [r._mapping["product_id"] for r in rows]) shaped = [] for r in rows: out = shape(r, overrides_of) out["differs_brand"] = bool(r._mapping["_differs_brand"]) shaped.append(out) wanted = chosen_columns(columns) return { "total": total, "with_fragment": with_fragment, "page": page, "per_page": per_page, "sort": sort, "dir": direction or "asc", "columns": wanted, "registry": registry(), "rows": [{k: v for k, v in row.items() if k in wanted or k in ("listing_id", "product_id")} for row in shaped], } def _overrides_of(db: Session, product_ids: list[int]) -> dict[str, set[str]]: if not product_ids: return {} from app.services.overrides import table_present if not table_present(db): return {} out: dict[str, set[str]] = {} for key, field in db.execute( select(Override.entity_key, Override.field) .where(Override.entity_type == "product", Override.entity_key.in_([str(i) for i in product_ids])) ): out.setdefault(key, set()).add(field) return out def csv_rows(db: Session, *, sort: str | None = None, direction: str | None = None, columns: str | None = None, **filters) -> Iterator[str]: """Every matching row as CSV text, one header, streamed in batches of `CSV_BATCH`.""" stmt, expr = build(db) stmt = stmt.add_columns(_brand_differs_expr(expr).label("_differs_brand")) where = conditions(db, expr, **filters) if where: stmt = stmt.where(*where) stmt = stmt.order_by(*order_for(expr, sort, direction)) wanted = chosen_columns(columns) buffer = io.StringIO() writer = csv.writer(buffer) writer.writerow(wanted) yield buffer.getvalue() offset = 0 while True: rows = db.execute(stmt.offset(offset).limit(CSV_BATCH)).all() if not rows: break overrides_of = _overrides_of(db, [r._mapping["product_id"] for r in rows]) buffer = io.StringIO() writer = csv.writer(buffer) for r in rows: out = shape(r, overrides_of) out["differs_brand"] = bool(r._mapping["_differs_brand"]) writer.writerow(["" if out.get(k) is None else out.get(k) for k in wanted]) yield buffer.getvalue() offset += CSV_BATCH if len(rows) < CSV_BATCH: break