jacob-mathison/query-orchestration
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

merge main

Browse Source
This commit is contained in:
jay brown
2025-04-22 12:09:51 -07:00
parent b23b17ac53 fee71e7740
commit f28561dac6
404 changed files with 211048 additions and 2821 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/clientSyncRunner/main.go
+1 -1
View File
@@ -31,7 +31,7 @@ func main() {
cfg.ControllerFunc = func() runner.Controller[clientsyncrunner.Body] {
svc := clientsync.New(cfg)
c := clientsyncrunner.New(cfg.GetValidator(), &clientsyncrunner.Services{
c := clientsyncrunner.New(&clientsyncrunner.Services{
ClientSync: svc,
})
cmd/docCleanRunner/main.go
+1 -1
View File
@@ -35,7 +35,7 @@ func main() {
cfg.ControllerFunc = func() runner.Controller[doccleanrunner.Body] {
clean := documentclean.New(cfg)
return doccleanrunner.New(cfg.GetValidator(), &doccleanrunner.Services{
return doccleanrunner.New(&doccleanrunner.Services{
Clean: clean,
})
}
cmd/docInitRunner/main.go
+1 -1
View File
@@ -41,7 +41,7 @@ func main() {
cfg.ControllerFunc = func() runner.Controller[docinitrunner.Body] {
docinit := documentinit.New(cfg)
return docinitrunner.New(cfg.GetValidator(), &docinitrunner.Services{
return docinitrunner.New(&docinitrunner.Services{
Document: docinit,
})
}
cmd/docSyncRunner/main.go
+1 -1
View File
@@ -42,7 +42,7 @@ func main() {
Client: cli,
})
return docsyncrunner.New(cfg.GetValidator(), &docsyncrunner.Services{
return docsyncrunner.New(&docsyncrunner.Services{
Document: docsync,
})
}
cmd/docTextProcessRunner/main.go
-66
View File
@@ -1,66 +0,0 @@
// **Listens For**: Body Containing Textract Result Document Metadata
//
// **Action**.
//
// Process textract output.
//
// An event with the document id is pushed to the QUERYSYNC queue.
package main
import (
"context"
"log/slog"
"os"
doctextprocessrunner "queryorchestration/api/docTextProcessRunner"
documenttext "queryorchestration/internal/document/text"
"queryorchestration/internal/server/runner"
"queryorchestration/internal/serviceconfig/aws"
"queryorchestration/internal/serviceconfig/objectstore"
"queryorchestration/internal/serviceconfig/queue/querysync"
"queryorchestration/internal/serviceconfig/textract"
_ "github.com/lib/pq"
)
type DocTextProcessConfig struct {
runner.BaseConfig[doctextprocessrunner.Body]
aws.AWSConfig
textract.TextractConfig
querysync.QuerySyncConfig
objectstore.ObjectStoreConfig
}
func main() {
ctx := context.Background()
cfg := &DocTextProcessConfig{}
cfg.ControllerFunc = func() runner.Controller[doctextprocessrunner.Body] {
text := documenttext.New(cfg)
return doctextprocessrunner.New(cfg.GetValidator(), &doctextprocessrunner.Services{
Text: text,
})
}
server, err := runner.New(ctx, cfg)
if err != nil {
slog.Error(err.Error())
os.Exit(1)
}
err = cfg.SetStoreClient(ctx)
if err != nil {
slog.Error(err.Error())
os.Exit(1)
}
err = cfg.SetTextractClient(ctx)
if err != nil {
slog.Error(err.Error())
os.Exit(1)
}
server.Listen(ctx)
}
cmd/docTextRunner/main.go
+7 -1
View File
@@ -37,7 +37,7 @@ func main() {
cfg.ControllerFunc = func() runner.Controller[doctextrunner.Body] {
text := documenttext.New(cfg)
return doctextrunner.New(cfg.GetValidator(), &doctextrunner.Services{
return doctextrunner.New(&doctextrunner.Services{
Text: text,
})
}
@@ -48,6 +48,12 @@ func main() {
os.Exit(1)
}
err = cfg.SetStoreClient(ctx)
if err != nil {
slog.Error(err.Error())
os.Exit(1)
}
err = cfg.SetTextractClient(ctx)
if err != nil {
slog.Error(err.Error())
cmd/queryRunner/main.go
+1 -1
View File
@@ -45,7 +45,7 @@ func main() {
Sync: sync,
})
c := queryrunner.New(cfg.GetValidator(), &queryrunner.Services{
c := queryrunner.New(&queryrunner.Services{
ResultSet: resset,
})
cmd/querySyncRunner/main.go
+1 -1
View File
@@ -45,7 +45,7 @@ func main() {
ResultSync: sync,
})
c := querysyncrunner.New(cfg.GetValidator(), &querysyncrunner.Services{
c := querysyncrunner.New(&querysyncrunner.Services{
QuerySync: svc,
})
cmd/queryVersionSyncRunner/main.go
+1 -1
View File
@@ -31,7 +31,7 @@ func main() {
cfg.ControllerFunc = func() runner.Controller[queryversionsyncrunner.Body] {
svc := queryversionsync.New(cfg)
c := queryversionsyncrunner.New(cfg.GetValidator(), &queryversionsyncrunner.Services{
c := queryversionsyncrunner.New(&queryversionsyncrunner.Services{
Sync: svc,
})
cmd/storeEventRunner/main.go
+2 -4
View File
@@ -4,7 +4,7 @@
//
// It will identify the type of event and the location of said event, and forward the task accordingly.
//
// It supports the document initialisation and document text processing stages. Events may be pushed to the DOCINIT and DOCTEXTPROCESS queues.
// It supports the document initialisation and document text processing stages. Events may be pushed to the DOCINIT queue.
package main
import (
@@ -16,7 +16,6 @@ import (
documentstore "queryorchestration/internal/document/store"
"queryorchestration/internal/server/runner"
"queryorchestration/internal/serviceconfig/queue/documentinit"
"queryorchestration/internal/serviceconfig/queue/documenttextprocess"
_ "github.com/lib/pq"
)
@@ -24,7 +23,6 @@ import (
type StoreEventConfig struct {
runner.BaseConfig[storeeventrunner.S3EventNotification]
documentinit.DocInitConfig
documenttextprocess.DocTextProcessConfig
}
func main() {
@@ -35,7 +33,7 @@ func main() {
cfg.ControllerFunc = func() runner.Controller[storeeventrunner.S3EventNotification] {
store := documentstore.New(cfg)
return storeeventrunner.New(cfg.GetValidator(), &storeeventrunner.Services{
return storeeventrunner.New(&storeeventrunner.Services{
Store: store,
})
}
cmd/textractGenerator_test/textdetection.go
+110 -17
View File
@@ -9,17 +9,40 @@ import (
"fmt"
"log/slog"
"os"
documenttext "queryorchestration/internal/document/text"
documenttypes "queryorchestration/internal/document/types"
"queryorchestration/internal/serviceconfig"
"queryorchestration/internal/serviceconfig/aws"
"queryorchestration/internal/serviceconfig/objectstore"
"queryorchestration/internal/serviceconfig/queue/querysync"
"queryorchestration/internal/serviceconfig/textract"
awstextract "github.com/aws/aws-sdk-go-v2/service/textract"
"github.com/aws/aws-sdk-go-v2/service/textract/types"
)
type Config struct {
serviceconfig.BaseConfig
textract.TextractConfig
querysync.QuerySyncConfig
objectstore.ObjectStoreConfig
}
func main() {
ctx := context.Background()
cfg := &textract.TextractConfig{}
if len(os.Args) != 3 {
fmt.Println("Error: expected at least two arguments")
fmt.Println("Usage:", os.Args[0], "<sample_name>", "<file_name>")
os.Exit(1)
}
sampleName := os.Args[1]
name := os.Args[2]
slog.Info("arguments", "sample_name", sampleName, "file_name", name)
cfg := &Config{}
profile := aws.Profile(os.Getenv("AWS_PROFILE"))
if profile == "" {
slog.Error("profile not available")
@@ -32,35 +55,105 @@ func main() {
os.Exit(1)
}
base := "../../assets/sampleGeneration"
name := "helloWorld"
filename := fmt.Sprintf("%s/%s.pdf", base, name)
content, err := os.ReadFile(filename)
root, err := os.OpenRoot("assets")
if err != nil {
slog.Error("error writing file", "error", err)
slog.Error("error opening root", "error", err)
os.Exit(1)
}
slog.Info("detecting text", "filename", filename)
res, err := cfg.GetTextractClient().DetectDocumentText(ctx, &awstextract.DetectDocumentTextInput{
Document: &types.Document{
Bytes: content,
},
})
root, err = root.OpenRoot(sampleName)
if err != nil {
slog.Error("error detecting text", "error", err)
slog.Error("error opening sample", "error", err)
os.Exit(1)
}
jsonContent, err := json.MarshalIndent(res, "", "\t")
pdfFile, err := root.Open(fmt.Sprintf("%s.pdf", name))
if err != nil {
slog.Error("error opening pdf file", "error", err)
os.Exit(1)
}
pdf, err := documenttypes.NewPDFFromReader(pdfFile)
if err != nil {
slog.Error("error creating pdf file", "error", err)
os.Exit(1)
}
textSvc := documenttext.New(cfg)
count, err := pdf.GetPageCount(ctx)
if err != nil {
slog.Error("error getting page count", "error", err)
os.Exit(1)
}
slog.Info("detecting text")
allResults := map[string][]awstextract.AnalyzeDocumentOutput{}
allResults["base"] = make([]awstextract.AnalyzeDocumentOutput, count)
allResults["table"] = []awstextract.AnalyzeDocumentOutput{}
for i := range count {
result, err := textSvc.GetBasePage(ctx, pdf, i)
if err != nil {
slog.Error("error detecting text", "error", err)
os.Exit(1)
}
for _, block := range result.Blocks {
feature := textSvc.GetBlockFeature(block)
if feature == documenttext.FeatureTable {
pageBytes, err := pdf.GetPageAsPNG(ctx, i)
if err != nil {
slog.Error("error getting page", "error", err)
os.Exit(1)
}
res, err := cfg.GetTextractClient().AnalyzeDocument(ctx, &awstextract.AnalyzeDocumentInput{
Document: &types.Document{
Bytes: pageBytes,
},
FeatureTypes: []types.FeatureType{
types.FeatureTypeTables,
},
})
if err != nil {
slog.Error("error detecting table", "error", err)
os.Exit(1)
}
allResults["table"] = append(allResults["table"], *res)
}
}
allResults["base"][i] = result
}
generatedBase := "generated"
if _, err := root.Stat(generatedBase); os.IsNotExist(err) {
err := root.Mkdir(generatedBase, 0755)
if err != nil {
slog.Error("error creating directory", "error", err)
os.Exit(1)
}
slog.Info("directory created", "directory", generatedBase)
}
jsonContent, err := json.MarshalIndent(allResults, "", "\t")
if err != nil {
slog.Error("error marshaling json", "error", err)
os.Exit(1)
}
outputFilename := fmt.Sprintf("%s/generated/%s.gen", base, name)
err = os.WriteFile(outputFilename, jsonContent, 0600)
outputFilename := fmt.Sprintf("%s/%s.gen", generatedBase, name)
file, err := root.Create(outputFilename)
if err != nil {
slog.Error("error creating file", "error", err)
os.Exit(1)
}
defer file.Close()
_, err = file.Write(jsonContent)
if err != nil {
slog.Error("error writing file", "error", err)
os.Exit(1)