source: trunk/amazonbot/amazonbot.py @ 17

#!/usr/bin/env python
# -*- coding: utf-8 -*-

__version__ = '$Revision$'
__author__ = 'Atzm WATANABE <sitosito@p.chan.ne.jp>'
__date__ = '$Date$'
__copyright__ = 'Copyright(C) 2006 Atzm WATANABE, all rights reserved.'
__license__ = 'Python'

import re
import sys
import time
import shlex
import random
import getopt

import MeCab
import nkf

from ircbot import SingleServerIRCBot
from irclib import nm_to_n

import config
config.init()

import my_amazon
my_amazon.setLocale(config.get('amazon', 'locale'))
my_amazon.setLicense(config.get('amazon', 'access_key'))

try:
        set, frozenset
except NameError:
        from sets import Set as set, ImmutableSet as frozenset

def uniq(sequence):
        """リストから重耇を取り陀く (é †ç•ªãŒç‹‚ã†ã®ã§æ³šæ„)
        """
        return list(set(sequence))

def unicoding(text):
        """text を匷制的に unicode オブゞェクトに倉換
        """
        if type(text) is unicode:
                return text
        return unicode(nkf.nkf('-w', text), 'utf-8')

def ununicoding(text, encoding='iso-2022-jp'):
        """text を指定された encoding で゚ンコヌドしraw str に匷制倉換
        """
        if type(text) is not unicode:
                return unicoding(text).encode(encoding)
        return text.encode(encoding)

def mecab_parse(text):
        """MeCab ã‚’äœ¿ã£ãŠåœ¢æ…‹çŽ è§£æžã—ïŒŒå›ºæœ‰åè©žãšäž€èˆ¬åè©žã ã‘ã‚’æŠœå‡ºã™ã‚‹
        """
        def choice_nominal(wlist):
                res = []
                for word, wtype in wlist:
                        wtypes = wtype.split('-')
                        if '固有名詞' in wtypes or ('名詞' in wtypes and '䞀般' in wtypes):
                                res.append(unicoding(word))
                return res

        text = ununicoding(text, 'utf-8')
        result = []
        tag = MeCab.Tagger('-Ochasen')
        for line in tag.parse(text).split('\n'):
                if not line or line == 'EOS':
                        break
                words = line.split()
                result.append((words[0], words[-1])) # word, word-type

        result = uniq(choice_nominal(result))
        return result

