Correlating a named file with Standard Input

#!ek9
defines module site.standardTypes.dataCorrelation

  defines program

    DataCorrelation()
      -> argv as List of String

      stdin <- Stdin()
      stdout <- Stdout()
      stderr <- Stderr()

      optionsHandler <- OptionsHandler()
      options <- optionsHandler.processCommandLine(argv)

      verboseMode <- options contains "-v"
      debugLevel <- optionsHandler.processDebugLevel(verboseMode, options.getOrDefault("-d", ""))

      //Use NamedFileProcessor for clean separation of concerns
      filename <- Optional(options.getOrDefault("-f", ""))
      namedFileContents <- NamedFileProcessor(verboseMode, debugLevel).processNamedFile(filename)

      if verboseMode
        stderr.println(`Process id is [${OS().pid()}]`)
      if verboseMode or debugLevel > 0
        stderr.println(`Loaded ${$ length namedFileContents} records from named file`)

      setupSignalHandling(verboseMode, debugLevel)

      validLine <- createLineValidator(stderr, debugLevel)
      toEntry <- createStdinLineHandler(stderr, debugLevel)
      byMerging <- createCustomerRecordMerger(stderr, debugLevel, namedFileContents)

      if debugLevel > 1
        stderr.println("Ready to start processing Standard Input")

      outputHeader(verboseMode, debugLevel, stdout)
      cat stdin | filter by validLine | map toEntry | filter by validEntry | map byMerging > stdout

      if debugLevel > 1
        stderr.println("Standard Input processing complete")

  defines function

    lineValidator() as pure abstract
      -> line as String
      <- rtn as Boolean?

    lineToCustomerRecord() as pure open
      -> line as String
      <- rtn as DictEntry of (String, CustomerRecord)?
      require line?
      rtn: DictEntry() of (String, CustomerRecord)

    customerRecordMerger() as abstract
      -> entry as DictEntry of (String, CustomerRecord)
      <- rtn as String?

    createLineValidator()
      ->
        stderr as Stderr
        debugLevel as Integer
      <-
        validator as lineValidator?

      validator: (stderr, debugLevel) is lineValidator as pure function
        rtn :=? line is not empty and #<line != '#'
        if debugLevel > 1 and not rtn
          stderr.println(`Discarding [${line}]`)

    createStdinLineHandler()
      ->
        stderr as Stderr
        debugLevel as Integer
      <-
        processor as lineToCustomerRecord?

      processor: (stderr, debugLevel) is lineToCustomerRecord as pure function
        rtn: DictEntry() of (String, CustomerRecord)
        traceLevel <- 3
        expectedFieldCount <- 4

        splitLine <- line.split(/,/)
        if debugLevel == traceLevel
          stderr.println(`About to split stdin line [${line}]`)

        if length of splitLine == expectedFieldCount
          id <- splitLine.getOrDefault(0, "").trim()
          firstname <- splitLine.getOrDefault(1, "").trim()
          lastname <- splitLine.getOrDefault(2, "").trim()
          dobStr <- splitLine.getOrDefault(3, "").trim()
          if id is not empty and firstname is not empty and lastname is not empty and dobStr is not empty
            dob <- $getAsDateTime(id, false, dobStr)
            if dob?
              entryValue <- CustomerRecord(id: id, firstName: firstname, lastName: lastname, email: String(), dob: dob, lastPurchase: String())
              if debugLevel == traceLevel
                stderr.println(`Line [${line}] processed`)
              rtn :=? DictEntry(id, entryValue)

        if not rtn?
          stderr.println(`Invalid line [${line}]`)

    getAsDateTime() as pure
      ->
        id as String
        throwException as Boolean
        aDateInput as String //expecting YYYYMMDD
      <-
        rtn as DateTime: DateTime()

      expectedDateParts <- 3
      group <- aDateInput.group(/(\d{4})(\d{2})(\d{2})/)
      if length group == expectedDateParts
        dateStr <- cat group | join with dashSeparated | collect as String
        rtn := DateTime(Date(dateStr))
      else if throwException
        throw Exception(`Id [${id}] Invalid date [${aDateInput}]`, 2)
      else
        Stderr().println(`Id [${id}] Invalid date [${aDateInput}]`)

    validEntry() as pure
      -> entry as DictEntry of (String, CustomerRecord)
      <- rtn as Boolean: entry?

    dashSeparated() as pure
      ->
        firstPart String
        secondPart String
      <-
        rtn as String: firstPart? and secondPart? <- `${firstPart}-${secondPart}` : String()

    outputFormatSeparated() as pure
      ->
        firstPart String
        secondPart String
      <-
        rtn as String: firstPart? and secondPart? <- `${firstPart}, ${secondPart}` : String()

    setupSignalHandling()
      ->
        verboseMode as Boolean
        debugLevel as Integer

      terminationHandler <- (verboseMode, debugLevel) is SignalHandler as function
        result: 1 //The error code to terminate with
        if verboseMode or debugLevel > 0
          Stderr().println(`Handled Terminal Signal [${signal}]`)

      terminations <- Signals().register(["HUP", "ABRT"], terminationHandler)
      if terminations not contains "HUP"
        Stderr().println("HUP Signal not supported")
      if terminations not contains "ABRT"
        Stderr().println("ABRT Signal not supported")

      debugHandler <- (verboseMode, debugLevel) is SignalHandler as function
        maxDebugLevel <- 3
        result: Integer() //Note the Integer is not set so process will not terminate
        if signal == "USR1" and debugLevel < maxDebugLevel
          debugLevel++
        else if signal == "USR2" and debugLevel > 0
          debugLevel--
        if verboseMode or debugLevel > 0
          Stderr().println(`Handled Info Signal [${signal}] Debug Level now [${debugLevel}]`)

      debugs <- Signals().register(["USR1", "USR2"], debugHandler)
      if debugs not contains "USR1"
        Stderr().println("USR1 Signal not supported")
      if debugs not contains "USR2"
        Stderr().println("USR2 Signal not supported")

    outputHeader()
      ->
        verboseMode as Boolean
        debugLevel as Integer
        stdout as Stdout

      detailedDebugLevel <- 2
      enGB <- Locale("en_GB")
      if verboseMode or debugLevel > detailedDebugLevel
        Stderr().println(`About to create output header with locale ${enGB}`)

      envVars <- EnvVars()
      user <- envVars contains "USER" <- envVars.get("USER") else envVars.get("USERNAME")
      dateTime <- enGB.mediumFormat(SystemClock().dateTime())

      stdout.println(`#Created by ${user} on ${dateTime}`)
      stdout.println("#id, firstName, lastName, emailAddress, dateOfBirth, dateLastPurchase")

    createCustomerRecordMerger()
      ->
        stderr as Stderr
        debugLevel as Integer
        namedFileContents as Dict of (String, CustomerRecord)
      <-
        merge as customerRecordMerger?

      merge: (stderr, debugLevel, namedFileContents) is customerRecordMerger
        traceLevel <- 3
        rtn: String() //Output will be unset by default and there for not output
        id <- entry.key()
        if debugLevel == traceLevel
          stderr.println(`Merging id [${id}]`)
        if namedFileContents contains id
          stdinEntry <- entry.value()
          namedFileEntry <- namedFileContents.getOrDefault(id, CustomerRecord())
          stdinEntry :~: namedFileEntry
          rtn := $stdinEntry
        else
          stderr.println(`Not merging [${id}] as named file does not contain id`)

  defines class

    OptionsHandler

      processCommandLine() as pure
        -> arguments as List of String
        <- options as Dict of (String, String)?

        getopts <- setupGetOpt(":")
        options :=? getopts.options(arguments)

      private setupGetOpt() as pure
        -> rqParam as String
        <- getopts as GetOpt?

        supportedOptions <- setupSupportedOptions(rqParam)
        usage <- setupUsage()
        getopts :=? GetOpt(rqParam, supportedOptions, usage)

      private setupSupportedOptions() as pure
        -> rqParam as String
        <- rtn as Dict of (String, String)?

        rtn :=? {
          "-v": String(),
          "-f": rqParam,
          "-d": rqParam
          }

      private setupUsage() as pure
        <- rtn as String?

        desc <- "Invalid option, only those list below are supported:"
        verbose <- "-v, verbose"
        filename <- "-f filename, use file of filename (mandatory option)"
        debug <- "-d level, use of debugging"
        rtn :=? `${desc}\n${verbose}\n${filename}\n${debug}`

      processDebugLevel()
        ->
          verboseMode as Boolean
          level as String
        <-
          rtn as Integer: 0 //default to zero

        maxDebugLevel <- 3
        debugLevel <- Integer(level)

        rtn := (debugLevel in 0 ... maxDebugLevel) <- debugLevel else 0

        if verboseMode and (debugLevel not in 0 ... maxDebugLevel)
          Stderr().println(`Debug level ${debugLevel} not supported - reverting to debug level '0'`)

    NamedFileProcessor
      verboseMode as Boolean: false
      debugLevel as Integer: Integer()

      NamedFileProcessor()
        ->
          verboseMode as Boolean
          debugLevel as Integer
        this.verboseMode = verboseMode
        this.debugLevel = debugLevel

      processNamedFile()
        -> filename as Optional of String
        <- rtn <- Dict() of (String, CustomerRecord)

        stderr <- Stderr()
        if filename?
          namedFile <- getNamedFile(filename.get())
          validLine <- createLineValidator(stderr, debugLevel)
          toEntry <- createNamedFileLineHandler(stderr, debugLevel)

          if verboseMode or debugLevel > 0
            stderr.println(`About to start processing [${namedFile}]`)

          cat namedFile.input() | filter by validLine | map toEntry > rtn

          if debugLevel > 1
            stderr.println(`Processing [${namedFile}] complete`)
        else
          throw Exception("Filename of 'named file' is required", 2)

      private getNamedFile()
        -> filename as String
        <- rtn as TextFile: TextFile()

        filePath <- FileSystemPath(filename)
        rtn := filePath.isAbsolute() <- TextFile(filePath) else TextFile(FileSystem().cwd() + filePath)
        if not rtn?
          throw Exception(`${filePath} is not an existing file`, 2)
        if not rtn.isReadable()
          throw Exception(`${rtn} is not readable`, 2)

      private createNamedFileLineHandler() as pure
        ->
          stderr as Stderr
          debugLevel as Integer
        <-
          processor as lineToCustomerRecord?

        processor :=? (stderr, debugLevel) is lineToCustomerRecord as pure function
          traceLevel <- 3
          expectedFieldCount <- 4
          splitLine <- line.split(/:/)
          if debugLevel == traceLevel
            stderr.println(`About to split line [${line}]`)

          if length of splitLine != expectedFieldCount
            throw Exception(`Invalid line [${line}]`, 2)
          id <- splitLine.getOrDefault(0, "").trim()
          if id is empty
            throw Exception(`Invalid line [${line}] empty ID`, 2)

          emailAddress <- splitLine.getOrDefault(1, "").trim()
          if emailAddress not matches /[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$/
            throw Exception(`Id [${id}] Invalid email address [${emailAddress}]`, 2)

          dateLastPurchase <- $getAsDateTime(id, true, splitLine.getOrDefault(3, "").trim())

          entryValue <- CustomerRecord(id: id, firstName: String(), lastName: String(), email: emailAddress, dob: String(), lastPurchase: dateLastPurchase)
          if debugLevel == traceLevel
            stderr.println(`Named file line [${line}] processed`)

          rtn :=? DictEntry(id, entryValue)

      default operator ?

  defines record

    CustomerRecord
      id <- String()
      firstName <- String()
      lastName <- String()
      email <- String()
      dob <- String()
      lastPurchase <- String()

      default CustomerRecord() as pure

      CustomerRecord() as pure
        ->
          id as String
          firstName as String
          lastName as String
          email as String
          dob as String
          lastPurchase as String
        this.id :=? String(id)
        this.firstName :=? String(firstName)
        this.lastName :=? String(lastName)
        this.email :=? String(email)
        this.dob :=? String(dob)
        this.lastPurchase :=? String(lastPurchase)

      override operator ? as pure
        <- rtn as Boolean: id?

      operator :~:
        -> other as CustomerRecord

        id :=? String(other.id)
        firstName :=? String(other.firstName)
        lastName :=? String(other.lastName)
        email :=? String(other.email)
        dob :=? String(other.dob)
        lastPurchase :=? String(other.lastPurchase)

      operator $ as pure
        <- rtn as String: cat [id, firstName, lastName, email, dob, lastPurchase]
          | join with outputFormatSeparated
          | collect as String