def _extract_used_lowest_display(self, item: Dict[

def _extract_used_lowest_display(self, item: Dict[str, Any]) -> Optional[str]: offers = item.get("offers") or {} sums = offers.get("summaries") or [] if isinstance(sums, dict): sums = [sums] for s in sums: cond_txt = _to_text(s.get("condition")).lower() if any(k in cond_txt for k in ("used", "refurb", "open", "مستعمل")): lp = (s.get("lowest_price") or {}) disp = lp.get("display_amount") or (str(lp.get("amount")) if lp.get("amount") is not None else None) if disp: return disp listings = offers.get("listings") or [] if isinstance(listings, dict): listings = [listings] for lst in listings: cond_txt = _to_text(lst.get("condition")).lower() if any(k in cond_txt for k in ("used", "refurb", "open", "مستعمل")): price = (lst.get("price") or {}) disp = price.get("display_amount") or (str(price.get("amount")) if price.get("amount") is not None else None) if disp: return disp return None def enrich_used_info(self, products: List[Dict[str, Any]], partner_tag: str, used_only: bool = False) -> None: if not products or self.api_disabled or not self.api: return need_asins: List[str] = [] for p in products: asin = p.get("asin") if not asin: continue used_disp = (p.get("used_lowest_display") or "").strip() # فقط ابحث إذا السعر المستعمل غير موجود if not used_disp: need_asins.append(asin) if not need_asins: return enriched = self.get_items_by_asins(list(dict.fromkeys(need_asins))) by_asin: Dict[str, Dict[str, Any]] = {} for it in enriched: d = _as_dict(it) a = d.get("asin") if a: by_asin[a] = d # فقط حدّث المنتجات اللي طلبنا بياناتها (اللي ما كان عندها used_disp) for p in products: asin = p.get("asin") if not asin: continue # إذا المنتج في by_asin، معناها جبنا بيانات جديدة إله if asin in by_asin: ui = self._extract_used_lowest_display(by_asin[asin]) if ui: p["used_lowest_display"] = _format_price(ui) p["has_used_offer"] = True if not p.get("link_used"): p["link_used"] = build_sa_affiliate_link(asin, partner_tag, used=True) # ⚠️ لا نستبدل price_after أبداً - نخليه السعر الجديد # السعر المستعمل موجود في used_lowest_display منفصل # ====== Keep-trying Category Deals ====== def fetch_category_deals(self, category_key: str, limit: int = DEFAULT_POST_COUNT, ensure_count: bool = True, used_only: bool = False) -> List[Dict[str, Any]]: """تجلب منتجات مخفّضة لفئة معينة. إذا used_only=True نحاول جلب المستعمل من المصدر أولاً.""" if self.api_disabled or not self.api: return [] need = clamp_count(limit) search_index = SEARCH_INDEX_BY_CATEGORY.get(category_key) if not search_index: return [] collected_raw: List[Dict[str, Any]] = [] seen_asin: set[str] = set() state_path = getattr(self.cfg, "state_path", "paapi_cursor.json") state = _state_load(state_path) def cursor_key(sort_name: str) -> str: return f"{category_key}::{sort_name}::USED::{int(used_only)}" cycles = 0 max_cycles = KEEP_SEARCH_MAX_CYCLES if ensure_count else 1 deadline = time.time() + KEEP_SEARCH_MAX_MINUTES * 60 if ensure_count else time.time() + 1

mapped.append( { "asin": it.get("asin"), "parent_asin": parent_asin, "title": title, "image": image, "all_images": all_variant_images if all_variant_images else [image], # صور الألوان المختلفة "price_before": price_before_txt, "price_after": price_after_txt, "discount_pct": float(pct) if pct is not None else 0.0, "savings_amount": prices["savings_amount"], "rating": rating_val, "rating_count": rating_cnt, "rating_text": (f"{round(rating_val,1)}/5" if rating_val is not None else None), "about": about, "link": build_sa_affiliate_link(it.get("asin"), self.cfg.pa_partner_tag) if it.get("asin") else None, "link_used": build_sa_affiliate_link(it.get("asin"), self.cfg.pa_partner_tag, used=True) if it.get("asin") and used_info["has_used_offer"] else None, "is_catch": bool((pct or 0) >= CATCH_MIN_PCT and (prices.get("savings_amount") or 0) >= CATCH_MIN_SAVE), # used "has_used_offer": used_info["has_used_offer"], "used_lowest_display": used_info["used_lowest_display"], "used_offer_count": used_info["used_offer_count"], "used_examples": used_info["used_examples"], # prime ✅ "is_prime_exclusive": prime_info["is_prime_exclusive"], "is_amazon_fulfilled": prime_info["is_amazon_fulfilled"], "is_prime_pantry": prime_info["is_prime_pantry"], # promo end (اختياري) "promo_end_iso": promo_end, } ) if not mapped: return [] # ⚠️ لا نفلتر has_used_offer لما used_only=True # لأن المنتجات أصلاً كلها مستعملة (جايين من condition="Used") mapped.sort( key=lambda x: ( -int(bool(x.get("has_used_offer"))), -(x.get("discount_pct") or 0), -(x.get("rating_count") or 0), -(x.get("rating") or 0), x.get("asin") or "", ) ) return mapped[:need] async def enhance_products_with_scraping(self, products: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """تحسين المنتجات بإضافة بيانات الكوبونات من web scraping""" if not self.enable_web_scraping or not products: return products # اختيار عينة من المنتجات للـ scraping (توفيراً للوقت والموارد) sample_size = min(self.cfg.scraping_sample_size, len(products)) selected_products = products[:sample_size] # استخراج ASINs asins_to_scrape = [p.get("asin") for p in selected_products if p.get("asin")] if not asins_to_scrape: return products # Web scraping مُعطل مؤقتاً - إرجاع المنتجات كما هي print("⚠️ نظام Web scraping مُعطل مؤقتاً") return products # ====== Keep-trying Global Catches ====== def fetch_global_catches(self, limit: int = 4, keep_trying: bool = True) -> List[Dict[str, Any]]: need = max(1, min(50, int(limit or 4))) collected: List[Dict[str, Any]] = [] seen_asin: set[str] = set() max_cycles = KEEP_SEARCH_MAX_CYCLES if keep_trying else 1 deadline = time.time() + KEEP_SEARCH_MAX_MINUTES * 60 if keep_trying else time.time() + 1 state_path = getattr(self.cfg, "state_path", "paapi_cursor.json") state = _state_load(state_path) def cursor_key(category_key: str, sort_name: str) -> str: return f"CATCH::{category_key}::{sort_name}" cycles = 0 while len(collected) cycles += 1

collected.append( { "asin": asin, "parent_asin": parent_asin, "title": title, "image": image, "all_images": all_variant_images if all_variant_images else [image], # صور الألوان المختلفة "price_before": price_before_txt, "price_after": price_after_txt, "final_price": final_price_txt, # السعر النهائي "discount_pct": float(pct), "savings_amount": sav, # العروض الجانبية "promotions": prices.get("promotions", []), "coupon_value": prices.get("coupon_value", 0.0), "lightning_deal": prices.get("lightning_deal", False), "deal_end_time": prices.get("deal_end_time"), "rating": rating_val, "rating_count": rating_cnt, "rating_text": (f"{round(rating_val,1)}/5" if rating_val is not None else None), "about": about, "link": build_sa_affiliate_link(asin, self.cfg.pa_partner_tag), "is_catch": True, # condition info "condition": "used", # prime ✅ "is_prime_exclusive": prime_info["is_prime_exclusive"], "is_amazon_fulfilled": prime_info["is_amazon_fulfilled"], "is_prime_pantry": prime_info["is_prime_pantry"], # promo end "promo_end_iso": promo_end, # البيانات الخام "_raw_amazon_data": it, } ) if len(collected) >= need: break if len(collected) >= need: break if len(collected) >= need: break if len(collected) time.sleep(KEEP_SEARCH_SLEEP_SECONDS) collected.sort(key=lambda x: (-(x.get("savings_amount") or 0), -(x.get("discount_pct") or 0))) return collected[:need] def get_product_variations(self, asin: str) -> Dict[str, Any]: try: print(f"🔍 get_product_variations called with ASIN: '{asin}'") if self.api_disabled or not self.api or not asin: print(f"🔍 API disabled or no ASIN") return {"success": False, "error": "API غير متاح"}

# 1) get parent print(f"🔍 Calling get_items with ASIN: '{asin}'") r = self.api.get_items(items=[asin]) print(f"🔍 get_items response type: {type(r)}") # التعامل مع list أو object items = None if isinstance(r, list): items = r print(f"🔍 Response is list with {len(items)} items") elif hasattr(r, 'items'): items = r.items print(f"🔍 Response has items attr with {len(items)} items") if not items: print(f"🔍 No items returned from get_items") return {"success": False, "error": "المنتج غير موجود"} main_item = _as_dict(items[0]) # التحقق من أن المنتج ليس مستعملاً condition = main_item.get("condition") or "" if isinstance(condition, dict): condition = condition.get("value", "").lower() else: condition = str(condition).lower() if any(k in condition for k in ("used", "refurbished", "مستعمل", "مجدد")): print(f"🔍 Product is used/refurbished, skipping variations") return {"success": False, "error": "المنتج مستعمل", "has_variations": False} parent_asin = main_item.get("parent_asin") or asin print(f"🔍 Got main_item, parent_asin: {parent_asin}") # 2) page through GetVariations variations, seen = [], set() for page in range(1, 6): # up to 5 pages gv = getattr(self.api, "get_variations", None) if not callable(gv): print(f"🔍 get_variations not callable") break try: res = self.api.get_variations( asin=parent_asin, variation_page=page, variation_count=min(self.cfg.paapi_items_per_page, 10) ) except TypeError: # fallback للمكتبات القديمة try: res = self.api.get_variations( asin=parent_asin, variation_page=page ) except Exception as e: print(f"🔍 get_variations error page {page}: {e}") break except Exception as e: # Amazon قد ترجع "No variation items have been found" print(f"🔍 get_variations exception page {page}: {e}") break # التعامل مع list أو object items = None if isinstance(res, list): items = res elif hasattr(res, 'items'): items = res.items if not items: print(f"🔍 No items in page {page}") break print(f"🔍 Page {page}: got {len(items)} items") for it in items: d = _as_dict(it) child = d.get("asin") if not child or child in seen: continue var_info = self._extract_variation_info(d, d.get("variation_attributes") or []) if var_info: variations.append(var_info) seen.add(child) print(f"🔍 Added variation: {var_info.get('variant_name')}") # 3) fallback: cautious keyword search constrained to same parent

"link": build_sa_affiliate_link(asin, self.cfg.pa_partner_tag), "price_before": prices.get("list_disp"), "price_after": prices.get("current_disp"), "price_before_amount": prices.get("list_amount"), "price_after_amount": prices.get("current_amount"), "savings_amount": prices.get("savings_amount"), "savings_pct": prices.get("savings_pct"), "has_discount": bool(prices.get("savings_pct") and prices.get("savings_pct") > 0) } except Exception as e: print(f"خطأ في استخراج variation info: {e}") return None