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.

356 lines
12 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
15 years ago
import urllib2
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._proxy_handler = None
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 user(self, network=None):
"""Return a :class:`NetworkUser` representation for the logged user in
the requested network. If no network is selected, will return the
first found."""
shortcut = self._targets(network).next() # need this due yield
return self.networks[shortcut].user
def _targets(self, shortcut=None):
"""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['NetworkManager']['save_count']
_log.debug('Operations %d, max %d', self._operations, max_count)
if self._operations > max_count:
self._options.save()
15 years ago
self._operations = 0
_log.debug('Config file saved')
return
def _proxy(self):
"""Check and install the proxy."""
if not self._options['NetworkManager']['proxy']:
return
if self._proxy_handler:
return
proxy = self._options['NetworkManager']['proxy']
if not '://' in proxy:
proxy = 'http://' + proxy
_log.debug('Installing proxy ' + proxy)
self._proxy_handler = urllib2.ProxyHandler({
'http': proxy,
'https': proxy})
opener = urllib2.build_opener(self._proxy_handler)
urllib2.install_opener(opener)
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):
"""Add the options for all networks and the network manager options
itself."""
# Remember that, in case you add or remove any options here, you
# should also update the CHEAT-CODES file.
for shortcut in self.networks:
conn = self.networks[shortcut]
conn.options(self._options)
# add Networks own options
self._options.add_group('NetworkManager', 'Network Manager')
self._options.add_option(
'--timeout',
group='NetworkManager',
option='timeout',
help='Timeout for requests in all networks.',
type='int',
metavar='SECONDS',
default=120)
self._options.add_option(
'--proxy',
group='NetworkManager',
option='proxy',
help='Proxy server (Note: This is not needed on Windows " \
"and OS X systems as Mitter will use the system " \
"value.)',
metavar='SERVER',
default='')
self._options.add_option(
group='NetworkManager',
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.)"""
self._proxy()
result = []
for shortcut in self._targets(network):
for message in self.networks[shortcut].messages():
message.network = shortcut
result.append(message)
self._operations += 1
15 years ago
# 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."""
self._proxy()
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
15 years ago
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))
self._proxy()
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._proxy()
self.networks[message.network].repost(message)
self._operations += 1
15 years ago
self._save()
return
def favourite(self, message):
"""Change the favourite status of the message."""
assert(isinstance(message, NetworkData))
self._proxy()
self.networks[message.network].favourite(message)
self._operations += 1
15 years ago
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._proxy()
self.networks[network].delete_message(message)
self._operations += 1
15 years ago
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)
self._proxy()
data = self.networks[network].message(message_id)
data.network = network
self._operations += 1
15 years ago
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)
15 years ago
def replies(self, network=None):
"""Return a list of NetworkData objects for the replies for the user
messages."""
result = []
self._proxy()
for shortcut in self._targets(network):
for message in self.networks[shortcut].replies():
message.network = shortcut
result.append(message)
self._operations += 1
15 years ago
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))
15 years ago
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)
15 years ago
# TODO: Function to return a regexp for usernames