PASSKEY

# configurations
.idea

# crawlee and apify storage folders
apify_storage
crawlee_storage
storage

# installed files
.venv
.env

# git folder
.git
.gitignore

.dockerignore

root = true

[*]
indent_style = space
indent_size = 2
charset = utf-8
trim_trailing_whitespace = true
insert_final_newline = true
end_of_line = lf


.editorconfig

# This file tells Git which files shouldn't be added to source control

.idea
.DS_Store

apify_storage
storage

.venv/
.env/
.env
__pypackages__
dist/
build/
*.egg-info/
*.egg

__pycache__

.mypy_cache
.dmypy.json
dmypy.json
.pytest_cache

.scrapy
*.log

# Added by Apify CLI
node_modules
.venv


.gitignore

apify ~= 1.1.1
beautifulsoup4 ~= 4.12.0
requests ~= 2.31.0
python-dotenv ~= 1.0.0
attrs ~= 23.1.0
exceptiongroup ~= 1.1.2
outcome ~= 1.2.0
pysocks ~= 1.7.1
selenium ~= 4.10.0
trio ~= 0.22.2
trio-websocket ~= 0.10.3
wsproto ~= 1.2.0

requirements.txt

# First, specify the base Docker image.
# You can see the Docker images from Apify at https://hub.docker.com/r/apify/.
# You can also use any other image from Docker Hub.
FROM apify/actor-python-selenium:3.11

# Second, copy just requirements.txt into the actor image,
# since it should be the only file that affects the dependency install in the next step,
# in order to speed up the build
COPY requirements.txt ./

# Install the packages specified in requirements.txt,
# Print the installed Python version, pip version
# and all installed packages with their versions for debugging
RUN echo "Python version:" \
 && python --version \
 && echo "Pip version:" \
 && pip --version \
 && echo "Installing dependencies:" \
 && pip install -r requirements.txt \
 && echo "All installed Python packages:" \
 && pip freeze

# Next, copy the remaining files and directories with the source code.
# Since we do this after installing the dependencies, quick build will be really fast
# for most source file changes.
COPY . ./

# Specify how to launch the source code of your actor.
# By default, the "python3 -m src" command is run
CMD ["python3", "-m", "src"]


.actor/Dockerfile

A scraper for pulling product data from product or cateogory URL on Sephora including reviews and questions/answers

sephora-scraper

URLs and Results

Sephora Reviews & Questions Scraper

.actor/actor.json

A scraper for pulling data from a product on Sephora including reviews and questions/answers

Select proxies to be used by your crawler

Proxy configuration

Product or category URLs from Sephora to scrape

URL List

.actor/input_schema.json

src/__init__.py

import asyncio
import logging

from apify.log import ActorLogFormatter

from .main import main

handler = logging.StreamHandler()
handler.setFormatter(ActorLogFormatter())

apify_client_logger = logging.getLogger('apify_client')
apify_client_logger.setLevel(logging.INFO)
apify_client_logger.addHandler(handler)

apify_logger = logging.getLogger('apify')
apify_logger.setLevel(logging.DEBUG)
apify_logger.addHandler(handler)

asyncio.run(main())


src/__main__.py

import requests
import json
from apify import Actor
from bs4 import BeautifulSoup
from src.utils import request_url, initialize_driver
from src.results import get_product_result
from selenium.webdriver.common.by import By
from selenium.common.exceptions import NoSuchElementException
import asyncio
from urllib.parse import urlparse

SHOP = '/shop/'
PRODUCT = '/product/'
LIMIT = 100

