#!/usr/bin/env python
#
# Copyright 2012 Ajay Narayan, Madhusudan C.S., Shobhit N.S.
#
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import cPickle
import os

from disco.core import Job
from disco.core import Params
from disco.func import chain_reader


import datasettings

from vectorizer import Vectorizer


def mapper(page, params):
    """The mapper function for the MapReduce which analyzes tweets.

    We do a word N-gram analysis after stripping accents, decoding to
    UTF-8 and then removing stop words.

    Args:
        page: The number of the twitter search page for the query that
           this mapper has to fetch and analyze.
        params: The parameter object that holds the map reduce
            initialization parameters. This contains the query string!
    """
    import json
    import requests

    trained_vectorizer = params.trained_vectorizer
    try:
        r = requests.get(
            'http://search.twitter.com/search.json',
            params={'q': params.query, 'rpp': 100, 'page': page})
        page_of_tweets = json.loads(
            r.text.decode(trained_vectorizer.charset,
                          trained_vectorizer.charset_error))
        tweets = page_of_tweets.get('results')
        if not tweets:
          return

        for tweet in tweets:
            tweet_text = tweet['text']
            tweet_id = tweet['id']
            analyze = trained_vectorizer.build_analyzer()
            tokens = analyze(tweet_text)
            for token in tokens:
                yield  (u" ".join(token), (tweet_id, 1))

    except requests.ConnectionError:
        pass


def reducer(klv, params):
    """Reducer of the MapReduce framework that builds the Counts matrix.

    This returns just the counts for each token, tweet id pair. We build
    the actual matrix after we get the results from the reducer.
    
    Args:
        klv: The key and the list of values given as input to the reducer.
        params: The parameters supplied while initializing the MapReduce.
    """
    from disco.util import kvgroup

    trained_vectorizer = params.trained_vectorizer

    for token, count_tuple_list in kvgroup(sorted(klv)):
        j = trained_vectorizer.vocabulary_.get(token)
        if j is not None:
            new_dict = {}
            for doc_id, count in count_tuple_list:
                if doc_id not in new_dict:
                    new_dict[doc_id] = count
                else:
                    new_dict[doc_id] += count
            yield j, new_dict


def bootstrap():
    """Bootstraps the MapReducer.
    """
    path = os.path.join(datasettings.DATA_DIRECTORY, 'buckets.txt')
    vectorizer_file = open(os.path.join(datasettings.DATA_DIRECTORY,
                                        'vectorizer.pickle'))

    trained_vectorizer = cPickle.load(vectorizer_file)
    job = Job()
    job.run(input=['tag://data:buckets'], map=mapper, reduce=reducer,
            map_reader=chain_reader, params=Params(query='Apple',
            trained_vectorizer=trained_vectorizer),
            required_modules=[('vectorizer',
                               os.path.join(datasettings.PROJECT_ROOT,
                                            'analyzer', 'vectorizer.py'),)])

    return trained_vectorizer.build_feature_matrix(job)


if __name__ == '__main__':
    bootstrap()

@@ -656,103 +656,8 @@
 os%0A%0A
-from disco.core import Job%0Afrom disco.core import Params%0Afrom disco.func import chain_reader%0A%0A%0A
 impo
@@ -1275,16 +1275,17 @@
 torizer%0A
+%0A
     try:
@@ -1629,16 +1629,17 @@
 sults')%0A
+%0A
         
@@ -1645,32 +1645,188 @@
  if not tweets:%0A
+          print ('No tweets were fetched in this mapper with page '%0A              'number %25s.' %25 (page))%0A          print 'HTTP status was: ', r.status_code%0A
           return
@@ -2088,26 +2088,14 @@
 eld 
- (u%22 %22.join
 (token
-)
 , (t
@@ -2158,11 +2158,44 @@
    p
-ass
+rint %22There was a connection error.%22
 %0A%0A%0Ad
@@ -2748,16 +2748,46 @@
 (klv)):%0A
+        token = token.strip()%0A
         
@@ -2825,21 +2825,38 @@
 ry_.get(
-token
+''.join(token.split())
 )%0A      
@@ -3142,819 +3142,4 @@
 ict%0A
-%0A%0Adef bootstrap():%0A    %22%22%22Bootstraps the MapReducer.%0A    %22%22%22%0A    path = os.path.join(datasettings.DATA_DIRECTORY, 'buckets.txt')%0A    vectorizer_file = open(os.path.join(datasettings.DATA_DIRECTORY,%0A                                        'vectorizer.pickle'))%0A%0A    trained_vectorizer = cPickle.load(vectorizer_file)%0A    job = Job()%0A    job.run(input=%5B'tag://data:buckets'%5D, map=mapper, reduce=reducer,%0A            map_reader=chain_reader, params=Params(query='Apple',%0A            trained_vectorizer=trained_vectorizer),%0A            required_modules=%5B('vectorizer',%0A                               os.path.join(datasettings.PROJECT_ROOT,%0A                                            'analyzer', 'vectorizer.py'),)%5D)%0A%0A    return trained_vectorizer.build_feature_matrix(job)%0A%0A%0Aif __name__ == '__main__':%0A    bootstrap()%0A

Remove all the code from mapreduce that have been moved to classification module and print some useful messages.