70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
"""Contract / quote product package_spec codec: JSON array with legacy scalar support."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any, Iterable, List, Optional
|
|
|
|
|
|
MAX_PACKAGE_SPECS_PER_LINE = 20
|
|
|
|
|
|
def normalize_package_specs_list(values: Optional[Iterable[Any]]) -> List[str]:
|
|
"""Trim, drop empties, dedupe while preserving order."""
|
|
out: List[str] = []
|
|
seen = set()
|
|
for raw in values or []:
|
|
s = str(raw or "").strip()
|
|
if not s or s in seen:
|
|
continue
|
|
seen.add(s)
|
|
out.append(s)
|
|
if len(out) >= MAX_PACKAGE_SPECS_PER_LINE:
|
|
break
|
|
return out
|
|
|
|
|
|
def decode_package_specs(value: Any) -> List[str]:
|
|
"""Decode DB/API stored package_spec into a list of specs.
|
|
|
|
- JSON array of strings → list (normalized)
|
|
- legacy non-empty scalar → one-element list
|
|
- empty / None → []
|
|
"""
|
|
if value is None:
|
|
return []
|
|
if isinstance(value, (list, tuple)):
|
|
return normalize_package_specs_list(value)
|
|
text = str(value).strip()
|
|
if not text:
|
|
return []
|
|
if text.startswith("["):
|
|
try:
|
|
parsed = json.loads(text)
|
|
if isinstance(parsed, list):
|
|
return normalize_package_specs_list(parsed)
|
|
except (json.JSONDecodeError, TypeError, ValueError):
|
|
pass
|
|
return [text]
|
|
|
|
|
|
def encode_package_specs(specs: Optional[Iterable[Any]]) -> str:
|
|
"""Encode list to JSON array string for DB; empty list → ''."""
|
|
normalized = normalize_package_specs_list(specs)
|
|
if not normalized:
|
|
return ""
|
|
return json.dumps(normalized, ensure_ascii=False)
|
|
|
|
|
|
def resolve_package_specs_from_payload(data: Optional[dict]) -> List[str]:
|
|
"""Prefer packageSpecs / package_specs; fallback to scalar packageSpec / package_spec."""
|
|
if not data:
|
|
return []
|
|
if "packageSpecs" in data and data.get("packageSpecs") is not None:
|
|
return normalize_package_specs_list(data.get("packageSpecs"))
|
|
if "package_specs" in data and data.get("package_specs") is not None:
|
|
return normalize_package_specs_list(data.get("package_specs"))
|
|
for key in ("packageSpec", "package_spec"):
|
|
if key in data and data.get(key) is not None:
|
|
return decode_package_specs(data.get(key))
|
|
return []
|