"""Turning collected listings into catalog rows. This is where cross-retailer matching happens. A barcode, when we have one, makes that a join; the normalised brand+name+size key is the fallback for the minority of records without one. """ import html import json import logging from dataclasses import asdict from datetime import UTC, datetime, timedelta from sqlalchemy import inspect, select, text from sqlalchemy.exc import IntegrityError, SQLAlchemyError from sqlalchemy.orm import Session from app.models import ( Merge, Brand, CollectionRun, Listing, Shop, PriceObservation, ProductVariant, RawRecord, Retailer, Source, ) from app.models.catalog import IDENTITY_RULES_VERSION from app.services.collectors.base import ( Collector, ListingRef, RawListing, bind_skip_sink, option_attributes, quantity_of, unbind_skip_sink, ) from app.services.collected import write_listed from app.services.collectors.fetch import SourceBlocked from app.services.fx import FxRates, load_rates from app.services import keying from app.services.normalize import ( brand_key, flat_key, looks_exclusive, parse_concentration, parse_size, quantity_ml_of, ) from app.services.taxonomy import classify, vertical_of DEFAULT_VERTICAL = "liquor" #: What a NEW variant nothing placed is filed under (no shelf, no category of ours): never #: "liquor" by default, which once filed everything unknown with the drinks (plan section 6). UNPLACED_VERTICAL = "unknown" # A run still "running" this long after it started belongs to a process that # is gone: a deploy recreates the container and kills whatever was collecting. STUCK_AFTER = timedelta(hours=24) class RejectedListing(Exception): """Ingest refuses this listing; it must not be published or guessed at.""" def __init__(self, reason: str, detail: str = "") -> None: super().__init__(f"{reason}: {detail}" if detail else reason) self.reason = reason self.detail = detail def count_skip(run: CollectionRun, reason: str) -> None: """One skip, counted in the total and under its reason (migration #2).""" run.skipped_no_price += 1 counts = dict(run.skip_counts or {}) counts[reason] = counts.get(reason, 0) + 1 run.skip_counts = counts def record_rejection( db: Session, run: CollectionRun, *, stage: str, reason: str, raw: RawListing | None, detail: str = "", ) -> None: """The one place a refused listing is counted and, once the table exists, kept. Stream Q's migration #2 adds `rejected_observations(stage, reason, source_sku, url, payload, run_id)`; until then this logs and counts, and a schema mismatch there can never end a run. """ count_skip(run, reason) logger.warning( "listing_rejected stage=%s reason=%s run=%s sku=%s detail=%s", stage, reason, run.id, raw.source_sku if raw else None, detail, ) if not _has_table(db, "rejected_observations"): return savepoint = db.begin_nested() try: db.execute( text( "INSERT INTO rejected_observations (stage, reason, source_sku, url, payload, run_id)" " VALUES (:stage, :reason, :sku, :url, CAST(:payload AS jsonb), :run_id)" ), { "stage": stage, "reason": reason, "sku": raw.source_sku if raw else None, "url": raw.url if raw else None, "payload": json.dumps( {"listing": asdict(raw) if raw else None, "detail": detail}, default=str ), "run_id": run.id, }, ) savepoint.commit() except SQLAlchemyError as exc: savepoint.rollback() logger.error("rejection_not_stored reason=%s error=%s", reason, exc) _table_cache: dict[int, dict[str, bool]] = {} def _has_table(db: Session, name: str) -> bool: bind = db.get_bind() known = _table_cache.setdefault(id(bind), {}) if name not in known: known[name] = inspect(bind).has_table(name) return known[name] logger = logging.getLogger(__name__) # How many price rows to accumulate before committing. Small enough to keep lock # windows short under concurrency, large enough not to thrash. COMMIT_EVERY = 50 def _get_or_create(db: Session, model, defaults: dict, **lookup): """Fetch a row, creating it if absent, tolerating a concurrent creator. Collectors for the same retailer run concurrently (one process per airport, so each host's crawl delay applies independently), and they share retailer rows. Without this, whichever process loses the race dies on a unique constraint and its whole run is lost. """ instance = db.scalar(select(model).filter_by(**lookup)) if instance is not None: return instance savepoint = db.begin_nested() try: instance = model(**lookup, **defaults) db.add(instance) savepoint.commit() return instance except IntegrityError: savepoint.rollback() return db.scalar(select(model).filter_by(**lookup)) def ensure_source(db: Session, collector: Collector) -> Source: return _get_or_create( db, Source, {"name": collector.retailer_name}, slug=collector.slug ) def ensure_retailer_and_shops(db: Session, collector: Collector) -> dict[str, Shop]: retailer = _get_or_create( db, Retailer, { "name": collector.retailer_name, "operator": collector.operator, "homepage": collector.homepage, }, slug=collector.retailer_slug, ) shops: dict[str, Shop] = {} for spec in collector.shops(): shop = _get_or_create( db, Shop, { "iata": spec.iata, "name": spec.name, "city": spec.city, "country": spec.country, "currency": spec.currency, "is_catalogue_only": spec.is_catalogue_only, }, retailer_id=retailer.id, code=spec.code, ) sync_shop_currency(shop, spec.currency) shops[spec.code] = shop return shops def held_listing_refs(db: Session, shops: list[Shop]) -> list[ListingRef]: """Every listing with a URL the database holds at these shops. Handed to a collector that declares `wants_held_listings`, so a store whose listing pages may only be read one page deep (robots forbids the query string that pages them) re-reads the product URLs it already knows instead of letting page-2+ product variants fall silent. """ if not shops: return [] code_of = {shop.id: shop.code for shop in shops} rows = db.execute( select(Listing.source_sku, Listing.url, Listing.shop_id, ProductVariant.vertical) .join(ProductVariant, ProductVariant.id == Listing.variant_id) .where(Listing.shop_id.in_(list(code_of)), Listing.url.isnot(None)) .order_by(Listing.id) ).all() return [ ListingRef(source_sku=sku, url=url, shop_code=code_of[shop_id], vertical=vertical) for sku, url, shop_id, vertical in rows ] def sync_shop_currency(shop: Shop, declared: str) -> bool: """The collector's declared currency wins over a stale row. One store was registered as MXN, then found to quote in USD; the collector was corrected but the row was not, so every one of its USD observations disagreed with its shop for two weeks. The shop's currency is collector data, not a human value, so it follows the code. """ if shop.currency == declared: return False logger.warning( "shop_currency_synced code=%s was=%s now=%s", shop.code, shop.currency, declared ) shop.currency = declared return True def _sizes_disagree(a: int | None, b: int | None) -> bool: """Two known sizes that cannot be the same bottle. Feeds are sloppy at 700-vs-750ml, so small differences are tolerated; beyond ~12 percent they are different bottles no matter what the barcode says -- retailers do mislabel barcodes, and trusting one puts a 70cl liqueur's price on a 1L whisky's page. """ if not a or not b: return False return abs(a - b) / max(a, b) > 0.12 def _quantities_disagree(ours, theirs) -> bool: """Two known quantities that cannot be the same item, compared like with like: ml with ml (the 12 percent tolerance of `_sizes_disagree`), g with g, pcs with pcs; a different dimension is a disagreement (100 ml is never 100 g); an unknown on either side never vetoes.""" if ours is None or theirs is None or ours.state != "stated" or theirs.state != "stated": return False if ours.unit is None or theirs.unit is None or ours.value is None or theirs.value is None: return False if ours.unit != theirs.unit: return True a, b = float(ours.value), float(theirs.value) if not a or not b: return False return abs(a - b) / max(a, b) > 0.12 def write_quantity(product: ProductVariant, q, *, name: str | None = None) -> None: """The seven quantity columns and `name_key` on a product, from one Quantity.""" product.quantity_value = q.value product.quantity_unit = q.unit product.pack_count = q.pack_count product.pack_unit_value = q.pack_unit_value product.form = q.form product.set_contents = q.set_contents product.quantity_state = q.state product.name_key = flat_key(name if name is not None else product.name)[:400] def brand_slug(brand: str | None) -> str: """The brands.slug for a spelling: the fold key, hyphenated.""" return brand_key(brand).replace(" ", "-") def resolve_brand(db: Session, brand: str | None) -> Brand | None: """The brands row a spelling folds to, created on first sight (migration #3). The row's `name` is the first spelling seen; `backfill brands` picks the most common spelling across the catalogue, and a human value stays. """ slug = brand_slug(brand) if not slug or not _has_table(db, "brands"): return None return _get_or_create(db, Brand, {"name": brand.strip()}, slug=slug) def raw_attributes(raw: RawListing) -> dict: """The certain facts a listing declares beyond its typed columns: the options the shop published as fields (`option:`, identity rules v6: certain and identity, written once at the variant's creation), a clearly stated age (liquor), the parsed concentration (beauty; Decision 6's veto).""" from app.services import product_lines vertical = vertical_of(classify(raw.name, raw.brand, raw.feed_categories)) or raw.vertical out: dict = dict(option_attributes(raw.options)) if vertical == "beauty": concentration = parse_concentration(raw.name) if concentration: out["concentration"] = concentration if vertical == "liquor": age = product_lines.age_of(raw.name) if age is not None: out["age"] = age return out def attributes_disagree(ours: dict | None, theirs: dict | None) -> bool: """Two declared values for the same attribute that differ: not the same product. A missing value on either side is not a disagreement; most feeds omit the concentration, and a silent feed must never split a product. """ ours, theirs = ours or {}, theirs or {} # The canonical attribute subsumes the parsed concentration (identity rules v3): two # Elixir rows parsed as "parfum" and "edp" are one attribute, not two bottles. if ours.get("attribute") and theirs.get("attribute"): return ours["attribute"] != theirs["attribute"] return any(key in theirs and theirs[key] != value for key, value in ours.items() if value) def stated_size(raw: RawListing) -> tuple[float | None, str | None]: """The size as the name states it, kept only when it agrees with quantity_ml.""" stated = parse_size(raw.name) if stated and (raw.quantity_ml is None or quantity_ml_of(*stated) == raw.quantity_ml): return stated if raw.quantity_ml: return float(raw.quantity_ml), "ml" return None, None def survivor_of(db: Session, product: ProductVariant | None, limit: int = 8) -> ProductVariant | None: """The product a merged row forwards to; the row itself when it forwards nowhere.""" hops = 0 while product is not None and product.merged_into_id and hops < limit: product = db.get(ProductVariant, product.merged_into_id) hops += 1 return product def _placed(db: Session, raw: RawListing, shop: Shop | None, parser_version: str | None, gtin: str | None, quantity, attributes: dict) -> ProductVariant | None: """The variant a listing ALREADY sits on, while the shop's words for it are unchanged (identity rules v6): the rules place NEW listings; a placed one stays placed. Without this a rules change un-places by attrition. Rehearsed 17 Sep: of 10,008 listings that reach their variant only by the key, 699 key apart under the certain key (a category noun, a stated strength), and each would have left for a fresh variant at its next sighting, taking 114 comparisons with it, one silent split per collection. It also keeps a reviewed catalogue still: a person's line or merge decisions never depend on a key agreeing. "Unchanged" is the listed layer's own compare: the fragment's name key and variant text equal the ones stored at the last sighting; a listing with no listed layer yet stays only when the name it arrives under is the variant's own. A shop that reuses a SKU for another product changes those words and falls through to the key. Evidence that contradicts still wins: a different barcode, a quantity that cannot be the same item, a declared attribute that differs.""" if shop is None: return None listing = db.scalar(select(Listing).where(Listing.shop_id == shop.id, Listing.source_sku == raw.source_sku)) if listing is None: return None variant = survivor_of(db, db.get(ProductVariant, listing.pinned_variant_id or listing.variant_id)) if variant is None or variant.merged_into_id is not None: return None if listing.listed_name_key and raw.raw is not None: from app.services.collected import listed_fields now = listed_fields(parser_version, raw.raw) if (now.get("listed_name_key"), now.get("listed_variant")) != (listing.listed_name_key, listing.listed_variant): return None elif flat_key(raw.name)[:400] != (variant.name_key or flat_key(variant.name)[:400]): return None if gtin and variant.gtin and variant.gtin != gtin: return None if _quantities_disagree(keying.product_quantity(variant, keying.maps_for(db)), quantity): return None if attributes_disagree(variant.attributes, attributes) or _stated_numbers_disagree(variant.abv, raw.abv): return None return variant def _stated_numbers_disagree(ours, theirs) -> bool: """Two stated numbers that differ (an ABV of 40 against 43); numbers compare as numbers and a stated value against none is not a disagreement (plan W9).""" if ours is None or theirs is None: return False try: return abs(float(ours) - float(theirs)) > 1e-9 except (TypeError, ValueError): return str(ours) != str(theirs) def _resolve_product(db: Session, raw: RawListing, shop: Shop | None = None, parser_version: str | None = None) -> ProductVariant: """Find or create the canonical product a listing refers to: by barcode; else where the listing already sits (`_placed`); else by the certain key; else a person's merge at that key; else its own new variant on its own product line, offered to the nearest approved line as a proposal and never joined to it (`run_collector` writes the arrival proposal).""" # The keying call (identity rules v3, `services/keying.py`): the brand with its alias # followed and the canonical attribute, then the key. category = classify(raw.name, raw.brand, raw.feed_categories) quantity = quantity_of(raw, category=category) # The brand row first, and into the maps: a brand seen for the first time used to key with no # row, so its first variant got no product line until a backfill (the ring from Panama). first_brand = resolve_brand(db, raw.brand) if first_brand is not None: maps = keying.maps_for(db) if first_brand.id not in maps.brands: maps.brands[first_brand.id] = first_brand maps.by_slug[first_brand.slug] = first_brand key, attributes, product_line_id = keying.listing_key(db, raw, raw_attributes(raw), quantity) # A barcode is identity -- unless the quantities flatly disagree, in which case # the retailer mislabelled the barcode and we refuse the join. The listing # then resolves by brand+name+quantity and must NOT carry the disputed code. gtin = raw.gtin product: ProductVariant | None = None if gtin: product = survivor_of(db, db.scalar(select(ProductVariant).where(ProductVariant.gtin == gtin))) if product is not None and ( _quantities_disagree(keying.product_quantity(product, keying.maps_for(db)), quantity) or (product.quantity_state != "stated" and _sizes_disagree(product.quantity_ml, raw.quantity_ml)) ): # Refuse, do not re-route: falling back to brand+name+size here # once created two 7,000 ml Glenfiddichs from a "0.70cl" misread, # each a published product with a price. Contradictory evidence # is a rejection, and the human review decides which side lied. raise RejectedListing( "gtin_size_veto", f"gtin={gtin} product={product.id}({product.quantity_ml}ml) " f"listing={raw.source_sku}({raw.quantity_ml}ml)", ) if product is None: product = _placed(db, raw, shop, parser_version, gtin, quantity, attributes) if product is None: # The fallback key can join a barcode-bearing product (the key IS our # identity for barcode-less rows), but never one holding a DIFFERENT # barcode -- that would be a different bottle with the same name. # Taking only the first key-holder here once created a fresh duplicate # product per airport per run; consider every candidate. for candidate in db.scalars( select(ProductVariant) .where(ProductVariant.match_key == key, ProductVariant.merged_into_id.is_(None)) .order_by(ProductVariant.gtin.isnot(None).desc(), ProductVariant.id) ): if gtin and candidate.gtin and candidate.gtin != gtin: continue # The per-vertical attribute vetoes: an EDP and an EDT of one line # and size share a key and are different bottles. if attributes_disagree(candidate.attributes, attributes) or _stated_numbers_disagree(candidate.abv, raw.abv): continue product = candidate break if product is None: # A human product merge holds across the next collection: a tombstone at this key # forwards to its survivor (LT3 d). Without this, `_resolve_listing` re-pinned the # listing to a fresh product at the old key on every sighting (576 pairs on staging). product = _follow_tombstone(db, key, gtin) if product is None: # Two collectors running concurrently can stock the same bottle and both # reach here. Create inside a savepoint so the loser re-reads the winner's # row instead of poisoning the whole session on the barcode constraint. brand_row = resolve_brand(db, raw.brand) savepoint = db.begin_nested() try: quantity_stated_value, quantity_stated_unit = stated_size(raw) product = ProductVariant( gtin=gtin, match_key=key, brand=raw.brand, brand_id=brand_row.id if brand_row else None, product_line_id=product_line_id, name=raw.name, category=category, vertical=resolve_vertical(raw.vertical, category, None), quantity_stated_value=quantity_stated_value, quantity_stated_unit=quantity_stated_unit, quantity_ml=raw.quantity_ml, abv=raw.abv, attributes=attributes, identity_rules_version=IDENTITY_RULES_VERSION, country_of_origin=raw.country_of_origin, is_exclusive=raw.is_exclusive or looks_exclusive(raw.name), ) write_quantity(product, quantity) db.add(product) savepoint.commit() product._arrived = True # `run_collector` offers it to the nearest approved line return product except IntegrityError: savepoint.rollback() product = None if gtin: product = db.scalar(select(ProductVariant).where(ProductVariant.gtin == gtin)) if product is None: product = db.scalar(select(ProductVariant).where(ProductVariant.match_key == key)) if product is None: raise # Enrich: a later source often knows a fact the first one did not. The # barcode enrichment runs in a savepoint because a concurrent collector may # have created a product with this code since we looked. if product.gtin is None and gtin: savepoint = db.begin_nested() try: product.gtin = gtin db.flush() savepoint.commit() except IntegrityError: savepoint.rollback() enrich_from_raw(product, raw, quantity, keying.maps_for(db).decided.get(product.id, {})) if product.name_key is None: product.name_key = flat_key(product.name)[:400] if product.brand_id is None and product.brand: brand_row = resolve_brand(db, product.brand) product.brand_id = brand_row.id if brand_row else None # A shop's options are written at the variant's creation and never enriched onto a variant # another listing made: the key reads them, and a key that moved would strand the first # listing's next sighting. missing = {k: v for k, v in attributes.items() if k not in (product.attributes or {}) and not k.startswith("option:")} if missing: # Reassigned, never mutated in place, so the session sees the change. product.attributes = {**(product.attributes or {}), **missing} if product.identity_rules_version != IDENTITY_RULES_VERSION: product.identity_rules_version = IDENTITY_RULES_VERSION # Re-run classification on every sighting: a second source often supplies the # category hint the first one lacked. if product.category is None: product.category = classify(raw.name, raw.brand, raw.feed_categories) product.vertical = resolve_vertical(raw.vertical, product.category, product.vertical) if raw.is_exclusive or looks_exclusive(raw.name): product.is_exclusive = True return product def enrich_from_raw(product: ProductVariant, raw: RawListing, quantity, decided: dict | None) -> None: """A later source often knows a fact the first one did not: fill a column that is empty, never one a decision guards (`attributes.guarded_for`, derived from the registry). A decided EMPTY (an explicit null on `attribute:abv`) is a decision too, so the column is never refilled (spec test 19). The quantity is enriched only where nothing was read and nothing was decided.""" from app.services import attributes as attributes_service guarded = attributes_service.guarded_for(decided) for column in ("brand", "quantity_ml", "abv", "country_of_origin"): if column in guarded: continue if getattr(product, column) is None and getattr(raw, column) is not None: setattr(product, column, getattr(raw, column)) if product.quantity_stated_value is None and product.quantity_ml is not None and "quantity_stated_value" not in guarded: product.quantity_stated_value, product.quantity_stated_unit = stated_size(raw) if product.quantity_state != "stated" and quantity.state == "stated" and not (guarded & attributes_service.QUANTITY_COLUMNS): write_quantity(product, quantity) def _follow_tombstone(db: Session, key: str, gtin: str | None) -> ProductVariant | None: """The survivor of a HUMAN merge whose tombstone carries this key under the current rules: `merges.reason == "confirmed"` (never a rule's fold), the tombstone's `identity_rules_version` current (`rederive` rekeys tombstones too), its quantity slot not `unknown` (unknown never agrees with known), and the survivor holding no barcode different from the listing's. Several tombstones on one key follow the newest merge.""" parts = key.split("|") if len(parts) < 4 or parts[3] == "unknown": # `n/a` equals itself and may follow return None rows = db.execute( select(ProductVariant, Merge) .join(Merge, Merge.from_id == ProductVariant.id) .where(ProductVariant.match_key == key, ProductVariant.merged_into_id.isnot(None), ProductVariant.identity_rules_version == IDENTITY_RULES_VERSION, Merge.reason == "confirmed") .order_by(Merge.id.desc()) ).all() for tombstone, _merge in rows: survivor = survivor_of(db, tombstone) if survivor is None or survivor.merged_into_id is not None: continue if gtin and survivor.gtin and survivor.gtin != gtin: continue return survivor return None def _propose_arrival(db: Session, product: ProductVariant) -> None: """A new variant that matched nothing, offered to the nearest approved product line as a proposal (K3.5); a store problem never ends a run.""" if not _has_table(db, "proposals"): return savepoint = db.begin_nested() try: from app.services import proposal_rules proposal_rules.propose_arrival(db, product) savepoint.commit() except Exception as exc: # noqa: BLE001 - a proposal is advice; the listing is already stored savepoint.rollback() logger.warning("arrival_proposal_failed product=%s error=%s", product.id, exc) def resolve_vertical(claimed: str | None, category: str | None, current: str | None) -> str: """The family a product belongs to. Our own category is the strongest evidence (it was read from the name and the feed's labels); the collector's claim is which tree it walked; the current value is kept when neither says otherwise; a product nothing placed is `unknown` (`UNPLACED_VERTICAL`), which also gives it the `n/a` quantity slot. """ return vertical_of(category) or claimed or current or UNPLACED_VERTICAL def _resolve_listing(db: Session, product: ProductVariant, shop: Shop, raw: RawListing) -> Listing: listing = db.scalar( select(Listing).where( Listing.shop_id == shop.id, Listing.source_sku == raw.source_sku ) ) if listing is None: savepoint = db.begin_nested() try: listing = Listing( variant_id=product.id, shop_id=shop.id, source_sku=raw.source_sku, url=raw.url, ) db.add(listing) savepoint.commit() return listing except IntegrityError: savepoint.rollback() listing = db.scalar( select(Listing).where( Listing.shop_id == shop.id, Listing.source_sku == raw.source_sku, ) ) if listing is None: raise if listing is not None: # A pinned listing (a person's decision, LT3 e) stays on its pinned product's # survivor every sighting and is never re-pinned over by the rules. if listing.pinned_variant_id: pinned = survivor_of(db, db.get(ProductVariant, listing.pinned_variant_id)) if pinned is not None: listing.variant_id = pinned.id else: listing.variant_id = product.id if raw.url: listing.url = raw.url return listing def _implausibly_cheap( db: Session, product: ProductVariant, shop: Shop, price_usd: float | None ) -> bool: """A price far below what every other shop charges is a bug, not a bargain. Both real incidents so far -- a currency mix-up and a small-variant price recorded against the full-size bottle -- would have been caught here: the bad rows sat at under half the product's cross-shop median while genuine spreads (including border-shop discounts and 20 percent promotions) stay well above it. Quarantine means skip and log, never publish. """ if price_usd is None or product.id is None: return False peers = [ float(row) for row in db.scalars( select(PriceObservation.price_usd) .join(Listing, Listing.id == PriceObservation.listing_id) .where( Listing.variant_id == product.id, Listing.shop_id != shop.id, PriceObservation.price_usd.isnot(None), ) .order_by(Listing.shop_id, PriceObservation.observed_at.desc()) .distinct(Listing.shop_id) ) ] if len(peers) < 3: return False peers.sort() median = peers[len(peers) // 2] return price_usd < 0.45 * median def facts_only_detail(detail: dict) -> dict: from app.services.collectors.base import facts_only return facts_only(detail) def mark_stuck(run: CollectionRun, why: str) -> None: """A run whose process died is an error, not a collection in progress.""" run.status = "error" run.error = f"stuck: {why}" run.finished_at = run.finished_at or datetime.now(UTC) def is_stuck(run: CollectionRun, now: datetime) -> bool: return run.status == "running" and run.started_at is not None and now - run.started_at > STUCK_AFTER def run_collector( db: Session, collector: Collector, *, limit: int | None = None, delay: float | None = None, rates: FxRates | None = None, ) -> CollectionRun: """Collect one source into the database, recording the run either way.""" source = ensure_source(db, collector) # A source runs one process at a time, so any earlier run of it still # "running" belongs to a process that died (a deploy killed it). Eight # such runs once sat open for ten days and looked like live collections. for stale in db.scalars( select(CollectionRun).where( CollectionRun.source_id == source.id, CollectionRun.status == "running" ) ): mark_stuck(stale, "superseded by a new run of the same source") run = CollectionRun(source_id=source.id, started_at=datetime.now(UTC)) db.add(run) db.flush() if not source.enabled: run.status = "skipped" run.error = "source disabled" run.finished_at = datetime.now(UTC) db.commit() logger.info("collector_skipped source=%s reason=disabled", collector.slug) return run rates = rates or load_rates() run.fx_source = "fallback" if rates.is_fallback else "live" run.fx_fetched_at = rates.fetched_at shops = ensure_retailer_and_shops(db, collector) if getattr(collector, "wants_held_listings", False): collector.held_listings = held_listing_refs(db, list(shops.values())) effective_delay = float(source.delay_seconds) if delay is None else delay # The table arrives with migration #1; a collection may run on a container # that started before the upgrade, and must not die for want of it. keep_raw = _has_table(db, "raw_records") parser_version = getattr(collector, "parser_version", collector.slug) def collector_skipped(reason: str, detail: dict) -> None: record_rejection( db, run, stage="collector", reason=reason, raw=None, detail=json.dumps(facts_only_detail(detail), default=str)[:2000], ) token = bind_skip_sink(collector_skipped) try: for raw in collector.collect(limit=limit, delay=effective_delay): shop = shops.get(raw.shop_code) if shop is None: continue # A price of zero is not a price. Retailers list gift-with-purchase # items and unavailable lines at 0, and showing those as the cheapest # way to buy a bottle makes the whole comparison look broken. if raw.price is None or raw.price <= 0: count_skip(run, "no_price") continue # Feeds ship HTML entities in names ("Bacardí"); stored # verbatim they render literally everywhere. Decode at the one # boundary every collector passes through. raw.name = html.unescape(raw.name) if raw.brand: raw.brand = html.unescape(raw.brand) try: product = _resolve_product(db, raw, shop, parser_version) except RejectedListing as rejected: record_rejection( db, run, stage="ingest", reason=rejected.reason, raw=raw, detail=rejected.detail, ) continue listing = _resolve_listing(db, product, shop, raw) if getattr(product, "_arrived", False): product._arrived = False _propose_arrival(db, product) # Every sighting stamps `last_seen_at`; a fragment refreshes the listed columns as # one unit from that fragment (`collected.write_listed`). A sighting the price # quarantine refuses below still refreshes them: the shop showed the words. listing.last_seen_at = datetime.now(UTC) if keep_raw and raw.raw is not None: record = RawRecord( listing_id=listing.id, run_id=run.id, payload=raw.raw, parser_version=parser_version, ) db.add(record) db.flush() write_listed(listing, record.id, parser_version, raw.raw) price_usd = rates.to_usd(raw.price, raw.currency) if _implausibly_cheap(db, product, shop, price_usd): record_rejection( db, run, stage="ingest", reason="price_quarantined", raw=raw, detail=f"product={product.id} price_usd={price_usd}", ) continue # Some shops render a "was" price equal to (or below) the current # one. Showing that as a saving would be wrong, so drop it. was_price = raw.was_price if raw.was_price and raw.was_price > raw.price else None db.add( PriceObservation( listing_id=listing.id, price=raw.price, currency=raw.currency, price_usd=price_usd, was_price=was_price, price_type=raw.price_type, in_stock=raw.in_stock, # Stamped as each listing arrives, which is just after its # page was fetched: one timestamp per run made a # six-hour crawl look like a single instant. observed_at=datetime.now(UTC), run_id=run.id, fx_rate=rates.rates.get((raw.currency or "").upper()), ) ) run.products_seen += 1 run.prices_written += 1 # Commit in batches rather than holding one transaction for the whole # run. Collectors run concurrently and touch the same product rows, so # a long-held transaction deadlocks against its siblings; short ones # rarely do. It also means a run that fails part-way keeps what it # already collected instead of losing everything. if run.prices_written % COMMIT_EVERY == 0: db.commit() run.status = "ok" except SourceBlocked as exc: # The source declined. Record it and stop -- we do not work around a block. run.status = "blocked" run.error = str(exc) logger.warning("collector_blocked source=%s detail=%s", collector.slug, exc) except Exception as exc: run.status = "error" run.error = f"{type(exc).__name__}: {exc}" logger.exception("collector_failed source=%s", collector.slug) unbind_skip_sink(token) run.finished_at = datetime.now(UTC) db.commit() logger.info( "collector_finished source=%s status=%s prices=%d", collector.slug, run.status, run.prices_written, ) return run