"""Operational commands. python -m app.cli collect [--source SLUG] [--limit N] [--delay SECONDS] python -m app.cli backfill # idempotent data moves, run after a deploy python -m app.cli rederive # recompute match keys under the current identity rules python -m app.cli status Migrations are schema-only (build plan §2); every data move is a backfill here, safe to run twice. """ import argparse import logging import sys from sqlalchemy import func, select, update from pathlib import Path from app.db import SessionLocal from app.models import ( Account, Award, Brand, CollectionRun, Listing, Shop, PriceObservation, ProductVariant, ProductLine, RawRecord, Retailer, Source, AttributeAlias, ) from app.services.collectors.registry import COLLECTORS, get_collector from app.services.fx import load_rates from app.services.awards_import import import_awards from app.services.discussion_import import import_discussion from app.services.ingest import ( DEFAULT_VERTICAL, is_stuck, mark_stuck, run_collector, sync_shop_currency, ) from app.models.catalog import IDENTITY_RULES_VERSION from app.services.ingest import brand_slug, raw_attributes from app.services.collectors.base import RawListing from app.services import imagery, keying from app.services import product_lines from app.services.merges import merge_duplicates from app.services.normalize import parse_size, size_is_implausible, quantity_ml_of from app.services.taxonomy import classify, vertical_of from app.cli_pages import register as register_pages from app.cli_quality import register as register_quality from app.cli_editorial import register as register_editorial from app.cli_accounts import register as register_accounts from app.cli_refresh import register as register_refresh from app.cli_decisions import register as register_decisions from app.cli_index import register as register_index from app.cli_brands import register as register_brands from app.cli_pass import register as register_pass from app.cli_precedents import register as register_precedents from app.cli_proposals import register as register_proposals from app.cli_hours import register as register_hours from app.cli_images import register as register_images from app.cli_featured import register as register_featured from app.cli_places import register as register_places DEFAULT_WINNERS_PATH = Path("/srv/import/winners.json") DEFAULT_DISCUSSION_PATH = Path("/srv/import/discussion.json") logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") logger = logging.getLogger(__name__) def cmd_collect(args: argparse.Namespace) -> int: from app.config import settings from app.services.collectors.control import CollectorLocked slugs = [args.source] if args.source else list(COLLECTORS) # The deploy freeze (Stream AW4): a deploy kills every collector in the container, so while # its marker exists nothing starts. Exit 3 like the lock: not a failure of the source. marker = settings.collect_freeze_file if marker.exists() and not args.ignore_freeze: stamp = (marker.read_text().strip().splitlines() or ["no stamp"])[0] logging.error("collect_frozen marker=%s stamp=%r (use --ignore-freeze to override)", marker, stamp) return 3 rates = load_rates() if rates.is_fallback: logging.warning("using fallback exchange rates; USD comparisons may be stale") exit_code = 0 with SessionLocal() as db: for slug in slugs: try: collector = get_collector(slug) except KeyError: logging.error("unknown source %s (known: %s)", slug, ", ".join(COLLECTORS)) exit_code = 2 continue try: run = run_collector( db, collector, limit=args.limit, delay=args.delay, rates=rates, mode=args.mode, by=args.by, ) except CollectorLocked as locked: print(f"{slug}: locked ({locked})") exit_code = 3 continue print(f"{slug}: {run.status} prices={run.prices_written} error={run.error or '-'}") if run.status not in {"ok", "skipped", "stopped"}: exit_code = 1 return exit_code def cmd_awards(args: argparse.Namespace) -> int: path = Path(args.file) if args.file else DEFAULT_WINNERS_PATH if not path.is_file(): logging.error("winners file not found: %s", path) return 2 with SessionLocal() as db: stats = import_awards(db, path, rebuild=args.rebuild) print( f"winners={stats['winners']} matched={stats['matched']} " f"created={stats['created']} already_present={stats['skipped_existing']} " f"ambiguous={stats['ambiguous']}" ) # The reconcile counters (Stream F, e2d50a3): what a re-import changed. print( f"updated={stats.get('updated', 0)} corrected={stats.get('corrected', 0)} " f"removed={stats.get('removed', 0)} unchanged={stats.get('unchanged', 0)} " f"ambiguous_same_year={stats.get('ambiguous_same_year', 0)}" ) return 0 def cmd_discussion(args: argparse.Namespace) -> int: path = Path(args.file) if args.file else DEFAULT_DISCUSSION_PATH if not path.is_file(): logging.error("discussion file not found: %s", path) return 2 with SessionLocal() as db: stats = import_discussion(db, path) print(f"created={stats['created']} updated={stats['updated']}") return 0 def cmd_discussion_inbox(args: argparse.Namespace) -> int: """What rian said to Claude: every comment naming @claude on the running list (the owner's subjects), oldest first, with its deep link. A session reads this on its next run (agents.md, the checkpoint command); nobody is notified of these, by design.""" from datetime import UTC, datetime from app.services.discussion import claude_inbox if args.recipient != "claude": print(f"discussion inbox: the only recipient is claude, not {args.recipient!r}") return 2 since = datetime.fromisoformat(args.since).replace(tzinfo=UTC) if args.since else None with SessionLocal() as db: rows = claude_inbox(db, since) if not rows: print("inbox for claude: nothing" + (f" since {args.since}" if args.since else "")) return 0 for r in rows: where = r["label"] or f"{r['subject_type']}:{r['subject_id']}" state = " (resolved)" if r["resolved"] else "" print(f"[{(r['at'] or '')[:16]}] {r['author']} on {where}{state}\n {r['body']}\n {r['url']}") print(f"{len(rows)} comment(s) for claude") return 0 def cmd_discussion_rethread(args: argparse.Namespace) -> int: """Apply a re-threading map once (T18): the topics it names, one move per comment, the listed soft deletes, every action attributed to the owner account. `--check` prints what would happen and writes nothing; `--apply` does it; a second `--apply` reports every entry as already done. The map is a file rian may edit before applying (`import/rethread-*.json`).""" import json as _json from sqlalchemy import select as _select from app.config import settings as _settings from app.models import Account from app.services.discussion import rethread from app.services.identity import Actor if not (args.check or args.apply) or (args.check and args.apply): print("discussion rethread: pass exactly one of --check and --apply") return 2 # The workspace's import/ is mounted read-only at /srv/import inside the container (the plan # and items routes read it the same way), so a relative default resolves there first. candidates = [Path(args.file), Path("/srv/import") / Path(args.file).name] path = next((c for c in candidates if c.is_file()), None) if path is None: print(f"discussion rethread: no such file: {args.file} (looked in /srv/import too)") return 2 plan = _json.loads(path.read_text()) with SessionLocal() as db: owner = db.scalar(_select(Account).where(Account.username == (_settings.account_owner or ""))) if owner is None: print("discussion rethread: no owner account to attribute the moves to (ACCOUNT_OWNER)") return 2 lines = rethread(db, plan, actor=Actor(owner.id, owner.username, owner.display_name), check=args.check) for line in lines: print(line) if args.check: db.rollback() # the walk wrote nothing; this only ends the transaction print(f"[check, nothing written] {len(lines)} entr{'y' if len(lines) == 1 else 'ies'} from {path.name}") return 0 db.commit() print(f"discussion rethread: applied {len(lines)} entr{'y' if len(lines) == 1 else 'ies'} from {path.name}") return 0 OWNER_DISPLAY_NAME = "rian" def backfill_accounts(db) -> str: """Seed the owner's account row (migration #1) and give it its username and status (migration #4) where they are NULL. Email is set later, by hand.""" existing = db.scalar(select(Account).where(Account.display_name == OWNER_DISPLAY_NAME)) if existing is None: db.add(Account(display_name=OWNER_DISPLAY_NAME, username=OWNER_DISPLAY_NAME, status="active")) db.commit() return "accounts: owner row created" changed = [] if existing.username is None: existing.username = OWNER_DISPLAY_NAME changed.append("username") if not existing.status: existing.status = "active" changed.append("status") db.commit() if changed: return f"accounts: owner row id={existing.id} gained {', '.join(changed)}" return f"accounts: owner row already present (id={existing.id})" def backfill_levels(db) -> str: """Seed the two launch levels (accounts plan §4.6): insert if absent, never update an edited level (the kit has no rename, so a name is fixed at its first seed).""" from app.models import AccountLevel from app.services.accounts import SEED_LEVELS created = [] for name, spec in SEED_LEVELS.items(): if db.get(AccountLevel, name) is None: db.add(AccountLevel(name=name, permissions=list(spec["permissions"]), assignable=list(spec["assignable"]))) created.append(name) db.commit() return f"levels: created {', '.join(created) if created else 'none'} (present: {', '.join(sorted(SEED_LEVELS))})" def backfill_level_permissions(db) -> str: """Grant a level a permission that joined its seed after the level was created. `backfill levels` inserts and never updates, so a host seeded before a permission existed keeps the old set forever and the page-status badge would never appear for its `admin` accounts. This adds the difference, and ONLY where the level still holds exactly what the previous seed gave it: a level someone edited on /admin is a human value, and a human value is never overwritten by a machine (it is named in the message instead). Idempotent: the second run reports nothing to do.""" from app.models import AccountLevel from app.services.accounts import PREVIOUS_SEED_PERMISSIONS, SEED_LEVELS granted, current, edited, absent = [], [], [], [] for name, spec in SEED_LEVELS.items(): row = db.get(AccountLevel, name) wanted = list(spec["permissions"]) if row is None: absent.append(name) continue held = list(row.permissions or []) if set(held) == set(wanted): current.append(name) elif set(held) == set(PREVIOUS_SEED_PERMISSIONS.get(name, [])): row.permissions = wanted granted.append(f"{name} (+{', '.join(sorted(set(wanted) - set(held)))})") else: edited.append(name) db.commit() bits = [f"level_permissions: granted {', '.join(granted) if granted else 'none'}"] if current: bits.append(f"{', '.join(current)} already current") if absent: bits.append(f"{', '.join(absent)} not seeded yet (`backfill levels` first)") for name in edited: bits.append(f"{name} was edited by hand: grant it on /admin") return "; ".join(bits) # Tables whose typed who-column links by folded text. The two the SPA silently defaulted to # "Adam" for an anonymous actor link only under --include-defaulted (rian's call, 10 Sep). AUTHOR_COLUMNS = ( ("discussion_comments", "author", "author_id"), ("quote_requests", "author", "author_id"), ("client_todos", "completed_by", "completed_by_id"), ("client_uploads", "uploaded_by", "uploaded_by_id"), ("owner_item_states", "acted_by", "acted_by_id"), ) DEFAULTED_AUTHOR_COLUMNS = ( ("feature_priorities", "author", "author_id"), ("quote_selections", "author", "author_id"), ) def backfill_authors(db, args=None) -> str: """Link typed names to accounts: `--map Adam=adam` (repeatable) folds the typed text and sets the who-column's id where it is NULL; every other spelling stays text with a NULL link (empty beats guessed; no guest accounts are minted). `--rename` (rian, 14 Sep) also rewrites each mapped spelling to the account's display name on the rows it links or has linked ("Mark (by email, 7 Sep)" reads "Mark"): the one deliberate overwrite of a typed value, asked for by the person who typed it, never implied by `--map` alone. Idempotent: a second run changes zero rows.""" from sqlalchemy import table as sa_table, column as sa_column pairs = list(getattr(args, "map", None) or []) if not pairs: return "authors: nothing to do (pass --map Typed=username, repeatable)" folded: dict[str, Account] = {} for pair in pairs: typed, _, username = pair.partition("=") if not typed or not username: raise ValueError(f"authors: --map expects Typed=username, got {pair!r}") account = db.scalar(select(Account).where(Account.username == username.strip().lower())) if account is None: raise ValueError(f"authors: no account with username {username.strip().lower()!r}") folded[typed.strip().lower()] = account rename = bool(getattr(args, "rename", False)) columns = AUTHOR_COLUMNS + (DEFAULTED_AUTHOR_COLUMNS if getattr(args, "include_defaulted", False) else ()) report = [] for tname, text_col, id_col in columns: t = sa_table(tname, sa_column(text_col), sa_column(id_col)) n = renamed = 0 for key, account in folded.items(): result = db.execute( update(t).where(t.c[id_col].is_(None), func.lower(func.trim(t.c[text_col])) == key) .values({id_col: account.id}) ) n += result.rowcount or 0 if rename and account.display_name and key != account.display_name.strip().lower(): result = db.execute( update(t).where(t.c[id_col] == account.id, func.lower(func.trim(t.c[text_col])) == key) .values({text_col: account.display_name}) ) renamed += result.rowcount or 0 report.append(f"{tname}={n}" + (f" renamed={renamed}" if rename else "")) db.commit() return "authors: linked " + " ".join(report) def backfill_overrides(db) -> str: """Kept for the after-deploy chains that name it; reports the ledger and writes nothing.""" from app.models import Decision from app.services.decisions.effective import table_present n = db.scalar(select(func.count()).select_from(Decision)) if table_present(db) else 0 return f"overrides: 0 changed ({n or 0} row(s) present)" def backfill_vertical(db) -> str: """product_variants.vertical from our own category (taxonomy family); the default otherwise.""" changed = 0 for product in db.scalars(select(ProductVariant)): wanted = vertical_of(product.category) or product.vertical or DEFAULT_VERTICAL if product.vertical != wanted: product.vertical = wanted changed += 1 db.commit() return f"vertical: {changed} product(s) updated" def backfill_stuck_runs(db) -> str: """Runs still 'running' a day after they started belong to dead processes.""" from datetime import UTC, datetime now = datetime.now(UTC) changed = 0 for run in db.scalars(select(CollectionRun).where(CollectionRun.status == "running")): if is_stuck(run, now): mark_stuck(run, "process ended without finishing (marked by backfill)") changed += 1 db.commit() return f"stuck_runs: {changed} run(s) marked error" #: The message the fetch port wrote when it read a page that is not there as a refusal. Only #: these two: 401, 403, 429, a challenge page and a robots refusal are real refusals and this #: backfill must never touch their rows. FALSE_REFUSAL_MARKERS = ("refused with HTTP 404", "refused with HTTP 410") def backfill_false_refusals(db) -> str: """Runs recorded as `blocked` because a page was gone are our error, not a refusal. An empty-bodied HTTP 404 satisfied the block test, so a run that met one delisted product ended `blocked` with " refused with HTTP 404". Every reader of that word treats the whole source as closed to us: the live page says `refused` with Start disabled, the Start route answers `SOURCE_REFUSED` (final until rian records a decision), the sweep plan skips the source and verification will not check it. This moves those rows to `error`, keeping the text the run recorded and appending why it was reclassified. Safe to repeat: a second run finds no `blocked` row matching and reports zero. """ from datetime import UTC, datetime note = ( " [reclassified from blocked to error by `backfill false_refusals` on " f"{datetime.now(UTC).date().isoformat()}: HTTP 404 and 410 mean the page is gone, " "not the host refusing us]" ) changed: list[int] = [] for run in db.scalars( select(CollectionRun).where(CollectionRun.status == "blocked").order_by(CollectionRun.id) ): recorded = run.error or "" if not any(marker in recorded for marker in FALSE_REFUSAL_MARKERS): continue run.status = "error" run.error = f"{recorded}{note}" changed.append(run.id) db.commit() ids = ", ".join(str(run_id) for run_id in changed) if changed else "none" return f"false_refusals: {len(changed)} run(s) moved from blocked to error ({ids})" def backfill_shops(db) -> str: """Existing shop rows take their currency from the collector that owns them.""" changed = 0 by_key = {} for collector in COLLECTORS.values(): for spec in collector.shops(): by_key[(collector.retailer_slug, spec.code)] = spec.currency for shop in db.scalars(select(Shop).join(Retailer)): declared = by_key.get((shop.retailer.slug, shop.code)) if declared and sync_shop_currency(shop, declared): changed += 1 db.commit() return f"shops: {changed} currency value(s) synced from collectors" def orphan_tile_ids(db) -> list[int]: """Listings for a multi-size parent tile that never got its per-size rows. Such a row carries the FAMILY'S cheapest price against no particular size, which is exactly the misread the per-size rows were introduced to stop. """ parents = db.scalars(select(Listing).where(Listing.source_sku.like("%-P"))).all() children = set( db.execute( select(Listing.shop_id, Listing.source_sku).where(Listing.source_sku.like("%::%")) ).all() ) child_parents = {(loc, sku.split("::", 1)[0]) for loc, sku in children} return [l.id for l in parents if (l.shop_id, l.source_sku) not in child_parents] def backfill_orphan_tiles(db) -> str: """Purge parent-tile listings with no per-size sibling, and product variants left empty. Purged, not re-resolved: re-resolving needs the product page, which is a crawl. The product page is read on the next collection of that store and the per-size rows are created then. """ ids = orphan_tile_ids(db) if not ids: return "orphan_tiles: nothing to purge" variant_ids = set(db.scalars(select(Listing.variant_id).where(Listing.id.in_(ids)))) db.execute(RawRecord.__table__.delete().where(RawRecord.listing_id.in_(ids))) db.execute(PriceObservation.__table__.delete().where(PriceObservation.listing_id.in_(ids))) db.execute(Listing.__table__.delete().where(Listing.id.in_(ids))) db.flush() still_listed = set(db.scalars(select(Listing.variant_id).where(Listing.variant_id.in_(variant_ids)))) awarded = set(db.scalars(select(Award.variant_id).where(Award.variant_id.in_(variant_ids)))) empty = variant_ids - still_listed - awarded if empty: db.execute(ProductVariant.__table__.delete().where(ProductVariant.id.in_(empty))) db.commit() return f"orphan_tiles: {len(ids)} listing(s) purged, {len(empty)} empty product(s) removed" def backfill_image_provenance(db) -> str: """Split the old single attribution into barcode-keyed and name-matched. Approximate for rows found before the split: a product with a barcode is assumed found by it. Rows enriched after the split carry the exact value. """ changed = 0 for product in db.scalars(select(ProductVariant).where(ProductVariant.image_source == "Open Food Facts")): product.image_source = "Open Food Facts (barcode)" if product.gtin else "Open Food Facts (name)" changed += 1 db.commit() return f"image_provenance: {changed} product(s) relabelled" def backfill_image_sources(db, args=None) -> str: """Move every product variant's picture provenance onto the controlled vocabulary (`services/imagery.py`, Stream AW3.2, after migration `aw3b1c2d3e4f`): "Open Food Facts (barcode)" becomes `public:openfoodfacts:barcode`, "(name)" `public:openfoodfacts:name`, a bare "Open Food Facts" one or the other by whether the row holds a barcode; a row with a picture and no `image_level` is stamped `variant`, what an Open Food Facts photo depicts. A string outside the vocabulary is left and counted, never guessed. Idempotent: a second run reports zero. Reads and writes product variants only; brands and product lines have no picture until `images import`.""" from collections import Counter counts: Counter[str] = Counter() for variant in db.scalars(select(ProductVariant).where(ProductVariant.image_url.isnot(None))): new = imagery.normalise_source(variant.image_source, has_gtin=bool(variant.gtin)) if new != variant.image_source: variant.image_source = new counts["moved"] += 1 elif not imagery.valid_source(new): counts["outside_vocabulary"] += 1 if variant.image_level is None: variant.image_level = "variant" counts["level_stamped"] += 1 db.commit() return (f"image_sources: {counts['moved']} source(s) moved to the vocabulary, " f"{counts['level_stamped']} level(s) stamped variant, {counts['outside_vocabulary']} left outside it") def backfill_implausible_sizes(db) -> str: """Null a size no single item comes in when the product's own name does not state it. The first audit found two Extime mists at 7,624 and 5,924 ml (the page's own capacity field; the collection-time cross-check now refuses it) and two whiskies at 7,000 ml. The size goes, the match key follows it, and the next collection re-reads the row through the corrected parser. """ changed = 0 maps = keying.load_maps(db) for product in db.scalars(select(ProductVariant).where(ProductVariant.quantity_ml.isnot(None))): if size_is_implausible(product.vertical, product.name, product.quantity_ml): logger.info("size_nulled product=%s name=%r was=%s", product.id, product.name, product.quantity_ml) product.quantity_ml = None product.match_key, _ = keying.product_key(product, maps) changed += 1 db.commit() return f"implausible_sizes: {changed} product(s) set to unknown size" def display_spelling(counter) -> str: """The spelling to show for a brand: mixed case over SHOUTING, then the most common. One storefront capitalises every brand ("SISLEY", "DIOR") and lists more product variants than the rest, so the raw majority would shout on every brand page. Trademark glyphs are not part of a name. """ def rank(item): spelling, count = item return (spelling.isupper(), -count, spelling) best = sorted(counter.items(), key=rank)[0][0] return best.replace("\u00ae", "").replace("\u2122", "").strip() or best def backfill_brands(db) -> str: """One brands row per fold key, product variants pointed at it (migration #3). A row is created with the spelling most product variants carry; an existing row's name is never changed here (a human may have set it). Products keep the spelling they were collected with in `brand`; `brand_id` is the fold. """ from collections import Counter, defaultdict spellings: dict[str, Counter] = defaultdict(Counter) for brand, n in db.execute( select(ProductVariant.brand, func.count(ProductVariant.id)).where(ProductVariant.brand.isnot(None)).group_by(ProductVariant.brand) ): slug = brand_slug(brand) if slug: spellings[slug][brand.strip()] += n existing = {b.slug: b for b in db.scalars(select(Brand))} created = 0 for slug, counter in spellings.items(): if slug not in existing: row = Brand(slug=slug, name=display_spelling(counter)) db.add(row) existing[slug] = row created += 1 db.flush() pointed = 0 for product in db.scalars(select(ProductVariant).where(ProductVariant.brand.isnot(None), ProductVariant.brand_id.is_(None))): row = existing.get(brand_slug(product.brand)) if row is not None: product.brand_id = row.id pointed += 1 db.commit() return f"brands: {created} brand row(s) created, {pointed} product(s) pointed at their brand" def backfill_sizes(db) -> str: """quantity_stated_value/quantity_stated_unit as the name states it, for rows that have a quantity_ml and no stated size.""" changed = 0 for product in db.scalars( select(ProductVariant).where(ProductVariant.quantity_ml.isnot(None), ProductVariant.quantity_stated_value.is_(None)) ): stated = parse_size(product.name) if stated and quantity_ml_of(*stated) == product.quantity_ml: product.quantity_stated_value, product.quantity_stated_unit = stated else: product.quantity_stated_value, product.quantity_stated_unit = float(product.quantity_ml), "ml" changed += 1 db.commit() return f"sizes: {changed} product(s) given a stated size" def backfill_categories(db) -> str: """Classify the uncategorised again under the widened rules and the brand map. Only rows with no category are touched (a category, once set, is never overwritten here); the vertical follows the new category. Rows nothing matches stay uncategorised: no category reads better than a wrong one. """ changed = 0 by_vertical: dict[str, int] = {} for product in db.scalars(select(ProductVariant).where(ProductVariant.category.is_(None))): category = classify(product.name, product.brand) if category is None: continue product.category = category wanted = vertical_of(category) or product.vertical if product.vertical != wanted: product.vertical = wanted by_vertical[category] = by_vertical.get(category, 0) + 1 changed += 1 db.commit() detail = ", ".join(f"{k} {v}" for k, v in sorted(by_vertical.items(), key=lambda kv: -kv[1])) return f"categories: {changed} product(s) categorised ({detail or 'none'})" def backfill_merges(db, args=None) -> str: """The duplicate groups as recorded merges; the unsettled ones as candidates. Run after `rederive`, so the groups are read under the current identity rules. Safe to repeat: a merged row is skipped, a candidate pair is recorded once. With `--check` (Stream L) the group list is printed first, one line per group (key, member ids, names, barcodes; a candidate group with its conflicts), so the groups can be read before a real run. """ from app.services.merges import duplicate_groups if args is not None and getattr(args, "check", False): groups = duplicate_groups(db) for group in sorted(groups, key=lambda g: (g.mergeable, g.match_key)): kind = "MERGE" if group.mergeable else "CANDIDATE " + ",".join(group.conflicts) members = "; ".join(f"{p.id} {p.name!r}" + (f" gtin={p.gtin}" if p.gtin else "") for p in group.members) print(f"{kind} {group.match_key} {members}") print(f"{sum(1 for g in groups if g.mergeable)} mergeable group(s), {sum(1 for g in groups if not g.mergeable)} candidate group(s)") counts = merge_duplicates(db) db.commit() return ( f"merges: {counts['groups']} group(s) merged ({counts['merged_rows']} row(s) now forward), " f"{counts['candidate_groups']} group(s) left for a human ({counts['candidates_added']} new candidate pair(s))" ) def cmd_rederive(args: argparse.Namespace) -> int: """Recompute every product's match key and attributes under the current identity rules. A rules change must reach the rows resolved before it, or two spellings of one brand (v2) or of one attribute (v3) stay two product variants forever. Rows are re-keyed in place through the same call ingest uses (`keying.product_key`: the brand with its alias followed, the canonical attribute from the alias table, then the v3 key) and stamped with the rules version; `backfill merges` then folds whatever now shares a key. Idempotent: a second run changes nothing. """ if _collection_running(getattr(args, "force", False)): return 2 rekeyed = stamped = restamped = relined = 0 with SessionLocal() as db: maps = keying.load_maps(db) # Every row, tombstones included (Stream L): a human merge's tombstone must carry # the current rules' key for ingest to follow it; `merged_into_id` is never touched. # The decided layer is read first inside `product_key` (overrides on the name's # collected spelling, the line, the attribute, the quantity). for product in db.scalars(select(ProductVariant).order_by(ProductVariant.id)): # The line is re-derived here too (identity rules v5), through the same call # ingest uses, so the after-deploy order (`backfill lines` before `rederive`) is a # convenience for the display names and never load-bearing: a product keyed under # v5 always hangs from the v5 line, tombstones included, and a `product_line_id` a person # decided is never moved. if "product_line_id" not in maps.decided.get(product.id, {}) and product.brand_id is not None: brand = maps.brand_of(product.brand_id, product.brand, product.vertical) line = keying.product_line_for(db, maps, brand, keying.keyed_name(product, maps), product.brand, product.vertical, create=True, category=product.category) if line is not None and product.product_line_id != line.id: product.product_line_id = line.id relined += 1 key, attributes = keying.product_key(product, maps) if product.match_key != key: logging.info("rekeyed product=%s %r -> %r", product.id, product.match_key, key) product.match_key = key rekeyed += 1 if product.vertical == "beauty" and not (product.attributes or {}).get("concentration"): attrs = raw_attributes(RawListing( source_sku="", name=product.name, price=1.0, currency="EUR", shop_code="", brand=product.brand, vertical=product.vertical, )) attributes = {**attributes, **{k: v for k, v in attrs.items() if k not in attributes}} if attributes != (product.attributes or {}): product.attributes = attributes restamped += 1 if product.identity_rules_version != IDENTITY_RULES_VERSION: product.identity_rules_version = IDENTITY_RULES_VERSION stamped += 1 db.commit() print( f"rederive: {rekeyed} match key(s) changed, {relined} product(s) moved to their line, " f"{restamped} attribute set(s) changed, {stamped} product(s) stamped rules v{IDENTITY_RULES_VERSION}" ) return 0 def _collection_running(force: bool = False) -> bool: """Whether a collection is running, in which case a data move refuses (the constitution: every shared row tolerates a race; a rekey or a prune under a live collector does not). `--force` overrides, for a run known to be stuck.""" from app.models import CollectionRun with SessionLocal() as db: running = db.scalar(select(func.count(CollectionRun.id)).where(CollectionRun.status == "running")) or 0 if running and not force: print(f"refused: {running} collection run(s) are running; wait, or --force if the run is stuck " "(`backfill stuck_runs` closes one)") return True return False # name -> function(db) -> summary line. Each must be safe to run twice. def backfill_perfume(db) -> str: """Rename the stored category Fragrance to Perfume (structure review, 9 Sep: "Search is always our driver"; the searched word has 19x the volume). Idempotent: a row already reading Perfume is untouched, and a second run reports zero. The vertical (beauty) does not change; `backfill categories` cannot do this because it touches only uncategorised rows. """ changed = ( db.execute( update(ProductVariant).where(ProductVariant.category == "Fragrance").values(category="Perfume") ).rowcount or 0 ) db.commit() return f"perfume: {changed} product(s) renamed from Fragrance to Perfume" def _backfill_threads(db, args=None) -> str: """Migration #6: a thread per legacy comment key, then the legacy mentions rung (app/services/discussion.py). The mention step resolves names through the accounts kit, which the API process wires at startup and this process must wire itself: the first rehearsal on a copy of staging raised NOT_INITIALIZED here.""" from app.services import accounts from app.services.discussion import backfill_threads accounts.init() return backfill_threads(db, args) def _backfill_thread_reads_seed(db, args=None) -> str: """The one-time read seed: everything before today is read for everyone (app/services/discussion.py, the workflow plan §3).""" from app.services.discussion import backfill_thread_reads_seed return backfill_thread_reads_seed(db, args) def _backfill_legacy_mentions(db, args=None) -> str: """T14: a mention notification for every legacy comment naming a current account that never rang, under the write path's own dedupe key (app/services/discussion.py). Wires the accounts kit first, as `_backfill_threads` does.""" from app.services import accounts from app.services.discussion import backfill_legacy_mentions accounts.init() return backfill_legacy_mentions(db, args) def backfill_lines(db) -> str: """One product_lines row per brand and line key (`services/product_lines.py`), product variants pointed at it. The standard brand is the product's brand row with its alias followed; a product with no brand gets no line. A row is created with the spelling the most common product name gives the key's words; an existing row's name is never changed here (a person may have set it), and a product joins the canonical line when its row is an alias. A merged row (tombstone) is pointed too, so `rederive` keys it under the current line and ingest can still follow a confirmed merge (identity rules v5 moved every Makeup shade's line); it never names a line. Safe to repeat: a second run creates nothing and points nothing. """ from collections import Counter, defaultdict brands = {b.id: b for b in db.scalars(select(Brand))} rows = list(db.scalars(select(ProductLine))) by_key = {(row.brand_id, row.key): row for row in rows} by_id = {row.id: row for row in rows} slugs = {row.slug for row in rows} keyed = [] samples: dict[tuple[int, str], Counter] = defaultdict(Counter) spellings: dict[tuple[int, str], set[str]] = defaultdict(set) categories: dict[tuple[int, str], str | None] = {} from app.services import overrides as overrides_service decided = overrides_service.read(db, "product", fields=("name", "product_line_id")) for product in db.scalars(select(ProductVariant).where(ProductVariant.brand_id.isnot(None)).order_by(ProductVariant.id)): brand = product_lines.resolve_alias(brands, product.brand_id) if brand is None: continue rows = decided.get(str(product.id), {}) if "product_line_id" in rows: continue # a person chose the line; the rules never point it elsewhere # A renamed product keys on the spelling its listings arrive under (LT3 b). name = str(rows["name"].collected_value) if "name" in rows and rows["name"].collected_value else product.name key = product_lines.product_line_key(name, listed_brand=product.brand, brand=brand.name, vertical=product.vertical, category=product.category) keyed.append((product, brand, key)) if product.merged_into_id is not None: continue # a tombstone is pointed at its line but never names one samples[(brand.id, key)][name] += 1 spellings[(brand.id, key)].update({product.brand or "", brand.name}) categories.setdefault((brand.id, key), product.category) created = pointed = 0 moved_from: dict[int, set[int]] = defaultdict(set) # old line id -> the rows its product variants went to for product, brand, key in keyed: row = by_key.get((brand.id, key)) if row is None: base = product_lines.line_slug(brand.slug, key)[:230] slug, n = base, 2 while slug in slugs: slug, n = f"{base}-{n}", n + 1 row = ProductLine( brand_id=brand.id, key=key, slug=slug, name=product_lines.display_line_name( key, samples[(brand.id, key)] or Counter({product.name: 1}), brand.name, brand_words=product_lines.brand_words_of(*spellings[(brand.id, key)]), vertical=product.vertical, category=categories.get((brand.id, key)), )[:200], ) db.add(row) db.flush() by_key[(brand.id, key)] = row by_id[row.id] = row slugs.add(slug) created += 1 target = product_lines.resolve_alias(by_id, row.id) or row if product.product_line_id != target.id: if product.product_line_id is not None and product.merged_into_id is None: moved_from[product.product_line_id].add(target.id) product.product_line_id = target.id pointed += 1 db.flush() # A person's decisions on a line (a preferred name, a review, an alias) follow its product variants # when a rule change empties the row and every live product landed on ONE new row (the # catalogue decisions §2.2: a decision is about the line, not the row id the rules minted); # where the product variants scattered the decisions stay put and are reported, for a person. decisions_moved = decisions_stranded = 0 if moved_from and overrides_service.table_present(db): decided_lines = overrides_service.read(db, "line") still_live = {lid for lid, in db.execute( select(ProductVariant.product_line_id).where(ProductVariant.merged_into_id.is_(None), ProductVariant.product_line_id.in_(list(moved_from))).distinct() )} for old_id, targets in moved_from.items(): if str(old_id) not in decided_lines or old_id in still_live: continue if len(targets) == 1: moved = overrides_service.move(db, "line", str(old_id), str(next(iter(targets)))) decisions_moved += len(moved["moved"]) decisions_stranded += len(moved["kept"]) else: decisions_stranded += len(decided_lines[str(old_id)]) logging.warning("line %s carried a decision and its product variants scattered to %s; the decision stays", old_id, sorted(targets)) db.commit() return (f"lines: {created} line(s) created, {pointed} product(s) pointed at their line" + (f", {decisions_moved} line decision(s) moved with their product variants" if decisions_moved else "") + (f", {decisions_stranded} line decision(s) left on an emptied row for a person" if decisions_stranded else "")) def backfill_prune_lines(db) -> str: """Derived line rows nothing references are pruned (the catalogue decisions, §2.3). A line row is a pure function of the collected name, so a rule change (identity rules v5 took the shade out of the Makeup line) leaves the rows the old rule made with nothing on them. Such a row is a cache, not a record: it is deleted here and nowhere else in a backfill, and only when NOTHING points at it. What counts as a reference is `merges.referenced_lines`, shared with the brand split, which empties rows the same way. Safe to repeat: a second run deletes zero. """ from app.services import merges pruned = merges.prune_lines(db) db.commit() return f"prune_lines: {pruned} empty line row(s) nothing referenced deleted" def backfill_quantities(db, args=None) -> str: """Every live product's standard quantity (identity rules v4, `services/quantity.py`): the name parsed with the stored v3 pair as the hint the text outranks (the pair is our past parse; 7293 held 75.00 ml for a 75 g stick), the seven quantity columns and `name_key` written, `quantity_ml` filled where the dimension is ml and it was NULL, and `quantity_ml` (with the derived pair) CLEARED only where the parsed dimension is g or pcs and the stored `quantity_ml` equals the same figure read as ml, the misread's signature; never otherwise. A product with a decided `quantity` override is never touched. Run before `rederive`, which keys on these columns. Safe to repeat: a second run writes zero. """ from app.services import overrides as overrides_service from app.services.ingest import write_quantity from app.services.quantity import parse_quantity decided = overrides_service.read(db, "product", fields=("quantity", "name")) written = cleared = filled = skipped = 0 misreads: list[str] = [] def snapshot(p): return (None if p.quantity_value is None else float(p.quantity_value), p.quantity_unit, p.pack_count, None if p.pack_unit_value is None else float(p.pack_unit_value), p.form, p.set_contents, p.quantity_state, p.name_key, p.quantity_ml, None if p.quantity_stated_value is None else float(p.quantity_stated_value), p.quantity_stated_unit) for product in db.scalars(select(ProductVariant).where(ProductVariant.merged_into_id.is_(None)).order_by(ProductVariant.id)): rows = decided.get(str(product.id), {}) if "quantity" in rows: skipped += 1 continue name = str(rows["name"].collected_value) if "name" in rows and rows["name"].collected_value else product.name if product.quantity_stated_value is not None and product.quantity_stated_unit: hint = (float(product.quantity_stated_value), product.quantity_stated_unit) elif product.quantity_ml: hint = (product.quantity_ml, "ml") else: hint = None before = snapshot(product) q = parse_quantity(name, hint=hint, category=product.category) write_quantity(product, q, name=name) if q.state == "stated" and q.unit == "ml" and product.quantity_ml is None and q.ml: product.quantity_ml = q.ml filled += 1 elif (q.state == "stated" and q.unit in ("g", "pcs") and q.value is not None and product.quantity_ml is not None and product.quantity_ml == round(float(q.value))): misreads.append(f"{product.id} {product.name!r} {product.quantity_ml} ml -> {q.value:g} {q.unit}") product.quantity_ml = product.quantity_stated_value = product.quantity_stated_unit = None cleared += 1 if snapshot(product) != before: written += 1 db.commit() for line in misreads[:40]: logging.info("quantity_ml cleared: %s", line) return (f"quantities: {written} product(s) written, {filled} quantity_ml filled from a millilitre name, " f"{cleared} quantity_ml cleared (a gram or piece figure once read as ml), {skipped} decided and untouched") def backfill_listed(db, args=None) -> str: """Every listing's LISTED columns from its newest fragment (`collected.listed_fields`, the same reader ingest uses), `listed_record_id` set to that fragment; every column left NULL where no fragment exists (NULL is the honest value: a copy of our own text would be labelled as the shop's); `last_seen_at` filled where NULL from the newest of the listing's price observations and raw records. The newest fragment is `max(id)` per listing through a portable subquery, read slim (`payload - 'tile_html'` on Postgres). Safe to repeat: a second run writes zero. """ from app.models import Listing, PriceObservation, RawRecord from app.services.collected import write_listed newest = ( select(RawRecord.listing_id.label("listing_id"), func.max(RawRecord.id).label("record_id")) .group_by(RawRecord.listing_id).subquery() ) slim = RawRecord.payload.op("-")("tile_html") if db.get_bind().dialect.name == "postgresql" else RawRecord.payload written = 0 for listing, record_id, parser_version, payload in db.execute( select(Listing, RawRecord.id, RawRecord.parser_version, slim) .join(newest, newest.c.listing_id == Listing.id) .join(RawRecord, RawRecord.id == newest.c.record_id) .order_by(Listing.id) ): if write_listed(listing, record_id, parser_version, payload): written += 1 db.flush() seen = ( select(Listing.id.label("listing_id"), func.max(PriceObservation.observed_at).label("seen")) .join(PriceObservation, PriceObservation.listing_id == Listing.id) .where(Listing.last_seen_at.is_(None)).group_by(Listing.id).subquery() ) seen_raw = ( select(RawRecord.listing_id.label("listing_id"), func.max(RawRecord.created_at).label("seen")) .group_by(RawRecord.listing_id).subquery() ) stamped = 0 latest: dict[int, object] = {} for listing_id, at in db.execute(select(seen.c.listing_id, seen.c.seen)): latest[listing_id] = at for listing_id, at in db.execute(select(seen_raw.c.listing_id, seen_raw.c.seen)): if at is not None and (listing_id not in latest or latest[listing_id] is None or at > latest[listing_id]): latest[listing_id] = at if latest: for listing in db.scalars(select(Listing).where(Listing.last_seen_at.is_(None), Listing.id.in_(list(latest)))): listing.last_seen_at = latest[listing.id] stamped += 1 db.commit() without = int(db.scalar(select(func.count(Listing.id)).where(Listing.listed_record_id.is_(None))) or 0) return (f"listed: {written} listing(s) written from their newest fragment, {stamped} last_seen_at filled, " f"{without} listing(s) with no fragment (listed columns NULL)") def backfill_options(db, args=None) -> str: """The options a shop published as fields, onto the variants created while the Shopify collector still glued them into the name (identity rules v6, Stream K3). For every variant, live or a merge's forwarding row, whose name is exactly a stored fragment's product title plus its variant title (how the collector built it until 2026-09-17): the name becomes the product title, and each option that is not the quantity is written as `option:` (`collectors.base.option_attributes`, the reader ingest uses). A forwarding row is reached through the listing ids its merge recorded. Without this, an old variant would key on the " / " tail fallback and its own listing's next sighting on the option field, and every Attenza variant would be minted a second time. A variant whose name a person decided, or whose name is not the glued form (another shop's listing made it), is left alone; with `--check` nothing is written. Safe to repeat: a second run writes zero. Run before `rederive`.""" import html as html_module from app.models import Listing, Merge, RawRecord from app.services.collected import _platform from app.services.collectors.base import option_attributes from app.services.collectors.shopify import published_options from app.services.normalize import flat_key check = bool(args is not None and getattr(args, "check", False)) maps = keying.load_maps(db) newest = ( select(RawRecord.listing_id.label("listing_id"), func.max(RawRecord.id).label("record_id")) .group_by(RawRecord.listing_id).subquery() ) # listing id -> the row that listing made: the forwarding row a merge moved it from, else its variant. made_by: dict[int, int] = {} for merge in db.scalars(select(Merge).order_by(Merge.id)): for listing_id in (merge.detail or {}).get("listings") or []: made_by.setdefault(int(listing_id), merge.from_id) written = renamed = decided = 0 seen: set[int] = set() for listing_id, variant_id, parser_version, payload in db.execute( select(Listing.id, Listing.variant_id, RawRecord.parser_version, RawRecord.payload) .join(newest, newest.c.listing_id == Listing.id) .join(RawRecord, RawRecord.id == newest.c.record_id) .where(RawRecord.parser_version.like("shopify%")) .order_by(Listing.id) ): if _platform(parser_version) != "shopify" or not isinstance(payload, dict): continue product, variant_json = payload.get("product") or {}, payload.get("variant") or {} title = html_module.unescape(str(product.get("title") or "")).strip() variant_title = str(variant_json.get("title") or "").strip() if not title or not variant_title or variant_title.lower() == "default title": continue glued = f"{title} {html_module.unescape(variant_title)}" for row_id in {made_by.get(listing_id, variant_id), variant_id}: row = db.get(ProductVariant, row_id) if row is None or row.id in seen or row.name != glued: continue seen.add(row.id) if "name" in maps.decided.get(row.id, {}): decided += 1 continue options = option_attributes(published_options(product, variant_json)) if not check: row.name = title row.name_key = flat_key(title)[:400] missing = {k: v for k, v in options.items() if k not in (row.attributes or {})} if missing: row.attributes = {**(row.attributes or {}), **missing} renamed += 1 written += 1 if options else 0 if not check: db.commit() return (f"options: {renamed} variant(s) {'would be ' if check else ''}named by their product title, {written} given the shop's " f"option field(s), {decided} left alone for a person's name decision") def backfill_rule_proposals(db, args=None) -> str: """What every word list WOULD have grouped or read, written as proposals (Stream K3.3): one pass per list (`rule::`), through the store the AI pass uses; nothing is applied. Run after `rederive` and `backfill merges`. `--check` counts without writing. Safe to repeat: rows upsert on (pass, natural key, field), so a second run inserts none. A list that proves bad is withdrawn alone: `proposals withdraw --pass rule::6`.""" from app.services import proposal_rules check = bool(args is not None and getattr(args, "check", False)) counts = proposal_rules.generate(db, check=check) if not check: db.commit() detail = ", ".join(f"{rule} {n}" for rule, n in counts.items()) return f"rule_proposals: {sum(counts.values())} proposal(s) {'would be ' if check else ''}written, none applied ({detail})" def backfill_attributes(db) -> str: """The attribute vocabulary seeded from what the names say, and every product of a vertical with attribute_values stamped with its canonical one. The wording found in a name (`lines.attribute_of`, "elixir parfum intense") becomes a `attribute_aliases` row if absent, canonical by the rule ("elixir"); a row a person decided keeps its canonical and display, and that wins for the product. `attributes.attribute` is set (the dict reassigned, never mutated in place); the parsed concentration stays as the raw record. Safe to repeat: a second run seeds nothing and stamps nothing. """ aliases = {(a.vertical, a.raw): a for a in db.scalars(select(AttributeAlias))} seeded = stamped = kinds = 0 for product in db.scalars( select(ProductVariant).where( ProductVariant.merged_into_id.is_(None), ProductVariant.vertical.in_(sorted(product_lines.ATTRIBUTE_VERTICALS)), ) ): # The category reaches the rule (Stream L): a Makeup shade is a color, a Skincare # " / Grasa" tail is a skin type and gets none; the flavor wordings come from # confectionery names. raw, canonical = product_lines.attribute_of(product.name, product.vertical, category=product.category) kind = product_lines.attribute_kind_of(product.name, product.vertical, category=product.category) if raw else "" if raw: alias = aliases.get((product.vertical, raw)) if alias is None: alias = AttributeAlias( vertical=product.vertical, raw=raw, canonical=canonical, display=product_lines.display_attribute(canonical), kind=kind or None, ) db.add(alias) aliases[(product.vertical, raw)] = alias seeded += 1 elif not alias.kind and kind: alias.kind = kind kinds += 1 canonical = alias.canonical kind = alias.kind or kind attrs = dict(product.attributes or {}) wanted = {k: v for k, v in attrs.items() if k not in ("attribute", "attribute_kind")} if canonical: wanted["attribute"] = canonical if kind: wanted["attribute_kind"] = kind if wanted != attrs: product.attributes = wanted stamped += 1 db.commit() return (f"attribute_values: {seeded} wording(s) seeded, {stamped} product(s) stamped with their attribute" + (f", {kinds} kind(s) recorded on existing wordings" if kinds else "")) def backfill_attribute_keys(db) -> str: """The two JSON keys on `product_variants.attributes` in the vocabulary's word (Stream K1, rename map row 8): `variation` becomes `attribute`, `variation_kind` becomes `attribute_kind`, value for value. Data, so it moves here and never in the migration; idempotent, the second run reports zero. Storage by kind (`attributes.concentration`) is the attribute registry's valve (K2): on the 16 Sep copy 1,745 variants carry both the canonical value and the parsed base concentration and 294 differ, so folding them here would change what the merge veto compares.""" from sqlalchemy import select from app.models import ProductVariant moved = 0 for row in db.scalars(select(ProductVariant).where(ProductVariant.attributes.isnot(None))): attrs = dict(row.attributes or {}) if "variation" not in attrs and "variation_kind" not in attrs: continue out = {k: v for k, v in attrs.items() if k not in ("variation", "variation_kind")} if attrs.get("variation"): out["attribute"] = attrs["variation"] if attrs.get("variation_kind"): out["attribute_kind"] = attrs["variation_kind"] row.attributes = out moved += 1 db.commit() return f"attribute keys: {moved} variant(s) moved from variation/variation_kind to attribute/attribute_kind" def backfill_places(db) -> str: """One airport place per IATA code from `shops.iata` (Stream K2; plan W19): the place's slug and name are the airport page's (its first shop's), its identifiers carry `(iata, code)`, and every shop of that airport gets its `shop_places` row with role `primary`. A catalogue-only shop (no IATA) is the online kind and needs no primary place. Idempotent: a second run reports zero.""" from app.models.places import Place, ShopPlace from app.services.urls import airport_path created = pointed = 0 by_iata: dict[str, list[Shop]] = {} for shop in db.scalars(select(Shop).where(Shop.iata.isnot(None)).order_by(Shop.id)): by_iata.setdefault(shop.iata, []).append(shop) primary = {s: p for s, p in db.execute(select(ShopPlace.shop_id, ShopPlace.place_id).where(ShopPlace.role == "primary"))} airports: dict[str, Place] = {} for row in db.scalars(select(Place).where(Place.kind == "airport")): for ident in row.identifiers or []: if ident.get("scheme") == "iata": airports[ident["value"]] = row for iata, shops in sorted(by_iata.items()): first = shops[0] place = airports.get(iata) if place is None: slug = airport_path(iata, first.city, first.name).rstrip("/").rsplit("/", 1)[-1] place = Place(slug=slug, kind="airport", name=first.name, identifiers=[{"scheme": "iata", "value": iata}], attributes={"city": first.city, "country": first.country}) db.add(place) db.flush() airports[iata] = place created += 1 for shop in shops: if shop.id not in primary: db.add(ShopPlace(shop_id=shop.id, place_id=place.id, role="primary")) primary[shop.id] = place.id pointed += 1 db.commit() return f"places: {created} airport place(s) created, {pointed} shop(s) given their primary place" def backfill_suggestion_decisions(db) -> str: """The two suggestion decision values in rian's words (Stream K1, rename map row 6): `merged` becomes `same`, `kept_apart` becomes `separate`. Data, so it moves here and never in the migration; idempotent, the second run reports zero.""" from sqlalchemy import update from app.models import Suggestion same = db.execute(update(Suggestion).where(Suggestion.decision == "merged").values(decision="same")).rowcount separate = db.execute(update(Suggestion).where(Suggestion.decision == "kept_apart").values(decision="separate")).rowcount db.commit() return f"suggestion decisions: {same} merged -> same, {separate} kept_apart -> separate" #: The backfills that rekey, re-point or delete catalogue rows a running collector also #: writes; they refuse while a collection runs (`_collection_running`). DATA_MOVES = frozenset({"lines", "prune_lines", "attributes", "quantities", "merges", "brands", "sizes", "categories"}) BACKFILLS = { "accounts": backfill_accounts, "levels": backfill_levels, "level_permissions": backfill_level_permissions, # a permission that joined a level's seed later "authors": backfill_authors, "overrides": backfill_overrides, "vertical": backfill_vertical, "stuck_runs": backfill_stuck_runs, "false_refusals": backfill_false_refusals, # a run blocked by a 404 or 410 becomes `error` "shops": backfill_shops, "orphan_tiles": backfill_orphan_tiles, "image_provenance": backfill_image_provenance, "image_sources": backfill_image_sources, # Stream AW3: after migration aw3b1c2d3e4f, before `images import` "implausible_sizes": backfill_implausible_sizes, "brands": backfill_brands, "sizes": backfill_sizes, "categories": backfill_categories, "perfume": backfill_perfume, "merges": backfill_merges, "threads": _backfill_threads, "legacy_mentions": _backfill_legacy_mentions, "thread_reads_seed": _backfill_thread_reads_seed, "options": backfill_options, # Stream K3: before `lines` and `rederive` "rule_proposals": backfill_rule_proposals, # Stream K3: after `rederive` and `merges`; applies nothing "lines": backfill_lines, "prune_lines": backfill_prune_lines, # rules v5: after rederive and merges "attributes": backfill_attributes, "suggestion_decisions": backfill_suggestion_decisions, "attribute_keys": backfill_attribute_keys, # K1: the JSON keys in the vocabulary's word "places": backfill_places, # K2: one airport place per IATA code, every shop's primary place "quantities": backfill_quantities, # Stream L: before rederive "listed": backfill_listed, # Stream L: after merges } def cmd_propose(args: argparse.Namespace) -> int: """File a proposals file as merge candidates with reason `proposed` (Stream L): the session's own reading of the data, for a person to approve on the desk; the rules never withdraw or rescore them; nothing is merged. `--check` reports the counts and writes nothing. The file: `{"proposed_by": "...", "entries": [{level, left_id, right_id, score, why}, ...]}`; `import/` is mounted read-only at `/srv/import` in the container.""" from app.services import merge_desk entries, proposed_by = merge_desk.load_proposals(args.file) with SessionLocal() as db: result = merge_desk.propose(db, entries, proposed_by=args.by or proposed_by, check=args.check) if args.check: db.rollback() else: db.commit() prefix = "[check, nothing written] " if args.check else "" print(f"{prefix}propose: {result['inserted']} pair(s) inserted, {result['endorsed']} rule pair(s) endorsed, " f"{result['skipped_present']} already present or decided, " f"{result['skipped_missing']} with a side gone, {result['invalid']} invalid, of {len(entries)} in {args.file}") for note in result["notes"]: print(" ", note) return 0 def cmd_review(args: argparse.Namespace) -> int: """The human publish gate on a brand or line page (the catalogue decisions §2.8): `checked` stamps that a person looked and it is one brand or one line with a clean name; `hidden` takes an eligible page off the site; `clear` withdraws the decision. Recorded in the one ledger (`overrides`) against the named account; the brand floor reads `hidden`.""" from app.models import Account from app.services import overrides as overrides_service with SessionLocal() as db: account = db.scalar(select(Account).where(Account.username == args.by.strip().lower())) if account is None: print(f"review: no account named {args.by!r}") return 2 state = None if args.state == "clear" else args.state prior = overrides_service.fields_of(db, args.entity, str(args.id)).get("review") prior_state = prior.value if prior is not None else None try: result = overrides_service.review(db, args.entity, args.id, state, set_by=account.id, reason=args.reason) except overrides_service.Refused as exc: print(f"review: {exc.code}: {exc.summary}") return 2 from app.services import audit_log audit_log.record("review.set", entity_type=args.entity, entity_key=str(args.id), account_id=account.id, detail={"review": state, "prior": prior_state, "reason": args.reason}) print(f"review: {args.entity} {args.id} {result['review'] or 'cleared'} by {account.username}") return 0 def cmd_suggest(args: argparse.Namespace) -> int: """Fill the merge queue at brand, line and product level from the rules (`services/suggest.py`). Idempotent: a pair is inserted once, an undecided pair is refreshed, a decided pair is never touched, a pair one side of which is gone is closed.""" from app.services import suggest with SessionLocal() as db: if args.check: db.commit = db.flush # type: ignore[method-assign] counts = suggest.generate(db) if args.check: db.rollback() else: db.commit() print( f"{'[check, nothing written] ' if args.check else ''}suggest: {counts['brand']} brand, {counts['line']} line, " f"{counts['product']} product pair(s) added; {counts['refreshed']} refreshed, {counts['superseded']} closed as " f"superseded, {counts['withdrawn']} withdrawn" ) return 0 def cmd_backfill(args: argparse.Namespace) -> int: """Run one backfill. `--check` prints what a run would change and writes nothing: every commit inside the backfill becomes a flush, and the transaction is rolled back at the end, so a backfill written before the flag existed honours it too.""" import inspect if args.name != "authors" and (args.map or args.include_defaulted or getattr(args, "rename", False)): print(f"backfill {args.name}: --map, --include-defaulted and --rename belong to `authors` only") return 2 if args.name in DATA_MOVES and not args.check and _collection_running(getattr(args, "force", False)): return 2 fn = BACKFILLS[args.name] with SessionLocal() as db: if args.check: db.commit = db.flush # type: ignore[method-assign] try: takes_args = len(inspect.signature(fn).parameters) >= 2 message = fn(db, args) if takes_args else fn(db) except ValueError as exc: db.rollback() print(exc) return 2 if args.check: db.rollback() message = f"[check, nothing written] {message}" print(message) return 0 def match_shops(shops: list, tokens: list[str]) -> list: """Resolve user-typed codes to shops by code or IATA, case-insensitively. A token that matches nothing is an error (raise, don't guess): a demo set silently missing an airport is worse than a rejected command. """ wanted = {t.upper() for t in tokens} matched, hit = [], set() for loc in shops: keys = {loc.code.upper()} | ({loc.iata.upper()} if loc.iata else set()) if keys & wanted: matched.append(loc) hit |= keys & wanted missing = wanted - hit if missing: raise SystemExit(f"no shop matches: {', '.join(sorted(missing))}") return matched def cmd_shops(args: argparse.Namespace) -> int: with SessionLocal() as db: shops = list(db.scalars(select(Shop).order_by(Shop.code))) if args.only: chosen = {loc.id for loc in match_shops(shops, args.only)} for loc in shops: loc.visible = loc.id in chosen elif args.show or args.hide: for loc in match_shops(shops, args.show or []): loc.visible = True for loc in match_shops(shops, args.hide or []): loc.visible = False db.commit() counts = dict( db.execute( select(Listing.shop_id, func.count(Listing.id)).group_by( Listing.shop_id ) ).all() ) for loc in shops: state = "visible" if loc.visible else "HIDDEN" iata = loc.iata or "-" print(f" {loc.code:28s} {iata:4s} {state:8s} {counts.get(loc.id, 0):5d} listings") return 0 def cmd_status(_: argparse.Namespace) -> int: with SessionLocal() as db: multi = ( select(Listing.variant_id) .group_by(Listing.variant_id) .having(func.count(func.distinct(Listing.shop_id)) > 1) .subquery() ) print(f"product variants: {db.scalar(select(func.count(ProductVariant.id))) or 0}") print(f"comparable: {db.scalar(select(func.count()).select_from(multi)) or 0}") print(f"observations: {db.scalar(select(func.count(PriceObservation.id))) or 0}") # The state word is `collector_view.state_of`'s, the same the page shows (Stream AW4): `off` # for the kill switch, `paused` for the cooperative state. The detail column keeps the # last run's status, which the deploy gate's scripts read for `running`. from app.services import collector_view states = {c.slug: c.state for c in collector_view.live(db).collectors} for source in db.scalars(select(Source).order_by(Source.slug)): last = db.scalar( select(CollectionRun) .where(CollectionRun.source_id == source.id) .order_by(CollectionRun.started_at.desc()) .limit(1) ) state = states.get(source.slug, "enabled" if source.enabled else "off") detail = f"{last.status} ({last.prices_written} prices)" if last else "never run" print(f" {source.slug:22s} {state:8s} {detail}") return 0 def cmd_sweep_plan(_: argparse.Namespace) -> int: """What a collection sweep will actually reach, read from the sources and their last run, never typed into a document (K12.1). A source whose last run was a refusal is never contacted again until a person records that permission changed; the sweep skips it and the review cites its collected names instead (`docs/REVIEW-PROCESS.md`).""" with SessionLocal() as db: print("before the sweep: app.cli backfill stuck_runs (a crashed run would otherwise block every approval)") print(" app.cli backfill prune_lines (empty product lines from earlier rules and folds)") print("then, per source, one process each:") reach, refused = [], [] for source in db.scalars(select(Source).order_by(Source.slug)): last = db.scalar(select(CollectionRun).where(CollectionRun.source_id == source.id).order_by(CollectionRun.started_at.desc()).limit(1)) when = last.finished_at.date().isoformat() if last and last.finished_at else "never" if not source.enabled: refused.append((source.slug, "paused", when, "")) elif last is not None and last.status == "blocked": refused.append((source.slug, "refused", when, (last.error or "")[:90])) else: reach.append((source.slug, last.status if last else "never run", when)) for slug, status, when in reach: print(f" collect --source {slug:24s} last {status} {when}") print(f"{len(reach)} source(s) a sweep reaches; {len(refused)} it does not:") for slug, why, when, error in refused: print(f" {slug:24s} {why:8s} since {when} {error}") return 0 def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="app.cli") sub = parser.add_subparsers(dest="command", required=True) collect = sub.add_parser("collect", help="run collectors into the database") collect.add_argument("--source", help="collector slug; default is every enabled collector") collect.add_argument("--limit", type=int, default=None, help="max listings per source") collect.add_argument("--delay", type=float, default=None, help="politeness delay in seconds") collect.add_argument("--mode", choices=("discover", "recheck"), default="discover", help="discover reads the listing pages; recheck reads back the held listings") collect.add_argument("--by", default="shell", help="who started it, recorded on the run (the page passes the user)") collect.add_argument("--ignore-freeze", action="store_true", help="run even while the deploy freeze marker exists") collect.set_defaults(func=cmd_collect) awards = sub.add_parser("awards", help="import competition medals onto product variants") awards.add_argument("--file", help=f"winners JSON (default {DEFAULT_WINNERS_PATH})") awards.add_argument( "--rebuild", action="store_true", help="drop competition-network medals and re-match from scratch", ) awards.set_defaults(func=cmd_awards) discussion = sub.add_parser("discussion", help="the decision cards import, and the inbox a session reads") dsub = discussion.add_subparsers(dest="discussion_command", required=True) dimport = dsub.add_parser("import", help="import the decisions/realities list onto /discuss") dimport.add_argument("--file", help=f"discussion JSON (default {DEFAULT_DISCUSSION_PATH})") dimport.set_defaults(func=cmd_discussion) dinbox = dsub.add_parser("inbox", help="what rian said to @claude on the running list, oldest first") dinbox.add_argument("--for", dest="recipient", required=True, metavar="HANDLE", help="claude") dinbox.add_argument("--since", metavar="DATE", help="ISO date; only comments on or after it") dinbox.set_defaults(func=cmd_discussion_inbox) drethread = dsub.add_parser("rethread", help="apply a re-threading map once: topics, moves, soft deletes (T18)") drethread.add_argument("--file", default="import/rethread-2026-09-13.json", help="the map (default: the 13 Sep map)") drethread.add_argument("--check", action="store_true", help="print what would happen; write nothing") drethread.add_argument("--apply", action="store_true", help="apply the map, attributed to the owner account") drethread.set_defaults(func=cmd_discussion_rethread) backfill = sub.add_parser("backfill", help="run one idempotent data move after a deploy") backfill.add_argument("name", choices=sorted(BACKFILLS)) backfill.add_argument("--check", action="store_true", help="print what would change; write nothing") backfill.add_argument("--force", action="store_true", help="run a data move even while a collection is running") backfill.add_argument("--map", action="append", default=[], metavar="TYPED=USERNAME", help="authors only: link this typed name to this account (repeatable)") backfill.add_argument("--rename", action="store_true", help="authors only: also rewrite each mapped spelling to the account's display name") backfill.add_argument("--include-defaulted", action="store_true", help="authors only: also link feature_priorities and quote_selections, " "where the page defaulted an anonymous actor to a name") backfill.set_defaults(func=cmd_backfill) rederive = sub.add_parser("rederive", help="recompute match keys under the current identity rules") rederive.add_argument("--force", action="store_true", help="run even while a collection is running") rederive.set_defaults(func=cmd_rederive) review = sub.add_parser("review", help="hide or clear the human publish gate on a brand or product line page (checked is retired: an approved sheet is the quality fact)") review.add_argument("entity", choices=("brand", "line")) review.add_argument("id", type=int, help="the brand or line id") review.add_argument("state", choices=("hidden", "clear")) review.add_argument("--reason", default=None, help="why, kept with the decision") review.add_argument("--by", default="rian", help="the account username the decision is recorded against (default rian)") review.set_defaults(func=cmd_review) suggest = sub.add_parser("suggest", help="fill the merge queue at brand, line and product level from the rules") suggest.add_argument("--check", action="store_true", help="print what would change; write nothing") suggest.set_defaults(func=cmd_suggest) propose = sub.add_parser("propose", help="file a proposals JSON as merge candidates a person approves on the desk") propose.add_argument("--file", required=True, help="the proposals file (import/proposals/-.json)") propose.add_argument("--by", default=None, help="who proposed (default: the file's proposed_by)") propose.add_argument("--check", action="store_true", help="report the counts; write nothing") propose.set_defaults(func=cmd_propose) sweep = sub.add_parser("sweep", help="a collection sweep: what it reaches and what comes first") ssub = sweep.add_subparsers(dest="sweep_command", required=True) ssub.add_parser("plan", help="the sources a sweep reaches, the refused ones it never contacts, and the two backfills that come first").set_defaults(func=cmd_sweep_plan) status = sub.add_parser("status", help="show catalog and collector status") status.set_defaults(func=cmd_status) shops = sub.add_parser( "shops", help="list shops and toggle site visibility (collectors keep running)", ) shops.add_argument("--show", nargs="+", metavar="CODE", help="make these visible") shops.add_argument("--hide", nargs="+", metavar="CODE", help="hide these from the site") shops.add_argument( "--only", nargs="+", metavar="CODE", help="make exactly these visible and hide everything else", ) shops.set_defaults(func=cmd_shops) register_quality(sub) # audit, verify: app/cli_quality.py (Stream Q) register_pages(sub) # indexnow: app/cli_pages.py (Stream B) register_editorial(sub) # articles, subscribers: app/cli_editorial.py (Stream D) register_accounts(sub) # accounts, sessions, audit-log: app/cli_accounts.py (Stream R) register_refresh(sub) # staging-refresh export | apply: app/cli_refresh.py (Stream R2) register_hours(sub) # hours collect/set/show: app/cli_hours.py (Stream G) register_images(sub) # images fetch/manifest: app/cli_images.py (Stream AW3) register_featured(sub) # featured evidence: app/cli_featured.py (Stream AW2) register_places(sub) # places guide import/export, places list: app/cli_places.py (Stream G2) register_decisions(sub) # decisions list/undo-batch/verify/export/replay: app/cli_decisions.py (Stream K2) register_proposals(sub) # proposals load/withdraw/sheet/approve: app/cli_proposals.py (Stream K4) register_precedents(sub) # precedents list/show/overturn/export: app/cli_precedents.py (Stream K12) register_pass(sub) # pass packet/status/fingerprint/parked/unpark: app/cli_pass.py (Stream K12) register_index(sub) # index suggest/list/approve/remove: app/cli_index.py (Stream K6) register_brands(sub) # brands spellings/split: app/cli_brands.py (Stream K9) args = parser.parse_args(argv) return args.func(args) if __name__ == "__main__": sys.exit(main())