10
\$\begingroup\$

Task

I have a large spreadsheet where each line should include:

  • The URL of a social media account
  • A field indicating whether the account is "active"
  • A name and UID number for each account

I have to find if the activity field is accurate for every account on the list, and fix any mismatches.


Code

##### IMPORTS

import requests
from bs4 import BeautifulSoup
import datetime
import re
from selenium import webdriver
import tkinter
import csv
from time import sleep

##### METHODS TO SCRAPE DATE OF MOST RECENT ACTIVITY FROM SOCIAL PAGES

# A method to get the date of last activity from a given Twitter feed.
def getMostRecentActivityTwitter(url):

    # Grab the page, raising an exception if it doesn't work
    response = requests.get(url)
    if '40' in str(response):
        raise Exception('Couldn''t find page!')

    # Parse the response into Beautiful Soup
    html = response.text
    soup = BeautifulSoup(html, 'html.parser')

    # Grab all the tweets
    tweets = soup.find_all('div', class_='tweet')

    # Take a look at the first two (if there's a pinned tweet, we want both)
    try:
        tweet1 = tweets[0]
        tweet2 = tweets[1]

        # Get their dates
        dateString1 = tweet1.find('a', class_='tweet-timestamp').attrs['title'].split('-')[1]
        dateString2 = tweet2.find('a', class_='tweet-timestamp').attrs['title'].split('-')[1]
        date1 = datetime.datetime.strptime(dateString1, ' %d %b %Y')
        date2 = datetime.datetime.strptime(dateString2, ' %d %b %Y')

        # Return the more recent of the two
        if date1-date2>datetime.timedelta(0):
            return date1
        else:
            return date2

    # Raise an exception if it didn't work
    except Exception:
        raise Exception('Parsing failed!')

# A method to get the date of last activity from a given Facebook page.
def getMostRecentActivityFB(url):

    # Grab the page, raising an exception if it doesn't work
    response = requests.get(url)
    if '40' in str(response):
        raise Exception('Couldn''t find page!')

    # Parse the response into Beautiful Soup
    html = response.text
    soup = BeautifulSoup(html, 'html.parser')

    # Grab the most recent post's timestamp
    try:
        mostRecentTimestamp = soup.find('abbr', class_=re.compile('_5ptz'))

        # Get an appropriately formatted datetime object out of it
        dateList = mostRecentTimestamp.attrs['title'].split(' ')
        dateString = dateList[1]+' '+dateList[2]+' '+dateList[3]
        date = datetime.datetime.strptime(dateString, '%B %d, %Y')

        # Return the result
        return date

    # Raise an exception if it didn't work
    except Exception:
        raise Exception('Parsing failed!')

# A method to get the date of last activity from a given YouTube channel.
def getMostRecentActivityYouTube(url):

    # Reformat the URL to make sure we're getting the Videos page
    splitURL = url.split('/')
    baseURL = splitURL[2]
    pageType = splitURL[3]
    channelName = splitURL[4]
    fixedURL = 'https://'+baseURL+'/'+pageType+'/'+channelName+'/videos'

    # Grab the page, raising an exception if it doesn't work
    response = requests.get(fixedURL)
    if '40' in str(response):
        raise Exception('Couldn''t find page!')

    # Parse the response into Beautiful Soup
    html = response.text
    soup = BeautifulSoup(html, 'html.parser')

    # Get the 'x units ago' timestamp from the most recent video
    try:
        howRecentString = soup.find(text=re.compile('ago'))

        # Turn it into a timedelta object
        howRecentList = howRecentString.split(' ')
        amount = howRecentList[0]
        unit = howRecentList[1]
        if unit[-1] != 's':
            unit = unit + 's'
        if unit == 'weeks':
            unit = 'days'
            amount = str(int(amount)*7)
        elif unit == 'months':
            unit = 'days'
            amount = str(int(amount)*30)
        elif unit == 'years':
            unit = 'days'
            amount = str(int(amount)*365)
        howRecent = datetime.timedelta(**{unit:float(amount)})

        # Calculate the date from today's, and return it
        todaysDate = datetime.datetime.today()
        date = (todaysDate - howRecent).replace(hour=0,minute=0,second=0,microsecond=0)
        return date

    # Raise an exception if it didn't work
    except Exception:
        raise Exception('Parsing failed!')

