Subversion Repositories SmartDukaan

#!/usr/bin/python

import threading
import time

import MySQLdb
from urlparse import urlparse
from urlparse import parse_qs
import requests
import json
import optparse
import urllib2
import base64
import urllib
import logging
from dtr.utils.utils import get_mongo_connection, to_java_date
from datetime import datetime
import traceback

parser = optparse.OptionParser()
parser.add_option("-C", "--chunksize", dest="chunksize",
                  default="100",
                  type="int", help="The requsets a single thread handles",
                  metavar="CHUNKSIZE")
parser.add_option("-M", "--mongoHost", dest="mongoHost",
                  default="localhost",
                  type="str", help="The requsets a single thread handles",
                  metavar="MONGOHOST")
parser.add_option("-m", "--mysqlHost", dest="mysqlHost",
                  default="127.0.0.1",
                  type="str", help="The requsets a single thread handles",
                  metavar="MONGOHOST")

(options, args) = parser.parse_args()

GCM_URL = "https://android.googleapis.com/gcm/send"
FCM_URL = "https://fcm.googleapis.com/fcm/send"
GOOGLE_API_KEY = "AIzaSyBGL29sWCKm8OweP62ywrfv4jv4VIN8u0o"
FCM_API_KEY = "AAAASAjNcn4:APA91bG6fWRIgYJI0L9gCjP5ynaXz2hJHYKtD9dfH7Depdv31Nd9APJwhx-OPkAJ1WSz4BGNYG8lHThLFSjDGFxIwUZv241YcAJEGDLgt86mxq9FXJe-yBRu-S0_ZwHqmX-QaVKl5F_A"
headers_gcm = {'content-type':'application/json', "authorization":"key=" + GOOGLE_API_KEY}
headers_fcm = {'content-type':'application/json', "authorization":"key=" + FCM_API_KEY}
aff_url_headers = { 
            'User-agent':'Mozilla/5.0 (X11; Linux i686) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/37.0.2062.120 Safari/537.36',
            'Accept' : 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',      
            'Accept-Language' : 'en-US,en;q=0.8',                     
            'Accept-Charset' : 'ISO-8859-1,utf-8;q=0.7,*;q=0.3',
            'Connection':'keep-alive'
        }

pendingNotificationSet = {'_id':1,'user_id':1,'gcm_id':1,'notification_campaign_id':1}
pendingCampaignSet = {'_id':1,'name':1,'title':1,'message':1,'type':1,'url':1,'sql':1,'expiresat':1,'image':1} 

pendingNotificationEntryMap = {}
campaignUsersMap = {}
notificationCampaignsMap = {}
userGcmRegIdMap = {}
gcmIdGcmRegIdMap = {}
domainStoresMap = {}
failedGcm = []


db = MySQLdb.connect(options.mysqlHost,"root","shop2020","dtr" )
cursor = db.cursor()

ALL_STORES_SQL = "select * from stores"
GCM_REG_ID_SQL1 = "select x.user_id, x.gcm_regid, x.id, x.androidid, x.created, x.notification_type from (select gc.* from gcm_users gc join users u on gc.user_id=u.id where gc.user_id in (%s) and gc.androidid is not null and u.activated=1 and gc.failurecount < 2 order by gc.id desc) as x group by x.user_id, x.gcm_regid, x.androidid"
GCM_REG_ID_SQL2 = "select x.user_id, x.gcm_regid, x.id, x.androidid, x.created, x.imeinumber,x.notification_type from (select gc.* from gcm_users gc join users u on u.id=gc.user_id where gc.user_id in (%s) and gc.androidid is null and gc.imeinumber is not null and gc.failurecount < 2 and u.activated=1 order by id desc) as x group by x.user_id, x.imeinumber"
FETCH_GCM_REG_SQL = "select id, gcm_regid, notification_type from gcm_users where id in (%s)"

cursor.execute(ALL_STORES_SQL)
result_stores = cursor.fetchall()
domainStoresMap = {}
for rec in result_stores:
    domainStoresMap[rec[2]] = rec
    
logging.basicConfig(level=logging.DEBUG,
                    format='[%(levelname)s] (%(threadName)-10s) %(message)s',
                    )

class __PushNotification:
    def __init__(self, notification_campaign_id, user_id, message, type, sms_type, sms_id, android_id, pushed_by, sent_timestamp, receive_timestamp, open_timestamp, sms_timestamp, gcm_id, created_timestamp, gcm_expired, notification_accounted):
        self.notification_campaign_id = notification_campaign_id
        self.user_id = user_id
        self.message = message
        self.type = type
        self.sms_type = sms_type
        self.sms_id = sms_id
        self.android_id = android_id
        self.pushed_by = pushed_by
        self.sent_timestamp = sent_timestamp
        self.receive_timestamp = receive_timestamp
        self.open_timestamp = open_timestamp
        self.sms_timestamp = sms_timestamp
        self.gcm_id = gcm_id
        self.created_timestamp = created_timestamp
        self.gcm_expired = gcm_expired
        self.notification_accounted = notification_accounted



