feat: downloads, RSS, subtitles, Bangumi, watcher, stats, profile, audit

Backend
  - service/bangumi.go: Bangumi (bgm.tv) scraper for anime libraries.
  - service/episode_parser.go: SxxExx / NxE / EPxx / 第NN集 parser with
    unit tests; consumed by the scanner for tv/anime libraries.
  - service/scraper.go: provider chain orchestrator picks Bangumi for
    anime libraries (TMDb fallback), TMDb for everything else; emits
    'no_match' rows so we don't retry forever; AnyEnabled() for the
    scanner kick.
  - service/scanner.go: writes season/episode numbers for tv/anime libs
    and reports a 'probed' counter alongside 'added'.
  - service/transcoder.go (existing): unchanged, stays per-media.
  - service/subtitle.go: discovers external subtitles next to the source
    file (.srt/.vtt/.ass/.ssa) and converts SRT/ASS to WebVTT on the fly.
  - service/qbittorrent.go: thread-safe qBittorrent v2 Web UI client
    (login / add / list / delete) with cookie-jar reuse.
  - service/downloads.go: persists download tasks, reads runtime
    qbittorrent.* settings, polls /torrents/info every 5 s and pushes
    the result to WS subscribers ('download' topic).
  - service/subscription.go: 10-minute RSS poller with regex filter,
    GUID dedup persisted in the settings table, and 'subscription' WS
    events on enqueue.
  - service/watcher.go: fsnotify watcher with 5 s coalescing debouncer
    that triggers per-library rescans on create/rename/remove.
  - service/stats.go: dashboard snapshot — totals, recently added,
    gopsutil-driven CPU/mem/disk readings.
  - service/profile.go: non-credential profile patch + admin role mutator.
  - service/audit.go: best-effort writer for the access_logs table.
  - service/service.go: container wires every new service; Boot() spins
    up watcher / downloads poller / subscription scheduler; Close()
    tears them down on graceful shutdown.

Handlers
  - new files: downloads.go, subscriptions.go, subtitles.go, series.go,
    stats.go, profile.go, util.go.
  - handler.go: registers PATCH /me, /libraries/:id/seasons,
    /media/:id/subtitles, /subtitles/:id, /downloads*, /subscriptions*,
    /stats, /admin/users/:id/role.
  - auth.go / media.go: write audit rows for login + library CRUD and
    refresh the watcher when libraries change.

Frontend
  - api: new helpers for downloads, subscriptions, profile, series, stats,
    subtitles; library helper gained scrape().
  - hooks/useWebSocket.ts: shared connection with 3 s reconnect.
  - components/GlobalEvents.tsx: surfaces scan / scrape / subscription
    completion as toasts (mounted at app root).
  - pages: Library now switches to a season-grouped layout for tv/anime
    libraries; Player attaches WebVTT <track> elements; new pages for
    Downloads (live torrent table), Subscriptions, Profile, Stats.
  - components/Layout.tsx + App.tsx: sidebar groups (媒体库 / 自动化 /
    账号 / 管理) and routes for the new pages; /stats and /admin remain
    admin-only.
  - types/index.ts: new types — Subscription, DownloadTask, QBitTorrent,
    Hardware, StatsSnapshot.

