"""The review (Stream K4; plan W10, W13, W15, W16; the escalation-2 spec §6 and §7): a proposals file loaded under a named pass, the sheet a person reads, the approval that turns proposals into decisions in one batch, the reject that is a Keep separate for a pair, and the two undos. Sources of truth: this module, `proposals_schema.json` (the file's shape), `proposals_store.py` (the thin upsert every pass writes through), `services/decisions/` (the one writer, undo and the effective read), `main/docs/REVIEW-PROCESS.md` (§4 the spot-check rule this module implements, §6 the file), `tests/test_proposals_load.py`, `tests/test_review_approval.py`. Four rules hold everywhere here. A proposal carries natural keys only, so a file written on staging loads on production unchanged (the `propose --file` trap). The sheet read writes nothing: the effective decision, the drift and the staleness are computed at read. Staleness is a change in what was read (a cited listing gone or ignored, or its text at the span changed), never a rule's rederive. A machine never applies a proposal: approval is a person's act, one batch per brand per call, undone as one. """ from __future__ import annotations import hashlib import json import math import pathlib import re import uuid as _uuid from collections import Counter, defaultdict from datetime import UTC, datetime from types import SimpleNamespace from typing import Any from sqlalchemy import func, select from sqlalchemy.orm import Session from app.models import (Account, Brand, Decision, DecisionBatch, Listing, ProductLine, ProductVariant, Proposal, ProposalPass, Retailer, Shop) from app.services import proposal_rules, proposals_store from app.services.decisions import natural_keys from app.services.decisions.effective import effective from app.services.decisions.writer import Refused, jsonable, refuse_if_collecting SCHEMA_PATH = pathlib.Path(__file__).with_name("proposals_schema.json") #: REVIEW-PROCESS.md §4, as numbers. Changing one is a version bump of that file. SPOT_LOWEST = 5 SPOT_BELOW = 0.7 SPOT_SAMPLE_SHARE = 0.05 SPOT_SAMPLE_MIN = 3 OPEN_STATES = ("open", "stale", "parked") #: The pass kinds that are a person's queue. Rian's ruling of 17 Sep: the programmatic stage may #: act, must err toward separate, and is NEVER reviewed by him, so a `rule` pass is not a question #: -- it is a hint the AI pass may read (`hints()`, REVIEW-PROCESS.md §1 and §2). Pairing W14's #: option B with option D put 938 of those in front of him before the AI pass had ever run. An #: `arrival` pass is a new listing joining an approved line, which is the pass's input too. REVIEW_KINDS = ("session",) class FileInvalid(ValueError): """The file does not have the spec §6.1 shape; the message names the row and the field.""" def __init__(self, where: str, message: str) -> None: super().__init__(f"{where}: {message}") self.where, self.message = where, message # --------------------------------------------------------------------------- the file's schema def _schema() -> dict: return json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) _TYPES = {"object": dict, "array": list, "string": str, "integer": int, "number": (int, float), "boolean": bool, "null": type(None)} def _type_ok(value: Any, name: str) -> bool: if name in ("integer", "number") and isinstance(value, bool): return False return isinstance(value, _TYPES[name]) def _check(schema: dict, value: Any, path: str) -> None: """The subset of JSON Schema the file's schema uses (type, enum, required, properties, additionalProperties, items, min/max, pattern, lengths). The server image carries no validator library; the schema file stays standard so one can be swapped in.""" types = schema.get("type") if types is not None: names = types if isinstance(types, list) else [types] if not any(_type_ok(value, n) for n in names): raise FileInvalid(path, f"expected {' or '.join(names)}, got {type(value).__name__}") if "enum" in schema and value not in schema["enum"]: raise FileInvalid(path, f"{value!r} is not one of {schema['enum']}") if isinstance(value, str): if len(value) < schema.get("minLength", 0) or len(value) > schema.get("maxLength", 10 ** 9): raise FileInvalid(path, f"length {len(value)} is outside {schema.get('minLength', 0)}..{schema.get('maxLength', '')}") if "pattern" in schema and not re.search(schema["pattern"], value): raise FileInvalid(path, f"{value!r} does not match {schema['pattern']}") if isinstance(value, (int, float)) and not isinstance(value, bool): if "minimum" in schema and value < schema["minimum"]: raise FileInvalid(path, f"{value} is below {schema['minimum']}") if "maximum" in schema and value > schema["maximum"]: raise FileInvalid(path, f"{value} is above {schema['maximum']}") if isinstance(value, dict): for key in schema.get("required", []): if key not in value: raise FileInvalid(path, f"missing {key!r}") props = schema.get("properties", {}) for key, item in value.items(): if key in props: _check(props[key], item, f"{path}.{key}") elif schema.get("additionalProperties") is False: raise FileInvalid(path, f"unexpected field {key!r}") if isinstance(value, list): if len(value) < schema.get("minItems", 0) or len(value) > schema.get("maxItems", 10 ** 9): raise FileInvalid(path, f"{len(value)} item(s), outside {schema.get('minItems', 0)}..{schema.get('maxItems', '')}") if "items" in schema: for i, item in enumerate(value): _check(schema["items"], item, f"{path}[{i}]") def validate_file(data: Any) -> None: """Raise `FileInvalid` naming the row (`proposals[12]`, its entity key) and the field.""" schema = _schema() if not isinstance(data, dict): raise FileInvalid("file", "a proposals file is one JSON object") head = {k: v for k, v in data.items() if k != "proposals"} _check({**schema, "required": [r for r in schema["required"] if r != "proposals"], "properties": {k: v for k, v in schema["properties"].items() if k != "proposals"}, "additionalProperties": True}, head, "file") if "proposals" not in data: raise FileInvalid("file", "missing 'proposals'") if not isinstance(data["proposals"], list): raise FileInvalid("file.proposals", "expected array") row_schema = schema["properties"]["proposals"]["items"] seen: dict[tuple, int] = {} for i, row in enumerate(data["proposals"]): key = row.get("entity", {}).get("key") if isinstance(row, dict) and isinstance(row.get("entity"), dict) else None where = f"proposals[{i}]" + (f" ({key})" if key else "") try: _check(row_schema, row, where) except FileInvalid as exc: raise FileInvalid(exc.where, exc.message) from None entity = row["entity"] try: kind, _ = natural_keys.parse(entity["key"]) except ValueError as exc: raise FileInvalid(f"{where}.entity.key", str(exc)) from None if kind != entity["type"]: raise FileInvalid(f"{where}.entity.key", f"a {kind} key under type {entity['type']}") for j, ev in enumerate(row["evidence"]): start, end = ev["span"] if end <= start: raise FileInvalid(f"{where}.evidence[{j}].span", "the span ends before it starts") if end - start != len(ev["text"]): raise FileInvalid(f"{where}.evidence[{j}].text", f"the span is {end - start} characters, the text {len(ev['text'])}") ident = (entity["type"], entity["key"], row["field"]) if ident in seen: raise FileInvalid(f"{where}.field", f"the same entity and field as proposals[{seen[ident]}]") seen[ident] = i # --------------------------------------------------------------------------- shared reads def canon(value: Any) -> str: return json.dumps(jsonable(value), sort_keys=True, ensure_ascii=False) def _brand_family(db: Session, slug: str) -> tuple[Brand | None, list[Brand]]: """The brand a slug resolves to (aliases followed) and every brand row that is it or points at it.""" row = db.scalar(select(Brand).where(Brand.slug == slug)) hops = 0 while row is not None and row.alias_of_id is not None and hops < 6: row, hops = db.get(Brand, row.alias_of_id), hops + 1 if row is None: return None, [] family = {row.id: row} frontier = [row.id] while frontier: nxt = list(db.scalars(select(Brand).where(Brand.alias_of_id.in_(frontier)))) frontier = [b.id for b in nxt if b.id not in family] family.update({b.id: b for b in nxt}) return row, list(family.values()) class ListingCache: """Listing rows by natural key, loaded in one query per call (the sheet cites hundreds).""" def __init__(self, db: Session) -> None: self.db, self.rows = db, {} def warm(self, keys) -> None: wanted = {} for key in set(keys) - set(self.rows): try: _, parts = natural_keys.parse(key) except ValueError: self.rows[key] = None continue wanted[(parts["retailer"], parts["code"], parts["sku"])] = key skus = {sku for _, _, sku in wanted} found = {} for i in range(0, len(skus), 500): chunk = list(skus)[i:i + 500] for listing, code, retailer in self.db.execute( select(Listing, Shop.code, Retailer.slug).join(Shop, Listing.shop_id == Shop.id) .join(Retailer, Shop.retailer_id == Retailer.id).where(Listing.source_sku.in_(chunk))): found[(retailer, code, listing.source_sku)] = listing for triple, key in wanted.items(): self.rows[key] = found.get(triple) def get(self, key: str) -> Listing | None: if key not in self.rows: self.warm([key]) return self.rows[key] def _listed_text(listing: Listing, source: str) -> str | None: if source.startswith("option:"): # A shop's option field lands in the listed layer's variant column (REVIEW-PROCESS.md §1 item 5). return listing.listed_variant return getattr(listing, source, None) def evidence_state(cache: ListingCache, evidence: list) -> tuple[bool, str | None, list[dict]]: """Whether every cited span still reads as it did: `(ok, why, per-span)`. A listing gone, ignored, or its text at the span changed makes the row stale (spec §6.3).""" cache.warm(ev.get("listing") for ev in evidence or [] if ev.get("listing")) spans, why = [], None for ev in evidence or []: listing = cache.get(ev.get("listing", "")) span = ev.get("span") if isinstance(ev.get("span"), list) else None state = {"listing": ev.get("listing"), "source": ev.get("source"), "span": span, "text": ev.get("text"), "matched": ev.get("matched"), "listed_text": None, "ok": True, "why": None} if not ev.get("listing"): # A rule's evidence cites the name it read, not one listing's span: there is nothing # positional to re-check, and calling it gone made every rule row on the sheet stale. pass elif listing is None: state.update(ok=False, why="the listing is gone") elif listing.ignored_at is not None: state.update(ok=False, why="the listing is ignored", listed_text=_listed_text(listing, ev.get("source", ""))) else: text = _listed_text(listing, ev.get("source", "")) or "" state["listed_text"] = text if span is None: if ev.get("text") and ev["text"] not in text: state.update(ok=False, why="the listed text no longer contains it") else: start, end = span[:2] if text[start:end] != ev.get("text"): state.update(ok=False, why=f"the listed text at {start}..{end} now reads {text[start:end]!r}") if not state["ok"] and why is None: why = f"{ev.get('listing')}: {state['why']}" spans.append(state) return why is None, why, spans def _entity(db: Session, entity_type: str, entity_id: int | None): from app.services.decisions.writer import MODELS return db.get(MODELS[entity_type], entity_id) if entity_id is not None else None def column_value(db: Session, entity_type: str, entity, field_name: str) -> Any: """The entity's column for a field now, in the ledger's value shape (a reference as its key).""" from app.services import attributes from app.services.decisions import undo as undo_service from app.services.product_lines import resolve_alias if entity is None: return None value = undo_service._column_value(db, entity_type, entity, field_name, attributes, resolve_alias) if field_name in ("product_line", "alias_of", "merged_into", "pinned_to") and value is not None: from app.services.decisions.writer import MODELS, REFERENCE_FIELDS target_type = REFERENCE_FIELDS.get(field_name) or entity_type row = db.get(MODELS[target_type], value) return natural_keys.build(row)[0] if row is not None else value return jsonable(value) def _rule_value(db: Session, maps, entity_type: str, entity, field_name: str) -> Any: """What the rules have for the field (spec §1.7), through the writer's own reading.""" from app.services.decisions import writer if entity is None or not writer.field_ok(entity_type, field_name): return None prev = effective(db, entity_type, [entity.id], [field_name]).get((entity.id, field_name)) try: value = writer._rule_value(SimpleNamespace(db=db, maps=maps), entity_type, entity, field_name, prev) except Exception: # noqa: BLE001 - informational only; a rule that cannot read leaves the cell empty return None if field_name == "product_line" and isinstance(value, int): line = db.get(ProductLine, value) return natural_keys.build(line)[0] if line is not None else None return jsonable(value) def _is_merge(p) -> bool: field_name, value, entity_type = (p["field"], p["value"], p["entity_type"]) if isinstance(p, dict) else (p.field, p.value, p.entity_type) return (entity_type == "suggestion" and isinstance(value, dict) and value.get("decision") == "same") \ or field_name in ("merged_into", "pinned_to", "attribute:quantity") def spot_checks(pass_name: str, rows: list[dict]) -> set[tuple[str, str, str]]: """REVIEW-PROCESS.md §4 over one brand's rows of one pass, deterministic per pass name: (1) every proposal that merges variants or changes a comparison: a pair confirmed same, a `merged_into`, a listing pin, a quantity; (2) the five lowest-confidence rows and every one below 0.7; (3) a sample of five percent of the rows, at least three, drawn from the rest by a hash seeded with the pass name, so a re-load draws the same sample.""" ident = lambda r: (r["entity_type"], r["natural_key"], r["field"]) # noqa: E731 marked = {ident(r) for r in rows if _is_merge(r)} scored = sorted((r for r in rows if r.get("confidence") is not None), key=lambda r: (float(r["confidence"]), r.get("position", 0), ident(r))) marked |= {ident(r) for r in scored[:SPOT_LOWEST]} marked |= {ident(r) for r in scored if float(r["confidence"]) < SPOT_BELOW} want = max(SPOT_SAMPLE_MIN, math.ceil(SPOT_SAMPLE_SHARE * len(rows))) rest = sorted((r for r in rows if ident(r) not in marked), key=lambda r: hashlib.sha256("|".join((pass_name, *ident(r))).encode()).hexdigest()) marked |= {ident(r) for r in rest[:want]} return marked # --------------------------------------------------------------------------- the load def _read(file) -> tuple[bytes, Any, str | None]: if isinstance(file, (dict, list)): raw = json.dumps(file, sort_keys=True, ensure_ascii=False).encode("utf-8") return raw, file, None path = pathlib.Path(file) raw = path.read_bytes() try: return raw, json.loads(raw.decode("utf-8")), str(path) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise FileInvalid("file", f"not JSON: {exc}") from None def _account(db: Session, by) -> Account | None: if by is None or isinstance(by, Account): return by if isinstance(by, int): return db.get(Account, by) return db.scalar(select(Account).where(Account.username == str(by).strip().lower())) def _pass_decisions(db: Session, pass_id: int) -> dict[tuple[str, str], Decision]: """The newest decision of a pass per (proposal key, field): what a replay carried here.""" out: dict[tuple[str, str], Decision] = {} for row in db.scalars(select(Decision).where(Decision.pass_id == pass_id, Decision.effect == "set", Decision.reverses_id.is_(None)) .order_by(Decision.decided_at, Decision.id)): key = (row.detail or {}).get("proposal_key") if key: out[(key, row.field)] = row return out def load(db: Session, file, *, check: bool = False, as_user=None, partial: bool = False) -> dict[str, Any]: """Load one brand's proposals under its pass (spec §6.2). The pass is upserted by name with the file's sha256 (an identical file is a no-op); every row is upserted on its natural identity through `proposals_store.write`, resolved on this host (`parked` when it cannot be, visible and never approvable), its rule and current values captured, its evidence checked (a span the listing does not contain loads stale), spot-checks marked by REVIEW-PROCESS.md §4 (never cleared), a previously rejected identical row noted, and a row this pass already decided here (it arrived by replay) marked approved. Open rows missing from the file go stale unless `partial`: the file is the pass's whole statement. `check` rolls everything back.""" from app.services import keying raw, data, source = _read(file) validate_file(data) sha = hashlib.sha256(raw).hexdigest() head, brand_slug = data["pass"], data["brand"] brand, family = _brand_family(db, brand_slug) if brand is None: raise Refused("BRAND_MISSING", f"No brand {brand_slug!r} on this host.") user = _account(db, as_user) if as_user is not None and user is None: raise Refused("ENTITY_NOT_FOUND", f"No account named {as_user!r}.") existing = db.scalar(select(ProposalPass).where(ProposalPass.name == head["name"])) if existing is not None and existing.withdrawn_at is not None: raise Refused("PASS_WITHDRAWN", f"Pass {head['name']} was withdrawn on {existing.withdrawn_at:%Y-%m-%d}; load under a new name.") counts: dict[str, Any] = {"pass": head["name"], "brand": brand.slug, "rows": len(data["proposals"]), "check": check} if existing is not None and existing.file_sha256 == sha: counts["unchanged"] = True return counts specs = [] for i, p in enumerate(data["proposals"]): entity = p["entity"] specs.append(proposals_store.ProposalRow( entity_type=entity["type"], natural_key=entity["key"], natural_key_detail=entity.get("detail"), field=p["field"], value=p["value"], reason=p.get("reason"), evidence=p.get("evidence") or [], confidence=p.get("confidence"), brand_slug=brand_slug, sheet_line_ref=p.get("sheet_line_ref"), position=p.get("position", i), attention=p.get("attention"))) ours = {(s.entity_type, s.natural_key, s.field) for s in specs} # A deferred question is re-proposed ONLY by a pass that has read its note (K11.4). Rian's # note is the whole value of deferring: if a later pass could ask the same question without # having read it, the loop that turns his answers into the reasoning rules does not exist and # the set/coffret question is asked identically forever. The row names the note it answers. answers = {(row["entity"]["type"], row["entity"]["key"], row["field"]): str(row.get("answers_deferred") or "") for row in data["proposals"]} unread = [] for r in db.scalars(select(Proposal).where(Proposal.status == "deferred", Proposal.natural_key.in_([s.natural_key for s in specs] or [""]))): ident = (r.entity_type, r.natural_key, r.field) if ident in ours and answers.get(ident) != str(r.uid): unread.append((ident, r)) if unread: (ident, r) = unread[0] raise Refused("DEFERRED_NOT_READ", f"{len(unread)} row(s) re-propose a question rian deferred without naming the note. " f"{ident[1]} {ident[2]}: \"{(r.resolution_note or '')[:120]}\" " f"(read them with `proposals notes`, then set answers_deferred to {r.uid} on the row).") before = {(r.entity_type, r.natural_key, r.field): r.status for r in db.scalars(select(Proposal).where(Proposal.pass_id == existing.id))} if existing is not None else {} pass_row = proposals_store.write(db, head["name"], specs, kind=head["kind"], generator=head["generator"], rules_version=head["rules_version"], process_version=head["process_version"], scope_brand_slug=brand_slug, note=head.get("note"), loaded_by=user, source_file=source, file_sha256=sha) pass_row.file_sha256, pass_row.source_file, pass_row.note = sha, source, head.get("note") pass_row.process_version, pass_row.rules_version, pass_row.generator = head["process_version"], head["rules_version"], head["generator"] pass_row.loaded_at, pass_row.loaded_by = datetime.now(UTC), getattr(user, "id", None) pass_row.loaded_by_username = getattr(user, "username", None) rows = {(r.entity_type, r.natural_key, r.field): r for r in db.scalars(select(Proposal).where(Proposal.pass_id == pass_row.id))} maps = keying.load_maps(db) cache = ListingCache(db) cache.warm(ev["listing"] for s in specs for ev in s.evidence) replayed = _pass_decisions(db, pass_row.id) marked = spot_checks(pass_row.name, [{"entity_type": s.entity_type, "natural_key": s.natural_key, "field": s.field, "value": s.value, "confidence": s.confidence, "position": s.position} for s in specs]) rejected_before: dict[tuple, str] = {} for r in db.scalars(select(Proposal).where(Proposal.status == "rejected", Proposal.pass_id != pass_row.id, Proposal.natural_key.in_([s.natural_key for s in specs] or [""]))): rejected_before[(r.entity_type, r.natural_key, r.field, canon(r.value))] = str(r.uid) tally = Counter() for spec in specs: ident = (spec.entity_type, spec.natural_key, spec.field) row = rows[ident] prior = before.get(ident) tally["inserted" if prior is None else ("kept" if prior not in OPEN_STATES else "updated")] += 1 if row.status not in OPEN_STATES: tally[f"status_{row.status}"] += 1 continue detail = dict(row.detail or {}) if spec.entity_type == "suggestion" and row.status == "parked" and (row.resolution_note or "").startswith("PAIR_MISSING"): # No rule offered the pair: the approval mints its suggestion row (spec §2), so it is creatable, not parked. row.status, row.resolution, row.resolution_note = "open", "creatable", None entity = _entity(db, row.entity_type, row.entity_id) if row.resolution == "resolved" else None row.current_value = column_value(db, row.entity_type, entity, row.field) row.rule_value = _rule_value(db, maps, row.entity_type, entity, row.field) row.spot_check = bool(row.spot_check) or ident in marked rejected = rejected_before.get((*ident, canon(row.value))) if rejected: detail["previously_rejected"] = rejected done = replayed.get((row.natural_key, row.field)) if done is not None: now = effective(db, done.entity_type, [done.entity_id], [done.field]).get((done.entity_id, done.field)) if now is not None and now.id == done.id: # A pair's reject is a person's Keep separate naming the proposal it answered. rejected_here = bool((done.detail or {}).get("rejected_proposal")) row.status, row.decision_id, row.resolved_at = ("rejected" if rejected_here else "approved"), done.id, datetime.now(UTC) row.resolution_note = done.reason if rejected_here else "decided by this pass on another host; arrived by replay" row.detail = detail tally["approved_on_load"] += 1 continue if row.status == "open": ok, why, _ = evidence_state(cache, row.evidence) if not ok: row.status, row.resolution_note = "stale", f"EVIDENCE_MISMATCH: {why}" tally["evidence_mismatch"] += 1 row.detail = detail tally[row.resolution] += 1 tally[row.status] += 1 if row.spot_check and row.status == "open": tally["spot_check"] += 1 if rejected: tally["previously_rejected"] += 1 if not partial: for ident, row in rows.items(): if ident not in ours and row.status in ("open", "parked"): row.status, row.resolution_note = "stale", "not in the latest file of this pass" tally["stale_missing"] += 1 _derive_absorbs(db, brand, family, rows.values()) counts.update(dict(tally)) pass_row.counts = {k: v for k, v in counts.items() if k not in ("check",)} | {"at": datetime.now(UTC).isoformat()} db.flush() if check: db.rollback() keying.invalidate() else: db.commit() return counts def _derive_absorbs(db: Session, brand: Brand, family: list[Brand], rows) -> None: """Spec §6.4 step 4: a line header absorbs every existing line of the brand whose every live member is proposed into it, unless the file named its absorbs itself.""" headers = {r.natural_key: r for r in rows if r.entity_type == "product_line" and r.field == "name"} moves: dict[int, set[str]] = defaultdict(set) members = [r for r in rows if r.entity_type == "product_variant" and r.field == "product_line" and r.entity_id is not None] for r in members: variant = db.get(ProductVariant, r.entity_id) if variant is not None and variant.product_line_id is not None: moves[variant.product_line_id].add((str(r.value), variant.id)) for product_line_id, pairs in moves.items(): targets = {t for t, _ in pairs} if len(targets) != 1: continue target = next(iter(targets)) header = headers.get(target) line = db.get(ProductLine, product_line_id) if header is None or line is None or line.alias_of_id is not None: continue live = set(db.scalars(select(ProductVariant.id).where(ProductVariant.product_line_id == product_line_id, ProductVariant.merged_into_id.is_(None)))) if live and live <= {vid for _, vid in pairs}: key = natural_keys.build(line)[0] if key == target or (header.entity_id is not None and header.entity_id == product_line_id): continue detail = dict(header.detail or {}) derived = sorted(set(detail.get("absorbs_derived", [])) | {key}) detail["absorbs_derived"] = derived header.detail = detail def withdraw(db: Session, pass_name: str, reason: str, by) -> dict[str, Any]: """Withdraw a pass (spec §6.2): its open, stale and parked rows go `withdrawn`, the pass is stamped, and the batches that approved its rows are listed with the undo command for each. The ledger is never touched here: undoing an approval is a person's own act.""" pass_row = db.scalar(select(ProposalPass).where(ProposalPass.name == pass_name)) if pass_row is None: raise Refused("PASS_NOT_FOUND", f"No pass {pass_name!r}.") if not (reason or "").strip(): raise Refused("VALUE_INVALID", "A withdrawal says why.") user = _account(db, by) if user is None: raise Refused("ENTITY_NOT_FOUND", "A withdrawal is a person's act: name an account.") n = 0 for row in db.scalars(select(Proposal).where(Proposal.pass_id == pass_row.id, Proposal.status.in_(OPEN_STATES))): row.status, row.resolution_note, row.resolved_at, row.resolved_by = "withdrawn", f"pass withdrawn: {reason.strip()}", datetime.now(UTC), user.id n += 1 if pass_row.withdrawn_at is None: pass_row.withdrawn_at, pass_row.withdrawn_by, pass_row.withdrawn_reason = datetime.now(UTC), user.id, reason.strip() batches = [] ids = sorted({bid for bid in db.scalars(select(Decision.batch_id).where(Decision.pass_id == pass_row.id))}) for b in db.scalars(select(DecisionBatch).where(DecisionBatch.id.in_(ids or [0])).order_by(DecisionBatch.opened_at)): undone = db.scalar(select(DecisionBatch.uid).where(DecisionBatch.reverses_batch_id == b.id)) batches.append({"uid": str(b.uid), "opened_at": b.opened_at.isoformat() if b.opened_at else None, "by": b.by_username, "undone_by": str(undone) if undone else None, "undo": None if undone else f"python -m app.cli decisions undo-batch {b.uid} --reason \"pass {pass_name} withdrawn\" --by "}) db.commit() return {"pass": pass_name, "withdrawn_rows": n, "approved_batches": batches} # --------------------------------------------------------------------------- the sheet read def _decision_out(dec: Decision | None, passes: dict[int, ProposalPass], batch_uids: dict[int, str]) -> dict | None: if dec is None: return None p = passes.get(dec.pass_id) if dec.pass_id else None return {"id": dec.id, "uid": str(dec.uid), "value": dec.value, "by": dec.decided_by_username, "at": dec.decided_at, "mode": dec.mode, "origin": dec.origin, "pass_name": p.name if p else None, "process_version": p.process_version if p else None, "batch_uid": batch_uids.get(dec.batch_id), "reason": dec.reason} def brand_index(db: Session, *, q: str | None = None, limit: int = 100, offset: int = 0, has_brand_decision: bool = False, has_suggestions: bool = False) -> dict[str, Any]: """EVERY brand, not only the ones with something waiting (rian, 18 Sep), with every count a column could show. The page picks which columns to draw; this says what they would hold. Why all of them: two rows that are really one company stay two brands until a person folds them, and a brand with nothing pending is still one he may want to open, rename or publish. An index that hid the empty ones could not show him Paco Rabanne beside Rabanne, which is the pair he is trying to look at. Writes nothing. """ from sqlalchemy import Integer, case, or_ live = select(ProposalPass.id).where(ProposalPass.withdrawn_at.is_(None), ProposalPass.kind.in_(REVIEW_KINDS)) counted = (select( Proposal.brand_slug.label("slug"), func.count(Proposal.id).filter(Proposal.status == "open").label("pending"), func.count(Proposal.id).filter(Proposal.status == "open", Proposal.spot_check.is_(True)).label("spot_check"), func.count(Proposal.id).filter(Proposal.status == "open", Proposal.against_decision_id.isnot(None)).label("held"), func.count(Proposal.id).filter(Proposal.status == "parked").label("parked"), func.count(Proposal.id).filter(Proposal.status == "stale").label("stale"), ).where(Proposal.pass_id.in_(live)).group_by(Proposal.brand_slug).subquery()) shape = (select(ProductVariant.brand_id.label("bid"), func.count(ProductVariant.id).label("variants"), func.count(func.distinct(ProductVariant.product_line_id)).label("lines")) .where(ProductVariant.merged_into_id.is_(None)).group_by(ProductVariant.brand_id).subquery()) listed = (select(ProductVariant.brand_id.label("bid"), func.count(Listing.id).label("n")) .join(Listing, Listing.variant_id == ProductVariant.id) .where(ProductVariant.merged_into_id.is_(None), Listing.ignored_at.is_(None)) .group_by(ProductVariant.brand_id).subquery()) # Which brands a brand fold names. It is filed under one slug and is about two, so both are # marked: the filter has to find the half whose own file never carried the question. folds: dict[str, list[str]] = {} for f in db.scalars(select(Proposal).where(Proposal.pass_id.in_(live), Proposal.status == "open", Proposal.entity_type == "suggestion", Proposal.natural_key.like("pair:brand:%"))): for side in _variant_sides(f): if side.startswith("brand:"): folds.setdefault(side[len("brand:"):], []).append(str(f.uid)) n_pending = func.coalesce(counted.c.pending, 0) n_listings = func.coalesce(listed.c.n, 0) base = (select(Brand.slug, Brand.name, Brand.id, n_pending.label("pending"), func.coalesce(counted.c.spot_check, 0).label("spot_check"), func.coalesce(counted.c.held, 0).label("held"), func.coalesce(counted.c.parked, 0).label("parked"), func.coalesce(counted.c.stale, 0).label("stale"), n_listings.label("listings"), func.coalesce(shape.c.variants, 0).label("product_variants"), func.coalesce(shape.c.lines, 0).label("product_lines")) .select_from(Brand) .outerjoin(counted, counted.c.slug == Brand.slug) .outerjoin(shape, shape.c.bid == Brand.id) .outerjoin(listed, listed.c.bid == Brand.id) .where(Brand.alias_of_id.is_(None))) if q and q.strip(): like = f"%{q.strip()}%" base = base.where(or_(Brand.name.ilike(like), Brand.slug.ilike(like))) if has_suggestions: base = base.where(n_pending > 0) if has_brand_decision: base = base.where(Brand.slug.in_(list(folds) or [""])) total = db.scalar(select(func.count()).select_from(base.subquery())) or 0 waiting = db.scalar(select(func.count()).select_from(base.where(n_pending > 0).subquery())) or 0 rows = db.execute(base.order_by(case((n_pending > 0, 0), else_=1).cast(Integer), n_pending.desc(), n_listings.desc(), Brand.slug) .limit(limit).offset(offset)).all() # The value-order metric, and the newest pass, only for the brands that have something waiting: # both mean nothing on an empty brand and cost a query each. gain: dict[str, int] = {} newest: dict[str, str] = {} busy = [r.slug for r in rows if r.pending] if busy: all_passes = {p.id: p for p in db.scalars(select(ProposalPass))} for r in db.scalars(select(Proposal).where(Proposal.pass_id.in_(live), Proposal.status == "open", Proposal.brand_slug.in_(busy))): p_row = all_passes.get(r.pass_id) if p_row is not None: newest[r.brand_slug] = p_row.name if not _is_merge(r): continue uids = [] for side in _variant_sides(r): if side.startswith("variant:"): try: uids.append(_uuid.UUID(side[len("variant:"):])) except ValueError: continue if uids: barcoded = set(db.scalars(select(ProductVariant.uid).where( ProductVariant.uid.in_(uids), ProductVariant.gtin.isnot(None)))) gain[r.brand_slug] = gain.get(r.brand_slug, 0) + len(uids) - len(barcoded) return {"total": int(total), "with_pending": int(waiting), "limit": limit, "offset": offset, "rows": [{"brand_slug": r.slug, "brand_name": r.name, "pending": int(r.pending), "spot_check": int(r.spot_check), "held": int(r.held), "parked": int(r.parked), "stale": int(r.stale), "listings": int(r.listings), "product_lines": int(r.product_lines), "product_variants": int(r.product_variants), "comparison_gain": int(gain.get(r.slug, 0)), "newest_pass": newest.get(r.slug), "brand_decisions": len(folds.get(r.slug, []))} for r in rows]} def sheets(db: Session) -> list[dict[str, Any]]: """Every brand with proposals still waiting, in REVIEW-PROCESS.md §5's value order: the comparison gain first (barcode-less variants the open merge proposals would bring into one comparison), then the brand's listing count, then the slug. Writes nothing. Only a review kind counts (`REVIEW_KINDS`): a word list's reading is the pass's input, not a brand's queue, so a brand whose only waiting rows are `rule` rows is not on the index at all.""" live_passes = {p.id: p for p in db.scalars(select(ProposalPass).where(ProposalPass.withdrawn_at.is_(None), ProposalPass.kind.in_(REVIEW_KINDS)))} if not live_passes: return [] rows = list(db.scalars(select(Proposal).where(Proposal.pass_id.in_(list(live_passes)), Proposal.status.in_(OPEN_STATES)))) by_brand: dict[str, dict] = {} slug_to_brand: dict[str, Brand | None] = {} for r in rows: if r.brand_slug not in slug_to_brand: slug_to_brand[r.brand_slug] = _brand_family(db, r.brand_slug)[0] brand = slug_to_brand[r.brand_slug] slug = brand.slug if brand is not None else r.brand_slug entry = by_brand.setdefault(slug, {"brand_slug": slug, "brand_name": brand.name if brand else r.brand_slug, "open": 0, "spot_check": 0, "held": 0, "parked": 0, "stale": 0, "gain": set(), "newest_pass": None, "_newest": None, "passes": set()}) entry["passes"].add(live_passes[r.pass_id].name) loaded = live_passes[r.pass_id].loaded_at if entry["_newest"] is None or (loaded and loaded > entry["_newest"]): entry["_newest"], entry["newest_pass"] = loaded, live_passes[r.pass_id].name if r.status == "parked": entry["parked"] += 1 continue if r.status == "stale": entry["stale"] += 1 continue entry["open"] += 1 entry["spot_check"] += int(bool(r.spot_check)) if r.against_decision_id is not None: entry["held"] += 1 if _is_merge(r): for side in _variant_sides(r): entry["gain"].add(side) brands = [b for b in slug_to_brand.values() if b is not None] listings = dict(db.execute(select(ProductVariant.brand_id, func.count(Listing.id)).join(Listing, Listing.variant_id == ProductVariant.id) .where(ProductVariant.brand_id.in_([b.id for b in brands] or [0])).group_by(ProductVariant.brand_id)).all()) gtins = {} for entry in by_brand.values(): keys = [k for k in entry["gain"] if k.startswith("variant:")] uids = [] for k in keys: try: uids.append(_uuid.UUID(k[len("variant:"):])) except ValueError: continue barcoded = set(db.scalars(select(ProductVariant.uid).where(ProductVariant.uid.in_(uids), ProductVariant.gtin.isnot(None)))) if uids else set() gtins[entry["brand_slug"]] = len(uids) - len(barcoded) out = [] for slug, entry in by_brand.items(): brand = next((b for b in brands if b.slug == slug), None) out.append({"brand_slug": slug, "brand_name": entry["brand_name"], "open": entry["open"], "spot_check": entry["spot_check"], "held": entry["held"], "parked": entry["parked"], "stale": entry["stale"], "comparison_gain": gtins.get(slug, 0), "listings": int(listings.get(brand.id, 0)) if brand else 0, "newest_pass": entry["newest_pass"], "passes": sorted(entry["passes"])}) out.sort(key=lambda e: (-e["comparison_gain"], -e["listings"], e["brand_slug"])) return out def hints(db: Session, brand_slug: str, *, rule: str | None = None, limit: int = 500) -> dict[str, Any]: """What every word list would have suggested for one brand, and why: the AI pass's input. This is the read that replaces the word lists' 938 rows in rian's queue (K11.2). The rows are not deleted and this does not regenerate them -- `proposal_rules.generate` already wrote one pass per list, and withdrawing a pass keeps every row. So a pass reading a brand asks here what the lists saw, with the words each one matched and the reason it gave, and decides for itself whether any of it is right. Nothing here is a question for a person and nothing is applied. Withdrawn passes are included deliberately: withdrawal takes a list out of the human queue, not out of the pass's evidence. Writes nothing. """ brand, family = _brand_family(db, brand_slug) if brand is None: raise Refused("BRAND_MISSING", f"No brand {brand_slug!r}.") slugs = [b.slug for b in family] passes = {p.id: p for p in db.scalars(select(ProposalPass).where(ProposalPass.kind.notin_(REVIEW_KINDS)))} if not passes: return {"brand": {"slug": brand.slug, "name": brand.name}, "lists": [], "total": 0} rows = list(db.scalars(select(Proposal).where(Proposal.brand_slug.in_(slugs), Proposal.pass_id.in_(list(passes))) .order_by(Proposal.pass_id, Proposal.position, Proposal.id))) # A grouping list's reading is always about TWO product lines, and a pass cannot weigh # "these two read as one" without knowing which two: the reason alone says neither. Resolved # in two queries for the whole brand, as the sheet does (`_side_names`). names = _side_names(db, rows) by_list: dict[str, dict] = {} for r in rows: source = passes[r.pass_id] # `rule::` -- the list is what a pass wants to know it came from. parts = source.name.split(":") name = parts[1] if len(parts) >= 2 and parts[0] == "rule" else source.name if rule is not None and name != rule: continue entry = by_list.setdefault(name, {"list": name, "pass_name": source.name, "kind": source.kind, "withdrawn": source.withdrawn_at is not None, "withdrawn_reason": source.withdrawn_reason, "rules_version": source.rules_version, "count": 0, "suggestions": []}) entry["count"] += 1 if len(entry["suggestions"]) >= limit: continue evidence = r.evidence if isinstance(r.evidence, list) else ([r.evidence] if r.evidence else []) entry["suggestions"].append({ "uid": str(r.uid), "entity_type": r.entity_type, "natural_key": r.natural_key, "field": r.field, "would_suggest": jsonable(r.value), "current_value": jsonable(r.current_value), "reason": r.reason, "confidence": r.confidence, "status": r.status, "matched": [e.get("matched") for e in evidence if isinstance(e, dict) and e.get("matched")], "read_from": [e.get("text") for e in evidence if isinstance(e, dict) and e.get("text")], "sides": [names[k] for k in _variant_sides(r) if k in names], }) out = sorted(by_list.values(), key=lambda e: (-e["count"], e["list"])) return {"brand": {"slug": brand.slug, "name": brand.name, "aliases": sorted(b.slug for b in family if b.id != brand.id)}, "lists": out, "total": sum(e["count"] for e in out)} def deferred_notes(db: Session, brand_slug: str | None = None, *, field: str | None = None, natural_key: str | None = None, limit: int = 200) -> dict[str, Any]: """Every note rian left when he deferred a question, for the pass about to ask it again (K11.4). This is the loop that turns his answers into the reasoning rules. Without it every pass starts from zero and the set/coffret question is asked identically forever -- a judgement he has NOT settled had nowhere to live but "no", and the sentence on a reject was never read by anything. `brand_slug` narrows to one brand's notes (its aliases included); `field` and `natural_key` narrow to the question in front of the pass, which is how a pass asking about one product line finds what it was told about that line last time. With no argument at all it answers every deferred note, which is what a pass reads before it proposes across a brand it has not seen. A rejection's sentence rides along, marked by its status: "no, because" is reasoning too, and the standing rule is that a pair a person kept separate is never proposed as one again. `wants` carries the one thing a pass most needs to know: whether the note asks for something ELSE or only for later. Answering "suggest something else" writes that marker on the proposal, but the notes this returns dropped it, so the two were indistinguishable to the pass that read them -- and a pass told to come back with a different suggestion would have re-proposed the same one. Writes nothing. """ slugs: list[str] | None = None brand = None if brand_slug is not None: brand, family = _brand_family(db, brand_slug) if brand is None: raise Refused("BRAND_MISSING", f"No brand {brand_slug!r}.") slugs = [b.slug for b in family] q = select(Proposal).where(Proposal.status.in_(("deferred", "rejected")), Proposal.resolution_note.isnot(None)) if slugs is not None: q = q.where(Proposal.brand_slug.in_(slugs)) if field is not None: q = q.where(Proposal.field == field) if natural_key is not None: q = q.where(Proposal.natural_key == natural_key) rows = list(db.scalars(q.order_by(Proposal.resolved_at.desc().nullslast(), Proposal.id.desc()).limit(limit))) passes = {p.id: p for p in db.scalars(select(ProposalPass).where(ProposalPass.id.in_({r.pass_id for r in rows} or {0})))} who = {a.id: a.username for a in db.scalars(select(Account).where(Account.id.in_({r.resolved_by for r in rows if r.resolved_by} or {0})))} names = _side_names(db, rows) out = [] for r in rows: out.append({"uid": str(r.uid), "status": r.status, "note": r.resolution_note, "wants": (r.detail or {}).get("wants"), "brand_slug": r.brand_slug, "entity_type": r.entity_type, "natural_key": r.natural_key, "field": r.field, "proposed": jsonable(r.value), "sheet_line_ref": r.sheet_line_ref, "by": who.get(r.resolved_by), "at": r.resolved_at, "pass_name": passes[r.pass_id].name if r.pass_id in passes else None, "sides": [names[k] for k in _variant_sides(r) if k in names]}) return {"brand": {"slug": brand.slug, "name": brand.name} if brand is not None else None, "notes": out, "deferred": sum(1 for r in out if r["status"] == "deferred"), "total": len(out)} def _variant_sides(p) -> list[str]: keys = [] if p.entity_type == "suggestion": detail = p.natural_key_detail or {} keys = [detail.get("left"), detail.get("right")] if not all(keys): try: _, parts = natural_keys.parse(p.natural_key) keys = [parts["left"], parts["right"]] except ValueError: keys = [] elif p.field == "merged_into": keys = [p.natural_key, p.value if isinstance(p.value, str) else None] return [k for k in keys if isinstance(k, str)] def _is_brand_fold(p: Proposal) -> bool: """A proposal that two brand rows are one. It is the outer question on a sheet: everything else on the brand assumes the brands as they stand (rian, 18 Sep).""" return p.entity_type == "suggestion" and p.field == "decision" and p.natural_key.startswith("pair:brand:") def brand_fold_rows(db: Session, slugs: list[str]) -> list[Proposal]: """Every open brand fold naming one of these brands on EITHER side. A fold is filed under one brand's slug, so it used to appear on that brand's sheet and nowhere else: the other half of the question was invisible from the other half of the pair. It belongs to both (rian: "The same brand fold suggestion should be on all relevant brands").""" live = select(ProposalPass.id).where(ProposalPass.withdrawn_at.is_(None), ProposalPass.kind.in_(REVIEW_KINDS)) out = [] for p in db.scalars(select(Proposal).where(Proposal.pass_id.in_(live), Proposal.status == "open", Proposal.entity_type == "suggestion", Proposal.natural_key.like("pair:brand:%"))): sides = set(_variant_sides(p)) if any(f"brand:{slug}" in sides for slug in slugs): out.append(p) return out def _side_names(db: Session, rows: list[Proposal]) -> dict[str, dict[str, Any]]: """Every pair side and merge target on the sheet, resolved to the name a person reads. A pair row asks "are these two the same?". Without both names ON the row the question cannot be answered at all: the sheet showed the pair's natural key and left a person comparing two uuids (rian, 17 Sep, reviewing Lancome: "product line 1196afb2 is not useful"). Batched by uid, so a sheet of sixteen pairs costs two queries rather than thirty-two. """ keys: dict[str, tuple[str, str]] = {} for p in rows: for key in _variant_sides(p): try: kind, parts = natural_keys.parse(key) except ValueError: continue if kind in ("product_line", "product_variant") and parts.get("uid"): keys[key] = (kind, parts["uid"]) elif kind == "brand" and parts.get("slug"): # A BRAND pair is the flagship case of REVIEW-PROCESS section 2.2 (a rename no rule # can find), and it reached the page as `pair:brand:brand:a||brand:b` because this # resolver knew only uid-keyed rows. Found by the first session pass, on Rabanne. keys[key] = ("brand", parts["slug"]) found: dict[tuple[str, str], str] = {} slugs = {uid for k, (kind, uid) in keys.items() if kind == "brand"} if slugs: for row in db.scalars(select(Brand).where(Brand.slug.in_(slugs))): found[("brand", row.slug)] = row.name for kind, model in (("product_line", ProductLine), ("product_variant", ProductVariant)): uids = set() for key, (k, uid) in keys.items(): if k != kind: continue try: uids.add(_uuid.UUID(str(uid))) except (ValueError, AttributeError, TypeError): continue if not uids: continue for row in db.scalars(select(model).where(model.uid.in_(uids))): found[(kind, str(row.uid))] = row.name return {key: {"key": key, "kind": kind, "name": found.get((kind, uid))} for key, (kind, uid) in keys.items()} def _passes_of(db: Session, slugs: list[str]) -> list[ProposalPass]: ids = set(db.scalars(select(Proposal.pass_id).where(Proposal.brand_slug.in_(slugs)).distinct())) return list(db.scalars(select(ProposalPass).where(ProposalPass.id.in_(ids or {0})).order_by(ProposalPass.loaded_at.desc(), ProposalPass.id.desc()))) def brand_batches(db: Session, slugs: list[str]) -> list[dict[str, Any]]: """The sheet's history: every batch whose scope names the brand, newest first, undone or not.""" out = [] candidates = db.scalars(select(DecisionBatch).where(DecisionBatch.kind == "sheet").order_by(DecisionBatch.opened_at.desc(), DecisionBatch.id.desc())) for b in candidates: if (b.scope or {}).get("brand_slug") not in slugs: continue undo = db.scalar(select(DecisionBatch).where(DecisionBatch.reverses_batch_id == b.id)) p = db.get(ProposalPass, b.pass_id) if b.pass_id else None summary = b.summary or {} out.append({"uid": str(b.uid), "opened_at": b.opened_at, "by": b.by_username, "mode": b.mode, "kind": b.kind, "pass_name": p.name if p else None, "note": b.note, "scope": b.scope, "decisions": int(summary.get("decisions", 0)), "held": int(summary.get("held", 0)), "refused": int(summary.get("refused", 0)), "undone": undo is not None, "undone_by_uid": str(undo.uid) if undo is not None else None, "undone_at": undo.opened_at if undo is not None else None}) return out def sheet(db: Session, brand_slug: str, pass_name: str | None = None) -> dict[str, Any]: """One brand's sheet (spec §6.3), for one pass (the newest by default). Rows grouped by the proposed product line in the pass's own order, brand rows first; each cell with the proposed, corrected, rule and current values (current re-read now, drift against the load), confidence, evidence re-read now (staleness), the spot-check mark and the effective decision computed now, with the disagreement flag. Writes nothing.""" brand, family = _brand_family(db, brand_slug) if brand is None: raise Refused("BRAND_MISSING", f"No brand {brand_slug!r}.") slugs = [b.slug for b in family] passes = _passes_of(db, slugs) # A brand's work is ONE queue. Each word list loads as its own pass, so showing the newest # pass alone showed a fraction of the brand and hid the rest behind a selector: one brand's # nineteen waiting rows arrived as three passes and the sheet drew three of them, while the # index (which counts every pass) said nineteen. A person cannot review what the page does # not draw, so with no pass named EVERY live pass for the brand is shown, and naming one # narrows to it. chosen = next((p for p in passes if p.name == pass_name), None) if pass_name else None if pass_name and chosen is None: raise Refused("PASS_NOT_FOUND", f"No pass {pass_name!r} for {brand.slug}.") # ...and only a review kind: a word list is never a question for a person (`REVIEW_KINDS`). # Naming a pass explicitly still draws it, so a withdrawn rule pass stays readable on purpose. shown = [chosen] if chosen is not None else [p for p in passes if p.withdrawn_at is None and p.kind in REVIEW_KINDS] result: dict[str, Any] = { "brand": {"slug": brand.slug, "name": brand.name, "aliases": sorted(b.slug for b in family if b.id != brand.id)}, "passes": [{"name": p.name, "kind": p.kind, "process_version": p.process_version, "rules_version": p.rules_version, "loaded_at": p.loaded_at, "loaded_by": p.loaded_by_username, "withdrawn": p.withdrawn_at is not None, "counts": p.counts or {}} for p in passes], "pass_name": chosen.name if chosen else None, "counts": {}, "spot_check_uids": [], "groups": [], "brand_gate": None, "batches": brand_batches(db, slugs), } # A brand fold is filed under one brand's slug, so the OTHER brand of the pair has no pass of # its own and returned an empty sheet: the half of the question filed elsewhere was invisible # from the half it is about. Its pass joins this brand's (rian, 18 Sep). if chosen is None: extra = {f.pass_id for f in brand_fold_rows(db, slugs)} - {p.id for p in shown} if extra: more = list(db.scalars(select(ProposalPass).where(ProposalPass.id.in_(extra)))) shown = shown + more result["passes"] = result["passes"] + [ {"name": p.name, "kind": p.kind, "process_version": p.process_version, "rules_version": p.rules_version, "loaded_at": p.loaded_at, "loaded_by": p.loaded_by_username, "withdrawn": p.withdrawn_at is not None, "counts": p.counts or {}} for p in more if p.name not in {q["name"] for q in result["passes"]}] if not shown: return result shown_ids = [p.id for p in shown] rows = list(db.scalars(select(Proposal).where(Proposal.pass_id.in_(shown_ids), Proposal.brand_slug.in_(slugs)) .order_by(Proposal.pass_id, Proposal.position, Proposal.id))) # A brand fold belongs to BOTH brands it names, whichever one's file carried it. have = {r.id for r in rows} rows = [r for r in brand_fold_rows(db, slugs) if r.id not in have and r.pass_id in shown_ids] + rows all_passes = {p.id: p for p in db.scalars(select(ProposalPass))} side_names = _side_names(db, rows) from app.services import attention as attention_service, review_detail levels = attention_service.for_rows(db, rows) cache = ListingCache(db) cache.warm([ev.get("listing") for r in rows for ev in (r.evidence or []) if ev.get("listing")] + [k for r in rows for k in ((r.natural_key_detail or {}).get("listings") or [])]) decided: dict[tuple[str, int, str], Decision] = {} by_type: dict[str, set[int]] = defaultdict(set) for r in rows: if r.entity_id is not None: by_type[r.entity_type].add(r.entity_id) fields = {r.field for r in rows} for et, ids in by_type.items(): for (eid, f), dec in effective(db, et, ids, fields).items(): decided[(et, eid, f)] = dec batch_ids = {d.batch_id for d in decided.values()} | {d.batch_id for d in db.scalars(select(Decision).where(Decision.id.in_( [r.decision_id for r in rows if r.decision_id] or [0])))} batch_uids = {b.id: str(b.uid) for b in db.scalars(select(DecisionBatch).where(DecisionBatch.id.in_(batch_ids or {0})))} shops = {s.id: s for s in db.scalars(select(Shop))} counts = Counter() groups: dict[tuple[int, str | None], dict] = {} order: list[tuple[int, str | None]] = [] for r in rows: entity = _entity(db, r.entity_type, r.entity_id) now = column_value(db, r.entity_type, entity, r.field) if entity is not None else None ok, why, spans = evidence_state(cache, r.evidence) if r.status == "open" else (True, None, evidence_state(cache, r.evidence)[2]) dec = decided.get((r.entity_type, r.entity_id, r.field)) if r.entity_id is not None else None eff = _decision_out(dec, all_passes, batch_uids) value_now = r.corrected_value if r.corrected_value is not None else r.value disagrees = dec is not None and dec.mode == "bulk" and dec.pass_id != r.pass_id and canon(dec.value) != canon(value_now) held = dec is not None and dec.effect == "set" and canon(dec.value) != canon(value_now) and (dec.origin == "person" or dec.mode == "individual") listing_rows = [] for key in (r.natural_key_detail or {}).get("listings") or []: listing = cache.get(key) shop = shops.get(listing.shop_id) if listing is not None else None listing_rows.append({"key": key, "shop": shop.code if shop else None, "shop_name": shop.name if shop else None, "listed_name": listing.listed_name if listing else None, "listed_variant": listing.listed_variant if listing else None, "listed_quantity_text": listing.listed_quantity_text if listing else None, "gone": listing is None, "ignored": bool(listing is not None and listing.ignored_at is not None)}) reopened = r.status in ("approved", "rejected") and r.decision_id is not None and waiting(db, r) status = "open" if reopened else r.status if reopened: ok, why, spans = evidence_state(cache, r.evidence) stale_now = r.status == "stale" or (status == "open" and not ok) row_pass = all_passes[r.pass_id].name if r.pass_id in all_passes else None out = { "uid": str(r.uid), "entity_type": r.entity_type, "natural_key": r.natural_key, "entity_id": r.entity_id, "resolution": r.resolution, "field": r.field, "proposed": r.value, "corrected": r.corrected_value, "rule_value": r.rule_value, "current_value": now, "loaded_value": r.current_value, # Drift is a change a person has not made here: an approved row's column moved because of the approval. "drift": status == "open" and entity is not None and canon(now) != canon(r.current_value), "confidence": float(r.confidence) if r.confidence is not None else None, "reason": r.reason, "evidence": spans, "spot_check": bool(r.spot_check), "status": status, "stale_now": stale_now, "reopened": reopened, "stale_why": why if status == "open" else (r.resolution_note if r.status == "stale" else None), "resolution_note": r.resolution_note, "listings": listing_rows, "effective": eff, "disagrees_with_bulk": disagrees, # What the row is ABOUT, so the question can be read without decoding a key: the row's # own entity, and for a pair or a merge both sides by name (REVIEW-PROCESS.md §3). "subject_name": getattr(entity, "name", None), "sides": [side_names[k] for k in _variant_sides(r) if k in side_names], "pass_name": row_pass, "held": bool(held and status == "open"), "previously_rejected": (r.detail or {}).get("previously_rejected"), "generator": r.generator, "collapsed": r.generator == "arrival" or bool((r.detail or {}).get("previously_rejected")), # A rejected fold offers the split instead of dying (K9); it acts only when a person runs it. # The ROW's pass, never the sheet's: the sheet shows every live pass at once, so # `chosen` is None on the whole-brand view and the offer read it as a name. "split_offer": split_offer(db, r, row_pass) if status == "rejected" else None, "decision_id": r.decision_id, "position": r.position, "sheet_line_ref": r.sheet_line_ref, "is_header": r.entity_type == "product_line" and r.field == "name", } out["is_brand_fold"] = _is_brand_fold(r) out["blocked_by_brand_fold"] = False # How much of a person this row needs, which is not how sure the pass was. out["attention"] = levels.get(r.id) out["category"] = review_detail.category_of(r) out["category_label"] = review_detail.CATEGORIES.get(out["category"], "Waiting for an answer") counts[status] += 1 if status == "open": counts["spot_check"] += int(bool(r.spot_check)) counts["held"] += int(out["held"]) counts["stale_now"] += int(stale_now) if r.spot_check: result["spot_check_uids"].append(out["uid"]) ref = None if r.entity_type == "brand" else (r.sheet_line_ref or (r.natural_key if out["is_header"] else None)) # Keyed by the pass too: several passes are shown at once and a rule's rows all carry a # null line ref, so keying by the ref alone piled every list's rows into one card with # nothing saying which list asked what, and one approval button that fits only one pass. key = (r.pass_id, ref) if key not in groups: groups[key] = {"ref": ref, "pass_name": out["pass_name"], "name": None, "header_uid": None, "absorbs": [], "rows": []} order.append(key) group = groups[key] if out["is_header"] and (r.sheet_line_ref in (None, r.natural_key)): group["name"], group["header_uid"] = (r.corrected_value or r.value), out["uid"] for key in sorted(set((r.natural_key_detail or {}).get("absorbs") or []) | set((r.detail or {}).get("absorbs_derived") or [])): line = natural_keys.resolve(db, "product_line", key, None) if isinstance(line, natural_keys.Parked): group["absorbs"].append({"key": key, "name": None, "slug": None, "members": 0, "aliased": False, "missing": True}) continue members = db.scalar(select(func.count(ProductVariant.id)).where(ProductVariant.product_line_id == line.id, ProductVariant.merged_into_id.is_(None))) or 0 group["absorbs"].append({"key": key, "name": line.name, "slug": line.slug, "members": int(members), "aliased": line.alias_of_id is not None, "missing": False}) group["rows"].append(out) # Brand rows first, then each pass in the order its rows were written; biggest queue first so # a person meets the bulk of the brand's work before its tail. ordered = sorted(order, key=lambda k: (k[1] is not None, -len(groups[k]["rows"]), min((row["position"] for row in groups[k]["rows"]), default=0))) result["groups"] = [groups[k] for k in ordered] result["counts"] = dict(counts) result["attention_counts"] = attention_service.summary( {r.id: levels[r.id] for r in rows if r.id in levels and r.status == "open"}) # The brand fold is answered first. Until it is, every other row on the sheet is drawn and # none of them can be decided: they all assume the brands as they stand, and the fold is what # decides whether that assumption holds (rian, 18 Sep). gate = [r for g in result["groups"] for r in g["rows"] if r["is_brand_fold"] and r["status"] == "open"] if gate: for g in result["groups"]: for row in g["rows"]: row["blocked_by_brand_fold"] = not row["is_brand_fold"] and row["status"] == "open" result["brand_gate"] = { "uids": [r["uid"] for r in gate], "sides": gate[0]["sides"], "blocked": sum(1 for g in result["groups"] for r in g["rows"] if r.get("blocked_by_brand_fold")), } return result def line_table(db: Session, brand_slug: str, line_ref: str, pass_name: str | None = None) -> dict[str, Any]: """What a pass emits, as rian described it: ONE proposed product line, and beneath it every listing that would sit in it with the attribute population proposed for each (K11.5). His words (walk-through W10): *"the AI pass would present a recommendation that we have one product line called Rouge Allure. then below that it lists all the listings in a table with proposed attribute population. The human scans through it, makes any corrections, and approves."* And W11: grouping and attributes are ONE job, so a table that shows the grouping without the attributes is only half the review. A view, not a store: every value here is already a `proposals` row, read back grouped by the thing a person actually decides about. The columns are the attribute kinds that VARY inside the line -- the quantity variations and the shades -- because a kind with one value everywhere tells a reviewer nothing. Each proposed fold is called out with the reason it gave, and the shops carry their listed name, their price and a link to the source so a spot-check is one click. Writes nothing. """ data = sheet(db, brand_slug, pass_name) group = next((g for g in data["groups"] if g["ref"] == line_ref), None) if group is None: raise Refused("LINE_NOT_FOUND", f"No product line {line_ref!r} on {brand_slug}'s sheet.") header = next((r for r in group["rows"] if r.get("is_header")), None) rows = group["rows"] # One entry per variant the line would hold, keyed by the proposal's natural key. variants: dict[str, dict] = {} def slot(key: str, subject: str | None) -> dict: return variants.setdefault(key, {"key": key, "name": subject, "membership": None, "attributes": {}, "listings": [], "rows": []}) folds: list[dict] = [] for r in rows: if r.get("is_header"): continue cell = {"uid": r["uid"], "field": r["field"], "proposed": r["proposed"], "corrected": r["corrected"], "current": r["current_value"], "confidence": r["confidence"], "reason": r["reason"], "status": r["status"], "spot_check": r["spot_check"], "stale_now": r["stale_now"], "held": r["held"], "disagrees_with_bulk": r["disagrees_with_bulk"], "evidence": r["evidence"]} # A fold is a claim about two things, so it is called out on its own with both sides named. if r["entity_type"] == "suggestion" or r["field"] in ("merged_into", "alias_of"): folds.append({**cell, "kind": r["field"], "sides": r["sides"], "natural_key": r["natural_key"]}) continue entry = slot(r["natural_key"], r.get("subject_name")) entry["rows"].append(cell) if r["field"] == "product_line": entry["membership"] = cell elif r["field"].startswith("attribute:"): entry["attributes"][r["field"].split(":", 1)[1]] = cell elif r["field"] in ("quantity", "name"): entry["attributes"][r["field"]] = cell for listing in r["listings"]: if listing["key"] not in {existing["key"] for existing in entry["listings"]}: entry["listings"].append(dict(listing)) # The shops, their prices and a link to the source: a spot-check is one click, never a retype. keys = [listing["key"] for entry in variants.values() for listing in entry["listings"]] cache = ListingCache(db) cache.warm(keys) shops = {s.id: s for s in db.scalars(select(Shop))} listing_ids = {} for key in keys: row = cache.get(key) if row is not None: listing_ids[key] = row.id newest: dict[int, Any] = {} if listing_ids: from app.models import PriceObservation for obs in db.scalars(select(PriceObservation).where(PriceObservation.listing_id.in_(set(listing_ids.values()))) .order_by(PriceObservation.listing_id, PriceObservation.observed_at.desc())): newest.setdefault(obs.listing_id, obs) prices: list[float] = [] for entry in variants.values(): for listing in entry["listings"]: row = cache.get(listing["key"]) shop = shops.get(row.shop_id) if row is not None else None obs = newest.get(row.id) if row is not None else None listing["url"] = row.url if row is not None else None listing["iata"] = shop.iata if shop is not None else None listing["quantity_text"] = listing.get("listed_quantity_text") listing["price"] = ({"price": float(obs.price) if obs.price is not None else None, "currency": obs.currency, "price_usd": float(obs.price_usd) if obs.price_usd is not None else None, "was_price": float(obs.was_price) if obs.was_price is not None else None, "observed_at": obs.observed_at, "in_stock": obs.in_stock} if obs is not None else None) if obs is not None and obs.price_usd is not None: prices.append(float(obs.price_usd)) # Only the kinds that VARY inside the line are columns: one value everywhere is not a variation. kinds: dict[str, set] = defaultdict(set) for entry in variants.values(): for kind, cell in entry["attributes"].items(): kinds[kind].add(canon(cell["corrected"] if cell["corrected"] is not None else cell["proposed"])) columns = sorted(k for k, values in kinds.items() if len(values) > 1) ordered = sorted(variants.values(), key=lambda e: (e["name"] or "", e["key"])) return { "brand": data["brand"], "pass_name": data["pass_name"], "ref": line_ref, "line": {"ref": line_ref, "header_uid": group.get("header_uid"), "name": group.get("name"), "proposed_name": header["proposed"] if header else None, "current_name": header["current_value"] if header else None, "confidence": header["confidence"] if header else None, "reason": header["reason"] if header else None, "status": header["status"] if header else None, "spot_check": header["spot_check"] if header else False, "evidence": header["evidence"] if header else [], "absorbs": group.get("absorbs") or []}, "columns": columns, "variants": ordered, "folds": folds, "prices": {"count": len(prices), "low_usd": min(prices) if prices else None, "high_usd": max(prices) if prices else None, "shops": len({listing.get("shop") for e in ordered for listing in e["listings"] if listing.get("shop")})}, "counts": {"variants": len(ordered), "listings": sum(len(e["listings"]) for e in ordered), "folds": len(folds), "open": sum(1 for e in ordered for c in e["rows"] if c["status"] == "open") + len([f for f in folds if f["status"] == "open"]), "spot_check": sum(1 for e in ordered for c in e["rows"] if c["spot_check"]) + len([f for f in folds if f["spot_check"]])}, } def batch_detail(db: Session, uid: str) -> dict[str, Any]: try: b = db.scalar(select(DecisionBatch).where(DecisionBatch.uid == _uuid.UUID(str(uid)))) except ValueError: b = None if b is None: raise Refused("ENTITY_NOT_FOUND", f"No batch {uid}.") undo = db.scalar(select(DecisionBatch).where(DecisionBatch.reverses_batch_id == b.id)) p = db.get(ProposalPass, b.pass_id) if b.pass_id else None reverses = db.get(DecisionBatch, b.reverses_batch_id) if b.reverses_batch_id else None rows = list(db.scalars(select(Decision).where(Decision.batch_id == b.id).order_by(Decision.id))) return {"uid": str(b.uid), "kind": b.kind, "mode": b.mode, "by": b.by_username, "opened_at": b.opened_at, "closed_at": b.closed_at, "pass_name": p.name if p else None, "scope": b.scope, "note": b.note, "summary": b.summary or {}, "undone": undo is not None, "undone_by_uid": str(undo.uid) if undo else None, "reverses_uid": str(reverses.uid) if reverses else None, "decisions": [decision_out(db, d, {b.id: str(b.uid)}) for d in rows]} def decision_out(db: Session, d: Decision, batch_uids: dict[int, str] | None = None) -> dict[str, Any]: batch_uids = batch_uids or {} if d.batch_id not in batch_uids: b = db.get(DecisionBatch, d.batch_id) batch_uids[d.batch_id] = str(b.uid) if b else None p = db.get(ProposalPass, d.pass_id) if d.pass_id else None return {"id": d.id, "uid": str(d.uid), "decided_at": d.decided_at, "entity_type": d.entity_type, "entity_id": d.entity_id, "natural_key": d.natural_key, "field": d.field, "effect": d.effect, "value": d.value, "rule_value": d.rule_value, "prior_value": d.prior_value, "origin": d.origin, "mode": d.mode, "by": d.decided_by_username, "reason": d.reason, "pass_name": p.name if p else None, "batch_uid": batch_uids.get(d.batch_id), "reverses_id": d.reverses_id, "restores_id": d.restores_id, "caused_by_id": d.caused_by_id, "replayed_from": d.replayed_from} def decisions_for(db: Session, entity_type: str, entity_id: int, limit: int = 200) -> list[dict[str, Any]]: """The ledger for one entity, newest first (K2's `decisions list --entity` as JSON).""" rows = list(db.scalars(select(Decision).where(Decision.entity_type == entity_type, Decision.entity_id == entity_id) .order_by(Decision.decided_at.desc(), Decision.id.desc()).limit(limit))) uids = {b.id: str(b.uid) for b in db.scalars(select(DecisionBatch).where(DecisionBatch.id.in_({r.batch_id for r in rows} or {0})))} return [decision_out(db, r, uids) for r in rows] # --------------------------------------------------------------------------- the approval def _phase(p: Proposal) -> int: """Spec §6.4 step 4: brand rows; line headers; memberships; attribute and name cells; the absorbed lines' aliases (derived, after this sort); merges and pairs; pins and ignores.""" if p.entity_type == "brand": return 0 if p.entity_type == "product_line" and p.field == "name": return 1 if p.entity_type == "product_variant" and p.field == "product_line": return 2 if p.entity_type == "product_line" and p.field == "alias_of": return 4 if p.entity_type == "suggestion" or p.field == "merged_into": return 5 if p.entity_type == "listing": return 6 return 3 def _held(db: Session, p: Proposal, value: Any) -> Decision | None: """The writer's human-first rule recomputed (spec §0 rule 5): the effective row a bulk act would not overrule. The writer checks again inside the batch; this only narrows `all`.""" if p.entity_id is None: return None dec = effective(db, p.entity_type, [p.entity_id], [p.field]).get((p.entity_id, p.field)) if dec is None or dec.effect != "set" or not (dec.origin == "person" or dec.mode == "individual"): return None if p.field in ("product_line", "alias_of", "merged_into", "pinned_to") and isinstance(value, str) and dec.value_ref_id is not None: kind = {"product_line": "product_line", "merged_into": "product_variant", "pinned_to": "product_variant"}.get(p.field, p.entity_type) target = natural_keys.resolve(db, kind, value, None) if not value.startswith("line:new:") else None if target is not None and not isinstance(target, natural_keys.Parked) and target.id == dec.value_ref_id: return None return dec if canon(dec.value) != canon(value) else None def _live_variant(db: Session, p: Proposal) -> ProductVariant: """A membership or cell re-resolved from its natural key, a merged variant followed to its survivor (spec §6.4 step 4), so a decision lands on the row that holds the listings now.""" found = natural_keys.resolve(db, "product_variant", p.natural_key, p.natural_key_detail) if isinstance(found, natural_keys.Parked): raise Refused(found.code, f"{p.natural_key}: {found.detail or found.code}") hops = 0 while found.merged_into_id is not None and hops < 8: found, hops = db.get(ProductVariant, found.merged_into_id), hops + 1 return found def _vertical_of(db: Session, family_ids: list[int]) -> str | None: row = db.execute(select(ProductVariant.vertical, func.count()).where(ProductVariant.brand_id.in_(family_ids), ProductVariant.merged_into_id.is_(None)) .group_by(ProductVariant.vertical).order_by(func.count().desc()).limit(1)).first() return row[0] if row else None def _adopt_or_mint(b, p: Proposal, name: str, brand: Brand, family_ids: list[int]) -> tuple[ProductLine, dict]: """A sheet line adopts before it mints (spec §2): the brand's line already holding the slug, else the brand's line holding the chosen name's rule key (its address moves to the slug, the old one redirected), else a new row. Whatever it lands on is keyed `decided:`, a namespace no rule produces, so an arrival never joins an approved line. A slug held by another brand's line refuses `SLUG_TAKEN` naming the holder.""" from app.services import product_lines, publish db = b.db if not p.natural_key.startswith("line:new:"): line = natural_keys.resolve(db, "product_line", p.natural_key, p.natural_key_detail) if isinstance(line, natural_keys.Parked): raise Refused(line.code, f"{p.natural_key}: {line.detail or line.code}") if line.brand_id not in family_ids: raise Refused("LINE_HOUSES_DIFFER", f"{line.slug} belongs to another brand.") old = line.key if not old.startswith("decided:"): line.key = f"decided:{line.slug}" db.flush() return line, {"adopted_from_key": old} slug = p.natural_key[len("line:new:"):] holder = db.scalar(select(ProductLine).where(ProductLine.slug == slug)) if holder is not None: if holder.brand_id not in family_ids: owner = db.get(Brand, holder.brand_id) raise Refused("SLUG_TAKEN", f"The address {slug} belongs to {owner.name if owner else 'another brand'}'s product line {holder.name!r}.") old = holder.key if not old.startswith("decided:"): holder.key = f"decided:{slug}" db.flush() return holder, {"adopted_from_key": old, "adopted": "slug"} rule_key = product_lines.product_line_key(name, brand=brand.name, vertical=_vertical_of(db, family_ids)) candidate = db.scalar(select(ProductLine).where(ProductLine.brand_id.in_(family_ids), ProductLine.key == rule_key, ProductLine.alias_of_id.is_(None))) if rule_key else None if candidate is not None: old_key, old_slug = candidate.key, candidate.slug candidate.key, candidate.slug = f"decided:{slug}", slug db.flush() publish.redirect_write(db, "product_line", old_slug, slug) return candidate, {"adopted_from_key": old_key, "adopted": "rule_key", "slug_was": old_slug} line = ProductLine(brand_id=brand.id, key=f"decided:{slug}", slug=slug, name=name[:200]) db.add(line) db.flush() if b.maps is not None: b.maps.remember_line(line) return line, {"minted": True} def _resolve_scope(rows: list[Proposal], scope, state: dict[str, str]) -> tuple[str, list[Proposal], dict]: by_uid = {str(r.uid): r for r in rows} if scope == "all": return "bulk", [r for r in rows if state[str(r.uid)] == "open" and not r.spot_check], {"proposal_uids": "all"} if isinstance(scope, dict) and scope.get("sheet_line_ref"): ref = scope["sheet_line_ref"] picked = [r for r in rows if state[str(r.uid)] == "open" and not r.spot_check and (r.sheet_line_ref == ref or r.natural_key == ref)] return "bulk", picked, {"sheet_line_ref": ref} if isinstance(scope, dict) and scope.get("proposal_uids") is not None: uids = [str(u) for u in scope["proposal_uids"]] unknown = [u for u in uids if u not in by_uid] if unknown: raise Refused("VALUE_INVALID", f"Not proposals of this sheet and pass: {', '.join(unknown[:5])}.") return "individual", [by_uid[u] for u in uids], {"proposal_uids": uids} if scope is None: return "individual", [], {"proposal_uids": []} raise Refused("VALUE_INVALID", "scope is \"all\", {sheet_line_ref} or {proposal_uids}.") def approve(db: Session, brand_slug: str, pass_name: str, *, scope, by, corrections: dict | None = None, reject: list | None = None, defer: list | None = None, counter: list | None = None, note: str | None = None, force: bool = False) -> dict[str, Any]: """The approval (spec §6.4): refuse while a collection runs; one batch (`sheet`; `bulk` for all or a line, `individual` for explicit uids), one brand, one transaction; the selection rule (`all` leaves out spot-checks, parked, stale and held rows); the order within the batch so each step's rows exist for the next; each row through the writer (a correction as a person's individual act), a refusal leaving that row open and named with its code; the absorbed lines aliased once they are empty; rejects (a pair's reject is Keep separate); the counts said loudly. Nothing is approved that the person did not select.""" from app.services.decisions import writer refuse_if_collecting(db, force) pass_row = db.scalar(select(ProposalPass).where(ProposalPass.name == pass_name)) if pass_row is None: raise Refused("PASS_NOT_FOUND", f"No pass {pass_name!r}.") if pass_row.withdrawn_at is not None: raise Refused("PASS_WITHDRAWN", f"Pass {pass_name} was withdrawn.") brand, family = _brand_family(db, brand_slug) if brand is None: raise Refused("BRAND_MISSING", f"No brand {brand_slug!r}.") user = _account(db, by) if user is None: raise Refused("ENTITY_NOT_FOUND", "An approval is a person's act: name an account.") slugs = [f.slug for f in family] family_ids = [f.id for f in family] rows = list(db.scalars(select(Proposal).where(Proposal.pass_id == pass_row.id, Proposal.brand_slug.in_(slugs)) .order_by(Proposal.position, Proposal.id))) by_uid = {str(r.uid): r for r in rows} corrections = {str(k): v for k, v in (corrections or {}).items()} reject = [str(u) for u in (reject or [])] # A counter-proposal is a defer that says what to do instead, so it rides the same path. counter = [str(u) for u in (counter or [])] defer = [str(u) for u in (defer or [])] + [u for u in counter if u not in (defer or [])] unknown = [u for u in list(corrections) + reject + defer if u not in by_uid] if unknown: raise Refused("VALUE_INVALID", f"Not proposals of this sheet and pass: {', '.join(unknown[:5])}.") if reject and not (note or "").strip(): raise Refused("VALUE_INVALID", "A reject says why in one sentence.") # The brand fold is answered first. Every other row assumes the brands as they stand, so while # a fold is open nothing else on the brand may be decided (rian, 18 Sep). Refused, not ignored. gate = [g for g in brand_fold_rows(db, slugs) if g.status == "open"] if gate and scope != "gate": gate_uids = {str(g.uid) for g in gate} asked = set(reject) | set(corrections) if isinstance(scope, dict) and scope.get("proposal_uids"): asked |= {str(u) for u in scope["proposal_uids"]} elif scope is not None: asked |= {""} if asked - gate_uids: sides = " and ".join(sorted({s for g in gate for s in _variant_sides(g)})) raise Refused("BRAND_FOLD_FIRST", f"Answer the brand question first: {sides}. Everything else on this brand " f"assumes the brands as they stand.") # The note is the whole point of a defer: it is what the next pass reads. A defer without one # is indistinguishable from silence and would ask the same question again forever. if defer and not (note or "").strip(): raise Refused("VALUE_INVALID", "A defer says why in one sentence: the note is what the next pass reads.") if counter and not (note or "").strip(): raise Refused("VALUE_INVALID", "Suggesting something else needs the something else: say what to do instead.") both = set(reject) & set(defer) if both: raise Refused("VALUE_INVALID", f"A proposal is rejected or deferred, not both: {', '.join(sorted(both)[:5])}.") # A proposal whose decision was undone since waits again (`waiting`), without a write on read. state = {str(r.uid): ("open" if r.status in ("approved", "rejected") and waiting(db, r) else r.status) for r in rows} mode, selected, scope_record = _resolve_scope(rows, scope, state) for uid in corrections: if by_uid[uid] not in selected: selected.append(by_uid[uid]) selected = [r for r in selected if str(r.uid) not in reject and str(r.uid) not in defer] cache = ListingCache(db) cache.warm(ev.get("listing") for r in selected + [by_uid[u] for u in reject] for ev in (r.evidence or []) if ev.get("listing")) counts = Counter({"approved": 0, "held": 0, "refused": 0, "stale": 0, "parked": 0, "rejected": 0, "deferred": 0}) refusals: list[dict] = [] held: list[dict] = [] def refuse(p: Proposal, code: str, summary: str, bucket: str = "refused") -> None: counts[bucket] += 1 item = {"uid": str(p.uid), "natural_key": p.natural_key, "field": p.field, "code": code, "summary": summary} (held if bucket == "held" else refusals).append(item) ready: list[Proposal] = [] for p in selected: if p.status == "parked": refuse(p, "PARKED", p.resolution_note or "unresolvable on this host", "parked") continue if state[str(p.uid)] != "open": refuse(p, f"STATUS_{p.status.upper()}", f"this proposal is {p.status}", "skipped") continue ok, why, _ = evidence_state(cache, p.evidence) if not ok: p.status, p.resolution_note = "stale", f"EVIDENCE_CHANGED: {why}" refuse(p, "STALE", why, "stale") continue value = corrections.get(str(p.uid), p.value) if mode == "bulk" and str(p.uid) not in corrections: dec = _held(db, p, value) if dec is not None: refuse(p, "DECISION_HELD", f"{p.field} was decided by {dec.decided_by_username or 'a person'} ({dec.mode}); a bulk approval leaves it.", "held") continue ready.append(p) reject_rows = [by_uid[u] for u in reject] defer_rows = [by_uid[u] for u in defer] for r in reject_rows + defer_rows: if r.status not in OPEN_STATES and state[str(r.uid)] != "open": raise Refused("VALUE_INVALID", f"Proposal {r.uid} is {r.status}; only a waiting proposal is rejected or deferred.") def _defer_all() -> None: """A defer is never a decision: no ledger row, no application, just the note and the exit. `counter` is a defer that asks for something ELSE rather than for later (rian, 18 Sep: "Paco is its own brand, but the other two can be combined"). Same exit, same note, and a flag so the next pass can tell "not yet" from "do it differently" and come back with a changed proposal instead of the same one.""" for r in defer_rows: r.status, r.resolution_note, r.resolved_at, r.resolved_by = "deferred", note.strip(), now, user.id if str(r.uid) in counter: r.detail = {**(r.detail or {}), "wants": "a different suggestion"} counts["deferred"] += 1 headers_done = [r for r in rows if r.status == "approved" and state[str(r.uid)] == "approved" and r.entity_type == "product_line" and r.field == "name" and r.decision_id] needs_batch = bool(ready) or any(r.entity_type == "suggestion" for r in reject_rows) now = datetime.now(UTC) result: dict[str, Any] = {"brand": brand.slug, "pass_name": pass_name, "mode": mode, "batch_uid": None, "undo": None} if not needs_batch: for r in reject_rows: r.status, r.resolution_note, r.resolved_at, r.resolved_by = "rejected", note.strip(), now, user.id counts["rejected"] += 1 _defer_all() db.commit() return _response(result, counts, refusals, held, []) absorb_report: list[dict] = [] with writer.batch(db, "sheet", mode, user, scope={"brand_slug": brand.slug, **scope_record}, pass_id=pass_row.id, note=note) as b: lines_by_ref: dict[str, ProductLine] = {} refused_refs: dict[str, tuple[str, str]] = {} for h in headers_done: dec = db.get(Decision, h.decision_id) line = db.get(ProductLine, dec.entity_id) if dec is not None else None if line is not None: lines_by_ref[h.natural_key] = line ordered = sorted(ready, key=lambda r: (_phase(r), r.position, r.id)) absorbed = False for p in ordered + [None]: if not absorbed and (p is None or _phase(p) >= 5): # The absorbed lines' aliases after the memberships and cells, before the merges. absorb_report = _absorb(b, db, rows, lines_by_ref, family_ids) absorbed = True if p is None: break ref = p.sheet_line_ref if _phase(p) > 1 and ref in refused_refs: refuse(p, *refused_refs[ref]) continue corrected = str(p.uid) in corrections value = corrections[str(p.uid)] if corrected else p.value detail: dict[str, Any] = {"proposal_key": p.natural_key} if corrected: detail.update(corrected=True, proposed_value=p.value) savepoint = db.begin_nested() mark = len(b.decisions) touched = set(b.touched) try: entity: Any = p.natural_key if _phase(p) == 1: if not isinstance(value, str) or not value.strip(): raise Refused("VALUE_INVALID", "A product line's name is non-empty text.") entity, extra = _adopt_or_mint(b, p, value.strip(), brand, family_ids) detail.update(extra) elif p.entity_type == "product_variant": entity = _live_variant(db, p) if p.field == "product_line": _refuse_if_header_was_answered_otherwise(db, p, rows) if p.field == "product_line" and isinstance(value, str) and value.startswith("line:new:"): target = lines_by_ref.get(value) if target is None: raise Refused("LINE_NOT_APPROVED", f"Approve the product line {value} before its members.") value = target row = writer.record(b, p.entity_type, entity, p.field, value, origin="person" if corrected else "proposal", proposal=p, reason=p.reason, detail=detail, mode="individual" if corrected else None) savepoint.commit() except Refused as exc: savepoint.rollback() del b.decisions[mark:] b.touched = touched if _phase(p) == 1 and p.natural_key == (ref or p.natural_key): refused_refs[p.natural_key] = (exc.code, f"its product line was refused: {exc.summary}") refuse(p, exc.code, exc.summary, "held" if exc.code == "DECISION_HELD" else "refused") continue if _phase(p) == 1: lines_by_ref[p.natural_key] = entity if p.decision_id is not None: p.detail = {**(p.detail or {}), "undone_decision_ids": [*(p.detail or {}).get("undone_decision_ids", []), p.decision_id]} p.status, p.decision_id, p.resolved_at, p.resolved_by = "approved", row.id, now, user.id p.corrected_value = value if corrected and not isinstance(value, ProductLine) else p.corrected_value p.resolution_note = None counts["approved"] += 1 for r in reject_rows: if r.entity_type == "suggestion": savepoint = db.begin_nested() mark = len(b.decisions) try: row = writer.record(b, "suggestion", r.natural_key, "decision", {"decision": "separate", "survivor": None, "name": None, "note": note.strip()}, origin="person", mode="individual", proposal=r, reason=note.strip(), detail={"proposal_key": r.natural_key, "rejected_proposal": str(r.uid)}) savepoint.commit() r.decision_id = row.id except Refused as exc: savepoint.rollback() del b.decisions[mark:] refuse(r, exc.code, exc.summary) continue r.status, r.resolution_note, r.resolved_at, r.resolved_by = "rejected", note.strip(), now, user.id counts["rejected"] += 1 _defer_all() b.held.extend(held) b.refused.extend(refusals) result["batch_uid"] = str(b.uid) result["undo"] = f"python -m app.cli decisions undo-batch {result['batch_uid']} --reason \"\" --by {user.username}" folded = _collapse_after_brand_fold(db, selected, user) if folded: result.update(folded) return _response(result, counts, refusals, held, absorb_report) def _refuse_if_header_was_answered_otherwise(db: Session, member: Proposal, rows: list[Proposal]) -> None: """A member cannot be approved when its own header was kept separate or deferred. A header ("Million Gold absorbs One Million Gold") and the members under it (the two One Million Gold variants moving to Million Gold) are two routes to ONE outcome. Nothing tied them, so keeping the lines separate and then approving the members moved the very variants the rejection was about, and the fold happened by the back door with a rejection recorded beside it. `LINE_NOT_APPROVED` did not catch it: that guard only fires for a line that does not exist yet, and every target here already existed. Deferred counts too. A deferred question is one a person has not answered, and applying half its consequence is answering it for them. """ ref = member.sheet_line_ref if not ref: return header = next((r for r in rows if r.entity_type == "product_line" and r.field == "name" and r.natural_key == ref and r.id != member.id), None) if header is None or header.status not in ("rejected", "deferred"): return word = "kept separate" if header.status == "rejected" else "deferred" raise Refused("HEADER_ANSWERED_OTHERWISE", f"The product line this moves into was {word}, and this move would fold it anyway. " f"Answer the product line first, or undo that answer.") def _collapse_after_brand_fold(db: Session, rows: list[Proposal], user) -> dict[str, Any]: """After a brand fold is confirmed, the suggestions follow the brands (rian, 18 Sep). Keeping them apart would leave a sheet of questions filed under a brand that no longer exists. So the loser's open proposals are repointed at the survivor, two that say the same thing become one, and the caller is told where to send the person: the brand they chose. """ moved = duplicates = 0 survivor_slug = None for r in rows: if not (_is_brand_fold(r) and r.status == "approved"): continue value = r.corrected_value if r.corrected_value is not None else r.value if not isinstance(value, dict) or value.get("decision") != "same": continue sides = _variant_sides(r) keep = (value.get("survivor") or sides[0]).partition(":")[2] lose = [k.partition(":")[2] for k in sides if k.partition(":")[2] != keep] survivor_slug = keep for slug in lose: others = list(db.scalars(select(Proposal).where(Proposal.brand_slug == slug, Proposal.status == "open"))) here = {(p.entity_type, p.natural_key, p.field, canon(p.value)) for p in db.scalars(select(Proposal).where(Proposal.brand_slug == keep, Proposal.status == "open"))} for other in others: sig = (other.entity_type, other.natural_key, other.field, canon(other.value)) if sig in here: other.status, other.resolution_note = "withdrawn", f"the same suggestion already waits on {keep}" other.resolved_at, other.resolved_by = datetime.now(UTC), user.id duplicates += 1 continue here.add(sig) other.brand_slug = keep moved += 1 if survivor_slug is None: return {} db.flush() return {"brand_folded_into": survivor_slug, "proposals_moved": moved, "duplicates_merged": duplicates, "redirect_to": f"/review?tab=cleaning&brand={survivor_slug}"} # --------------------------------------------------------------------------- the split a rejection offers def split_offer(db: Session, row: Proposal, pass_name: str | None) -> dict[str, Any] | None: """What a REJECTED `rule:brand_trailers` proposal offers instead of dying. That proposal asks a person to confirm that the spellings on a brand row belong to it, and they sit there because `normalize.brand_key` dropped a trailing listed word. Rejecting it used to record a rejection and move nothing, which left the wrong fold exactly where it was; `undo.unmerge` works on variants only and the fold is not a decision, so there was no way back at all. Now the rejection carries the split the person can then run in one click (`POST /api/review/brands//split`, `services/brands.split_brand`). It is an OFFER, never an act: a rejection still splits nothing by itself. The spellings the proposal named are ticked, the rest of the row's spellings are offered beside them, and the name is the person's to type. """ from app.services import brands as brands_service # Any rejected brand-level proposal, whatever pass asked it. It was keyed to # `rule:brand_trailers` when that pass was on rian's sheet; K11.2 took the rule passes off # the sheet, so keeping the guard would have made the offer unreachable the moment the two # streams met. What it means is unchanged: you have said these spellings are not one brand, # so here is how to take them apart. if row.status != "rejected" or row.entity_type != "brand": return None brand = db.scalar(select(Brand).where(Brand.slug == (row.natural_key or "").partition(":")[2])) if brand is None or brand.alias_of_id is not None: return None named = {brands_service.spelling_key(ev.get("text")) for ev in (row.evidence or []) if ev.get("text")} held = brands_service.spellings_on(db, brand) if len(held) < 2: return None return { "brand_slug": brand.slug, "brand_name": brand.name, "spellings": [{"spelling": spelling, "variants": n, "folded": brands_service.spelling_key(spelling) in named, "words": proposal_rules.trailer_words(spelling)} for spelling, n in sorted(held.items(), key=lambda kv: (-kv[1], kv[0]))], } def run_split(db: Session, brand_slug: str, spellings: list[str], *, name: str, by, force: bool = False) -> dict[str, Any]: """The split a person runs from the sheet, after rejecting a fold (K9.4). One batch of one, `kind = route`, undoable like any other; `services/brands.split_brand` does the work and owns every refusal. Nothing here decides anything: the spellings and the name come from the person, and a rejection on its own never reaches this.""" from app.services import brands as brands_service from app.services.decisions import writer refuse_if_collecting(db, force) brand = db.scalar(select(Brand).where(Brand.slug == brand_slug)) if brand is None: raise Refused("BRAND_MISSING", f"No brand {brand_slug!r}.") user = _account(db, by) report = brands_service.split_brand(db, brand, spellings, new_name=name, decided_by=user) batch = db.get(DecisionBatch, db.get(Decision, report["decision_id"]).batch_id) return {**report, "batch_uid": str(batch.uid), "from_brand": brand.slug, "undo": f"python -m app.cli decisions undo-batch {batch.uid} --reason \"\" --by {user.username}", "message": f"{report['product_variants_moved']} product variant(s) moved to {report['brand_slug']}, " f"{report['lines_created']} product line(s) created, {report['lines_pruned']} pruned"} def _absorb(b, db: Session, rows: list[Proposal], lines_by_ref: dict[str, ProductLine], family_ids: list[int]) -> list[dict]: """Alias every line an approved header absorbs, once no live variant is left on it (spec §2: its address forwards and prune keeps it). A line still holding a member waits for the next approval, reported; it is never aliased with members the person has not approved.""" from app.services.decisions import writer report = [] for h in rows: if h.status != "approved" or h.entity_type != "product_line" or h.field != "name": continue target = lines_by_ref.get(h.natural_key) if target is None: continue keys = sorted(set((h.natural_key_detail or {}).get("absorbs") or []) | set((h.detail or {}).get("absorbs_derived") or [])) for key in keys: line = natural_keys.resolve(db, "product_line", key, None) if isinstance(line, natural_keys.Parked): report.append({"line": key, "state": "missing", "detail": line.code}) continue if line.id == target.id or line.alias_of_id == target.id: continue if line.brand_id not in family_ids: report.append({"line": key, "state": "refused", "detail": "LINE_HOUSES_DIFFER"}) continue left = db.scalar(select(func.count(ProductVariant.id)).where(ProductVariant.product_line_id == line.id, ProductVariant.merged_into_id.is_(None))) or 0 if left: report.append({"line": key, "state": "waiting", "detail": f"{left} live variant(s) not yet moved"}) continue savepoint = db.begin_nested() mark = len(b.decisions) try: writer.record(b, "product_line", line, "alias_of", target, origin="proposal", proposal=h, reason=f"absorbed into {target.name}", detail={"proposal_key": h.natural_key, "absorbed": key}) savepoint.commit() report.append({"line": key, "state": "aliased", "detail": target.slug}) except Refused as exc: savepoint.rollback() del b.decisions[mark:] report.append({"line": key, "state": "refused", "detail": exc.code}) return report def _response(result: dict, counts: Counter, refusals: list, held: list, absorbs: list) -> dict[str, Any]: counts = {k: int(v) for k, v in counts.items()} result.update(counts=counts, refusals=refusals, held=held, absorbs=absorbs, message=f"{counts.get('approved', 0)} approved, {counts.get('held', 0)} held, {counts.get('refused', 0)} refused" + (f", {counts['stale']} stale" if counts.get("stale") else "") + (f", {counts['parked']} parked" if counts.get("parked") else "") + (f", {counts['rejected']} rejected" if counts.get("rejected") else "")) return result # --------------------------------------------------------------------------- undo def undo_batch(db: Session, batch_uid: str, by, reason: str, *, force: bool = False) -> dict[str, Any]: """K2's batch undo behind the review's route: one transaction, a batch of kind `undo`.""" from app.services.decisions import undo as undo_service user = _account(db, by) if user is None: raise Refused("ENTITY_NOT_FOUND", "An undo is a person's act: name an account.") if not (reason or "").strip(): raise Refused("VALUE_INVALID", "An undo says why.") row = undo_service.undo_batch(db, batch_uid, user, reason.strip(), force=force) summary = row.summary or {} return {"undo_batch_uid": str(row.uid), "reversed_batch_uid": str(batch_uid), "decisions": int(summary.get("decisions", 0)), "skipped": summary.get("skipped", [])} def undo_decision(db: Session, decision_id: int, by, reason: str, *, force: bool = False) -> dict[str, Any]: """K2's single undo behind the ledger view's route: a batch of one (`route`, individual).""" from app.services.decisions import undo as undo_service from app.services.decisions.writer import batch as open_batch refuse_if_collecting(db, force) user = _account(db, by) if user is None: raise Refused("ENTITY_NOT_FOUND", "An undo is a person's act: name an account.") if not (reason or "").strip(): raise Refused("VALUE_INVALID", "An undo says why.") if db.get(Decision, decision_id) is None: raise Refused("ENTITY_NOT_FOUND", f"No decision {decision_id}.") skipped: list[dict] = [] with open_batch(db, "route", "individual", user, scope={"undo_decision": decision_id}, note=reason.strip()) as b: row = undo_service.undo(b, decision_id, reason.strip(), skipped=skipped) uid = str(b.uid) return {"undo_batch_uid": uid, "decision_uid": str(row.uid), "skipped": skipped} def in_force(db: Session, p: Proposal, now: Decision | None | bool = False) -> bool: """Whether an approved or rejected proposal's decision still stands: the effective row on its field is that decision, or an undo of an undo restoring it. A decision undone (by the sheet's route or `decisions undo-batch`) leaves the proposal waiting again, computed at read, so both undo paths agree and the table's CHECK (approved has a decision) is never broken.""" if p.decision_id is None: return p.status == "rejected" dec = db.get(Decision, p.decision_id) if dec is None: return False if now is False: now = effective(db, dec.entity_type, [dec.entity_id], [dec.field]).get((dec.entity_id, dec.field)) return now is not None and (now.id == dec.id or now.restores_id == dec.id) def waiting(db: Session, p: Proposal, now: Decision | None | bool = False) -> bool: """Open, or approved/rejected with its decision undone since.""" if p.status == "open": return True if p.status in ("approved", "rejected") and p.decision_id is not None: return not in_force(db, p, now) return False