"""Serialize Python/master values for Shopify metafield GraphQL APIs.""" from __future__ import annotations import json from decimal import Decimal, InvalidOperation from typing import Any, List, Optional, Sequence from channel.attribute_value_resolver import resolve_shopify_choice_label def choices_from_metafield_definition(raw_json: Optional[dict]) -> Optional[List[str]]: """Parse Shopify metafield definition validations into allowed choice labels.""" if not isinstance(raw_json, dict): return None for validation in raw_json.get("validations") or []: if not isinstance(validation, dict): continue if str(validation.get("name") or "").strip().lower() != "choices": continue raw = validation.get("value") if isinstance(raw, list): choices = [str(item).strip() for item in raw if str(item).strip()] return choices or None if isinstance(raw, str) and raw.strip(): try: parsed = json.loads(raw) except json.JSONDecodeError: continue if isinstance(parsed, list): choices = [str(item).strip() for item in parsed if str(item).strip()] return choices or None return None def serialize_shopify_metafield_value( value: Any, data_type: str, *, choices: Optional[Sequence[str]] = None, ) -> Optional[str]: """Return a Shopify-compatible metafield value string, or None to omit the field.""" if value is None: return None dtype = str(data_type or "single_line_text_field").strip().lower() if "metaobject" in dtype: return _serialize_metaobject_reference(value) if dtype == "boolean": return _serialize_boolean(value) if dtype == "number_integer": return _serialize_integer(value) if dtype == "number_decimal": return _serialize_decimal(value) if dtype == "json": return _serialize_json(value) if dtype == "rich_text_field": return _serialize_rich_text(value) if dtype.startswith("list."): if choices: resolved = _resolve_list_choices(value, choices, multi=dtype.startswith("list.list")) if not resolved: return None return json.dumps(resolved, ensure_ascii=False) return _serialize_list(value) if isinstance(value, bool): return "true" if value else "false" if isinstance(value, (list, dict)): if not value: return None return json.dumps(value, ensure_ascii=False) text = str(value).strip() return text or None def _serialize_boolean(value: Any) -> Optional[str]: if isinstance(value, bool): return "true" if value else "false" text = str(value).strip().lower() if text in {"", "null", "none", "nan"}: return None if text in {"1", "true", "yes", "y", "on"}: return "true" if text in {"0", "false", "no", "n", "off"}: return "false" return None def _serialize_integer(value: Any) -> Optional[str]: if isinstance(value, bool): return "1" if value else "0" if isinstance(value, int) and not isinstance(value, bool): return str(value) if isinstance(value, float): if value != value: # NaN return None if value.is_integer(): return str(int(value)) return None text = str(value).strip() if not text: return None try: number = Decimal(text.replace(",", "")) except InvalidOperation: return None if number != number.to_integral_value(): return None return str(int(number)) def _serialize_decimal(value: Any) -> Optional[str]: if isinstance(value, bool): return "1" if value else "0" if isinstance(value, (int, float, Decimal)): if isinstance(value, float) and value != value: return None text = format(Decimal(str(value)).normalize(), "f") return text.rstrip("0").rstrip(".") if "." in text else text text = str(value).strip().replace(",", "") if not text: return None try: number = Decimal(text) except InvalidOperation: return None text = format(number.normalize(), "f") return text.rstrip("0").rstrip(".") if "." in text else text def _serialize_rich_text(value: Any) -> Optional[str]: if isinstance(value, dict): if not value: return None return json.dumps(value, ensure_ascii=False) text = str(value).strip() if not text: return None try: parsed = json.loads(text) except json.JSONDecodeError: parsed = None if isinstance(parsed, dict) and parsed.get("type") == "root": return text return json.dumps( { "type": "root", "children": [ { "type": "paragraph", "children": [{"type": "text", "value": text}], } ], }, ensure_ascii=False, ) def _serialize_json(value: Any) -> Optional[str]: if isinstance(value, (list, dict)): if not value: return None return json.dumps(value, ensure_ascii=False) text = str(value).strip() if not text: return None try: parsed = json.loads(text) except json.JSONDecodeError: return json.dumps({"value": text}, ensure_ascii=False) if isinstance(parsed, (dict, list)): return text return json.dumps({"value": parsed}, ensure_ascii=False) def _serialize_metaobject_reference(value: Any) -> Optional[str]: text = str(value).strip() if text.startswith("gid://shopify/Metaobject/"): return text return None def _resolve_list_choices(value: Any, choices: Sequence[str], *, multi: bool) -> Optional[List[str]]: if isinstance(value, list): parts = [str(item).strip() for item in value if str(item).strip()] else: text = str(value).strip() if not text: return None if text.startswith("[") and text.endswith("]"): try: parsed = json.loads(text) if isinstance(parsed, list): parts = [str(item).strip() for item in parsed if str(item).strip()] else: parts = [text] except json.JSONDecodeError: parts = [part.strip() for part in text.replace("|", ",").split(",") if part.strip()] else: parts = [part.strip() for part in text.replace("|", ",").split(",") if part.strip()] resolved: List[str] = [] for part in parts: match = resolve_shopify_choice_label(part, choices) if match and match not in resolved: resolved.append(match) if not resolved: return None if multi: return resolved return [resolved[0]] def _serialize_list(value: Any) -> Optional[str]: if isinstance(value, list): items = [str(v).strip() for v in value if str(v).strip()] else: text = str(value).strip() if not text: return None if text.startswith("[") and text.endswith("]"): try: parsed = json.loads(text) if isinstance(parsed, list): return text except json.JSONDecodeError: pass items = [part.strip() for part in text.replace("|", ",").split(",") if part.strip()] if not items: return None return json.dumps(items, ensure_ascii=False)