def populatePendingNotificationEntriesToBeSent():
    global pendingNotificationEntryMap
    toRefetchGcmId = []
    offset = 0
    old_campaigns = []
    while(True):
        pendingNotificationEntries = list(get_mongo_connection(host=options.mongoHost).User.pushnotifications.find({'type':'pending'}, pendingNotificationSet).skip(offset).limit(10000))
        if not pendingNotificationEntries:
            break
        for entry in pendingNotificationEntries:
            try:
                gcm_regid = gcmIdGcmRegIdMap.get(entry.get('gcm_id'))[0]
            except:
                gcm_regid = None
            if gcm_regid is None:
                toRefetchGcmId.append(entry.get('gcm_id'))
            if not notificationCampaignsMap.has_key(long(entry['notification_campaign_id'])):
                old_campaigns.append(long(entry['notification_campaign_id']))
            if pendingNotificationEntryMap.has_key(entry['notification_campaign_id']):
                entries = pendingNotificationEntryMap.get(entry['notification_campaign_id'])
                entries.append(entry)
                pendingNotificationEntryMap[entry['notification_campaign_id']] = entries
            else:
                entries = []
                entries.append(entry)
                pendingNotificationEntryMap[entry['notification_campaign_id']] = entries
        offset = offset + 10000
    logging.debug("FETCH_GCM_REG_SQL:- "+FETCH_GCM_REG_SQL%(",".join(map(str,toRefetchGcmId))))
    if not toRefetchGcmId:
        print "No gcm to refetch"
        return
    cursor.execute(FETCH_GCM_REG_SQL%(",".join(map(str,toRefetchGcmId))))
    result_data = cursor.fetchall()
    if result_data:
        for data in result_data:
            gcmIdGcmRegIdMap[data[0]] = [data[1],data[2]]
            
    if old_campaigns:
        print "yes, old campaigns exist"
        campaigns = list(get_mongo_connection(host=options.mongoHost).User.notificationcampaigns.find({'_id':{"$in":old_campaigns}},pendingCampaignSet))
        for campaign in campaigns:
            notificationCampaignsMap[long(campaign['_id'])] = campaign
    
    

def populateCampaignsMap(pendingCampaigns):
    global campaignUsersMap
    global notificationCampaignsMap
    for campaign in pendingCampaigns:
        cursor.execute(str(campaign['sql']))
        user_records = cursor.fetchall()
        userids = []
        for record in user_records:
            if str(record[0]) not in userids:
                userids.append(str(record[0]))
        campaignUsersMap[long(campaign['_id'])]=userids
        notificationCampaignsMap[long(campaign['_id'])] = campaign
    
def getPendingCampaigns():
    campaigns = list(get_mongo_connection(host=options.mongoHost).User.notificationcampaigns.find({'notification_processed':0,'expiresat':{'$gte':to_java_date(datetime.now())}},pendingCampaignSet))
    return campaigns

