#!/usr/bin/python2.5 # -*- coding: utf-8 -*- # Mitter, a simple client for Twitter # Copyright (C) 2007, 2008 The Mitter Contributors # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU 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 General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . import os.path import glob import logging from mitterlib.network.networkbase import NetworkData _log = logging.getLogger('mitterlib.network.Networks') # List of files that are not networks SKIPPABLES = ('__init__.py', 'networkbase.py') #-------------------------------------------------------------------- # Helper functions #-------------------------------------------------------------------- def _import_name(module): (name, _) = os.path.splitext(module) return 'mitterlib.network.%s' % (name) def _networks(): network_dir = os.path.dirname(__file__) networks = glob.glob(os.path.join(network_dir, '*.py')) for network in networks: network_name = os.path.basename(network) if network_name in SKIPPABLES: # not a real network continue module_name = _import_name(network_name) yield module_name #-------------------------------------------------------------------- # Exceptions #-------------------------------------------------------------------- class NetworksError(Exception): """Basic Networks exception.""" pass class NetworksNoSuchNetworkError(NetworksError): """The request network does not exists.""" def __init__(self, network): self._network = network def __str__(self): return 'Unknown network %s' % (self._network) pass class NetworksNoNetworkSetupError(NetworksError): """There are no networks set up.""" pass class NetworksErrorLoadingNetwork(NetworksError): """Error loadig one of the networks.""" def __init__(self, network, exception): self._network = network self._exception = str(exception) def __str__(self): return "Couldn't load %s (%s)" % (self._network, self._exception) class Networks(object): """Network transparency layer: Keeps a list of available networks and send requests to all those who have been setup.""" def __init__(self, options): self._networks = {} self._options = options self.options() @property def networks(self): if self._networks: return self._networks for module_name in _networks(): try: module = __import__(module_name, fromlist=[module_name]) connection = module.Connection(self._options) self._networks[connection.SHORTCUT] = connection except ImportError, ex: raise NetworksErrorLoadingNetwork(module_name, ex) return self._networks def _targets(self, shortcut): """Select a network based on the shortcut. If the shortcut is None, returns all available network shortcuts.""" if shortcut: if not shortcut in self.networks: raise NetworksNoSuchNetworkError(shortcut) else: targets = [shortcut] else: targets = self.networks setup = False for target in targets: if self.networks[target].is_setup(): setup = True yield target if not setup: raise NetworksNoNetworkSetupError def settings(self): """Return a dictionary with the options that the interfaces need to setup.""" result = [] for shortcut in self.networks: settings = { 'shortcut': shortcut, 'name': self.networks[shortcut].NAMESPACE, 'options': self.networks[shortcut].AUTH, } result.append(settings) return result def options(self): """Request all networks to add their options.""" for shortcut in self.networks: conn = self.networks[shortcut] conn.options(self._options) return def name(self, shortcut): """Return the name of a network based on the shortcut.""" try: name = self.networks[shortcut].NAMESPACE except KeyError: raise NetworksNoSuchNetworkError(shortcut) return name # This is basically a copy of all methods available in NetworkBase, with # the additional parameter "network" (to request data from just one # source) def messages(self, network=None): """Return a list of NetworkData objects for the main "timeline" (the default list presented to the user.)""" result = [] for shortcut in self._targets(network): for message in self.networks[shortcut].messages(): message.network = shortcut result.append(message) return result def update(self, status, reply_to=None, network=None): """Update the user status. Must return the id for the new status.""" if reply_to and isinstance(reply_to, NetworkData): # If you pass a NetworkData object, we get the proper network network = reply_to.network for shortcut in self._targets(network): self.networks[shortcut].update(status, reply_to) return None def delete_message(self, message, network=None): """Delete an update. Message can be a NetworkData object, in which case network is not necessary; otherwise a network must be provided.""" if isinstance(message, NetworkData): network = message.network self.networks[network].delete_message(message) return def message(self, message_id, network): """Return a single NetworkData object for a specified message.""" if not network in self.networks: raise NetworksNoSuchNetworkError(network) data = self.networks[network].message(message_id) data.network = network return data def replies(self, network=None): """Return a list of NetworkData objects for the replies for the user messages.""" result = [] for shortcut in self._targets(network): for message in self.networks[shortcut].replies(): message.network = shortcut result.append(message) return result def inbox(self, network=None): """Return a list of NetworkData objects for the direct messages/inbox of the user.""" return [] def available_requests(self, network=None): """Return a dictionary with the available requests the user can make to each network before getting capped.""" result = {} for shortcut in self._targets(network): requests = self.networks[shortcut].available_requests() result[shortcut] = requests return result