Spaces:
Sleeping
Sleeping
File size: 3,218 Bytes
b2be963 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 | 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
|