def insertPushNotificationEntriesToSent():
    global userGcmRegIdMap
    global gcmIdGcmRegIdMap
    for campaignId, user_list in campaignUsersMap.items():
        if len(user_list)==0:
            continue
        logging.debug("GCM_REG_SQL_1:- "+GCM_REG_ID_SQL1%(",".join(map(str,user_list))))
        cursor.execute(GCM_REG_ID_SQL1%(",".join(map(str,user_list))))
        result_data = cursor.fetchall()
        
        if result_data and len(result_data)>0:
                
            for dataRec in result_data:
                
                gcmIdGcmRegIdMap[dataRec[2]] = [dataRec[1], dataRec[5]]
                
                if userGcmRegIdMap.has_key(dataRec[0]):
                    gcmRegIdMap = userGcmRegIdMap.get(dataRec[0])
                    gcmRegIdMap[dataRec[2]]= {'gcm_regid':dataRec[1],'android_id':dataRec[3]}
                    userGcmRegIdMap[dataRec[0]] = gcmRegIdMap
                else:
                    userGcmRegIdMap[dataRec[0]] = {dataRec[2]:{'gcm_regid':dataRec[1],'android_id':dataRec[3]}}

        logging.debug("GCM_REG_SQL_2:- "+GCM_REG_ID_SQL2%(",".join(map(str,user_list))))       
        cursor.execute(GCM_REG_ID_SQL2%(",".join(map(str,user_list))))
        result_data = cursor.fetchall()
        for dataRec in result_data:
            
            gcmIdGcmRegIdMap[dataRec[2]] = [dataRec[1], dataRec[6]]
            
            if userGcmRegIdMap.has_key(dataRec[0]):
                gcmRegIdMap = userGcmRegIdMap.get(dataRec[0])
                gcmRegIdMap[dataRec[2]]= {'gcm_regid':dataRec[1],'android_id':dataRec[3]}
                userGcmRegIdMap[dataRec[0]] = gcmRegIdMap
            else:
                userGcmRegIdMap[dataRec[0]] = {dataRec[2]:{'gcm_regid':dataRec[1],'android_id':None}}
                    
        
    for campaignId, userList in campaignUsersMap.items():
        bulkArrayPushNotificationObj = []
        bulkCount = 0
        for userId in userList:
            gcmRegIdMap = userGcmRegIdMap.get(long(userId))
            if gcmRegIdMap is None:
                logging.debug("No value exist in gcm map for user_id"+str(userId))
                continue
            for gcm_id, details in gcmRegIdMap.items():
                android_id = None
                if details.has_key('android_id'):
                    android_id = details['android_id']
                pushNotificationObj = __PushNotification(long(campaignId), long(userId), None, 'pending', \
                                    None, None, android_id, "php", None, None, None, None, gcm_id, to_java_date(datetime.now()),0,0)
                bulkArrayPushNotificationObj.append(pushNotificationObj.__dict__)
                bulkCount = bulkCount + 1
            if bulkCount >= 10000:
                get_mongo_connection(host=options.mongoHost).User.pushnotifications.insert(bulkArrayPushNotificationObj)
                bulkCount = 0
                bulkArrayPushNotificationObj = []
                
        if bulkArrayPushNotificationObj:
            get_mongo_connection(host=options.mongoHost).User.pushnotifications.insert(bulkArrayPushNotificationObj)

def initiateNotificationThreadProcess(chunkSize):
    count = 1
    logging.debug('Starting Push Notification Job....'+str(datetime.now()))
    for entries in pendingNotificationEntryMap.values():
        campaign_receivers_list = list(chunks(entries, chunkSize))
        for sublist in campaign_receivers_list:
            thread = NotificationThread(count, "Thread-"+str(count), sublist)
            thread.start()
            thread.join()
            count = count +1
    logging.debug('Stopping Push Notification Job....'+str(datetime.now()))

class NotificationThread (threading.Thread):
    def __init__(self, threadID, name, recordsList):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.recordsList = recordsList
        
    def run(self):
        logging.debug('Starting')
        handleCampaignRequest(self.name, self.recordsList)
        logging.debug('Completed')