async def main():
  async with Actor:
    # Read actor input
    actor_input = await Actor.get_input() or {}

    # Get data from actor input
    url_list = actor_input.get('url_list', None)
    get_similars = actor_input.get('get_similars', False)
    proxy_settings = actor_input.get('proxy_config', None)
    parsed_proxy = None
    if proxy_settings:
      proxy_configuration = await Actor.create_proxy_configuration(actor_proxy_input=proxy_settings)
      proxy_url = await proxy_configuration.new_url()
      Actor.log.info(f'proxy_url: {proxy_url}')
      parsed_proxy = urlparse(proxy_url)
      Actor.log.info(f'parsed_proxy: {parsed_proxy}')

    # Exit if no url is given
    if not url_list:
      Actor.log.info('No URL specified in actor input, exiting...')
      await Actor.exit()

    # Enqueue the URLs in the default request queue
    default_queue = await Actor.open_request_queue()
    for url in url_list:
      if SHOP in url:
        await default_queue.add_request({
          'url': url,
          'userData': {
            'get_similars': False,
            'first': True
          }
        })
      else:
        Actor.log.info(f'Enqueuing: {url}')
        await default_queue.add_request({
          'url': url,
          'userData': {
            'get_similars': get_similars,
            'first': True
          }
        })


    is_headless = True if Actor.config.headless else False
    driver = initialize_driver(is_headless, parsed_proxy)

    # Process the requests in the queue one by one
    while request := await default_queue.fetch_next_request():
      url = request['url']
      is_get_similars = request["userData"]['get_similars']
      first = request["userData"]['first']

      try:
        # Get the base URL by removing query parameters
        if "?" in url:
          url = url.split('?')[0]

        # check if URL is valid
        if SHOP not in url and PRODUCT not in url:
          raise Exception

        # go to url and close modal if exists
        if is_get_similars or first:
          modal_ids = ['modal1Dialog', 'modal2Dialog', 'modal3Dialog']
          driver.get(url)
          await asyncio.sleep(3)
          imageBuffer = driver.get_screenshot_as_png()
          await Actor.set_value('OUTPUT', imageBuffer, content_type='image/png')
          for modal_id in modal_ids:
            try:
              modal_close = driver.find_element(By.XPATH, f'//*[@id="{modal_id}"]/button')
              driver.execute_script("arguments[0].click();", modal_close)
              await asyncio.sleep(1)
              break  # Exit the loop once a modal is found and closed
            except NoSuchElementException:
              continue  # If the modal wasn't found, move on to the next one

        # decide if category or single product
        is_product = False if SHOP in url else True

        if is_product:
          Actor.log.info(f'Scraping product: {url}')

          # Extract product id from URL
          last_dash = url.rfind('-')
          product_id = url[last_dash + 1:]

          # fetch url to get price and love_count
          headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.82 Safari/537.36"}
          try:
            webpage = requests.get(url, timeout=5, headers=headers)
            content = BeautifulSoup(webpage.content, 'html.parser')
          except requests.exceptions.ReadTimeout:
            content = None

          # fetch api results data and prepare for creating result
          offset = 0
          r_soup = request_url("Reviews", product_id, LIMIT, offset)
          reviews_data = [json.loads(r_soup.text)]
          reviev_count = reviews_data[0]["TotalResults"]
          r_pages = reviev_count // 100
          if r_pages > 0:
            for _ in range(r_pages):
              offset += 100
              soup_new = request_url("Reviews", product_id, LIMIT, offset)
              reviews_data.append(json.loads(soup_new.text))
          
          # fetch api questions data and prepare for creating result
          offset = 0
          q_soup = request_url("Questions", product_id, LIMIT, offset)
          questions_data = [json.loads(q_soup.text)]
          question_count = questions_data[0]["TotalResults"]
          if question_count == 0:
            questions_data = []
          q_pages = question_count // 100
          if q_pages > 0:
            for _ in range(q_pages):
              offset += 100
              soup_new = request_url("Questions", product_id, LIMIT, offset)
              questions_data.append(json.loads(soup_new.text))
          
          # build result dict
          result = await get_product_result(product_id, reviews_data, questions_data)

          # get price and love_count from content
          try:
            if content:
              priceTag = content.find("p", {"data-comp": "Price "})
              price = priceTag.find("b").text
              result["info"]["price"] = price
              # result["info"]["price"] = None
            else:
              result["info"]["price"] = None
          except AttributeError:
            result["info"]["price"] = None
          try:
            if content:
              love_count = content.find("div", {"data-comp": "LovesCount "}).text
              result["info"]["love_count"] = love_count
              # result["info"]["love_count"] = None
            else:
              result["info"]["love_count"] = None
          except AttributeError:
            result["info"]["love_count"] = None

          # look for nested links and enqueue their targets
          if is_get_similars:
            for _ in range(3):
              driver.execute_script("window.scrollBy(0, 1000);")
              await asyncio.sleep(1)
            try:
              aa = driver.find_element(By.TAG_NAME, "main").text
              Actor.log.info(f'aa: {aa}')
              
              similars_careousel_headline = driver.find_element(By.XPATH, '//h2[text()="You May Also Like"]')
              Actor.log.info(f'similars_careousel_headline: {similars_careousel_headline}')
              similars_careousel = driver.execute_script(
                """return arguments[0].parentNode.parentNode.nextElementSibling;""",
                similars_careousel_headline)
              similars_list = similars_careousel.find_elements(By.TAG_NAME, 'li')
              Actor.log.info(f'similars_list: {similars_list}')

            except NoSuchElementException or TimeoutError:
              similars_list = None
            if similars_list:
              for link in similars_list:
                try:
                  href = link.find_element(By.TAG_NAME, 'a').get_attribute('href')
                  if not href.startswith('https://www.sephora.com'):
                    href = "https://www.sephora.com" + href
                  Actor.log.info(f'Enqueuing {href}')
                  await default_queue.add_request({
                    'url': href,
                    'userData': {
                      'get_similars': False,
                      'first': False
                    }
                  })
                except NoSuchElementException:
                  continue
              Actor.log.info(f'Enqueued {len(similars_list)} similar product URL')
            else:
              Actor.log.info(f'No similar product found')

          # Push the title of the page into the default dataset
          await Actor.push_data({ 'url': url, 'result': result })

        else:
          Actor.log.info(f'Scraping category: {url}')
          try:
            product_count = 0
            main = driver.find_element(By.TAG_NAME, 'main')
            for paragraph in main.find_elements(By.TAG_NAME, 'p'):
              if ' Result' in paragraph.text:
                product_count = int(str(paragraph.text).split(' ')[0])
                all_product_component = driver.execute_script(
                  """return arguments[0].parentNode.parentNode.nextElementSibling;""",
                  paragraph
                )
                break
          except NoSuchElementException:
            product_count = 0
          click_count = product_count // 60
          try:
            show_more = None
            for button in main.find_elements(By.TAG_NAME, 'button'):
              if button.text == 'Show More Products':
                show_more = button
            if show_more:
              for _ in range(click_count):
                driver.execute_script("arguments[0].scrollIntoView();", show_more)
                await asyncio.sleep(1)
                driver.execute_script("arguments[0].click();", show_more)
                await asyncio.sleep(3)
          except NoSuchElementException:
            pass
          
          try:
            if all_product_component:
              all_products = all_product_component.find_elements(By.XPATH, './div')
            else:
              all_products = []
          except NoSuchElementException:
            Actor.log.info(f'No product found for this category')
            all_products = []

          products_count = 0
          for product in all_products:
            try:
              driver.execute_script("arguments[0].scrollIntoView();", product)
              href = product.find_element(By.XPATH, './a').get_attribute('href')
              if '?' in href:
                href = href.split('?')[0]
              await default_queue.add_request({
                'url': href,
                'userData': {
                  'get_similars': False,
                  'first': False
                }
              })
              products_count += 1
            except NoSuchElementException:
              continue
          if all_products:
            Actor.log.info(f'Enqueued {products_count} product URL')
      except:
        Actor.log.exception(f'Cannot extract data from {url}')
      finally:
        # Mark the request as handled so it's not processed again
        await default_queue.mark_request_as_handled(request)

