pip install crawl4ai pydantic deepseek requests python-dotenv

# for clonning DeepSeek
git clone https://github.com/deepseek-ai/DeepSeek-LLM.git

# for local model management
pip install lmstudio

export DEEPSEEK_API_KEY="your_deepseek_key"
export SCRAPEDO_API_KEY="your_scrapedo_key"
export CRAWL4AI_API_KEY="your_crawl4ai_key" # This line is usually unnecessary unless Crawl4AI changes architecture

import asyncio
import os
import json
import re
from typing import List
from pydantic import BaseModel
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode, LLMConfig
from crawl4ai.extraction_strategy import LLMExtractionStrategy

# Load API key (currently hardcoded)
DEEPSEEK_API_KEY = "sk-3b566c3a561a49c0bbd0e642adcd5385" # Replace with your actual key or load from env

if not DEEPSEEK_API_KEY:
    print("Warning: DEEPSEEK_API_KEY not found. LLM extraction will be skipped.")

# Define the schema for individual products
class ProductItem(BaseModel):
    title: str
    price: str
    rating: str
    link: str

# Define the schema for the complete response (list of products)
class ProductList(BaseModel):
    products: List[ProductItem]

# Improved extraction prompt
extraction_prompt = """
You are an AI that extracts product listings from an Amazon search results page.
Analyze the content and extract all visible product listings.

For each product, extract:
- title: The product name/title
- price: The price (include currency symbol, use "N/A" if not found)
- rating: Star rating (use "N/A" if not found)
- link: The product URL (make it a complete URL starting with https://www.amazon.com)

Return the results as a JSON object with a "products" key containing an array of product objects.
Only include products that have at least a title and are clearly product listings.

Example format:
{
  "products": [
    {
      "title": "Gaming Chair XYZ",
      "price": "$299.99",
      "rating": "4.5 stars",
      "link": "https://www.amazon.com/dp/B123456789"
    }
  ]
}
"""

async def scrape_amazon_with_llm(search_query="gaming+chairs"):
    """
    Scrape Amazon product listings using LLM extraction.
    Assumes DEEPSEEK_API_KEY is available (checked by the caller).
    """
    url = f"https://www.amazon.com/s?k={search_query}"

    try:
        llm_config = LLMConfig(
            provider="openai/gpt-4o-mini",
            api_token=DEEPSEEK_API_KEY,
        )
        extraction_strategy = LLMExtractionStrategy(
            llm_config=llm_config,
            schema=ProductList.model_json_schema(),
            extraction_type="schema",
            instruction=extraction_prompt,
            chunk_token_threshold=2000,
            overlap_rate=0.1,
            apply_chunking=True,
            input_format="markdown",
            extra_args={"temperature": 0.1, "max_tokens": 2000}
        )
        crawl_config = CrawlerRunConfig(
            extraction_strategy=extraction_strategy,
            cache_mode=CacheMode.BYPASS
        )
        browser_config = BrowserConfig(
            headless=True,
            browser_type="chromium"
        )
        print(f"Scraping Amazon with LLM extraction for '{search_query}'...")
        async with AsyncWebCrawler(config=browser_config) as crawler:
            result = await crawler.arun(
                url=url,
                config=crawl_config,
                js_code=[
                    "await new Promise(resolve => setTimeout(resolve, 3000));",
                    "window.scrollTo(0, document.body.scrollHeight/2);",
                    "await new Promise(resolve => setTimeout(resolve, 2000));"
                ],
                wait_for="css:[data-component-type='s-search-result']",
                page_timeout=30000
            )
            if result.success and result.extracted_content:
                try:
                    extracted_data = json.loads(result.extracted_content)
                    if 'products' in extracted_data:
                        products = extracted_data['products']
                        print(f"Successfully extracted {len(products)} products using LLM.")
                        extraction_strategy.show_usage()
                        return products
                    else:
                        print("No products found in LLM extracted data.")
                        return None
                except json.JSONDecodeError as e:
                    print(f"Failed to parse LLM extracted JSON: {e}")
                    print("Raw extracted content (first 500 chars):", str(result.extracted_content)[:500])
                    return None
            else:
                print("LLM extraction failed to retrieve or extract content.")
                if result.error_message:
                    print(f"Error during LLM crawl: {result.error_message}")
                return None
    except Exception as e:
        print(f"An error occurred during LLM extraction: {str(e)}")
        return None