# A method to get the date of last activity from a given Instagram page.
def getMostRecentActivityInstagram(url):

    # Use Selenium to open the page, and read it into Beautiful Soup
    driver = webdriver.Ie()
    driver.get(url)
    soup = BeautifulSoup(driver.page_source, 'html.parser')
    driver.quit()

    # Find the link to the most recent post
    try:
        href = soup.find('a', href=re.compile('/p')).attrs['href']

        # Build the URL for the most recent post
        splitHREF = href.split('/')
        newURL = 'https://instagram.com/p/'+splitHREF[2]

        # Use Selenium to open the post, and read it into Beautiful Soup
        driver = webdriver.Ie()
        driver.get(newURL)
        newSoup = BeautifulSoup(driver.page_source, 'html.parser')
        driver.quit()

        # Find, interpret, and return the timestamp associated with the post
        dateString = newSoup.find('time').attrs['title']
        date = datetime.datetime.strptime(dateString, '%b %d, %Y')
        return date

    # Raise an exception if it didn't work
    except Exception:
        raise Exception('Parsing failed!')

# A method to get the date of last activity from a given Google Plus page.
def getMostRecentActivityGP(url):

    # Grab the page, raising an exception if it doesn't work
    response = requests.get(url)
    if '40' in str(response):
        raise Exception('Couldn''t find page!')

    # Parse the response into Beautiful Soup
    html = response.text
    soup = BeautifulSoup(html, 'html.parser')

    # Find, interpret, and return the timestamp of the most recent post
    try:
        dateString = soup.find('a', class_='o-U-s').text
        date = datetime.datetime.strptime(dateString, '%Y-%m-%d')
        return date

    # Raise an exception if it didn't work
    except Exception:
        raise Exception('Parsing failed!')

# A method to get the date of last activity from a given LinkedIn page.
def getMostRecentActivityLinkedIn(url):

    # Use Selenium to open the page, and read it into Beautiful Soup
    driver = webdriver.Ie()
    driver.get(url)
    soup = BeautifulSoup(driver.page_source, 'html.parser')
    driver.quit()

    # Get the 'x units ago' timestamp from the most recent post
    try:
        howRecentString = soup.find('a', class_='nus-timestamp').text

        # Turn it into a timedelta object
        howRecentList = howRecentString.split(' ')
        amount = howRecentList[0]
        unit = howRecentList[1]
        if unit[-1] != 's':
            unit = unit + 's'
        if unit == 'weeks':
            unit = 'days'
            amount = str(int(amount)*7)
        elif unit == 'months':
            unit = 'days'
            amount = str(int(amount)*30)
        elif unit == 'years':
            unit = 'days'
            amount = str(int(amount)*365)
        howRecent = datetime.timedelta(**{unit:float(amount)})

        # Calculate the date from today's, and return it
        todaysDate = datetime.datetime.today()
        date = (todaysDate - howRecent).replace(hour=0,minute=0,second=0,microsecond=0)
        return date

    # Raise an exception if it didn't work
    except Exception:
        raise Exception('Parsing failed!')