Verified: go build, go vet, go test (incl. ParseEpisode + srtToVTT +
stripASSTags) all pass; tsc -b && vite build emits 17 route chunks plus
the deferred hls chunk; main bundle 247 KB / 83 KB gzipped.
This commit is contained in:
Kiro
2026-05-14 16:07:49 +00:00
parent 0f30c34463
commit 7bd6bcfb1b
50 changed files with 3134 additions and 146 deletions
+1
View File
@@ -41,6 +41,7 @@ func loginHandler(svc *service.Container) gin.HandlerFunc {
"token": token,
"user": u,
})
svc.Audit.Record(c.Request.Context(), u.ID, "auth.login", u.Username, c.ClientIP(), "")
}
}
+70
View File
@@ -0,0 +1,70 @@
// Package handler — download manager endpoints.
package handler
import (
"net/http"
"github.com/gin-gonic/gin"
"github.com/ShukeBta/MediaStationGo/internal/middleware"
"github.com/ShukeBta/MediaStationGo/internal/service"
)
type addDownloadReq struct {
URL string `json:"url" binding:"required"`
SavePath string `json:"save_path"`
}
func addDownloadHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
var req addDownloadReq
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
uid, _ := c.Get(middleware.CtxUserID)
t, err := svc.Downloads.AddDownload(c.Request.Context(), uid.(string), req.URL, req.SavePath)
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
svc.Audit.Record(c.Request.Context(), uid.(string), "download.add", req.URL, c.ClientIP(), "")
c.JSON(http.StatusOK, t)
}
}
func listDownloadsHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
rows, live, err := svc.Downloads.List(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{
"tasks": rows,
"torrents": live,
})
}
}
func deleteDownloadHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
hash := c.Param("hash")
withFiles := c.Query("delete_files") == "true"
if err := svc.Downloads.Delete(c.Request.Context(), hash, withFiles); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.Status(http.StatusNoContent)
}
}
func reloadDownloadConfigHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
if err := svc.Downloads.ReloadConfig(c.Request.Context()); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.Status(http.StatusNoContent)
}
}
+23
View File
@@ -32,8 +32,10 @@ func Register(r *gin.Engine, cfg *config.Config, log *zap.Logger, svc *service.C
authed.Use(middleware.AuthRequired(cfg.Secrets.JWTSecret))
{
authed.GET("/me", meHandler(svc))
authed.PATCH("/me", updateProfileHandler(svc))
authed.POST("/me/password", changePasswordHandler(svc))
// Libraries.
authed.GET("/libraries", listLibrariesHandler(svc))
authed.POST("/libraries", middleware.AdminRequired(), createLibraryHandler(svc))
authed.DELETE("/libraries/:id", middleware.AdminRequired(), deleteLibraryHandler(svc))
@@ -41,10 +43,15 @@ func Register(r *gin.Engine, cfg *config.Config, log *zap.Logger, svc *service.C
authed.POST("/libraries/:id/scrape", middleware.AdminRequired(), scrapeLibraryHandler(svc))
authed.GET("/libraries/:id/media", listMediaHandler(svc))
authed.GET("/libraries/:id/seasons", listSeasonsHandler(svc))
// Media.
authed.GET("/media/:id", getMediaHandler(svc))
authed.GET("/media", searchMediaHandler(svc))
authed.POST("/media/:id/scrape", middleware.AdminRequired(), scrapeOneHandler(svc))
authed.POST("/media/:id/probe", middleware.AdminRequired(), reprobeHandler(svc))
authed.GET("/media/:id/subtitles", listSubtitlesHandler(svc))
authed.GET("/subtitles/:id", serveSubtitleHandler(svc))
// Streaming.
authed.GET("/stream/:id", streamHandler(svc))
@@ -69,6 +76,21 @@ func Register(r *gin.Engine, cfg *config.Config, log *zap.Logger, svc *service.C
authed.DELETE("/playlists/:id/items/:media_id", removePlaylistItemHandler(svc))
authed.DELETE("/playlists/:id", deletePlaylistHandler(svc))
// Downloads.
authed.GET("/downloads", listDownloadsHandler(svc))
authed.POST("/downloads", addDownloadHandler(svc))
authed.DELETE("/downloads/:hash", middleware.AdminRequired(), deleteDownloadHandler(svc))
authed.POST("/downloads/reload", middleware.AdminRequired(), reloadDownloadConfigHandler(svc))
// Subscriptions.
authed.GET("/subscriptions", listSubscriptionsHandler(svc))
authed.POST("/subscriptions", createSubscriptionHandler(svc))
authed.DELETE("/subscriptions/:id", deleteSubscriptionHandler(svc))
authed.POST("/subscriptions/:id/run", runSubscriptionHandler(svc))
// Stats / dashboard.
authed.GET("/stats", statsHandler(svc))
authed.GET("/ws", wsHandler(svc))
}
@@ -77,6 +99,7 @@ func Register(r *gin.Engine, cfg *config.Config, log *zap.Logger, svc *service.C
admin.Use(middleware.AuthRequired(cfg.Secrets.JWTSecret), middleware.AdminRequired())
{
admin.GET("/users", listUsersHandler(svc))
admin.PATCH("/users/:id/role", adminUpdateRoleHandler(svc))
admin.DELETE("/users/:id", deleteUserHandler(svc))
admin.GET("/settings", listSettingsHandler(svc))
admin.PUT("/settings", updateSettingHandler(svc))
+7
View File
@@ -40,6 +40,10 @@ func createLibraryHandler(svc *service.Container) gin.HandlerFunc {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
uid, _ := c.Get("ctx_user_id")
svc.Audit.Record(c.Request.Context(), toString(uid), "library.create", l.ID, c.ClientIP(), l.Path)
// Refresh fsnotify watcher to pick up the new library root.
go func() { _ = svc.Watcher.Refresh(c.Request.Context()) }()
c.JSON(http.StatusOK, l)
}
}
@@ -51,6 +55,9 @@ func deleteLibraryHandler(svc *service.Container) gin.HandlerFunc {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
uid, _ := c.Get("ctx_user_id")
svc.Audit.Record(c.Request.Context(), toString(uid), "library.delete", id, c.ClientIP(), "")
go func() { _ = svc.Watcher.Refresh(c.Request.Context()) }()
c.Status(http.StatusNoContent)
}
}
+48
View File
@@ -0,0 +1,48 @@
// Package handler — user profile endpoints.
package handler
import (
"net/http"
"github.com/gin-gonic/gin"
"github.com/ShukeBta/MediaStationGo/internal/middleware"
"github.com/ShukeBta/MediaStationGo/internal/service"
)
func updateProfileHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
var patch service.ProfileUpdate
if err := c.ShouldBindJSON(&patch); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
uid, _ := c.Get(middleware.CtxUserID)
u, err := svc.Profile.UpdateProfile(c.Request.Context(), uid.(string), patch)
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, u)
}
}
type adminUpdateRoleReq struct {
Role string `json:"role" binding:"required"`
}
func adminUpdateRoleHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
var req adminUpdateRoleReq
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
u, err := svc.Profile.AdminUpdateRole(c.Request.Context(), c.Param("id"), req.Role)
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, u)
}
}
+47
View File
@@ -0,0 +1,47 @@
// Package handler — TV series endpoints.
//
// These return episode lists grouped by season number for a library that
// holds TV episodes. Series rows are distinct from Movies — the front
// end uses /api/libraries/:id/seasons to render a season selector.
package handler
import (
"net/http"
"sort"
"github.com/gin-gonic/gin"
"gorm.io/gorm"
"github.com/ShukeBta/MediaStationGo/internal/model"
"github.com/ShukeBta/MediaStationGo/internal/service"
)
// seasonGroup is the JSON returned to the React UI per season.
type seasonGroup struct {
Season int `json:"season"`
Episodes []model.Media `json:"episodes"`
}
func listSeasonsHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
libID := c.Param("id")
var rows []model.Media
err := svc.Repo.DB.Where(&model.Media{LibraryID: libID}).
Order("season_num asc, episode_num asc").
Find(&rows).Error
if err != nil && err != gorm.ErrRecordNotFound {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
buckets := make(map[int][]model.Media)
for _, r := range rows {
buckets[r.SeasonNum] = append(buckets[r.SeasonNum], r)
}
out := make([]seasonGroup, 0, len(buckets))
for s, items := range buckets {
out = append(out, seasonGroup{Season: s, Episodes: items})
}
sort.Slice(out, func(i, j int) bool { return out[i].Season < out[j].Season })
c.JSON(http.StatusOK, gin.H{"seasons": out})
}
}
+21
View File
@@ -0,0 +1,21 @@
// Package handler — stats / dashboard endpoints.
package handler
import (
"net/http"
"github.com/gin-gonic/gin"
"github.com/ShukeBta/MediaStationGo/internal/service"
)
func statsHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
snap, err := svc.Stats.Compute(c.Request.Context(), svc.Cfg.App.DataDir)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, snap)
}
}
+78
View File
@@ -0,0 +1,78 @@
// Package handler — RSS subscription endpoints.
package handler
import (
"net/http"
"github.com/gin-gonic/gin"
"github.com/ShukeBta/MediaStationGo/internal/middleware"
"github.com/ShukeBta/MediaStationGo/internal/model"
"github.com/ShukeBta/MediaStationGo/internal/service"
)
type subscriptionReq struct {
Name string `json:"name" binding:"required"`
FeedURL string `json:"feed_url" binding:"required"`
Filter string `json:"filter"`
Enabled *bool `json:"enabled"`
}
func createSubscriptionHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
var req subscriptionReq
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
uid, _ := c.Get(middleware.CtxUserID)
enabled := true
if req.Enabled != nil {
enabled = *req.Enabled
}
s := &model.Subscription{
UserID: uid.(string),
Name: req.Name,
FeedURL: req.FeedURL,
Filter: req.Filter,
Enabled: enabled,
}
if err := svc.Subscription.Create(c.Request.Context(), s); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, s)
}
}
func listSubscriptionsHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
items, err := svc.Subscription.List(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"items": items})
}
}
func deleteSubscriptionHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
if err := svc.Subscription.Delete(c.Request.Context(), c.Param("id")); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.Status(http.StatusNoContent)
}
}
func runSubscriptionHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
n, err := svc.Subscription.RunNow(c.Request.Context(), c.Param("id"))
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"queued": n})
}
}
+37
View File
@@ -0,0 +1,37 @@
// Package handler — subtitle endpoints.
package handler
import (
"net/http"
"github.com/gin-gonic/gin"
"github.com/ShukeBta/MediaStationGo/internal/service"
)
func listSubtitlesHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
tracks, err := svc.Subtitle.Discover(c.Request.Context(), c.Param("id"))
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"tracks": tracks})
}
}
func serveSubtitleHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
path := c.Query("path")
if path == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "missing path"})
return
}
c.Header("Content-Type", "text/vtt; charset=utf-8")
c.Header("Cache-Control", "public, max-age=3600")
if err := svc.Subtitle.Serve(c.Request.Context(), c.Param("id"), path, c.Writer); err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
return
}
}
}
+14
View File
@@ -0,0 +1,14 @@
// Package handler — small utilities shared across handlers.
package handler
// toString converts a gin context value to a string, returning an empty
// string when the value is missing or of the wrong type.
func toString(v any) string {
if v == nil {
return ""
}
if s, ok := v.(string); ok {
return s
}
return ""
}
+41
View File
@@ -0,0 +1,41 @@
// Package service — audit log helper.
//
// AuditService writes structured AccessLog rows for sensitive actions
// (login, library CRUD, scrape / scan triggers, download enqueue, etc).
// It deliberately swallows write errors so audit failures never bubble
// up to the caller.
package service
import (
"context"
"go.uber.org/zap"
"github.com/ShukeBta/MediaStationGo/internal/model"
"github.com/ShukeBta/MediaStationGo/internal/repository"
)
// AuditService is the only sanctioned writer for the access_logs table.
type AuditService struct {
log *zap.Logger
repo *repository.Container
}
// NewAuditService is the constructor.
func NewAuditService(log *zap.Logger, repo *repository.Container) *AuditService {
return &AuditService{log: log, repo: repo}
}
// Record persists one audit row.
func (a *AuditService) Record(ctx context.Context, userID, action, target, ip, detail string) {
row := &model.AccessLog{
UserID: userID,
Action: action,
Target: target,
IP: ip,
Detail: detail,
}
if err := a.repo.Log.Create(ctx, row); err != nil {
a.log.Debug("audit write failed", zap.Error(err))
}
}
+126
View File
@@ -0,0 +1,126 @@
// Package service — Bangumi metadata provider.
//
// Bangumi (https://bgm.tv) is the Chinese anime / manga / game database
// most users in mainland China prefer. Its public REST API is documented
// at https://bangumi.github.io/api/.
//
// We implement the minimal subset needed to enrich anime libraries:
//
// GET /search/subject/{keywords}?type=2&responseGroup=small
// GET /v0/subjects/{id} (cover)
//
// The provider gracefully no-ops when bangumi_access_token is empty.
package service
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/url"
"strings"
"time"
"go.uber.org/zap"
"github.com/ShukeBta/MediaStationGo/internal/config"
)
// BangumiProvider talks to https://api.bgm.tv.
type BangumiProvider struct {
cfg *config.Config
log *zap.Logger
client *http.Client
base string
}
// NewBangumiProvider is the constructor.
func NewBangumiProvider(cfg *config.Config, log *zap.Logger) *BangumiProvider {
return &BangumiProvider{
cfg: cfg,
log: log,
base: "https://api.bgm.tv",
client: &http.Client{Timeout: 15 * time.Second},
}
}
// Enabled reports whether a token is configured. Bangumi works without
// auth for read-only endpoints, but configuring a token raises the rate
// limit so we treat presence as a soft "enabled" flag.
func (b *BangumiProvider) Enabled() bool {
// Bangumi search works without auth — keep the provider enabled
// unconditionally, but pass the token when we have one.
return true
}
// Search runs a Bangumi keyword search and returns the top match. Type 2
// = anime; pass 1 (book) / 6 (real) when extending later.
func (b *BangumiProvider) Search(ctx context.Context, query string) (*Match, error) {
if query == "" {
return nil, nil
}
u := fmt.Sprintf("%s/search/subject/%s?type=2&responseGroup=small",
b.base, url.PathEscape(query))
type subject struct {
ID int `json:"id"`
Name string `json:"name"`
NameCN string `json:"name_cn"`
Summary string `json:"summary"`
Air string `json:"air_date"`
Rating struct {
Score float32 `json:"score"`
} `json:"rating"`
Images struct {
Large string `json:"large"`
} `json:"images"`
}
type page struct {
Results int `json:"results"`
List []subject `json:"list"`
}
var p page
if err := b.getJSON(ctx, u, &p); err != nil {
return nil, err
}
if p.Results == 0 || len(p.List) == 0 {
return nil, nil
}
r := p.List[0]
title := r.NameCN
if title == "" {
title = r.Name
}
m := &Match{
BangumiID: r.ID,
Title: title,
Overview: r.Summary,
PosterURL: r.Images.Large,
Rating: r.Rating.Score,
}
if len(r.Air) >= 4 {
fmt.Sscanf(r.Air[:4], "%d", &m.Year)
}
return m, nil
}
func (b *BangumiProvider) getJSON(ctx context.Context, u string, out any) error {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
if err != nil {
return err
}
req.Header.Set("User-Agent", "MediaStationGo/0.1 (https://github.com/ShukeBta/MediaStationGo)")
if t := strings.TrimSpace(b.cfg.Secrets.BangumiToken); t != "" {
req.Header.Set("Authorization", "Bearer "+t)
}
resp, err := b.client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode >= 400 {
return fmt.Errorf("bangumi %s: %d", u, resp.StatusCode)
}
return json.NewDecoder(resp.Body).Decode(out)
}
+148
View File
@@ -0,0 +1,148 @@
// Package service — download manager.
//
// DownloadService persists user-initiated downloads, dispatches them to
// the configured client (currently qBittorrent) and pushes live progress
// to the WS hub so the React UI can render a live table.
//
// Settings consumed (system Setting table):
//
// qbittorrent.url e.g. http://127.0.0.1:8080
// qbittorrent.username qBittorrent WebUI user
// qbittorrent.password qBittorrent WebUI password
// qbittorrent.savepath optional default save dir
//
// Settings can be updated at runtime via the admin UI; ReloadConfig()
// re-reads them and re-authenticates.
package service
import (
"context"
"errors"
"sync"
"time"
"go.uber.org/zap"
"github.com/ShukeBta/MediaStationGo/internal/model"
"github.com/ShukeBta/MediaStationGo/internal/repository"
)
// DownloadService is the single download orchestrator.
type DownloadService struct {
log *zap.Logger
repo *repository.Container
hub *Hub
qb *QBitClient
mu sync.Mutex
stopCh chan struct{}
pollOnce sync.Once
}
// NewDownloadService is the constructor.
func NewDownloadService(log *zap.Logger, repo *repository.Container, hub *Hub) *DownloadService {
return &DownloadService{
log: log,
repo: repo,
hub: hub,
qb: NewQBitClient(log, QBitConfig{}),
stopCh: make(chan struct{}),
}
}
// Start kicks off the background poller (idempotent).
func (d *DownloadService) Start(ctx context.Context) {
d.pollOnce.Do(func() {
_ = d.ReloadConfig(ctx)
go d.poll(ctx)
})
}
// Stop terminates the poller.
func (d *DownloadService) Stop() {
close(d.stopCh)
}
// ReloadConfig rebuilds the qBittorrent client from the system settings.
func (d *DownloadService) ReloadConfig(ctx context.Context) error {
cfg := QBitConfig{}
for _, key := range []struct{ from, into *string }{} {
_ = key
}
get := func(k string) string {
v, _ := d.repo.Setting.Get(ctx, k)
return v
}
cfg.BaseURL = get("qbittorrent.url")
cfg.Username = get("qbittorrent.username")
cfg.Password = get("qbittorrent.password")
d.qb.Configure(cfg)
return nil
}
// AddDownload accepts a magnet URL / HTTP URL and persists a tracking row.
func (d *DownloadService) AddDownload(ctx context.Context, userID, urlStr, savePath string) (*model.DownloadTask, error) {
if urlStr == "" {
return nil, errors.New("empty url")
}
if savePath == "" {
savePath, _ = d.repo.Setting.Get(ctx, "qbittorrent.savepath")
}
if err := d.qb.AddTorrent(ctx, urlStr, savePath); err != nil {
return nil, err
}
t := &model.DownloadTask{
UserID: userID,
Source: "qbittorrent",
URL: urlStr,
SavePath: savePath,
Status: "queued",
}
if err := d.repo.Download.Create(ctx, t); err != nil {
return nil, err
}
return t, nil
}
// List returns every persisted download task augmented with live data
// from qBittorrent when available.
func (d *DownloadService) List(ctx context.Context) ([]model.DownloadTask, []QBitTorrent, error) {
rows, err := d.repo.Download.List(ctx)
if err != nil {
return nil, nil, err
}
live, err := d.qb.List(ctx, "")
if err != nil {
// Network failure shouldn't break the page — return rows with no
// live data and let the UI render the persisted snapshot.
d.log.Debug("qbittorrent list failed", zap.Error(err))
return rows, nil, nil
}
return rows, live, nil
}
// Delete removes a torrent (and optionally its files) from qBittorrent.
func (d *DownloadService) Delete(ctx context.Context, hash string, withFiles bool) error {
return d.qb.Delete(ctx, hash, withFiles)
}
// poll fans out qBittorrent /torrents/info every 5 s as WS events. The
// payload is opaque to the client; the React store merges by hash.
func (d *DownloadService) poll(ctx context.Context) {
t := time.NewTicker(5 * time.Second)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-d.stopCh:
return
case <-t.C:
}
live, err := d.qb.List(ctx, "")
if err != nil {
continue
}
d.hub.Publish("download", map[string]any{"torrents": live})
}
}
+58
View File
@@ -0,0 +1,58 @@
// Package service — episode parser for TV series.
//
// Detects season + episode numbers from filenames. Recognised patterns:
//
// S01E02 / s1e2
// 1x02 / 01x02
// EP02 / E02
// 第2集 / 第02集
//
// When neither a season nor an episode marker is present, returns (0, 0).
package service
import (
"path/filepath"
"regexp"
"strconv"
"strings"
)
var (
patSEnE = regexp.MustCompile(`(?i)s(\d{1,2})e(\d{1,3})`)
patNxE = regexp.MustCompile(`(\d{1,2})x(\d{1,3})`)
patEP = regexp.MustCompile(`(?i)(?:^|[^a-z])(?:e|ep)\.?\s*(\d{1,3})(?:[^0-9]|$)`)
patCN = regexp.MustCompile(`第\s*(\d{1,3})\s*[集话話]`)
)
// ParseEpisode tries to extract (season, episode) from an arbitrary filename.
// Returns (0, 0) when nothing recognisable is found.
func ParseEpisode(path string) (season, episode int) {
name := strings.TrimSuffix(filepath.Base(path), filepath.Ext(path))
if m := patSEnE.FindStringSubmatch(name); len(m) == 3 {
season = mustAtoi(m[1])
episode = mustAtoi(m[2])
return
}
if m := patNxE.FindStringSubmatch(name); len(m) == 3 {
season = mustAtoi(m[1])
episode = mustAtoi(m[2])
return
}
if m := patEP.FindStringSubmatch(name); len(m) >= 2 {
season = 1
episode = mustAtoi(m[1])
return
}
if m := patCN.FindStringSubmatch(name); len(m) >= 2 {
season = 1
episode = mustAtoi(m[1])
return
}
return 0, 0
}
func mustAtoi(s string) int {
v, _ := strconv.Atoi(s)
return v
}
+29
View File
@@ -0,0 +1,29 @@
package service
import "testing"
func TestParseEpisode(t *testing.T) {
cases := []struct {
in string
wantS, wantE int
}{
{"Breaking.Bad.S01E02.1080p.mkv", 1, 2},
{"breaking.bad.s5e14.mkv", 5, 14},
{"Friends 1x02.mp4", 1, 2},
{"Friends 10x24 - The One Where.mkv", 10, 24},
{"Some Anime - EP05 [1080p].mkv", 1, 5},
{"Some Anime - E12.mkv", 1, 12},
{"日剧 第03集.mkv", 1, 3},
{"日剧 第12话.mkv", 1, 12},
{"Movie.2020.1080p.mkv", 0, 0},
}
for _, tc := range cases {
t.Run(tc.in, func(t *testing.T) {
s, e := ParseEpisode(tc.in)
if s != tc.wantS || e != tc.wantE {
t.Errorf("ParseEpisode(%q) = (%d, %d), want (%d, %d)",
tc.in, s, e, tc.wantS, tc.wantE)
}
})
}
}
+67
View File
@@ -0,0 +1,67 @@
// Package service — user profile management.
package service
import (
"context"
"errors"
"strings"
"go.uber.org/zap"
"github.com/ShukeBta/MediaStationGo/internal/model"
"github.com/ShukeBta/MediaStationGo/internal/repository"
)
// ProfileService handles non-credential user mutations.
type ProfileService struct {
log *zap.Logger
repo *repository.Container
}
// NewProfileService is the constructor.
func NewProfileService(log *zap.Logger, repo *repository.Container) *ProfileService {
return &ProfileService{log: log, repo: repo}
}
// ProfileUpdate is the patch object accepted by UpdateProfile. Empty
// fields are ignored so the same payload can be reused across screens.
type ProfileUpdate struct {
Email *string `json:"email,omitempty"`
AvatarURL *string `json:"avatar_url,omitempty"`
}
// UpdateProfile applies a non-credential patch to the user.
func (p *ProfileService) UpdateProfile(ctx context.Context, userID string, patch ProfileUpdate) (*model.User, error) {
if userID == "" {
return nil, errors.New("missing user id")
}
updates := map[string]any{}
if patch.Email != nil {
v := strings.TrimSpace(*patch.Email)
updates["email"] = v
}
if patch.AvatarURL != nil {
updates["avatar_url"] = strings.TrimSpace(*patch.AvatarURL)
}
if len(updates) > 0 {
if err := p.repo.DB.Model(&model.User{}).Where("id = ?", userID).
Updates(updates).Error; err != nil {
return nil, err
}
}
return p.repo.User.FindByID(ctx, userID)
}
// AdminUpdateRole lets administrators promote / demote another user. The
// caller is expected to gate the route with AdminRequired.
func (p *ProfileService) AdminUpdateRole(ctx context.Context, userID, role string) (*model.User, error) {
role = strings.ToLower(strings.TrimSpace(role))
if role != "admin" && role != "user" {
return nil, errors.New("role must be admin or user")
}
if err := p.repo.DB.Model(&model.User{}).Where("id = ?", userID).
Update("role", role).Error; err != nil {
return nil, err
}
return p.repo.User.FindByID(ctx, userID)
}
+233
View File
@@ -0,0 +1,233 @@
// Package service — qBittorrent Web UI client.
//
// QBitClient is a thin wrapper around the qBittorrent /api/v2 REST API
// (https://github.com/qbittorrent/qBittorrent/wiki/WebUI-API).
//
// We only need three operations for the download flow:
//
// POST /auth/login
// POST /torrents/add (multipart, accepts magnet URL or .torrent bytes)
// GET /torrents/info (filtered by hash)
//
// The client stores the SID cookie returned by /auth/login and reuses it
// across calls. Re-auth happens transparently on 403.
package service
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"mime/multipart"
"net/http"
"net/http/cookiejar"
"net/url"
"strings"
"sync"
"time"
"go.uber.org/zap"
)
// QBitConfig holds the connection settings (typically loaded from the
// system Setting table or an env var).
type QBitConfig struct {
BaseURL string
Username string
Password string
}
// QBitTorrent is the subset of /torrents/info we surface to the API.
type QBitTorrent struct {
Hash string `json:"hash"`
Name string `json:"name"`
State string `json:"state"`
Progress float32 `json:"progress"`
DLSpeed int64 `json:"dlspeed"`
UpSpeed int64 `json:"upspeed"`
NumSeeds int `json:"num_seeds"`
NumLeech int `json:"num_leechs"`
Size int64 `json:"size"`
SavePath string `json:"save_path"`
}
// QBitClient is a thread-safe qBittorrent v2 API client.
type QBitClient struct {
log *zap.Logger
mu sync.Mutex
cfg QBitConfig
client *http.Client
}
// NewQBitClient builds a fresh client, applying default URL if blank.
func NewQBitClient(log *zap.Logger, cfg QBitConfig) *QBitClient {
if cfg.BaseURL == "" {
cfg.BaseURL = "http://localhost:8080"
}
jar, _ := cookiejar.New(nil)
return &QBitClient{
log: log,
cfg: cfg,
client: &http.Client{Jar: jar, Timeout: 20 * time.Second},
}
}
// Configure rotates the client to a new endpoint and re-auths next call.
func (q *QBitClient) Configure(cfg QBitConfig) {
q.mu.Lock()
defer q.mu.Unlock()
q.cfg = cfg
jar, _ := cookiejar.New(nil)
q.client.Jar = jar
}
// Login performs POST /api/v2/auth/login.
func (q *QBitClient) Login(ctx context.Context) error {
if q.cfg.BaseURL == "" {
return errors.New("qbittorrent base url not configured")
}
form := url.Values{}
form.Set("username", q.cfg.Username)
form.Set("password", q.cfg.Password)
req, err := http.NewRequestWithContext(ctx, http.MethodPost,
strings.TrimRight(q.cfg.BaseURL, "/")+"/api/v2/auth/login",
strings.NewReader(form.Encode()),
)
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
req.Header.Set("Referer", q.cfg.BaseURL)
resp, err := q.client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
if resp.StatusCode >= 400 || strings.TrimSpace(string(body)) != "Ok." {
return fmt.Errorf("qbittorrent login failed: %s", strings.TrimSpace(string(body)))
}
return nil
}
// AddTorrent submits a magnet URL or HTTP(S) URL to qBittorrent.
func (q *QBitClient) AddTorrent(ctx context.Context, magnetOrURL, savePath string) error {
q.mu.Lock()
defer q.mu.Unlock()
if err := q.ensureAuth(ctx); err != nil {
return err
}
body := &bytes.Buffer{}
w := multipart.NewWriter(body)
_ = w.WriteField("urls", magnetOrURL)
if savePath != "" {
_ = w.WriteField("savepath", savePath)
}
_ = w.Close()
req, err := http.NewRequestWithContext(ctx, http.MethodPost,
strings.TrimRight(q.cfg.BaseURL, "/")+"/api/v2/torrents/add", body,
)
if err != nil {
return err
}
req.Header.Set("Content-Type", w.FormDataContentType())
req.Header.Set("Referer", q.cfg.BaseURL)
resp, err := q.client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode >= 400 {
raw, _ := io.ReadAll(resp.Body)
return fmt.Errorf("qbittorrent add: %d: %s", resp.StatusCode, strings.TrimSpace(string(raw)))
}
return nil
}
// List returns every torrent (optionally filtered by status: all / downloading / completed).
func (q *QBitClient) List(ctx context.Context, filter string) ([]QBitTorrent, error) {
q.mu.Lock()
defer q.mu.Unlock()
if err := q.ensureAuth(ctx); err != nil {
return nil, err
}
u := strings.TrimRight(q.cfg.BaseURL, "/") + "/api/v2/torrents/info"
if filter != "" {
u += "?filter=" + url.QueryEscape(filter)
}
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
if err != nil {
return nil, err
}
req.Header.Set("Referer", q.cfg.BaseURL)
resp, err := q.client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode >= 400 {
return nil, fmt.Errorf("qbittorrent list: %d", resp.StatusCode)
}
var out []QBitTorrent
if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
return nil, err
}
return out, nil
}
// Delete removes a torrent (optionally with its files).
func (q *QBitClient) Delete(ctx context.Context, hash string, deleteFiles bool) error {
q.mu.Lock()
defer q.mu.Unlock()
if err := q.ensureAuth(ctx); err != nil {
return err
}
form := url.Values{}
form.Set("hashes", hash)
if deleteFiles {
form.Set("deleteFiles", "true")
} else {
form.Set("deleteFiles", "false")
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost,
strings.TrimRight(q.cfg.BaseURL, "/")+"/api/v2/torrents/delete",
strings.NewReader(form.Encode()),
)
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
req.Header.Set("Referer", q.cfg.BaseURL)
resp, err := q.client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode >= 400 {
return fmt.Errorf("qbittorrent delete: %d", resp.StatusCode)
}
return nil
}
// ensureAuth makes sure we have a valid SID cookie. Cheap on the happy
// path; logs in transparently otherwise.
func (q *QBitClient) ensureAuth(ctx context.Context) error {
u, err := url.Parse(q.cfg.BaseURL)
if err != nil {
return err
}
if cookies := q.client.Jar.Cookies(u); len(cookies) > 0 {
for _, c := range cookies {
if strings.EqualFold(c.Name, "SID") && c.Value != "" {
return nil
}
}
}
return q.Login(ctx)
}
+16 -15
View File
@@ -2,11 +2,13 @@
//
// ScannerService walks the configured library roots looking for video
// files, then upserts a model.Media row per file. Each upsert also runs
// ffprobe (when available) and queues a TMDb lookup for newly added rows.
// ffprobe (when available) and queues a metadata lookup for newly added
// rows.
//
// The scan is synchronous from the HTTP layer's point of view, but it
// publishes WebSocket progress events on the "scan" topic so the React
// UI can render a live counter / spinner.
// For TV / anime libraries we extract season + episode numbers from the
// filename via ParseEpisode and store them on the Media row. A future
// pass groups episodes into Series rows; the current scaffold lets the
// frontend group by `series_id`.
package service
import (
@@ -73,15 +75,6 @@ type ScanResult struct {
}
// ScanLibrary walks the library root and persists discovered media files.
//
// Workflow per file:
// 1. fast filename-based title cleanup.
// 2. ffprobe → duration / resolution / codecs (best effort).
// 3. upsert into the media table.
// 4. publish progress over the WS hub.
//
// After the walk we kick off the TMDb scraper for every still-pending
// row in the same library. The scraper has its own throttle.
func (s *ScannerService) ScanLibrary(ctx context.Context, libraryID string) (*ScanResult, error) {
lib, err := s.repo.Library.FindByID(ctx, libraryID)
if err != nil || lib == nil {
@@ -113,6 +106,13 @@ func (s *ScannerService) ScanLibrary(ctx context.Context, libraryID string) (*Sc
Container: strings.TrimPrefix(ext, "."),
}
// Detect season/episode for TV / anime libraries.
if lib.Type == "tv" || lib.Type == "anime" {
s, e := ParseEpisode(path)
m.SeasonNum = s
m.EpisodeNum = e
}
// Best-effort ffprobe; failure does not abort the file.
if s.probe != nil {
if probe, err := s.probe.Probe(ctx, path); err == nil && probe != nil {
@@ -157,8 +157,9 @@ func (s *ScannerService) ScanLibrary(ctx context.Context, libraryID string) (*Sc
"probed": res.Probed,
})
// Fire-and-forget metadata enrichment when a TMDb key is configured.
if s.scraper != nil && s.scraper.tmdb != nil && s.scraper.tmdb.Enabled() {
// Fire-and-forget metadata enrichment when at least one provider is
// configured.
if s.scraper != nil && s.scraper.AnyEnabled() {
go func(libID string) {
if _, err := s.scraper.EnrichLibrary(context.Background(), libID); err != nil {
s.log.Warn("scraper enrich failed", zap.Error(err))
+95 -35
View File
@@ -1,12 +1,15 @@
// Package service — scraper orchestrator.
//
// ScraperService takes a Media row and tries to enrich it with metadata
// from one or more providers (currently TMDb only). It is invoked at the
// end of every scan cycle for media items whose `scrape_status` is still
// "pending"; it can also be re-triggered manually from the admin UI.
// from one or more providers. Selection is driven by the library type:
//
// The orchestrator is deliberately stateless: it loops media → provider →
// repository, publishing scrape progress events to the WS hub.
// library.type == "anime" -> Bangumi (fallback: TMDb)
// library.type == "tv" -> TMDb (movies) — TV episodes inherit
// series metadata; episode-level scraping
// is left as a future step
// default -> TMDb
//
// The orchestrator publishes scrape progress events on the WS hub.
package service
import (
@@ -26,23 +29,33 @@ import (
// ScraperService coordinates metadata enrichment across providers.
type ScraperService struct {
cfg *config.Config
log *zap.Logger
repo *repository.Container
tmdb *TMDbProvider
hub *Hub
cfg *config.Config
log *zap.Logger
repo *repository.Container
tmdb *TMDbProvider
bangumi *BangumiProvider
hub *Hub
}
// NewScraperService is the constructor.
func NewScraperService(cfg *config.Config, log *zap.Logger, repo *repository.Container, tmdb *TMDbProvider, hub *Hub) *ScraperService {
return &ScraperService{cfg: cfg, log: log, repo: repo, tmdb: tmdb, hub: hub}
func NewScraperService(
cfg *config.Config,
log *zap.Logger,
repo *repository.Container,
tmdb *TMDbProvider,
bangumi *BangumiProvider,
hub *Hub,
) *ScraperService {
return &ScraperService{
cfg: cfg, log: log, repo: repo,
tmdb: tmdb, bangumi: bangumi, hub: hub,
}
}
// yearPattern extracts a 4-digit year from a filename (1900-2099).
// yearPattern extracts a 4-digit year (1900-2099).
var yearPattern = regexp.MustCompile(`(?:^|[^\d])(19\d{2}|20\d{2})(?:[^\d]|$)`)
// noiseTokens are aggressively stripped from filenames before search.
// Keep in sync with nowen-video's filename_parser.go intent.
// noiseTokens are stripped before search.
var noiseTokens = []string{
"1080p", "2160p", "4k", "720p", "480p",
"hdrip", "bluray", "blu-ray", "webrip", "web-dl", "web",
@@ -52,8 +65,7 @@ var noiseTokens = []string{
"hkfree", "yify", "rarbg", "ettv", "fgt",
}
// bracketedTag matches "[anything]" or "(anything)" segments, which are
// almost always release-group / encoder tags in scene filenames.
// bracketedTag matches "[anything]" or "(anything)" segments.
var bracketedTag = regexp.MustCompile(`[\[\(][^\]\)]*[\]\)]`)
// CleanQuery converts a filename like "Inception.2010.1080p.BluRay.x264.mkv"
@@ -70,13 +82,18 @@ func CleanQuery(raw string) (title string, year int) {
}
}
// 2. Drop everything inside square / round brackets — those are tags.
// 2. Drop everything inside brackets.
lower = bracketedTag.ReplaceAllString(lower, " ")
// 3. Drop episode markers (S01E02 / 1x02 / EP05 / 第03集).
lower = patSEnE.ReplaceAllString(lower, " ")
lower = patNxE.ReplaceAllString(lower, " ")
lower = patEP.ReplaceAllString(lower, " ")
lower = patCN.ReplaceAllString(lower, " ")
for _, t := range noiseTokens {
lower = strings.ReplaceAll(lower, t, " ")
}
// collapse separators / spaces
for _, sep := range []string{".", "_", "-", "[", "]", "(", ")"} {
lower = strings.ReplaceAll(lower, sep, " ")
}
@@ -85,11 +102,15 @@ func CleanQuery(raw string) (title string, year int) {
return strings.TrimSpace(title), year
}
// EnrichOne runs the provider chain for a single media row.
// EnrichOne runs the provider chain for a single media row. The library's
// type decides which provider goes first; a fallback runs when the primary
// returns nothing.
func (s *ScraperService) EnrichOne(ctx context.Context, m *model.Media) error {
if s.tmdb == nil || !s.tmdb.Enabled() {
return nil
lib, err := s.repo.Library.FindByID(ctx, m.LibraryID)
if err != nil {
return err
}
query := m.Title
if query == "" {
query, _ = CleanQuery(m.Path)
@@ -100,10 +121,15 @@ func (s *ScraperService) EnrichOne(ctx context.Context, m *model.Media) error {
if year == 0 {
_, year = CleanQuery(filepath.Base(m.Path))
}
match, err := s.tmdb.SearchMovie(ctx, query, year)
if err != nil || match == nil {
return err
match := s.lookup(ctx, lib, query, year)
if match == nil {
// Mark explicitly so we don't retry forever.
_ = s.repo.DB.Model(&model.Media{}).Where("id = ?", m.ID).
Update("scrape_status", "no_match").Error
return nil
}
updates := map[string]any{
"title": match.Title,
"overview": match.Overview,
@@ -111,28 +137,51 @@ func (s *ScraperService) EnrichOne(ctx context.Context, m *model.Media) error {
"backdrop_url": match.BackdropURL,
"rating": match.Rating,
"year": match.Year,
"tmdb_id": match.TMDbID,
"scrape_status": "matched",
}
if match.TMDbID > 0 {
updates["tmdb_id"] = match.TMDbID
}
if match.BangumiID > 0 {
updates["bangumi_id"] = match.BangumiID
}
if err := s.repo.DB.Model(&model.Media{}).Where("id = ?", m.ID).
Updates(updates).Error; err != nil {
return err
}
s.hub.Publish("scrape", map[string]any{
"media_id": m.ID,
"title": match.Title,
"tmdb_id": match.TMDbID,
"media_id": m.ID,
"title": match.Title,
"tmdb_id": match.TMDbID,
"bangumi_id": match.BangumiID,
})
return nil
}
// EnrichLibrary runs the provider chain for every "pending" media in a
// library. It throttles to 4 RPS to stay below TMDb's rate limit and
// publishes a summary event when done.
func (s *ScraperService) EnrichLibrary(ctx context.Context, libraryID string) (int, error) {
if s.tmdb == nil || !s.tmdb.Enabled() {
return 0, nil
// lookup runs the provider chain. When the library is missing we fall
// back to TMDb only.
func (s *ScraperService) lookup(ctx context.Context, lib *model.Library, query string, year int) *Match {
kind := ""
if lib != nil {
kind = lib.Type
}
if kind == "anime" && s.bangumi != nil {
if m, err := s.bangumi.Search(ctx, query); err == nil && m != nil {
return m
}
s.log.Debug("bangumi miss, falling back to tmdb", zap.String("query", query))
}
if s.tmdb != nil && s.tmdb.Enabled() {
if m, err := s.tmdb.SearchMovie(ctx, query, year); err == nil && m != nil {
return m
}
}
return nil
}
// EnrichLibrary runs the provider chain for every "pending" media in a
// library. It throttles to 4 RPS and publishes a summary event when done.
func (s *ScraperService) EnrichLibrary(ctx context.Context, libraryID string) (int, error) {
var rows []model.Media
q := s.repo.DB.Where("scrape_status = ?", "pending")
if libraryID != "" {
@@ -162,3 +211,14 @@ func (s *ScraperService) EnrichLibrary(ctx context.Context, libraryID string) (i
})
return matched, nil
}
// AnyEnabled reports whether at least one provider can run.
func (s *ScraperService) AnyEnabled() bool {
if s.tmdb != nil && s.tmdb.Enabled() {
return true
}
if s.bangumi != nil && s.bangumi.Enabled() {
return true
}
return false
}
+82 -30
View File
@@ -5,6 +5,8 @@
package service
import (
"context"
"go.uber.org/zap"
"github.com/ShukeBta/MediaStationGo/internal/config"
@@ -14,20 +16,31 @@ import (
// Container holds every service initialized at startup. Handlers receive a
// pointer to it and pick the relevant fields.
type Container struct {
Cfg *config.Config
Log *zap.Logger
Repo *repository.Container
WSHub *Hub
Auth *AuthService
Media *MediaService
Scan *ScannerService
Stream *StreamService
Transcoder *TranscoderService
FFprobe *FFprobeService
TMDb *TMDbProvider
Scraper *ScraperService
Playback *PlaybackService
ImageProxy *ImageProxy
Cfg *config.Config
Log *zap.Logger
Repo *repository.Container
WSHub *Hub
Auth *AuthService
Media *MediaService
Scan *ScannerService
Stream *StreamService
Transcoder *TranscoderService
FFprobe *FFprobeService
TMDb *TMDbProvider
Bangumi *BangumiProvider
Scraper *ScraperService
Playback *PlaybackService
ImageProxy *ImageProxy
Watcher *WatcherService
Downloads *DownloadService
Subscription *SubscriptionService
Subtitle *SubtitleService
Stats *StatsService
Profile *ProfileService
Audit *AuditService
stopCtx context.Context
stopCancel context.CancelFunc
}
// New builds the service container.
@@ -37,30 +50,69 @@ func New(cfg *config.Config, log *zap.Logger, repos *repository.Container) *Cont
probe := NewFFprobeService(cfg, log)
tmdb := NewTMDbProvider(cfg, log)
scraper := NewScraperService(cfg, log, repos, tmdb, hub)
bangumi := NewBangumiProvider(cfg, log)
scraper := NewScraperService(cfg, log, repos, tmdb, bangumi, hub)
transcoder := NewTranscoderService(cfg, log, repos, hub)
scanner := NewScannerService(cfg, log, repos, hub, probe, scraper)
downloads := NewDownloadService(log, repos, hub)
subscription := NewSubscriptionService(log, repos, downloads, hub)
watcher := NewWatcherService(log, repos, scanner)
ctx, cancel := context.WithCancel(context.Background())
return &Container{
Cfg: cfg,
Log: log,
Repo: repos,
WSHub: hub,
Auth: NewAuthService(cfg, log, repos),
Media: NewMediaService(cfg, log, repos),
Scan: NewScannerService(cfg, log, repos, hub, probe, scraper),
Stream: NewStreamService(cfg, log, repos, transcoder),
Transcoder: transcoder,
FFprobe: probe,
TMDb: tmdb,
Scraper: scraper,
Playback: NewPlaybackService(log, repos),
ImageProxy: NewImageProxy(cfg, log),
Cfg: cfg,
Log: log,
Repo: repos,
WSHub: hub,
Auth: NewAuthService(cfg, log, repos),
Media: NewMediaService(cfg, log, repos),
Scan: scanner,
Stream: NewStreamService(cfg, log, repos, transcoder),
Transcoder: transcoder,
FFprobe: probe,
TMDb: tmdb,
Bangumi: bangumi,
Scraper: scraper,
Playback: NewPlaybackService(log, repos),
ImageProxy: NewImageProxy(cfg, log),
Watcher: watcher,
Downloads: downloads,
Subscription: subscription,
Subtitle: NewSubtitleService(log, repos),
Stats: NewStatsService(log, repos),
Profile: NewProfileService(log, repos),
Audit: NewAuditService(log, repos),
stopCtx: ctx,
stopCancel: cancel,
}
}
// Boot kicks off background workers (watcher, downloads poller,
// subscription scheduler). Called once after AutoMigrate.
func (c *Container) Boot() {
if err := c.Watcher.Start(c.stopCtx); err != nil {
c.Log.Warn("watcher start failed", zap.Error(err))
}
c.Downloads.Start(c.stopCtx)
c.Subscription.Start(c.stopCtx)
}
// Close releases any resources held by services (websocket hub, ffmpeg
// transcodes).
// transcodes, fsnotify, background pollers).
func (c *Container) Close() {
if c.stopCancel != nil {
c.stopCancel()
}
if c.Watcher != nil {
c.Watcher.Stop()
}
if c.Subscription != nil {
c.Subscription.Stop()
}
if c.Downloads != nil {
c.Downloads.Stop()
}
if c.Transcoder != nil {
c.Transcoder.StopAll()
}
+113
View File
@@ -0,0 +1,113 @@
// Package service — statistics aggregator.
//
// StatsService computes the dashboard numbers for the admin / home page:
// - total libraries, media items, users
// - total disk size and durations
// - top recently-watched media
// - process metadata (CPU / memory) via gopsutil
package service
import (
"context"
"runtime"
"time"
"github.com/shirou/gopsutil/v3/cpu"
"github.com/shirou/gopsutil/v3/disk"
"github.com/shirou/gopsutil/v3/mem"
"go.uber.org/zap"
"github.com/ShukeBta/MediaStationGo/internal/model"
"github.com/ShukeBta/MediaStationGo/internal/repository"
)
// StatsService computes aggregate stats.
type StatsService struct {
log *zap.Logger
repo *repository.Container
}
// NewStatsService is the constructor.
func NewStatsService(log *zap.Logger, repo *repository.Container) *StatsService {
return &StatsService{log: log, repo: repo}
}
// Snapshot is the JSON returned by /api/stats.
type Snapshot struct {
Libraries int64 `json:"libraries"`
MediaCount int64 `json:"media_count"`
UsersCount int64 `json:"users_count"`
TotalSizeBytes int64 `json:"total_size_bytes"`
TotalSeconds int64 `json:"total_seconds"`
RecentlyAdded []model.Media `json:"recently_added"`
Hardware Hardware `json:"hardware"`
GeneratedAt time.Time `json:"generated_at"`
}
// Hardware is the live CPU / memory / disk readings.
type Hardware struct {
CPUPercent float64 `json:"cpu_percent"`
MemoryUsed uint64 `json:"memory_used"`
MemoryTotal uint64 `json:"memory_total"`
DiskUsed uint64 `json:"disk_used"`
DiskTotal uint64 `json:"disk_total"`
GoVersion string `json:"go_version"`
Goroutines int `json:"goroutines"`
}
// Compute builds a fresh snapshot.
func (s *StatsService) Compute(ctx context.Context, dataDir string) (*Snapshot, error) {
snap := &Snapshot{GeneratedAt: time.Now()}
if err := s.repo.DB.Model(&model.Library{}).Count(&snap.Libraries).Error; err != nil {
return nil, err
}
if err := s.repo.DB.Model(&model.Media{}).Count(&snap.MediaCount).Error; err != nil {
return nil, err
}
if err := s.repo.DB.Model(&model.User{}).Count(&snap.UsersCount).Error; err != nil {
return nil, err
}
type sumRow struct {
Size int64
Seconds int64
}
var sum sumRow
if err := s.repo.DB.Model(&model.Media{}).
Select("COALESCE(SUM(size_bytes),0) as size, COALESCE(SUM(duration_sec),0) as seconds").
Scan(&sum).Error; err != nil {
return nil, err
}
snap.TotalSizeBytes = sum.Size
snap.TotalSeconds = sum.Seconds
if err := s.repo.DB.Model(&model.Media{}).
Order("created_at desc").Limit(12).
Find(&snap.RecentlyAdded).Error; err != nil {
return nil, err
}
snap.Hardware = readHardware(dataDir)
return snap, nil
}
func readHardware(dataDir string) Hardware {
hw := Hardware{
GoVersion: runtime.Version(),
Goroutines: runtime.NumGoroutine(),
}
if usage, err := cpu.Percent(0, false); err == nil && len(usage) > 0 {
hw.CPUPercent = usage[0]
}
if v, err := mem.VirtualMemory(); err == nil {
hw.MemoryUsed = v.Used
hw.MemoryTotal = v.Total
}
if dataDir == "" {
dataDir = "/"
}
if d, err := disk.Usage(dataDir); err == nil {
hw.DiskUsed = d.Used
hw.DiskTotal = d.Total
}
return hw
}
+243
View File
@@ -0,0 +1,243 @@
// Package service — RSS subscriptions for automated downloads.
//
// SubscriptionService periodically polls every Subscription row, fetches
// the configured RSS / Atom feed, and queues new items into the
// DownloadService. Items are deduplicated by GUID stored as a Setting key
// "subscription.<id>.last_guid" so the same episode is never re-queued.
package service
import (
"context"
"encoding/xml"
"errors"
"fmt"
"io"
"net/http"
"regexp"
"strings"
"time"
"go.uber.org/zap"
"github.com/ShukeBta/MediaStationGo/internal/model"
"github.com/ShukeBta/MediaStationGo/internal/repository"
)
// SubscriptionService runs the polling loop.
type SubscriptionService struct {
log *zap.Logger
repo *repository.Container
downloads *DownloadService
hub *Hub
stop chan struct{}
}
// NewSubscriptionService is the constructor.
func NewSubscriptionService(log *zap.Logger, repo *repository.Container, downloads *DownloadService, hub *Hub) *SubscriptionService {
return &SubscriptionService{
log: log, repo: repo, downloads: downloads, hub: hub,
stop: make(chan struct{}),
}
}
// Start runs the polling loop in the background.
func (s *SubscriptionService) Start(ctx context.Context) {
go s.loop(ctx)
}
// Stop shuts the loop down.
func (s *SubscriptionService) Stop() { close(s.stop) }
// rssFeed is the minimal RSS subset we need to decode.
type rssFeed struct {
XMLName xml.Name `xml:"rss"`
Channel struct {
Items []struct {
Title string `xml:"title"`
Link string `xml:"link"`
GUID string `xml:"guid"`
Description string `xml:"description"`
Enclosure struct {
URL string `xml:"url,attr"`
} `xml:"enclosure"`
} `xml:"item"`
} `xml:"channel"`
}
// Create persists a new subscription.
func (s *SubscriptionService) Create(ctx context.Context, sub *model.Subscription) error {
if sub.Name == "" || sub.FeedURL == "" {
return errors.New("name and feed_url required")
}
return s.repo.Subscription.Create(ctx, sub)
}
// List returns every subscription rule.
func (s *SubscriptionService) List(ctx context.Context) ([]model.Subscription, error) {
return s.repo.Subscription.List(ctx)
}
// Delete removes a subscription.
func (s *SubscriptionService) Delete(ctx context.Context, id string) error {
return s.repo.DB.Where("id = ?", id).Delete(&model.Subscription{}).Error
}
// RunNow forces a poll for one subscription, ignoring its schedule. Used
// by the admin UI's "test now" button.
func (s *SubscriptionService) RunNow(ctx context.Context, id string) (int, error) {
var sub model.Subscription
if err := s.repo.DB.Where("id = ?", id).First(&sub).Error; err != nil {
return 0, err
}
return s.runOne(ctx, &sub)
}
// loop polls every 10 minutes.
func (s *SubscriptionService) loop(ctx context.Context) {
t := time.NewTicker(10 * time.Minute)
defer t.Stop()
// First run shortly after startup.
first := time.NewTimer(30 * time.Second)
defer first.Stop()
for {
select {
case <-ctx.Done():
return
case <-s.stop:
return
case <-first.C:
case <-t.C:
}
s.runAll(ctx)
}
}
func (s *SubscriptionService) runAll(ctx context.Context) {
subs, err := s.repo.Subscription.List(ctx)
if err != nil {
s.log.Warn("subscription list failed", zap.Error(err))
return
}
for i := range subs {
if !subs[i].Enabled {
continue
}
if n, err := s.runOne(ctx, &subs[i]); err != nil {
s.log.Warn("subscription run failed",
zap.String("name", subs[i].Name), zap.Error(err))
} else if n > 0 {
s.log.Info("subscription queued items",
zap.String("name", subs[i].Name), zap.Int("count", n))
}
}
}
func (s *SubscriptionService) runOne(ctx context.Context, sub *model.Subscription) (int, error) {
feed, err := s.fetch(ctx, sub.FeedURL)
if err != nil {
return 0, err
}
filter := compileFilter(sub.Filter)
guidKey := fmt.Sprintf("subscription.%s.seen", sub.ID)
seenRaw, _ := s.repo.Setting.Get(ctx, guidKey)
seen := splitNonEmpty(seenRaw)
seenSet := make(map[string]struct{}, len(seen))
for _, g := range seen {
seenSet[g] = struct{}{}
}
queued := 0
for _, item := range feed.Channel.Items {
guid := item.GUID
if guid == "" {
guid = item.Link
}
if _, ok := seenSet[guid]; ok {
continue
}
if filter != nil && !filter.MatchString(item.Title) {
continue
}
download := item.Enclosure.URL
if download == "" {
download = item.Link
}
if download == "" {
continue
}
if _, err := s.downloads.AddDownload(ctx, sub.UserID, download, ""); err != nil {
s.log.Warn("subscription enqueue failed",
zap.String("title", item.Title), zap.Error(err))
continue
}
queued++
seen = append(seen, guid)
}
// Remember the last 200 GUIDs so the seen set doesn't grow forever.
if len(seen) > 200 {
seen = seen[len(seen)-200:]
}
_ = s.repo.Setting.Set(ctx, guidKey, strings.Join(seen, "\n"))
now := time.Now()
_ = s.repo.DB.Model(sub).Updates(map[string]any{"last_run_at": &now}).Error
if queued > 0 {
s.hub.Publish("subscription", map[string]any{
"id": sub.ID,
"name": sub.Name,
"queued": queued,
})
}
return queued, nil
}
func (s *SubscriptionService) fetch(ctx context.Context, feedURL string) (*rssFeed, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, feedURL, nil)
if err != nil {
return nil, err
}
req.Header.Set("User-Agent", "MediaStationGo/0.1")
resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode >= 400 {
return nil, fmt.Errorf("rss %s: %d", feedURL, resp.StatusCode)
}
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, err
}
var f rssFeed
if err := xml.Unmarshal(body, &f); err != nil {
return nil, err
}
return &f, nil
}
func compileFilter(pat string) *regexp.Regexp {
pat = strings.TrimSpace(pat)
if pat == "" {
return nil
}
if r, err := regexp.Compile("(?i)" + pat); err == nil {
return r
}
return nil
}
func splitNonEmpty(s string) []string {
if s == "" {
return nil
}
out := make([]string, 0)
for _, p := range strings.Split(s, "\n") {
p = strings.TrimSpace(p)
if p != "" {
out = append(out, p)
}
}
return out
}
+227
View File
@@ -0,0 +1,227 @@
// Package service — subtitle handling.
//
// SubtitleService finds external subtitle files next to a media file and
// converts SRT to WebVTT on the fly so the browser <track> element can
// load them directly.
//
// External-subtitle discovery rules (matching MediaStation Python defaults):
//
// 1. Same directory, same basename, different extension.
// 2. Same directory, ".sub/" or "subs/" subdirectory.
// 3. Sibling languages e.g. movie.zh.srt / movie.en.srt → exposed as
// ?lang=zh / ?lang=en.
//
// Supported extensions: .srt, .ass, .ssa, .vtt.
package service
import (
"context"
"errors"
"fmt"
"io"
"os"
"path/filepath"
"regexp"
"strings"
"go.uber.org/zap"
"github.com/ShukeBta/MediaStationGo/internal/repository"
)
// SubtitleService is the discovery + conversion entry point.
type SubtitleService struct {
log *zap.Logger
repo *repository.Container
}
// NewSubtitleService is the constructor.
func NewSubtitleService(log *zap.Logger, repo *repository.Container) *SubtitleService {
return &SubtitleService{log: log, repo: repo}
}
// SubtitleTrack describes one external subtitle file.
type SubtitleTrack struct {
Lang string `json:"lang"`
Label string `json:"label"`
Path string `json:"path"`
URL string `json:"url"`
Codec string `json:"codec"`
}
// extToCodec maps the file extension to the inner codec name.
var extToCodec = map[string]string{
".srt": "srt",
".vtt": "vtt",
".ass": "ass",
".ssa": "ssa",
}
// Discover lists every external subtitle file for a media row. The URL is
// relative; the caller should prepend /api/subtitles/<media_id>?path=...
// when serializing for the frontend.
func (s *SubtitleService) Discover(ctx context.Context, mediaID string) ([]SubtitleTrack, error) {
m, err := s.repo.Media.FindByID(ctx, mediaID)
if err != nil {
return nil, err
}
if m == nil {
return nil, errors.New("media not found")
}
dir := filepath.Dir(m.Path)
base := strings.TrimSuffix(filepath.Base(m.Path), filepath.Ext(m.Path))
candidates := make([]string, 0, 16)
candidates = append(candidates, dir)
for _, sub := range []string{"subs", "Subs", "sub", ".sub"} {
candidates = append(candidates, filepath.Join(dir, sub))
}
var tracks []SubtitleTrack
for _, c := range candidates {
entries, err := os.ReadDir(c)
if err != nil {
continue
}
for _, e := range entries {
if e.IsDir() {
continue
}
ext := strings.ToLower(filepath.Ext(e.Name()))
codec, ok := extToCodec[ext]
if !ok {
continue
}
fullName := strings.TrimSuffix(e.Name(), ext)
if !strings.HasPrefix(strings.ToLower(fullName), strings.ToLower(base)) &&
c == dir {
// In the same directory we require a basename match;
// inside subs/ subdirs we accept anything.
continue
}
lang := detectLang(fullName, base)
tracks = append(tracks, SubtitleTrack{
Lang: lang,
Label: lang,
Path: filepath.Join(c, e.Name()),
Codec: codec,
})
}
}
return tracks, nil
}
// langTag matches the .zh / .zh-cn / .chs language sub-extensions.
var langTag = regexp.MustCompile(`(?i)\.([a-z]{2,3}(?:[-_][a-z]{2,4})?)$`)
func detectLang(name, base string) string {
suffix := strings.TrimPrefix(name, base)
suffix = strings.TrimPrefix(suffix, ".")
if m := langTag.FindStringSubmatch("." + suffix); len(m) >= 2 {
return strings.ToLower(m[1])
}
if suffix == "" {
return "und" // undetermined
}
return strings.ToLower(suffix)
}
// Serve writes the subtitle file as WebVTT (.vtt). SRT/SSA files are
// converted minimally on the fly. Returns ErrSubtitleNotFound when the
// path is rejected (path traversal / not in the media directory).
func (s *SubtitleService) Serve(ctx context.Context, mediaID, sub string, w io.Writer) error {
m, err := s.repo.Media.FindByID(ctx, mediaID)
if err != nil || m == nil {
return errors.New("media not found")
}
abs, err := filepath.Abs(sub)
if err != nil {
return err
}
mediaDir, _ := filepath.Abs(filepath.Dir(m.Path))
if !strings.HasPrefix(abs, mediaDir) {
return fmt.Errorf("path escape")
}
f, err := os.Open(abs)
if err != nil {
return err
}
defer f.Close()
body, err := io.ReadAll(f)
if err != nil {
return err
}
switch strings.ToLower(filepath.Ext(abs)) {
case ".vtt":
_, err = w.Write(body)
case ".srt":
_, err = w.Write([]byte(srtToVTT(string(body))))
case ".ass", ".ssa":
_, err = w.Write([]byte(assToVTT(string(body))))
default:
return errors.New("unsupported subtitle format")
}
return err
}
// srtToVTT performs the minimal SRT → WebVTT transformation: prepend
// "WEBVTT\n\n" and replace ',' with '.' in the timecode separators.
func srtToVTT(body string) string {
body = strings.ReplaceAll(body, "\r\n", "\n")
out := strings.Builder{}
out.WriteString("WEBVTT\n\n")
for _, line := range strings.Split(body, "\n") {
if strings.Contains(line, "-->") {
line = strings.ReplaceAll(line, ",", ".")
}
out.WriteString(line)
out.WriteByte('\n')
}
return out.String()
}
// assToVTT extracts the dialogue lines from an ASS/SSA subtitle. Styling
// is dropped — the goal is to produce something usable in <track> rather
// than a pixel-perfect render.
func assToVTT(body string) string {
out := strings.Builder{}
out.WriteString("WEBVTT\n\n")
for i, line := range strings.Split(body, "\n") {
line = strings.TrimSpace(line)
if !strings.HasPrefix(line, "Dialogue:") {
continue
}
parts := strings.SplitN(line, ",", 10)
if len(parts) < 10 {
continue
}
fmt.Fprintf(&out, "%d\n%s --> %s\n%s\n\n",
i,
normaliseTimecode(parts[1]),
normaliseTimecode(parts[2]),
stripASSTags(parts[9]),
)
}
return out.String()
}
func normaliseTimecode(t string) string {
t = strings.TrimSpace(t)
parts := strings.Split(t, ":")
if len(parts) != 3 {
return t
}
hh := parts[0]
if len(hh) == 1 {
hh = "0" + hh
}
return hh + ":" + parts[1] + ":" + strings.ReplaceAll(parts[2], ".", ".")
}
var assTag = regexp.MustCompile(`\{[^}]*\}`)
func stripASSTags(s string) string {
return assTag.ReplaceAllString(s, "")
}
+36
View File
@@ -0,0 +1,36 @@
package service
import (
"strings"
"testing"
)
func TestSrtToVTT(t *testing.T) {
in := "1\n00:00:01,000 --> 00:00:02,500\nHello world\n\n"
out := srtToVTT(in)
if !strings.HasPrefix(out, "WEBVTT") {
t.Fatalf("missing WEBVTT prefix: %q", out)
}
if !strings.Contains(out, "00:00:01.000 --> 00:00:02.500") {
t.Fatalf("comma timecode not converted: %q", out)
}
if !strings.Contains(out, "Hello world") {
t.Fatalf("dialogue lost: %q", out)
}
}
func TestStripASSTags(t *testing.T) {
cases := []struct {
in, want string
}{
{"{\\an8}hello", "hello"},
{"plain", "plain"},
{"a{\\fad(0,500)}b{\\b1}c", "abc"},
}
for _, c := range cases {
got := stripASSTags(c.in)
if got != c.want {
t.Errorf("stripASSTags(%q) = %q, want %q", c.in, got, c.want)
}
}
}
+4 -1
View File
@@ -61,9 +61,12 @@ func NewTMDbProvider(cfg *config.Config, log *zap.Logger) *TMDbProvider {
// Enabled reports whether the operator has supplied an API key.
func (t *TMDbProvider) Enabled() bool { return t.cfg.Secrets.TMDbAPIKey != "" }
// Match describes a successful metadata match.
// Match describes a successful metadata match. The same struct is reused
// across providers; provider-specific IDs sit side-by-side so the scraper
// orchestrator can write them all into a single update.
type Match struct {
TMDbID int `json:"tmdb_id"`
BangumiID int `json:"bangumi_id"`
Title string `json:"title"`
Overview string `json:"overview"`
PosterURL string `json:"poster_url"`
+194
View File
@@ -0,0 +1,194 @@
// Package service — filesystem watcher.
//
// WatcherService observes every enabled library root with fsnotify and
// debounces incoming events into per-library re-scans. New / renamed
// files become Media rows; deletes remove them.
//
// The watcher runs in the background and is started after migrations
// complete. It survives library add / delete via Refresh().
package service
import (
"context"
"path/filepath"
"sync"
"time"
"github.com/fsnotify/fsnotify"
"go.uber.org/zap"
"github.com/ShukeBta/MediaStationGo/internal/repository"
)
// WatcherService is a thin orchestrator on top of fsnotify.
type WatcherService struct {
log *zap.Logger
repo *repository.Container
scanner *ScannerService
mu sync.Mutex
watcher *fsnotify.Watcher
watched map[string]string // dir -> libraryID
pending map[string]time.Time
stop chan struct{}
}
// NewWatcherService is the constructor.
func NewWatcherService(log *zap.Logger, repo *repository.Container, scanner *ScannerService) *WatcherService {
return &WatcherService{
log: log,
repo: repo,
scanner: scanner,
watched: make(map[string]string),
pending: make(map[string]time.Time),
stop: make(chan struct{}),
}
}
// Start initialises the underlying fsnotify watcher and registers every
// library root currently in the database.
func (w *WatcherService) Start(ctx context.Context) error {
fw, err := fsnotify.NewWatcher()
if err != nil {
return err
}
w.watcher = fw
if err := w.Refresh(ctx); err != nil {
w.log.Warn("watcher refresh failed", zap.Error(err))
}
go w.loop(ctx)
go w.debouncer(ctx)
return nil
}
// Stop tears down the watcher (called on graceful shutdown).
func (w *WatcherService) Stop() {
close(w.stop)
if w.watcher != nil {
_ = w.watcher.Close()
}
}
// Refresh reads the library list and adjusts the set of watched
// directories. Idempotent — safe to call after every CRUD.
func (w *WatcherService) Refresh(ctx context.Context) error {
libs, err := w.repo.Library.List(ctx)
if err != nil {
return err
}
w.mu.Lock()
defer w.mu.Unlock()
current := make(map[string]string)
for _, l := range libs {
if !l.Enabled {
continue
}
current[l.Path] = l.ID
}
// Remove disappeared paths.
for path := range w.watched {
if _, ok := current[path]; !ok {
_ = w.watcher.Remove(path)
delete(w.watched, path)
}
}
// Add new ones (top-level only — fsnotify is non-recursive).
for path, id := range current {
if _, ok := w.watched[path]; ok {
continue
}
if err := w.watcher.Add(path); err != nil {
w.log.Warn("watch add failed", zap.String("path", path), zap.Error(err))
continue
}
w.watched[path] = id
}
return nil
}
// loop drains fsnotify events and pushes the affected library into the
// pending map. The actual rescan happens in the debouncer goroutine.
func (w *WatcherService) loop(ctx context.Context) {
if w.watcher == nil {
return
}
for {
select {
case <-ctx.Done():
return
case <-w.stop:
return
case ev, ok := <-w.watcher.Events:
if !ok {
return
}
if ev.Op&(fsnotify.Create|fsnotify.Remove|fsnotify.Rename|fsnotify.Write) == 0 {
continue
}
lib := w.findLibrary(ev.Name)
if lib == "" {
continue
}
w.mu.Lock()
w.pending[lib] = time.Now()
w.mu.Unlock()
case err, ok := <-w.watcher.Errors:
if !ok {
return
}
w.log.Warn("watcher error", zap.Error(err))
}
}
}
// findLibrary maps a path back to the watching library ID, taking the
// shortest matching prefix.
func (w *WatcherService) findLibrary(path string) string {
w.mu.Lock()
defer w.mu.Unlock()
dir := filepath.Dir(path)
for {
if id, ok := w.watched[dir]; ok {
return id
}
parent := filepath.Dir(dir)
if parent == dir {
return ""
}
dir = parent
}
}
// debouncer drains the pending set every 5 s and triggers a rescan per
// library. Coalescing avoids storming the disk on bulk operations
// (mass-rename, large copies).
func (w *WatcherService) debouncer(ctx context.Context) {
t := time.NewTicker(5 * time.Second)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-w.stop:
return
case <-t.C:
}
w.mu.Lock()
due := make([]string, 0, len(w.pending))
now := time.Now()
for id, ts := range w.pending {
if now.Sub(ts) >= 5*time.Second {
due = append(due, id)
delete(w.pending, id)
}
}
w.mu.Unlock()
for _, id := range due {
w.log.Info("watcher triggered rescan", zap.String("library_id", id))
if _, err := w.scanner.ScanLibrary(ctx, id); err != nil {
w.log.Warn("watcher rescan failed", zap.Error(err))
}
}
}
}