src/main.py

from urllib.parse import unquote

async def get_product_result(product_id, reviews_data, questions_data):
  # create result structure
  result = {
    "info": {},
    "product_variants": [],
    "statistics": {},
    "reviews": [],
    "questions": []
  }
  
  product_data = reviews_data[0]["Includes"]
  products = product_data["ProductsOrder"] if product_data else []
  main_product = product_data["Products"][product_id] if product_data else {}

  # info
  if main_product:
    main_is_available = False
    if "AVAILABILITY" in main_product["AttributesOrder"]:
      if main_product["Attributes"]["AVAILABILITY"]["Values"]:
        main_is_available = main_product["Attributes"]["AVAILABILITY"]["Values"][0]["Value"]
    general_info = {
      "id": product_id,
      "name": main_product["Name"],
      "image": str(main_product["ImageUrl"]).split("?")[0],
      "description": main_product["Description"],
      "is_available": main_is_available,
      "brand": main_product["Brand"]["Name"]
    }
    result["info"] = general_info


  # product_variants
  for prod in products:
    if prod == product_id:
      continue
    pr = product_data["Products"][prod]
    is_available = False
    if "AVAILABILITY" in pr["AttributesOrder"]:
      if pr["Attributes"]["AVAILABILITY"]["Values"]:
        is_available = pr["Attributes"]["AVAILABILITY"]["Values"][0]["Value"]
    variant_product = {
      "variant_id": pr["Id"],
      "variant_description": pr["Description"],
      "is_variant_available": is_available,
      "variant_name": pr["Name"],
      "variant_image": str(pr["ImageUrl"]).split("?")[0]
    }
    result["product_variants"].append(variant_product)


  # statistics
  commentCount = reviews_data[0]["TotalResults"]
  if commentCount:
    ReviewStatistics = main_product["ReviewStatistics"]
    ContextDataDistributionOrder = ReviewStatistics["ContextDataDistributionOrder"]
    result_statistics = []
    for elt in ContextDataDistributionOrder:
      distribution = ReviewStatistics["ContextDataDistribution"][elt]
      statistics = distribution["Values"]
      result_statistic = []
      for stat in statistics:
        result_statistic.append({
          "label": stat["ValueLabel"],
          "count": stat["Count"],
          "percentage": round((int(stat["Count"]) / int(commentCount)) * 100, 2)
        })
      result_statistics.append({
        "statistics_type": distribution["Label"],
        "statistics": result_statistic
      })
    rating_dist_raw = ReviewStatistics["RatingDistribution"]
    rating_dist = []
    for rating in rating_dist_raw:
      rating_dist.append({
        "rating": rating["RatingValue"],
        "count": rating["Count"],
        "percentage": round((int(rating["Count"]) / int(commentCount)) * 100, 2)
      })
    result["statistics"] |= {
      "average_rating": round(ReviewStatistics["AverageOverallRating"], 2),
      "helpful_vote_count": ReviewStatistics["HelpfulVoteCount"],
      "not_helpful_vote_count": ReviewStatistics["NotHelpfulVoteCount"],
      "recommended_review_count": ReviewStatistics["RecommendedCount"],
      "not_recommended_review_count": ReviewStatistics["NotRecommendedCount"],
      "review_count": commentCount,
      "variant_count": len(products) - 1,
      "rating_distribution": rating_dist,
      "review_distribution": result_statistics
    }

  
  # reviews
  for page in reviews_data:
    comments = page["Results"]
    for comment in comments:
      # reivewer_info build
      reviewer_info = {"nickname": comment["UserNickname"]}
      ContextDataValues = comment["ContextDataValues"]
      ContextDataValuesOrder = comment["ContextDataValuesOrder"]
      if ContextDataValuesOrder:
        for feature in ContextDataValuesOrder:
          try:
            reviewer_info |= {feature: ContextDataValues[feature]["ValueLabel"]}
          except:
            pass
      
      # badges build
      BadgesOrder = comment["BadgesOrder"]
      is_verified_purchase = False
      if BadgesOrder:
        if "verifiedPurchaser" in BadgesOrder:
          is_verified_purchase = True

      review = {
        "rating": comment["Rating"],
        "review_text": unquote(str(comment["ReviewText"])),
        "review_title": comment["Title"],
        "product_id": comment["ProductId"],
        "is_recommended": comment["IsRecommended"],
        "number_of_feedbacks_of_review": comment["TotalFeedbackCount"],
        "number_of_positive_feedbacks_of_review": comment["TotalPositiveFeedbackCount"],
        "number_of_negative_feedbacks_of_review": comment["TotalNegativeFeedbackCount"],
        "submitted_at": comment["SubmissionTime"],
        "reviewer_info": reviewer_info,
        "is_verified_purchase": is_verified_purchase,
      }
      result["reviews"].append(review)


  # questions
  if questions_data:
    question_info = questions_data[0]["Includes"]["Products"][product_id]
    result["statistics"]["question_count"] = question_info["TotalQuestionCount"]
    result["statistics"]["answer_count"] = question_info["TotalAnswerCount"]
    full_questions_resp = []
    
    for page in questions_data:
      answers = page["Includes"]["Answers"]
      questions_all = page["Results"]
      for quest in questions_all:
        q_elt = {
          "product_id": quest["ProductId"],
          "answers_count": quest["TotalAnswerCount"],
          "question": unquote(str(quest["QuestionDetails"])),
          "asked_at": quest["SubmissionTime"],
          "answers": []
        }
        answerIds = quest["AnswerIds"]

        for ans in answerIds:
          answer_elt = answers[ans]
          q_elt["answers"].append({
            "answer": unquote(str(answer_elt["AnswerText"])),
            "username": answer_elt["UserNickname"],
            "answered_at": answer_elt["SubmissionTime"],
            "positive_feedback_count": answer_elt["TotalPositiveFeedbackCount"],
            "negative_feedback_count": answer_elt["TotalNegativeFeedbackCount"]
          })
        full_questions_resp.append(q_elt)
        
    result["questions"] = full_questions_resp

  return result

