googlemaps.py

# -*- coding: utf-8 -*-
#from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
#from selenium.common.exceptions import NoSuchElementException
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from bs4 import BeautifulSoup
from datetime import datetime
import time
#import re
import logging
import traceback

GM_WEBPAGE = 'https://www.google.com/maps/search/'
MAX_WAIT = 10
MAX_RETRY = 10
MAX_SCROLLS = 40

class GoogleMapsScraper:

    def __init__(self):
        self.driver = self.__get_driver()
        self.logger = self.__get_logger()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, tb):
        if exc_type is not None:
            traceback.print_exception(exc_type, exc_value, tb)

        self.driver.close()
        self.driver.quit()

        return True

    def sort_by_date(self, url):
        self.driver.get(url)
        wait = WebDriverWait(self.driver, MAX_WAIT)
        
        clicked_all = False
        tries_all = 0
        while not clicked_all and tries_all < MAX_RETRY:
            try:
                menu_bt_all = wait.until(EC.element_to_be_clickable((By.CSS_SELECTOR, 'button.allxGeDnJMl__button.allxGeDnJMl__button-text')))
                menu_bt_all.click()
                
                clicked_all = True
                time.sleep(3)
            except Exception:
                tries_all += 1
                self.logger.warn('Failed to click all reviews button')
                
            if tries_all == MAX_RETRY:
                return -1

        # open dropdown menu
        clicked = False
        tries = 0
        while not clicked and tries < MAX_RETRY:
            try:
                menu_bt = wait.until(EC.element_to_be_clickable((By.CSS_SELECTOR, 'div.cYrDcjyGO77__container')))  # //button[@data-value=\'Sort\'] XPath with graphical interface
                menu_bt.click()

                clicked = True
                time.sleep(3)
            except Exception:
                tries += 1
                self.logger.warn('Failed to click recent button')

            # failed to open the dropdown
            if tries == MAX_RETRY:
                return -1

        # second element of the list: most recent
        recent_rating_bt = self.driver.find_elements_by_xpath('//div[@role=\'menuitem\']')[1]
        recent_rating_bt.click()

        # wait to load review (ajax call)
        time.sleep(5)

        return 0
    
    def get_place_url(self, name):
        name_spl = name.split(';')[1] + ' ' + name.split(';')[2]
        name_spl = name_spl.replace(" ", "+")
        url = GM_WEBPAGE + name_spl
        return url
        
    
    def get_num_reviews(self):
        self.__scroll()
        time.sleep(4)
        self.__expand_reviews()
        response = BeautifulSoup(self.driver.page_source, 'html.parser')
        rblock = response.find_all('div', class_='section-review-content')
        return len(rblock)

    def get_reviews(self, offset):

        # scroll to load reviews
        error = self.__scroll()
        parsed_reviews = []
        if error == -1:
            return []
        else:

        # wait for other reviews to load (ajax)
            time.sleep(4)

        # expand review text
            self.__expand_reviews()

        # parse reviews
            response = BeautifulSoup(self.driver.page_source, 'html.parser')
            rblock = response.find_all('div', class_='section-review-content')
            for index, review in enumerate(rblock):
                if index >= offset:
                    parsed_reviews.append(self.__parse(review))

            return parsed_reviews


    def get_account(self, url):

        self.driver.get(url)

        # ajax call also for this section
        time.sleep(4)

        resp = BeautifulSoup(self.driver.page_source, 'html.parser')

        place_data = self.__parse_place(resp)

        return place_data


    def __parse(self, review):

        item = {}

        id_review = review.find('button', class_='section-review-action-menu')['data-review-id']
        username = review.find('div', class_='section-review-title').find('span').text

        try:
            review_text = self.__filter_string(review.find('span', class_='section-review-text').text)
        except Exception:
            # print e
            review_text = None

        rating = float(review.find('span', class_='section-review-stars')['aria-label'].split(' ')[1][0])
        relative_date = review.find('span', class_='section-review-publish-date').text

        try:
            n_reviews_photos = review.find('div', class_='section-review-subtitle').find_all('span')[1].text
            metadata = n_reviews_photos.split('\xe3\x83\xbb')
            if len(metadata) == 3:
                n_photos = int(metadata[2].split(' ')[0].replace('.', ''))
            else:
                n_photos = 0

            idx = len(metadata)
            n_reviews = int(metadata[idx - 1].split(' ')[0].replace('.', ''))

        except Exception:
            n_reviews = 0
            n_photos = 0

        user_url = review.find('a')['href']
        
        item['id_review'] = id_review
        item['caption'] = review_text

        # depends on language, which depends on geolocation defined by Google Maps
        # custom mapping to transform into date shuold be implemented
        item['relative_date'] = relative_date

        # store datetime of scraping and apply further processing to calculate
        # correct date as retrieval_date - time(relative_date)
        item['retrieval_date'] = datetime.now()
        item['rating'] = rating
        item['username'] = username
        item['n_review_user'] = n_reviews
        item['n_photo_user'] = n_photos
        item['url_user'] = user_url

        return item


    def __parse_place(self, response):

        place = {}
        place['overall_rating'] = float(response.find('div', class_='gm2-display-2').text.replace(',', '.'))
        place['n_reviews'] = int(response.find('div', class_='gm2-caption').text.replace('.', '').replace(',','').split(' ')[0])

        return place

    # expand review description
    def __expand_reviews(self):
        # use XPath to load complete reviews
        links = self.driver.find_elements_by_xpath('//a[@class=\'section-expand-review blue-link\']')
        for l in links:
            l.click()
        time.sleep(2)


    def __scroll(self):
        #scrollable_div = self.driver.find_element_by_css_selector('div.section-layout.section-scrollbox.scrollable-y.scrollable-show')
        wait = WebDriverWait(self.driver, MAX_WAIT)
        
        scrolled = False
        tries = 0
        while not scrolled and tries < MAX_RETRY:
            try:
                scrollable_div = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, 'div.section-layout.section-scrollbox.scrollable-y.scrollable-show')))
                self.driver.execute_script('arguments[0].scrollTop = arguments[0].scrollHeight', scrollable_div)
                
                scrolled = True
                time.sleep(3)
            except Exception:
                tries += 1
                print('Failed to scroll')
                #self.logger.warn('Failed to scroll')
                
            if tries == MAX_RETRY:
                return -1
            
        time.sleep(5)
        return 0
        #self.driver.execute_script('arguments[0].scrollTop = arguments[0].scrollHeight', scrollable_div)


    def __get_logger(self):
        # create logger
        logger = logging.getLogger('googlemaps-scraper')
        logger.setLevel(logging.DEBUG)

        # create console handler and set level to debug
        fh = logging.FileHandler('gm-scraper.log')
        fh.setLevel(logging.DEBUG)

        # create formatter
        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')

        # add formatter to ch
        fh.setFormatter(formatter)

        # add ch to logger
        logger.addHandler(fh)

        return logger


    def __get_driver(self, debug=False):
        options = Options()
        if not debug:
            options.add_argument("--headless")
        options.add_argument("--window-size=1366,768")
        options.add_argument("--disable-notifications")
        options.add_argument("--lang=en")
        input_driver = webdriver.Chrome(chrome_options=options)

        return input_driver


    # util function to clean special characters
    def __filter_string(self, str):
        strOut = str.replace('\r', ' ').replace('\n', ' ').replace('\t', ' ')
        return strOut