from __future__ import print_function

import sys
from dateutil import parser
from datetime import *

from pyspark import SparkContext
from pyspark.sql import SQLContext
from pyspark.sql.functions import col, regexp_replace, split, udf

def normdate(s):
    try:
        return parser.parse(s, default=datetime(1800,1,1), fuzzy=True).date().isoformat()
    except ValueError:
        return None

if __name__ == "__main__":
    if len(sys.argv) != 4:
        print("Usage: moa-load.py <input xml> <series json> <output parquet>", file=sys.stderr)
        exit(-1)
    sc = SparkContext(appName="MoA Load")
    sqlContext = SQLContext(sc)

    raw = sqlContext.read.format('com.databricks.spark.xml').options(rowTag='doc').load(sys.argv[1])

    series = sqlContext.read.json(sys.argv[2])

    convdate = udf(lambda s: normdate(s))
    mkurl = udf(lambda ser, id: 'http://ebooks.library.cornell.edu/cgi/t/text/text-idx?c=%s;idno=%s' % (ser, id))

    df = raw.select((split(col('docno'), '_')[0]).alias('moaseries'),
                    (split(col('docno'), '_')[1]).alias('id'),
                    convdate(col('date')).alias('date'),
                    regexp_replace(regexp_replace(col('text'), '&', '&amp;'),
                                   '<', '&lt;').alias('text'))\
            .withColumn('issue', col('id'))\
            .withColumn('url', mkurl(col('moaseries'), col('id')))

    df.join(series, (df.moaseries == series.moaseries) \
            & (df.date >= series.startdate) & (df.date <= series.enddate))\
        .drop('moaseries').drop('startdate').drop('enddate')\
        .write.save(sys.argv[3])
    
    sc.stop()

@@ -228,16 +228,57 @@
 lit, udf
+%0Afrom pyspark.sql.types import StringType
 %0A%0Adef no
@@ -402,18 +402,17 @@
 ept 
-ValueError
+Exception
 :%0A  
@@ -862,16 +862,30 @@
 mdate(s)
+, StringType()
 )%0A    mk
@@ -1568,16 +1568,30 @@
 enddate)
+, 'left_outer'
 )%5C%0A     
@@ -1680,20 +1680,16 @@
 rgv%5B3%5D)%0A
-    
 %0A    sc.

Use left outer join in moa-load.py to handle missing metadata.