# A method to get the date of last activity from a given Tumblr page.
def getMostRecentActivityTumblr(url):

    # Grab the page, raising an exception if it doesn't work
    response = requests.get(url)
    if '40' in str(response):
        raise Exception('Couldn''t find page!')

    # Parse the response into Beautiful Soup
    html = response.text
    soup = BeautifulSoup(html, 'html.parser')

    # Build the URL for the most recent post
    try:
        splitURL = url.split('/')
        baseURL = splitURL[2]
        postID = soup.find('article').attrs['data-post-id']
        newURL = 'http://'+baseURL+'/post/'+postID

        # Try to grab it, and parse the response into Beautiful Soup
        newResponse = requests.get(newURL)
        if '40' in str(newResponse):
            raise Exception('Couldn''t find page!')
        newHTML = newResponse.text
        newSoup = BeautifulSoup(newHTML, 'html.parser')

        # Find, interpret, and return the date of the post
        dateString = newSoup.find('a', class_=re.compile('post-date')).text
        dateList = dateString.split(',')
        fixedDateString = dateList[0][0:-2]+dateList[1]
        date = datetime.datetime.strptime(fixedDateString, '%b %d %Y')
        return date

    # Raise an exception if it didn't work
    except Exception:
        raise Exception('Parsing failed!')

# A method to get the date of last activity from a given Vkontakte page.
def getMostRecentActivityVK(url):

    # Grab the page, raising an exception if it doesn't work
    response = requests.get(url)
    if '40' in str(response):
        raise Exception('Couldn''t find page!')

    # Parse the response into Beautiful Soup
    html = response.text
    soup = BeautifulSoup(html, 'html.parser')

    # Find the date of the most recent post
    try:
        dateString = soup.find('a', class_='wi_date').text

        # Figure out the date from the string
        if 'ago' in dateString:
            date = datetime.datetime.today()
        elif 'today' in dateString:
            date = datetime.datetime.today()
        elif 'yesterday' in dateString:
            date = datetime.datetime.today() - datetime.timedelta(days=1)
        elif ':' in dateString:
            dateList= dateString.split(' ')
            fixedDateString = str(dateList[0])+' '+str(dateList[1])+' '+str(datetime.datetime.today().year)
            date = datetime.datetime.strptime(fixedDateString, '%d %b %Y')
        else:
            date = datetime.datetime.strptime(dateString, '%d %b %Y')

        # Clean up the date and return it
        date = date.replace(hour=0,minute=0,second=0,microsecond=0)
        return date

    # Raise an exception if it didn't work
    except Exception:
        raise Exception('Parsing failed!')

##### OTHER METHODS

# A method to clean up URL's
def cleanURL(url):
    strippedURL = url.strip().lower()
    coreURL = strippedURL.split('//')[-1].split('?')[0].strip('www.').strip('/')
    rebuiltURL = 'https://'+coreURL
    if 'facebook' in coreURL:
        rebuiltURL = rebuiltURL+'/?brand_redir=DISABLE'
    return(rebuiltURL)

