"""Review decisions for V1 → V2 attribute seed candidates. Discover writes candidates. Operators approve, modify, merge, or exclude them. Apply copies only approved/modified records into the live v2 attribute registry. """ from __future__ import annotations from datetime import datetime, timezone from typing import Any import re from sqlalchemy import select from sqlalchemy.exc import OperationalError, ProgrammingError from sqlalchemy.orm import Session from db.models import MagentoAttributeOptionRegistry, MagentoAttributeRegistry, MasterProductAttributeValue, ShopifyAttributeRegistry from db.models_v2 import ( V2AttributeSeedCandidate, V2AttributeSeedMappingCandidate, V2AttributeValueSeedCandidate, ) from db.source_imports import normalize_column_key from db.v2_attribute_types import ( ADMIN_UI_CONTROLS, APPLY_READY_STATUSES, CANONICAL_DATA_TYPES, LIFECYCLE_STATUSES, OWNER_SCOPES, REVIEW_STATUSES, SELECTABLE_TYPES, VALUE_REVIEW_STATUSES, channel_compatibility, default_ui_control, magento_frontend_input, map_canonical_data_type, shopify_metafield_type, ) from db.v2_import.matching import normalize_lookup_value from db.v2_attribute_seed import candidate_ownership, content_composition_sources, infer_owner_scope from db.channel_attribute_options import channel_attribute_options_bundle class AttributeReviewError(ValueError): pass def list_seed_candidates( session: Session, *, review_status: str | None = None, search: str | None = None, owner_scope: str | None = None, ) -> list[V2AttributeSeedCandidate]: stmt = select(V2AttributeSeedCandidate).order_by(V2AttributeSeedCandidate.source_code.asc()) if review_status: stmt = stmt.where(V2AttributeSeedCandidate.review_status == review_status) if owner_scope: stmt = stmt.where(V2AttributeSeedCandidate.owner_scope == owner_scope) rows = list(session.scalars(stmt).all()) term = str(search or "").strip().lower() if term: rows = [ row for row in rows if term in " ".join( str(item or "").lower() for item in ( row.source_code, row.proposed_code, row.proposed_label, row.proposed_data_type, getattr(row, "owner_scope", ""), ) ) ] return rows def get_seed_candidate(session: Session, candidate_id: int) -> V2AttributeSeedCandidate: row = session.get(V2AttributeSeedCandidate, int(candidate_id)) if row is None: raise AttributeReviewError("Attribute seed candidate not found.") return row def candidate_detail(session: Session, candidate: V2AttributeSeedCandidate) -> dict[str, Any]: refresh_seed_review_state(session, candidate) values = _ordered_values_for_candidate(session, candidate) mappings = list( session.scalars( select(V2AttributeSeedMappingCandidate).where( V2AttributeSeedMappingCandidate.attribute_source_code == candidate.source_code ) ).all() ) type_errors = validate_type_change(session, candidate, candidate.proposed_data_type) return { "candidate": serialize_candidate(candidate), "values": [serialize_value(item) for item in values], "mappings": [serialize_mapping(item) for item in mappings], "impact": candidate_impact(session, candidate), "type_validation": type_errors, "unmapped_product_values": unmapped_product_value_report(session, candidate), } def candidate_value_previews( session: Session, source_codes: list[str], *, limit: int = 8, ) -> dict[str, list[str]]: if not source_codes: return {} rows = session.scalars( select(V2AttributeValueSeedCandidate) .where(V2AttributeValueSeedCandidate.attribute_source_code.in_(source_codes)) .order_by( V2AttributeValueSeedCandidate.attribute_source_code.asc(), V2AttributeValueSeedCandidate.sort_order.asc(), V2AttributeValueSeedCandidate.proposed_label.asc(), ) ).all() grouped: dict[str, list[str]] = {} for row in rows: bucket = grouped.setdefault(str(row.attribute_source_code), []) if len(bucket) >= limit: continue label = str(row.proposed_label or row.source_value_code or "").strip() if label: bucket.append(label) return grouped def serialize_candidate(row: V2AttributeSeedCandidate, *, value_preview: list[str] | None = None) -> dict[str, Any]: return { "id": int(row.id), "source_code": row.source_code, "proposed_code": row.proposed_code, "proposed_label": row.proposed_label, "discovered_data_type": row.discovered_data_type, "proposed_data_type": row.proposed_data_type, "proposed_ui_control": row.proposed_ui_control, "is_variant_axis": bool(row.is_variant_axis), "is_channel_required": bool(row.is_channel_required), "lifecycle_status": row.lifecycle_status, "owner_scope": getattr(row, "owner_scope", None) or infer_owner_scope( code=row.source_code, magento_code=row.magento_attribute_code, label=row.proposed_label ), "review_status": row.review_status, "merge_into_code": row.merge_into_code, "sources": row.sources or [], "aliases": row.aliases or [], "product_usage_count": int(row.product_usage_count or 0), "value_count": int(row.value_count or 0), "value_preview": list(value_preview or []), "channel_map_count": int(row.channel_map_count or 0), "magento_frontend_input": row.magento_frontend_input, "magento_attribute_code": row.magento_attribute_code, "shopify_representation": row.shopify_representation or {}, "conflict_reasons": row.conflict_reasons or [], "proposed_action": row.proposed_action, "duplicate_group": row.duplicate_group, "notes": row.notes, "ownership": candidate_ownership(row.source_code, row.proposed_label, row.proposed_code), "composition_sources": content_composition_sources(row.source_code) or content_composition_sources(row.proposed_code or ""), "applied_at": row.applied_at.isoformat() if row.applied_at else None, "updated_at": row.updated_at.isoformat() if row.updated_at else None, } def serialize_value(row: V2AttributeValueSeedCandidate) -> dict[str, Any]: return { "id": int(row.id), "attribute_source_code": row.attribute_source_code, "source_value_code": row.source_value_code, "proposed_value_code": row.proposed_value_code, "proposed_label": row.proposed_label, "sort_order": int(row.sort_order or 0), "review_status": row.review_status, "merge_into_value_code": row.merge_into_value_code, "aliases": row.aliases or [], "product_usage_count": int(row.product_usage_count or 0), "lifecycle_status": row.lifecycle_status, } def serialize_mapping(row: V2AttributeSeedMappingCandidate) -> dict[str, Any]: return { "id": int(row.id), "channel": row.channel, "channel_attribute_code": row.channel_attribute_code, "channel_attribute_id": row.channel_attribute_id, "representation": row.representation or {}, "compatibility_status": row.compatibility_status, "is_manual": bool(getattr(row, "is_manual", False)), } def seed_channel_options(session: Session, owner_scope: str | None = None) -> dict[str, list[dict[str, Any]]]: try: bundle = channel_attribute_options_bundle(session) except (OperationalError, ProgrammingError): bundle = {"magento": [], "shopify": []} return { "magento": _filter_channel_options(bundle.get("magento") or [], "magento", owner_scope), "shopify": _filter_channel_options(bundle.get("shopify") or [], "shopify", owner_scope), } def _filter_channel_options( options: list[dict[str, Any]], channel: str, owner_scope: str | None, ) -> list[dict[str, Any]]: if not owner_scope: return list(options) wanted_category = owner_scope == "category" filtered: list[dict[str, Any]] = [] for option in options: if _option_is_category_related(channel, option) == wanted_category: filtered.append(option) return filtered def _option_is_category_related(channel: str, option: dict[str, Any]) -> bool: code = str(option.get("code") or "").strip().lower() owner = str(option.get("owner_type") or "").strip().lower() if channel == "magento": return code.startswith("hs_") return "collection" in owner or "category" in owner def upsert_candidate_mapping( session: Session, candidate_id: int, payload: dict[str, Any], ) -> V2AttributeSeedMappingCandidate: candidate = get_seed_candidate(session, candidate_id) session.flush() channel = str(payload.get("channel") or "").strip().lower() if channel not in {"magento", "shopify"}: raise AttributeReviewError("Map Magento or Shopify.") code = str(payload.get("channel_attribute_code") or "").strip() if not code: raise AttributeReviewError("Choose or type an outbound attribute code.") representation, channel_attribute_id = _representation_for_outbound( session, channel, code, candidate.proposed_data_type ) representation["channel_attribute_code"] = code representation["sync_mode"] = str(payload.get("sync_mode") or "use_existing").strip() or "use_existing" representation["is_optional"] = False compat = channel_compatibility(candidate.proposed_data_type, channel, representation) now = datetime.now(timezone.utc) row = session.scalar( select(V2AttributeSeedMappingCandidate).where( V2AttributeSeedMappingCandidate.attribute_source_code == candidate.source_code, V2AttributeSeedMappingCandidate.channel == channel, ) ) if row is None: row = V2AttributeSeedMappingCandidate( attribute_source_code=candidate.source_code, channel=channel, channel_attribute_code=code, channel_attribute_id=channel_attribute_id, representation=representation, compatibility_status=compat, is_manual=True, created_at=now, updated_at=now, ) session.add(row) else: row.channel_attribute_code = code row.channel_attribute_id = channel_attribute_id row.representation = representation row.compatibility_status = compat row.is_manual = True row.updated_at = now session.flush() if candidate.review_status in {"pending_review", "approved", "conflict"}: candidate.review_status = "modified" candidate.channel_map_count = len( list( session.scalars( select(V2AttributeSeedMappingCandidate).where( V2AttributeSeedMappingCandidate.attribute_source_code == candidate.source_code ) ).all() ) ) return row def delete_candidate_mapping(session: Session, mapping_id: int) -> V2AttributeSeedCandidate: row = session.get(V2AttributeSeedMappingCandidate, int(mapping_id)) if row is None: raise AttributeReviewError("Channel mapping not found.") source_code = str(row.attribute_source_code) session.delete(row) session.flush() candidate = session.scalar( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == source_code) ) if candidate is None: raise AttributeReviewError("Attribute seed candidate not found.") now = datetime.now(timezone.utc) if candidate.review_status in {"pending_review", "approved"}: candidate.review_status = "modified" candidate.updated_at = now candidate.channel_map_count = len( list( session.scalars( select(V2AttributeSeedMappingCandidate).where( V2AttributeSeedMappingCandidate.attribute_source_code == source_code ) ).all() ) ) return candidate def _representation_for_outbound( session: Session, channel: str, code: str, data_type: str, ) -> tuple[dict[str, Any], str | None]: if channel == "magento": magento = _lookup_magento_attribute(session, code) representation = { "frontend_input": str(magento.frontend_input or "").strip() if magento is not None else magento_frontend_input(data_type), "backend_type": str(magento.backend_type or "").strip() if magento is not None else None, "attribute_id": str(magento.attribute_id) if magento is not None else None, "attribute_code": code, } return representation, str(magento.attribute_id) if magento is not None else None shopify = _lookup_shopify_attribute(session, code) representation = { "kind": "metafield", "attribute_code": code, "metafield_type": shopify.data_type if shopify is not None else shopify_metafield_type(data_type), "namespace": shopify.namespace if shopify is not None else None, "key": shopify.key if shopify is not None else code, "name": shopify.name if shopify is not None else code, } if shopify is not None and str(shopify.owner_type or "").lower() in {"product", "variant"}: representation["kind"] = "product_option" return representation, None def _lookup_magento_attribute(session: Session, code: str) -> MagentoAttributeRegistry | None: try: return session.scalar( select(MagentoAttributeRegistry) .where(MagentoAttributeRegistry.attribute_code == code) .order_by(MagentoAttributeRegistry.id.desc()) ) except (OperationalError, ProgrammingError): return None def _lookup_shopify_attribute(session: Session, code: str) -> ShopifyAttributeRegistry | None: try: return session.scalar( select(ShopifyAttributeRegistry) .where(ShopifyAttributeRegistry.attribute_code == code) .order_by(ShopifyAttributeRegistry.id.desc()) ) except (OperationalError, ProgrammingError): return None def candidate_impact(session: Session, candidate: V2AttributeSeedCandidate) -> dict[str, Any]: values = list( session.scalars( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == candidate.source_code ) ).all() ) mappings = list( session.scalars( select(V2AttributeSeedMappingCandidate).where( V2AttributeSeedMappingCandidate.attribute_source_code == candidate.source_code ) ).all() ) return { "product_usage_count": int(candidate.product_usage_count or 0), "value_count": len(values), "channel_map_count": len(mappings), "magento_mapped": any(row.channel == "magento" for row in mappings), "shopify_mapped": any(row.channel == "shopify" for row in mappings), "aliases": list(candidate.aliases or []) + [candidate.source_code], "incompatible_channels": [ row.channel for row in mappings if row.compatibility_status == "incompatible" ], } def update_candidate(session: Session, candidate_id: int, payload: dict[str, Any]) -> V2AttributeSeedCandidate: row = get_seed_candidate(session, candidate_id) if "proposed_code" in payload: row.proposed_code = normalize_column_key(payload.get("proposed_code") or row.source_code) if "proposed_label" in payload: row.proposed_label = str(payload.get("proposed_label") or "").strip() or row.proposed_label if "proposed_data_type" in payload: data_type = map_canonical_data_type(payload.get("proposed_data_type")) if data_type not in CANONICAL_DATA_TYPES: raise AttributeReviewError("Unsupported canonical type.") errors = validate_type_change(session, row, data_type) if errors: raise AttributeReviewError(errors[0]) row.proposed_data_type = data_type if "proposed_ui_control" not in payload: row.proposed_ui_control = default_ui_control(data_type) if "proposed_ui_control" in payload: control = str(payload.get("proposed_ui_control") or "").strip() if control not in ADMIN_UI_CONTROLS: raise AttributeReviewError("Unsupported admin UI control.") row.proposed_ui_control = control if "is_variant_axis" in payload: row.is_variant_axis = bool(payload.get("is_variant_axis")) if "is_channel_required" in payload: row.is_channel_required = bool(payload.get("is_channel_required")) if "lifecycle_status" in payload: status = str(payload.get("lifecycle_status") or "").strip() if status not in LIFECYCLE_STATUSES: raise AttributeReviewError("Unsupported lifecycle status.") row.lifecycle_status = status if "notes" in payload: row.notes = str(payload.get("notes") or "").strip() or None if "owner_scope" in payload: scope = str(payload.get("owner_scope") or "").strip().lower() if scope not in OWNER_SCOPES: raise AttributeReviewError("Owner scope must be product or category.") row.owner_scope = scope if "aliases" in payload and isinstance(payload.get("aliases"), list): row.aliases = [str(item).strip() for item in payload["aliases"] if str(item).strip()] row.review_status = "modified" row.updated_at = datetime.now(timezone.utc) refresh_seed_review_state(session, row) return row def refresh_seed_review_state(session: Session, candidate: V2AttributeSeedCandidate) -> None: """Recompute stored discover warnings and channel compatibility from the operator's proposed type.""" proposed = map_canonical_data_type(candidate.proposed_data_type) if candidate.review_status in {"modified", "approved"}: candidate.conflict_reasons = [ reason for reason in (candidate.conflict_reasons or []) if not str(reason).startswith("source types disagree:") ] now = datetime.now(timezone.utc) for mapping in session.scalars( select(V2AttributeSeedMappingCandidate).where( V2AttributeSeedMappingCandidate.attribute_source_code == candidate.source_code ) ).all(): representation = dict(mapping.representation or {}) if mapping.channel_attribute_code and not representation.get("channel_attribute_code"): representation["channel_attribute_code"] = mapping.channel_attribute_code mapping.representation = representation mapping.compatibility_status = channel_compatibility(proposed, mapping.channel, representation) mapping.updated_at = now def approve_candidate(session: Session, candidate_id: int) -> V2AttributeSeedCandidate: row = get_seed_candidate(session, candidate_id) errors = validate_type_change(session, row, row.proposed_data_type) if errors: row.review_status = "conflict" row.conflict_reasons = errors raise AttributeReviewError(errors[0]) row.review_status = "approved" row.proposed_action = "keep" row.updated_at = datetime.now(timezone.utc) for value in _values_for_candidate(session, row): if value.review_status == "pending_review": value.review_status = "approved" value.updated_at = row.updated_at return row def exclude_candidate(session: Session, candidate_id: int) -> V2AttributeSeedCandidate: row = get_seed_candidate(session, candidate_id) row.review_status = "excluded" row.proposed_action = "exclude" row.updated_at = datetime.now(timezone.utc) for value in _values_for_candidate(session, row): if value.review_status == "pending_review": value.review_status = "excluded" value.updated_at = row.updated_at return row def bulk_approve_candidates(session: Session, candidate_ids: list[int]) -> dict[str, Any]: approved: list[int] = [] errors: list[dict[str, Any]] = [] for candidate_id in candidate_ids: try: row = approve_candidate(session, int(candidate_id)) approved.append(int(row.id)) except AttributeReviewError as exc: errors.append({"id": int(candidate_id), "error": str(exc)}) return {"approved_ids": approved, "errors": errors} def bulk_exclude_candidates(session: Session, candidate_ids: list[int]) -> dict[str, Any]: excluded: list[int] = [] errors: list[dict[str, Any]] = [] for candidate_id in candidate_ids: try: row = exclude_candidate(session, int(candidate_id)) excluded.append(int(row.id)) except AttributeReviewError as exc: errors.append({"id": int(candidate_id), "error": str(exc)}) return {"excluded_ids": excluded, "errors": errors} def merge_candidate(session: Session, candidate_id: int, merge_into_code: str) -> V2AttributeSeedCandidate: row = get_seed_candidate(session, candidate_id) target_code = normalize_column_key(merge_into_code) if not target_code or target_code == row.source_code: raise AttributeReviewError("Choose a different surviving attribute code.") target = session.execute( select(V2AttributeSeedCandidate).where( (V2AttributeSeedCandidate.source_code == target_code) | (V2AttributeSeedCandidate.proposed_code == target_code) ) ).scalars().first() if target is None: raise AttributeReviewError("Merge target is not in the seed review set.") row.review_status = "merged" row.merge_into_code = target.proposed_code or target.source_code row.proposed_action = "merge" row.updated_at = datetime.now(timezone.utc) aliases = list(target.aliases or []) if row.source_code not in aliases: aliases.append(row.source_code) target.aliases = aliases if target.review_status == "merged": target.review_status = "modified" target.merge_into_code = None target.proposed_action = "keep" if row.is_variant_axis: target.is_variant_axis = True _revive_survivor_values(session, target) _fold_merged_children(session, row, target) session.flush() return row def bulk_merge_candidates(session: Session, candidate_ids: list[int], survivor_id: int) -> dict[str, Any]: ids = [int(item) for item in candidate_ids] survivor_id = int(survivor_id) if survivor_id not in ids: raise AttributeReviewError("The surviving attribute must be one of the selected rows.") if len(set(ids)) < 2: raise AttributeReviewError("Select at least two attributes to merge.") survivor = get_seed_candidate(session, survivor_id) if survivor.review_status == "merged": survivor.review_status = "modified" survivor.merge_into_code = None survivor.proposed_action = "keep" survivor.updated_at = datetime.now(timezone.utc) _revive_survivor_values(session, survivor) merged_ids: list[int] = [] for candidate_id in ids: if candidate_id == survivor_id: continue row = merge_candidate(session, candidate_id, survivor.source_code) merged_ids.append(int(row.id)) session.flush() if survivor.review_status == "pending_review": survivor.review_status = "modified" survivor.updated_at = datetime.now(timezone.utc) return { "survivor_id": int(survivor.id), "survivor_code": survivor.proposed_code or survivor.source_code, "merged_ids": merged_ids, } def _revive_survivor_values(session: Session, survivor: V2AttributeSeedCandidate) -> None: now = datetime.now(timezone.utc) for row in _values_for_attribute(session, survivor.source_code): if row.review_status != "merged": continue row.review_status = "modified" row.merge_into_value_code = None row.updated_at = now def _values_for_attribute(session: Session, source_code: str) -> list[V2AttributeValueSeedCandidate]: wanted = str(source_code or "").strip() rows: dict[int, V2AttributeValueSeedCandidate] = {} for row in session.scalars( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == wanted ) ).all(): rows[int(row.id)] = row for obj in list(session.identity_map.values()): if isinstance(obj, V2AttributeValueSeedCandidate) and str(obj.attribute_source_code) == wanted: rows[int(obj.id)] = obj return list(rows.values()) def _expand_value_key(raw: object) -> list[str]: text = normalize_lookup_value(raw) if not text: return [] text = ( text.replace("\u00a0", " ") .replace("\u2033", '"') .replace("\u201d", '"') .replace("\u2019", "'") ) keys = [text] dequoted = re.sub(r'["\'″”′]', "", text) dequoted = re.sub(r"\s+", " ", dequoted).strip() if dequoted and dequoted not in keys: keys.append(dequoted) underscored = dequoted.replace(" ", "_").replace("/", "_") if underscored and underscored not in keys: keys.append(underscored) dotted = dequoted.replace("_", ".") if re.fullmatch(r"\d+(?:\.\d+)?", dotted): number = float(dotted) compact = str(int(number)) if number.is_integer() else format(number, "f").rstrip("0").rstrip(".") keys.extend([compact, compact.replace(".", "_"), dotted]) measured = _measurement_key(raw) or _measurement_key(dequoted) if measured: keys.append(measured) keys.append(measured.replace(".", "_")) return keys def _measurement_key(raw: object) -> str: try: from db.canonical_attributes import normalize_depth_option_label except Exception: return "" measured = normalize_depth_option_label(raw) return normalize_lookup_value(measured) if measured else "" def _value_lookup_keys(row: V2AttributeValueSeedCandidate) -> list[str]: seen: set[str] = set() keys: list[str] = [] for item in (row.proposed_label, row.proposed_value_code, row.source_value_code, *(row.aliases or [])): for key in _expand_value_key(item): if key not in seen: seen.add(key) keys.append(key) return keys def _index_value( by_norm: dict[str, V2AttributeValueSeedCandidate], by_source_code: dict[str, V2AttributeValueSeedCandidate], row: V2AttributeValueSeedCandidate, ) -> None: by_source_code[str(row.source_value_code)] = row for key in _value_lookup_keys(row): by_norm[key] = row def _match_folded_value( item: V2AttributeValueSeedCandidate, by_norm: dict[str, V2AttributeValueSeedCandidate], by_source_code: dict[str, V2AttributeValueSeedCandidate], ) -> V2AttributeValueSeedCandidate | None: exact = by_source_code.get(str(item.source_value_code)) if exact is not None: return exact for key in _value_lookup_keys(item): match = by_norm.get(key) if match is not None: return match return None def _alias_value_into( match: V2AttributeValueSeedCandidate, item: V2AttributeValueSeedCandidate, now: datetime, ) -> None: aliases = list(match.aliases or []) for alias in (item.source_value_code, item.proposed_label, item.proposed_value_code, *(item.aliases or [])): if alias and alias not in aliases: aliases.append(alias) match.aliases = aliases item.review_status = "merged" item.merge_into_value_code = match.proposed_value_code or match.source_value_code item.updated_at = now def _fold_merged_children( session: Session, source: V2AttributeSeedCandidate, target: V2AttributeSeedCandidate, ) -> None: now = datetime.now(timezone.utc) session.flush() target_values = _values_for_attribute(session, target.source_code) by_norm: dict[str, V2AttributeValueSeedCandidate] = {} by_source_code: dict[str, V2AttributeValueSeedCandidate] = {} for row in target_values: _index_value(by_norm, by_source_code, row) source_values = _values_for_attribute(session, source.source_code) for item in source_values: match = _match_folded_value(item, by_norm, by_source_code) if match is not None and match is not item: _alias_value_into(match, item, now) continue if item.attribute_source_code == target.source_code: _index_value(by_norm, by_source_code, item) continue item.attribute_source_code = target.source_code aliases = list(item.aliases or []) if item.source_value_code not in aliases: aliases.append(item.source_value_code) item.aliases = aliases item.updated_at = now _index_value(by_norm, by_source_code, item) target.value_count = len(_values_for_attribute(session, target.source_code)) target_maps = { str(row.channel): row for row in session.scalars( select(V2AttributeSeedMappingCandidate).where( V2AttributeSeedMappingCandidate.attribute_source_code == target.source_code ) ).all() } for mapping in session.scalars( select(V2AttributeSeedMappingCandidate).where( V2AttributeSeedMappingCandidate.attribute_source_code == source.source_code ) ).all(): if mapping.channel in target_maps: continue mapping.attribute_source_code = target.source_code mapping.updated_at = now target_maps[mapping.channel] = mapping target.channel_map_count = len(target_maps) recount_candidate_usage(session, target) target.updated_at = now def update_value(session: Session, value_id: int, payload: dict[str, Any]) -> V2AttributeValueSeedCandidate: row = session.get(V2AttributeValueSeedCandidate, int(value_id)) if row is None: raise AttributeReviewError("Attribute value candidate not found.") if "proposed_label" in payload: row.proposed_label = str(payload.get("proposed_label") or "").strip() or row.proposed_label if "proposed_value_code" not in payload: derived = value_code_from_display(row.proposed_label) if derived: _assert_value_code_available( session, row.attribute_source_code, derived, ignore_id=int(row.id) ) _remember_value_alias(row, row.proposed_value_code, derived) row.proposed_value_code = derived if "proposed_value_code" in payload: forced = value_code_from_display(payload.get("proposed_value_code") or row.proposed_label) if not forced: raise AttributeReviewError("Enter a V2 option code.") _assert_value_code_available(session, row.attribute_source_code, forced, ignore_id=int(row.id)) _remember_value_alias(row, row.proposed_value_code, forced) row.proposed_value_code = forced if "sort_order" in payload: row.sort_order = int(payload.get("sort_order") or 0) if "lifecycle_status" in payload: status = str(payload.get("lifecycle_status") or "").strip() if status not in LIFECYCLE_STATUSES: raise AttributeReviewError("Unsupported lifecycle status.") row.lifecycle_status = status if "review_status" in payload: status = str(payload.get("review_status") or "").strip() if status not in VALUE_REVIEW_STATUSES: raise AttributeReviewError("Unsupported value review status.") row.review_status = status else: row.review_status = "modified" if row.review_status == "pending_review" else row.review_status if "merge_into_value_code" in payload: row.merge_into_value_code = str(payload.get("merge_into_value_code") or "").strip() or None if row.merge_into_value_code: row.review_status = "merged" row.updated_at = datetime.now(timezone.utc) return row def bulk_merge_values(session: Session, value_ids: list[int], survivor_id: int) -> dict[str, Any]: ids = [int(item) for item in value_ids] survivor_id = int(survivor_id) if survivor_id not in ids: raise AttributeReviewError("The surviving value must be one of the selected rows.") if len(set(ids)) < 2: raise AttributeReviewError("Select at least two values to merge.") survivor = session.get(V2AttributeValueSeedCandidate, survivor_id) if survivor is None: raise AttributeReviewError("Value candidate not found.") if survivor.review_status in {"merged", "excluded"}: survivor.review_status = "modified" survivor.merge_into_value_code = None now = datetime.now(timezone.utc) merged_ids: list[int] = [] owner = _candidate_owning_value(session, survivor) for value_id in ids: if value_id == survivor_id: continue row = session.get(V2AttributeValueSeedCandidate, value_id) if row is None: raise AttributeReviewError("Value candidate not found.") if not _same_review_attribute(session, survivor, row, owner): raise AttributeReviewError("Values must belong to the same attribute.") _alias_value_into(survivor, row, now) merged_ids.append(int(row.id)) if survivor.review_status == "pending_review": survivor.review_status = "modified" survivor.updated_at = now if owner is not None: recount_candidate_usage(session, owner) session.flush() return { "survivor_id": int(survivor.id), "survivor_code": survivor.proposed_value_code or survivor.source_value_code, "merged_ids": merged_ids, } def value_code_from_display(label: object) -> str: """Turn a display like 90, 21.5, or 21 1/2 into a stable option code.""" return normalize_column_key(str(label or "").strip()) def _remember_value_alias(row: V2AttributeValueSeedCandidate, old: object, new_code: object) -> None: text = str(old or "").strip() if not text or text == str(new_code or "").strip(): return aliases = list(row.aliases or []) if text not in aliases: aliases.append(text) row.aliases = aliases def _assert_value_code_available( session: Session, attribute_source_code: str, proposed_code: str, *, ignore_id: int | None = None, ) -> None: wanted = str(proposed_code or "").strip() if not wanted: return for row in session.scalars( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == attribute_source_code ) ).all(): if ignore_id is not None and int(row.id) == int(ignore_id): continue if row.review_status in {"merged", "excluded"}: continue if str(row.proposed_value_code or "").strip() == wanted: raise AttributeReviewError( f'V2 option code "{wanted}" is already used by "{row.proposed_label}". Merge into that option or pick a different code.' ) def create_value( session: Session, candidate_id: int, payload: dict[str, Any], ) -> V2AttributeValueSeedCandidate: candidate = get_seed_candidate(session, candidate_id) label = str(payload.get("proposed_label") or payload.get("label") or "").strip() if not label: raise AttributeReviewError("Enter a display label for the new option.") proposed_code = value_code_from_display( payload.get("proposed_value_code") or payload.get("code") or label ) if not proposed_code: raise AttributeReviewError("Enter a V2 option code.") source_code = value_code_from_display( payload.get("source_value_code") or proposed_code ) existing = _find_value_row(session, candidate, label) or _find_value_row(session, candidate, source_code) if existing is not None: if existing.review_status in {"excluded", "merged"}: existing.review_status = "modified" existing.merge_into_value_code = None _assert_value_code_available( session, candidate.source_code, proposed_code, ignore_id=int(existing.id) ) _remember_value_alias(existing, existing.proposed_value_code, proposed_code) _remember_value_alias(existing, existing.source_value_code, proposed_code) existing.proposed_label = label existing.proposed_value_code = proposed_code existing.updated_at = datetime.now(timezone.utc) _mark_candidate_modified(candidate) return existing _assert_value_code_available(session, candidate.source_code, proposed_code) now = datetime.now(timezone.utc) values = _values_for_candidate(session, candidate) sort_order = max((int(row.sort_order or 0) for row in values), default=-10) + 10 row = V2AttributeValueSeedCandidate( attribute_source_code=candidate.source_code, source_value_code=source_code, proposed_value_code=proposed_code, proposed_label=label, sort_order=sort_order, review_status="modified", aliases=[], product_usage_count=0, lifecycle_status="active", created_at=now, updated_at=now, ) session.add(row) session.flush() candidate.value_count = len(_values_for_candidate(session, candidate)) _mark_candidate_modified(candidate) return row def create_and_merge_values( session: Session, candidate_id: int, *, label: str, merge_values: list[str], value_code: str | None = None, ) -> dict[str, Any]: display = str(label or "").strip() if not display: raise AttributeReviewError("Enter a display label for the new option.") survivor = create_value( session, candidate_id, { "proposed_label": display, "proposed_value_code": value_code or display, }, ) candidate = get_seed_candidate(session, candidate_id) ids = [int(survivor.id)] now = datetime.now(timezone.utc) aliases = list(survivor.aliases or []) for raw in merge_values: text = str(raw or "").strip() if not text: continue match = _find_value_row(session, candidate, text) if match is None: if text not in aliases: aliases.append(text) continue if int(match.id) == int(survivor.id): continue ids.append(int(match.id)) survivor.aliases = aliases survivor.updated_at = now result = {"survivor_id": int(survivor.id), "survivor_code": survivor.proposed_value_code, "merged_ids": []} if len(set(ids)) >= 2: result = bulk_merge_values(session, ids, int(survivor.id)) session.flush() return result def _mark_candidate_modified(candidate: V2AttributeSeedCandidate) -> None: if candidate.review_status in {"pending_review", "approved", "conflict"}: candidate.review_status = "modified" candidate.updated_at = datetime.now(timezone.utc) def _ordered_values_for_candidate( session: Session, candidate: V2AttributeSeedCandidate, ) -> list[V2AttributeValueSeedCandidate]: rows = _values_for_candidate(session, candidate) return sorted( rows, key=lambda row: ( 1 if row.review_status == "merged" else 0, int(row.sort_order or 0), str(row.proposed_label or ""), int(row.id or 0), ), ) def _values_for_candidate( session: Session, candidate: V2AttributeSeedCandidate, ) -> list[V2AttributeValueSeedCandidate]: live, excluded = _value_rows_for_codes(session, _candidate_source_codes(candidate)) return live + excluded def _lookup_candidate_by_code(session: Session, code: str) -> V2AttributeSeedCandidate | None: wanted = str(code or "").strip() if not wanted: return None return session.execute( select(V2AttributeSeedCandidate).where( (V2AttributeSeedCandidate.source_code == wanted) | (V2AttributeSeedCandidate.proposed_code == wanted) ) ).scalars().first() def _resolve_review_attribute( session: Session, code: str, *, seen: set[str] | None = None, ) -> V2AttributeSeedCandidate | None: wanted = str(code or "").strip() if not wanted: return None seen = seen if seen is not None else set() if wanted in seen: return None seen.add(wanted) direct = _lookup_candidate_by_code(session, wanted) if direct is not None: if direct.review_status == "merged" and direct.merge_into_code: survivor = _resolve_review_attribute(session, str(direct.merge_into_code), seen=seen) if survivor is not None: return survivor return direct for candidate in session.scalars(select(V2AttributeSeedCandidate)).all(): if candidate.review_status == "merged": continue aliases = {str(alias).strip() for alias in (candidate.aliases or [])} if wanted in aliases: return candidate return None def _candidate_owning_value( session: Session, row: V2AttributeValueSeedCandidate, ) -> V2AttributeSeedCandidate | None: return _resolve_review_attribute(session, str(row.attribute_source_code or "").strip()) def _same_review_attribute( session: Session, survivor: V2AttributeValueSeedCandidate, row: V2AttributeValueSeedCandidate, owner: V2AttributeSeedCandidate | None, ) -> bool: if survivor.attribute_source_code == row.attribute_source_code: return True other = _candidate_owning_value(session, row) if owner is not None and other is not None and int(owner.id) == int(other.id): return True if owner is not None and str(row.attribute_source_code or "").strip() in _candidate_source_codes(owner): return True return False def refold_merged_candidates(session: Session) -> None: """Re-attach values that rediscover created on a merged-away attribute.""" by_source = {str(row.source_code): row for row in session.scalars(select(V2AttributeSeedCandidate)).all()} for row in list(by_source.values()): if row.review_status != "merged" or not row.merge_into_code: continue survivor = _resolve_review_attribute(session, str(row.merge_into_code)) if survivor is None or int(survivor.id) == int(row.id): continue _fold_merged_children(session, row, survivor) def _find_value_row( session: Session, candidate: V2AttributeSeedCandidate, raw: object, ) -> V2AttributeValueSeedCandidate | None: wanted = set(_expand_value_key(raw)) if not wanted: return None matches: list[V2AttributeValueSeedCandidate] = [] for row in _values_for_candidate(session, candidate): if wanted.intersection(_value_lookup_keys(row)): matches.append(row) if not matches: return None live = [row for row in matches if row.review_status not in {"excluded", "merged"}] return live[0] if live else matches[0] def validate_type_change( session: Session, candidate: V2AttributeSeedCandidate, data_type: str, ) -> list[str]: canonical = map_canonical_data_type(data_type) if canonical not in SELECTABLE_TYPES: return [] report = unmapped_product_value_report(session, candidate) if not report: return [] excluded = [item["value"] for item in report if item["reason"] == "excluded"] unresolved = [item["value"] for item in report if item["reason"] != "excluded"] hint = "Create the V2 option and merge these values into it." if excluded: preview = ", ".join(excluded[:8]) extra = f" (+{len(excluded) - 8} more)" if len(excluded) > 8 else "" return [ f"{len(excluded)} product value(s) are excluded but still used: {preview}{extra}. Un-exclude them, or {hint[0].lower() + hint[1:]}" ] preview = ", ".join(unresolved[:8]) extra = f" (+{len(unresolved) - 8} more)" if len(unresolved) > 8 else "" return [ f"{len(unresolved)} product value(s) are not mapped to an approved option: {preview}{extra}. {hint}" ] def unmapped_product_value_report( session: Session, candidate: V2AttributeSeedCandidate, ) -> list[dict[str, Any]]: canonical = map_canonical_data_type(candidate.proposed_data_type) if canonical not in SELECTABLE_TYPES: return [] attribute_codes = _candidate_source_codes(candidate) live_rows, excluded_rows = _value_rows_for_codes(session, attribute_codes) approved_keys: set[str] = set() for row in live_rows: if row.review_status == "merged": continue approved_keys.update(_value_lookup_keys(row)) excluded_keys: set[str] = set() for row in excluded_rows: excluded_keys.update(_value_lookup_keys(row)) if row.merge_into_value_code: approved_keys.update(_value_lookup_keys(row)) counts: dict[str, int] = {} reasons: dict[str, str] = {} for value in session.execute( select(MasterProductAttributeValue.value).where( MasterProductAttributeValue.attribute_code.in_(sorted(attribute_codes)) ) ).scalars().all(): label = str(value or "").strip() if not label: continue value_keys = set(_expand_value_key(label)) if approved_keys.intersection(value_keys): continue counts[label] = counts.get(label, 0) + 1 if excluded_keys.intersection(value_keys): reasons[label] = "excluded" else: reasons.setdefault(label, "unmapped") out: list[dict[str, Any]] = [] magento_by_id = _magento_label_by_option_id(session, candidate) for value, count in sorted(counts.items()): suggested = magento_by_id.get(value) or value out.append( { "value": value, "product_count": count, "reason": reasons.get(value, "unmapped"), "suggested_label": suggested, "looks_like_magento_option_id": bool(value.isdigit() and len(value) >= 4), } ) return out def _candidate_source_codes(candidate: V2AttributeSeedCandidate) -> set[str]: codes = {str(candidate.source_code or "").strip()} for alias in candidate.aliases or []: text = str(alias).strip() if text: codes.add(text) return {code for code in codes if code} def recount_candidate_usage(session: Session, candidate: V2AttributeSeedCandidate) -> None: """Count distinct SKUs across the survivor and its merged aliases, not summed sources.""" codes = _candidate_source_codes(candidate) if not codes: candidate.product_usage_count = 0 return skus_all: set[str] = set() skus_by_key: dict[str, set[str]] = {} for sku, value in session.execute( select(MasterProductAttributeValue.sku, MasterProductAttributeValue.value).where( MasterProductAttributeValue.attribute_code.in_(sorted(codes)) ) ).all(): sku_text = str(sku or "").strip() label = str(value or "").strip() if not sku_text or not label: continue skus_all.add(sku_text) for key in _expand_value_key(label): skus_by_key.setdefault(key, set()).add(sku_text) candidate.product_usage_count = len(skus_all) for row in _values_for_candidate(session, candidate): matched: set[str] = set() for key in _value_lookup_keys(row): matched.update(skus_by_key.get(key, set())) row.product_usage_count = len(matched) def _value_rows_for_codes( session: Session, attribute_codes: set[str], ) -> tuple[list[V2AttributeValueSeedCandidate], list[V2AttributeValueSeedCandidate]]: rows = list( session.scalars( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code.in_(sorted(attribute_codes)) ) ).all() ) live = [row for row in rows if row.review_status != "excluded"] excluded = [row for row in rows if row.review_status == "excluded"] return live, excluded def _magento_mapping_codes(session: Session, candidate: V2AttributeSeedCandidate) -> set[str]: codes = { str(candidate.magento_attribute_code or "").strip(), str(candidate.source_code or "").strip(), str(candidate.proposed_code or "").strip(), } for mapping in session.scalars( select(V2AttributeSeedMappingCandidate).where( V2AttributeSeedMappingCandidate.attribute_source_code.in_( sorted(_candidate_source_codes(candidate)) ), V2AttributeSeedMappingCandidate.channel == "magento", ) ).all(): text = str(mapping.channel_attribute_code or "").strip() if text: codes.add(text) return {code for code in codes if code} def _magento_option_rows(session: Session, candidate: V2AttributeSeedCandidate) -> list[MagentoAttributeOptionRegistry]: codes = _magento_mapping_codes(session, candidate) if not codes: return [] try: return list( session.scalars( select(MagentoAttributeOptionRegistry).where( MagentoAttributeOptionRegistry.attribute_code.in_(sorted(codes)) ) ).all() ) except (OperationalError, ProgrammingError): return [] def _magento_label_by_option_id(session: Session, candidate: V2AttributeSeedCandidate) -> dict[str, str]: labels: dict[str, str] = {} for row in _magento_option_rows(session, candidate): option_id = str(row.value_index).strip() label = str(row.option_label or "").strip() if option_id and label: labels[option_id] = label return labels def export_override_payload(session: Session) -> dict[str, Any]: attributes: dict[str, Any] = {} values: dict[str, Any] = {} for row in session.scalars(select(V2AttributeSeedCandidate)).all(): if row.review_status == "pending_review": continue attributes[row.source_code] = { "review_status": row.review_status, "proposed_code": row.proposed_code, "proposed_label": row.proposed_label, "proposed_data_type": row.proposed_data_type, "proposed_ui_control": row.proposed_ui_control, "is_variant_axis": bool(row.is_variant_axis), "is_channel_required": bool(row.is_channel_required), "lifecycle_status": row.lifecycle_status, "merge_into_code": row.merge_into_code, "notes": row.notes, } for row in session.scalars(select(V2AttributeValueSeedCandidate)).all(): if row.review_status == "pending_review": continue values.setdefault(row.attribute_source_code, {})[row.source_value_code] = { "review_status": row.review_status, "proposed_value_code": row.proposed_value_code, "proposed_label": row.proposed_label, "sort_order": int(row.sort_order or 0), "merge_into_value_code": row.merge_into_value_code, "lifecycle_status": row.lifecycle_status, } return {"attributes": attributes, "values": values} def apply_override_payload(session: Session, payload: dict[str, Any]) -> dict[str, int]: attrs = payload.get("attributes") if isinstance(payload, dict) else {} values = payload.get("values") if isinstance(payload, dict) else {} updated_attrs = 0 updated_values = 0 by_code = { str(row.source_code): row for row in session.scalars(select(V2AttributeSeedCandidate)).all() } if isinstance(attrs, dict): for code, item in attrs.items(): row = by_code.get(normalize_column_key(code)) if row is None or not isinstance(item, dict): continue _apply_attribute_override(row, item) updated_attrs += 1 if isinstance(values, dict): value_rows = list(session.scalars(select(V2AttributeValueSeedCandidate)).all()) index = {(str(row.attribute_source_code), str(row.source_value_code)): row for row in value_rows} for attr_code, items in values.items(): if not isinstance(items, dict): continue for value_code, item in items.items(): row = index.get((normalize_column_key(attr_code), normalize_column_key(value_code))) if row is None: row = index.get((normalize_column_key(attr_code), str(value_code))) if row is None or not isinstance(item, dict): continue _apply_value_override(row, item) updated_values += 1 return {"attributes": updated_attrs, "values": updated_values} def _apply_attribute_override(row: V2AttributeSeedCandidate, item: dict[str, Any]) -> None: if item.get("review_status") in REVIEW_STATUSES: row.review_status = item["review_status"] if item.get("proposed_code"): row.proposed_code = normalize_column_key(item["proposed_code"]) if item.get("proposed_label"): row.proposed_label = str(item["proposed_label"]).strip() if item.get("proposed_data_type"): row.proposed_data_type = map_canonical_data_type(item["proposed_data_type"]) if item.get("proposed_ui_control") in ADMIN_UI_CONTROLS: row.proposed_ui_control = item["proposed_ui_control"] if "is_variant_axis" in item: row.is_variant_axis = bool(item["is_variant_axis"]) if "is_channel_required" in item: row.is_channel_required = bool(item["is_channel_required"]) if item.get("lifecycle_status") in LIFECYCLE_STATUSES: row.lifecycle_status = item["lifecycle_status"] if "merge_into_code" in item: row.merge_into_code = str(item.get("merge_into_code") or "").strip() or None if "notes" in item: row.notes = str(item.get("notes") or "").strip() or None row.updated_at = datetime.now(timezone.utc) def _apply_value_override(row: V2AttributeValueSeedCandidate, item: dict[str, Any]) -> None: if item.get("review_status") in VALUE_REVIEW_STATUSES: row.review_status = item["review_status"] if item.get("proposed_value_code"): row.proposed_value_code = normalize_column_key(item["proposed_value_code"]) if item.get("proposed_label"): row.proposed_label = str(item["proposed_label"]).strip() if "sort_order" in item: row.sort_order = int(item.get("sort_order") or 0) if item.get("lifecycle_status") in LIFECYCLE_STATUSES: row.lifecycle_status = item["lifecycle_status"] if "merge_into_value_code" in item: row.merge_into_value_code = str(item.get("merge_into_value_code") or "").strip() or None row.updated_at = datetime.now(timezone.utc) def summary_counts(session: Session) -> dict[str, Any]: rows = list(session.scalars(select(V2AttributeSeedCandidate)).all()) by_status: dict[str, int] = {} for row in rows: by_status[row.review_status] = by_status.get(row.review_status, 0) + 1 ready = sum(1 for row in rows if row.review_status in APPLY_READY_STATUSES) excluded_with_usage = [ {"source_code": row.source_code, "product_usage_count": int(row.product_usage_count or 0)} for row in rows if row.review_status == "excluded" and int(row.product_usage_count or 0) > 0 ] return { "candidate_count": len(rows), "review_status_counts": by_status, "ready_to_apply": ready, "excluded_with_usage": excluded_with_usage, }