"""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 import os import signal import threading from dataclasses import asdict from functools import lru_cache from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any from sqlalchemy import func, 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 import control as run_control from app.services.collectors.base import gone from app.services.collectors.fetch import FetchError, PageGone, 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) #: A running row whose heartbeat is older than this is dead: six times the slowest crawl delay #: met (60 s), and a render caps at 90 s, so a slow host never makes a live run dead. The hook #: writes the heartbeat at every request boundary and every poll of a pause (Stream AW4). STUCK_HEARTBEAT = timedelta(hours=1) #: "not given", so a caller may pass boot=None to mean "do not apply the boot rule". _UNSET: Any = object() 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: """Whether a table exists, asked through the SESSION's own connection. Never `inspect(bind)`: an inspector on the engine borrows a second connection, and on a single-connection pool (the tests' SQLite) that IS the session's connection, whose release rolls back whatever the caller has not committed. `keying.load_maps` carries the same note; here it silently undid an uncommitted brand split before the claim could be read (K9). """ bind = db.get_bind() known = _table_cache.setdefault(id(bind), {}) if name not in known: known[name] = inspect(db.connection()).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, vertical: str | None = None) -> str: """The brands.slug for a spelling: the fold key, hyphenated, scoped to the vertical the listing sits in (`normalize.trailers_for`), so a vertical nobody wrote a trailer list for folds nothing beyond a corporate suffix.""" return brand_key(brand).replace(" ", "-") def resolve_brand(db: Session, brand: str | None, vertical: str | None = None) -> Brand | None: """The brands row a spelling belongs to, created on first sight (migration #3). Three things are asked, in this order, and only the last is the fold. 1. A spelling a person SPLIT off is claimed: `brand_key("Appleton Rum")` is "appleton", the very row the split emptied, so resolving by the fold would put the next sighting straight back and mint a duplicate beside the moved variants (K9, `services/brands.py`). 2. Where the spelling already sits, when its live variants agree on one row (`Maps.placed_spellings`). A change to the trailer list then cannot silently re-home a spelling that is already placed; it only decides where an unseen one goes. Scoping the list per vertical (K9.5) would otherwise have moved three real spellings on the 17 Sep catalogue -- 'Souvenir de Paris', 'Distillerie De Paris' and 'Huda Beauty', 78 variants, all of them classified liquor -- at the next collection, with no decision behind it. 3. The fold key for the listing's vertical. The row's `name` is the first spelling seen; `backfill brands` picks the most common spelling across the catalogue, and a human value stays. """ from app.services import keying if not brand or not brand.strip() or not _has_table(db, "brands"): return None maps = keying.maps_for(db) # A split's claim first, then where the spelling ALREADY sits, then the fold. known = maps.spelling_claims.get(flat_key(brand)) or maps.placed_spellings.get(flat_key(brand)) if known is not None: return known slug = brand_slug(brand, vertical) if not slug: 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) vertical = vertical_of(category) or raw.vertical 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, vertical) 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, vertical) 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.vertical) 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) @lru_cache(maxsize=4) def process_boot(proc: str = "/proc") -> datetime | None: """When the process tree that hosts collectors started, or None where /proc is not Linux's. Collectors run inside the app container and nowhere else: the database publishes no port and the render sidecar sits on an internal network, so nothing outside the container can collect. PID 1 there is the container's own start, and a `docker exec` collector dies with it. A run that began before this moment is therefore dead however recently it started -- which STUCK_AFTER, a 24-hour guess at the same boundary, cannot say. The 18 Sep host reboot killed two runs three hours old, and the review's approval gate went on calling them live collections, refusing every approval, for the 21 hours left on the timer. """ try: stat = Path(proc, "1", "stat").read_text() after_comm = stat[stat.rindex(")") + 2:].split() ticks = int(after_comm[19]) # field 22, starttime, in clock ticks since boot for line in Path(proc, "stat").read_text().splitlines(): if line.startswith("btime "): boot = int(line.split()[1]) break else: return None return datetime.fromtimestamp(boot + ticks / os.sysconf("SC_CLK_TCK"), UTC) except (OSError, ValueError, IndexError): return None def _aware(value: datetime) -> datetime: return value if value.tzinfo else value.replace(tzinfo=UTC) def is_stuck(run: CollectionRun, now: datetime, boot: datetime | None = _UNSET, *, heartbeat: bool = True) -> bool: """Dead if it outlived the timer, if its heartbeat is older than `STUCK_HEARTBEAT` (a NULL heartbeat falls through: a run from before the hook is judged by time alone), or if it began before the container it ran in. Never reads `/proc` for the pid: the host dev server sees another pid namespace, and a reused pid would be believed. `heartbeat=False` is the rule as the review's approval gate applies it (`decisions/writer.refuse_if_collecting` recomputes the cutoff itself; an issue on the running list), so the page can say what holds approvals.""" if run.status != "running" or run.started_at is None: return False started = _aware(run.started_at) if now - started > STUCK_AFTER: return True beat = getattr(run, "heartbeat_at", None) if heartbeat and beat is not None and now - _aware(beat) > STUCK_HEARTBEAT: return True boot = process_boot() if boot is _UNSET else boot return boot is not None and started < boot def _try_lock(engine, slug: str): """The child's advisory lock on a dedicated connection, held until the process ends. This, not the API's running-row check, is what closes two Starts at once (a double click, two tabs, the page racing the admit-queue): `pg_try_advisory_lock` on `dfp.collect.` is atomic, and a session lock dies with the connection, so a killed collector never leaves it held. Returns the connection to close in `finally`, or raises CollectorLocked. On SQLite (the tests) there is no lock and every caller wins. """ if engine is None or engine.dialect.name != "postgresql": return None conn = engine.connect().execution_options(isolation_level="AUTOCOMMIT") try: got = conn.execute( text("SELECT pg_try_advisory_lock(hashtext(:key))"), {"key": f"dfp.collect.{slug}"} ).scalar() except SQLAlchemyError: conn.close() raise if not got: conn.close() logger.warning("collector_locked source=%s (another process holds the collector lock)", slug) raise run_control.CollectorLocked(f"{slug}: another collector process holds the lock") return conn def _release_lock(conn, slug: str) -> None: if conn is None: return try: conn.execute(text("SELECT pg_advisory_unlock(hashtext(:key))"), {"key": f"dfp.collect.{slug}"}) except SQLAlchemyError: pass finally: conn.close() def expected_total_for(db: Session, source: Source, mode: str, limit: int | None, refs_count: int | None) -> int | None: """What the progress bar divides by. Recheck: the exact count of held listings to walk. Discover: `products_seen` of the newest run that ended ok WITHOUT a limit (a `--limit 30` rehearsal must never become the denominator), capped by this run's limit; None when no such run exists, and the page then shows no percent.""" if mode == "recheck": total = refs_count or 0 else: total = db.scalar( select(CollectionRun.products_seen) .where(CollectionRun.source_id == source.id, CollectionRun.status == "ok", CollectionRun.limit_n.is_(None)) .order_by(CollectionRun.started_at.desc()).limit(1) ) if total is None: return None return min(limit, total) if limit else total def _price_key(price, currency) -> tuple[float, str] | None: """What "the price moved" compares: the figure to the cent and the currency, never a Numeric against a float.""" if price is None: return None return (round(float(price), 2), (currency or "").upper()) class _Counted: """What the run's counters are measured against, loaded once before the loop (Stream AW4, D5): the listings held at the run's shops, each one's newest price, and every brand and product line id. Counted after resolution in the loop body, never from ORM events, so a savepoint the barcode race rolls back inflates nothing.""" def __init__(self, db: Session, shop_ids: list[int]) -> None: from app.models import ProductLine self.held_at_start: set[int] = set() self.last_price: dict[int, tuple[float, str] | None] = {} self.seen_ids: set[int] = set() if shop_ids: self.held_at_start = set(db.scalars(select(Listing.id).where(Listing.shop_id.in_(shop_ids)))) # The newest observation per listing, as a window so it reads the same on SQLite. newest = func.row_number().over( partition_by=PriceObservation.listing_id, order_by=(PriceObservation.observed_at.desc(), PriceObservation.id.desc()), ).label("rn") ranked = ( select(PriceObservation.listing_id, PriceObservation.price, PriceObservation.currency, newest) .join(Listing, Listing.id == PriceObservation.listing_id) .where(Listing.shop_id.in_(shop_ids)) .subquery() ) for listing_id, price, currency in db.execute( select(ranked.c.listing_id, ranked.c.price, ranked.c.currency).where(ranked.c.rn == 1) ): self.last_price[listing_id] = _price_key(price, currency) self.known_brand_ids: set[int] = set(db.scalars(select(Brand.id))) if _has_table(db, "brands") else set() self.known_line_ids: set[int] = set(db.scalars(select(ProductLine.id))) if _has_table(db, "product_lines") else set() def count(self, run: CollectionRun, listing: Listing, product: ProductVariant, raw: RawListing, *, arrived: bool) -> None: """One resolved sighting. A second sighting of a listing in the same run counts nothing more (a multi-size tile family); a brand or line first seen this run counts once.""" if arrived: run.new_variants += 1 if listing.id not in self.seen_ids: if listing.id in self.held_at_start: run.existing_checked += 1 if self.last_price.get(listing.id) != _price_key(raw.price, raw.currency): run.existing_changed += 1 else: run.new_found += 1 self.seen_ids.add(listing.id) if product.brand_id is not None and product.brand_id not in self.known_brand_ids: self.known_brand_ids.add(product.brand_id) run.new_brands += 1 if product.product_line_id is not None and product.product_line_id not in self.known_line_ids: self.known_line_ids.add(product.product_line_id) run.new_lines += 1 def missing(self) -> int: """The held listings the walk never met: a measure only of an unlimited discover run that ended ok, which is the one kind that saw the whole shop.""" return len(self.held_at_start - self.seen_ids) def _listings(collector: Collector, *, mode: str, limit: int | None, delay: float, refs: list[ListingRef], run: CollectionRun): """The generator seam between the collector and the loop body (Stream AW4, D5). `discover` yields what `collect()` yields. `recheck` walks the held listings through `read_one` under the same hook: a listing the source no longer offers counts as missing, a retryable fetch failure is a counted skip, a block or an unreadable robots.txt propagates and ends the run as it always did. No listing state is written here. """ if mode != "recheck": yield from collector.collect(limit=limit, delay=delay) return for ref in refs[:limit] if limit else refs: try: raw = collector.read_one(ref) except PageGone as exc: # Exactly what `existing_missing` counts, and the same outcome as a read_one that # answers None: the shop no longer offers this listing. It is counted and named, # never a skip (that would double-count it) and never the end of the run: one # 404 on a held product once ended a whole recheck `blocked`, which is a word # reserved for the host refusing us and disables Start until rian lifts it. logger.info("recheck_listing_gone source=%s detail=%s", collector.slug, exc) run.existing_missing += 1 continue except FetchError as exc: if gone(exc): run.existing_missing += 1 else: count_skip(run, "recheck_fetch_error") continue if raw is None: run.existing_missing += 1 continue yield raw def run_collector( db: Session, collector: Collector, *, limit: int | None = None, delay: float | None = None, rates: FxRates | None = None, mode: str = "discover", by: str | None = None, sleeper=None, clock=None, ) -> CollectionRun: """Collect one source into the database, recording the run either way. The control plane (Stream AW4): the child's advisory lock comes first, then the run row with its pid, mode, limit and who started it, committed at once so the page sees it within a poll. The hook is bound for the whole loop; a SIGTERM only sets its flag (the raise happens at the next poll, never mid-flush), and a cooperative end commits the batch and writes `stopped` with `stopped_by`. `sleeper` and `clock` are for the tests. """ source = ensure_source(db, collector) db.commit() # the source row must exist before the lock's own connection looks for a run lock_conn = _try_lock(db.get_bind(), collector.slug) # 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), pid=os.getpid(), mode=mode, limit_n=limit, started_by=by or "shell", ) db.add(run) db.flush() if not source.enabled: run.status = "skipped" run.error = "source disabled" run.finished_at = datetime.now(UTC) db.commit() _release_lock(lock_conn, collector.slug) 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) refs: list[ListingRef] = [] if mode == "recheck" or getattr(collector, "wants_held_listings", False): refs = held_listing_refs(db, list(shops.values())) if mode != "recheck": collector.held_listings = refs run.expected_total = expected_total_for(db, source, mode, limit, len(refs)) db.commit() if mode == "recheck" and not refs: run.status = "skipped" run.error = "nothing to recheck: no held listing with a URL" run.finished_at = datetime.now(UTC) db.commit() _release_lock(lock_conn, collector.slug) return run effective_delay = float(source.delay_seconds) if delay is None else delay counted = _Counted(db, [shop.id for shop in shops.values()]) # 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], ) hook = run_control.RunControl( run_id=run.id, source_id=source.id, slug=collector.slug, db=db, run_started_at=run.started_at, **({"sleeper": sleeper} if sleeper else {}), **({"clock": clock} if clock else {}), ) previous_handler = None on_main_thread = threading.current_thread() is threading.main_thread() if on_main_thread: def _on_term(signum, frame): # noqa: ARG001 hook.stop_flag.set() # never raises here: the hook raises at the next poll logger.warning("collector_signalled source=%s signal=%s", collector.slug, signum) previous_handler = signal.signal(signal.SIGTERM, _on_term) token = bind_skip_sink(collector_skipped) control_token = run_control.bind_control(hook) try: for raw in _listings(collector, mode=mode, limit=limit, delay=effective_delay, refs=refs, run=run): 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) arrived = bool(getattr(product, "_arrived", False)) if arrived: product._arrived = False _propose_arrival(db, product) # The counters, after resolution (D5): what this sighting was, not what it wrote. counted.count(run, listing, product, raw, arrived=arrived) # 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" # Only a walk that saw the whole shop can say what it no longer offers; a limited run # or a recheck (which counts its own misses as it goes) leaves this "not measured". if mode != "recheck" and limit is None: run.existing_missing = counted.missing() except run_control.RunStopped as stopped: # A cooperative end: the batch first, then the row. If the batch cannot be committed, # the row still says stopped and why the batch was lost. logger.warning("collector_stopped source=%s reason=%s by=%s", collector.slug, stopped.reason, stopped.by) try: db.commit() except SQLAlchemyError as exc: db.rollback() run.error = f"batch lost at stop: {type(exc).__name__}" if stopped.reason != "superseded": _end_stopped(db, run, stopped.by) else: run = db.get(CollectionRun, run.id) or run except PageGone as exc: # A page that is not there is OUR problem to skip, never the host refusing us. Every # walk skips one and carries on, so reaching here means the walk could not: the run # read nothing, which is `error` (ours, retried next run), never `blocked` (theirs, # final until rian lifts it). The URL is in the message. run.status = "error" run.error = f"a page the walk could not skip is gone: {exc}" logger.warning("collector_page_gone source=%s detail=%s", collector.slug, exc) 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) finally: run_control.unbind_control(control_token) unbind_skip_sink(token) if on_main_thread and previous_handler is not None: signal.signal(signal.SIGTERM, previous_handler) _release_lock(lock_conn, collector.slug) if run.finished_at is None: run.finished_at = datetime.now(UTC) db.commit() # The hook wrote heartbeat_at and requests_made past this object with a guarded UPDATE; # one SELECT gives the caller the row as the database holds it. db.refresh(run) logger.info( "collector_finished source=%s status=%s prices=%d", collector.slug, run.status, run.prices_written, ) return run def _end_stopped(db: Session, run: CollectionRun, by: str) -> None: """`stopped` written with the same guard the heartbeat uses: only a row still `running` is ours to close, so a row another process already closed is left as it wrote it.""" from sqlalchemy import update result = db.execute( update(CollectionRun) .where(CollectionRun.id == run.id, CollectionRun.status == "running") .values(status="stopped", stopped_by=by[:80], finished_at=datetime.now(UTC), error=run.error, products_seen=run.products_seen, prices_written=run.prices_written, skipped_no_price=run.skipped_no_price, skip_counts=run.skip_counts, existing_checked=run.existing_checked, existing_changed=run.existing_changed, existing_missing=run.existing_missing, new_found=run.new_found, new_brands=run.new_brands, new_lines=run.new_lines, new_variants=run.new_variants) .execution_options(synchronize_session=False) ) db.commit() if result.rowcount: run.status, run.stopped_by = "stopped", by[:80] run.finished_at = run.finished_at or datetime.now(UTC) else: db.refresh(run)