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.
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.
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.NOTE: If you are using a bookmarking service that is not available, please let me know: http://fullread.com/contacts/?subject=Bookmarking%20service%20support.
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.
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()
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()
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
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]