def extract_products_from_markdown(markdown_content):
    products = []
    lines = markdown_content.split('\n')
    current_product = {}
    for line in lines:
        line = line.strip()
        if line.startswith('## ') or line.startswith('### '):
            if current_product.get('title'):
                products.append(current_product)
                current_product = {}
            current_product['title'] = line.replace('## ', '').replace('### ', '').strip()
        elif '$' in line and 'price' not in current_product:
            price_match = re.search(r'\$[\d,]+\.?\d*', line)
            if price_match:
                current_product['price'] = price_match.group()
        elif ('star' in line.lower() or 'rating' in line.lower()) and 'rating' not in current_product:
            current_product['rating'] = line.strip()
        elif (line.startswith('http') or 'amazon.com' in line) and 'link' not in current_product:
            if 'amazon.com/dp/' in line or 'amazon.com/.*/dp/' in line or '/gp/product/' in line:
                 url_match = re.search(r'(https://www\.amazon\.com[^ ]+)', line)
                 if url_match:
                    current_product['link'] = url_match.group(1)
    if current_product.get('title'):
        products.append(current_product)
    for product in products:
        product.setdefault('title', 'N/A')
        product.setdefault('price', 'N/A')
        product.setdefault('rating', 'N/A')
        product.setdefault('link', 'N/A')
    return products

async def scrape_amazon_simple(search_query="gaming+chairs"):
    url = f"https://www.amazon.com/s?k={search_query}"
    try:
        browser_config = BrowserConfig(headless=True, browser_type="chromium")
        print(f"Scraping Amazon (simple extraction) for '{search_query}'...")
        async with AsyncWebCrawler(config=browser_config) as crawler:
            result = await crawler.arun(
                url=url,
                js_code=[
                    "await new Promise(resolve => setTimeout(resolve, 3000));",
                    "window.scrollTo(0, document.body.scrollHeight/2);",
                    "await new Promise(resolve => setTimeout(resolve, 2000));"
                ],
                wait_for="css:[data-component-type='s-search-result']",
                page_timeout=30000
            )
            if result.success and result.markdown:
                print("Successfully retrieved content for simple extraction.")
                products = extract_products_from_markdown(result.markdown)
                print(f"Successfully extracted {len(products)} products using simple parsing.")
                return products
            else:
                print("Failed to retrieve content for simple extraction.")
                if result.error_message:
                    print(f"Error during simple crawl: {result.error_message}")
                return None
    except Exception as e:
        print(f"An error occurred during simple extraction: {str(e)}")
        return None

def display_products(products, method_name=""):
    if not products:
        print(f"No products to display from {method_name} method.")
        return
    print(f"\n{'='*60}")
    print(f"FOUND {len(products)} PRODUCTS (using {method_name})")
    print(f"{'='*60}")
    for i, product_data in enumerate(products, 1):
        product = product_data.model_dump() if isinstance(product_data, BaseModel) else product_data
        print(f"\n{i}. {product.get('title', 'N/A')}")
        print(f"   Price: {product.get('price', 'N/A')}")
        print(f"   Rating: {product.get('rating', 'N/A')}")
        print(f"   Link: {product.get('link', 'N/A')}")
        print("-" * 50)

# --- Function to save individual product Markdown files ---
def save_products_as_markdown_files(products, base_filename_prefix="product"):
    """
    Save each product as an individual Markdown (.md) file in 'crawled_pages'.
    """
    if not products:
        return

    output_dir = "crawled_pages"
    os.makedirs(output_dir, exist_ok=True)
    print(f"\nSaving individual product Markdown files to '{output_dir}/'...")

    count_saved = 0
    for i, product_data in enumerate(products, 1):
        product = product_data.model_dump() if isinstance(product_data, BaseModel) else product_data

        title = product.get('title', f'product_{i:02d}').replace('/', '_').replace('\\', '_') # Sanitize title for filename
        # Further sanitize by removing other problematic characters for filenames
        sanitized_title_segment = re.sub(r'[^\w\s-]', '', title.lower().replace(' ', '_'))[:50] # Limit length

        md_content = f"""## {product.get('title', 'N/A')}

**Price**: {product.get('price', 'N/A')}
**Rating**: {product.get('rating', 'N/A')}
**Link**: [{product.get('link', 'N/A')}]({product.get('link', 'N/A')})

---
*Extracted from: Amazon Search*
*Search Query used in script: [Not directly available here, defined in main]*
*Source URL: {product.get('link', 'N/A')}*
"""
        # Use a combination of base prefix and a sanitized title segment or index for filename
        filename = f"{output_dir}/{base_filename_prefix}_{i:02d}_{sanitized_title_segment}.md"

        try:
            with open(filename, "w", encoding="utf-8") as f:
                f.write(md_content)
            count_saved +=1
        except Exception as e:
            print(f"Could not save file {filename}: {e}")

    print(f"Saved {count_saved} Markdown files to {output_dir}/")
# --- End of save_products_as_markdown_files ---

