Key-Value Map-Reduce SEO

from unittest import TestCase, main
from thrift import Thrift
from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TBinaryProtocol

from hbase import Hbase
from hbase.ttypes import ColumnDescriptor, Mutation, BatchMutation, NotFound
class HbaseWriter:

    def __init__(self, netloc, port, table="webpages"):
        self.tableName = table

        self.transport = TTransport.TBufferedTransport(
            TSocket.TSocket(netloc, port))
        self.protocol = TBinaryProtocol.TBinaryProtocol(self.transport)
        self.client = Hbase.Client(self.protocol)
        self.transport.open()

        tables = self.client.getTableNames()
        if self.tableName not in tables:
            self.__createTable()

    def __del__(self):
        self.transport.close()

    def __createTable(self):
        self.client.createTable(self.tableName,
                                [ColumnDescriptor(name="contents:", maxVersions=1, compression="BLOCK"),])

    def reverseUrl(self, url):
        link = filter(None, url.split("//"))[-1]
        hops = filter(None, link.split("/"))
        domain = hops[0].split(".")
        domain.reverse()
        domain = '.'.join(domain)
        hops[0] = domain
        return '/'.join(hops)

    def write(self, url, content):
        row = self.reverseUrl(url)
        mutations = [Mutation(column="contents:", value=content)]
        self.client.mutateRow(self.tableName, row, mutations)


class TestHbaseWriter(TestCase):
    def setUp(self):
        self.writer = HbaseWriter("192.168.0.1", 9090, "test")

    def tearDown(self):
        name = self.writer.tableName
        client = self.writer.client
        client.disableTable(name)
        client.deleteTable(name)

    def testReverseUrl(self):
        self.assertEquals(self.writer.reverseUrl("http://www.a.com"), "com.a.www")
        self.assertEquals(self.writer.reverseUrl("http://www.a.com/"), "com.a.www")
        self.assertEquals(self.writer.reverseUrl("http://a.com"), "com.a")
        self.assertEquals(self.writer.reverseUrl("http://www.b.com/foo"), "com.b.www/foo")
        self.assertEquals(self.writer.reverseUrl("aaa.bbb.ccc.com.cn/foo1/foo2"), "cn.com.ccc.bbb.aaa/foo1/foo2")

    def testCreate(self):
        tableName = self.writer.tableName
        client = self.writer.client
        self.assertTrue(self.writer.tableName in client.getTableNames())
        columns = dict()
        columns["contents"] = ColumnDescriptor(name="contents", maxVersions=1, compression="BLOCK")
        cds = client.getColumnDescriptors(tableName)
        for name,column in cds.items():
            self.assertTrue(column.name in columns)

    def testWrite(self):
        tableName = self.writer.tableName
        client = self.writer.client
        data = {"http://www.a.com":"com.a.www",
                "http://www.a.com/bbb":"com.a.www/bbb",
                "http://www.foo.com/foo":"foo"}
        for url, content in data.items():
            self.writer.write(url, content)

        scannerId = client.scannerOpen(tableName, "", ["contents:",])
        while True :
            try:
                result = client.scannerGet(scannerId)
            except NotFound:
                break
            row = result.row
            contents = result.columns["contents:"].value
            url = "http://" + self.writer.reverseUrl(row)
            self.assertTrue(url in data)
            self.assertEqual(data[url], contents)
        client.scannerClose(scannerId)


if __name__ == "__main__":
    main()