# -.- coding: utf-8 -.- # Zeitgeist # # Copyright © 2010 Seif Lotfy # Copyright © 2010 Stuart Langridge # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Lesser General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Lesser General Public License for more details. # # You should have received a copy of the GNU Lesser General Public License # along with this program. If not, see . import logging import socket import gobject import gnomekeyring import json import httplib2 import urlparse import cgi import urllib import os from subprocess import * from couchdb.client import uri as couchdburi from oauth import oauth from gi.repository import Soup try: from ubuntu_sso.main import SSOCredentials except ImportError: SSOCredentials = None from oauth import oauth from desktopcouch.records.record import Record as CouchRecord from desktopcouch.records.server import CouchDatabase try: from _zeitgeist.engine.datamodel import Event from _zeitgeist.engine.extension import Extension AmExtension = True except ImportError: # _zeitgeist is only available if you're the extension # so define an Extension class so that in the subprocess we don't error class Extension(): pass from zeitgeist.client import ZeitgeistClient from zeitgeist.datamodel import TimeRange, Event, Subject try: CLIENT = ZeitgeistClient() except: CLIENT = None logging.basicConfig(level=logging.DEBUG) log = logging.getLogger("zeitgeist.ubuntu_one") class DesktopCouchGatewayLauncher(Extension): PUBLIC_METHODS = ["pointless"] # have to define one def __init__(self, engine): devnull = open(os.devnull, "w") # Launch my own file as a separate subprocess # The subprocess launches as __main__ log.debug("Launching DesktopCouch Gateway subprocess...") p = Popen(["python", __file__], stdin=devnull) log.debug("DesktopCouch Gateway subprocess launched as pid %s" % p.pid) def pointless(self): """A pointless public method, because I have to have one :-)""" pass class DesktopCouchGateway(): """Synchronise all zeitgeist events to and from desktopcouch. This enables your zeitgeist event log to be available on all your machines. """ def __init__ (self): self.ids = [] self.c_ids = [] if CLIENT: self.token, self.id = self._get_U1_oauth_token() if self.token and self.id: self.computer = socket.gethostname() self.database = CouchDatabase("zeitgeist_event_log", create=True) CLIENT.install_monitor(TimeRange.always(), [], self.push_events_to_couch, self.remove_ids) def _get_U1_oauth_token(self): """Get the token from the keyring""" gobject.set_application_name("Zeitgeist DesktopCouch Gateway") if SSOCredentials is not None: creds = SSOCredentials('Ubuntu One') info = creds.find_credentials('Ubuntu One') if not info: return(None, None) consumer = oauth.OAuthConsumer(info['consumer_key'], info['consumer_secret']) secret='oauth_token=%s&oauth_token_secret=%s' % (info['token'], info['token_secret']) else: consumer = oauth.OAuthConsumer("ubuntuone", "hammertime") items = [] items = gnomekeyring.find_items_sync( gnomekeyring.ITEM_GENERIC_SECRET, {'ubuntuone-realm': "https://ubuntuone.com", 'oauth-consumer-key': consumer.key}) if len(items) == 0: return (None, None) secret = items[0].secret token = oauth.OAuthToken.from_string(secret) log.debug("U1 token for this machine is %s"% token.key) # now get their ubuntu one ID url = "https://one.ubuntu.com/api/account" oauth_request = oauth.OAuthRequest.from_consumer_and_token(http_url=url, http_method="GET", oauth_consumer=consumer, token=token) oauth_request.sign_request(oauth.OAuthSignatureMethod_PLAINTEXT(), consumer, token) oauth_header = oauth_request.to_header() client = httplib2.Http() resp, content = client.request(url, "GET", headers=oauth_header) if resp['status'] == "200": doc = json.loads(content) log.debug("User's U1 id is %s" % doc["id"]) return token.key, doc["id"] else: return (None,None) def push_events_to_couch(self, timerange, events): # Create a new record for event for event in events: self.push_to_couch(event) def remove_ids(self, time_range, ids): pass def _serialize_event_to_json(self, event): subjects = [] for subject in event.subjects: j_subject = { "uri":subject.uri, "interpretation": subject.interpretation, "manifestation": subject.manifestation, "mimetype": subject.mimetype, "text": subject.text, "origin": subject.origin } subjects.append(j_subject) j_event = { "id": event.id, "timestamp": int(event.timestamp), "actor": event.actor, "interpretation": event.interpretation, "manifestation": event.manifestation, "subjects": subjects, "payload": event.payload } return j_event def _deserialize_event_from_json(self, json): d = json["event"] ev = Event() ev.timestamp = int(d.get("timestamp", "")) ev.interpretation = str(d.get("interpretation", "").encode("UTF-8")) ev.manifestation = str(d.get("manifestation", "").encode("UTF-8")) ev.actor = str(d.get("actor", "").encode("UTF-8")) #del json["event"] subjects = d.get("subjects", []) for sd in subjects: subj = Subject() subj.uri = str(sd.get("uri", "").encode("UTF-8")) subj.interpretation = str(sd.get("interpretation", "").encode("UTF-8")) subj.manifestation = str(sd.get("manifestation", "").encode("UTF-8")) subj.origin = str(sd.get("origin", "").encode("UTF-8")) subj.mimetype = str(sd.get("mimetype", "").encode("UTF-8")) subj.text = str(sd.get("text", "").encode("UTF-8")) subj.storage = str(sd.get("storage", "").encode("UTF-8")) ev.append_subject(subj) return ev def fake_callback(self, x=None, y=None, z=None): pass def push_to_couch(self, event): record = CouchRecord({ "u1_id": self.id, "u1_computer": self.computer, "u1_token": self.token, "event": self._serialize_event_to_json(event) }, "http://zeitgeist-project.com/documentation/desktopcouch/event") # Put the record into the database if not event.id in self.ids: self.database.put_record(record) self.ids.append(event.id) log.debug("Pushed into CouchDB event.id: %i" % self.id) def push_to_local(self, record): if not record["u1_token"] == self.token: event = self._deserialize_event_from_json(record) CLIENT.insert_event(event, self.fake_callback, None) else: log.debug("got an event generated by myself; ignoring it") class ZeitgeistMonitor(object): def __init__(self, dbname): # First, we have to get an authenticated URL for the changes feed for our database self.db = db = CouchDatabase(dbname) changes_feed_uri = couchdburi( db._server.resource.uri, db.db.name, "_changes", since=0, # if you remember the last sequence number you saw you can fill that in here feed="continuous", heartbeat=10000 # send a newline every 10 seconds to keep the connection alive ) oauth_data = db._server.resource.http.oauth_data consumer = oauth.OAuthConsumer(oauth_data["consumer_key"], oauth_data["consumer_secret"]) token = oauth.OAuthToken(oauth_data["token"], oauth_data["token_secret"]) schema, netloc, path, params, query, fragment = \ urlparse.urlparse(changes_feed_uri) querystr_as_dict = dict(cgi.parse_qsl(query)) req = oauth.OAuthRequest.from_consumer_and_token( consumer, token, http_method = "GET", http_url = changes_feed_uri, parameters = querystr_as_dict ) req.sign_request(oauth.OAuthSignatureMethod_HMAC_SHA1(), consumer, token) signed_changes_feed_uri = req.to_url() # Now that we have an authenticated URL, use libsoup to fetch it, async, and # continue to fetch it as changes happen self.gateway = DesktopCouchGateway() self.session = Soup.SessionAsync() msg = Soup.Message.new("GET", signed_changes_feed_uri) self.session.queue_message(msg, self.complete, signed_changes_feed_uri) msg.connect("got-chunk", self.got_chunk) def got_chunk(self, msg, chunkbuffer): data = ''.join([chr(x) for x in chunkbuffer.get_data()]) if not data.strip(): # got a heartbeat; this keeps the HTTP connection alive return try: actual_data = json.loads(data) except ValueError: print "weird, we got something that wasn't JSON: '%s'" % data return self.gateway.push_to_local(self.db.get_record(actual_data["id"])) def complete(*args): # should never get called, because connection is long-lived print "complete: this shouldn't happen. if it does, restart the connection." if __name__ == "__main__": # This is being run as a subprocess # So, start the monitor print "in the child" monitor = ZeitgeistMonitor("zeitgeist_event_log") print "set up monitor ok" gobject.MainLoop().run()