""" Small, dependency-free helpers shared by every analyzer. AliExpress payloads are inconsistent (strings for numbers, lists wrapped in single-key dicts, missing keys), so all extraction goes through these defensive utilities. """ from __future__ import annotations import re from typing import Any, Iterable, Optional def safe_get(data: Any, path: str, default: Any = None) -> Any: """Read a nested value using a dotted path, e.g. ``"a.b.c"``. Returns ``default`` when any segment is missing or not a dict. """ current = data for key in path.split("."): if not isinstance(current, dict) or key not in current: return default current = current[key] return current def first_present(data: dict, keys: Iterable[str], default: Any = None) -> Any: """Return the first non-empty value among several candidate keys.""" for key in keys: value = safe_get(data, key) if value not in (None, "", [], {}): return value return default def to_float(value: Any, default: float = 0.0) -> float: """Coerce AliExpress numeric strings ('27.78', '19,00', 'USD 5.99') to float.""" if value is None: return default if isinstance(value, (int, float)): return float(value) match = re.search(r"-?\d+(?:[.,]\d+)?", str(value)) if not match: return default return float(match.group().replace(",", ".")) def to_int(value: Any, default: int = 0) -> int: return int(to_float(value, float(default))) def round2(value: float) -> float: """Round monetary / percentage values to two decimal places.""" return round(float(value) + 0.0, 2) def clamp(value: float, low: float = 0.0, high: float = 100.0) -> float: """Keep a score inside its valid range.""" return max(low, min(high, value)) def as_list(value: Any) -> list: """Normalize AliExpress list containers. The API often wraps lists as ``{"some_key_dto": [...]}`` or returns a single dict instead of a list. This flattens all of those to a plain list. """ if value is None: return [] if isinstance(value, list): return value if isinstance(value, dict): # Unwrap single-key containers such as {"ae_item_sku_info_d_t_o": [...]} if len(value) == 1: inner = next(iter(value.values())) if isinstance(inner, list): return inner if isinstance(inner, dict): return [inner] return [value] return [value] def linear_score(value: float, best: float, worst: float) -> float: """Map ``value`` to 0-100 where ``best`` earns 100 and ``worst`` earns 0. Works in both directions (best < worst means "lower is better"). """ if best == worst: return 100.0 if value == best else 0.0 ratio = (worst - value) / (worst - best) return clamp(ratio * 100.0) def ratio_score(value: float, full_score_at: float) -> float: """Map ``value`` to 0-100, saturating at ``full_score_at``.""" if full_score_at <= 0: return 0.0 return clamp((value / full_score_at) * 100.0) def unwrap_response(data: Any) -> dict: """Peel AliExpress API envelopes off a payload. Real API responses arrive wrapped, e.g.:: {"aliexpress_ds_product_wholesale_get_response": {"result": {...}}} {"aliexpress_ds_freight_query_response": {"result": {...}}} This walks down through any ``*_response`` wrapper and any ``result`` container until the actual data object is reached. Payloads that are already unwrapped pass through unchanged. """ if not isinstance(data, dict): return {} current = data for _ in range(4): # envelopes are at most a couple of levels deep descended = False for key, value in current.items(): if key.endswith("_response") and isinstance(value, dict): current = value descended = True break if isinstance(current.get("result"), dict): current = current["result"] descended = True if not descended: break return current def parse_delivery_days(raw: Any) -> tuple[Optional[int], Optional[int]]: """Parse AliExpress delivery time strings. Accepts values like ``"12-20"``, ``"15"``, ``"7 days"`` or integers. Returns ``(min_days, max_days)`` or ``(None, None)`` when unknown. """ if raw is None: return None, None if isinstance(raw, (int, float)): days = int(raw) return days, days numbers = [int(n) for n in re.findall(r"\d+", str(raw))] if not numbers: return None, None if len(numbers) == 1: return numbers[0], numbers[0] return min(numbers[0], numbers[1]), max(numbers[0], numbers[1]) def weighted_average(items: "list[tuple[Optional[float], float]]") -> float: """Weighted average of (score, weight) pairs on a 0-100 scale. Missing metrics are passed as ``None`` scores and are *excluded*; the remaining weights are renormalized so unavailable data never drags the result toward zero. Returns 0.0 when nothing is available. """ available = [(s, w) for s, w in items if s is not None and w > 0] total_weight = sum(w for _, w in available) if total_weight <= 0: return 0.0 return clamp(sum(s * w for s, w in available) / total_weight) def piecewise_score(value: float, anchors: "list[tuple[float, float]]") -> float: """Linear interpolation over (value, score) anchor points. Produces smooth, gradually increasing scores instead of hard cliffs. Anchors must be sorted by value; values outside the range clamp to the first / last anchor's score. """ if not anchors: return 0.0 if value <= anchors[0][0]: return clamp(anchors[0][1]) for (x1, y1), (x2, y2) in zip(anchors, anchors[1:]): if value <= x2: ratio = (value - x1) / (x2 - x1) if x2 != x1 else 1.0 return clamp(y1 + ratio * (y2 - y1)) return clamp(anchors[-1][1]) def saturating_score(value: float, full_score_at: float) -> float: """Square-root saturation curve on a 0-100 scale. Rewards early progress: 10% of the saturation point already earns ~32 points instead of 10. Used for volume metrics (sales, reviews) so that moderate history is not punished as if it were zero. """ if full_score_at <= 0 or value <= 0: return 0.0 return clamp(((value / full_score_at) ** 0.5) * 100.0)