From 899c504d1f3b590cc742233e7699b0465d64238f Mon Sep 17 00:00:00 2001 From: root Date: Thu, 9 Apr 2026 21:19:43 +0500 Subject: [PATCH] feat(import): move PDF parsing to backend; fix heartbeat/reap for import_tasks - parsePDF function restored in import.go (body was orphaned outside function) - ParseImportFile() called at upload time with 3-min timeout; chapters stored as JSON in MinIO - runner.go: prefer ChaptersKey path (read pre-parsed JSON) over BookImport.Import() - ImportChapterStore interface added; store wired in runner/main.go - HeartbeatTask and ReapStaleTasks now include import_tasks collection - parseImportTask now returns ChaptersKey in domain.ImportTask - asynq_runner.go handleImportTask passes ChaptersKey - pb-init-v3.sh: chapters_key field added to import_tasks schema --- backend/cmd/runner/main.go | 31 ++++----- backend/internal/asynqqueue/producer.go | 11 ++-- backend/internal/asynqqueue/tasks.go | 11 ++-- backend/internal/backend/handlers_import.go | 34 +++++++++- backend/internal/bookstore/bookstore.go | 4 ++ backend/internal/domain/domain.go | 1 + backend/internal/runner/asynq_runner.go | 9 +-- backend/internal/runner/runner.go | 69 ++++++++++++++------- backend/internal/storage/import.go | 34 ++++++++++ backend/internal/storage/store.go | 28 +++++++-- scripts/pb-init-v3.sh | 1 + 11 files changed, 174 insertions(+), 59 deletions(-) diff --git a/backend/cmd/runner/main.go b/backend/cmd/runner/main.go index 070c9fc..b4cd4f6 100644 --- a/backend/cmd/runner/main.go +++ b/backend/cmd/runner/main.go @@ -192,21 +192,22 @@ func run() error { deps := runner.Dependencies{ Consumer: consumer, - BookWriter: store, - BookReader: store, - AudioStore: store, - CoverStore: store, - TranslationStore: store, - BookImport: storage.NewBookImporter(store), - ChapterIngester: store, - SearchIndex: searchIndex, - Novel: novel, - Kokoro: kokoroClient, - PocketTTS: pocketTTSClient, - CFAI: cfaiClient, - LibreTranslate: ltClient, - Notifier: store, - Log: log, + BookWriter: store, + BookReader: store, + AudioStore: store, + CoverStore: store, + TranslationStore: store, + BookImport: storage.NewBookImporter(store), + ImportChapterStore: store, + ChapterIngester: store, + SearchIndex: searchIndex, + Novel: novel, + Kokoro: kokoroClient, + PocketTTS: pocketTTSClient, + CFAI: cfaiClient, + LibreTranslate: ltClient, + Notifier: store, + Log: log, } r := runner.New(rCfg, deps) diff --git a/backend/internal/asynqqueue/producer.go b/backend/internal/asynqqueue/producer.go index b9478c7..ad099a2 100644 --- a/backend/internal/asynqqueue/producer.go +++ b/backend/internal/asynqqueue/producer.go @@ -96,11 +96,12 @@ func (p *Producer) CreateImportTask(ctx context.Context, task domain.ImportTask) } payload := ImportPayload{ - PBTaskID: id, - Slug: task.Slug, - Title: task.Title, - FileType: task.FileType, - ObjectKey: task.ObjectKey, + PBTaskID: id, + Slug: task.Slug, + Title: task.Title, + FileType: task.FileType, + ObjectKey: task.ObjectKey, + ChaptersKey: task.ChaptersKey, } if err := p.enqueue(ctx, TypeImportBook, payload); err != nil { // Non-fatal: PB record exists; runner will pick it up on next poll. diff --git a/backend/internal/asynqqueue/tasks.go b/backend/internal/asynqqueue/tasks.go index 367b459..5c2c709 100644 --- a/backend/internal/asynqqueue/tasks.go +++ b/backend/internal/asynqqueue/tasks.go @@ -48,9 +48,10 @@ type ScrapePayload struct { // ImportPayload is the Asynq job payload for PDF/EPUB import tasks. type ImportPayload struct { - PBTaskID string `json:"pb_task_id"` - Slug string `json:"slug"` - Title string `json:"title"` - FileType string `json:"file_type"` // "pdf" or "epub" - ObjectKey string `json:"object_key"` // MinIO path to uploaded file + PBTaskID string `json:"pb_task_id"` + Slug string `json:"slug"` + Title string `json:"title"` + FileType string `json:"file_type"` // "pdf" or "epub" + ObjectKey string `json:"object_key"` // MinIO path to uploaded file + ChaptersKey string `json:"chapters_key"` // MinIO path to pre-parsed chapters JSON } diff --git a/backend/internal/backend/handlers_import.go b/backend/internal/backend/handlers_import.go index aa84e29..0747525 100644 --- a/backend/internal/backend/handlers_import.go +++ b/backend/internal/backend/handlers_import.go @@ -1,6 +1,7 @@ package backend import ( + "context" "encoding/json" "fmt" "io" @@ -45,6 +46,8 @@ func (s *Server) handleAdminImport(w http.ResponseWriter, r *http.Request) { ct := r.Header.Get("Content-Type") var req importRequest var objectKey string + var chaptersKey string + var chapterCount int if strings.HasPrefix(ct, "multipart/form-data") { if err := r.ParseMultipartForm(32 << 20); err != nil { @@ -96,7 +99,20 @@ func (s *Server) handleAdminImport(w http.ResponseWriter, r *http.Request) { return } - // Upload to MinIO for actual import + // Parse PDF/EPUB on the backend (with timeout) and store chapters as JSON. + // The runner only needs to ingest pre-parsed chapters — no PDF parsing on runner. + parseCtx, parseCancel := context.WithTimeout(r.Context(), 3*time.Minute) + defer parseCancel() + chapters, parseErr := storage.ParseImportFile(parseCtx, data, req.FileType) + if parseErr != nil || len(chapters) == 0 { + jsonError(w, http.StatusUnprocessableEntity, "could not parse file: "+func() string { + if parseErr != nil { return parseErr.Error() } + return "no chapters found" + }()) + return + } + + // Store raw file in MinIO (for reference/re-import). objectKey = fmt.Sprintf("imports/%d_%s", time.Now().Unix(), header.Filename) if s.deps.ImportFileStore == nil { jsonError(w, http.StatusInternalServerError, "storage not available") @@ -106,6 +122,15 @@ func (s *Server) handleAdminImport(w http.ResponseWriter, r *http.Request) { jsonError(w, http.StatusInternalServerError, "upload file: "+err.Error()) return } + + // Store pre-parsed chapters JSON in MinIO so runner can ingest without re-parsing. + chaptersJSON, _ := json.Marshal(chapters) + chaptersKey = fmt.Sprintf("imports/%d_%s_chapters.json", time.Now().Unix(), strings.TrimSuffix(header.Filename, filepath.Ext(header.Filename))) + if err := s.deps.ImportFileStore.PutImportChapters(r.Context(), chaptersKey, chaptersJSON); err != nil { + jsonError(w, http.StatusInternalServerError, "store chapters: "+err.Error()) + return + } + chapterCount = len(chapters) } else { if err := json.NewDecoder(r.Body).Decode(&req); err != nil { jsonError(w, http.StatusBadRequest, "parse body: "+err.Error()) @@ -141,6 +166,8 @@ func (s *Server) handleAdminImport(w http.ResponseWriter, r *http.Request) { BookStatus: req.BookStatus, FileType: req.FileType, ObjectKey: objectKey, + ChaptersKey: chaptersKey, + ChaptersTotal: chapterCount, InitiatorUserID: "", }) if err != nil { @@ -149,8 +176,9 @@ func (s *Server) handleAdminImport(w http.ResponseWriter, r *http.Request) { } writeJSON(w, 0, importResponse{ - TaskID: taskID, - Slug: slug, + TaskID: taskID, + Slug: slug, + Preview: &importPreview{Chapters: chapterCount}, }) } diff --git a/backend/internal/bookstore/bookstore.go b/backend/internal/bookstore/bookstore.go index 3d49bcc..e549790 100644 --- a/backend/internal/bookstore/bookstore.go +++ b/backend/internal/bookstore/bookstore.go @@ -221,4 +221,8 @@ type BookImporter interface { // without a concrete type assertion, regardless of which Producer is wired. type ImportFileStore interface { PutImportFile(ctx context.Context, objectKey string, data []byte) error + // PutImportChapters stores the pre-parsed chapters JSON under the given key. + PutImportChapters(ctx context.Context, key string, data []byte) error + // GetImportChapters retrieves the pre-parsed chapters JSON. + GetImportChapters(ctx context.Context, key string) ([]byte, error) } diff --git a/backend/internal/domain/domain.go b/backend/internal/domain/domain.go index c7b509e..301cb2d 100644 --- a/backend/internal/domain/domain.go +++ b/backend/internal/domain/domain.go @@ -178,6 +178,7 @@ type ImportTask struct { FileName string `json:"file_name"` FileType string `json:"file_type"` // "pdf" or "epub" ObjectKey string `json:"object_key,omitempty"` // MinIO path to uploaded file + ChaptersKey string `json:"chapters_key,omitempty"` // MinIO path to pre-parsed chapters JSON Author string `json:"author,omitempty"` CoverURL string `json:"cover_url,omitempty"` Genres []string `json:"genres,omitempty"` diff --git a/backend/internal/runner/asynq_runner.go b/backend/internal/runner/asynq_runner.go index cb5d723..643fcf6 100644 --- a/backend/internal/runner/asynq_runner.go +++ b/backend/internal/runner/asynq_runner.go @@ -199,10 +199,11 @@ func (r *Runner) handleImportTask(ctx context.Context, t *asynq.Task) error { return fmt.Errorf("unmarshal import payload: %w", err) } task := domain.ImportTask{ - ID: p.PBTaskID, - Slug: p.Slug, - Title: p.Title, - FileType: p.FileType, + ID: p.PBTaskID, + Slug: p.Slug, + Title: p.Title, + FileType: p.FileType, + ChaptersKey: p.ChaptersKey, } r.tasksRunning.Add(1) defer r.tasksRunning.Add(-1) diff --git a/backend/internal/runner/runner.go b/backend/internal/runner/runner.go index 16da7c6..7575f9b 100644 --- a/backend/internal/runner/runner.go +++ b/backend/internal/runner/runner.go @@ -15,6 +15,7 @@ package runner import ( "context" + "encoding/json" "fmt" "log/slog" "os" @@ -49,6 +50,11 @@ type ChapterIngester interface { IngestChapters(ctx context.Context, slug string, chapters []bookstore.Chapter) error } +// ImportChapterStore retrieves pre-parsed chapter JSON blobs from object storage. +type ImportChapterStore interface { + GetImportChapters(ctx context.Context, key string) ([]byte, error) +} + // Config tunes the runner behaviour. type Config struct { // WorkerID uniquely identifies this runner instance in PocketBase records. @@ -114,7 +120,12 @@ type Dependencies struct { // CoverStore stores book cover images in MinIO. CoverStore bookstore.CoverStore // BookImport handles PDF/EPUB file parsing and chapter extraction. + // Kept for backward compatibility when ChaptersKey is not set. BookImport bookstore.BookImporter + // ImportChapterStore retrieves pre-parsed chapter JSON blobs from MinIO. + // When set and the task has a ChaptersKey, the runner reads from here + // instead of calling BookImport.Import() (the new preferred path). + ImportChapterStore ImportChapterStore // ChapterIngester persists extracted chapters into MinIO/PocketBase. ChapterIngester ChapterIngester // Notifier creates notifications for users. @@ -675,6 +686,10 @@ func (r *Runner) runAudioTask(ctx context.Context, task domain.AudioTask) { } // runImportTask executes one PDF/EPUB import task. +// Preferred path: when task.ChaptersKey is set, it reads pre-parsed chapters +// JSON from MinIO (written by the backend at upload time) and ingests them. +// Fallback path: when ChaptersKey is empty, calls BookImport.Import() to +// parse the raw file on the runner (legacy behaviour, not used for new tasks). func (r *Runner) runImportTask(ctx context.Context, task domain.ImportTask, objectKey string) { ctx, span := otel.Tracer("runner").Start(ctx, "runner.import_task") defer span.End() @@ -682,10 +697,11 @@ func (r *Runner) runImportTask(ctx context.Context, task domain.ImportTask, obje attribute.String("task.id", task.ID), attribute.String("book.slug", task.Slug), attribute.String("file.type", task.FileType), + attribute.String("chapters_key", task.ChaptersKey), ) log := r.deps.Log.With("task_id", task.ID, "slug", task.Slug, "file_type", task.FileType) - log.Info("runner: import task starting") + log.Info("runner: import task starting", "chapters_key", task.ChaptersKey) hbCtx, hbCancel := context.WithCancel(ctx) defer hbCancel() @@ -714,15 +730,33 @@ func (r *Runner) runImportTask(ctx context.Context, task domain.ImportTask, obje } } - if r.deps.BookImport == nil { - fail("book import not configured (BookImport dependency missing)") - return - } + var chapters []bookstore.Chapter - chapters, err := r.deps.BookImport.Import(ctx, objectKey, task.FileType) - if err != nil { - fail(fmt.Sprintf("import file: %v", err)) - return + if task.ChaptersKey != "" && r.deps.ImportChapterStore != nil { + // New path: read pre-parsed chapters JSON uploaded by the backend. + raw, err := r.deps.ImportChapterStore.GetImportChapters(ctx, task.ChaptersKey) + if err != nil { + fail(fmt.Sprintf("get chapters JSON: %v", err)) + return + } + if err := json.Unmarshal(raw, &chapters); err != nil { + fail(fmt.Sprintf("unmarshal chapters JSON: %v", err)) + return + } + log.Info("runner: loaded pre-parsed chapters", "count", len(chapters)) + } else { + // Legacy path: parse the raw file on the runner. + if r.deps.BookImport == nil { + fail("book import not configured (BookImport dependency missing)") + return + } + var err error + chapters, err = r.deps.BookImport.Import(ctx, objectKey, task.FileType) + if err != nil { + fail(fmt.Sprintf("import file: %v", err)) + return + } + log.Info("runner: parsed chapters from file (legacy path)", "count", len(chapters)) } if len(chapters) == 0 { @@ -730,23 +764,12 @@ func (r *Runner) runImportTask(ctx context.Context, task domain.ImportTask, obje return } - // Store chapters via BookWriter - // Note: BookWriter.WriteChapters expects domain.Chapter, need conversion - var domainChapters []bookstore.Chapter - for _, ch := range chapters { - domainChapters = append(domainChapters, bookstore.Chapter{ - Number: ch.Number, - Title: ch.Title, - Content: ch.Content, - }) - } - - // Store chapters via ChapterIngester + // Persist chapters via ChapterIngester. if r.deps.ChapterIngester == nil { fail("chapter ingester not configured") return } - if err := r.deps.ChapterIngester.IngestChapters(ctx, task.Slug, domainChapters); err != nil { + if err := r.deps.ChapterIngester.IngestChapters(ctx, task.Slug, chapters); err != nil { fail(fmt.Sprintf("store chapters: %v", err)) return } @@ -786,7 +809,7 @@ func (r *Runner) runImportTask(ctx context.Context, task domain.ImportTask, obje log.Error("runner: FinishImportTask failed", "err", err) } - // Create notification for the user who initiated the import + // Notify the user who initiated the import. if r.deps.Notifier != nil { msg := fmt.Sprintf("Import completed: %d chapters from %s", len(chapters), task.Title) targetUser := task.InitiatorUserID diff --git a/backend/internal/storage/import.go b/backend/internal/storage/import.go index 1a32964..f0b387b 100644 --- a/backend/internal/storage/import.go +++ b/backend/internal/storage/import.go @@ -109,6 +109,40 @@ func decryptPDF(data []byte) ([]byte, error) { return out.Bytes(), nil } +// ParseImportFile parses a PDF or EPUB and returns chapters. +// Unlike AnalyzeFile it respects ctx cancellation so callers can apply a timeout. +// For PDFs it first attempts to strip encryption with an empty password. +func ParseImportFile(ctx context.Context, data []byte, fileType string) ([]bookstore.Chapter, error) { + type result struct { + chapters []bookstore.Chapter + err error + } + ch := make(chan result, 1) + go func() { + var chapters []bookstore.Chapter + var err error + switch fileType { + case "pdf": + chapters, err = parsePDF(data) + case "epub": + chapters, err = parseEPUB(data) + default: + err = fmt.Errorf("unsupported file type: %s", fileType) + } + ch <- result{chapters, err} + }() + select { + case <-ctx.Done(): + return nil, fmt.Errorf("parse timed out: %w", ctx.Err()) + case r := <-ch: + return r.chapters, r.err + } +} + +// parsePDF extracts chapters from PDF bytes using dslipak/pdf. +// It first attempts to decrypt the PDF with an empty password in case the file +// uses owner-only encryption (copy/print restrictions), which is common for +// publisher PDFs that open normally in readers but confuse raw parsers. func parsePDF(data []byte) ([]bookstore.Chapter, error) { // If the PDF is encrypted, try to decrypt it with an empty password. // Many publisher PDFs use owner-only encryption (copy/print restrictions) diff --git a/backend/internal/storage/store.go b/backend/internal/storage/store.go index 8ad02d2..e813f9a 100644 --- a/backend/internal/storage/store.go +++ b/backend/internal/storage/store.go @@ -654,13 +654,14 @@ func (s *Store) CreateImportTask(ctx context.Context, task domain.ImportTask) (s "file_name": task.Slug + "." + task.FileType, "file_type": task.FileType, "object_key": task.ObjectKey, + "chapters_key": task.ChaptersKey, "author": task.Author, "cover_url": task.CoverURL, "summary": task.Summary, "book_status": task.BookStatus, "status": string(domain.TaskStatusPending), "chapters_done": 0, - "chapters_total": 0, + "chapters_total": task.ChaptersTotal, "started": time.Now().UTC().Format(time.RFC3339), "initiator_user_id": task.InitiatorUserID, } @@ -914,7 +915,7 @@ func (s *Store) FailTask(ctx context.Context, id, errMsg string) error { } // HeartbeatTask updates the heartbeat_at field on a running task. -// Tries scraping_tasks first, then audio_jobs, then translation_jobs. +// Tries scraping_tasks, audio_jobs, translation_jobs, then import_tasks. func (s *Store) HeartbeatTask(ctx context.Context, id string) error { payload := map[string]any{ "heartbeat_at": time.Now().UTC().Format(time.RFC3339), @@ -925,7 +926,10 @@ func (s *Store) HeartbeatTask(ctx context.Context, id string) error { if err := s.pb.patch(ctx, fmt.Sprintf("/api/collections/audio_jobs/records/%s", id), payload); err == nil { return nil } - return s.pb.patch(ctx, fmt.Sprintf("/api/collections/translation_jobs/records/%s", id), payload) + if err := s.pb.patch(ctx, fmt.Sprintf("/api/collections/translation_jobs/records/%s", id), payload); err == nil { + return nil + } + return s.pb.patch(ctx, fmt.Sprintf("/api/collections/import_tasks/records/%s", id), payload) } // ReapStaleTasks finds all running tasks whose heartbeat_at is either missing @@ -943,7 +947,7 @@ func (s *Store) ReapStaleTasks(ctx context.Context, staleAfter time.Duration) (i } total := 0 - for _, collection := range []string{"scraping_tasks", "audio_jobs", "translation_jobs"} { + for _, collection := range []string{"scraping_tasks", "audio_jobs", "translation_jobs", "import_tasks"} { items, err := s.pb.listAll(ctx, collection, filter, "") if err != nil { return total, fmt.Errorf("ReapStaleTasks list %s: %w", collection, err) @@ -1185,6 +1189,7 @@ func parseImportTask(raw json.RawMessage) (domain.ImportTask, error) { FileName string `json:"file_name"` FileType string `json:"file_type"` ObjectKey string `json:"object_key"` + ChaptersKey string `json:"chapters_key"` Author string `json:"author"` CoverURL string `json:"cover_url"` Genres string `json:"genres"` // stored as comma-separated @@ -1219,6 +1224,7 @@ func parseImportTask(raw json.RawMessage) (domain.ImportTask, error) { FileName: rec.FileName, FileType: rec.FileType, ObjectKey: rec.ObjectKey, + ChaptersKey: rec.ChaptersKey, Author: rec.Author, CoverURL: rec.CoverURL, Genres: genres, @@ -1266,6 +1272,20 @@ func (s *Store) PutImportFile(ctx context.Context, key string, data []byte) erro return s.mc.putObject(ctx, "imports", key, "application/octet-stream", data) } +// PutImportChapters stores a pre-parsed chapters JSON blob in MinIO. +func (s *Store) PutImportChapters(ctx context.Context, key string, data []byte) error { + return s.mc.putObject(ctx, "imports", key, "application/json", data) +} + +// GetImportChapters retrieves the pre-parsed chapters JSON from MinIO. +func (s *Store) GetImportChapters(ctx context.Context, key string) ([]byte, error) { + data, err := s.mc.getObject(ctx, "imports", key) + if err != nil { + return nil, fmt.Errorf("get chapters object: %w", err) + } + return data, nil +} + func (s *Store) CoverExists(ctx context.Context, slug string) bool { return s.mc.coverExists(ctx, CoverObjectKey(slug)) } diff --git a/scripts/pb-init-v3.sh b/scripts/pb-init-v3.sh index 308fd1d..2647565 100755 --- a/scripts/pb-init-v3.sh +++ b/scripts/pb-init-v3.sh @@ -306,6 +306,7 @@ create "import_tasks" '{ {"name":"file_name", "type":"text"}, {"name":"file_type", "type":"text"}, {"name":"object_key", "type":"text"}, + {"name":"chapters_key", "type":"text"}, {"name":"author", "type":"text"}, {"name":"cover_url", "type":"text"}, {"name":"genres", "type":"text"},