"""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 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") 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))) ours = {(s.entity_type, s.natural_key, s.field) for s in specs} 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 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.""" live_passes = {p.id: p for p in db.scalars(select(ProposalPass).where(ProposalPass.withdrawn_at.is_(None)))} 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 _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 _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) chosen = next((p for p in passes if p.name == pass_name), None) if pass_name else next((p for p in passes if p.withdrawn_at is None), passes[0] if passes else None) if pass_name and chosen is None: raise Refused("PASS_NOT_FOUND", f"No pass {pass_name!r} for {brand.slug}.") 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": [], "batches": brand_batches(db, slugs), } if chosen is None: return result rows = list(db.scalars(select(Proposal).where(Proposal.pass_id == chosen.id, Proposal.brand_slug.in_(slugs)) .order_by(Proposal.position, Proposal.id))) all_passes = {p.id: p for p in db.scalars(select(ProposalPass))} 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[str | None, dict] = {} order: list[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 != chosen.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) 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, "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")), "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", } 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)) if ref not in groups: groups[ref] = {"ref": ref, "name": None, "header_uid": None, "absorbs": [], "rows": []} order.append(ref) group = groups[ref] 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) ordered = sorted(order, key=lambda ref: (ref is not None, min((row["position"] for row in groups[ref]["rows"]), default=0))) result["groups"] = [groups[ref] for ref in ordered] result["counts"] = dict(counts) return result 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, 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 [])] unknown = [u for u in list(corrections) + reject 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.") # 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] 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}) 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] for r in reject_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.") 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 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" 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 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}" return _response(result, counts, refusals, held, absorb_report) 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