natural language toolkit twitter corpus reader c 20012023 nltk project ewan klein ewaninf ed ac uk url https www nltk org for license information see license txt a reader for corpora that consist of tweets it is assumed that the tweets have been serialised into linedelimited json corpusview streambackedcorpusview def init self root fileidsnone wordtokenizertweettokenizer encodingutf8 corpusreader initself root fileids encoding for path in self abspathsself fileids if isinstancepath zipfilepathpointer pass elif os path getsizepath 0 raise valueerrorffile path is empty returns the full tweet objects as specified by twitter documentation on tweets https dev twitter comdocsplatformobjectstweets return the given files as a list of dictionaries deserialised from json rtype listdict returns only the text content of tweets in the files return the given files as a list of tweets rtype liststr return the given files as a list of the text content of tweets as as a list of words screenanames hashtags urls and punctuation symbols rtype listliststr assumes that each line in stream is a jsonserialised object natural language toolkit twitter corpus reader c 2001 2023 nltk project ewan klein ewan inf ed ac uk url https www nltk org for license information see license txt a reader for corpora that consist of tweets it is assumed that the tweets have been serialised into line delimited json reader for corpora that consist of tweets represented as a list of line delimited json individual tweets can be tokenized using the default tokenizer or by a custom tokenizer specified as a parameter to the constructor construct a new tweet corpus reader for a set of documents located at the given root directory if you made your own tweet collection in a directory called twitter files then you can initialise the reader as from nltk corpus import twittercorpusreader reader twittercorpusreader root path to twitter files json however the recommended approach is to set the relevant directory as the value of the environmental variable twitter and then invoke the reader as follows root os environ twitter reader twittercorpusreader root json if you want to work directly with the raw tweets the json library can be used import json for tweet in reader docs print json dumps tweet indent 1 sort_keys true the corpus view class used by this reader param root the root directory for this corpus param fileids a list or regexp specifying the fileids in this corpus param word_tokenizer tokenizer for breaking the text of tweets into smaller units including but not limited to words check that all user created corpus files are non empty returns the full tweet objects as specified by twitter documentation on tweets https dev twitter com docs platform objects tweets _ return the given file s as a list of dictionaries deserialised from json rtype list dict returns only the text content of tweets in the file s return the given file s as a list of tweets rtype list str return the given file s as a list of the text content of tweets as as a list of words screenanames hashtags urls and punctuation symbols rtype list list str assumes that each line in stream is a json serialised object
import json
import os

from nltk.corpus.reader.api import CorpusReader
from nltk.corpus.reader.util import StreamBackedCorpusView, ZipFilePathPointer, concat
from nltk.tokenize import TweetTokenizer


class TwitterCorpusReader(CorpusReader):
    r

    CorpusView = StreamBackedCorpusView
    

    def __init__(
        self, root, fileids=None, word_tokenizer=TweetTokenizer(), encoding="utf8"
    ):
        
        CorpusReader.__init__(self, root, fileids, encoding)

        for path in self.abspaths(self._fileids):
            if isinstance(path, ZipFilePathPointer):
                pass
            elif os.path.getsize(path) == 0:
                raise ValueError(f"File {path} is empty")
        

        self._word_tokenizer = word_tokenizer

    def docs(self, fileids=None):
        
        return concat(
            [
                self.CorpusView(path, self._read_tweets, encoding=enc)
                for (path, enc, fileid) in self.abspaths(fileids, True, True)
            ]
        )

    def strings(self, fileids=None):
        
        fulltweets = self.docs(fileids)
        tweets = []
        for jsono in fulltweets:
            try:
                text = jsono["text"]
                if isinstance(text, bytes):
                    text = text.decode(self.encoding)
                tweets.append(text)
            except KeyError:
                pass
        return tweets

    def tokenized(self, fileids=None):
        
        tweets = self.strings(fileids)
        tokenizer = self._word_tokenizer
        return [tokenizer.tokenize(t) for t in tweets]

    def _read_tweets(self, stream):
        
        tweets = []
        for i in range(10):
            line = stream.readline()
            if not line:
                return tweets
            tweet = json.loads(line)
            tweets.append(tweet)
        return tweets