hatenab.py

# -*- coding: utf-8 -*-
# Based on http://d.hatena.ne.jp/kenkitii/20060429/p1
import re
import datetime, time

# IronPython
import clr

from System import *
from System.Text import *
from System.Net import *
from System.Security.Cryptography import *
from System.Collections.Generic import *
from System.Diagnostics import Trace

channelName = "#HatenaB"

# Crigate or TwitterIrcGateway
isCrigate = False
try:
    import Misuzilla.Crigate
    import Misuzilla.Crigate.Scripting
    from Misuzilla.Crigate import ConsoleHandler, Console, Context
    from Misuzilla.Crigate.Configuration import IConfiguration
    from Misuzilla.Crigate.Scripting import DLRIntegrationHandler, DLRBasicConfiguration, DLRContextHelper
    isCrigate = True
except:
    import Misuzilla.Applications.TwitterIrcGateway
    import Misuzilla.Applications.TwitterIrcGateway.AddIns
    import Misuzilla.Applications.TwitterIrcGateway.AddIns.Console
    from Misuzilla.Applications.TwitterIrcGateway.AddIns import IConfiguration
    from Misuzilla.Applications.TwitterIrcGateway.AddIns.Console import ConsoleAddIn, Console, Context
    from Misuzilla.Applications.TwitterIrcGateway.AddIns.DLRIntegration import DLRIntegrationAddIn, DLRBasicConfiguration, DLRContextHelper

class AtomClient:
    def __init__(self):
        self.endopoint = None
        self.wsse = None
        self._sha1 = SHA1.Create()
        self.encoding = UTF8Encoding(False)

    def base64(self, digest):
        return Convert.ToBase64String(digest)

    def sha1(self, value):
        if value.__class__ == Array[Byte]:
            return self._sha1.ComputeHash(value)
        else:
            return self._sha1.ComputeHash(self.getBytes(value))

    def getBytes(self, value):
        return self.encoding.GetBytes(value)
        
    def escapeHtml(self, value):
        return re.sub('"', "&quot;", re.sub(">", "&gt;", re.sub("<", "&lt;", re.sub("&", "&amp;", value))))

    def credentials(self, endpoint, user, password):
        nonce = self.sha1(str(time.time() + Random().Next()))
        now = datetime.datetime.now().isoformat() + "Z"
        digest = self.sha1(nonce + self.getBytes(str(now)) + self.getBytes(str(password)))

        wsse = 'UsernameToken Username="%(u)s", PasswordDigest="%(p)s", Nonce="%(n)s", Created="%(c)s"'
        value = dict(u = user, p = self.base64(digest),
                     n = self.base64(nonce), c = now)

        self.endpoint = endpoint
        self.wsse = wsse % value

    def atomRequest(self, method, URI, body):
        ServicePointManager.Expect100Continue = False
        webClient = WebClient()
        webClient.Headers['X-WSSE']       = self.wsse
        webClient.Headers['Content-Type'] = "text/xml"
        webClient.Headers['User-Agent']   = "Crigate/1.0"
        webClient.Encoding = self.encoding

        # FIXME: TENUKI
        return webClient.UploadString(("http://%s%s" % (self.endpoint, URI)), method, body)

class HatenaBClient(AtomClient):
    def bookmark(self, url, comment = ""):
        return self.atomRequest("POST", "/atom/post", """<entry xmlns="http://purl.org/atom/ns#">
          <title>dummy</title>
          <link rel="related" type="text/html" href="%s" />
          <summary type="text/plain">%s</summary>
        </entry>
        """ % (self.escapeHtml(url), self.escapeHtml(comment)))

class HatenaBContext(Context):
    def Initialize(self):
        self.url_re = re.compile(r"^https?://")
        self.config = DLRBasicConfiguration(self.CurrentSession, "HatenaBContext", Dictionary[String,String]({ "Username": "ユーザ名", "Password":"パスワード" }))
        pass

    def GetCommands(self):
        dict = Context.GetCommands(self)
        dict["Bookmark"] = "Bookmark"
        return dict

    def OnUninitialize(self):
        pass

    def get_Configurations(self):
        return Array[IConfiguration]([ self.config ])

    def OnCallMissingCommand(self, commandName, rawInputLine):
        if self.url_re.match(rawInputLine):
            self.Bookmark(rawInputLine)
            return True
        else:
            return False

    # Implementation
    def Bookmark(self, args):
        self.hatebClient = HatenaBClient()
        self.hatebClient.credentials("b.hatena.ne.jp", self.config.GetValue("Username"), self.config.GetValue("Password"))
        args = args.split(" ", 1)
        try:
            res = self.hatebClient.bookmark(args[0], (args[1] if len(args) > 1 else ""))
            title = re.compile("<title>([^<]*)</title>").findall(res)[0]
            self.Console.NotifyMessage("%s をブックマークしました。" % title)
        except Exception, e:
            self.Console.NotifyMessage("エラー: " + e.Message)

# コンソールチャンネルを追加する
console = None
if isCrigate:
    console = Misuzilla.Crigate.Console()
    console.Initialize(CurrentSession)
    console.Attach(channelName, DLRContextHelper.Wrap(CurrentSession, "HatenaB", HatenaBContext))
    CurrentSession.HandlerLoader.GetHandler[DLRIntegrationHandler]().BeforeUnload += lambda sender, e: console.Detach()
else:
    console = Misuzilla.Applications.TwitterIrcGateway.AddIns.Console.Console()
    console.Attach(channelName, CurrentServer, CurrentSession, DLRContextHelper.Wrap(CurrentSession, "HatenaB", HatenaBContext))
    CurrentSession.AddInManager.GetAddIn[DLRIntegrationAddIn]().BeforeUnload += lambda sender, e: console.Detach()