def handleCampaignRequest(threadName, recordsList ):
    global failedGcm
    for record in recordsList:
        try:
            #userGcmRegIdDetails = userGcmRegIdMap.get(long(record.get('user_id')))
            campaign = notificationCampaignsMap.get(long(record.get('notification_campaign_id'))) 
            gcm_id = record.get('gcm_id')
            gcm_regid = gcmIdGcmRegIdMap.get(gcm_id)[0]
            if gcm_regid is None:
                print "Ideally should never raise"
                raise
        except:
            logging.debug('Error while getting GCM Details for User Id:- '+ str(record.get('user_id'))+" and Notification Id:- "+str(record.get('notification_campaign_id')))
            traceback.print_exc() 
            continue
        result_url = ""
        if campaign.get('type')=='url':
            parsed_uri = urlparse(campaign.get('url'))
            domain = '{uri.netloc}'.format(uri=parsed_uri)
            store = domainStoresMap.get(domain)
            if store is not None:
                url_params = { 'url' : campaign.get('url'),  'userId' : record.get('user_id'), 'storeId' : store[0] }
                encoded_url_params = urllib.urlencode(url_params)
                DTR_API_BASIC_AUTH = base64.encodestring('%s:%s' % ("dtr", "dtr18Feb2015")).replace('\n', '')
                
                pushpostrequest = urllib2.Request('http://api.profittill.com/pushnotifications/generateAffiliateUrl', encoded_url_params, headers=aff_url_headers)
                pushpostrequest.add_header("Authorization", "Basic %s" % DTR_API_BASIC_AUTH)
                json_result =  json.loads(urllib2.urlopen(pushpostrequest).read())
                result_url = json_result['url']
            else:
                queryString = urlparse(campaign.get('url').strip()).query
                parsed_url = parse_qs(queryString)
                if not parsed_url.has_key('user_id'):
                    if len(queryString)>0:
                        result_url = campaign.get('url').strip()+'&user_id='+str(record.get('user_id'))
                    else:
                        result_url = campaign.get('url').strip()+'?user_id='+str(record.get('user_id'))
                else:
                    result_url = campaign.get('url').strip()
        if campaign.get('url') is None or str(campaign.get('url'))=='':
            result_url = 'http://api.profittill.com/deals?user_id='+str(record.get('user_id'))
        data = {"message":campaign.get('message'),"cid":str(campaign.get('_id'))+"_"+str(record.get('_id')),"title":campaign.get('title'),
                "type":campaign.get('type'),"url":result_url.strip(),"vibrate":1,"sound":1,"largeIcon":"large_icon",
                "smallIcon":"small_icon","priority":"high","time_to_live":long(campaign.get('expiresat'))-long(time.mktime(datetime.now().timetuple())),"image":campaign.get("image")}
        
        post_data = {}
        
        post_data['data'] = data
        
        if gcmIdGcmRegIdMap.get(gcm_id)[1] == 'gcm':
        
            regIds = []
            regIds.append(gcm_regid)
            post_data['registration_ids'] = regIds
             
            post_data_json = json.dumps(post_data)
            logging.debug('User Id:- '+str(record.get('user_id'))+' Post Data Json :- '+str(post_data_json))
            response = requests.post(GCM_URL, data=post_data_json, headers=headers_gcm)
            
        else:
            
            post_data['to'] = gcm_regid
            post_data['priority'] = 'high'
            post_data['sound'] = 'default'
            post_data['time_to_live'] = 2419200 
            post_data_json = json.dumps(post_data)
            logging.debug('User Id:- '+str(record.get('user_id'))+' Post Data Json :- '+str(post_data_json))
            response = requests.post(FCM_URL, data=post_data_json, headers=headers_fcm)
            
        logging.debug('User Id:- '+str(record.get('user_id'))+' GCM_ID:- '+str(gcm_id)+' Response :-'+str(response.text))
        result = json.loads(response.text)
        if result["success"]:
            get_mongo_connection(host=options.mongoHost).User.pushnotifications.update({'_id':record.get('_id')},{"$set":{'message':'success','type':'sent','sent_timestamp':to_java_date(datetime.now())}})
        else:
            get_mongo_connection(host=options.mongoHost).User.pushnotifications.update({'_id':record.get('_id')},{"$set":{'message':result["results"][0]["error"],'type':'failed','sent_timestamp':to_java_date(datetime.now())}})
            failedGcm.append(gcm_id)
#             updateGcmUserSql = "update gcm_users set failurecount=failurecount+1 where gcm_regid='%s'"%(gcm_regid)
#             logging.debug('Update GCM User Query :-'+str(updateGcmUserSql))
#             try:
#                 dtrdb = MySQLdb.connect('localhost',"root","shop2020","dtr" )
#                 cursor = dtrdb.cursor()
#                 cursor.execute(updateGcmUserSql)
#                 dtrdb.commit()
#                 session.commit()
#                 dtrdb.close()
#             except:
#                 dtrdb.rollback()
#                 dtrdb.close()

def chunks(l, n):
    """Yield successive n-sized chunks from l."""
    for i in xrange(0, len(l), n):
        yield l[i:i+n]

def markNotificationCampaignsProcessed():
    for campaign in notificationCampaignsMap.values():
        logging.debug('Notification Campaign....'+str(campaign.get('_id'))+"...Marked Processed. "+str(datetime.now()))
        get_mongo_connection(host=options.mongoHost).User.notificationcampaigns.update({'_id':campaign.get('_id')},{"$set":{'notification_processed':1}})

def updateFailedGcm():
    updateGcmUserSql = "update gcm_users set failurecount=failurecount+1 where id in (%s)"
    logging.debug("updateGcmUserSql:- "+updateGcmUserSql%(",".join(map(str,failedGcm))))
    if not failedGcm:
        return
    cursor.execute(updateGcmUserSql%(",".join(map(str,failedGcm))))
    
    try:
        db.commit()
    except:
        traceback.print_exc()
        db.rollback()


def main():
    try:
        pendingCampaigns = getPendingCampaigns()
        populateCampaignsMap(pendingCampaigns)
        insertPushNotificationEntriesToSent()
        markNotificationCampaignsProcessed()
        populatePendingNotificationEntriesToBeSent()
        initiateNotificationThreadProcess(options.chunksize)
        updateFailedGcm()
    except:
        traceback.print_exc()
    finally:
        db.close()


if __name__ == '__main__':
    main()