src/results.py

import requests
from bs4 import BeautifulSoup
import dotenv
import os
from selenium.common.exceptions import WebDriverException
from selenium.webdriver.chrome.options import Options as ChromeOptions
from selenium import webdriver

dotenv.load_dotenv()


def request_url(type: str, product_id: str, limit: int, offset: int):
  base_url = f"https://api.bazaarvoice.com/data/{type.lower()}.json"

  headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.82 Safari/537.36",
  }
  params = {
    'Filter': ['contentlocale:en*', f'ProductId:{product_id}'],
    'Sort': 'SubmissionTime:desc',
    'Limit': limit,
    'Offset': offset,
    'Include': 'Products,Comments,Answers',
    'Stats': type,
    'passkey': os.environ.get("PASSKEY"),
    'apiversion': '5.4',
    'Locale': 'en_US'
  }

  response = requests.get(base_url, params=params, headers=headers)
  soup = BeautifulSoup(response.content, 'html.parser')

  return soup


def initialize_driver(is_headless, parsed_proxy):
    chrome_options = ChromeOptions()
    if is_headless:
      chrome_options.add_argument('--headless')
    chrome_options.add_argument('--no-sandbox')
    chrome_options.add_argument('--disable-dev-shm-usage')
    user_agent = 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.50 Safari/537.36'    
    chrome_options.add_argument('user-agent={0}'.format(user_agent))
    chrome_options.add_argument("--disable-blink-features=AutomationControlled")
    chrome_options.add_argument("--enable-webgl")
    chrome_options.add_argument("--ignore-gpu-blacklist")
    if parsed_proxy:
      try:
        creds, address = parsed_proxy.netloc.split('@')
        hostname, port = address.split(':')
        chrome_options.add_argument(f"--proxy-server={parsed_proxy.scheme}://{hostname}:{port}")
      except:
        pass

    driver = webdriver.Chrome(options=chrome_options)

    return driver

src/utils.py

The Sephora Reviews & Questions scraper is designed to extract detailed product information, including reviews and questions/answers, from either individual product or category URLs on Sephora's website.

Sephora Advanced Scraper

Sephora Advanced Scraper · Apify

Input · Sephora Advanced Scraper · Apify

API · Sephora Advanced Scraper · Apify

Related Actors · Sephora Advanced Scraper · Apify

The Sephora Reviews & Questions scraper is designed to extract detailed product information, including reviews and questions/answers, from either individual product or category URLs on Sephora's we...

Where next?