How I Processed 666K Pages of Flattened PDFs into a Full Text Search Engine

In 2017 the National Archives and Records Administration (NARA) released the JFK files in an unsearchable manner 🔍. I tried doing manual research 🕵🏻. I relied on their provided CSV file of metadata to look for relevant documents to discover something - but I was looking for a needle in the haystack. I didn't know where to begin - but at the very least, I wanted to be able to search the contents therein. At least the National Archives allowed me to bulk download the PDFs. From that, I was able to birth the Apario Writer.

In 2020, I began with rails new phoenixvault 🐦‍🔥 and I proceeded on a Zoom call with DJ Nicke - a former animator at Disney - to watch me build the proof of concept of the crowd sourcing declass utility that I envisioned. You see, when I was 7 years old, I had a dream after watching a space focused science program on TV that involved me sitting at the home computer, but interacting with an advanced interface that would help me uncover the mysteries of the day and time of the era. In Stargate SG-1, this concept was explored with the Tolan where Nareem was shocked to discover what Teal'c found in the records buried within a full text interface. Connecting it back to the JFK files released by NARA, they were unsearchable. Agenda on why aside, what could I do about it? This proof of concept grew into a SaaS platform that cost me $7,000 per month to operate over 12 bare meta servers in a private cloud using ESXi. This interface worked, but it was going to be replaced by a cost saving solution architected from the ground up in Go to reduce the dependency graph of the SaaS solution down to a single binary.

In order to do this, I needed to create a pipeline. Looking at the SaaS model, I had a series of sidekiq jobs that compiled the assets. In order to improve the performance of that process, running off from Ruby code, I needed to build a new binary from the ground up using Go. I took the course on YouTube from Matt Holiday called Programming In Go and watched the episodes on repeat several times until the concepts were burned into my per-character programming go-to of choice. I was reminded of the machine detail in circuit design and lower level embedded systems design with the way that Matt explained the concepts behind Go - and they clicked for me. I took that teaching and I applied it to the $7K/mo SaaS model that was complex to manage and costly for a topic that not enough people wanted to continue to subsidize.

But lets take a step back and look at the broader picture first, because Microsoft released on GitHub a Typescript JFK files search engine and it was designed to work with their Cloud Services provided on the Azure cloud. Before I built the proof of concept, I spoke with Microsoft Sales representatives and was quoted tens of thousands per month in order to serve the 666K pages of NARA released JFK files in their full format. The problem was, the interface they built was sloppy and a sales pitch for somebody like me who wanted to dig into the JFK topic using technology - but needed a lunch pad. What Microsoft did was above board and inspirational - even if it was rooted in profit based motives. It actually demonstrated how to use Azure services natively in a manner that if an organization is comfortable with an eight digit annual budget, then c'est le vie! However, I was able to get $7K/mo, but I wasn't working to get eight digits for just the JFK files. So, I built something bigger and better and bolder in a way that has never been done before.

I took concepts that I learned over decades of professional experience at Cisco Systems and Oracle that brought me into WB Games that were translated into the single binary that became the reader that ingests the output of the writer that can also be used with the search index algorithm of 13 unique index types including Gematria. That three binary pipeline needed somewhere to begin. That was the writer. Even though the reader displays the PhoenixVault logo and is the front-end to the concept that I built in 2020.

The Writer

This component compiles the Apario Database that the reader consumes and displays an interactive web application that is self contained with Go powered SSL capabilities. To compile this database, you'll need a Collection of Records in the form of PDF files that you want to gether into a single collection of data that will have each component part of it.

mkdir -p ~/work/projectapario
cd ~/work/projectapario
git clone [e-mail]:ProjectApario/writer.git
cd writer

The writer component is 3344 lines of code across 15 .go files. The install.sh script loads the dependencies onto your system so you can compile your own Apario Database.

chmod +x install.sh && sudo ./install.sh

I use the config.yml strategy with the writer. For example, if I am working on teslafiles.info then I'd have a writer.yml file that I'd use for my config file. To begin, lets ensure that some directories exist:

mkdir -p "~/apario/teslafiles.info/{logs,config,workspace,data,import,app,search,ssl}"

This will create:

~/apario
~/apario/teslafiles.info
~/apario/teslafiles.info/logs
~/apario/teslafiles.info/config
~/apario/teslafiles.info/workspace
~/apario/teslafiles.info/data
~/apario/teslafiles.info/import
~/apario/teslafiles.info/search
~/apario/teslafiles.info/app
~/apario/teslafiles.info/ssl

