- or go directly to \037http://anidb.info\017" % Settings.trigger invalid_command = "\002Usage\017:\t/%s [-pb] <title> - or go directly to \037http://anidb.info\017" % Settings.command connection_error = "\002Error\017:\tUnable to retrieve AniDB info [Error: %s]" prefix = "\002AniDB\017: " noresults = "No results found for '%s'!" url = "\037%s\017" class Helper: _last_message = 0 @staticmethod def clear_nick(nickname): """ Almost identical to xchat.strip for older X-Chat/Python plugin versions. """ if hasattr(xchat, 'strip'): return xchat.strip(nickname) else: if nickname.startswith("\x03"): return nickname[3:] else: return nickname @staticmethod def say(text, context=None): """ Delays a SAY command and makes sure it is actually run in the X-Chat main thread. """ def _say_timer(userdata): if time.time() - Helper._last_message < Settings.message_delay: return 1 else: context, text = userdata context.command("SAY " + text) Helper._last_message = time.time() return 0 if not context: context = xchat.get_context() xchat.hook_timer(100, _say_timer, (context, text)) class Hooks: @staticmethod def initialize(): """ Adds the required X-Chat hooks for various events. """ if Settings.trigger: xchat.hook_print("Channel Message", Hooks.channeltext, "Channel Message") xchat.hook_print("Your Message", Hooks.channeltext, "Your Message") if Settings.command: xchat.hook_command(Settings.command, Hooks.command, help="/%s searchstring" % Settings.command) @staticmethod def channeltext(word, word_eol, userdata): if xchat.get_info("channel") in Settings.excluded_channels or \ not len(word) >= 2 or \ not word[1].startswith(Settings.trigger): return xchat.EAT_NONE # Clean arguments user, message = Helper.clear_nick(word[0]), word[1] words = [s.strip() for s in message.split(" ") if s.strip()] # No argument was passed if len(words) < 2: if userdata == "Channel Message": xchat.command("NOTICE %s %s" % (user, Messages.invalid_trigger)) else: print Messages.invalid_trigger return xchat.EAT_PLUGIN searchstring = " ".join(words[1:]) HttpThread(searchstring, xchat.get_context()).start() return xchat.EAT_PLUGIN @staticmethod def command(word, word_eol, userdata): if len(word) < 2 or len(word_eol) < 2 or not word_eol[1].strip(): print Messages.invalid_command return xchat.EAT_ALL # Using /anidb -p <search> will SAY the result in current channel parser = optparse.OptionParser(usage=optparse.SUPPRESS_USAGE, add_help_option=False, version=False) # Suppress default behaviour of exiting on error def error(msg): print parser.get_usage() parser.error = error parser.add_option("-p", "--public", action="store_true", dest="public", default=False) parser.add_option("-b", "--browser", action="store_true", dest="browser", default=False) options, args = parser.parse_args(args=word[1:]) # Evaluate switches if options.public: output = "public" else: output = "private" if options.browser: callback = Callbacks.openurl else: callback = None if len(args) < 1: print Messages.invalid_command else: HttpThread(" ".join(args), xchat.get_context(), output=output, callback=callback).start() return xchat.EAT_ALL class Callbacks: @staticmethod def connectionerror(error, context, output="private"): if output == "public": Helper.say(Messages.connection_error % error, context) else: print Messages.connection_error % error @staticmethod def noresultserror(searchstring, context, output="private"): message = Messages.prefix + (Messages.noresults % searchstring) if output == "public": Helper.say(message, context) else: print message @staticmethod def displayresult(results, context, output, search_url=None): for result in results: msg_parts = [] # Main title if result.title: msg_parts.append(result.title) # Official japanese title if result.japanesetitle: msg_parts.append(result.japanesetitle) # Episode count if result.epcount: if result.epcount == 1: eptext = "episode" else: eptext = "episodes" msg_parts.append("%i %s" % (result.epcount, eptext)) # URL msg_parts.append(Messages.url % result.url) message = Messages.prefix + " - ".join(msg_parts) if output == "public": Helper.say(message, context) else: print message if len(results) > 1 and search_url: message = Messages.prefix + "Remaining results at " + Messages.url % search_url if output == "public": Helper.say(message, context) else: print message @staticmethod def openurl(result): if sys.version >= "2.5": method = 2 else: method = 0 webbrowser.open(result.url, method) class HttpThread(threading.Thread): def __init__(self, searchstring, context, output="public", callback=None): threading.Thread.__init__(self) self.searchstring_ = searchstring self.output_ = output self.callback_ = callback self.context_ = context def start(self): """ This doesn't really start a thread on win32. Why? Because I can't figure out how to use threads in X-Chat without it crashing. """ if sys.platform == 'win32': self.run() else: threading.Thread.start(self) def run(self): url = Settings.anidb_searchurl % urllib.quote_plus(self.searchstring_) try: request = HttpThread._build_connection(url) # Open stream opener = urllib2.build_opener(urllib2.HTTPHandler) connection = opener.open(request) data = HttpThread._decode_gzip_stream(connection) connection.close() except Exception, e: Callbacks.connectionerror(e, self.context_) return if "No results." in data: Callbacks.noresultserror(self.searchstring_, self.context_, self.output_) return # Populate result by using regexps results = [AnidbResult(self.searchstring_, url)] m = re.search(Settings.regexp_titleid, data) if m: # Found single match for search query result = results[0] result.title = m.group(1) result.url_ = m.group(2) m = re.search(Settings.regexp_epcount, data) if m: result.epcount = int(m.group(1)) m = re.search(Settings.regexp_japanese_title, data) if m: result.japanesetitle = m.group(1) else: # Since the former RegExp didn't match, we assume that more than one result was returned anime_ids = set() search_results = [] for match in re.finditer(Settings.regexp_searchlinks, data): if len(search_results) >= Settings.number_of_results: break aid, title = int(match.group(1)), match.group(2) if aid not in anime_ids: anime_ids.add(aid) search_results.append(AnidbResult(self.searchstring_, Settings.anidb_shortcut % aid, title)) if len(search_results) > 0: results = search_results Callbacks.displayresult(results, self.context_, self.output_, url) if self.callback_: self.callback_(result) @staticmethod def _build_connection(url): request = urllib2.Request(url) request.add_header("Accept-Encoding", "gzip") request.add_header("Accept-Charset", "utf-8;q=1.0, *;q=0.5") # Add login cookies if supplied if Settings.adbautouser and Settings.adbautopass: request.add_header("Cookie", "adbautouser=%s; adbautopass=%s" % (Settings.adbautouser, Settings.adbautopass)) return request @staticmethod def _decode_gzip_stream(connection): raw_data = connection.read() # Decompress gzip stream stream = cStringIO.StringIO(raw_data) gzipper = gzip.GzipFile(fileobj=stream) return gzipper.read() class AnidbResult: def __init__(self, searchstring, url, title="", epcount=0): self.searchstring_ = searchstring self.url_ = url self.title_ = title self.epcount_ = int(epcount) self.japanesetitle_ = "" @property def searchstring(self): return self.searchstring_ @property def url(self): return self.url_ def gettitle(self): return self.title_ def settitle(self, title): self.title_ = title title = property(gettitle, settitle) def getepcount(self): return self.epcount_ def setepcount(self, epcount): self.epcount_ = int(epcount) epcount = property(getepcount, setepcount) def getjapanesetitle(self): return self.japanesetitle_ def setjapanesetitle(self, title): self.japanesetitle_ = title japanesetitle = property(getjapanesetitle, setjapanesetitle) Hooks.initialize() print __module_name__, __module_version_