class AmazonBotBase(SingleServerIRCBot):
        """アマゟンボットのベヌスクラス
        å˜äœ“ã§ã¯ïŒŒå—ã‘å–ã£ãŸãƒ¡ãƒƒã‚»ãƒŒã‚žã®åœ¢æ…‹çŽ è§£æžãšåè©žæŠœå‡ºãŸã§ã—ã‹ã‚„ã‚‰ãªã„
        サブクラスで process_keyword を実装しお Amazon ぞク゚リを投げるべし

        サブクラスには onmsg_HOGEHOGE(self, conn, ev, to, args) ãƒ¡ã‚œãƒƒãƒ‰ã‚’äœœã‚‹ã“ãšã§ã‚³ãƒžãƒ³ãƒ‰è¿œåŠ å¯èƒœ
        コマンド曞匏は !HOGEHOGE arg [, arg2, ...] ずなる
        ヘルプはメ゜ッドに docstring を曞けば OK
        """
        def __init__(self):
                _server = [(config.get('irc', 'server'), config.get('irc', 'port', 'int'))]
                _nick = config.get('bot', 'nick')

                self._prev_time = time.time() - config.get('bot', 'freq', 'int')
                self._silent = False
                SingleServerIRCBot.__init__(self, _server, _nick, _nick)

        def start(self):
                try:
                        SingleServerIRCBot.start(self)
                except KeyboardInterrupt:
                        self.die(ununicoding(config.get('bot', 'bye')))

        def on_welcome(self, c, e):
                c.join(config.get('irc', 'channel'))
                if __debug__:
                        print >> sys.stderr, 'DEBUG> Joined %s' % config.get('irc', 'channel')

        def on_nicknameinuse(self, c, e):
                c.nick(c.get_nickname() + '_')

        def on_privmsg(self, c, e):
                return self.on_pubmsg(c, e, to=nm_to_n(e.source()))

        def on_pubmsg(self, c, e, to=config.get('irc', 'channel')):
                msg = unicoding(e.arguments()[0])

                if __debug__:
                        try:
                                print >> sys.stderr, 'DEBUG> pubmsg incoming "%s", should be reply to %s' % (ununicoding(msg, 'euc-jp'), to)
                        except:
                                print >> sys.stderr, 'DEBUG> pubmsg incoming, but could not en/decode message. should be reply to %s' % to

                if msg[0] == '!':
                        try:
                                words = shlex.split(ununicoding(msg, 'utf-8')[1:])
                        except:
                                return False
                        if not words:
                                return False
                        method = getattr(self, 'onmsg_%s' % words[0], lambda *arg: False)
                        return method(c, e, to, words[1:]) # words[0] == command name

                _current_time = time.time()
                if _current_time < self._prev_time + config.get('bot', 'freq', 'int'):
                        if __debug__:
                                cur = time.strftime('%H:%M:%S', time.localtime(_current_time))
                                go = time.strftime('%H:%M:%S', time.localtime(self._prev_time + config.get('bot', 'freq', 'int')))
                                print >> sys.stderr, 'DEBUG> Not expired: now %s, be expired at: %s' % (cur, go)
                        return False
                self._prev_time = _current_time

                self.silence(msg, c, e, to)
                if self._silent:
                        return False

                nominals = mecab_parse(msg)
                if not nominals:
                        if __debug__:
                                print >> sys.stderr, "DEBUG> Couldn't find nominal words"
                        return False

                title, url = self.process_keyword(' '.join(nominals))
                if title and url:
                        content = unicoding(config.get('bot', 'content'))
                        try:
                                message = ununicoding(': '.join([content, title, url]))
                        except UnicodeError, err:
                                # なぜかたたに unicode オブゞェクトを iso-2022-jp で゚ンコヌドできない
                                if __debug__:
                                        print >> sys.stderr, 'DEBUG> %s' % str(err)
                                return False

                        c.notice(to, message)
                        return True
                return False

        ACTIVE_PATTERN = re.compile(unicoding(config.get('bot', 'active_pattern')))
        SILENT_PATTERN = re.compile(unicoding(config.get('bot', 'silent_pattern')))
        def silence(self, msg, c, e, to):
                active = self.ACTIVE_PATTERN.search(msg)
                silent = self.SILENT_PATTERN.search(msg)
                if __debug__:
                        print >> sys.stderr, 'DEBUG> ACT_PATT: %s, SIL_PATT: %s' % (str(active), str(silent))

                if active:
                        self._silent = False
                        c.notice(to, ununicoding(config.get('bot', 'thanks')))
                elif silent:
                        self._silent = True
                        c.notice(to, ununicoding(config.get('bot', 'sorry')))

        def process_keyword(self, keyword):
                return [None, None]