Directory	Purpose
~/apario/<collection-domain>/logs	Contains log files
~/apario/<collection-domain>/config	Contains `[reader\
{% raw %}~/apario/<collection-domain>/workspace	Binary will use this as a temporary workspace
~/apario/<collection-domain>/data	When importing data into the Apario Database, you place the originals here.
~/apario/<collection-domain>/import	When you're importing data, your structured data goes in here such as your .csv or .xlsx files.
~/apario/<collection-domain>/app	The database(s) live inside of here.
~/apario/<collection-domain>/search	The index files for the Apario Database gematria included binary data.
~/apario/<collection-domain>/ssl	The TLS

The writer.yml file in ~/apario/teslafiles.info/config for teslafiles.info:

---
log: ~/apario/teslafiles.info/logs/writer.log
database-directory: ~/apario/teslafiles.info/app
no-clam: true
language: eng

When using a .csv file, you'd add:

import-csv: ~/apario/teslafiles.info/import/teslafiles.csv
csv-column-url: URL
csv-column-path: PATH
csv-column-record-number: ID
csv-column-title: TITLE

Your .csv file would have:

ID,TITLE,URL,PATH
1,Document One,https://example.com/document1.pdf,~/apario/teslafiles.info/data/document1.pdf
2,Document Two,https://example.com/document2.pdf,~/apario/teslafiles.info/data/document2.pdf
3,Document Three,https://example.com/document3.pdf,~/apario/teslafiles.info/data/document3.pdf

Then running the writer is a matter of:

go build -o writer .
chmod +x writer
./writer -config ~/apario/teslafiles.info/config/writer.yml

Then, the application will begin streaming detailed logging information into the logs: directory choice from the writer.yml setting destination and minimal information to the STDOUT of the ./writer ... invocation.

How The Writer Was Built From A Blank Text File

Now, let's dig into the Go code shall we, and understand how I ingested 666K pages of unsearchable PDFs and built a full text search engine to provide advanced search capabilities that could parse (president kennedy or the president) not (blank page or intentionally left blank) and (top secret or ts/sci or ts//sci) not cover page and get legitimate results. I built that, twice. First in Ruby on Rails, in front of DJ Nicke, and second behind closed doors during holiday breaks when I had the chance to get some uninterrupted programming flow back into swing.

Dependencies

// Binary Dependencies
sl_required_binaries = []string{
    "pdfcpu",
    "gs",
    "pdftotext",
    "convert",
    "pdftoppm",
    "tesseract",
    "clamscan", // optional with `-no-clam` flag
}

Project	License Type	URL
pdfcpu	Apache License 2.0	https://github.com/pdfcpu/pdfcpu
gs (Ghostscript)	AGPL-3.0 (dual-licensed; commercial available)	https://www.ghostscript.com/
pdftotext & pdftoppm (Poppler)	GPL v2 or v3	https://poppler.freedesktop.org/
convert (ImageMagick)	ImageMagick License (Apache-2.0-derived)	https://imagemagick.org/license/
tesseract	Apache License 2.0	https://github.com/tesseract-ocr/tesseract
clamscan (ClamAV)	GPL-2.0	https://github.com/Cisco-Talos/clamav

The pipeline shells out to a mix of copyleft projects. GPL-3.0 was picked because it's compatible with almost everything the project shells out to (Apache-2.0 tools, dual-licensed Poppler, AGPL-3.0 Ghostscript) — the one exception, ClamAV's GPL-2.0-only, doesn't matter since it's just called as a subprocess, not linked in.

To begin, we'll work with the main.go file first:

Before I get to the main() func, there is a directory of assets that contains a JSON file of cryptonyms commonly used throughout the JFK files. This data is particular to the data set itself, but I embed the JSON file itself into the Go program at the package main level below the import () block.

//go:embed bundled/*
var fs_references embed.FS

When we get to building the main() func, its 280 lines long, so we're going to break it down into segments about what its doing. You can follow along in the main.go file on GitHub. I want to focus on the logging component, because how I built the pipeline, the logging matters. It's not one log for the entire pipeline. There is an info log, an error log and a debug log. Some log messages are meant for watching the progress of the compilation. Others are meant for intervention in the event that there is a corrupted PDF file.

log_files = make(map[string]*os.File)

// Initialize log files with truncation
debugFile, err := os.OpenFile(filepath.Join(logDir, "debug.log"), os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644)
if err != nil {
    log.Fatalf("failed to open debug log: %v", err)
}

log_files[cDebugLog] = debugFile

infoFile, err := os.OpenFile(filepath.Join(logDir, "info.log"), os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644)
if err != nil {
    closeLogFiles()
    log.Fatalf("failed to open info log: %v", err)
}

log_files[cInfoLog] = infoFile

errorFile, err := os.OpenFile(filepath.Join(logDir, "error.log"), os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644)
if err != nil {
    closeLogFiles()
    log.Fatalf("failed to open error log: %v", err)
}

log_files[cErrorLog] = errorFile

// Initialize loggers
log_debug = NewCustomLogger(debugFile, "DEBUG: ", log.Ldate|log.Ltime|log.Llongfile, 10)

log_info = NewCustomLogger(infoFile, "INFO: ", log.Ldate|log.Ltime|log.Lshortfile, 10)

log_error = NewCustomLogger(errorFile, "ERROR: ", log.Ldate|log.Ltime|log.Lshortfile, 10)

Next, I need to handle various interrupt signals and run receive_watchdog_signal in the event that they are received. This means Ctrl+C invokes this method.

// interrupt Ctrl+C and other SIGINT/SIGTERM/SIGKILL related signals to the application to quit gracefully
watchdog := make(chan os.Signal, 1)
signal.Notify(watchdog, os.Kill, syscall.SIGTERM, os.Interrupt)
go receive_watchdog_signal(watchdog, logFile, cancel)

Let's look into what this function is doing - because in order to understand why I am using a watchdog, looking into what this function does shows you the tear-down and cleanup process when Ctrl+C is invoked on the writer while its processing.

func receive_watchdog_signal(watchdog chan os.Signal, logFile *os.File, cancel context.CancelFunc) {
    <-watchdog
    log.SetOutput(os.Stdout)
    err := logFile.Close()
    if err != nil {
        log_error.Printf("failed to close the logFile due to error: %v", err)
    }
    defer cancel()

    ch_ImportedRow.Close()       // step 01
    ch_ExtractText.Close()       // step 02
    ch_ExtractPages.Close()      // step 03
    ch_GeneratePng.Close()       // step 04
    ch_GenerateLight.Close()     // step 05
    ch_GenerateDark.Close()      // step 06
    ch_ConvertToJpg.Close()      // step 07
    ch_PerformOcr.Close()        // step 08
    ch_AnalyzeText.Close()       // step 09
    ch_AnalyzeCryptonyms.Close() // step 10
    ch_CompletedPage.Close()     // step 11
    ch_CompiledDocument.Close()  // step 12

    fmt.Printf("Completed running in %d", time.Since(startedAt))

    // excluded pid termination lines L333-L351
}

The binary is designed to support being compiled and executed in a universal format that supports macOS, Linux and Windows natively allowing you compile an Apario Database directory on any system that has the downstream dependencies defined. Getting everything working on Windows is extremely hard and I have not successfully tested it without using WSL2. In that case, the binary is compiled for Linux and not Windows, and is not an .exe - but Go allows you to compile this program into an .exe and run it - even if its not recommended 🤣.

What I hope you noticed was the 12 step process here 👀:

ch_ImportedRow.Close()       // step 01
ch_ExtractText.Close()       // step 02
ch_ExtractPages.Close()      // step 03
ch_GeneratePng.Close()       // step 04
ch_GenerateLight.Close()     // step 05
ch_GenerateDark.Close()      // step 06
ch_ConvertToJpg.Close()      // step 07
ch_PerformOcr.Close()        // step 08
ch_AnalyzeText.Close()       // step 09
ch_AnalyzeCryptonyms.Close() // step 10
ch_CompletedPage.Close()     // step 11
ch_CompiledDocument.Close()  // step 12

This exposes the actual pipeline of what the writer actually does.

FIRST: You begin with a CSV file or an XLSX file that contains the following type of data:

RECORDNO	TITLE	URL	PATH
<string>	<string>	<string>	<string>
R1	Example Title	https://archive.gov/example.pdf	./ingest/example.pdf

This assumes that you have ./ingest as a directory in the same location as the writer binary in this example - but you get what I am doing here. That PATH is local to the machine running the writer. The URL is the original source that should end in .gov or .mil if its going to be authentic OSINT that the PhoenixVault was originally built to serve. However, it's actually not required. You can use URLs like https://mywebsite.com/assets/example.pdf and the interface will use whatever link you provided in the column. This is so your users can see the original document in its original location - perhaps to verify authenticity - perhaps to compare results. The RECORDNO column is a unique-required string field. Some OSINT releases provide reference identifiers, and this <string> field is designed for that value.

What writer does, is take this .csv or .xlsx (sheet 1 only) result and use these 4 columns in the CLI flags when the writer runs. Depending on which technology you're using, the flags change. Let's see what usage would look like based on these two type of ingestion file types. First lets get the source code of the writer on our machine.

mkdir -p ~/workspace/projectapario
cd ~/workspace/projectapario
git clone [e-mail]:ProjectApario/writer.git 
cd writer
mkdir -p bin
go build -o bin/writer .
chmod +x bin/writer
sudo mv bin/writer /usr/local/bin/writer
which writer
mkdir -p ingest # place PDFs in this folder
touch ingest.csv # populate with your documents metadata

Now, lets run it with a CSV file called ingest.csv and a directory of PDFs called ingest exists next to bin. These are the registered flag definitions:

// Import .csv collections
flag_s_import_csv               = config.NewString("import-csv", "", "relative path to an excel spreadsheet where output is a comma separated table of urls and metadata properties. use additional args to associate columns to key data points.")
flag_s_csv_column_url           = config.NewString("csv-column-url", "", "value of row 1 whose column correlates to URLs to download PDF files from")
flag_s_csv_column_path          = config.NewString("csv-column-path", "", "value of row 1 whose column correlates to absolute paths of PDF files")
flag_s_csv_column_record_number = config.NewString("csv-column-record-number", "", "value of row 1 whose column correlates to a unique record identifier or number")
flag_s_csv_column_title         = config.NewString("csv-column-title", "", "value of row 1 whose column correlates to the title of the document")
flag_s_pdf_metadata_json        = config.NewString("metadata-json", "", "json key value map[string]string")

Using it is as simple as:

writer \ # CSV flags 👇🏻
       -import-csv ./ingest.csv \
       -csv-column-url URL \
       -csv-column-path PATH \
       -csv-column-record-number RECORDNO \
       -csv-column-title TITLE \
       -metadata-json "{\"collection\": \"JFK Files\"}"

It looks identical for XLSX with the slight change.

writer \ # XLSX flags 👇🏻
       -import-csv ./ingest.xlsx \
       -xlsx-column-url URL \
       -xlsx-column-path PATH \
       -xlsx-column-record-number RECORDNO \
       -xlsx-column-title TITLE \
       -metadata-json "{\"collection\": \"JFK Files\"}"

Each row in the CSV is gathered into a struct {} and then passed into the ch_ImportedRow channel for processing. What we're doing, is going from a row in the CSV to a compiled JSON record object for the entire processed document and store that file in a structured directory that the reader ingests by determinatively looking into a path as a structured Apario Database that is a directory of files.

// start a bunch of receiver functions to handle when data is ready to be processed
// each of these functions are like black boxes that ONE PAGE from a document is ingested into until it reaches the end

At this point, we need to go back to the main.go file and look at how each of these channels are established.

As each row from the CSV or XSLX is passed into the writer itself, the actual setup of this channel is found in data.go of the project, but in main() I spawned goroutines for each of the pipeline steps. Traditionally, goroutines are for smaller tasks and such, but they can also be used for longer running tasks too.

The ch_ImportedRow Channel

go receiveImportedRow(ctx, ch_ImportedRow.Chan())            // step 01 - runs validate_result_data_record before sending into ch_ExtractText

The actual receiver function uses a for { select {} } pattern in Go to listen on the context.Context's .Done() channel and the incoming <-ch that is an interface{} data type (anything).

func receiveImportedRow(ctx context.Context, ch <-chan interface{}) {
    var err error
    for {
        select {
        case <-ctx.Done():
            return
        case ird, ok := <-ch:
            if ok {
                rd, valid := ird.(ResultData)
                if !valid {
                    log_error.Printf("not valid typecasting for ird to rd.(ResultData)")
                    return
                } // endif
                rd, err = validate_result_data_record(ctx, rd)
                if err != nil {
                    log_error.Tracef("received error on validate_result_data_record for rd.URL %v ; err = %v", rd.URL, err)
                } else {
                    log_info.Printf("validated the downloaded PDF %v from URL %v, sending rd into ch_ExtractText", filepath.Base(rd.PDFPath), rd.URL)
                    if ch_ExtractText.CanWrite() {
                        err := ch_ExtractText.Write(rd)
                        if err != nil {
                            log_error.Tracef("failed to write to ch_ExtractText channel due to error: %v", err)
                            return
                        } // end if
                    } // end if
                } // end if-else
            } // end if ok
        } // end select
    } // end for
} // end func

This pattern is used throughout each of the steps. The resource intensive task in this case is called in the middle of this receiver function.

rd, err = validate_result_data_record(ctx, rd)

When we look into this function, located inside of the pipeline.go file, we can see:

func validate_result_data_record(ctx context.Context, record ResultData) (ResultData, error) {
    log_info.Printf("started validate_result_data_record(%v) = %v", record.Identifier, record.PDFPath)
    // analyze, repair on error, then re-analyze if necessary
    pdf_info, analyze_err := analyze_then_repair_pdf(record.PDFPath)
    if analyze_err != nil {
        return record, log_error.TraceReturn(analyze_err)
    }
    // fix total pages
    if pdf_info.Infos != nil && pdf_info.Infos[0].Pages == 0 && pdf_info.Infos[0].Pages != pdf_info.Infos[0].PageCount {
        pdf_info.Infos[0].Pages = pdf_info.Infos[0].PageCount
    }
    // validate total pages
    if pdf_info.Infos[0].Pages == 0 {
        return record, log_error.TraceReturnf("failed to set pdf_info.Pages to pdf_info.PageCount\npdf_info = %+v", pdf_info)
    }
    // validate pdf
    validate_err := validate_pdf(record.PDFPath)
    if validate_err != nil {
        return record, log_error.TraceReturn(validate_err)
    }
    // optimize pdf
    optimize_err := optimize_pdf(record.PDFPath)
    if optimize_err != nil {
        return record, log_error.TraceReturn(optimize_err)
    }
    return record, nil
}

Here we're running analyze_then_repair_pdf then using the metadata from the result of the call to populate some structured data. Then we call the validate_pdf to confirm, then optimize_pdf. When we have a problem, we're using the verbose.TraceReturn directly as the log_error handler that is a verbose logger with secrets protection and stack trace analysis capabilities. These can spawn child processes and take several seconds to complete depending on the size of the PDF file being analyzed. If it's gigabytes in size, like this writer has processed successfully, then it'll take a few minutes to complete. That's normal. In the info_log you'll see the begin and end time per document and you'll see some documents take several minutes to complete. This means that the pipeline built within the writer is capable of managing many system resources on a bare metal machine, utilize 100% of its capabilities while safely managing resources across the system that include spawned projects where some are single threaded, and some are multi-threaded. When a multi-threaded tesseract process is spawning, the writer has semaphores in place to ensure that its not going to use too many system resources and that tesseract is contained within a lane of processing compute power proportional to the impact that the pipeline has on the greater runtime.

The ch_ExtractText Channel

This is step 2 of the pipeline where we have to use a third party binary in order to extract the text out of the PDF. I didn't want to reinvent the wheel - I wanted to ingest 666K pages of JFK files as fast as possible, and if the tool works and its license is compatible, let's go!

go receiveOnExtractTextCh(ctx, ch_ExtractText.Chan())        // step 02 - runs extractPlainTextFromPdf before sending into ch_ExtractPages

func receiveOnExtractTextCh(ctx context.Context, ch <-chan interface{}) {
    for {
        select {
        case <-ctx.Done():
            return
        case ird, ok := <-ch:
            if ok {
                rd, ok := ird.(ResultData)
                if !ok {
                    log_error.Tracef("failed to assert the ird from ch_ExtractText as type ResultData")
                    return
                }
                log_info.Printf("received rd from ch_ExtractText for URL %v, running extractPlainTextFromPdf(%v)", rd.URL, rd.Identifier)
                go extractPlainTextFromPdf(ctx, rd)
            } else {
                log_debug.Println("ch_ExtractText is closed but received some data")
                return
            }
        }
    }
}

Here you can see that the bulk of the processing takes place in the go extractPlainTextFromPdf(ctx, rd) spawning of another goroutine; in this case, we aren't going to wait for it in this particular loop of receiveOnExtractTextCh's for { select {} } lifecycle until the context.Context is canceled. That context is then passed into the extractPlainTextFromPdf, where an external tool is invoked.

func extractPlainTextFromPdf(ctx context.Context, record ResultData) {
    defer func() {
        log_info.Printf("finished extracting the text from the PDF %v, now sending rd into ch_ExtractPages", filepath.Base(record.PDFPath))
        if ch_ExtractPages.CanWrite() {
            err := ch_ExtractPages.Write(record)
            if err != nil {
                log_error.Tracef("failed to write record %v into the ch_ExtractPages due to error %v", record, err)
            }
        }
    }()
    log_info.Printf("started extractPlainTextFromPdf(%v) = %v", record.Identifier, record.PDFPath)
    if ok, err := fileHasData(record.ExtractedTextPath); !ok || err != nil {
        /*
            pdftotext REPLACE_WITH_FILE_PATH REPLACE_WITH_TEXT_OUTPUT_FILE_PATH
        */
        cmd_extract_text_pdf := exec.Command(m_required_binaries["pdftotext"], record.PDFPath, record.ExtractedTextPath)
        var cmd4_extract_text_pdf_stdout bytes.Buffer
        var cmd4_extract_text_pdf_stderr bytes.Buffer
        cmd_extract_text_pdf.Stdout = &cmd4_extract_text_pdf_stdout
        cmd_extract_text_pdf.Stderr = &cmd4_extract_text_pdf_stderr
        sem_pdftotext.Acquire()
        cmd_extract_text_pdf_err := cmd_extract_text_pdf.Run()
        sem_pdftotext.Release()
        if cmd_extract_text_pdf_err != nil {
            log_error.Tracef("Failed to execute command `pdftotext %v %v` due to error: %s\n", record.PDFPath, record.ExtractedTextPath, cmd_extract_text_pdf_err)
            return
        }
    }
}

This uses the semaphore package in order to .Acquire() and .Release() on an invocation of .Run() from an exec.Command on the required m_required_binaries["pdftotext"] pdftotext binary invocation. We're going to use a bytes.Buffer to collect on STDOUT and STDERR from the child process and we're using a semaphore to control how many times this particular function is allowed to be called system wide while writer compiles.

The ch_ExtractPages Channel

go receiveOnExtractPagesCh(ctx, ch_ExtractPages.Chan())      // step 03 - runs extractPagesFromPdf before sending PendingPage into ch_GeneratePng

func receiveOnExtractPagesCh(ctx context.Context, ch <-chan interface{}) {
    for {
        select {
        case <-ctx.Done():
            return
        case ird, ok := <-ch:
            if ok {
                rd, ok := ird.(ResultData)
                if !ok {
                    log_error.Tracef("ch_ExtractPages receive an ird but cannot cast it as a .(ResultData) type")
                    return
                }
                log_info.Printf("received on ch_ExtractPages URL %v, running extractPagesFromPdf(%v)", rd.URL, rd.Identifier)
                go extractPagesFromPdf(ctx, rd)
            } else {
                log_debug.Trace("ch_ExtractPages is closed but received some data")
                return
            }
        }
    }
}

This function is quite large. Let's look at it and then talk about it.

func extractPagesFromPdf(ctx context.Context, record ResultData) {
   log_info.Printf("started extractPagesFromPdf(%v) = %v", record.Identifier, record.PDFPath)
   /*
      pdfcpu extract -mode page REPLACE_WITH_FILE_PATH REPLACE_WITH_OUTPUT_DIRECTORY
   */
   pagesDir := filepath.Join(record.DataDir, "pages")
   sm_page_directories.Store(record.Identifier, pagesDir)
   _, pagesDirExistsErr := os.Stat(pagesDir)
   performPagesExtract := false
   if os.IsNotExist(pagesDirExistsErr) {
      performPagesExtract = true
   } else {
      ok, err := DirHasPDFs(pagesDir)
      if err == nil && ok {
         performPagesExtract = true
      }
   }
RETRY:
   if performPagesExtract {
      pagesDirErr := os.MkdirAll(pagesDir, 0755)
      if pagesDirErr != nil {
         log_error.Tracef("failed to create directory %v due to error %v", pagesDir, pagesDirErr)
         return
      }
      cmd_extract_pages_in_pdf := exec.Command(m_required_binaries["pdfcpu"], "extract", "-mode", "page", record.PDFPath, pagesDir)
      var cmd_extract_pages_in_pdf_stdout bytes.Buffer
      var cmd_extract_pages_in_pdf_stderr bytes.Buffer
      cmd_extract_pages_in_pdf.Stdout = &cmd_extract_pages_in_pdf_stdout
      cmd_extract_pages_in_pdf.Stderr = &cmd_extract_pages_in_pdf_stderr
      sem_pdfcpu.Acquire()
      cmd_extract_pages_in_pdf_err := cmd_extract_pages_in_pdf.Run()
      sem_pdfcpu.Release()
      if cmd_extract_pages_in_pdf_err != nil {
         log_error.Tracef("Failed to execute command `pdfcpu extract -mode page %v %v` due to error: %s\n", record.PDFPath, pagesDir, cmd_extract_pages_in_pdf_err)
         return
      }
   } else {
      log_info.Printf("not performing `pdfcpu extrace -mode page %v %v` because the directory %v already has PDFs inside it", record.PDFPath, pagesDir, pagesDir)
      check := len(pagesDir) == len(pagesDir)

      if check {
         goto RETRY
      }
   }

   pagesDirWalkErr := filepath.Walk(pagesDir, func(path string, info os.FileInfo, err error) error {
      if err != nil {
         log_error.Tracef("Error accessing a path %q: %v\n", path, err)
         return err
      }

      if !info.IsDir() && strings.HasSuffix(info.Name(), ".pdf") {
         nameParts := strings.Split(info.Name(), "_page_")
         if len(nameParts) < 2 {
            return fmt.Errorf("incorrect filename provided as %v", info.Name())
         }
         pgNoStr := strings.ReplaceAll(nameParts[1], ".pdf", "")
         pgNo, pgNoErr := strconv.Atoi(pgNoStr)
         if pgNoErr != nil {
            return fmt.Errorf("failed to extract the pgNo from the PDF filename %v", info.Name())
         }
         identifier := NewIdentifier(9)
         pp := PendingPage{
            Identifier:       identifier,
            RecordIdentifier: record.Identifier,
            PageNumber:       pgNo,
            PagesDir:         pagesDir,
            PDFPath:          path,
            OCRTextPath:      filepath.Join(pagesDir, fmt.Sprintf("ocr.%06d.txt", pgNo)),
            ManifestPath:     filepath.Join(pagesDir, fmt.Sprintf("page.%06d.json", pgNo)),
            PNG: PNG{
               Light: Images{
                  Original: filepath.Join(pagesDir, fmt.Sprintf("page.light.%06d.original.png", pgNo)),
                  Large:    filepath.Join(pagesDir, fmt.Sprintf("page.light.%06d.large.png", pgNo)),
                  Medium:   filepath.Join(pagesDir, fmt.Sprintf("page.light.%06d.medium.png", pgNo)),
                  Small:    filepath.Join(pagesDir, fmt.Sprintf("page.light.%06d.small.png", pgNo)),
                  Social:   filepath.Join(pagesDir, fmt.Sprintf("page.light.%06d.social.png", pgNo)),
               },
               Dark: Images{
                  Original: filepath.Join(pagesDir, fmt.Sprintf("page.dark.%06d.original.png", pgNo)),
                  Large:    filepath.Join(pagesDir, fmt.Sprintf("page.dark.%06d.large.png", pgNo)),
                  Medium:   filepath.Join(pagesDir, fmt.Sprintf("page.dark.%06d.medium.png", pgNo)),
                  Small:    filepath.Join(pagesDir, fmt.Sprintf("page.dark.%06d.small.png", pgNo)),
                  Social:   filepath.Join(pagesDir, fmt.Sprintf("page.dark.%06d.social.png", pgNo)),
               },
            },
            JPEG: JPEG{
               Light: Images{
                  Original: filepath.Join(pagesDir, fmt.Sprintf("page.light.%06d.original.jpg", pgNo)),
                  Large:    filepath.Join(pagesDir, fmt.Sprintf("page.light.%06d.large.jpg", pgNo)),
                  Medium:   filepath.Join(pagesDir, fmt.Sprintf("page.light.%06d.medium.jpg", pgNo)),
                  Small:    filepath.Join(pagesDir, fmt.Sprintf("page.light.%06d.small.jpg", pgNo)),
                  Social:   filepath.Join(pagesDir, fmt.Sprintf("page.light.%06d.social.jpg", pgNo)),
               },
               Dark: Images{
                  Original: filepath.Join(pagesDir, fmt.Sprintf("page.dark.%06d.original.jpg", pgNo)),
                  Large:    filepath.Join(pagesDir, fmt.Sprintf("page.dark.%06d.large.jpg", pgNo)),
                  Medium:   filepath.Join(pagesDir, fmt.Sprintf("page.dark.%06d.medium.jpg", pgNo)),
                  Small:    filepath.Join(pagesDir, fmt.Sprintf("page.dark.%06d.small.jpg", pgNo)),
                  Social:   filepath.Join(pagesDir, fmt.Sprintf("page.dark.%06d.social.jpg", pgNo)),
               },
            },
         }
         sm_pages.Store(pp.Identifier, pp)
         err := WritePendingPageToJson(pp)
         if err != nil {
            return err
         }
         log_info.Printf("sending page %d (ID %v) from record %v URL %v into the ch_GeneratingPng", pgNo, identifier, record.Identifier, record.URL)
         if ch_GeneratePng.CanWrite() {
            err := ch_GeneratePng.Write(pp)
            if err != nil {
               log_error.Tracef("cannot send pp into ch_GeneratePng channel due to error %v", err)
               return err
            }
         }
      }

      return nil
   })

   if pagesDirWalkErr != nil {
      log_error.Tracef("Error walking the path ./pages: %v\n", pagesDirWalkErr)
      return
   }

   return
}

You can see RETRY: and goto RETRY? Well, thats a way of recursively calling a block within the function until something works - without having to deal with invoking and tearing down functions in between. In the case of recursion, I'm simply using the performPagesExtract as a variable to determine whether or not we should even consider using goto RETRY. When performPagesExtract is false, I reconsider what if true? If true, that means that the check passes and we actually did need to run it but we didn't have something we needed first - so we fix that - and then re-run it.

Next what I do is I build out a PendingPage{} struct with a bunch of metadata about the individual page from the given document. This is performed by using filepath.Walk(pagesDir, <walkFunc> {}) invocation on the output of the previous command that took the PDF file and extracted in a sub-directory one PDF per page in the original file. We're going to iterate over those compiled documents now, and we're going to populate this PendingPage{} record with the gathered metadata about this particular page.

We're using semaphores here, and we're also invoking WritePendingPageToJson(pp), which effectively compiles the Apario Database from scratch by writing individual JSON files to structured locations based on the page analysis output.

I could go into each of the channels and break them down - but they are very similar in structure to what we have just explored from start to finish. What felt like a short ride, can be a journey of discovery by studying the writer and either utilizing it for public service or contributing to its long-term stability.

The writer is responsible for ingesting a CSV file that can have as many as 72K rows of records inside of it that contain 666K pages of OSINT in unsearchable manner.

The full text analysis uses textee and gematria in order to take the OCR extract text, or the PDF's containing text and break sentences like:

The lazy brown dog jumps over the quick fox.

Into:

The (2)
The lazy (1)
The lazy brown (1)
lazy (1)
lazy brown (1)
lazy brown dog (1)
brown (1)
brown dog (1)
brown dog jumps (1)
dog (1)
dog jumps (1)
dog jumps over (1)
jumps (1)
jumps over (1)
jumps over the (1)
over (1)
over the (1)
over the quick (1)
the quick (1)
the quick fox (1)
quick (1)
quick fox (1)
fox (1)

Notice the The (2) and the way that between over the quick (1) and the quick (1) we omit the (1) by making the first The a 2. That's what textee does.

The gematria package is responsible for calculating the Simple, English, Jewish, Eights, Majestic and Mystery values.

The pipeline operates in this manner:

1. Rows from a CSV are ingested
2. PDF is scanned, analyzed, optimized and rendered into a directory of individual page PDF files
3. Each page PDF file is individually processed - keeping track of its original directory and original extracted document
4. The .pdf is converted into a .png
5. The .png is compiled into a progressive light JPEG
6. The .png is converted into dark mode
7. The dark mode .png is compiled into a progressive dark JPEG
8. The .pdf is processed through pdftotext to extract text
9. The .pdf is processed through tesseract to extract text
10. The extracted text is analyzed against 7 regular expressions to find dates and cryptonyms
11. The PendingPage{} struct is passed into aggregatePendingPage in the aggregate.go file

When the dates are analyzed, I opted to use regular expressions for it as defined in the data.go file:

re_date1 = regexp.MustCompile(`(?i)(\d{1,2})(st|nd|rd|th)?\s(?:of\s)?(January|Jan|February|Feb|March|Mar|April|Apr|May|June|Jun|July|Jul|August|Aug|September|Sep|October|Oct|November|Nov|December|Dec),?\s(\d{2,4})`)

re_date2 = regexp.MustCompile(`(?i)(\d{1,2})\/(\d{1,2})\/(\d{2,4})`)

re_date3 = regexp.MustCompile(`(?i)(January|Jan|February|Feb|March|Mar|April|Apr|May|June|Jun|July|Jul|August|Aug|September|Sep|October|Oct|November|Nov|December|Dec),?\s(\d{2,4})`)

re_date5 = regexp.MustCompile(`(?i)(January|Jan|February|Feb|March|Mar|April|Apr|May|June|Jun|July|Jul|August|Aug|September|Sep|October|Oct|November|Nov|December|Dec)\s(\d{1,2})(st|nd|rd|th)?,?\s(\d{2,4})`)

re_date4 = regexp.MustCompile(`(?i)(January|Jan|February|Feb|March|Mar|April|Apr|May|June|Jun|July|Jul|August|Aug|September|Sep|October|Oct|November|Nov|December|Dec)\s(\d{4})`)

re_date6 = regexp.MustCompile(`(\d{4})`)

Then inside of dates.go I am consuming them like so:

match1 := re_date1.FindAllStringSubmatch(in, -1)

match2 := re_date2.FindAllStringSubmatch(in, -1)

match3 := re_date3.FindAllStringSubmatch(in, -1)

match4 := re_date4.FindAllStringSubmatch(in, -1)

match5 := re_date5.FindAllStringSubmatch(in, -1)

match6 := re_date6.FindAllStringSubmatch(in, -1)

Since time is an illusion and we have multiple types, we can look at:

day, dayErr := strconv.Atoi(m[?])

month := getMonthFromString(m[?])
month, _ := strconv.Atoi(m[?])

year, yearErr := strconv.Atoi(m[?])

I conclude analyzing each of these matches with this function:

func uniqueTimes(times []time.Time) []time.Time {
    seen := make(map[time.Time]bool)
    var unique []time.Time
    for _, t := range times {
        if t.Year() < 1800 || t.Year() > time.Now().Year() {
            continue
        }
        t = time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, t.Location())
        if _, ok := seen[t]; !ok {
            unique = append(unique, t)
            seen[t] = true
        }
    }
    return unique
}

Aggregating PendingPage {} In Pipeline

func aggregatePendingPage(ctx context.Context, pp PendingPage) {
   data_rd, result_data_found := sm_resultdatas.Load(pp.RecordIdentifier)
   // process missing from sync.Map

   rd, rd_cast_ok := data_rd.(ResultData)
   // process invalid ResultData cast

   document_data, document_found := sm_documents.Load(pp.RecordIdentifier)
   // process missing from sync.Map

   document, document_cast_ok := document_data.(Document)
   // process invalid Document cast

   // print document page count results

   mu := DocumentLocker(document.Identifier)
   mu.Lock()
   if document.Pages == nil {
      document.Pages = make(map[int64]Page)
   }
   document.Pages[int64(pp.PageNumber)] = Page{
      Identifier:         pp.Identifier,
      DocumentIdentifier: pp.RecordIdentifier,
      PageNumber:         int64(pp.PageNumber),
   }
   defer mu.Unlock()
   ticker := time.NewTicker(333 * time.Millisecond)
   timer := time.NewTimer(999 * time.Second)
   if int64(len(document.Pages)) == document.TotalPages {
      for {
         select {
         case <-ctx.Done():
            log_debug.Printf("context canceled on aggregatePendingPage %+v", pp)
            return
         case <-timer.C:
            log_error.Printf("FATAL TO DOCUMENT aggregatePendingPage failed to ingest %+v into the ch_CompiledDocument", pp)
            return
         case <-ticker.C:
            err := ch_CompiledDocument.Write(document)
            log_info.Printf("ch_CompiledDocument.CanWrite = %v", ch_CompiledDocument.CanWrite())
            if err != nil {
               log_error.Printf("FATAL TO DOCUMENT - err in writing to channel: %v", err)
            }
            return
         }
      }
   } else {
      log_debug.Printf("aggregatePendingPage document %v page %d received but the document.Pages are at %d of %d so waiting before sending into ch_CompiledDocument",
         pp.RecordIdentifier, pp.PageNumber, len(document.Pages), document.TotalPages)
   }
   return
}

As you can see, we're using a sync.Map called sm_resultdatas.Load(pp.RecordIdentifier) in order to retrieve the PendingPage{} struct out of memory according to the RECORDNO column from the original CSV or XLSX file.

We're using the DocumentLocker in order to safely write to document.Pages[int64(pp.PageNumber)] in a concurrent multi-threaded environment where resources are shared across threads and child processes.

type Document struct {
    Identifier          string         `json:"identifier"`
    URL                 string         `json:"url"`
    Pages               map[int64]Page `json:"pages"`
    TotalPages          int64          `json:"total_pages"`
    CoverPageIdentifier string         `json:"cover_page_identifier"`
    Collection          Collection     `json:"collection"`
}

type Page struct {
    Identifier         string            `json:"identifier"`
    DocumentIdentifier string            `json:"document_identifier"`
    PageNumber         int64             `json:"page_number"`
    Metadata           map[string]string `json:"metadata"`
    FullTextGematria   gem.Gematria      `json:"full_text_gematria"`
    FullText           string            `json:"full_text"`
}

The lockers are defined in the lockers.go file, and the init() func is what I want to point your attention to next.

func init() {
    // Checksums
    cslMu = &sync.RWMutex{}
    checksumLockers = make(map[string]*sync.RWMutex)

    // Documents
    dlMu = &sync.RWMutex{}
    documentLockers = make(map[string]*sync.RWMutex)

    // Pages
    pageLockers = make(map[string]*sync.RWMutex)
    plMu = &sync.RWMutex{}
}

Here, we're registering a couple of map[string]*sync.RWMutex for the documentLockers and pageLockers in the package.

There are three types; the first is for checksums ; used to take the URL, generate a SHA512 string and use that as the Apario Database subdirectory to store the structured contents into. Rendered assets are stored here and JSON files written to disk live here. In order to treat the filesystem like this as a database, I needed to have various lockers. A checksum level locker - for work being done to the root level of the document record. The DocumentLocker for per ID records in the CSV or XLSX file ingested.

// Document Locker
var dlMu *sync.RWMutex
var documentLockers map[string]*sync.RWMutex
var DocumentLocker = func(id string) *sync.RWMutex {
    dlMu.Lock()
    log.Printf("DocumentLocker(%s) Locked...", id)
    defer func() {
        log.Printf("... DocumentLocker(%s) Unlocked!", id)
        dlMu.Unlock()
    }()
    if _, found := documentLockers[id]; !found {
        documentLockers[id] = &sync.RWMutex{}
    }
    return documentLockers[id]
}

The DocumentLocker variable is a function that intentionally can be overwritten with a new function of the same signature type. The dlMu is a documentLockers's mutex, different from a record's locker defined as documentLockers[id] = &sync.RWMutex{}.

This pattern allowed me to create lockers for pages that were dynamically belonging to documents in your original CSV or XLSX that was passed into the arguments of the writer when it executed.

for {
   select {
   case <-ctx.Done():
      elapsed := time.Since(startedAt)
      log.Printf("Completed task in %.0f seconds", elapsed.Seconds())
      return
   case <-ch_Done:
      log.SetOutput(os.Stdout)
      log.Printf("done processing everything... time to end things now!")
      watchdog <- os.Kill
   case id, ok := <-ch_CompiledDocument.Chan():
      if ok {
         d, ok := id.(Document)
         if !ok {
            log_error.Printf("cannot typecast the final result for %s as a .(Document)", d.Identifier)
         }
         a_i_received_documents.Add(1)
         log_info.Printf("a_i_total_documents == a_i_received_documents ; %d == %d",
            a_i_total_documents.Load(), a_i_received_documents.Load())

         if a_i_total_documents.Load() == a_i_received_documents.Load() {
            log.SetOutput(os.Stdout)
            log.Printf("Completed processing document %v", d)
            ch_Done <- struct{}{}
         }
      }
   }
}

I chose to end the main() func with a for { select {} } that listens on the background context's .Done() channel, the ch_Done channel being written to, or a new completed row from the CSV or XLSX is received compiled via <-ch_CompiledDocument.Chan().

I created a variable called a_i_received_documents to track the total received documents. The a_ is for atomic and i_ is for int, so the atomic.Int is called received_documents Once every row has been processed, the struct{}{} is written to the ch_Done channel.

If you've noticed the .Chan() throughout this, that is the smartchan that I wrote.

package go_smartchan

import (
    "errors"
    `fmt`
    "sync"
    `sync/atomic`
)

type SmartChan struct {
    ch     chan interface{}
    mu     sync.RWMutex
    closed atomic.Bool
    cnt    atomic.Int64
}

func NewSmartChan(i int) *SmartChan {}

func (sc *SmartChan) Write(thing interface{}) (err error) {}
func (sc *SmartChan) Read() (interface{}, error) {}
func (sc *SmartChan) Count() int64 {}
func (sc *SmartChan) Close() {}
func (sc *SmartChan) Chan() chan interface{} {}
func (sc *SmartChan) CanWrite() bool {}

This is used throughout the writer and throughout a bunch of other projects that I've worked with. Asking .CanWrite() on a channel is a task that allows you to build in pipelines when channels become unwritable and then writable. It attempts to remove the panic() from using channels in Go.

I didn't want to close out on the piece before walking you through some of the more complex AI generated functions that I did use. At this time, ChatGPT was the utility that I was using, and these functions served their purposes for what they were intended to perform. Given their open source nature, discussing them here is relevant to the context of this piece.

func nrgbaToRGBA(src *image.NRGBA) *image.RGBA {
   b := src.Bounds()
   dst := image.NewRGBA(b)

   for y := b.Min.Y; y < b.Max.Y; y++ {
      for x := b.Min.X; x < b.Max.X; x++ {
         r, g, b, a := src.At(x, y).RGBA()

         // Convert from non-premultiplied to premultiplied alpha
         r = (r * a) / 0xffff
         g = (g * a) / 0xffff
         b = (b * a) / 0xffff

         dst.SetRGBA(x, y, color.RGBA{
            R: uint8(r >> 8),
            G: uint8(g >> 8),
            B: uint8(b >> 8),
            A: uint8(a >> 8),
         })
      }
   }

   return dst
}

A little bot named Red lived in Paint Town. Red loved to help kids make their pictures look pretty!

One day, tiny Tim brought a clear sheet with colors on it. "I want to show my friends!" said Tim.

"I can help!" beeped Red. "Watch me make it pretty!"
Red took a white paper and looked at each tiny spot on Tim's clear sheet.

"When I see light spots," beeped Red, "I make them soft like clouds."

"When I see bright spots," beeped Red, "I make them strong like rainbows!"

Spot by spot, Red made Tim's picture. Some colors were soft, some were bright!

"Wow!" smiled Tim. "Now my friends can see my picture too!"

Red did a happy beep-boop dance. Helping make pictures was Red's favorite thing!

The End.

Conclusion

This application, the first of three in the Project Apario trio of open source software takes a .csv or .xlsx (first sheet only) and processes the structured data RECORDNO,TITLE,URL,PATH into an Apario Database that the reader renders into an ultra-fast self-contained web application called PhoenixVault.

The original project written was a Ruby on Rails implementation that was built on a Zoom call with a former Disney animator over a 17 day flow state. I have a witness for what I built. DJ watched me built the pipeline in Rails using sidekiq making each of the system calls for each of the steps. When I took a chance on myself over Christmas 2021 to rewrite the Rails implementation that was still costing me over $2,000 per month - even after removing all of the SaaS and going completely in-house private cloud for everything. This meant that I spent a year migrating the external SaaS model that used MongoDB Atlas, ElasticSearch As A Service called ELK, and DigitalOcean Managed Redis running behind a private cloud built behind ESXi and Rocky 9 Linux with SELinux enabled. I reduced my OPEX from $7K down to $2K per month. That comes out to a $180K savings over 3 years. I wasn't going to keep asking the American people to pay me $180K to keep the PhoenixVault online for you. Instead, what I built in the trio application, was a solution that can run on an OVH Bare Metal host for $33 per month.

It's simple. The $33/mo host is sufficient. When you get it, set it up like so:

ssh -i ~/.ssh/mykey rocky@17.76.17.76 # replace with your IP address and replace ~/.ssh/mykey with your SSH key
rocky@ip-17-76-17-76: 

Then you can set it up like so:

sudo mkdir -p /apario/{reader,writer,search}/{workspace,config,logs}

Then you can do:

cd /apario/reader
git clone [e-mail]:ProjectApario/reader.git src

cd /apario/writer
git clone [e-mail]:ProjectApario/writer.git src

cd /apario/search
git clone [e-mail]:ProjectApario/search.git src 

Then you'll need to render a YAML script for each of the instances that you're running. If your project is called acme-disclosure then you're have:

/apario/reader/src
/apario/writer/src
/apario/search/src

In addition to that, you'll also create a symbiotic relationship between:

/apario/reader/acme-disclosure/database

and

/apario/writer/acme-disclosure/ingest

such that the writer writes to the /apario/reader/acme-disclosure/database path and the reader reads the same path. The writer processes the flattened or non-optimized PDFs from the /apario/writer/acme-disclosure/ingest.

This is useful for ACME (a fictional company) to release a disclosure of information in the form of a PDF dump. It's common with documentation and with OSINT practices. For the context of the JFK files from the National Archives, I wrote the writer to ingest those files into a useable format that I ended up writing a consumer for.

This is the first of three binaries and the project is incomplete because of life. When the time comes to dust off what I have built and continue the torch, the time will come. Until that time arrives, I type to share what I learned so that the youth can pick up the torch.

The vision that I had as a child wasn't for me alone to experience. It was for the world to discover the beauty that dwells within and how technology helps us understand the complex reality that we live inside. 💜