__module_name__ = "AniDBSearch" __module_version__ = "2.2" __module_description__ = "Searches AniDB for a given string and displays the URL. Triggered by .anidb" __module_author__ = "2points <2points at gmx dot org>" __module_url__ = "http://sector-5.net/archives/x-chat-plugin-anidb-search/" import xchat import urllib, urllib2, threading, time import cStringIO, gzip import re, optparse, webbrowser, sys class Settings: """ Settings used for this script If you want it to properly display content not usually accessible to unregistered users (e.g. Hentai), you'll need to supply the cookie values of your AniDB login cookie below. For everything else, these values are OPTIONAL. """ # Cookies for AniDB account adbautouser = "" adbautopass = "" # Channels where the script won't trigger. excluded_channels = [] # If you do not want to use the trigger/command, just use empty strings, # eg. trigger = "" and command = "" effectively disables this plugin. trigger = ".anidb" command = "anidb" # Number of results to fetch/post when the title doesn't match exactly one anime. # Values larger than 3 aren't recommended unless you want to annoy # everyone around with huge amounts of text spam. number_of_results = 3 # Number of seconds to wait between printing new lines (for flood protection). message_delay = 3 # WARNING: Do not modify these strings unless you know what you're doing. I mean it. # They're meant for easy fixing when AniDB changes some little HTML around again. anidb_searchurl = "http://anidb.net/perl-bin/animedb.pl?show=animelist&adb.search=%s&do.search=search" anidb_shortcut = "http://anidb.net/a%i" regexp_searchlinks = r'\s+([^<]+)' regexp_titleid = r'(.+?)\s+\(' regexp_japanese_title = r'ja\s+\s+([^<]+)' regexp_epcount = r'[^,]+, (\d+) episodes' class Messages: # Same here. invalid_trigger = "\002Usage\017: %s - or go directly to \037http://anidb.info\017" % Settings.trigger invalid_command = "\002Usage\017:\t/%s [-pb] <title> - or go directly to \037http://anidb.info\017" % Settings.command connection_error = "\002Error\017:\tUnable to retrieve AniDB info [Error: %s]" prefix = "\002AniDB\017: " noresults = "No results found for '%s'!" url = "\037%s\017" class Helper: _last_message = 0 @staticmethod def clear_nick(nickname): """ Almost identical to xchat.strip for older X-Chat/Python plugin versions. """ if hasattr(xchat, 'strip'): return xchat.strip(nickname) else: if nickname.startswith("\x03"): return nickname[3:] else: return nickname @staticmethod def say(text, context=None): """ Delays a SAY command and makes sure it is actually run in the X-Chat main thread. """ def _say_timer(userdata): if time.time() - Helper._last_message < Settings.message_delay: return 1 else: context, text = userdata context.command("SAY " + text) Helper._last_message = time.time() return 0 if not context: context = xchat.get_context() xchat.hook_timer(100, _say_timer, (context, text)) class Hooks: @staticmethod def initialize(): """ Adds the required X-Chat hooks for various events. """ if Settings.trigger: xchat.hook_print("Channel Message", Hooks.channeltext, "Channel Message") xchat.hook_print("Your Message", Hooks.channeltext, "Your Message") if Settings.command: xchat.hook_command(Settings.command, Hooks.command, help="/%s searchstring" % Settings.command) @staticmethod def channeltext(word, word_eol, userdata): if xchat.get_info("channel") in Settings.excluded_channels or \ not len(word) >= 2 or \ not word[1].startswith(Settings.trigger): return xchat.EAT_NONE # Clean arguments user, message = Helper.clear_nick(word[0]), word[1] words = [s.strip() for s in message.split(" ") if s.strip()] # No argument was passed if len(words) < 2: if userdata == "Channel Message": xchat.command("NOTICE %s %s" % (user, Messages.invalid_trigger)) else: print Messages.invalid_trigger return xchat.EAT_PLUGIN searchstring = " ".join(words[1:]) HttpThread(searchstring, xchat.get_context()).start() return xchat.EAT_PLUGIN @staticmethod def command(word, word_eol, userdata): if len(word) < 2 or len(word_eol) < 2 or not word_eol[1].strip(): print Messages.invalid_command return xchat.EAT_ALL # Using /anidb -p <search> will SAY the result in current channel parser = optparse.OptionParser(usage=optparse.SUPPRESS_USAGE, add_help_option=False, version=False) # Suppress default behaviour of exiting on error def error(msg): print parser.get_usage() parser.error = error parser.add_option("-p", "--public", action="store_true", dest="public", default=False) parser.add_option("-b", "--browser", action="store_true", dest="browser", default=False) options, args = parser.parse_args(args=word[1:]) # Evaluate switches if options.public: output = "public" else: output = "private" if options.browser: callback = Callbacks.openurl else: callback = None if len(args) < 1: print Messages.invalid_command else: HttpThread(" ".join(args), xchat.get_context(), output=output, callback=callback).start() return xchat.EAT_ALL class Callbacks: @staticmethod def connectionerror(error, context, output="private"): if output == "public": Helper.say(Messages.connection_error % error, context) else: print Messages.connection_error % error @staticmethod def noresultserror(searchstring, context, output="private"): message = Messages.prefix + (Messages.noresults % searchstring) if output == "public": Helper.say(message, context) else: print message @staticmethod def displayresult(results, context, output, search_url=None): for result in results: msg_parts = [] # Main title if result.title: msg_parts.append(result.title) # Official japanese title if result.japanesetitle: msg_parts.append(result.japanesetitle) # Episode count if result.epcount: if result.epcount == 1: eptext = "episode" else: eptext = "episodes" msg_parts.append("%i %s" % (result.epcount, eptext)) # URL msg_parts.append(Messages.url % result.url) message = Messages.prefix + " - ".join(msg_parts) if output == "public": Helper.say(message, context) else: print message if len(results) > 1 and search_url: message = Messages.prefix + "Remaining results at " + Messages.url % search_url if output == "public": Helper.say(message, context) else: print message @staticmethod def openurl(result): if sys.version >= "2.5": method = 2 else: method = 0 webbrowser.open(result.url, method) class HttpThread(threading.Thread): def __init__(self, searchstring, context, output="public", callback=None): threading.Thread.__init__(self) self.searchstring_ = searchstring self.output_ = output self.callback_ = callback self.context_ = context def start(self): """ This doesn't really start a thread on win32. Why? Because I can't figure out how to use threads in X-Chat without it crashing. """ if sys.platform == 'win32': self.run() else: threading.Thread.start(self) def run(self): url = Settings.anidb_searchurl % urllib.quote_plus(self.searchstring_) try: request = HttpThread._build_connection(url) # Open stream opener = urllib2.build_opener(urllib2.HTTPHandler) connection = opener.open(request) data = HttpThread._decode_gzip_stream(connection) connection.close() except Exception, e: Callbacks.connectionerror(e, self.context_) return if "No results." in data: Callbacks.noresultserror(self.searchstring_, self.context_, self.output_) return # Populate result by using regexps results = [AnidbResult(self.searchstring_, url)] m = re.search(Settings.regexp_titleid, data) if m: # Found single match for search query result = results[0] result.title = m.group(1) result.url_ = m.group(2) m = re.search(Settings.regexp_epcount, data) if m: result.epcount = int(m.group(1)) m = re.search(Settings.regexp_japanese_title, data) if m: result.japanesetitle = m.group(1) else: # Since the former RegExp didn't match, we assume that more than one result was returned anime_ids = set() search_results = [] for match in re.finditer(Settings.regexp_searchlinks, data): if len(search_results) >= Settings.number_of_results: break aid, title = int(match.group(1)), match.group(2) if aid not in anime_ids: anime_ids.add(aid) search_results.append(AnidbResult(self.searchstring_, Settings.anidb_shortcut % aid, title)) if len(search_results) > 0: results = search_results Callbacks.displayresult(results, self.context_, self.output_, url) if self.callback_: self.callback_(result) @staticmethod def _build_connection(url): request = urllib2.Request(url) request.add_header("Accept-Encoding", "gzip") request.add_header("Accept-Charset", "utf-8;q=1.0, *;q=0.5") # Add login cookies if supplied if Settings.adbautouser and Settings.adbautopass: request.add_header("Cookie", "adbautouser=%s; adbautopass=%s" % (Settings.adbautouser, Settings.adbautopass)) return request @staticmethod def _decode_gzip_stream(connection): raw_data = connection.read() # Decompress gzip stream stream = cStringIO.StringIO(raw_data) gzipper = gzip.GzipFile(fileobj=stream) return gzipper.read() class AnidbResult: def __init__(self, searchstring, url, title="", epcount=0): self.searchstring_ = searchstring self.url_ = url self.title_ = title self.epcount_ = int(epcount) self.japanesetitle_ = "" @property def searchstring(self): return self.searchstring_ @property def url(self): return self.url_ def gettitle(self): return self.title_ def settitle(self, title): self.title_ = title title = property(gettitle, settitle) def getepcount(self): return self.epcount_ def setepcount(self, epcount): self.epcount_ = int(epcount) epcount = property(getepcount, setepcount) def getjapanesetitle(self): return self.japanesetitle_ def setjapanesetitle(self, title): self.japanesetitle_ = title japanesetitle = property(getjapanesetitle, setjapanesetitle) Hooks.initialize() print __module_name__, __module_version__, "loaded."