shanxia2.0/backend/app/utils/crm_package_spec.py

"""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 []