"""Top-level orchestrator for the Product Decision Engine. Workflow: Receive product JSON -> extract SKUs -> match freight per SKU -> analyze each SKU -> compare SKUs -> product-level analysis -> final recommendation -> response dict. Everything runs in memory; nothing is persisted. """ from __future__ import annotations from typing import Any, Optional from ..constants import constants as C from ..utils.helpers import (as_list, clamp, first_present, saturating_score, unwrap_response) from .analyzers.market_analyzer import MarketAnalyzer from .analyzers.product_analyzer import ProductAnalysis, ProductAnalyzer from .analyzers.recommendation_engine import RecommendationEngine from .analyzers.risk_analyzer import RiskAnalyzer from .analyzers.store_analyzer import StoreAnalyzer from .analyzers.strength_analyzer import StrengthAnalyzer from .analyzers.sku.sku_analyzer import SkuAnalyzer from .analyzers.sku.sku_decision_engine import SkuDecision, SkuDecisionEngine class DecisionEngine: """Coordinates every analyzer and assembles the final API response.""" def __init__( self, product_json: dict, freight_results: list[dict], profit_margin: float, ) -> None: # Accept both the raw aliexpress_ds_product_wholesale_get_response # envelope and an already-unwrapped result object. self._product_json = unwrap_response(product_json or {}) self._freight_by_sku = self._index_freight(freight_results or []) self._profit_margin = float(profit_margin) self._product_analyzer = ProductAnalyzer() self._store_analyzer = StoreAnalyzer() self._sku_engine = SkuDecisionEngine() self._market_analyzer = MarketAnalyzer() self._risk_analyzer = RiskAnalyzer() self._strength_analyzer = StrengthAnalyzer() self._recommendation_engine = RecommendationEngine() # ------------------------------------------------------------------ # def run(self) -> dict: """Execute the full workflow and return the response payload.""" product = self._product_analyzer.analyze(self._product_json) store = self._store_analyzer.analyze(self._product_json) popularity_score = self._popularity_score(product) store_signal = store.score if store.score > 0 else None sku_decisions = self._analyze_skus(popularity_score, store_signal) rankings = self._rank_skus(sku_decisions) market = self._market_analyzer.analyze(product, store, sku_decisions) risk = self._risk_analyzer.analyze(product, store, sku_decisions) strength = self._strength_analyzer.analyze(product, store, sku_decisions) recommendation = self._recommendation_engine.recommend( product, store, sku_decisions, market, risk, strength ) return self._build_response( product, store, sku_decisions, rankings, market, risk, strength, recommendation, ) # ------------------------------------------------------------------ # # SKU pipeline # ------------------------------------------------------------------ # def _analyze_skus(self, popularity_score: "float | None", store_score: "float | None") -> list[SkuDecision]: sku_dtos = self._extract_sku_dtos() max_supplier_price = self._max_supplier_price(sku_dtos) decisions: list[SkuDecision] = [] for sku_dto in sku_dtos: sku_id = str(first_present(sku_dto, ("sku_id", "skuId", "id"), default="")) decisions.append( self._sku_engine.analyze( sku_dto=sku_dto, freight_json=self._freight_by_sku.get(sku_id), profit_margin=self._profit_margin, popularity_score=popularity_score, store_score=store_score, max_supplier_price=max_supplier_price, ) ) return decisions def _extract_sku_dtos(self) -> list[dict]: container = first_present( self._product_json, ("ae_item_sku_info_dtos.ae_item_sku_info_d_t_o", "ae_item_sku_info_dtos", "sku_info_dtos", "skus"), ) return [dto for dto in as_list(container) if isinstance(dto, dict)] @staticmethod def _max_supplier_price(sku_dtos: list[dict]) -> float: """Anchor used to compare SKU prices against their siblings.""" analyzer = SkuAnalyzer() prices = [analyzer.analyze(dto).supplier_price for dto in sku_dtos] return max(prices, default=0.0) @staticmethod def _index_freight(freight_results: list[dict]) -> dict[str, dict]: index: dict[str, dict] = {} for entry in freight_results: if not isinstance(entry, dict): continue sku_id = str(first_present(entry, ("skuId", "sku_id"), default="")) freight = first_present(entry, ("freightJson", "freight_json", "freight")) if sku_id and isinstance(freight, dict): index[sku_id] = freight return index # ------------------------------------------------------------------ # # SKU comparison / rankings # ------------------------------------------------------------------ # @staticmethod def _rank_skus(decisions: list[SkuDecision]) -> dict[str, Optional[str]]: """Find the winner per business dimension. ``None`` when no SKU exists.""" if not decisions: return {key: None for key in ( "bestOverallSku", "lowestCostSku", "highestProfitSku", "highestMarginSku", "highestStockSku", "fastestDeliverySku", "bestValueSku", "lowestShippingCostSku", )} def winner(key_fn, reverse: bool = True) -> str: best = sorted(decisions, key=key_fn, reverse=reverse)[0] return best.sku.sku_id return { "bestOverallSku": winner(lambda d: d.result.score), "lowestCostSku": winner(lambda d: d.pricing.total_cost, reverse=False), "highestProfitSku": winner(lambda d: d.pricing.estimated_profit), "highestMarginSku": winner(lambda d: d.pricing.margin), "highestStockSku": winner(lambda d: d.sku.stock), "fastestDeliverySku": winner( lambda d: d.freight.delivery_days_avg if d.freight.delivery_days_avg is not None else 999.0, reverse=False, ), # Best value: most decision-score per unit of landed cost. "bestValueSku": winner( lambda d: d.result.score / d.pricing.total_cost if d.pricing.total_cost > 0 else 0.0 ), "lowestShippingCostSku": winner( lambda d: d.freight.shipping_cost, reverse=False ), } # ------------------------------------------------------------------ # # Shared signals # ------------------------------------------------------------------ # @staticmethod def _popularity_score(product: ProductAnalysis) -> "float | None": """Product-level popularity shared by every SKU: sales + reviews. Uses a saturating (square-root) curve so moderate history earns a moderate score. Returns ``None`` when neither metric is available, letting downstream weights renormalize instead of scoring zero. """ if product.sales_count <= 0 and product.review_count <= 0: return None return clamp( saturating_score(product.sales_count, C.SALES_FOR_FULL_POPULARITY) * 0.6 + saturating_score(product.review_count, C.REVIEWS_FOR_FULL_POPULARITY) * 0.4 ) # ------------------------------------------------------------------ # # Response assembly # ------------------------------------------------------------------ # def _build_response( self, product, store, sku_decisions: list[SkuDecision], rankings: dict, market, risk, strength, recommendation, ) -> dict: best = ( max(sku_decisions, key=lambda d: d.result.score) if sku_decisions else None ) return { "decision": recommendation.decision, "confidence": round(recommendation.confidence), "overallScore": round(recommendation.overall_score), "bestSku": ( {"skuId": best.sku.sku_id, "reason": "Highest overall score"} if best else None ), "rankings": rankings, "market": market.to_dict(), "pricing": best.pricing.to_dict() if best else None, "store": store.to_dict(), "product": product.to_dict(), "scores": { "overall": round(recommendation.overall_score, 2), "breakdown": recommendation.score_breakdown and {k: round(v, 2) for k, v in recommendation.score_breakdown.items()}, }, "skuAnalysis": [d.to_dict() for d in sku_decisions], "strengths": strength.strengths, "risks": risk.risks, "summary": recommendation.summary, "confidenceReasons": recommendation.confidence_reasons, "marketRecommendation": recommendation.market_recommendation, "skuRecommendation": recommendation.sku_recommendation, }