Spaces:
Sleeping
Sleeping
| import json | |
| import os | |
| import random | |
| import httpx | |
| from typing import List, Dict, Any | |
| from sqlalchemy.orm import Session | |
| from app.models.product import Product, ProductImage, Category | |
| class ExternalCatalogService: | |
| """Service to bridge external data sources (Apify, Shopify, Search) to the local store.""" | |
| def __init__(self, db: Session): | |
| self.db = db | |
| self.discovery_file = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "discovered_products.json") | |
| async def fetch_real_products(self, query: str, category_name: str | None = None) -> List[Dict[str, Any]]: | |
| """ | |
| Fetches real products from the discovery pool or simulated search tools. | |
| """ | |
| if os.path.exists(self.discovery_file): | |
| with open(self.discovery_file, "r", encoding="utf-8") as f: | |
| pool = json.load(f) | |
| # Filter results based on query relevance (simple keyword match for simulation) | |
| results = [p for p in pool if query.lower() in p["title"].lower() or query.lower() in p.get("name_ar", "").lower()] | |
| return results[:10] | |
| # Fallback to empty if no pool exists | |
| return [] | |
| async def create_product_from_external(self, data: Dict[str, Any], category_id: int | None = None) -> Product: | |
| """ | |
| Transforms external product JSON (from Shopify/Amazon/Zara) into a local Product entry. | |
| """ | |
| # Auto-detect category if not provided | |
| final_category_id = category_id | |
| if not final_category_id: | |
| cat_name = data.get("category_name") | |
| if cat_name: | |
| cat = self.db.query(Category).filter(Category.name_ar == cat_name).first() | |
| if cat: | |
| final_category_id = cat.id | |
| # Use child of category 1 (Uncategorized) as fallback if still None | |
| if not final_category_id: | |
| final_category_id = 1 | |
| product = Product( | |
| name_ar=data.get("name_ar", data.get("title", "منتج جديد")), | |
| name_en=data.get("name_en", data.get("title", "New Product")), | |
| description_ar=data.get("description", ""), | |
| description_en=data.get("description", ""), | |
| price=float(data.get("price", 0)), | |
| compare_price=float(data.get("compare_price", 0)) or float(data.get("price", 0)) * 1.2, | |
| stock=random.randint(5, 50), | |
| category_id=final_category_id, | |
| rating=float(data.get("rating", 4.5)), | |
| rating_count=random.randint(5, 100), | |
| specs=data.get("specs", {}), | |
| is_active=True, | |
| is_featured=random.choice([True, False]) | |
| ) | |
| self.db.add(product) | |
| self.db.flush() | |
| images = data.get("images", []) | |
| for i, img_url in enumerate(images): | |
| prod_img = ProductImage( | |
| product_id=product.id, | |
| image_url=img_url, | |
| alt_text=product.name_ar, | |
| sort_order=i | |
| ) | |
| self.db.add(prod_img) | |
| return product | |