fuse-python で遊んだ


RSS 内のアイテムを1つづつローカルの HTML ファイルにするだけのスクリプト。これからもうちょっと便利になるように改良してみたいところ。あまりサンプルがないようなので貼付けておきます。git-hub とかは僕には難しすぎてよくわかりません。
例によって日本語の扱いですこしはまったのが残念。成長しねぇ。

#!/usr/bin/env python

import os, stat, errno, sys
import urllib2
from xml.dom import minidom
import types

try:
    import _find_fuse_parts
except ImportError:
    pass
import fuse
from fuse import Fuse


if not hasattr(fuse, '__version__'):
    raise RuntimeError, \
        "your fuse-py doesn't know of fuse.__version__, probably it's too old."

fuse.fuse_python_api = (0, 2)


class MyStat(fuse.Stat):
    def __init__(self):
        self.st_mode = 0
        self.st_ino = 0
        self.st_dev = 0
        self.st_nlink = 0
        self.st_uid = 0
        self.st_gid = 0
        self.st_size = 0
        self.st_atime = 0
        self.st_mtime = 0
        self.st_ctime = 0

def tagText(node, tagName):
    return node.getElementsByTagName(tagName)[0].firstChild.nodeValue

class RSSReader:
    def __init__(self, target_url):
        req = urllib2.Request(target_url)
        try:
            e = minidom.parse(file=urllib2.urlopen(req))
        except Exception, ex:
            sys.stderr.write(str(ex)) + "\n"
        if type(e) != types.NoneType:
            self.rss = e
            self.items = self.rss.getElementsByTagName('item')
            self.titles = self.rss.getElementsByTagName('title')
            self.dirs = []
            self.descrs = []
            self.contents = []

    def get_items(self): # items :: DOM
        return self.items
    def get_titles(self): # titles :: DOM
        return self.titles

    def get_dirs(self):
        if self.dirs == []:
            def f(i):
                return '/'+tagText(i,'title').encode('utf-8')
            self.dirs = map(f, self.items)
        return self.dirs
    def get_descs(self):
        if self.descrs == []:
            def f(i):
                return tagText(i,'description').encode('utf-8')
            self.descrs = map(f, self.items)
        return self.descrs
    def get_contents(self):
        if self.contents == []:
            def f(i):
                pre = len("![CDATA[")
                post = len("]]")-1
                ctt = tagText(i, 'content:encoded')[pre:-post]
                return ("<html><body>%s</body></html>" % ctt).encode('utf-8')
            self.contents = map(f, self.items)
        return self.contents


class RSSFS(Fuse):
    def __init__(self, url, *args, **kw):
        Fuse.__init__(self, *args, **kw)
        self.root = '/'
        self.rss = RSSReader(url)
        self.rss.get_contents()

    def getattr(self, path):
        st = MyStat()
        if path == '/':
            st.st_mode = stat.S_IFDIR | 0755
            st.st_nlink = 2 + len(self.rss.get_dirs())
        elif path in self.rss.get_dirs():
            idx = self.rss.get_dirs().index(path)
            st.st_mode = stat.S_IFREG | 0444
            st.st_nlink = 1
            st.st_size = len(self.rss.get_contents()[idx])
        else:
            return -errno.ENOENT
        return st

    def readdir(self, path, offset):
        dirs = map(lambda name: name[1:], self.rss.get_dirs())
        for r in  ['.', '..'] + dirs:
            yield fuse.Direntry(r)

    def open(self, path, flags):
        if path in self.rss.get_dirs():
            accmode = os.O_RDONLY | os.O_WRONLY | os.O_RDWR
            if (flags & accmode) != os.O_RDONLY:
                return -errno.EACCES
        else:
            return -errno.ENOENT

    def read(self, path, size, offset):
        if path in self.rss.get_dirs():
            idx = self.rss.get_dirs().index(path)
            contents = self.rss.get_contents()[idx]
            slen = len(contents)
            if offset < slen:
                if offset + size > slen:
                    size = slen - offset
                buf = contents[offset:offset+size]
            else:
                buf = ''
            return buf
        else:
            return -errno.ENOENT

def main(url):
    usage = Fuse.fusage
    server = RSSFS(url,
                   version="%prog " + fuse.__version__,
                   usage=usage,
                   dash_s_do='setsingle')

    server.parse(errex=1)
    server.main()

if __name__ == '__main__':
    main("http://d.hatena.ne.jp/hatenacinnamon/rss")

最近全然 Haskell してないのが心残り。しなもんかわいすぎる。