from __future__ import print_function
import os, sys
from pyspark.sql import SparkSession, Row
from pyspark.sql.functions import col, collect_list, concat_ws, input_file_name, \
    posexplode, split, sort_array, udf

def idPrefix(path):
    dirs = path.split('/')
    issue = dirs[4]
    if len(issue) != 10:
        return ''
    date = '-'.join([issue[0:4], issue[4:6], issue[6:8]])
    ed = 'ed-' + issue[8:10].lstrip('0')
    return '/'.join(['', 'lccn', dirs[2], date, ed])

if __name__ == '__main__':
    if len(sys.argv) != 3:
        print('Usage: ca-image-ids.py <input> <output>', file=sys.stderr)
        exit(-1)
    spark = SparkSession.builder.appName('Scan CA manifests').getOrCreate()

    ## Spark already munges paths to use '/'
    dirname = udf(lambda path: path.split('/')[-2])

    id_prefix = udf(lambda path: idPrefix(path))
    
    spark.read.text(sys.argv[1]) \
        .select(concat_ws('/', dirname(input_file_name()),
                          split('value', '\s+')[1]).alias('file')) \
        .filter(col('file').endswith('.jp2')) \
        .withColumn('prefix', id_prefix('file')) \
        .filter(col('prefix') != '') \
        .groupBy('prefix') \
        .agg(sort_array(collect_list('file')).alias('files')) \
        .select('prefix', posexplode('files').alias('seq', 'file')) \
        .select(concat_ws('/seq-', 'prefix', col('seq')+1).alias('id'), 'file') \
        .write.json(sys.argv[2])
    spark.stop()

@@ -140,16 +140,29 @@
 ct_list,
+ collect_set,
  concat_
@@ -199,16 +199,38 @@
 explode,
+ regexp_replace, size,
  split, 
@@ -255,24 +255,31 @@
 ef idPrefix(
+batch, 
 path):%0A    d
@@ -285,24 +285,32 @@
 dirs = path.
+lower().
 split('/')%0A 
@@ -308,16 +308,56 @@
 it('/')%0A
+    if len(dirs) %3C 5:%0A        return ''%0A
     issu
@@ -365,17 +365,18 @@
  = dirs%5B
-4
+-2
 %5D%0A    if
@@ -391,13 +391,11 @@
 ue) 
-!= 10
+%3C 9
 :%0A  
@@ -406,24 +406,99 @@
   return ''%0A
+    series = (%5Bs for s in dirs%5B1:%5D if s.startswith('sn')%5D or %5Bdirs%5B2%5D%5D)%5B0%5D%0A
     date = '
@@ -614,22 +614,26 @@
 ', '
-lccn', dirs%5B2%5D
+ca', batch, series
 , da
@@ -984,24 +984,31 @@
 = udf(lambda
+ batch,
  path: idPre
@@ -1011,16 +1011,23 @@
 dPrefix(
+batch, 
 path))%0A 
@@ -1074,35 +1074,33 @@
        .
-select(concat_ws('/
+withColumn('batch
 ', dirna
@@ -1124,10 +1124,85 @@
 e())
-,%0A
+) %5C%0A        .select(col('batch'), concat_ws('/', col('batch'),%0A              
     
@@ -1310,24 +1310,46 @@
 ('.jp2')) %5C%0A
+        .distinct() %5C%0A
         .wit
@@ -1377,20 +1377,97 @@
 prefix('
-file
+batch', 'file')) %5C%0A        .withColumn('dir', regexp_replace('file', '/%5B%5E/%5D+$', '
 ')) %5C%0A  
@@ -1545,16 +1545,50 @@
    .agg(
+collect_set('dir').alias('dirs'), 
 sort_arr
@@ -1626,24 +1626,61 @@
 'files')) %5C%0A
+        .filter(size('dirs') == 1) %5C%0A
         .sel
@@ -1792,17 +1792,19 @@
 l('seq')
-+
+ + 
 1).alias

Update CA manifest extraction.