1
\$\begingroup\$

I submitted a previous version of this program but I've completely rewritten it in an Object Oriented style. This is only my second attempt at OO programming so I'm interested in hearing how I can improve things.

Here's the main program:

import tweepy
import sqlite3
import json
import re
import time
import datetime
import math
import jinja2
import PyRSS2Gen


class Tweets (object):
    def __init__(self, account_data, params):
        auth = tweepy.auth.OAuthHandler(account_data['consumer_key'],
                account_data['consumer_secret'])
        auth.set_access_token(account_data['access_token_key'],
                account_data['access_token_secret'])
        api = tweepy.API(auth)
        self.db = params['db']
        self.tweets = []
        for tweet in api.home_timeline(count=100, include_rts=0):
            try:
                url = tweet.entities['urls'][0]['expanded_url']
            except IndexError:
                url = False
            if (url is not False):
                self.tweets.append((
                        tweet.id_str,
                        self.extract_urls(tweet.text),
                        url,
                        str(tweet.created_at).replace(' ', 'T'),
                        tweet.retweet_count,
                        tweet.user.screen_name,
                        tweet.user.followers_count
                        ))

    def save(self):
        tdb = TweetDatabase(self.db)
        tdb.save(self.tweets)
        tdb.purge()

    def extract_urls(self, text):
        text = re.sub('http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|'
                '[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', '',
                text).strip()
        text = re.sub('\:$', '', text)
        return text

class TweetDatabase (object):
    def __init__(self, db):
        self.conn = sqlite3.connect(db)
        self.c = self.conn.cursor()

    def create(self):
        try:
            self.c.execute('''CREATE TABLE tweets
                                    ( id int not null unique,
                                    text text,
                                    url text,
                                    created_at text,
                                    retweet_count int,
                                    screen_name text,
                                    followers_count int);''')
            self.conn.commit()
            return True
        except sqlite3.OperationalError:
            return False

    def load(self):
        return self.c.execute('''select * from tweets;''')

    def save(self, data):
        self.create()
        self.c.executemany('''INSERT OR REPLACE INTO tweets 
                    VALUES (?,?,?,?,?,?,?)''', data)
        self.conn.commit()
        return True

    def purge(self):
        self.c.execute('''
            delete from tweets
            where datetime(created_at) < date('now','-8 day');
            ''')
        self.c.execute('vacuum;')
        self.conn.commit()
        return True


class FilteredTweets (object):
    def __init__(self, params):
        self.filtered_tweets = []
        self.blacklist = params['blacklist']
        self.whitelist = params['whitelist']
        self.db = TweetDatabase(params['db'])
        self.tweets = self.db.load()
        for tweet in self.tweets:
            id, text, url, created_at, retweet_count, screen_name, \
                    followers_count = tweet
            score = self.build_score(retweet_count, followers_count)
            if (self.check_blacklist(text) and (
                    score > params['threshold'] or 
                    self.check_whitelist(screen_name))):
                self.filtered_tweets.append(
                        {
                        'id': id, 
                        'text': text, 
                        'url': url, 
                        'created_at': created_at,
                        'retweet_count': retweet_count, 
                        'screen_name': screen_name, 
                        'followers_count': followers_count, 
                        'score': score
                        })
            self.filtered_tweets = sorted(self.filtered_tweets, key=lambda 
                    tup: tup['score'], reverse=True)

    def check_blacklist(self, text):
        for phrase in self.blacklist:
            if phrase.strip() in text:
                return False
        return True

    def check_whitelist(self, screen_name):
        for whitelist_name in self.whitelist:
            if screen_name == whitelist_name:
                return True
        return False

    def build_score(self, retweet_count, followers_count):
        retweet_count -= 1
        if retweet_count > 2:
            retweet_score = pow(retweet_count, 1.5)
            raw_score = (retweet_score / followers_count)*100000
            score = round(math.log(raw_score, 1.09))
        else:
            score = 0
        return int(score)

    def load_by_date(self, close, far):
        self.date_filtered_tweets=[]
        counter = 0
        for tweet in self.filtered_tweets:
            self.created_at_object = datetime.datetime.strptime(
                    tweet['created_at'], '%Y-%m-%dT%H:%M:%S')
            if (self.created_at_object > self.build_date(far) 
                    and self.created_at_object < self.build_date(close)
                    and counter < 40):
                self.date_filtered_tweets.append(tweet)
                counter += 1
        return self.date_filtered_tweets

    def build_date(self, day_delta):
        filter_date = (
                datetime.datetime.today() - 
                datetime.timedelta(days=day_delta)).replace(
                        hour=0, minute=0, second=0, microsecond=0)
        return filter_date

