#!/usr/bin/spark-submit

import sys
from operator import add
from pyspark import SparkContext
from pyspark import join
import socket, struct
from bisect import bisect_left

if __name__ == "__main__":

    infileIp =  's3://gu-anly502/maxmind/GeoLite2-Country-Blocks-IPv4.csv'
    infileCountry = 's3://gu-anly502/maxmind/GeoLite2-Country-Locations-en.csv'

    sc     = SparkContext( appName="Wikipedia Geolocate" )
    ## Part A
    linesIp = sc.textFile(infileIp)
    linesCountry = sc.textFile(infileCountry)

    fieldsIp =  linesIp.map(lambda line: line.split(',')).filter(lambda line: line[0]!='network') 
    fieldsCountry =  linesCountry.map(lambda line: line.split(',')).filter(lambda line: line[0]!='geoname_id') 

    geonameIdIp = fieldsIp.map(lambda line: line[1])
    networkIp = fieldsIp.map(lambda line: line[0].split('/'))
    networkIp = networkIp.map(lambda line: line[0])

    geonameIdCountry = fieldsCountry.map(lambda line: line[0])
    nameCountry = fieldsCountry.map(lambda line: line[5])

    geonameNetwork = geonameIdIp.zip(networkIp)
    geonameCountry = geonameIdCountry.zip(nameCountry)

    maxMind = geonameNetwork.join(geonameCountry)

    ## Part B; from http://stackoverflow.com/questions/9590965/convert-an-ip-string-to-a-number-and-vice-versa

    ## Convert an IP string to long
    def ip2long(ip):
        packedIP = socket.inet_aton(ip)
        return struct.unpack("!L", packedIP)[0]

    ipLong = maxMind.map(lambda ip: (ip2long(ip[1][0]),ip[1][1])).collect()

    ## Part C
    bcIpLong = sc.broadcast(ipLong)      
    
    ## Part D
    
    infileFwiki = 's3://gu-anly502/ps03/forensicswiki.2012.txt'
    linesFwiki = sc.textFile(infileFwiki)

    ## Part E

    # Convert IPs in fwiki to long
    fwiki = linesFwiki.map(lambda line: line.split(' - -'))
    fwikiIpLong = fwiki.map(lambda line: (ip2long(line[0]),line[1]))



