A micro-blogging tool with multiple interfaces.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

222 lines
7.2 KiB

#!/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 <http://www.gnu.org/licenses/>.
import logging
import os.path
from mitterlib.network.networkbase import NetworkData
from mitterlib import module_search
_log = logging.getLogger('mitterlib.network.Networks')
# List of files that are not networks
SKIPPABLES = ('__init__.py', 'networkbase.py')
#--------------------------------------------------------------------
# Helper functions
#--------------------------------------------------------------------
def _import_name(module):
"""Based on the name of the module, return the proper "import"
statement."""
(name, _) = os.path.splitext(module)
return 'mitterlib.network.%s' % (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 module_search(__file__, SKIPPABLES):
import_name = _import_name(module_name)
module = __import__(import_name, fromlist=[import_name])
connection = module.Connection(self._options)
self._networks[connection.SHORTCUT] = connection
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
def is_self(self, message):
"""Return True if the message belongs to the logged user."""
if self.networks[message.network].username() == message.username:
return True
return False
# 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