Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/missionmatch / cmd/opportunity-extract/main.go |
| Date | Not specified in source |
| Repository | Janady13/missionmatch |
| Revision | 720d47fae476b0d38477ae4401697d372b1ea2bc |
| Branch | feature/opportunity-to-outcome |
| Path or archive member | cmd/opportunity-extract/main.go |
| Original SHA-256 | 85ae4bfae37946d56ba7c47d1c714acf4ae3f20215ca76bab791a8f306e37484 |
| Git blob SHA | 15f1ed1f8f9216e99074b78ec5ecdab554fe0d01 |
| Reading-copy SHA-256 | 85ae4bfae37946d56ba7c47d1c714acf4ae3f20215ca76bab791a8f306e37484 |
Recovered source transcript
Download complete source text ↓717 lines · 27.8 KB
// Command opportunity-extract turns an uploaded document into citable chunks. Directive Section 5.1.
//
// THIS RUNS BEFORE ANY MODEL, and the ordering is the design rather than an optimization: what a
// solicitation states plainly should be found by parsing it, so the model is left to NORMALIZE language
// rather than to extract facts. Everything this writes is a literal span of the source with a content hash,
// so a scope item derived from it can be traced to the exact place it came from.
//
// PROVENANCE IS NEVER SUPPLIED BY THIS PROGRAM. Migration 0081 makes an opportunity_document_chunks row born
// tenant_upload, and the INSERT below deliberately omits the column so that is visible at the call site. A
// chunk of a customer's package can therefore never claim to be public, whatever this code believes.
//
// IT SURVIVES BAD INPUT RATHER THAN CRASHING ON IT. A document that is malformed, oversized, empty, locked,
// or simply not text is marked failed WITH A REASON and the run continues to the next one. A batch job that
// dies on the first bad file leaves every later file unprocessed and tells you about one problem instead of
// all of them.
//
// Usage:
//
// opportunity-extract -dsn <migrator dsn> [-limit 20] [-document <uuid>] [-dry-run]
package main
import (
"bytes"
"compress/zlib"
"context"
"flag"
"io"
"log"
"os"
"strings"
"time"
"unicode"
"unicode/utf8"
"github.com/jackc/pgx/v5/pgxpool"
pdflib "github.com/ledongthuc/pdf"
"github.com/missionworkforce/missionmatch/internal/modules/opportunities"
"github.com/missionworkforce/missionmatch/internal/platform/storage"
)
// Failure reasons are a CLOSED vocabulary written to extraction_error_code, so an operator can group
// failures rather than reading prose. Each one names a condition, never a stack trace.
const (
errBytesMissing = "bytes_missing" // the row points at storage and storage does not have it
errEmpty = "empty_document" // zero bytes, or nothing but whitespace after decoding
errTooLarge = "too_large" // beyond what this job will hold in memory
errEncrypted = "encrypted_or_locked"
errUnsupported = "unsupported_format"
errNoTextLayer = "no_text_layer" // a PDF of scanned images, which needs OCR this build does not have
errNoChunks = "no_chunks_found" // text extracted, and nothing citable in it
)
// maxExtractBytes bounds what this job will read into memory per document. Smaller than the upload cap on
// purpose: a batch running twenty documents at 64 MiB each is a memory incident waiting for a busy day.
const maxExtractBytes = 32 << 20
func main() {
// THE ISOLATED PDF WORKER IS CHECKED BEFORE flag.Parse, AND THAT ORDER IS LOAD BEARING. Checking it after
// meant the flag package saw an unknown flag, printed its usage, and USAGE PRINTS DEFAULTS. One of those
// defaults was the migrator DSN read from the environment, so a single unrecognised flag disclosed a live
// password into the log. See the DSN handling below, which no longer allows that to happen at all.
if len(os.Args) > 1 && os.Args[1] == pdfWorkerFlag {
selfLimit()
os.Exit(runPDFTextWorker())
}
// THE DSN IS NOT A FLAG DEFAULT. flag usage output prints every default, so a DSN default means any usage
// error, any typo, any unknown flag, discloses a live credential. It is read from the environment AFTER
// parsing instead, and the help text names the variable rather than showing its value.
dsn := flag.String("dsn", "", "migrator DSN, defaults to MM_MIGRATOR_DATABASE_URL")
limit := flag.Int("limit", 20, "how many pending documents to process in this run")
one := flag.String("document", "", "process exactly one document id, ignoring extraction_status")
dryRun := flag.Bool("dry-run", false, "extract and report, write nothing")
storeRoot := flag.String("storage-root", "", "local object storage root, defaults to MM_STORAGE_ROOT")
// flag.Parse WAS DELETED BY AN EARLIER EDIT OF MINE and this is the restoration. Removing a duplicated
// worker check block took the parse call and the environment fallbacks with it, so every flag stayed at its
// zero value and the DSN check below fatalled on every run. The binary was DEPLOYED in that state.
//
// IT FAILED LOUDLY, WHICH IS THE ONLY REASON IT WAS CAUGHT WITHIN MINUTES: "-dsn or
// MM_MIGRATOR_DATABASE_URL is required" on a host where that variable was plainly set. A binary that had
// instead defaulted to something plausible would have run against the wrong target silently.
flag.Parse()
if *dsn == "" {
*dsn = os.Getenv("MM_MIGRATOR_DATABASE_URL")
}
if *storeRoot == "" {
*storeRoot = os.Getenv("MM_STORAGE_ROOT")
}
if *dsn == "" {
log.Fatal("opportunity-extract: -dsn or MM_MIGRATOR_DATABASE_URL is required")
}
if *storeRoot == "" {
log.Fatal("opportunity-extract: -storage-root or MM_STORAGE_ROOT is required, the bytes live there")
}
store, err := storage.NewLocal(*storeRoot)
if err != nil {
log.Fatalf("opportunity-extract: storage: %v", err)
}
ctx := context.Background()
pool, err := pgxpool.New(ctx, *dsn)
if err != nil {
log.Fatalf("opportunity-extract: pool: %v", err)
}
defer pool.Close()
docs, err := selectDocuments(ctx, pool, *one, *limit)
if err != nil {
log.Fatalf("opportunity-extract: select: %v", err)
}
if len(docs) == 0 {
log.Print("opportunity-extract: nothing pending")
return
}
log.Printf("opportunity-extract: %d document(s), dry_run=%t", len(docs), *dryRun)
var okCount, failCount, chunkCount int
for _, d := range docs {
n, reason, err := extractOne(ctx, pool, store, d, *dryRun)
switch {
case err != nil:
// An UNEXPECTED error still does not stop the batch. It is logged loudly and the document is left
// pending so the next run retries it, which is different from a document that is definitively bad.
log.Printf(" document %s: UNEXPECTED %v, left pending for retry", d.ID, err)
failCount++
case reason != "":
log.Printf(" document %s (%s): FAILED %s", d.ID, d.FileName, reason)
failCount++
default:
log.Printf(" document %s (%s): %d chunks", d.ID, d.FileName, n)
okCount++
chunkCount += n
}
}
log.Printf("opportunity-extract: DONE ok=%d failed=%d chunks=%d", okCount, failCount, chunkCount)
}
type doc struct {
OrgID, ID, FileName, MediaType, StorageKey string
SizeBytes int64
}
func selectDocuments(ctx context.Context, pool *pgxpool.Pool, one string, limit int) ([]doc, error) {
q := `SELECT organization_id::text, id::text, file_name, media_type, storage_key, size_bytes
FROM opportunity_documents
WHERE extraction_status = 'pending' AND is_current
ORDER BY created_at LIMIT $1`
args := []any{limit}
if strings.TrimSpace(one) != "" {
q = `SELECT organization_id::text, id::text, file_name, media_type, storage_key, size_bytes
FROM opportunity_documents WHERE id = $1`
args = []any{strings.TrimSpace(one)}
}
rows, err := pool.Query(ctx, q, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var out []doc
for rows.Next() {
var d doc
if e := rows.Scan(&d.OrgID, &d.ID, &d.FileName, &d.MediaType, &d.StorageKey, &d.SizeBytes); e != nil {
return nil, e
}
out = append(out, d)
}
return out, rows.Err()
}
// extractOne returns (chunks written, failure reason, unexpected error). A failure REASON is a document this
// job has decided about; an unexpected ERROR is a document it has not, and the two are kept apart because
// only one of them should be retried forever.
func extractOne(ctx context.Context, pool *pgxpool.Pool, store storage.Storer, d doc, dry bool) (int, string, error) {
if d.SizeBytes > maxExtractBytes {
return 0, errTooLarge, markFailed(ctx, pool, d, errTooLarge, dry)
}
raw, err := store.Get(ctx, d.StorageKey)
if err != nil {
// The row promises bytes that storage does not have. That is a definite state, not a transient one,
// so it is recorded rather than retried until someone notices the log.
return 0, errBytesMissing, markFailed(ctx, pool, d, errBytesMissing, dry)
}
text, reason := textFrom(raw, d.MediaType, d.FileName)
if reason != "" {
return 0, reason, markFailed(ctx, pool, d, reason, dry)
}
chunks := opportunities.ExtractDeterministic(text)
// THE QUALITY GATE, and it exists because the first run against REAL federal PDFs produced sixteen
// chunks of binary noise that looked like data. Inflating every stream picks up font programs and image
// data along with the page text, and a scanner looking for parenthesised strings finds plenty of them
// inside a compressed font. Those chunks stored cleanly, carried correct provenance, and were complete
// garbage.
//
// GARBAGE CHUNKS ARE WORSE THAN NO CHUNKS, because a reviewer cites them and a citation is the one thing
// this pipeline promises. So every chunk must look like prose, and a document that cannot produce enough
// prose is reported as having no text layer rather than being filled with noise.
kept := chunks[:0]
for _, c := range chunks {
if looksLikeProse(c.Text) {
kept = append(kept, c)
}
}
dropped := len(chunks) - len(kept)
chunks = kept
if dropped > 0 {
log.Printf(" %s: dropped %d chunk(s) that did not look like prose", d.FileName, dropped)
}
// TABLES, directive Section 5.1, which lists them alongside headings and numbered clauses. They are read
// by a SECOND, STRUCTURAL reader and appended AFTER the prose gate rather than passed through it.
//
// WHY THEY BYPASS THE GATE RATHER THAN THE GATE BEING LOOSENED. A rate table is rows of codes and numbers
// and legitimately does not read like language, so any threshold low enough to admit it also readmits the
// decompressed font data the gate was written to stop. ExtractTables validates by SHAPE instead, requiring
// gutter separated cells, a content composition that font blobs fail, and no long unbroken runs. The prose
// gate is therefore exactly as strict as it was, and the wage determination stops being invisible.
//
// THIS IS THE DIFFERENCE BETWEEN 1 CHUNK AND THE WHOLE RATE GRID on the single most staffing relevant
// document in a solicitation package: the legal floor under every bill rate in the proposal.
tables := opportunities.ExtractTables(text)
if len(tables) > 0 {
chunks = append(chunks, tables...)
rows := 0
for _, t := range tables {
rows += strings.Count(t.Text, "\n") + 1
}
log.Printf(" %s: %d table(s) recovered, %d row(s)", d.FileName, len(tables), rows)
if rates := opportunities.ExtractRateRows(text); len(rates) > 0 {
log.Printf(" %s: %d labor category rate row(s), for example %s %s at %s",
d.FileName, len(rates), rates[0].Code, rates[0].Title, rates[0].HourlyRate)
}
}
if len(chunks) == 0 {
reason := errNoChunks
if dropped > 0 {
reason = errNoTextLayer
}
return 0, reason, markFailed(ctx, pool, d, reason, dry)
}
if dry {
return len(chunks), "", nil
}
return len(chunks), "", writeChunks(ctx, pool, d, chunks)
}
// textFrom converts stored bytes to text, or says why it cannot. It is deliberately CONSERVATIVE: a format
// this build cannot read honestly is refused by name rather than run through a decoder that emits plausible
// garbage. Garbage chunks are worse than no chunks, because a reviewer cites them.
func textFrom(raw []byte, mediaType, fileName string) (string, string) {
if len(raw) == 0 {
return "", errEmpty
}
switch {
case bytes.HasPrefix(raw, []byte("%PDF")):
t, reason := pdfText(raw)
if reason != "" {
return "", reason
}
// SANITIZE BEFORE RETURNING. Real federal PDFs carry text in encodings that are not UTF-8, and the
// raw bytes reached Postgres and were refused with "invalid byte sequence for encoding UTF8". The
// document was then left pending and retried forever, failing identically every run. Text that
// cannot be stored is not text.
t = toValidUTF8(t)
if strings.TrimSpace(t) == "" {
return "", errNoTextLayer
}
return t, ""
case strings.HasPrefix(mediaType, "text/"), strings.HasSuffix(strings.ToLower(fileName), ".txt"):
if !utf8.Valid(raw) {
return "", errUnsupported
}
t := strings.TrimSpace(string(raw))
if t == "" {
return "", errEmpty
}
return t, ""
case bytes.HasPrefix(raw, []byte("PK\x03\x04")):
// docx, xlsx, and a plain zip all start this way. This build has no office format reader, and saying
// so is more useful than emitting the XML skeleton as if it were prose.
return "", errUnsupported
default:
if utf8.Valid(raw) && printableRatio(raw) > 0.85 {
return strings.TrimSpace(string(raw)), ""
}
return "", errUnsupported
}
}
// pdfText extracts a PDF's text layer.
//
// It uses github.com/ledongthuc/pdf, chosen by measured bake off against rsc.io/pdf and pdfcpu over 29 real
// federal PDFs; see docs/STACK_DECISIONS.md for the numbers. The hand rolled reader that preceded it is
// KEPT as a fallback, because it costs nothing and it handles the simple uncompressed case that generated
// documents produce, and because a second opinion is useful when the first returns nothing.
//
// EVERY CALL IS PANIC CONTAINED, mandatory regardless of which library won. Two of the three candidates
// panicked on real input, and a batch job that dies on one bad file takes every later document with it.
func pdfText(raw []byte) (string, string) {
// PERMISSION RESTRICTED IS NOT ENCRYPTED, and conflating them refused four of the first nine real
// federal documents this pipeline ever saw. An OWNER password restricts printing or editing and the
// text is readable by anyone; a USER password actually encrypts the content. Nearly every agency
// publishes with the former. So the library is given its chance FIRST, and /Encrypt only becomes a
// verdict when the reader also cannot get text out.
if t, ok := pdfTextViaLibrary(raw); ok {
t = toValidUTF8(t)
if strings.TrimSpace(t) != "" {
return t, ""
}
}
if bytes.Contains(raw, []byte("/Encrypt")) {
return "", errEncrypted
}
return pdfTextFallback(raw)
}
// pdfTextViaLibrary returns (text, true) when the reader completed, whether or not it found anything, and
// (empty, false) when it panicked or refused the file.
// pdfTextViaLibraryInProcess is the ORIGINAL reader, now only ever called INSIDE the isolated worker. It keeps
// its panic recovery, because a panic is still a fact about the file and is cheaper to survive than a restart.
func pdfTextViaLibraryInProcess(raw []byte) (text string, ok bool) {
defer func() {
if rec := recover(); rec != nil {
// A panic is a fact about the file, recorded and survived. It is deliberately NOT fatal and
// deliberately NOT silent.
log.Printf(" pdf reader panicked, falling back: %v", rec)
text, ok = "", false
}
}()
rd, err := pdflib.NewReader(bytes.NewReader(raw), int64(len(raw)))
if err != nil {
return "", false
}
var b strings.Builder
for i := 1; i <= rd.NumPage(); i++ {
p := rd.Page(i)
if p.V.IsNull() {
continue
}
// GetPlainText preserves the document's own spacing. Joining positioned glyph runs by hand and
// adding a space after each one turns "Delivery" into "D e l i v e r y", which reads as garbage to
// the prose gate. That mistake was made in the bake off harness and is not repeated here.
if txt, e := p.GetPlainText(nil); e == nil {
b.WriteString(txt)
b.WriteString("\n")
continue
}
rows, e := p.GetTextByRow()
if e != nil {
continue
}
for _, row := range rows {
for _, w := range row.Content {
b.WriteString(w.S)
}
b.WriteString("\n")
}
}
return b.String(), true
}
// pdfTextFallback is the original hand rolled reader, kept for the simple uncompressed case.
//
// ITS LIMITS ARE THE POINT AND ARE REPORTED, NOT HIDDEN. It reads only uncompressed streams, so a PDF whose
// content is Flate compressed yields nothing and is reported as no_text_layer rather than as an empty
// document. An ENCRYPTED pdf is detected and named. A scanned pdf has no text layer at all and is named too.
// Each of those is a different problem for the person holding the file, and collapsing them into "failed"
// would waste their time.
func pdfTextFallback(raw []byte) (string, string) {
// MOST REAL PDFs COMPRESS THEIR CONTENT STREAMS. The first version read only uncompressed streams and
// reported no_text_layer for two of the first four genuine federal attachments it ever saw, which would
// have meant the pipeline silently covered the documents nobody sends and missed the ones everybody
// does. FlateDecode is zlib and is in the standard library, so the streams are inflated and appended to
// what is scanned. A stream that will not inflate is skipped rather than fatal: one bad object should
// not lose a whole document.
inflated, budgetBlown := inflateStreams(raw)
if budgetBlown {
// THE AGGREGATE BUDGET WAS EXHAUSTED, so what came back is TRUNCATED. It is refused rather than
// scanned, because a partial extraction that looks finished is the failure shape this build keeps
// meeting: it stores plausible chunks, reports success, and nobody can tell which pages are missing.
// A named refusal on a 97 KB file that inflates past 48 MB is the honest answer.
return "", errTooLarge
}
raw = append(append([]byte(nil), raw...), inflated...)
var out bytes.Buffer
// Text between BT and ET, taking the literal strings inside parentheses. This is the simple case and it
// is the case a generated or exported PDF usually produces.
for _, seg := range bytes.Split(raw, []byte("BT")) {
end := bytes.Index(seg, []byte("ET"))
if end < 0 {
continue
}
body := seg[:end]
for {
o := bytes.IndexByte(body, '(')
if o < 0 {
break
}
c := indexUnescaped(body[o+1:], ')')
if c < 0 {
break
}
out.Write(unescapePDF(body[o+1 : o+1+c]))
rest := body[o+1+c:]
// LINE STRUCTURE IS RECONSTRUCTED FROM THE POSITIONING OPERATORS, and this is not cosmetic.
// The first version joined every string with a space, so a whole page arrived as ONE line and
// the numbered clause splitter (anchored to a line start) matched nothing: two real PDFs each
// produced a single chunk covering everything. Chunking exists so a scope item can cite C.3.2
// rather than "the document", so losing the line breaks loses the entire point.
//
// Td, TD, T* and the quote operators all move to a new line in PDF text space. If one appears
// before the next string, that is a line break in the source.
if nextOpenParen := bytes.IndexByte(rest, '('); movesToNewLine(rest, nextOpenParen) {
out.WriteByte('\n')
} else {
out.WriteByte(' ')
}
body = rest
}
out.WriteByte('\n')
}
t := strings.TrimSpace(out.String())
if t == "" {
if bytes.Contains(raw, []byte("/Image")) || bytes.Contains(raw, []byte("/DCTDecode")) {
return "", errNoTextLayer
}
if bytes.Contains(raw, []byte("/FlateDecode")) {
// Honest about the reason: the text is there and this reader cannot decompress it.
return "", errNoTextLayer
}
return "", errEmpty
}
return t, ""
}
// movesToNewLine reports whether a text positioning operator appears in b before the next string starts.
// limit is the index of the next open parenthesis, or negative when there is no further string.
func movesToNewLine(b []byte, limit int) bool {
window := b
if limit >= 0 {
window = b[:limit]
}
for _, op := range [][]byte{[]byte(" Td"), []byte(" TD"), []byte("T*"), []byte(" '"), []byte(" \"")} {
if bytes.Contains(window, op) {
return true
}
}
return false
}
// maxInflateTotal bounds the AGGREGATE inflated output across every stream in one document.
//
// THE PER STREAM BOUND WAS NOT ENOUGH AND A 97 KB FILE PROVED IT. Each stream was already limited to
// maxExtractBytes, and the comment beside that limit claimed a decompression bomb was ruled out. It was not:
// the limit was PER STREAM with no ceiling on the sum, and 07_TE6_Deliverables.pdf, 97,642 bytes, drove the
// heap past 512 MB and was OOM KILLED at exit 137. It never printed a verdict and never reached the DONE line,
// so one small file took the whole batch down.
//
// That directly violated this program's own promise, written at the top of this file: a document it cannot
// read is marked failed WITH A REASON while the run continues. Being killed by the kernel is not a reason.
//
// 48 MB is generous against the 32 MB document cap, so no honest solicitation reaches it, while a ratio like
// this one's five thousand to one stops immediately.
const maxInflateTotal = 48 << 20
// inflateStreams returns the concatenated inflate of every FlateDecode stream in the file, and reports whether
// it stopped early because the aggregate budget was exhausted. A truncated result is NOT returned as if it were
// complete: the caller refuses the document, because a partial extraction that looks finished is worse than a
// named failure.
func inflateStreams(raw []byte) ([]byte, bool) {
var out bytes.Buffer
rest := raw
for {
if out.Len() >= maxInflateTotal {
return out.Bytes(), true
}
i := bytes.Index(rest, []byte("stream"))
if i < 0 {
break
}
body := rest[i+len("stream"):]
for len(body) > 0 && (body[0] == '\r' || body[0] == '\n') {
body = body[1:]
}
e := bytes.Index(body, []byte("endstream"))
if e < 0 {
break
}
zr, err := zlib.NewReader(bytes.NewReader(body[:e]))
if err == nil {
// Bounded per stream AND against the remaining aggregate budget, so the sum cannot run away.
remaining := int64(maxInflateTotal - out.Len())
if dec, rerr := io.ReadAll(io.LimitReader(zr, remaining)); rerr == nil {
out.Write(dec)
out.WriteByte(0x0a)
}
zr.Close()
}
rest = body[e:]
}
return out.Bytes(), false
}
// toValidUTF8 drops byte sequences that are not valid UTF-8 and collapses control characters, so extracted
// text is storable. It replaces rather than rejects, because a page with one bad glyph is still evidence.
func toValidUTF8(s string) string {
s = strings.ToValidUTF8(s, "")
return strings.Map(func(r rune) rune {
if r == '\n' || r == '\t' {
return r
}
if unicode.IsControl(r) {
return -1
}
return r
}, s)
}
func indexUnescaped(b []byte, target byte) int {
for i := 0; i < len(b); i++ {
if b[i] == '\\' {
i++
continue
}
if b[i] == target {
return i
}
}
return -1
}
func unescapePDF(b []byte) []byte {
out := make([]byte, 0, len(b))
for i := 0; i < len(b); i++ {
if b[i] == '\\' && i+1 < len(b) {
i++
switch b[i] {
case 'n':
out = append(out, '\n')
case 'r':
out = append(out, '\r')
case 't':
out = append(out, '\t')
default:
out = append(out, b[i])
}
continue
}
out = append(out, b[i])
}
return out
}
// looksLikeProse decides whether an extracted span is text a human wrote or bytes that happened to sit
// between parentheses. It is deliberately blunt: real solicitation prose is overwhelmingly letters, digits,
// spaces, and ordinary punctuation, and decompressed font or image data is not.
func looksLikeProse(s string) bool {
t := strings.TrimSpace(s)
if utf8.RuneCountInString(t) < 12 {
return false
}
// A READABLE WORD IS REQUIRED, and this was added after the real SCA wage determination was re extracted.
// That PDF's text layer decodes to character soup because its embedded font carries no usable character
// map, and one span of that soup scored well enough on letter and space ratios to pass every test below.
// It was stored, and a stored garbage chunk is the exact thing this gate exists to prevent, because a
// reviewer cites it.
//
// The bar is deliberately the weakest thing that separates language from soup: somewhere in the span there
// is a run of at least four letters containing a vowel. Every sentence of English clears it. "%* +"% !"!,"
// does not. It cannot cost a legitimate document anything, which is why it is safe to add here rather than
// only in the table reader.
if !hasReadableWord(t) {
return false
}
var letters, spaces, digits, total, runs int
var longestWord, current int
for _, r := range t {
total++
switch {
case unicode.IsLetter(r):
letters++
current++
if current > longestWord {
longestWord = current
}
case unicode.IsSpace(r):
spaces++
current = 0
runs++
case unicode.IsDigit(r):
digits++
current = 0
case strings.ContainsRune(".,;:()/-'\"%$&+#", r):
current = 0
default:
// Anything else is a symbol that prose uses rarely and binary uses constantly.
current = 0
runs--
}
}
if total == 0 {
return false
}
letterRatio := float64(letters) / float64(total)
spaceRatio := float64(spaces) / float64(total)
digitRatio := float64(digits) / float64(total)
// PROSE, the original case: mostly letters, broken by spaces, no implausibly long unbroken runs, which
// is what a base64 like blob looks like once symbols are stripped.
if letterRatio >= 0.55 && spaceRatio >= 0.08 && spaceRatio <= 0.35 && longestWord <= 30 {
return true
}
// TABULAR CONTENT, added after a wage determination was refused entirely. A rate table is short rows of
// labor category names and numbers: fewer letters, more digits, still spaced like language and still
// free of the long unbroken runs that mark decompressed font data. It is the most staffing relevant
// document in a solicitation, so refusing all of it was worse than the noise the gate exists to stop.
// Font noise fails this too: it has almost no digits AND almost no spacing.
return letterRatio+digitRatio >= 0.60 && digitRatio >= 0.05 &&
spaceRatio >= 0.05 && spaceRatio <= 0.45 && longestWord <= 30
}
func printableRatio(b []byte) float64 {
if len(b) == 0 {
return 0
}
n := 0
for _, r := range string(b) {
if unicode.IsPrint(r) || unicode.IsSpace(r) {
n++
}
}
return float64(n) / float64(utf8.RuneCount(b))
}
// writeChunks inserts the spans and flips the document to done, IN ONE TRANSACTION with the org GUC set.
// Partial extraction output with a document still marked pending is the worst of both states: the next run
// duplicates every chunk.
func writeChunks(ctx context.Context, pool *pgxpool.Pool, d doc, chunks []opportunities.Chunk) error {
tx, err := pool.Begin(ctx)
if err != nil {
return err
}
defer func() { _ = tx.Rollback(ctx) }()
if _, err := tx.Exec(ctx, `SELECT set_config('app.current_org_id', $1, true)`, d.OrgID); err != nil {
return err
}
if _, err := tx.Exec(ctx,
`UPDATE opportunity_documents SET extraction_status='running' WHERE id=$1`, d.ID); err != nil {
return err
}
for _, c := range chunks {
var page any
if c.PageNumber > 0 {
page = c.PageNumber
}
// PROVENANCE IS OMITTED DELIBERATELY. Migration 0081 stamps tenant_upload on this row.
if _, err := tx.Exec(ctx, `
INSERT INTO opportunity_document_chunks
(organization_id, document_id, page_number, section_label, text_content,
content_hash, extraction_method, confidence)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8)`,
d.OrgID, d.ID, page, c.SectionLabel, c.Text, c.ContentHash, c.Method, c.Confidence); err != nil {
return err
}
}
if _, err := tx.Exec(ctx,
`UPDATE opportunity_documents SET extraction_status='done', extraction_error_code='' WHERE id=$1`,
d.ID); err != nil {
return err
}
return tx.Commit(ctx)
}
func markFailed(ctx context.Context, pool *pgxpool.Pool, d doc, reason string, dry bool) error {
if dry {
return nil
}
c, cancel := context.WithTimeout(ctx, 15*time.Second)
defer cancel()
tx, err := pool.Begin(c)
if err != nil {
return err
}
defer func() { _ = tx.Rollback(c) }()
if _, err := tx.Exec(c, `SELECT set_config('app.current_org_id', $1, true)`, d.OrgID); err != nil {
return err
}
if _, err := tx.Exec(c,
`UPDATE opportunity_documents SET extraction_status='failed', extraction_error_code=$2 WHERE id=$1`,
d.ID, reason); err != nil {
return err
}
return tx.Commit(c)
}
// hasReadableWord reports whether a span contains at least one run of four or more letters that includes a
// vowel. It is the minimum evidence that bytes are language rather than mis mapped character codes.
func hasReadableWord(s string) bool {
run, vowels := 0, 0
for _, r := range strings.ToLower(s) {
if r >= 'a' && r <= 'z' {
run++
if strings.ContainsRune("aeiouy", r) {
vowels++
}
if run >= 4 && vowels > 0 {
return true
}
continue
}
run, vowels = 0, 0
}
return false
}