# A method to check a given URL, and write the results to the output table
def checkAndRecord(rowID, accountName, url, putativeActivity, writer):

    # Grab the domain name part of the URL
    domainName = url.split('/')[2]

    # If it's from one of the services we can work with...
    if 'twitter' in domainName:

        # Attempt to scrape the date...
        try:
            date = getMostRecentActivityTwitter(url)

        # ...or record the error if we can't
        except Exception as errorMessage:
            writer.writerow([rowID, accountName, url, 'Not Checked', errorMessage, ''])
            return

    # Do the same for each service
    elif 'facebook' in domainName:
        try:
            date = getMostRecentActivityFB(url)
        except Exception as errorMessage:
            writer.writerow([rowID, accountName, url, 'Not Checked', errorMessage, ''])
            return
    elif 'youtube' in domainName:
        try:
            date = getMostRecentActivityYouTube(url)
        except Exception as errorMessage:
            writer.writerow([rowID, accountName, url, 'Not Checked', errorMessage, ''])
            return
    elif 'instagram' in domainName:
        try:
            date = getMostRecentActivityInstagram(url)
        except Exception as errorMessage:
            writer.writerow([rowID, accountName, url, 'Not Checked', errorMessage, ''])
            return
    elif 'google' in domainName:
        try:
            date = getMostRecentActivityGP(url)
        except Exception as errorMessage:
            writer.writerow([rowID, accountName, url, 'Not Checked', errorMessage, ''])
            return
    elif 'linkedin' in domainName:
        try:
            date = getMostRecentActivityLinkedIn(url)
        except Exception as errorMessage:
            writer.writerow([rowID, accountName, url, 'Not Checked', errorMessage, ''])
            return
    elif 'tumblr' in domainName:
        try:
            date = getMostRecentActivityTumblr(url)
        except Exception as errorMessage:
            writer.writerow([rowID, accountName, url, 'Not Checked', errorMessage, ''])
            return
    elif 'vk' in domainName:
        try:
            date = getMostRecentActivityVK(url)
        except Exception as errorMessage:
            writer.writerow([rowID, accountName, url, 'Not Checked', errorMessage, ''])
            return

    # If it's not from one of our services, record that in the output
    else:
        writer.writerow([rowID, accountName, url, 'Not Checked', 'No scraper for this service!', ''])
        return

    # If we managed to get a date, check if it's active...
    if datetime.datetime.today() - date < datetime.timedelta(days=365):

        # ...check if it's supposed to be dormant, and record the results...
        if putativeActivity == 'Dormant':
            writer.writerow([rowID, accountName, url, 'Active', '', 'Yes'])
            return
        else:
            writer.writerow([rowID, accountName, url, 'Active', '', 'No'])
            return

    # ...or do the same if it's dormant
    else:
        if putativeActivity == 'Active':
            writer.writerow([rowID, accountName, url, 'Inactive', '', 'Yes'])
            return
        else:
            writer.writerow([rowID, accountName, url, 'Inactive', '', 'No'])
            return

##### MAIN METHOD

# Make and hide tkinter window
root = tkinter.Tk()
root.withdraw()

# Prompt user for input file, and open it
csvToRead = tkinter.filedialog.askopenfile(parent=root, title='Select input csv file.')
with open(csvToRead.name, 'r', encoding='utf8') as fileToRead:
    reader = csv.reader(fileToRead)
    next(reader) # Skip the first line (header)

    # Prompt user for output directory, and initialize output file
    outputDir = tkinter.filedialog.askdirectory(parent=root, title='Select output directory:')
    with open(outputDir+'/recency_check_output.csv', 'w', encoding='utf8') as fileToWrite:
        writer = csv.writer(fileToWrite, lineterminator='\n')
        writer.writerow(['ID', 'Account Name', 'URL', 'Activity', 'Reason', 'Mismatch?'])

        # For each row of the table...
        for row in reader:

            # Get and clean the URL
            url = cleanURL(row[3])

            # Assign the other stuff we care about
            rowID = row[0]
            accountName = row[2]
            putativeActivity = row[8]

            # Keep track of what number we're on
            print(rowID)

            # Wait a second to avoid tripping bot detectors
            sleep(1)

            # Check the page's activity and write the results to the output table
            checkAndRecord(rowID, accountName, url, putativeActivity, writer)

Concerns

I'm looking for general pointers and advice on how to improve my code... both this specific code, and my coding style in general. Some specific concerns I have:

  • Because each social media service presents its post time-stamps differently, I've had to write a separate method for each one. However, some logic is duplicated, are there ways I could consolidate this to make my code more concise?

  • Are there better scraping methods I could use?

  • Is there a better way to manipulate the text?

  • Am I using exceptions properly?

  • Do I comment too much? Or make bad comments?

\$\endgroup\$
5
+100
\$\begingroup\$

Handling HTTP errors

This code seems to be intended to handle 404 Not Found response when downloading a url:

# Grab the page, raising an exception if it doesn't work
response = requests.get(url)
if '40' in str(response):
    raise Exception('Couldn''t find page!')

