#!/usr/bin/python
# -*- coding: utf-8 -*-
# Mitter, micro-blogging client
# 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 logging
import cmd
import datetime
import warnings
import mitterlib.ui.helpers.console_utils as console_utils
import mitterlib.constants
from mitterlib.network import NetworksNoNetworkSetupError, NetworksError
from mitterlib.network.networkbase import NetworkError, \
NetworkPermissionDeniedError
_log = logging.getLogger('ui.cmd')
class Interface(cmd.Cmd):
"""The command line interface for Mitter."""
NAMESPACE = 'cmd'
# -----------------------------------------------------------------------
# Methods required by cmd.Cmd (our commands)
# -----------------------------------------------------------------------
def do_config(self, line=None):
"""Setup the networks."""
options = self._connection.settings()
console_utils.authorization(options, self._options)
return
def do_timeline(self, line):
"""Return a list of new messages in your friends timeline."""
try:
self._show_messages(self._connection.messages())
except NetworksNoNetworkSetupError:
# call the config
self.do_config()
except NetworkError:
print 'Network failure. Try again in a few minutes.'
return
def do_replies(self, line):
"""Get a list of replies to you."""
try:
self._show_messages(self._connection.replies(), is_timeline=False)
except NetworksNoNetworkSetupError:
self.do_config()
except NetworkError:
print 'Network failure. Try again in a few minutes.'
return
def do_update(self, line):
"""Update your status."""
if self._update(line):
print 'Status updated'
else:
print 'Failed to update your status. Try again in a few minutes.'
def do_exit(self, line):
"""Quit the application."""
_log.debug('Exiting application')
return True
def do_EOF(self, line):
"""Quit the application (it's the same as "exit"). You can also use
Ctrl+D."""
print # Cmd doesn't add an empty line after the ^D
return self.do_exit(None)
def do_rt(self, line):
""""Retweet" a message in your list."""
pos = int(line)
if not self._check_message(pos):
return
original_message = self._messages[pos-1]
new_message = console_utils.make_retweet(original_message)
return self.do_update(new_message)
def do_r(self, line):
"""Same as "reply"."""
return self.do_reply(line)
def do_reply(self, line):
"""Reply to a message. Use "reply"/"r" ."""
line_split = line.split()
pos = int(line_split[0]) # message (cmd strips the
# command already)
if not self._check_message(pos):
return
message = self._messages[pos - 1]
if self._update(' '.join(line_split[1:]), reply_to=message):
print 'Reply sent.'
else:
print "Couldn't send your reply. Try again in a few minutes."
return
def do_delete(self, line):
"""Delete a message. You must provide the number of the displayed\
message."""
message_id = int(line)
real_message_id = self._messages[message_id - 1]
try:
self._connection.delete_message(real_message_id)
except NetworkPermissionDeniedError:
print 'Permission denied.'
return
print 'Message deleted.'
return
def do_thread(self, line):
"""Retrieves the thread about a single message (like a reply.) Be
aware that this may consume a lot of your hourly requests if the
thread is too long."""
message_id = int(line)
_log.debug('Message in pos %d', message_id)
if not self._check_message(message_id):
return
message = self._messages[message_id - 1]
thread = console_utils.fetch_thread(message, self._connection)
console_utils.print_thread(thread, self._connection)
return
def emptyline(self):
"""Called when the user doesn't call any command. Default is to repeat
the last command; we are going to call timeline() again."""
return self.do_timeline(None)
def default(self, line):
"""Called when we receive an unknown command; default is error
message, we are going to call update() instead."""
return self.do_update(line)
# -----------------------------------------------------------------------
# Helper functions
# -----------------------------------------------------------------------
def _check_message(self, message_id):
"""Check if a message is valid in the current list."""
if message_id < 1 or message_id > len(self._messages):
print
print 'No such message.'
print
return False
return True
def _show_messages(self, data, is_timeline=True):
"""Function called after we receive the list of messages."""
if is_timeline:
self._last_update = datetime.datetime.now()
self._messages = data
console_utils.print_messages(data, self._connection,
show_numbers=True)
self._update_prompt()
return
def _post_delete(self, data, error):
"""Function called after we delete a message."""
if error:
if error == 403:
# Ok, we are *assuming* that, if you get a Forbidden
# error, it means it's not your message.
print "You can't delete this message."
# TODO: we are using Logging.Error in the Twitter
# object when we get this error. So the user will
# see connection errors instead of this simple
# message.
else:
print 'Error deleting message.'
else:
print 'Message deleted.'
self._update_prompt()
return
def _update(self, status, reply_to=None):
"""Send the update to the server."""
try:
self._connection.update(status, reply_to=reply_to)
except (NetworksError, NetworkError):
# TODO: capture the proper exception.
# TODO: Also, NetworkError's should never get here. Networks
# should catch that (leaving the status kinda messed.)
return False
except MessageTooLongWarning:
print 'Your message is too long. Update NOT send.'
return False
self._update_prompt()
return True
def _update_prompt(self):
"""Update the command line prompt."""
# check the requests limits for every network
requests = self._connection.available_requests()
available = []
for network in requests:
if requests[network] >= 0:
# just show information for networks that count that
available.append('%s (%s): %d' % (
self._connection.name(network),
network,
requests[network]))
if self._last_update:
update_text = self._last_update.strftime('%H:%M')
else:
update_text = 'Never'
self.prompt = ('Last update: %s [%s]\nMitter> ' %
(update_text, ', '.join(available)))
return
# -----------------------------------------------------------------------
# Methods required by the main Mitter code
# -----------------------------------------------------------------------
def __init__(self, connection, options):
"""Class initialization."""
cmd.Cmd.__init__(self)
self._options = options
self._last_update = None
self._connection = connection
self._messages = []
intro = ['Welcome to Mitter %s.' % (mitterlib.constants.version),
'',
'To get a list of available commands, type "help".',
'',
"If you start a line with something that it's not a command, " \
'it will be considered ' \
"a status update (so you don't need to type any commands to " \
'just update your status.',
'',
'An empty line will retrieve the latest updates from your ' \
'friends.',
'',
'']
import textwrap
wrapper = textwrap.TextWrapper()
intros = []
for line in intro:
if not line:
intros.append('') # textwrap doesn't like empty lines
else:
for reident in wrapper.wrap(line):
intros.append(reident)
self.intro = '\n'.join(intros)
self.prompt = 'Mitter> '
return
def __call__(self):
"""Make the object callable; that's the only requirement for
Mitter."""
warnings.simplefilter('error') # Warnings are exceptions
self.cmdloop()
return
@classmethod
def options(self, options):
# no options for this interface
return