Nuno Mariz Weblog

My goal was to archive and display my internet lifestream. My first approach was writing a client for each API of the social networks that I'm in.
This turned out to be a complete waste of time and effort. All that I needed after all was a FriendFeed account that would centralize all my feeds.

Archiving and displaying your entries with Django is quite simple.
First of all, you need to download the Python FriendFeed API client. Then start a new application in your project, lets call it lifestream:

./manage.py startapp lifestream

On the settings.py add the lifestream project to the INSTALLED_APPS and a variable to store your FriendFeed username:

FRIENDFEED_USERNAME = 'your_username'

In the models.py add a model named Entry:

from django.db import models

class Entry(models.Model):
    id = models.CharField(max_length=255, primary_key=True)
    service_id = models.CharField(max_length=50, null=True, blank=True)
    service_name = models.CharField(max_length=50, null=True, blank=True)
    service_icon = models.URLField(max_length=255, verify_exists=False, null=True, blank=True)
    service_profile = models.URLField(max_length=255, verify_exists=False, null=True, blank=True)
    title = models.CharField(max_length=255, null=True, blank=True)
    link = models.URLField(max_length=255, verify_exists=False, null=True, blank=True)
    updated = models.DateTimeField(null=True, blank=True)
    published = models.DateTimeField(null=True, blank=True)
    media_title = models.CharField(max_length=255, null=True, blank=True)
    media_link = models.URLField(max_length=255, verify_exists=False, null=True, blank=True)
    media_thumbnail = models.URLField(max_length=255, verify_exists=False, null=True, blank=True)
    created = models.DateTimeField(auto_now_add=True)

    def __unicode__(self):
        return self.title

    class Meta:
        ordering = ['-published']
        verbose_name = 'Entry'
        verbose_name_plural = 'Entries'

    class Admin:
        list_display = ['title', 'service_name', 'published']
        list_filter = ['service_name']
        date_hierarchy = 'published'

Create an url.py on the lifestream folder:

from django.conf.urls.defaults import *
from lifestream.models import Entry
 
entry_list_dict = {
    'queryset' : Entry.objects.all(),
    'paginate_by' : 30,
}

urlpatterns = patterns('',   
    (r'^$', 'django.views.generic.list_detail.object_list', entry_list_dict),
)

As you can see, I've used a generic view. You can also use the date based generic views and pagination to build an archive like mine.

Add to your project root urls.py:

(r'^lifestream/', include('lifestream.urls'))

Create a template lifestream/entry_list.html:

{% for entry in object_list %}
<div class="source">
  <a href="{{ entry.service_profile }}" title="{{ entry.service_name }}"><img src="{{ entry.service_icon }}" alt="{{ entry.service_name }}" alt="{{ entry.service_name }}" /></a>
</div>
<div class="details">
  <ul>
    <li><a href="{{ entry.link }}">{{ entry.title }}</a></li>
    <li>{{ entry.published|timesince }} ago</li>
    {% if entry.media_thumbnail %}<li><a href="{{ entry.media_link }}"><img src="{{ entry.media_thumbnail }}" alt="{{ entry.media_title }}" /></a></li>{% endif %}
  </ul>
</div>
{% endfor %}

Finally, create a script to synchronize your feeds:

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
import os

ROOT_PATH = os.path.realpath(os.path.dirname(__file__))
PROJECT_PATH, PROJECT_DIR = os.path.split(ROOT_PATH)

sys.path.insert(0, ROOT_PATH)
sys.path.insert(1, PROJECT_PATH)

os.environ['DJANGO_SETTINGS_MODULE'] = '%s.settings' % PROJECT_DIR

from friendfeed import FriendFeed
from django.conf import settings
from lifestream.models import Entry

ff = FriendFeed()
feed = ff.fetch_user_feed(settings.FRIENDFEED_USERNAME)

for e in feed.get('entries'):
    entry, created = Entry.objects.get_or_create(id=e.get('id'))
    if created:
        service = e.get('service')
        entry.service_id = service.get('id')
        entry.service_name = service.get('name')
        entry.service_icon = service.get('iconUrl')
        entry.service_profile = service.get('profileUrl')
        entry.title = e.get('title')
        entry.link = e.get('link')
        entry.updated = e.get('updated')
        entry.published = e.get('published')
        media = e.get('media')
        if media:
            entry.media_title = media[0].get('title')
            entry.media_link = media[0].get('player') or entry.link
            thumbnails = media[0].get('thumbnails')
            entry.media_thumbnail = thumbnails[0].get('url')
        entry.save()

If you want, you can add a job in your crontab:

# synchronize every 15 mins
*/15 * * * *   root   /path/to/your/application/lifestream_cron.py

See my lifestream as the working example.

UPDATE: Friendfeed sends the time in UTC, if you want to use your timezone you have do some hacking:

Install pytz:

easy_install pytz

Import and assign your timezone to a variable:

import pytz
tz = pytz.timezone(settings.TIME_ZONE)

And replace entry.updated and entry.published with:

updated = e.get('updated')
updated = updated.replace(tzinfo=pytz.utc).astimezone(tz)
published = e.get('published')
published = published.replace(tzinfo=pytz.utc).astimezone(tz)
if settings.DATABASE_ENGINE == 'mysql': # http://code.djangoproject.com/ticket/5304
    updated = updated.replace(tzinfo=None)
    published = published.replace(tzinfo=None)
entry.updated = updated
entry.published = published

Thanks to Chris Kelly that send me an email reporting this.

Django has a code error notifications mechanism when a view raises an exception. It will email the people in ADMIN tuple(settings documentation) in settings.py with the full exception information and displays the default 500.html template.
This only happens when DEBUG=False in the settings.py.

It's possible to set a handle that change this behavior with a handler500 variable in the root urls.py.
So, we can easily write a simple view that sends an error notification to our jabber account.

First of all, you need to install xmpppy and dnspython:

$ easy_install xmpppy
$ easy_install dnspython

Add to settings.py the jabber parameters such as the jabber id, password, recipient, etc.:

JABBER_ERROR_NOTIFICATION = True
JABBER_ID = 'your_jabberid@jabberdomain.com'
JABBER_PASSWORD = 'your_jabber_password'
JABBER_RECIPIENT = 'recipient@jabberdomain.com'
JABBER_ERROR_TEXT = 'An error occurred in "Project Name", please check your email.'

Start a new app named errors or something else and add it to the INSTALLED_APPS tuple in the settings.py:

python manage.py startapp errors

Add a handler500 variable with the view in the root urls.py:

handler500 = 'errors.views.server_error_jabber'

Finally add the view in errors.views that sends a jabber notification and returns a 500 error page:

from django.views.defaults import server_error
from django.conf import settings
import xmpp, time

def server_error_jabber(request, template_name='500.html'):
   if settings.JABBER_ERROR_NOTIFICATION:
       jid = xmpp.protocol.JID(settings.JABBER_ID)
       cl = xmpp.Client(jid.getDomain(), debug=[])
       conn = cl.connect()
       if conn:
           auth = cl.auth(jid.getNode(), settings.JABBER_PASSWORD, resource=jid.getResource())
           if auth:
               id = cl.send(xmpp.protocol.Message(settings.JABBER_RECIPIENT, settings.JABBER_ERROR_TEXT))
               # Some older servers will not send the message if you disconnect immediately after sending
               time.sleep(1)
   return server_error(request, template_name)

NOTE: Don't forget to set DEBUG=False in the settings.py.

A simple templatetag for adding to the template context a variable with the user timeline from Twitter.
It uses the CachedContextUpdatingNode snippet for caching from Jacob Kaplan-Moss.
The reason that is necessary to cache content is because Twitter limits the number of accesses to the API.
This only works if the cache is enabled on your settings.py.

class TwitterNode(CachedContextUpdatingNode):

    cache_timeout = 1800 # 30 Minutes, maybe you want to change this
    
    def __init__(self, username, varname):
        self.username = username
        self.varname = varname

    def make_datetime(self, created_at):
        return datetime.fromtimestamp(mktime(strptime(created_at, '%a %b %d %H:%M:%S +0000 %Y')))

    def get_cache_key(self, context):
        return 'twitter_user_timeline_cache'

    def get_content(self, context):
        try:
            response = urllib.urlopen('http://twitter.com/statuses/user_timeline/%s.json' % self.username).read()
            json = simplejson.loads(response)
        except:
            return {self.varname : None}
        for i in range(len(json)):
            json[i]['created_at'] = self.make_datetime(json[i]['created_at'])
        return {self.varname : json}
    
@register.tag
def twitter_user_timeline(parser, token):
    bits = token.contents.split()
    if len(bits) != 4:
        raise TemplateSyntaxError, "twitter_user_timeline tag takes exactly three arguments"
    if bits[2] != 'as':
        raise TemplateSyntaxError, "second argument to twitter_user_timeline tag must be 'as'"
    return TwitterNode(bits[1], bits[3])

Usage:

{% twitter_user_timeline username as twitter_entries %}
{% if twitter_entries %}
  {% for entry in twitter_entries %}
  {{ entry.created_at|date:"d M Y H:i" }} - {{ entry.text }}
  {% endfor %}
{% endif %}

Use the source, Luke.

Fullread updates
January 10, 2008 @ 10:34 a.m.
New features on Fullread:
- Feeds in the latest and user bookmarks.
  - http://fullread.com/feeds/latest/[rss/atom]/ (By latest - all users)
  - http://fullread.com/feeds/[username]/[rss/atom]/ (By user)
- List of users that bookmarked the same URL.
- Multiple bookmarking service support.
NOTE: If you are using a bookmarking service that is not available, please let me know: http://fullread.com/contacts/?subject=Bookmarking%20service%20support.
0 comments | Filed under projects, python | add to del.icio.us
Fullread
January 3, 2008 @ 7:53 p.m.