Unfortunately this is very inappropriate, for several reasons:

  • The code is counting on str(response) to look something like '<Response [N]>', where N is the HTTP status. The string representation of objects is not an API: don't try to parse str(...), always look for a better method than that

  • The message in the Exception indicates that the page was not found, but that's inaccurate. Status code 404 would mean the page was not found, but there are other status codes that start with 40 that mean something else, such as 400 Bad Request, 401 Unauthorized, 408 Request Timeout, to name just a few.

  • I suppose the message in the Exception is misleading, the real intention is what is in the comment: grab the page, raise an exception if it "doesn't work". However, there are many other ways a page cannot be grabbed, for example the status code could be another value in the range 4xx that doesn't contain "40", or it could be a 5xx (server side) error, or even something else. If you want to proceed only on successful download, you really should be checking for status code 200.

  • Instead of raising a generic Exception, it would be better to raise a more specific exception so that the caller function could identify the problem and handle it appropriately. If a suitable exception class doesn't exist, consider creating your own custom exception class.

A better way to write this part:

# Grab the page, raising an exception if it doesn't work
response = requests.get(url)
if response.status_code != 200:
    raise DownloadError("oops, could not grab the page", response)

Don't repeat yourself

Don't repeat the same operation for two variables like this:

tweet1 = tweets[0]
tweet2 = tweets[1]

# Get their dates
dateString1 = tweet1.find('a', class_='tweet-timestamp').attrs['title'].split('-')[1]
dateString2 = tweet2.find('a', class_='tweet-timestamp').attrs['title'].split('-')[1]
date1 = datetime.datetime.strptime(dateString1, ' %d %b %Y')
date2 = datetime.datetime.strptime(dateString2, ' %d %b %Y')

It's better to create a helper function parse_tweet_date, so that you can rewrite the above code as:

date1 = parse_tweet_date(tweets[0])
date2 = parse_tweet_date(tweets[1])

If the date parsing logic changes, you will be able to change it at one place, in the implementation of the parse_tweet_date helper.

The same goes for creating soups, multiple functions duplicating this code:

# Grab the page, raising an exception if it doesn't work
response = requests.get(url)
if '40' in str(response):
    raise Exception('Couldn''t find page!')

# Parse the response into Beautiful Soup
html = response.text
soup = BeautifulSoup(html, 'html.parser')

After improving this as I suggested above, move the code to a helper function so that it is not duplicated multiple times. The helper function can take a url as parameter and return a soup.

Working with dates

This is an unnatural way to compare dates:

# Return the more recent of the two
if date1-date2>datetime.timedelta(0):
    return date1
else:
    return date2

You can write naturally as:

if date1 > date2:
    return date1
else:
    return date2

Even better, as @dannnno pointed out in a comment, you can further simplify using max:

return max(date1, date2)

Strange numeric conversions

The numeric conversions here are very strange:

    amount = howRecentList[0]
    unit = howRecentList[1]
    if unit[-1] != 's':
        unit = unit + 's'
    if unit == 'weeks':
        unit = 'days'
        amount = str(int(amount)*7)
    elif unit == 'months':
        unit = 'days'
        amount = str(int(amount)*30)
    elif unit == 'years':
        unit = 'days'
        amount = str(int(amount)*365)
    howRecent = datetime.timedelta(**{unit:float(amount)})

Some of the conditional branches convert amount to int to perform a multiplication, then immediately convert back to str, until finally it's converted to float. It would be a lot simpler to convert it to float once at the beginning:

    amount = float(howRecentList[0])
    unit = howRecentList[1]
    if unit[-1] != 's':
        unit = unit + 's'
    if unit == 'weeks':
        unit = 'days'
        amount *= 7
    elif unit == 'months':
        unit = 'days'
        amount *= 30
    elif unit == 'years':
        unit = 'days'
        amount *= 365
    howRecent = datetime.timedelta(**{unit: amount})

String formatting

An expression easily becomes hard to read with multiple string concatenations. Instead of this:

dateString = dateList[1]+' '+dateList[2]+' '+dateList[3]

I suggest to use format:

dateString = '{} {} {}'.format(dateList[1], dateList[2], dateList[3])

The same thing, written a bit more compactly:

dateString = '{} {} {}'.format(*dateList[1:4])

But actually, since the values of dateList are strings, a simple join will be probably the best here:

