041099598b
Adds a scraping_tasks PocketBase collection with fields for kind, status, progress counters, timestamps, and error info. Exposes CreateScrapeTask, UpdateScrapeTask, and ListScrapeTasks on the Store interface with implementations in HybridStore and PocketBaseStore.
369 lines
13 KiB
Go
369 lines
13 KiB
Go
// hybrid.go implements the Store interface using PocketBase for structured data
|
|
// and MinIO for binary chapter/audio blobs.
|
|
package storage
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log/slog"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/libnovel/scraper/internal/scraper"
|
|
)
|
|
|
|
// HybridStore satisfies Store by routing structured data to PocketBase and
|
|
// binary objects (chapters, audio) to MinIO.
|
|
type HybridStore struct {
|
|
pb *PocketBaseStore
|
|
minio *MinioClient
|
|
log *slog.Logger
|
|
}
|
|
|
|
// NewHybridStore constructs a HybridStore. It connects to both backends and
|
|
// calls EnsureCollections to bootstrap any missing PocketBase collections.
|
|
func NewHybridStore(ctx context.Context, pbCfg PocketBaseConfig, minioCfg MinioConfig, log *slog.Logger) (*HybridStore, error) {
|
|
mc, err := NewMinioClient(ctx, minioCfg)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("storage: minio: %w", err)
|
|
}
|
|
pb := NewPocketBaseStore(pbCfg, log)
|
|
// Verify PocketBase credentials before proceeding.
|
|
if err := pb.Ping(ctx); err != nil {
|
|
return nil, fmt.Errorf("storage: pocketbase auth: %w", err)
|
|
}
|
|
if err := pb.EnsureCollections(ctx); err != nil {
|
|
// Non-fatal: 400/422 means collections already exist.
|
|
log.Warn("EnsureCollections returned an error (may be safe to ignore)", "err", err)
|
|
}
|
|
return &HybridStore{pb: pb, minio: mc, log: log}, nil
|
|
}
|
|
|
|
// ─── Book metadata ────────────────────────────────────────────────────────────
|
|
|
|
func (h *HybridStore) WriteMetadata(ctx context.Context, meta scraper.BookMeta) error {
|
|
return h.pb.UpsertBook(ctx,
|
|
meta.Slug, meta.Title, meta.Author, meta.Cover,
|
|
meta.Status, meta.Summary, meta.SourceURL,
|
|
meta.Genres, meta.TotalChapters, meta.Ranking,
|
|
)
|
|
}
|
|
|
|
func (h *HybridStore) ReadMetadata(ctx context.Context, slug string) (scraper.BookMeta, bool, error) {
|
|
rec, found, err := h.pb.GetBook(ctx, slug)
|
|
if err != nil || !found {
|
|
return scraper.BookMeta{}, found, err
|
|
}
|
|
return recToBookMeta(rec), true, nil
|
|
}
|
|
|
|
func (h *HybridStore) ListBooks(ctx context.Context) ([]scraper.BookMeta, error) {
|
|
rows, err := h.pb.ListBooks(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
books := make([]scraper.BookMeta, 0, len(rows))
|
|
for _, r := range rows {
|
|
books = append(books, recToBookMeta(r))
|
|
}
|
|
return books, nil
|
|
}
|
|
|
|
func (h *HybridStore) LocalSlugs(ctx context.Context) (map[string]bool, error) {
|
|
books, err := h.ListBooks(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
slugs := make(map[string]bool, len(books))
|
|
for _, b := range books {
|
|
slugs[b.Slug] = true
|
|
}
|
|
return slugs, nil
|
|
}
|
|
|
|
func (h *HybridStore) MetadataMtime(ctx context.Context, slug string) int64 {
|
|
t, err := h.pb.BookMetaUpdated(ctx, slug)
|
|
if err != nil {
|
|
h.log.Warn("MetadataMtime: BookMetaUpdated failed", "slug", slug, "err", err)
|
|
return 0
|
|
}
|
|
if t.IsZero() {
|
|
return 0
|
|
}
|
|
return t.Unix()
|
|
}
|
|
|
|
// ─── Chapters ─────────────────────────────────────────────────────────────────
|
|
|
|
func (h *HybridStore) ChapterExists(ctx context.Context, slug string, ref scraper.ChapterRef) bool {
|
|
return h.minio.ChapterExists(ctx, slug, ref.Volume, ref.Number)
|
|
}
|
|
|
|
func (h *HybridStore) WriteChapter(ctx context.Context, slug string, chapter scraper.Chapter) error {
|
|
content := "# " + chapter.Ref.Title + "\n\n" + chapter.Text + "\n"
|
|
if err := h.minio.PutChapter(ctx, slug, chapter.Ref.Volume, chapter.Ref.Number, content); err != nil {
|
|
return err
|
|
}
|
|
// Update chapter index in PocketBase.
|
|
title, dateLabel := splitChapterTitle(chapter.Ref.Title)
|
|
if err := h.pb.UpsertChapterIdx(ctx, slug, chapter.Ref.Number, title, dateLabel); err != nil {
|
|
h.log.Warn("WriteChapter: failed to upsert chapter index in PocketBase",
|
|
"slug", slug, "chapter", chapter.Ref.Number, "err", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (h *HybridStore) ReadChapter(ctx context.Context, slug string, n int) (string, error) {
|
|
return h.minio.GetChapter(ctx, slug, 0, n)
|
|
}
|
|
|
|
func (h *HybridStore) ListChapters(ctx context.Context, slug string) ([]ChapterInfo, error) {
|
|
rows, err := h.pb.ListChapterIdx(ctx, slug)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
infos := make([]ChapterInfo, 0, len(rows))
|
|
for _, r := range rows {
|
|
n := int(floatVal(r, "number"))
|
|
title, _ := r["title"].(string)
|
|
date, _ := r["date_label"].(string)
|
|
infos = append(infos, ChapterInfo{Number: n, Title: title, Date: date})
|
|
}
|
|
sort.Slice(infos, func(i, j int) bool { return infos[i].Number < infos[j].Number })
|
|
return infos, nil
|
|
}
|
|
|
|
func (h *HybridStore) CountChapters(ctx context.Context, slug string) int {
|
|
return h.pb.CountChapterIdx(ctx, slug)
|
|
}
|
|
|
|
// ─── Ranking ─────────────────────────────────────────────────────────────────
|
|
|
|
func (h *HybridStore) WriteRankingItem(ctx context.Context, item RankingItem) error {
|
|
return h.pb.UpsertRankingItem(ctx, item)
|
|
}
|
|
|
|
func (h *HybridStore) ReadRankingItems(ctx context.Context) ([]RankingItem, error) {
|
|
return h.pb.ListRankingItems(ctx)
|
|
}
|
|
|
|
func (h *HybridStore) RankingFreshEnough(ctx context.Context, maxAge time.Duration) (bool, error) {
|
|
last, err := h.pb.RankingLastUpdated(ctx)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if last.IsZero() {
|
|
return false, nil
|
|
}
|
|
return time.Since(last) < maxAge, nil
|
|
}
|
|
|
|
// ─── Audio cache ──────────────────────────────────────────────────────────────
|
|
|
|
func (h *HybridStore) GetAudioCache(ctx context.Context, cacheKey string) (string, bool) {
|
|
filename, ok, err := h.pb.GetAudioCache(ctx, cacheKey)
|
|
if err != nil {
|
|
h.log.Warn("GetAudioCache: PocketBase lookup failed", "cache_key", cacheKey, "err", err)
|
|
}
|
|
return filename, ok
|
|
}
|
|
|
|
func (h *HybridStore) SetAudioCache(ctx context.Context, cacheKey, filename string) error {
|
|
return h.pb.SetAudioCache(ctx, cacheKey, filename)
|
|
}
|
|
|
|
// ─── Reading progress ─────────────────────────────────────────────────────────
|
|
|
|
func (h *HybridStore) GetProgress(ctx context.Context, sessionID, slug string) (ReadingProgress, bool) {
|
|
ch, updated, ok, err := h.pb.GetProgress(ctx, sessionID, slug)
|
|
if err != nil {
|
|
h.log.Warn("GetProgress: PocketBase lookup failed", "slug", slug, "err", err)
|
|
return ReadingProgress{}, false
|
|
}
|
|
if !ok {
|
|
return ReadingProgress{}, false
|
|
}
|
|
return ReadingProgress{Slug: slug, Chapter: ch, UpdatedAt: updated}, true
|
|
}
|
|
|
|
func (h *HybridStore) SetProgress(ctx context.Context, sessionID string, p ReadingProgress) error {
|
|
return h.pb.SetProgress(ctx, sessionID, p.Slug, p.Chapter)
|
|
}
|
|
|
|
func (h *HybridStore) AllProgress(ctx context.Context, sessionID string) ([]ReadingProgress, error) {
|
|
rows, err := h.pb.AllProgress(ctx, sessionID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out := make([]ReadingProgress, 0, len(rows))
|
|
for _, r := range rows {
|
|
slug, _ := r["slug"].(string)
|
|
ch := int(floatVal(r, "chapter"))
|
|
var updated time.Time
|
|
if ts, ok := r["updated"].(string); ok {
|
|
updated, _ = time.Parse(time.RFC3339, ts)
|
|
}
|
|
out = append(out, ReadingProgress{Slug: slug, Chapter: ch, UpdatedAt: updated})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (h *HybridStore) DeleteProgress(ctx context.Context, sessionID, slug string) error {
|
|
return h.pb.DeleteProgress(ctx, sessionID, slug)
|
|
}
|
|
|
|
// ─── AudioObjectKey ───────────────────────────────────────────────────────────
|
|
|
|
func (h *HybridStore) AudioObjectKey(slug string, n int, voice string, speed float64) string {
|
|
return AudioObjectKey(slug, n, voice, speed)
|
|
}
|
|
|
|
// ─── PutAudio ─────────────────────────────────────────────────────────────────
|
|
|
|
func (h *HybridStore) PutAudio(ctx context.Context, key string, data []byte) error {
|
|
return h.minio.PutAudio(ctx, key, data)
|
|
}
|
|
|
|
// ─── Presigned URLs ───────────────────────────────────────────────────────────
|
|
|
|
func (h *HybridStore) PresignChapter(ctx context.Context, slug string, n int, expires time.Duration) (string, error) {
|
|
return h.minio.PresignChapter(ctx, slug, 0, n, expires)
|
|
}
|
|
|
|
func (h *HybridStore) PresignAudio(ctx context.Context, key string, expires time.Duration) (string, error) {
|
|
return h.minio.PresignAudio(ctx, key, expires)
|
|
}
|
|
|
|
// ─── Scraping tasks ───────────────────────────────────────────────────────────
|
|
|
|
func (h *HybridStore) CreateScrapeTask(ctx context.Context, kind, targetURL string) (string, error) {
|
|
return h.pb.CreateScrapingTask(ctx, kind, targetURL)
|
|
}
|
|
|
|
func (h *HybridStore) UpdateScrapeTask(ctx context.Context, id string, u ScrapeTaskUpdate) error {
|
|
data := map[string]interface{}{
|
|
"status": u.Status,
|
|
"books_found": u.BooksFound,
|
|
"chapters_scraped": u.ChaptersScraped,
|
|
"chapters_skipped": u.ChaptersSkipped,
|
|
"errors": u.Errors,
|
|
"error_message": u.ErrorMessage,
|
|
}
|
|
if !u.Finished.IsZero() {
|
|
data["finished"] = u.Finished.UTC().Format(time.RFC3339)
|
|
}
|
|
return h.pb.UpdateScrapingTask(ctx, id, data)
|
|
}
|
|
|
|
func (h *HybridStore) ListScrapeTasks(ctx context.Context) ([]ScrapeTask, error) {
|
|
rows, err := h.pb.ListScrapingTasks(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
tasks := make([]ScrapeTask, 0, len(rows))
|
|
for _, r := range rows {
|
|
t := ScrapeTask{
|
|
ID: strVal(r, "id"),
|
|
Kind: strVal(r, "kind"),
|
|
TargetURL: strVal(r, "target_url"),
|
|
Status: strVal(r, "status"),
|
|
BooksFound: int(floatVal(r, "books_found")),
|
|
ChaptersScraped: int(floatVal(r, "chapters_scraped")),
|
|
ChaptersSkipped: int(floatVal(r, "chapters_skipped")),
|
|
Errors: int(floatVal(r, "errors")),
|
|
ErrorMessage: strVal(r, "error_message"),
|
|
}
|
|
if ts, ok := r["started"].(string); ok {
|
|
t.Started, _ = time.Parse(time.RFC3339, ts)
|
|
}
|
|
if ts, ok := r["finished"].(string); ok && ts != "" {
|
|
t.Finished, _ = time.Parse(time.RFC3339, ts)
|
|
}
|
|
tasks = append(tasks, t)
|
|
}
|
|
return tasks, nil
|
|
}
|
|
|
|
// ─── helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
func recToBookMeta(rec map[string]interface{}) scraper.BookMeta {
|
|
m := scraper.BookMeta{
|
|
Slug: strVal(rec, "slug"),
|
|
Title: strVal(rec, "title"),
|
|
Author: strVal(rec, "author"),
|
|
Cover: strVal(rec, "cover"),
|
|
Status: strVal(rec, "status"),
|
|
Summary: strVal(rec, "summary"),
|
|
SourceURL: strVal(rec, "source_url"),
|
|
}
|
|
if tc := floatVal(rec, "total_chapters"); tc > 0 {
|
|
m.TotalChapters = int(tc)
|
|
}
|
|
if rk := floatVal(rec, "ranking"); rk > 0 {
|
|
m.Ranking = int(rk)
|
|
}
|
|
// Genres stored as JSON string or array.
|
|
switch v := rec["genres"].(type) {
|
|
case string:
|
|
_ = json.Unmarshal([]byte(v), &m.Genres)
|
|
case []interface{}:
|
|
for _, g := range v {
|
|
if s, ok := g.(string); ok {
|
|
m.Genres = append(m.Genres, s)
|
|
}
|
|
}
|
|
}
|
|
return m
|
|
}
|
|
|
|
func strVal(m map[string]interface{}, key string) string {
|
|
if v, ok := m[key].(string); ok {
|
|
return v
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// splitChapterTitle mirrors writer.SplitChapterTitle logic (simplified).
|
|
func splitChapterTitle(raw string) (title, date string) {
|
|
raw = strings.TrimSpace(raw)
|
|
// Strip leading numeric index.
|
|
if idx := strings.IndexFunc(raw, func(r rune) bool { return r == ' ' || r == '\t' }); idx > 0 {
|
|
prefix := raw[:idx]
|
|
allDigit := true
|
|
for _, c := range prefix {
|
|
if c < '0' || c > '9' {
|
|
allDigit = false
|
|
break
|
|
}
|
|
}
|
|
if allDigit {
|
|
raw = strings.TrimSpace(raw[idx:])
|
|
}
|
|
}
|
|
// Detect trailing relative date.
|
|
units := []string{"second", "minute", "hour", "day", "week", "month", "year"}
|
|
lower := strings.ToLower(raw)
|
|
for _, u := range units {
|
|
for _, suffix := range []string{u + "s ago", u + " ago"} {
|
|
if idx := strings.LastIndex(lower, suffix); idx > 0 {
|
|
// Find start of date token (digit before the unit).
|
|
start := strings.LastIndex(raw[:idx], " ")
|
|
if start < 0 {
|
|
start = 0
|
|
}
|
|
numPart := strings.TrimSpace(raw[start:idx])
|
|
fields := strings.Fields(numPart)
|
|
if len(fields) > 0 {
|
|
if _, err := strconv.Atoi(fields[0]); err == nil {
|
|
return strings.TrimSpace(raw[:start]), strings.TrimSpace(raw[start : idx+len(suffix)])
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return raw, ""
|
|
}
|