async def main():
    search_query = "gaming+chairs"
    print("=== Amazon Product Scraper ===\n")
    products = None
    extraction_method_used = ""

    if DEEPSEEK_API_KEY:
        print("Attempting LLM extraction...")
        products = await scrape_amazon_with_llm(search_query)
        extraction_method_used = "LLM"

    if not products:
        if DEEPSEEK_API_KEY and extraction_method_used == "LLM":
            print("\nLLM extraction did not yield results, falling back to simple extraction...\n")
        else:
            print("API key not available or LLM skipped, proceeding with simple extraction...\n")
        products = await scrape_amazon_simple(search_query)
        extraction_method_used = "Simple"

    if products:
        display_products(products, extraction_method_used)

        # Define base filename for outputs
        base_output_filename = f"amazon_{search_query.replace('+', '_')}_{extraction_method_used.lower()}_products"

        # Save main JSON file
        json_filename = f"{base_output_filename}.json"
        products_to_save = [p.model_dump() if isinstance(p, BaseModel) else p for p in products]
        with open(json_filename, "w", encoding="utf-8") as f:
            json.dump(products_to_save, f, indent=2, ensure_ascii=False)
        print(f"\nResults saved to {json_filename}")

        # Save individual Markdown files
        save_products_as_markdown_files(products_to_save, base_filename_prefix=base_output_filename)
    else:
        print("\nBoth LLM and simple extraction methods failed to find products.")

if __name__ == "__main__":
    asyncio.run(main())

import os
from deepseek import DeepSeek # Assuming this is your DeepSeek library/SDK

# --- Configuration ---
# IMPORTANT: Replace "your_deepseek_key" with your actual DeepSeek API key.
# Consider using environment variables for API keys in production.
DEEPSEEK_API_KEY = "your-deepseek-api"
DEEPSEEK_MODEL = "deepseek-chat" # Or your preferred DeepSeek model
TEMPERATURE = 0.3 # Adjust for desired creativity/factuality (0.0-1.0)
MARKDOWN_DIR = "crawled_pages"
SYSTEM_PROMPT = "You are an expert AI assistant. Your task is to accurately extract specific information from the provided text, which is the content of a product markdown page."

def extract_product_info(client, file_content, filename):
    """
    Uses DeepSeek to extract product name and price from the file content.
    """
    # You can customize this prompt further based on the exact structure of your .md files
    # For example, if you know the title is always a H2 heading (##) and price is bolded.
    user_prompt = f"""
From the following content of the file '{filename}', please extract:
1. The product name (usually the main title or heading).
2. The product price (including the currency symbol).

Return the information in a structured format, like:
Product Name: [The extracted product name]
Price: [The extracted price]

If information is missing, indicate "N/A".

Markdown Content:
---
{file_content}
---
"""
    try:
        response = client.chat.completions.create(
            model=DEEPSEEK_MODEL,
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": user_prompt}
            ],
            temperature=TEMPERATURE
        )
        return response.choices[0].message.content
    except Exception as e:
        print(f"Error processing file {filename} with DeepSeek: {e}")
        return None

if __name__ == "__main__":
    if DEEPSEEK_API_KEY == "your_deepseek_key":
        print("Please replace 'your_deepseek_key' with your actual DeepSeek API key in the script.")
        exit()

    if not os.path.isdir(MARKDOWN_DIR):
        print(f"Error: Directory '{MARKDOWN_DIR}' not found. Please ensure the crawled pages are in this directory.")
        exit()

    # Initialize the DeepSeek client
    # Based on your snippet, the DeepSeek class might take model and temperature
    # in its constructor. Adjust if your DeepSeek client initialization differs.
    # If these are set in the constructor and used as defaults, you might not
    # need to pass them in `completions.create()` unless you want to override.
    deepseek_client = DeepSeek(
        api_key=DEEPSEEK_API_KEY
        # model=DEEPSEEK_MODEL, # Potentially set here if your SDK supports it as default
        # temperature=TEMPERATURE # Potentially set here
    )

    print(f"Processing Markdown files from '{MARKDOWN_DIR}'...\n")

    markdown_files_processed = 0
    for filename in os.listdir(MARKDOWN_DIR):
        if filename.endswith(".md"):
            file_path = os.path.join(MARKDOWN_DIR, filename)
            print(f"--- Reading file: {file_path} ---")
            try:
                with open(file_path, "r", encoding="utf-8") as file:
                    content = file.read()

                if content.strip(): # Ensure content is not empty
                    extracted_info = extract_product_info(deepseek_client, content, filename)
                    if extracted_info:
                        print(f"Extracted Information for {filename}:\n{extracted_info}\n")
                    else:
                        print(f"Could not extract information for {filename}.\n")
                    markdown_files_processed += 1
                else:
                    print(f"File {filename} is empty. Skipping.\n")

            except Exception as e:
                print(f"Error reading or processing file {file_path}: {e}\n")

    if markdown_files_processed == 0:
        print(f"No Markdown files were found or processed in '{MARKDOWN_DIR}'.")
    else:
        print(f"Finished processing {markdown_files_processed} Markdown file(s).")

import json

with open("output.json", "w") as f:
    json.dump(data, f, indent=2)

import csv

with open("output.csv", "w", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=data[0].keys())
    writer.writeheader()
    writer.writerows(data)

0 9 * * * /usr/bin/python3 /home/user/deepseek_scraper/run_crawler.py >> /home/user/deepseek_scraper/logs/crawler.log 2>&1

from deepseek import DeepSeek

deepseek = DeepSeek(
    api_key="your_deepseek_key",
    model="deepseek-chat",
    temperature=0.5
)