dateString = ' '.join(dateList[1:4])

Your questions

Because each social media service presents its post time-stamps differently, I've had to write a separate method for each one. However, some logic is duplicated, are there ways I could consolidate this to make my code more concise?

As mentioned above, create helper functions. Anytime you find yourself copy-pasting code, stop right there, and move that piece of code to a helper function instead.

Are there better scraping methods I could use?

Extracting data using scraping is very fragile in general. Minor changes in the page layout can break your script. And such changes happen without warning, one day you will simply find that the script is no longer working, and the correction may involve a major rewrite.

The safe way is to learn to use the official API whenever possible. It's a bit more work, but the result is cleaner, easier to maintain, and doesn't break without notice.

Scraping is essentially a dirty hack, it's a good option only when there is no better option (no API), a last resort.

Am I using exceptions properly?

These are well known bad practices:

  • Raising generic Exception that doesn't indicate the nature of the problem, and doesn't allow the caller to handle that specific problem
  • Catching a generic Exception instead of the most specific possible. This leads to unexpected errors getting masked, and often obscure bugs that are painful to debug.

The general recommendation is this:

  • Always catch the most specific exception that may be raised
  • Always throw specific exceptions. If there is no appropriate exception for your purpose, create a custom exception class

Do I comment too much? Or make bad comments

It often happens that a comment and a block of code can be converted to a helper function, with the comment as the name of the function. This was the case with the code that grabs a page or else raises an error. By applying this technique, comments can simply disappear into self explanatory code. A common advice is that if the code is not self explanatory without a comment, then try to refactor until it is.

\$\endgroup\$
  • \$\begingroup\$ This is very helpful, thanks! Several things here (.join(), .format()) that I didn't know existed, and I didn't know that datetime objects were comparable with < and >. Followup question: When defining helper functions, where do you recommend I put them? Should they be defined right above the method that calls them? In a separate section that I label "helper functions" or some such? Somewhere else? Personal taste and it doesn't really matter? \$\endgroup\$ – A_S00 Jun 13 '16 at 22:52
  • \$\begingroup\$ It's good to put functions close to where they are used, but this can be a matter of personal taste. \$\endgroup\$ – Stop ongoing harm to Monica Jun 14 '16 at 10:30
  • \$\begingroup\$ Couldn't you change the date1 > date2 ? date1 : date2 (used c-style ternary for brevity...) to just max(date1, date2)? \$\endgroup\$ – Dannnno Jun 14 '16 at 13:03
  • \$\begingroup\$ @Dannnno yup you can, well-spotted. Updated my answer, thanks! \$\endgroup\$ – Stop ongoing harm to Monica Jun 14 '16 at 13:30
2
\$\begingroup\$

Others have already pointed out most of the code duplication and some other stuff, so I won't touch too much on that. But I think this problem is a great candidate for the strategy design pattern combined with a small factory!

First, we start off with an abstract base class for the strategies. I do this to enforce an interface on all the strategies using the @abstractmethod decorator.

class ActivityScraperStrategy:
    def __init__(self):
        pass

    @abstractmethod
    def get_most_recent_activity(self, url):
        """
        :type url: str
        :rtype: datetime.datetime
        """
        pass

    def _parse_url(self, url):
        """
        :type url: str
        :rtype: BeautifulSoup
        """
        response = requests.get(url)

        if response.status_code != 200:
            raise DownloadError("Unable to grab page, status code: {}".format(response.status_code))

        return BeautifulSoup(response.text, 'html.parser')

Besides creating the get_most_recent_activity() method, I have created a private method _parse_url() on this base class. Normally this isn't really how you should be using inheritance, but since I already had to use inheritance for the interface, I figured this would be a bit easier.

Now that we have the base class in order, we can create the first actual implementation of a strategy. Lets start with Twitter:

class TwitterActivityScraperStrategy(ActivityScraperStrategy):
    def get_most_recent_activity(self, url):
        """
        :type url: str
        :rtype: datetime.datetime
        """
        soup = self._parse_url(url)
        tweets = self._get_first_two_tweets(soup)

        if not tweets:
            raise NoActivityError("No tweets found on the page")

        first_tweet_date = self._get_datetime_from_tweet(tweets[0])
        second_tweet_date = self._get_datetime_from_tweet(tweets[1])

        if first_tweet_date > second_tweet_date:
            return first_tweet_date
        else:
            return second_tweet_date

    def _get_first_two_tweets(self, soup):
        """
        :type soup: BeautifulSoup
        :rtype: bs4.ResultSet
        """
        return soup.find_all('div', class_='tweet', limit=2)

    def _get_datetime_from_tweet(self, tweet):
        """
        :type tweet: bs4.Tag
        :rtype: datetime.datetime
        """
        tweet_timestamps = tweet.find('a', class_='tweet-timestamp')
        if not tweet_timestamps:
            raise HtmlChangedError("Unable to retrieve the timestamp tag in the tweets")

        title = tweet_timestamps.attrs['title']
        if not title:
            raise HtmlChangedError("Unable to retrieve title from tweet timestamp tag")

        date_string = title.split('-')[1]
        return datetime.datetime.strptime(date_string, ' %d %b %Y')

As you can see, I have attempted to split up certain responsibilities into separate (private) methods on the class. This greatly improves the readability of the get_most_recent_activity() method and the class overall.

I have also replaced the except Exception: with specific checks after each call on the Tag elements. This makes the reason you are throwing the exception and why the exception occurs more clear.

Another addition I have made is a specific exception called HtmlChangedError. This error is raised when the code fails due to the html/css on the website being changed. An error like this could specifically be caught somewhere down the chain to immediately inform you of the fact that some website's html has been changed and you need to adjust your scraper.

We can do exactly the same for Facebook:

class FacebookActivityScraperStrategy(ActivityScraperStrategy):
    def get_most_recent_activity(self, url):
        """
        :type url: str
        :rtype: datetime.datetime
        """
        soup = self._parse_url(url)
        most_recent_timestamp_tag = self._get_most_recent_timestamp_tag(soup)
        if not most_recent_timestamp_tag:
            raise NoActivityError("No timestamps found on page")

        return self._get_datetime_from_timestamp_tag(most_recent_timestamp_tag)

    def _get_most_recent_timestamp_tag(self, soup):
        """
        :type soup: BeautifulSoup
        :rtype: bs4.Tag
        """
        return soup.find('abbr', class_=re.compile('_5ptz'))

    def _get_datetime_from_timestamp_tag(self, most_recent_timestamp_tag):
        """
        :type most_recent_timestamp_tag: bs4.Tag
        :rtype: datetime.datetime
        """
        title = most_recent_timestamp_tag.attrs['title']
        if not title:
            raise HtmlChangedError("Unable to retrieve the title from the timestamp tag")

        date_list = title.split(' ')
        return self._get_datetime_from_date_list(date_list)

    def _get_datetime_from_date_list(self, date_list):
        """
        :type date_list: list
        :rtype: datetime.datetime
        """
        date_string = " ".join(date_list[1:4])
        return datetime.datetime.strptime(date_string, '%B %d, %Y')

Now you can imagine creating a class like this for each website you want to be able to scrape. For the sake of this review, I will keep it at these two strategy classes.

Now we need to instantiate the right strategy class based on the url. The factory design pattern is perfect for this:

class ActivityScraperStrategyFactory:
    @staticmethod
    def create_from_url(url):
        """
        :type url: str
        :rtype: ActivityScraperStrategy
        """
        domain = ActivityScraperStrategyFactory.get_domain_from_url(url)

        return ActivityScraperStrategyFactory.create_from_domain(domain)

    @staticmethod
    def get_domain_from_url(url):
        """
        :type url: str
        :rtype: str
        """
        return url.split('/')[2]

    @staticmethod
    def create_from_domain(domain):
        """
        :type domain: str
        :rtype: ActivityScraperStrategy
        """
        if "twitter" in domain:
            return TwitterActivityScraperStrategy()
        elif "facebook" in domain:
            return FacebookActivityScraperStrategy()

        raise UnknownDomainError("No scraper available for the domain {}".format(domain))