class AmazonBot(AmazonBotBase):
        """アマゟンボットの実装クラス
        process_keyword メ゜ッドで Amazon ぞク゚リを投げお結果を返す
        """
        _AVAIL_PRODUCT_LINES = {
                'books-jp': '(和曞, default)',
                'books-us': '(掋曞)',
                'music-jp': '(ポピュラヌ音楜)',
                'classical-jp': '(クラシック音楜)',
                'dvd-jp': '(DVD)',
                'vhs-jp': '(ビデオ)',
                'electronics-jp': '(゚レクトロニクス)',
                'kitchen-jp': '(ホヌムキッチン)',
                'software-jp': '(゜フトりェア)',
                'videogames-jp': '(ゲヌム)',
                'magazines-jp': '(雑誌)',
                'toys-jp': '(おもちゃホビヌ)',
        }

        def __init__(self):
                AmazonBotBase.__init__(self)

        def get_version(self):
                return 'AmazonBot by %s, based on python-irclib' % __author__

        def onmsg_isbn(self, c, e, to, args):
                """Syntax: !isbn <ISBN number>
                """
                return self.onmsg_asin(c, e, to, args)
        def onmsg_asin(self, c, e, to, args):
                """Syntax: !asin <ASIN number>
                """
                if __debug__:
                        print >> sys.stderr, 'DEBUG> in asin command: %s' % str(args)

                try:
                        data = my_amazon.searchByASIN(args[0])
                except my_amazon.AmazonError, err:
                        c.notice(to, ununicoding(config.get('bot', 'no_products')))
                        if __debug__:
                                print >> sys.stderr, 'DEBUG> Caught AmazonError in onmsg_asin: %s' % str(err)
                        return False
                except IndexError, err:
                        c.notice(to, 'Please specify an argument.')
                        return False

                return self._process_onmsg(c, e, to, data)

        def onmsg_k(self, c, e, to, args): return self.onmsg_keyword(c, e, to, args)
        def onmsg_keyword(self, c, e, to, args):
                """Syntax: !keyword [-h] [-t type] <keyword1> [, keyword2, ...]
                """
                if __debug__:
                        print >> sys.stderr, 'DEBUG> in keyword command: %s' % str(args)

                try:
                        options, rest = getopt.getopt(args, 't:h', ['type=', 'help'])
                except getopt.GetoptError, err:
                        if __debug__:
                                print >> sys.stderr, 'DEBUG> Caught GetoptError in onmsg_keyword: %s' % str(err)
                        return False

                keyword = ' '.join(rest).strip()
                product_line = 'books-jp'
                for opt, val in options:
                        if opt in ['-t', '--type']:
                                if val not in self._AVAIL_PRODUCT_LINES.keys():
                                        c.notice(to, 'Type "%s" is not available.' % val)
                                        return False

                                product_line = val
                                break

                        elif opt in ['-h', '--help']:
                                _from = nm_to_n(e.source()) # ログを流しおしたうのでヘルプは盎接送信元ぞ
                                c.notice(_from, ununicoding('Available types:'))

                                for key, val in self._AVAIL_PRODUCT_LINES.iteritems():
                                        time.sleep(1) # XXX: 連続投皿するず匟かれるこずがあるので暫定察凊
                                        c.notice(_from, ununicoding(' * %s: %s' % (key, val)))

                                return True

                if not keyword:
                        c.notice(to, 'Please specify keywords.')
                        return False

                if __debug__:
                        fmt = 'DEBUG> keyword="%s", product_line=%s'
                        print >> sys.stderr, fmt % (ununicoding(keyword, 'euc-jp'), product_line)

                try:
                        data = my_amazon.searchByKeyword(keyword, product_line=product_line)
                except my_amazon.AmazonError, err:
                        c.notice(to, ununicoding(config.get('bot', 'no_products')))
                        if __debug__:
                                print >> sys.stderr, 'DEBUG> Caught AmazonError in onmsg_amazon: %s' % str(err)
                        return False

                return self._process_onmsg(c, e, to, data)

        def onmsg_h(self, c, e, to, args): return self.onmsg_help(c, e, to, args)
        def onmsg_help(self, c, e, to, args):
                """Syntax: !help
                """
                if __debug__:
                        print >> sys.stderr, 'DEBUG> in help command: %s' % str(args)

                _from = nm_to_n(e.source()) # ログを流しおしたうのでヘルプは盎接送信元ぞ
                docs = []
                for key in dir(self):
                        val = getattr(self, key, '')
                        if __debug__:
                                print >> sys.stderr, 'DEBUG> key=%s, val=%s' % (key, ununicoding(str(val), 'euc-jp'))

                        if key[:6] != 'onmsg_':
                                continue

                        doc = val.__doc__
                        if doc:
                                doc = doc.strip()
                                if not doc:
                                        continue
                                time.sleep(1) # XXX: 連続投皿するず匟かれるっぜいので暫定察凊
                                c.notice(_from, doc)

                return True

        def _process_onmsg(self, c, e, to, data):
                if type(data.Details) is not list:
                        data.Details = [data.Details]

                detail = random.choice(data.Details)
                title = ununicoding(detail.ProductName)
                url = ununicoding(detail.URL)
                c.notice(to, '%(title)s: %(url)s' % locals())

                return True

        def process_keyword(self, keyword):
                keyword = ununicoding(keyword, 'utf-8')
                if __debug__:
                        print >> sys.stderr, 'DEBUG> KEYWORD: %s' % ununicoding(keyword, 'euc-jp')

                try:
                        data = my_amazon.searchByBlended(keyword)
                        if type(data.ProductLine) is not type([]):
                                data.ProductLine = [data.ProductLine]
                except my_amazon.AmazonError, err:
                        if __debug__:
                                print >> sys.stderr, 'DEBUG> Caught AmazonError: %s' % str(err)
                        return [None, None]

                product_line = random.choice(data.ProductLine)
                detail = random.choice(product_line.ProductInfo.Details)

                url = unicoding(getattr(detail, 'URL', None))
                product_name = unicoding(getattr(detail, 'ProductName', None))

                return [product_name, url]

if __name__ == '__main__':
        bot = AmazonBot()
        bot.start()
        print '> Bye ;)'