feat(server): track scrape jobs in scraping_tasks and expose GET /api/scrape/tasks
runAsync creates a scraping_tasks record on job start, flushes progress counters via OnProgress, and finalizes status (done/failed/cancelled) on completion. Adds GET /api/scrape/tasks to list all historical jobs. Also fixes relative cover URLs in parseBrowsePage.
This commit is contained in:
@@ -125,6 +125,7 @@ func (s *Server) ListenAndServe(ctx context.Context) error {
|
|||||||
mux.HandleFunc("GET /api/ranking", s.handleGetRanking)
|
mux.HandleFunc("GET /api/ranking", s.handleGetRanking)
|
||||||
// Scrape status
|
// Scrape status
|
||||||
mux.HandleFunc("GET /api/scrape/status", s.handleScrapeStatus)
|
mux.HandleFunc("GET /api/scrape/status", s.handleScrapeStatus)
|
||||||
|
mux.HandleFunc("GET /api/scrape/tasks", s.handleScrapeTasks)
|
||||||
// Progress API
|
// Progress API
|
||||||
mux.HandleFunc("GET /api/progress", s.handleGetProgress)
|
mux.HandleFunc("GET /api/progress", s.handleGetProgress)
|
||||||
mux.HandleFunc("POST /api/progress/{slug}", s.handleSetProgress)
|
mux.HandleFunc("POST /api/progress/{slug}", s.handleSetProgress)
|
||||||
@@ -699,9 +700,90 @@ func (s *Server) runAsync(w http.ResponseWriter, cfg orchestrator.Config) {
|
|||||||
ctx, cancel := context.WithTimeout(context.Background(), 24*time.Hour)
|
ctx, cancel := context.WithTimeout(context.Background(), 24*time.Hour)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
|
// Determine task kind and target.
|
||||||
|
kind := "catalogue"
|
||||||
|
targetURL := ""
|
||||||
|
if cfg.SingleBookURL != "" {
|
||||||
|
kind = "book"
|
||||||
|
targetURL = cfg.SingleBookURL
|
||||||
|
}
|
||||||
|
|
||||||
|
// Create the task record in PocketBase.
|
||||||
|
taskID, err := s.store.CreateScrapeTask(ctx, kind, targetURL)
|
||||||
|
if err != nil {
|
||||||
|
s.log.Warn("could not create scraping_tasks record", "err", err)
|
||||||
|
// Non-fatal: continue without task tracking.
|
||||||
|
}
|
||||||
|
|
||||||
|
// flush pushes the latest counters to PocketBase (best-effort).
|
||||||
|
flush := func(p orchestrator.Progress, status, errMsg string, finished bool) {
|
||||||
|
if taskID == "" {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
u := storage.ScrapeTaskUpdate{
|
||||||
|
Status: status,
|
||||||
|
BooksFound: p.BooksFound,
|
||||||
|
ChaptersScraped: p.ChaptersScraped,
|
||||||
|
ChaptersSkipped: p.ChaptersSkipped,
|
||||||
|
Errors: p.Errors,
|
||||||
|
ErrorMessage: errMsg,
|
||||||
|
}
|
||||||
|
if finished {
|
||||||
|
u.Finished = time.Now().UTC()
|
||||||
|
}
|
||||||
|
if updateErr := s.store.UpdateScrapeTask(ctx, taskID, u); updateErr != nil {
|
||||||
|
s.log.Warn("could not update scraping_tasks record", "task_id", taskID, "err", updateErr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
cfg.OnProgress = func(p orchestrator.Progress) {
|
||||||
|
flush(p, "running", "", false)
|
||||||
|
}
|
||||||
|
|
||||||
o := orchestrator.New(cfg, s.novel, s.log, s.store)
|
o := orchestrator.New(cfg, s.novel, s.log, s.store)
|
||||||
if err := o.Run(ctx); err != nil {
|
runErr := o.Run(ctx)
|
||||||
s.log.Error("scrape job failed", "err", fmt.Sprintf("%v", err))
|
|
||||||
|
// Determine final status.
|
||||||
|
finalStatus := "done"
|
||||||
|
errMsg := ""
|
||||||
|
if runErr != nil {
|
||||||
|
s.log.Error("scrape job failed", "err", fmt.Sprintf("%v", runErr))
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
finalStatus = "cancelled"
|
||||||
|
} else {
|
||||||
|
finalStatus = "failed"
|
||||||
|
}
|
||||||
|
errMsg = runErr.Error()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Best-effort: read last known progress counters via a zero-value
|
||||||
|
// OnProgress — we don't have a snapshot here, so re-use whatever the
|
||||||
|
// last OnProgress call delivered (the orchestrator calls notify() at
|
||||||
|
// the very end, so this is always accurate after Run returns).
|
||||||
|
// We issue one final flush with the terminal status and finished time.
|
||||||
|
if taskID != "" {
|
||||||
|
// Re-fetch current counters by listing the task (cheapest path).
|
||||||
|
tasks, listErr := s.store.ListScrapeTasks(ctx)
|
||||||
|
var last storage.ScrapeTaskUpdate
|
||||||
|
if listErr == nil {
|
||||||
|
for _, t := range tasks {
|
||||||
|
if t.ID == taskID {
|
||||||
|
last = storage.ScrapeTaskUpdate{
|
||||||
|
BooksFound: t.BooksFound,
|
||||||
|
ChaptersScraped: t.ChaptersScraped,
|
||||||
|
ChaptersSkipped: t.ChaptersSkipped,
|
||||||
|
Errors: t.Errors,
|
||||||
|
}
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
last.Status = finalStatus
|
||||||
|
last.ErrorMessage = errMsg
|
||||||
|
last.Finished = time.Now().UTC()
|
||||||
|
if updateErr := s.store.UpdateScrapeTask(ctx, taskID, last); updateErr != nil {
|
||||||
|
s.log.Warn("could not finalize scraping_tasks record", "task_id", taskID, "err", updateErr)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
@@ -863,6 +945,9 @@ func parseNovelItem(li *html.Node) (NovelListing, bool) {
|
|||||||
src = attrVal(n, "src")
|
src = attrVal(n, "src")
|
||||||
}
|
}
|
||||||
if src != "" && novel.Cover == "" {
|
if src != "" && novel.Cover == "" {
|
||||||
|
if !strings.HasPrefix(src, "http") {
|
||||||
|
src = novelFireBase + src
|
||||||
|
}
|
||||||
novel.Cover = src
|
novel.Cover = src
|
||||||
}
|
}
|
||||||
case "h4":
|
case "h4":
|
||||||
@@ -966,3 +1051,19 @@ func (s *Server) handleScrapeStatus(w http.ResponseWriter, _ *http.Request) {
|
|||||||
w.Header().Set("Content-Type", "application/json")
|
w.Header().Set("Content-Type", "application/json")
|
||||||
_ = json.NewEncoder(w).Encode(map[string]bool{"running": running})
|
_ = json.NewEncoder(w).Encode(map[string]bool{"running": running})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// handleScrapeTasks handles GET /api/scrape/tasks.
|
||||||
|
// Returns JSON array of all scraping_tasks records, newest first.
|
||||||
|
func (s *Server) handleScrapeTasks(w http.ResponseWriter, r *http.Request) {
|
||||||
|
tasks, err := s.store.ListScrapeTasks(r.Context())
|
||||||
|
if err != nil {
|
||||||
|
s.log.Error("handleScrapeTasks: list failed", "err", err)
|
||||||
|
http.Error(w, `{"error":"failed to list tasks"}`, http.StatusInternalServerError)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if tasks == nil {
|
||||||
|
tasks = []storage.ScrapeTask{}
|
||||||
|
}
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
_ = json.NewEncoder(w).Encode(tasks)
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user