class Output (object):

    def build_webpage(self, yesterdays_items, last_weeks_items, params):
        with open(params['html_template']) as f: 
            template = jinja2.Template(f.read())
        self.html_output = template.render(yesterdays_items = yesterdays_items, 
                last_weeks_items = last_weeks_items)
        with open(params['html_output'], 'w') as f: 
            f.write(self.html_output.encode('utf-8'))
        return True

    def build_rss(self, items, output_file):
        rss_items = []
        sorted_items = sorted(items, key=lambda 
                    tup: tup['created_at'], 
                    reverse=True)
        for item in sorted_items:
            rss_items.append(PyRSS2Gen.RSSItem(
                    title = '%s: %s - Score: %s' % (
                        item['screen_name'],
                        item['text'],
                        item['score']),
                    link = item['url'],
                    pubDate = datetime.datetime.strptime(
                            item['created_at'], '%Y-%m-%dT%H:%M:%S')))
        rss = PyRSS2Gen.RSS2(
            title = 'Mike\'s News',
            link = 'http://mikeshea.net/news/',
            description = 'Mike\'s personal news filtered by Tweet Threshold.',
            lastBuildDate = datetime.datetime.now(),
            items = rss_items
        )
        with open(output_file, 'w') as f:
            rss.write_xml(f)
        return True

    def build_json(self, items, output_file):
        with open(output_file, 'w') as f:
            f.write(json.dumps(items))
        return True

def main(accounts, params):
    for account in accounts:
        remote_tweets = Tweets(account, params)
        remote_tweets.save()
    tweets = FilteredTweets(params)
    wp = Output()
    wp.build_webpage(tweets.load_by_date(0,1), tweets.load_by_date(1,6), params)
    wp.build_rss(tweets.load_by_date(0,1), params['rss_output_file'])
    wp.build_json(tweets.load_by_date(0,7), params['json_output_file'])

Here's the script that executes it.

import tweet_threshold

TWITTER_ACCOUNT_DATA = [
    {'consumer_key': 'accountkey',
     'consumer_secret': 'accountkey',
     'access_token_key': 'accountkey',
     'access_token_secret': 'accountkey'},
    {'consumer_key': 'accountkey',
     'consumer_secret': 'accountkey',
     'access_token_key': 'accountkey',
     'access_token_secret': 'accountkey'}]

PARAMS = {
    'db': '/path/to/your/dir/tweet_threshold.sqlite',
    'html_output': '/path/to/your/dir/index.html',
    'html_template': '/path/to/your/dir/html_template.txt',
    'rss_output_file': '/path/to/your/dir/yesterday.xml',
    'json_output_file': '/path/to/your/dir/items.json',
    'threshold': 50,
    'blacklist': (
        'Congress',
        'Representative',
        'DHS',
        'Fox',
        'CISPA',
        'Republicans',
        '[Sponsor]'),
    'whitelist': (
        'mshea',
        'slyflourish',
        'yourwife',
        'yourbestfriend')}

tweet_threshold.main(TWITTER_ACCOUNT_DATA, PARAMS)
\$\endgroup\$

1 Answer 1

1
\$\begingroup\$

using '''...''' creates '/n' at the end of your lines, which you may find unexpected and is not explict.

A pythonic alternative is to use:

'CREATE TABLE tweets ('
'id int not null unique'
'text text,'
'url text,'

and so on. Python will join all there to one line, without errant new lines.

replace if (url is not False): with if url:

'''select * from tweets;''' is only on one line so use 'select * from tweets;'

beware of extracting from list like so:

id, text, url, created_at, retweet_count, screen_name, followers_count = tweet

because index is important here. If you change something elsewhere in the code is is easy to loose track. I suggest using inserting a dict instead and accessing id, text, and url by pulling out the dict, and doing tweet['id'], tweet['text'], and tweet['url']. If you cant store a dict instead store the dict as JSON. Makes this much more maintainable. You are doing this anyway in self.filtered_tweets.append({...})

replace datetime.datetime.strptime( tweet['created_at'], '%Y-%m-%dT%H:%M:%S') with self.created_at_object = tweet['created_at'].isoformat()

replace

if (self.created_at_object > self.build_date(far) 
                    and self.created_at_object < self.build_date(close)
                    and counter < 40):

with

if counter < 40 and self.build_date(close) > self.created_at_object > self.build_date(far):
\$\endgroup\$
7
  • \$\begingroup\$ Why avoid newlines in SQL? \$\endgroup\$ Apr 12, 2013 at 16:27
  • \$\begingroup\$ I said "which you may find unexpected and is not explicit." - remember explicit is better than implicit, - and making a habit of using '''...''' sparingly will prevent unexpected newlines in future. \$\endgroup\$ Apr 12, 2013 at 18:09
  • \$\begingroup\$ For me, the inconvenient thing about multiline string literals is the indentation that gets included. I don't see anything unexpected in line breaks between lines. \$\endgroup\$ Apr 12, 2013 at 20:03
  • \$\begingroup\$ convoluted example In : print ''.join('''multi line''') out: multi\nline ...just sayin' \$\endgroup\$ Apr 12, 2013 at 21:10
  • \$\begingroup\$ Another issue is that mistakes like this are easy to make and hard to spot: "select c" "from t" (forgot whitespace between c and from). \$\endgroup\$ Apr 13, 2013 at 8:50

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

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