Fullread is online. It's just a simple tool that helps you organize your online readings.
I use del.icio.us a lot, mainly for bookmarking, but right now my account is a complete mess because I’m constantly adding links that I want to read later and I don't delete them.
Fullread uses a simple concept: I add an URL to my account to read later, then I decide to archive, delete or add it to my del.icio.us account.
Feel free to send me any feature request or something else.
I hope you enjoy it.

17 comments | Filed under projects, python | add to del.icio.us
Smallr updates
November 23, 2007 @ 5:56 p.m.

After I've messed up about the domain issue(see this comments), I've just updated smallr with some features. Now provides an API to access the service by REST.
Right now ~~only~~ supports JSON, XML ~~will be~~ is also available ~~soon~~.
A bookmarklet is also available.

3 comments | Filed under projects, python | add to del.icio.us
Smallr
November 22, 2007 @ 7:51 p.m.

I've just released smallr. It lets you generate small web addresses with a keyword that make it easy to remember, will not break in emails and will never expire.
Basically is a TinyURL clone, with a keyword addon.
There will be an API for webservices with JSON and XML, eventually a Firefox extension.
Test it and send me some feedback.
I hope you enjoy it.

14 comments | Filed under projects, python | add to del.icio.us
iPhone says Welcome to Django
August 20, 2007 @ 8:50 p.m.

0 comments | Filed under django, iphone, python | add to del.icio.us

The Web Server Gateway Interface (WSGI) is a standard interface between web server software and web applications written in Python. Having a standard interface makes it easy to use an application that supports WSGI with a number of different web servers.

One implementation is wsgiref that was added to Python 2.5 Standard Library.
Here is a simple web application that writes "Hello World":

def index(environ, start_response):
    start_response("200 Ok", [('content-type', 'text/html')])
    return ['Hello World!']

if __name__ == "__main__":
    from wsgiref.simple_server import make_server
    server = make_server(HOST, PORT, index)
    print 'Starting up HTTP server on port %i...' % PORT
    server.serve_forever()

This is nice, but is not very useful for us, so lets add some kind of controller(or url dispatcher):

from dispatcher import Dispatcher

dispatcher = Dispatcher()
dispatcher.add(r'^/$', 'views.index')
dispatcher.add(r'^/hello/(?P<username>\w+)/$', 'views.hello')

HOST = 'localhost'
PORT = 8000

if __name__ == "__main__": 
    from wsgiref.simple_server import make_server
    server = make_server(HOST, PORT, dispatcher)
    print 'Starting up HTTP server on port %i...' % PORT
    server.serve_forever()

As you can see I've used a regular expression for the url mapping, just like Django uses.
Here is the dispatcher(dispatcher.py):

import re

class Dispatcher(object):
    def __init__(self, handle404 = None):
        self.urls = dict()
        self.request_path = ''
        if handle404:
            self.handle404 = handle404
        else:
            self.handle404 = self._404

    def __call__(self, environ, start_response):
        self.request_path = environ.get('PATH_INFO', '')
        for url in self.urls:            
            regex = re.compile(url)
            if regex.match(self.request_path):
                m = regex.match(self.request_path)                        
                mod_name, func_name = self._get_mod_func(self.urls[url])                
                try:
                    callback = getattr(__import__(mod_name, {}, {}, ['']), func_name)
                except ImportError, e:
                    raise Exception, "Could not import %s. Error was: %s" % (mod_name, str(e))
                except AttributeError, e:
                    raise Exception, "Tried %s in module %s. Error was: %s" % (func_name, mod_name, str(e))                
                args = (environ, start_response)
                kwargs = dict()                
                for i in regex.groupindex:
                    kwargs[i] = m.group(i)
                # Run callback with environ, start_response and args
                return callback(*args, **kwargs)
        # No match with the defined urls
        return self.handle404(environ, start_response)
            
    def _get_mod_func(self, callback):
        """
        Converts 'path.to.module.funtion' to ['path.to.module', 'function']
        """
        try:
            d = callback.rindex('.')
        except ValueError:
            return callback, ''
        return callback[:d], callback[d+1:]
    
    def _404(self, environ, start_response):
        start_response("404 Not Found", [('content-type', 'text/html')])
        return ['Not Found']

    def add(self, regex, handler):
        self.urls[regex] = handler

And here is the views(views.py):

def index(environ, start_response):
    start_response("200 Ok", [('content-type', 'text/html')])
    return ['Index']

def hello(environ, start_response, username):
    start_response("200 Ok", [('content-type', 'text/html')])
    return ['Hello %s' % username]

Simple eh?
This is just a start, now we can add more features, like encapsulate the start_response method and add some kind of HttpResponse.
I don't what do reinvent the wheel here and add another web framework to Python, just like Joe Gregorio says in this article. But to prove how trivial is to make a simple framework with WSGI and don't worry about the deployment at development phase.

Blog: "python" Posts

Internet lifestream with Django

Django error notification with jabber

Twitter user timeline with a Django templatetag

Fullread updates

Fullread

Smallr updates

Smallr

iPhone says Welcome to Django

A simple WSGI middleware dispatcher

About

CURRENTLY READING

Menu

Projects

Tags

Archives

Blogs

Last.fm

Social stuff

Latest comments

Most Popular Posts