import os import uuid import logging from typing import List, Optional from pydantic import BaseModel from apify_client import ApifyClient from src.agents.selector_agent import ProductIdea as AgentProductIdea from src.models.product_idea import ProductIdea as DBProductIdea, ProductStatus from src.models.database import get_session logger = logging.getLogger(__name__) class ScrapingCriteria(BaseModel): target_platform: str = "Amazon" min_reviews_per_month: int = 1000 max_bsr_rank: int = 10000 category: Optional[str] = None keyword: Optional[str] = "trending products" max_items: int = 10 def mock_amazon_scraper(criteria: ScrapingCriteria) -> List[dict]: """Fallback mock scraper if Apify fails or token is missing""" return [ { "title": "Ergonomic Office Chair with Lumbar Support", "source_url": "https://www.amazon.com/dp/B08MockChair1", "primary_image_url": "https://cbu01.alicdn.com/img/ibank/O1CN01fN1pC6228sM5kF2s3_!!2214643033501-0-cib.jpg", "reviews": 1250, "bsr": 4500, "est_cny": 150.0, "retail_usd": 89.99 } ] def apify_amazon_scraper(criteria: ScrapingCriteria) -> List[dict]: """ Real scraper that uses Apify to extract high-demand products from Amazon. """ token = os.environ.get("APIFY_API_TOKEN") if not token: logger.warning("APIFY_API_TOKEN is not set. Falling back to mock data for safety.") return mock_amazon_scraper(criteria) client = ApifyClient(token) logger.info(f"Running Apify Amazon Scraper for keyword: {criteria.keyword}...") # Using 'junglee/amazon-crawler' as the standard Amazon actor on Apify run_input = { "categoryOrProductUrls": [ {"url": f"https://www.amazon.com/s?k={criteria.keyword.replace(' ', '+')}"} ], "maxItemsPerStartUrl": criteria.max_items, "useCaptchaSolver": True, } # Run the Actor and wait for it to finish try: run = client.actor("junglee/amazon-crawler").call(run_input=run_input) raw_products = [] # In apify-client v3+, run is an object, not a dict dataset_id = getattr(run, "defaultDatasetId", getattr(run, "default_dataset_id", None)) if dataset_id is None and isinstance(run, dict): dataset_id = run.get("defaultDatasetId") for item in client.dataset(dataset_id).iterate_items(): # Parse the Apify Amazon Crawler output format # Ensure safe fallback using .get() for missing fields title = item.get("title", "Unknown Title") url = item.get("url", "") # Reviews parsing reviews_count = item.get("reviewsCount", 0) if not isinstance(reviews_count, (int, float)): try: reviews_count = int(str(reviews_count).replace(',', '')) except ValueError: reviews_count = 0 # Price parsing price_dict = item.get("price", {}) price_usd = price_dict.get("value", 0.0) if isinstance(price_dict, dict) else 0.0 # For cost_cny, we assume a mockup logic (e.g. 30% of retail price * 7.2) # In a real scenario, this comes from 1688 image search est_cny = (price_usd * 0.3) * 7.2 # BSR Parsing (Apify usually returns a list or string for rank) # We mock BSR if not perfectly extracted, as different actors format it differently bsr = 5000 # Default safe mock for BSR unless parsed # Extract Image URL image_url = item.get("thumbnailUrl") or item.get("image") or "" raw_products.append({ "title": title, "source_url": url, "primary_image_url": image_url, "reviews": reviews_count, "bsr": bsr, "est_cny": est_cny, "retail_usd": price_usd }) logger.info(f"Apify returned {len(raw_products)} raw items.") return raw_products except Exception as e: logger.error(f"Apify scraping failed: {str(e)}") logger.warning("Falling back to mock data.") return mock_amazon_scraper(criteria) def run_scraper_agent(criteria: ScrapingCriteria) -> List[AgentProductIdea]: """ The Autonomous Scraper Agent (Epic 8). Takes in scraping criteria, runs the extraction, and persists ProductIdea into Turso DB. """ logger.info(f"Scraper Agent started with criteria: {criteria.model_dump()}") raw_products = apify_amazon_scraper(criteria) ideas: List[AgentProductIdea] = [] session = next(get_session()) for raw in raw_products: if raw["reviews"] >= criteria.min_reviews_per_month and raw["bsr"] <= criteria.max_bsr_rank: idea_id = str(uuid.uuid4()) idea = AgentProductIdea( id=idea_id, source_url=raw["source_url"], sales_volume=raw["reviews"] * 3, est_total_cost_cny=raw["est_cny"], recommended_retail_price_usd=raw["retail_usd"], status="SCRAPED" ) ideas.append(idea) db_idea = DBProductIdea( id=idea_id, source_url=raw["source_url"], primary_image_url=raw.get("primary_image_url", ""), est_total_cost_cny=raw["est_cny"], recommended_retail_price_usd=raw["retail_usd"], zone_1_shipping_usd=10.0, status=ProductStatus.SCRAPED, shopify_title=raw.get("title", "Amazon Item") ) session.add(db_idea) logger.info(f"Generated new ProductIdea: {idea.id} from {idea.source_url}") else: logger.warning(f"Discarding product {raw['title']} - Failed criteria check (Reviews: {raw['reviews']}, BSR: {raw['bsr']})") session.commit() logger.info(f"Scraper Agent completed. Persisted {len(ideas)} new product ideas to DB.") return ideas if __name__ == "__main__": import argparse from dotenv import load_dotenv load_dotenv("../../.env") # Load the centralized env parser = argparse.ArgumentParser(description="Run Scraper Agent") parser.add_argument("--keyword", type=str, default="Home & Kitchen gadgets", help="Search keyword") parser.add_argument("--max-items", type=int, default=5, help="Max items to scrape") args = parser.parse_args() logging.basicConfig(level=logging.INFO) criteria = ScrapingCriteria(target_platform="Amazon", keyword=args.keyword, max_items=args.max_items) discovered_ideas = run_scraper_agent(criteria) for i in discovered_ideas: logger.info(i.model_dump_json(indent=2))