Das deutsche Python-Forum

Code: Alles auswählen

index = {
    wort-1 : {
        dateiindex-1 : [
            { "pos" : pos-1, "typ", fundtypID-1 }, ... , { "pos" : pos-n, "typ", fundtypID-n }
        ],
        ...,
        dateiindex-n : [...]
    },
    ...,
    wort-n : {...}
}

Code: Alles auswählen

#!/usr/bin/env python
import sys, shelve

_IDEM = ''.join(map(chr, xrange(256)))
_BAD_CHARS = ''.join(filter(lambda c: not (c.isalnum() or c.isspace()), _IDEM))


def clean_lines(lines, bad_characters):
    for line in lines:
        result = line.translate(_IDEM, bad_characters).strip()
        if result:
            yield result


def iter_words(lines):
    for line in lines:
        for word in line.split():
            yield word.lower()


class Index:
    def __init__(self, name):
        self.name = name
        self.document_ids = shelve.open(self.name + '_doc_id_index')
        self.index = shelve.open(self.name + '_words_index')
    
    def update(self, document_index):
        doc_id = str(len(self.document_ids) + 1)
        self.document_ids[doc_id] = document_index.name
        
        for word, count in document_index:
            if word in self.index:
                index_entry = self.index[word]
                index_entry[doc_id] = count
            else:
                index_entry = {doc_id: count}
            self.index[word] = index_entry

    def sync(self):
        self.document_ids.sync()
        self.index.sync()
    
    def close(self):
        self.document_ids.close()
        self.index.close()

    def search(self, words):
        if isinstance(words, basestring):
            words = [words]
        words = list(iter_words(clean_lines(words, _BAD_CHARS)))
        result = dict()
        for word in words:
            try:
                for doc_id, count in self.index[word].iteritems():
                    result[doc_id] = result.get(doc_id, 0) + count
            except KeyError:
                pass
        return [(count, self.document_ids[doc_id]) for doc_id, count
                in result.iteritems()]


class DocumentIndex:
    def __init__(self, name, words=tuple()):
        self.name = name
        self.words = dict()
        self.update(words)
    
    def __len__(self):
        return len(self.words)
    
    def __iter__(self):
        return self.words.iteritems()
    
    def update(self, words):
        for word in words:
            self.words[word] = self.words.get(word, 0) + 1


def update(filenames):
    filenames = sys.argv[1:]
    index = Index('rfc')
    
    for filename in filenames:
        print 'indexing %r' % filename
        
        input_file = open(filename, 'r')
        words = iter_words(clean_lines(input_file, _BAD_CHARS))
        document_index = DocumentIndex(filename, words)
        input_file.close()
        
        index.update(document_index)
    
    index.close()


def search(words):
    index = Index('rfc')
    result = index.search(words)
    result.sort()
    print ''.join(['%5d %s\n' % (count, filename)
                   for count, filename in result])
    index.close()


def info(dummy):
    # 280663 words in 3663 files.
    index = Index('rfc')
    print len(index.document_ids)
    index.close()


def main():
    command_func = {'-u': update, '-s': search, '-i': info }
    command_func[sys.argv[1]](sys.argv[2:])


if __name__ == '__main__':
    main()

Code: Alles auswählen

In [122]: list(indexer.clean_lines(['def spam(arg):'], indexer._BAD_CHARS))
Out[122]: ['def spamarg']

Code: Alles auswählen

def make_text_filter(replace_characters):
    """Creates a function that takes a text and replaces all characters in
    `replace_characters` with a space, deletes all characters from the text
    that are not alphanumeric or whitespace and finally strips leading and
    trailing whitespace.
    
    :returns: function(str) -> str
    """
    translate_chars = map(chr, xrange(256))
    for replace_character in replace_characters:
        translate_chars[ord(replace_character)] = ' '
    translate_chars = ''.join(translate_chars)
    delete_chars = ''.join(filter(lambda c: not (c.isalnum() or c.isspace()),
                                  translate_chars))
    return lambda text: text.translate(translate_chars, delete_chars).strip()


text_filter = make_text_filter('+-*/()[]{}.,;:<>|=')

Code: Alles auswählen

In [22]: indexer.text_filter('def spam(args): print args.map[key]')
Out[22]: 'def spam args   print args map key'

Code: Alles auswählen

__version__="0.0.1"

import sys, re, sgmllib


class MetaContentParser(sgmllib.SGMLParser):

    """
    Extrahiert den Text aus HTML-Seiten.
    Unterscheidet die folgende Daten:
        - title
        - meta-Tags-Informationen
        - Überschriften
        - nomaler Text
    """

    entitydefs = {
        "nbsp" : " ",
        "Auml" : "Ä",
        "auml" : "ä",
        "Ouml" : "Ö",
        "ouml" : "ö",
        "Uuml" : "Ü",
        "uuml" : "ü"
        }

    def __init__(self):
        # MetaTags die berücksichtigt werden sollen.
        self.metaTags = ["keywords","description","author"]

        sgmllib.SGMLParser.__init__(self, 0)

        self.data = {
            "title"     : " ",
            "meta"      : " ",
            "heads"     : " ",
            "content"   : " "
        }
        self.typ = "content"

    ###
    ### Default Funktionen
    ###

    def start_title( self, dummy ):
        self.typ = "title"

    def end_title( self ):
        self.typ = "content"

    def start_meta( self, metaData ):
        try:
            typ = metaData[0][1]
            content = metaData[1][1]
        except:
            return

        if not typ.lower() in self.metaTags:
            return

        content = " ".join( content.split(",") )
        self.data["meta"] += " " + content

    def handle_data( self, data ):
        self.data[self.typ] += data

    def unknown_starttag(self, tag, attrs):
        if tag[0] == "h":
            # Überschriften
            self.typ = "heads"
        else:
            self.typ = "content"

    def cleancontent( self ):
        "Leerzeichen verwerfen, nur Buchstaben zulassen"
        for typ in self.data:
            self.data[typ] = " ".join(
                    re.findall( r"[\wäöüßÄÖÜ]+",self.data[typ] )
                )

    ###
    ### Abfragen der Daten
    ###

    def get_data( self ):
        self.cleancontent()
        return self.data

    def dump( self ):
        for k,v in self.get_data().iteritems():
            print ">",k
            print v
            print "-"*80

Code: Alles auswählen

import urllib2

from ContentParser import MetaContentParser


TestURL = "http://www.heise.de"

print "rufe [%s] ab..." % TestURL,
c = urllib2.urlopen(TestURL)
HTMLside = c.read()
c.close()
print "OK"


print "Parse Daten...",
MyParser = MetaContentParser()
MyParser.feed( HTMLside )
print "OK"

print "-"*80
MyParser.dump()

Code: Alles auswählen

class compress:
    def __init__( self, indexfile, compresslevel = 9 ):
        self.index = shelve.open( indexfile, "r" )
        self.compresslevel = compresslevel

    def compress( self, outfile ):
        outfile = anydbm.open( outfile, "n" )

        for word in self.index.keys():
            data = pickle.dumps( self.index[word] )
            outfile[word] = bz2.compress( data, self.compresslevel )


compress( "wordindex.bin" ).compress1( "wordindexcompressed.bin" )

Code: Alles auswählen

class compaccess:
    def __init__( self, compressedIndexFile ):
        self.index = anydbm.open( compressedIndexFile, "r" )

    def get( self, word ):
        compresseddata = self.index[word]
        data = bz2.decompress( compresseddata )
        return pickle.loads( data )


MyIndex = compaccess( "wordindexcompressed.bin" )
print MyIndex.get("python")