The beauty of this is, that the outlying code no longer has to worry about what type of url is being scraped and which method to call:

def checkAndRecord(rowID, accountName, url, putativeActivity, writer):
    try:
        activity_scraper = ActivityScraperStrategyFactory.create_from_url(url)
        date = activity_scraper.get_most_recent_activity(url)

        if is_active_within_year(date):
            # Write active stuff
        else:
            # Write inactive stuff
    except ActivityScraperError as error:
        writer.writerow([rowID, accountName, url, 'Not Checked', error.message, ''])

The function is now much shorter and no longer cares what type of scraper strategy is used, only that it returns a datetime when calling the method get_most_recent_activity() and raises an exception if it fails.

I have created the exceptions like this:

class ActivityScraperError(StandardError):
    pass


class DownloadError(ActivityScraperError):
    pass


class HtmlChangedError(ActivityScraperError):
    pass


class NoActivityError(ActivityScraperError):
    pass


class UnknownDomainError(ActivityScraperError):
    pass

So that you can simple catch ActivityScraperError and log the error message into the spreadsheet, assuming that you use useful error messages.

If you want to create a scraper for a new website, all you need to do is create a new strategy class and change the factory to return that strategy based on the domain. Making your code much less tightly coupled.

\$\endgroup\$
6
\$\begingroup\$

Style

Case:

Python's official style guide, PEP8, states you should use snake_case naming for your variables instead of camelCase:

getMostRecentActivityTwitter

Magic numbers:

You should avoid unexplained numbers through your code, like this:

if '40'

this is confusing to read and would be better explained through a simple temporary variable.

Spacing

You should be using spacing between binary operators, like here:

dateList[1]+' '+dateList[2]+' '+dateList[3]

Spaces after commas in a parameter list:

date = (todaysDate - howRecent).replace(hour=0,minute=0,second=0,microsecond=0)

Miscellaneous

You can use the += operator here:

unit = unit + 's'

You can just return this, you don't need to assign it to a temporary date variable:

date = (todaysDate - howRecent).replace(hour=0,minute=0,second=0,microsecond=0)

There's a lot of these in the code:

except Exception as errorMessage:
    writer.writerow([rowID, accountName, url, 'Not Checked', errorMessage, ''])
    return

seeing as they all have the same error, it would be easier to wrap the entire if conditions block in one big try catch

You've got some duplicate code here too:

if putativeActivity == 'Active':
    writer.writerow([rowID, accountName, url, 'Inactive', '', 'Yes'])
    return
else:
    writer.writerow([rowID, accountName, url, 'Inactive', '', 'No'])
    return

Seeing as they're identical besides one string, it would be easier to have the if else only deal with the value of the string and have the rest of the function call separate.

if putativeActivity == 'Active':
    active = 'Yes'
else:
    active = 'No'
writer.writerow([rowID, accountName, url, 'Inactive', '', active])
return
\$\endgroup\$
  • \$\begingroup\$ Thanks for the pointers! Couple questions on your answer: 1) You say I should use a temp variable to catch 40X errors in my http responses, because otherwise it's not clear what '40' means, but that I shouldn't use a temp variable to store date (which I used because I thought it would be unclear what that big bit of logic was for otherwise). How do I tell the difference between these two cases in the future? 2) Any thoughts on the specific concerns listed at the bottom of the question, especially logic duplication in the scraper methods and making cleanURL less messy? \$\endgroup\$ – A_S00 Jun 9 '16 at 17:30
  • \$\begingroup\$ The date variable isn't a magic number in that sense, but I suppose it comes down to personal experience. I don't know about URL cleaning though \$\endgroup\$ – Quill Jun 9 '16 at 19:27

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.