feat: initial commit
This commit is contained in:
Admin
213
scraper/internal/orchestrator/orchestrator.go
Normal file
213
scraper/internal/orchestrator/orchestrator.go
Normal file
@@ -0,0 +1,213 @@
|
||||
// Package orchestrator coordinates the catalogue walk, metadata extraction,
|
||||
// chapter-list fetching, and parallel chapter scraping.
|
||||
//
|
||||
// Concurrency model
|
||||
// - One goroutine runs ScrapeCatalogue and feeds book URLs into a channel.
|
||||
// - For each book, a dedicated goroutine calls ScrapeMetadata (metadata goroutine).
|
||||
// - ScrapeChapterList is called in the metadata goroutine once metadata is done.
|
||||
// - N worker goroutines (default: runtime.NumCPU()) each pull ChapterRef values
|
||||
// from a shared work queue and call ScrapeChapterText.
|
||||
// - A sync.WaitGroup ensures all chapter workers finish before the orchestrator
|
||||
// signals completion.
|
||||
package orchestrator
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"runtime"
|
||||
"sync"
|
||||
|
||||
"github.com/libnovel/scraper/internal/scraper"
|
||||
"github.com/libnovel/scraper/internal/writer"
|
||||
)
|
||||
|
||||
// Config holds tunable parameters for the orchestrator.
|
||||
type Config struct {
|
||||
// Workers is the number of goroutines used to scrape chapters in parallel.
|
||||
// Defaults to runtime.NumCPU() when 0.
|
||||
Workers int
|
||||
|
||||
// StaticRoot is the path to the static/books output directory.
|
||||
StaticRoot string
|
||||
|
||||
// SingleBookURL when non-empty causes the orchestrator to scrape only
|
||||
// that one book instead of walking the full catalogue.
|
||||
SingleBookURL string
|
||||
}
|
||||
|
||||
// Orchestrator coordinates the full scrape pipeline.
|
||||
type Orchestrator struct {
|
||||
cfg Config
|
||||
novel scraper.NovelScraper
|
||||
writer *writer.Writer
|
||||
log *slog.Logger
|
||||
workers int
|
||||
}
|
||||
|
||||
// New returns a new Orchestrator.
|
||||
func New(cfg Config, novel scraper.NovelScraper, log *slog.Logger) *Orchestrator {
|
||||
workers := cfg.Workers
|
||||
if workers <= 0 {
|
||||
workers = runtime.NumCPU()
|
||||
}
|
||||
return &Orchestrator{
|
||||
cfg: cfg,
|
||||
novel: novel,
|
||||
writer: writer.New(cfg.StaticRoot),
|
||||
log: log,
|
||||
workers: workers,
|
||||
}
|
||||
}
|
||||
|
||||
// Run executes the full scrape pipeline and blocks until it is complete or ctx
|
||||
// is cancelled.
|
||||
func (o *Orchestrator) Run(ctx context.Context) error {
|
||||
o.log.Info("orchestrator starting",
|
||||
"source", o.novel.SourceName(),
|
||||
"workers", o.workers,
|
||||
"static_root", o.cfg.StaticRoot,
|
||||
)
|
||||
|
||||
// chapterWork is the shared queue consumed by chapter worker goroutines.
|
||||
type chapterJob struct {
|
||||
slug string
|
||||
ref scraper.ChapterRef
|
||||
}
|
||||
chapterWork := make(chan chapterJob, o.workers*4)
|
||||
|
||||
// Start chapter worker pool.
|
||||
var chapterWG sync.WaitGroup
|
||||
for i := 0; i < o.workers; i++ {
|
||||
chapterWG.Add(1)
|
||||
go func(workerID int) {
|
||||
defer chapterWG.Done()
|
||||
for job := range chapterWork {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
// Skip if already on disk.
|
||||
if o.writer.ChapterExists(job.slug, job.ref) {
|
||||
o.log.Debug("chapter already exists, skipping",
|
||||
"book", job.slug, "chapter", job.ref.Number)
|
||||
continue
|
||||
}
|
||||
|
||||
chapter, err := o.novel.ScrapeChapterText(ctx, job.ref)
|
||||
if err != nil {
|
||||
o.log.Error("chapter scrape failed",
|
||||
"book", job.slug,
|
||||
"chapter", job.ref.Number,
|
||||
"url", job.ref.URL,
|
||||
"err", err,
|
||||
)
|
||||
continue
|
||||
}
|
||||
|
||||
if err := o.writer.WriteChapter(job.slug, chapter); err != nil {
|
||||
o.log.Error("chapter write failed",
|
||||
"book", job.slug,
|
||||
"chapter", job.ref.Number,
|
||||
"err", err,
|
||||
)
|
||||
continue
|
||||
}
|
||||
|
||||
o.log.Info("chapter saved",
|
||||
"book", job.slug,
|
||||
"chapter", job.ref.Number,
|
||||
"worker", workerID,
|
||||
)
|
||||
}
|
||||
}(i)
|
||||
}
|
||||
|
||||
// processBook scrapes metadata + chapter list for one book, then enqueues
|
||||
// chapter jobs. It is called inside a goroutine per book.
|
||||
processBook := func(bookURL string) {
|
||||
// Metadata goroutine.
|
||||
meta, err := o.novel.ScrapeMetadata(ctx, bookURL)
|
||||
if err != nil {
|
||||
o.log.Error("metadata scrape failed", "url", bookURL, "err", err)
|
||||
return
|
||||
}
|
||||
|
||||
// Persist / update metadata.yaml.
|
||||
if err := o.writer.WriteMetadata(meta); err != nil {
|
||||
o.log.Error("metadata write failed", "slug", meta.Slug, "err", err)
|
||||
// Continue — chapters can still be scraped.
|
||||
}
|
||||
|
||||
o.log.Info("metadata saved", "slug", meta.Slug, "title", meta.Title)
|
||||
|
||||
// Fetch chapter list.
|
||||
refs, err := o.novel.ScrapeChapterList(ctx, bookURL)
|
||||
if err != nil {
|
||||
o.log.Error("chapter list scrape failed", "slug", meta.Slug, "err", err)
|
||||
return
|
||||
}
|
||||
|
||||
o.log.Info("chapter list fetched", "slug", meta.Slug, "chapters", len(refs))
|
||||
|
||||
// Enqueue chapter jobs.
|
||||
for _, ref := range refs {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case chapterWork <- chapterJob{slug: meta.Slug, ref: ref}:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if o.cfg.SingleBookURL != "" {
|
||||
// Single-book mode: skip catalogue entirely.
|
||||
o.log.Info("single-book mode", "url", o.cfg.SingleBookURL)
|
||||
processBook(o.cfg.SingleBookURL)
|
||||
} else {
|
||||
// Catalogue mode: stream every book.
|
||||
entries, catErrs := o.novel.ScrapeCatalogue(ctx)
|
||||
|
||||
// Drain catalogue errors in a separate goroutine.
|
||||
go func() {
|
||||
for err := range catErrs {
|
||||
o.log.Error("catalogue error", "err", err)
|
||||
}
|
||||
}()
|
||||
|
||||
var bookWG sync.WaitGroup
|
||||
for entry := range entries {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
break
|
||||
default:
|
||||
}
|
||||
|
||||
bookWG.Add(1)
|
||||
bookURL := entry.URL
|
||||
go func() {
|
||||
defer bookWG.Done()
|
||||
processBook(bookURL)
|
||||
}()
|
||||
}
|
||||
|
||||
// Wait for all book goroutines to enqueue their chapters before
|
||||
// closing the chapter work queue.
|
||||
bookWG.Wait()
|
||||
}
|
||||
|
||||
// Signal chapter workers there is no more work.
|
||||
close(chapterWork)
|
||||
|
||||
// Wait for all in-flight chapter scrapes to finish.
|
||||
chapterWG.Wait()
|
||||
|
||||
if ctx.Err() != nil {
|
||||
return fmt.Errorf("orchestrator: context cancelled: %w", ctx.Err())
|
||||
}
|
||||
|
||||
o.log.Info("orchestrator finished")
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user