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.
 
 

298 lines
10 KiB

#!/usr/bin/env python
# -*- 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._operations = 0
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
return
def _save(self):
"""Check the number of operations and, if it above the minimum value,
save the config file."""
max_count = self._options['Network_Manager']['save_count']
_log.debug('Operations %d, max %d', self._operations, max_count)
if self._operations > max_count:
self._options.save()
self._operations = 0
_log.debug('Config file saved')
return
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)
# add Networks own options
self._options.add_group('Network_Manager', 'Network Manager')
self._options.add_option(
group='Network_Manager',
option='save_count',
help='Number of network operations till the config file ' \
'is saved',
type='int',
metavar='OPERATIONS',
default=5,
is_cmd_option=False)
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)
self._operations += 1
# don't check the operations count inside the loop; if Mitter
# fails, all the last messages can still be retrieved ('cause the
# config file wasn't saved yet.)
self._save()
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
results = []
for shortcut in self._targets(network):
results.append(self.networks[shortcut].update(status, reply_to))
self._operations += 1
self._save()
return results
def reply_prefix(self, message):
"""Return the prefix to be used in the reply for that message."""
assert(isinstance(message, NetworkData))
return self.networks[message.network].reply_prefix(message)
def repost(self, message):
"""Repost a message in the user's timeline. The network used is
the same in the message."""
assert(isinstance(message, NetworkData))
self.networks[message.network].repost(message)
self._operations += 1
self._save()
return
def favourite(self, message):
"""Change the favourite status of the message."""
assert(isinstance(message, NetworkData))
self.networks[message.network].favourite(message)
self._operations += 1
self._save()
return
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)
self._operations += 1
self._save()
return
def message(self, message_id, network):
"""Return a single NetworkData object for a specified message."""
# TODO: This should probably be named parent() and receive a
# NetworkData. Still returns a single NetworkData object.
if not network in self.networks:
raise NetworksNoSuchNetworkError(network)
data = self.networks[network].message(message_id)
data.network = network
self._operations += 1
self._save()
return data
def link(self, message):
"""Returns a link directly to the message."""
assert(isinstance(message, NetworkData))
return self.networks[message.network].link(message)
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)
self._operations += 1
self._save()
return result
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
def can_delete(self, message):
"""Return True if the message can be deleted; False otherwise."""
assert(isinstance(message, NetworkData))
return self.networks[message.network].can_delete(message)
def can_reply(self, message):
"""Return True if the message can be replied; False otherwise."""
assert(isinstance(message, NetworkData))
return self.networks[message.network].can_reply(message)
def can_repost(self, message):
"""Return True if the message can be resposted; False otherwise."""
assert(isinstance(message, NetworkData))
return self.networks[message.network].can_repost(message)
def can_favourite(self, message):
"""Return True if the message can be favourited/unfavourited; False
otherwise."""
assert(isinstance(message, NetworkData))
return self.networks[message.network].can_favourite(message)
# TODO: Function to return a regexp for usernames