"""Turning messy retailer strings into comparable facts. Everything here operates on factual attributes (size, strength, brand, name). Retailer marketing copy is deliberately never parsed or stored. """ import re import unicodedata # Multipacks first: parse_quantity_ml's single-size patterns happily match the # "1L" inside "24 x 1L", recording a case of beer as one bottle. The pack # total is the comparable fact. _MULTIPACK_RE = re.compile( r"(\d{1,2})\s*(?:bottles?\s*|cans?\s*|pack\s*)?[x\u00d7]\s*" r"(\d+(?:[.,]\d+)?)\s*(litre|liter|ltr|lt|l|cl|ml)\b", re.I, ) _UNIT_FACTORS = {"litre": 1000.0, "liter": 1000.0, "ltr": 1000.0, "lt": 1000.0, "l": 1000.0, "cl": 10.0, "ml": 1.0} _SIZE_PATTERNS: list[tuple[re.Pattern[str], float]] = [ (re.compile(r"(\d+(?:[.,]\d+)?)\s*(?:litre|liter|ltr|lt)\b", re.I), 1000.0), (re.compile(r"(\d+(?:[.,]\d+)?)\s*l\b", re.I), 1000.0), (re.compile(r"(\d+(?:[.,]\d+)?)\s*cl\b", re.I), 10.0), (re.compile(r"(\d+(?:[.,]\d+)?)\s*ml\b", re.I), 1.0), ] # Not preceded by a digit: "250%" must not yield 50. _ABV_RE = re.compile(r"(? int | None: """Extract a size in millilitres from a product name (pack total for multipacks).""" if not text: return None pack = _MULTIPACK_RE.search(text) if pack: count = int(pack.group(1)) unit = float(pack.group(2).replace(",", ".")) * _UNIT_FACTORS[pack.group(3).lower()] millilitres = round(count * unit) if 10 <= millilitres <= 50000: return millilitres for pattern, factor in _SIZE_PATTERNS: match = pattern.search(text) if match: value = float(match.group(1).replace(",", ".")) # A sub-unit centilitre value is a typo for litres: "0.70cl" means # 0.70L; a genuine 7 ml bottle of spirits does not exist. if factor == 10.0 and value < 1: factor = 1000.0 millilitres = round(value * factor) if 10 <= millilitres <= 20000: return millilitres return None _UNIT_LABELS = {"litre": "l", "liter": "l", "ltr": "l", "lt": "l", "l": "l", "cl": "cl", "ml": "ml"} def parse_size(text: str | None) -> tuple[float, str] | None: """The size as the name states it: (70, "cl"), (100, "ml"), (1.5, "l"). The stated form is what a shopper reads and what `quantity_stated_value`/`quantity_stated_unit` store; `quantity_ml` is derived from it (`quantity_ml_of`). A multipack is stated as its total in ml, the only comparable fact about it. """ if not text: return None pack = _MULTIPACK_RE.search(text) if pack: total = parse_quantity_ml(text) return (float(total), "ml") if total else None for pattern, factor in _SIZE_PATTERNS: match = pattern.search(text) if match: value = float(match.group(1).replace(",", ".")) unit = _UNIT_LABELS[match.group(0)[len(match.group(1)):].strip().lower()] if unit == "cl" and value < 1: unit = "l" if 10 <= value * _UNIT_FACTORS[unit] <= 20000: return (value, unit) return None def quantity_ml_of(value: float | None, unit: str | None) -> int | None: """Millilitres for a stated volume; None for a weight or an unknown unit.""" factor = _UNIT_FACTORS.get(str(unit or "").lower()) if value is None or factor is None: return None return round(float(value) * factor) # Looser than the parsing pattern on purpose: a guard only needs to recognise # a pack, and a feed's own suffix glued to the unit ("24x0.33LDS") once hid one. _LOOKS_MULTIPACK_RE = re.compile(r"\d{1,2}\s*(?:bottles?\s*|cans?\s*|pack\s*)?[x\u00d7]\s*\d+(?:[.,]\d+)?\s*(?:l|cl|ml)", re.I) def is_multipack(text: str | None) -> bool: """Whether a name describes a case or pack ("24 x 330ml"), whose total is legitimately large.""" return bool(text and _LOOKS_MULTIPACK_RE.search(text)) def size_is_implausible(vertical: str | None, name: str, quantity_ml: int | None) -> bool: """A stored size no single item of that family comes in, unless the name itself says so (the ceilings live in `quantity.CEILING_ML`; this wrapper stays for its callers).""" from app.services import quantity # quantity builds on this module return quantity.is_implausible(vertical, name, quantity_ml) _URL_PACK_RE = re.compile(r"(? int | None: """A size named in a product URL's slug, for feeds whose display names omit it. Slugs write dots as hyphens ("0-7l") and can end in numeric product codes, so the whole path is scanned with the hyphen forms normalised first. """ if not url: return None path = url.split("://")[-1].split("?")[0].lower() path = path.split("/", 1)[1] if "/" in path else path path = re.sub(r"(\d)-(\d)", r"\1.\2", path).replace("-", " ").replace("/", " ") pack = _URL_PACK_RE.search(path) if pack: total = round(int(pack.group(1)) * float(pack.group(2)) * _UNIT_FACTORS[pack.group(3)]) return total if 10 <= total <= 50000 else None match = _URL_SIZE_RE.search(path) if not match: return None value, unit = float(match.group(1)), match.group(2) if unit == "cl" and value < 1: unit = "l" # Slugs drop decimal points: "armand-de-brignac-gold-15l" is the 1.5 litre # (the page's own anchor says #1.5l). A bare two-digit litre count is that # convention, not a fifteen-litre bottle. if unit == "l" and value >= 10 and "." not in match.group(1): value = value / 10 millilitres = round(value * _UNIT_FACTORS[unit]) return millilitres if 10 <= millilitres <= 20000 else None # Fragrance concentration, read from the name. It is the beauty vertical's # attribute in the sense of Decision 6: it VETOES a fallback match (an EDP and # an EDT of the same line and size are different bottles) and is never part of # the key, because most feeds omit it and a missing value must not split a # product. Order matters: "Eau de Parfum" is edp, not parfum. _CONCENTRATION_RULES: list[tuple[str, re.Pattern[str]]] = [ ("edp", re.compile(r"\b(eau\s+de\s+parfum|edp)\b", re.I)), ("edt", re.compile(r"\b(eau\s+de\s+toilette|edt)\b", re.I)), ("edc", re.compile(r"\b(eau\s+de\s+cologne|edc|cologne)\b", re.I)), ("mist", re.compile(r"\b(mist|brume|bruma)\b", re.I)), ("parfum", re.compile(r"\b(parfum|extrait|perfume)\b", re.I)), ] def parse_concentration(text: str | None) -> str | None: """edp | edt | edc | parfum | mist, or None when the name does not say.""" if not text: return None for label, pattern in _CONCENTRATION_RULES: if pattern.search(text): return label return None def parse_abv(text: str | None) -> float | None: """Extract alcohol strength as a percentage.""" if not text: return None match = _ABV_RE.search(text) if not match: return None value = float(match.group(1).replace(",", ".")) return value if 0 < value <= 100 else None def _has_valid_check_digit(digits: str) -> bool: """GS1 check-digit test (EAN-8/12/13/14). This matters more than it looks: some shops put a sequential internal SKU in the barcode field. Those pass a length test, and two shops numbering their own product variants from 1 would then "match" unrelated bottles. The check digit is what separates a real barcode from a counter. """ body, check = digits[:-1], int(digits[-1]) total = 0 for index, char in enumerate(reversed(body)): total += int(char) * (3 if index % 2 == 0 else 1) return (10 - total % 10) % 10 == check def clean_gtin(raw: object) -> str | None: """Normalise a barcode to digits, left-padded to 13, or None if it isn't one. Retailers pad inconsistently, so the same barcode arrives as 50196081, 0050196081 and 0000050196081. Normalising is what makes the join work. """ if raw is None: return None digits = re.sub(r"\D", "", str(raw)) trimmed = digits.lstrip("0") # Strip padding BEFORE validating: retailers zero-pad to varying widths, so # the same barcode arrives as 50196081, 0050196081 and 0005000267013602. if not 6 <= len(trimmed) <= 14: return None padded = trimmed.rjust(13, "0") if len(trimmed) < 13 else trimmed if not _has_valid_check_digit(padded): return None # GS1 reserves 02 and 20-29 prefixes for company-internal ("restricted # circulation") numbering. Two operators can put the same such code on # different bottles, so it is not a global identity and must not drive # cross-retailer joins -- one already paired a limited edition with the # regular bottle. These rows fall back to the brand+name+size key. if padded.startswith("02") or (len(padded) == 13 and padded[0] == "2"): return None return padded # The only widths a real GTIN comes in. _GTIN_LENGTHS = frozenset({8, 12, 13, 14}) def gtin_from_sku(raw: object) -> str | None: """A barcode read out of a SKU field, which is a different thing entirely. Some shops do put the EAN in the SKU field, so we cannot ignore it: 30% of the SKUs one platform publishes agree exactly with barcodes another retailer supplies for the same bottle. But a supplier code that merely looks numeric must never become a barcode. A 7-digit Magento code padded out to 13 has a one-in-ten chance of passing the check digit, and the damage from the ones that pass is worse than the miss: the fake permanently occupies the unique index the real barcode needs, it joins unrelated bottles, and it is published as a gtin13 that Google validates against GS1. So the rule is narrower than clean_gtin: a supplier code is a barcode only if it already IS one of the real GTIN widths. We never invent width for it. Measured against the catalogue, this rejects 61 codes of which exactly one was corroborated by a second retailer, and keeps 148 of the 149 that were. """ if raw is None: return None digits = re.sub(r"\D", "", str(raw)) if len(digits) not in _GTIN_LENGTHS: return None return clean_gtin(digits) # Words that may lead a name without being part of it: the brand's own # connectors and the article a shop puts in front of a brand name. _LEADING_FILLER = {"the", "and", "et", "y", "de", "by"} # Packaging that turns a bottle into a different offer: a set with a shower # gel is not the bottle, and must never fold into it (a Boss Bottled once did). _SET_RE = re.compile(r"\b(gift\s*set|set|coffret|estuche|kit|duo|trio|collection|bundle)\b", re.I) def looks_like_set(text: str | None) -> bool: """Whether a name describes a set or kit rather than a single item.""" return bool(text and _SET_RE.search(text)) def _strip_brand_prefix(tokens: list[str], brand_sequences: list[list[str]]) -> list[str]: """The name without the brand it repeats at its head. Strips the longest run that is a PREFIX of the brand's own word sequence ("Paris Saint Germain" off "Paris Saint Germain paris cap", leaving the second "paris" that tells the two caps apart), else the brand's last word alone ("Armani" off "Armani Code"). Never every leading brand word: "Nina Ricci Ricci Ricci" is a scent called Ricci Ricci, and a rule that ate all four words merged it with "Nina" (2026-09-05, dev rehearsal). """ while tokens and tokens[0] in _LEADING_FILLER: tokens = tokens[1:] best = 0 for sequence in brand_sequences: n = 0 while n < len(sequence) and n < len(tokens) and tokens[n] == sequence[n]: n += 1 best = max(best, n) if best == 0: for sequence in brand_sequences: if sequence and tokens and tokens[0] == sequence[-1]: best = 1 break stripped = tokens[best:] while stripped and stripped[0] in _LEADING_FILLER: stripped = stripped[1:] return stripped or tokens def match_key(listed_brand: str | None, name: str, quantity_ml: int | None, *, vertical: str | None = None, brand: str | None = None, attribute: str | None = None, line: str | None = None, quantity=None, category: str | None = None, options: dict | None = None) -> str: """The certain key of a product variant with no barcode: identity rules v6 (plan W14). `brand | residual name | identity attributes | quantity`. The automatic merge and ingest's fallback join fire only on this key agreeing, so every part is something the shop stated or a fold that cannot change what was stated: * the brand is the brand row's slug words with its alias followed when the caller resolved one (a Confirm same on a brand pair reaches every key under it), else the fold of the listed brand; * the residual name is `product_lines.product_line_key` (the certain removals only); `line` is the caller's line key when the line row is known, so a line alias a person confirmed reaches the key from both ingest and rederive; * the identity slot is `product_lines.identity_slot`: the concentration (or a stored name's marked tail), a percentage the name states, every shop-published option; `attribute` is the canonical wording when the caller resolved it through the alias table; * the quantity slot is spelled by `quantity.quantity_key` and nowhere else (`100ml`, `50g`, `200pcs`, a pack on its total, a set on its contents, `unknown` never equal to anything). The key is a pure function of the listed facts and the alias maps; a person's decision on a variant (its line, an attribute, its quantity) never moves it, or the variant's own listing would key elsewhere at its next sighting and mint a duplicate (the lesson `keyed_name` already carries for a typed name). History: v5 put the line rules' word lists between the name and this key (72 drink words, 66 format words, connectors), v3 keyed the quantity on a bare millilitre figure, v2 kept the concentration in the name so "1 Million EDT 100ml" and "1 Million Eau de Toilette 100 ml" were two variants at two airports for two weeks. `rederive` brings old rows onto this key. """ from app.services import product_lines, quantity as quantity_service # both build on this module brand_part = brand_key(brand) or brand_key(listed_brand) if line is None: line = product_lines.product_line_key(name, listed_brand=listed_brand, brand=brand, vertical=vertical, category=category) identity = product_lines.identity_slot(name, vertical, category=category, canonical=attribute, options=options) if quantity is None: quantity = quantity_service.parse_quantity(name, hint=(quantity_ml, "ml") if quantity_ml else None) parts = [brand_part.replace(" ", "-"), line.replace(" ", "-"), identity, quantity_service.quantity_key(quantity)] return "|".join(parts)[:255] _EXCLUSIVE_RE = re.compile(r"travel\s*(?:retail\s*)?exclusive|airport\s+exclusive", re.I) # Trademark and service-mark glyphs, removed BEFORE the NFKD fold. NFKD gives them a # compatibility decomposition to LETTERS -- U+2122 becomes "TM" and U+2120 becomes "SM" -- which # then survive the ASCII step and sit in the key: "CIROC(tm)" keyed as `ciroctm`, a different # brand row from `ciroc`, on seven pairs of rows in one dump. (U+00AE and U+00A9 decompose to # nothing and were always harmless; the defect is the two that become letters.) _TRADEMARK_RE = re.compile("[\u2122\u00ae\u00a9\u2120\u2117]") # Words a retailer appends to a brand that are not the brand: the category it # sells in, or a corporate suffix. NOT folded off the key (K11.3) -- the list is # read by `proposal_rules.trailer_words`, which proposes the join for a person. _BRAND_TRAILERS = { "whisky", "whiskey", "scotch", "bourbon", "gin", "vodka", "rum", "rhum", "tequila", "mezcal", "cognac", "brandy", "armagnac", "liqueur", "liqueurs", "wine", "wines", "champagne", "spirits", "distillery", "distillers", "distilleries", "brewery", "vineyards", "vineyard", "estate", "estates", "winery", "cellars", "co", "company", "ltd", "inc", "sa", "srl", "gmbh", "brands", "parfums", "parfum", "fragrances", "beauty", "cosmetics", "paris", "london", "the", } def brand_key(brand: str | None) -> str: """One key for every spelling of a brand: the brand row a listed spelling lands on. Case, accents, punctuation and the trademark glyphs are folded ("Moët & Chandon", "MOET ET CHANDON" and "moet-chandon" agree; "CÎROC™" and "Cîroc" agree); "&", "and" and "et" are dropped; a leading article is dropped. Empty when there is no brand. Deliberately does NOT fold spaces away: "Glen Moray" and "Glenmorangie" must never meet, and a hyphenated brand name keeps its parts. **It folds nothing else, and in particular no trailing word** (K11.3). It used to pop words off the end while they were in `_BRAND_TRAILERS`, so "Tanqueray Gin" and "Tanqueray" became one brand row with no decision behind it -- the programmatic stage JOINING, which rian's ruling of 17 Sep forbids: that stage may act only toward SEPARATE brands, and every join is the AI pass's to propose and a person's to make. The fold also mangled names outright, because the list holds ordinary words: "L'Oréal Paris" keyed as `loreal`, "Souvenir de Paris" as `souvenir-de`, "Au Vodka" as `au`. The list itself is kept and read by `proposal_rules.trailer_words`, which proposes the join with its reason. """ if not brand: return "" text = _TRADEMARK_RE.sub("", brand) text = unicodedata.normalize("NFKD", text).encode("ascii", "ignore").decode().lower() text = re.sub(r"['\u2019`.]", "", text) words = [w for w in _PUNCT_RE.sub(" ", text).split() if w not in {"and", "et", "y"}] while len(words) > 1 and words[0] == "the": words.pop(0) return " ".join(words) def looks_exclusive(text: str | None) -> bool: """Whether a product name itself declares it a travel-retail exclusive.""" return bool(text and _EXCLUSIVE_RE.search(text)) _STOPWORDS = { "the", "and", "of", "de", "la", "le", "el", "old", "years", "year", "yr", "yrs", "aged", "vol", "single", "blended", "blend", "premium", "reserve", "original", "whisky", "whiskey", "vodka", "gin", "rum", "tequila", "cognac", "liqueur", "scotch", "malt", "bourbon", "brandy", "wine", "spirits", "edition", } _FLAT_RE = re.compile(r"[^a-z0-9]+") def flat_key(text: str | None) -> str: """The flattened text two spellings of a name are compared on: lowercase ASCII, one space between words, nothing else (`product_variants.name_key`, `listings.listed_name_key`; the collectors page's "name differs").""" folded = unicodedata.normalize("NFKD", text or "").encode("ascii", "ignore").decode().lower() return " ".join(_FLAT_RE.sub(" ", folded).split()) def name_tokens(text: str) -> set[str]: """Significant words in a product name, for comparing two spellings of it.""" cleaned = unicodedata.normalize("NFKD", text or "").encode("ascii", "ignore").decode() cleaned = _NOISE_RE.sub(" ", cleaned.lower()) for pattern, _ in _SIZE_PATTERNS: cleaned = pattern.sub(" ", cleaned) cleaned = _ABV_RE.sub(" ", cleaned) words = {w for w in _PUNCT_RE.sub(" ", cleaned).split() if len(w) > 1} significant = words - _STOPWORDS # A name made entirely of stopwords ("Single Malt") still needs something to # compare on, so fall back to the raw words rather than an empty set. return significant or words