This commit is contained in:
truewhile
2026-08-27 23:40:12 +08:00
parent d3051eaffe
commit 6e8eac9887
26 changed files with 2018 additions and 95 deletions
+18
View File
@@ -20,6 +20,24 @@ func registerAdminRoutes(api *gin.RouterGroup, cfg *config.Config, svc *service.
registerAdminAPIConfigRoutes(admin, svc)
registerAdminRecognitionWordRoutes(admin, svc)
registerAdminStrmRoutes(admin, svc)
registerAdminScraperRoutes(admin, svc)
}
func registerAdminScraperRoutes(admin *gin.RouterGroup, svc *service.Container) {
admin.GET("/scraper/queue", listScrapeQueueHandler(svc))
admin.POST("/scraper/queue/:id/cancel", cancelScrapeTaskHandler(svc))
admin.POST("/scraper/queue/:id/retry", retryScrapeTaskHandler(svc))
admin.DELETE("/scraper/queue/:id", deleteScrapeTaskHandler(svc))
admin.POST("/scraper/queue/batch", batchActionScrapeTasksHandler(svc))
admin.POST("/scraper/queue/clear-done", clearDoneScrapeTasksHandler(svc))
admin.POST("/scraper/queue/clear-finished", clearFinishedScrapeTasksHandler(svc))
admin.POST("/scraper/queue/clear-canceled", clearCanceledScrapeTasksHandler(svc))
admin.POST("/scraper/queue/retry-failed", retryAllFailedScrapeTasksHandler(svc))
admin.POST("/scraper/queue/cancel-pending", cancelPendingScrapeTasksHandler(svc))
admin.POST("/scraper/queue/enqueue-library/:id", enqueueLibraryScrapeHandler(svc))
admin.POST("/scraper/queue/enqueue-all", enqueueAllScrapeHandler(svc))
admin.POST("/media/repair-rescrape", enqueueAllScrapeHandler(svc))
admin.POST("/libraries/:id/repair-rescrape", enqueueLibraryScrapeHandler(svc))
}
func registerAdminStrmRoutes(admin *gin.RouterGroup, svc *service.Container) {
+167
View File
@@ -0,0 +1,167 @@
package handler
import (
"net/http"
"strconv"
"github.com/gin-gonic/gin"
"github.com/ShukeBta/MMTL/internal/service"
)
func listScrapeQueueHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
page, _ := strconv.Atoi(c.DefaultQuery("page", "1"))
pageSize, _ := strconv.Atoi(c.DefaultQuery("page_size", "50"))
snap, err := svc.Scraper.ScrapeQueueSnapshot(c.Request.Context(), c.Query("status"), page, pageSize)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, snap)
}
}
func cancelScrapeTaskHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
if err := svc.Scraper.CancelScrapeTask(c.Request.Context(), c.Param("id")); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"ok": true})
}
}
func retryScrapeTaskHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
if err := svc.Scraper.RetryScrapeTask(c.Request.Context(), c.Param("id")); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"ok": true})
}
}
func deleteScrapeTaskHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
if err := svc.Scraper.DeleteScrapeTask(c.Request.Context(), c.Param("id")); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"ok": true})
}
}
func batchActionScrapeTasksHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
var req queueBatchReq
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
n, err := svc.Scraper.BatchActionScrapeTasks(c.Request.Context(), req.Action, req.IDs)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"affected": n, "action": req.Action})
}
}
func clearDoneScrapeTasksHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
n, err := svc.Scraper.ClearDoneScrapeTasks(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"deleted": n})
}
}
func clearFinishedScrapeTasksHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
n, err := svc.Scraper.ClearFinishedScrapeTasks(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"deleted": n})
}
}
func clearCanceledScrapeTasksHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
n, err := svc.Scraper.ClearCanceledScrapeTasks(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"deleted": n})
}
}
func retryAllFailedScrapeTasksHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
n, err := svc.Scraper.RetryAllFailedScrapeTasks(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"retried": n})
}
}
func cancelPendingScrapeTasksHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
n, err := svc.Scraper.CancelPendingScrapeTasks(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"canceled": n})
}
}
type enqueueScrapeReq struct {
EpisodeImages bool `json:"episode_images"`
EpisodeArtwork bool `json:"episode_artwork"`
RefreshMatched bool `json:"refresh_matched"`
IncludeMatched bool `json:"include_matched"`
}
func (r enqueueScrapeReq) toOptions() service.ScrapeOptions {
epArtwork := r.EpisodeImages || r.EpisodeArtwork
return service.ScrapeOptions{
EpisodeArtwork: &epArtwork,
IncludeMatched: r.IncludeMatched || r.RefreshMatched,
RetryNoMatch: true,
}
}
func enqueueLibraryScrapeHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
var req enqueueScrapeReq
_ = c.ShouldBindJSON(&req)
libID := c.Param("id")
n, err := svc.Scraper.EnqueueLibrary(c.Request.Context(), libID, req.toOptions())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"enqueued": n})
}
}
func enqueueAllScrapeHandler(svc *service.Container) gin.HandlerFunc {
return func(c *gin.Context) {
var req enqueueScrapeReq
_ = c.ShouldBindJSON(&req)
n, err := svc.Scraper.EnqueueAll(c.Request.Context(), req.toOptions())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"enqueued": n})
}
}
+8 -53
View File
@@ -2,7 +2,6 @@
package handler
import (
"context"
"errors"
"io"
"net/http"
@@ -135,28 +134,12 @@ func scrapeOneHandler(svc *service.Container) gin.HandlerFunc {
return
}
options.IncludeMatched = true
m, err := svc.Repo.Media.FindByID(c.Request.Context(), c.Param("id"))
if err != nil || m == nil {
c.JSON(http.StatusNotFound, gin.H{"error": "not found"})
return
}
task := startScrapeHTTPTask(svc, "手动刮削媒体", m.Title, m.Path)
if err := svc.Scraper.EnrichOneWithOptions(c.Request.Context(), m, options); err != nil {
finishHTTPTask(task, err, "scrape", "手动刮削媒体失败", nil, nil)
task, err := svc.Scraper.EnqueueMedia(c.Request.Context(), c.Param("id"), options)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
reclassified := reclassifyMediaAfterScrape(c.Request.Context(), svc, m.ID)
refreshed, _ := svc.Repo.Media.FindByID(c.Request.Context(), m.ID)
metrics := map[string]int64{"processed": 1}
if refreshed != nil && refreshed.ScrapeStatus == "matched" {
metrics["matched"] = 1
}
if reclassified > 0 {
metrics["reclassified"] = int64(reclassified)
}
finishHTTPTask(task, nil, "completed", "手动刮削媒体结束", metrics, nil)
c.JSON(http.StatusOK, refreshed)
c.JSON(http.StatusOK, task)
}
}
@@ -170,40 +153,12 @@ func scrapeLibraryHandler(svc *service.Container) gin.HandlerFunc {
return
}
options.IncludeMatched = true
var task *service.TaskHandle
if lib, err := svc.Repo.Library.FindByID(c.Request.Context(), libID); err == nil && lib != nil {
task = startScrapeHTTPTask(svc, "手动刮削媒体库", lib.Name, lib.Path)
} else {
task = startScrapeHTTPTask(svc, "手动刮削媒体库", libID, "")
n, err := svc.Scraper.EnqueueLibrary(c.Request.Context(), libID, options)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
// Run in the background so HTTP returns instantly; the WS hub
// pushes per-item progress on the "scrape" topic.
go func(libID string, task *service.TaskHandle, options service.ScrapeOptions) {
result, err := svc.Scraper.EnrichLibraryDetailedWithOptions(context.Background(), libID, options)
reclassified := 0
if result.Processed > 0 {
reclassified = reclassifyLibraryAfterScrape(context.Background(), svc, libID)
}
metrics := map[string]int64{
"matched": int64(result.Matched),
"processed": int64(result.Processed),
"candidates": int64(result.Candidates),
}
if reclassified > 0 {
metrics["reclassified"] = int64(reclassified)
}
if result.Failed > 0 {
metrics["errors"] = int64(result.Failed)
}
stage := "completed"
message := "手动刮削媒体库结束"
if err != nil {
stage = "scrape"
message = "手动刮削媒体库失败"
}
finishHTTPTask(task, err, stage, message, metrics, nil)
}(libID, task, options)
c.JSON(http.StatusAccepted, gin.H{"status": "scraping"})
c.JSON(http.StatusOK, gin.H{"status": "queued", "enqueued": n})
}
}
+1 -1
View File
@@ -9,7 +9,7 @@ type Library struct {
CoverURL string `gorm:"size:1024" json:"cover_url,omitempty"`
Enabled bool `gorm:"default:true" json:"enabled"`
SortOrder int `gorm:"index;default:0" json:"sort_order"` // 手动拖拽排序用,越小越靠前
CarouselEnabled bool `gorm:"default:true" json:"carousel_enabled"` // 是否参与首页海报轮播
CarouselEnabled bool `gorm:"default:false" json:"carousel_enabled"` // 是否参与首页海报轮播(默认不参与)
Roots []LibraryRoot `gorm:"foreignKey:LibraryID" json:"roots,omitempty"`
}
+1
View File
@@ -57,5 +57,6 @@ func AllModels() []interface{} {
&StrmDownloadTask{},
&StrmUploadTask{},
&StrmDirCache{},
&ScrapeTask{},
}
}
+34
View File
@@ -0,0 +1,34 @@
package model
import "time"
const (
ScrapeTaskPending = "pending"
ScrapeTaskRunning = "running"
ScrapeTaskDone = "done"
ScrapeTaskFailed = "failed"
ScrapeTaskCanceled = "canceled"
)
// ScrapeTask 表示一条持久化的媒体刮削任务。
type ScrapeTask struct {
Base
MediaID string `gorm:"index;size:36" json:"media_id"`
LibraryID string `gorm:"index;size:36" json:"library_id"`
LibraryName string `gorm:"size:128" json:"library_name"`
MediaTitle string `gorm:"size:255;not null" json:"media_title"`
MediaPath string `gorm:"size:1024;not null" json:"media_path"`
MediaType string `gorm:"size:16" json:"media_type"` // movie / tv / anime / adult
Provider string `gorm:"size:32" json:"provider"` // tmdb / douban / bangumi / thetvdb / metatube
MatchedTitle string `gorm:"size:255" json:"matched_title"`
MatchedYear int `json:"matched_year"`
PosterURL string `gorm:"size:1024" json:"poster_url"`
BackdropURL string `gorm:"size:1024" json:"backdrop_url"`
Status string `gorm:"index;size:16;default:pending" json:"status"` // pending / running / done / failed / canceled
Error string `gorm:"type:text" json:"error"`
RetryCount int `gorm:"default:0" json:"retry_count"`
EpisodeImages bool `gorm:"default:true" json:"episode_images"`
RefreshMatched bool `gorm:"default:false" json:"refresh_matched"`
StartedAt *time.Time `json:"started_at,omitempty"`
FinishedAt *time.Time `json:"finished_at,omitempty"`
}
+10
View File
@@ -15,6 +15,11 @@ type LibraryRepository struct{ db *gorm.DB }
// Create persists a new library row.
func (r *LibraryRepository) Create(ctx context.Context, l *model.Library) error {
if l != nil && l.SortOrder == 0 {
var maxSort int
_ = r.db.WithContext(ctx).Model(&model.Library{}).Select("COALESCE(MAX(sort_order), -1)").Scan(&maxSort)
l.SortOrder = maxSort + 1
}
return r.db.WithContext(ctx).Create(l).Error
}
@@ -23,6 +28,11 @@ func (r *LibraryRepository) CreateWithRoots(ctx context.Context, l *model.Librar
return r.Create(ctx, l)
}
return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if l != nil && l.SortOrder == 0 {
var maxSort int
_ = tx.Model(&model.Library{}).Select("COALESCE(MAX(sort_order), -1)").Scan(&maxSort)
l.SortOrder = maxSort + 1
}
if err := tx.Create(l).Error; err != nil {
return err
}
+5 -3
View File
@@ -30,9 +30,10 @@ type Container struct {
StrmSyncPath *StrmSyncPathRepository
StrmSyncRecord *StrmSyncRecordRepository
StrmDownload *StrmDownloadTaskRepository
StrmUpload *StrmUploadTaskRepository
StrmDirCache *StrmDirCacheRepository
}
StrmUpload *StrmUploadTaskRepository
StrmDirCache *StrmDirCacheRepository
ScrapeTask *ScrapeTaskRepository
}
// New 将每个 repository 连接到单个 *gorm.DB。
func New(db *gorm.DB) *Container {
@@ -60,5 +61,6 @@ func New(db *gorm.DB) *Container {
StrmDownload: &StrmDownloadTaskRepository{db: db},
StrmUpload: &StrmUploadTaskRepository{db: db},
StrmDirCache: &StrmDirCacheRepository{db: db},
ScrapeTask: &ScrapeTaskRepository{db: db},
}
}
@@ -0,0 +1,274 @@
package repository
import (
"context"
"errors"
"strings"
"sync"
"time"
"gorm.io/gorm"
"github.com/ShukeBta/MMTL/internal/model"
)
var scrapeClaimMu sync.Mutex
// ScrapeTaskRepository persists model.ScrapeTask.
type ScrapeTaskRepository struct{ db *gorm.DB }
func (r *ScrapeTaskRepository) Create(ctx context.Context, t *model.ScrapeTask) error {
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Create(t).Error
})
}
func (r *ScrapeTaskRepository) CreateBatch(ctx context.Context, tasks []model.ScrapeTask) error {
if len(tasks) == 0 {
return nil
}
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).CreateInBatches(tasks, 100).Error
})
}
func (r *ScrapeTaskRepository) FindByID(ctx context.Context, id string) (*model.ScrapeTask, error) {
var t model.ScrapeTask
err := r.db.WithContext(ctx).Where("id = ?", id).First(&t).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, nil
}
return &t, err
}
func (r *ScrapeTaskRepository) FindActiveByMediaID(ctx context.Context, mediaID string) (*model.ScrapeTask, error) {
var t model.ScrapeTask
err := r.db.WithContext(ctx).
Where("media_id = ? AND status IN ?", mediaID, []string{model.ScrapeTaskPending, model.ScrapeTaskRunning}).
First(&t).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, nil
}
return &t, err
}
func (r *ScrapeTaskRepository) List(ctx context.Context, status string, page, pageSize int) ([]model.ScrapeTask, int64, error) {
if page < 1 {
page = 1
}
if pageSize < 1 || pageSize > 200 {
pageSize = 50
}
q := r.db.WithContext(ctx).Model(&model.ScrapeTask{})
if strings.TrimSpace(status) != "" && status != "all" {
q = q.Where("status = ?", strings.TrimSpace(status))
}
var total int64
if err := q.Count(&total).Error; err != nil {
return nil, 0, err
}
var rows []model.ScrapeTask
err := q.Order("created_at desc").
Offset((page - 1) * pageSize).
Limit(pageSize).
Find(&rows).Error
return rows, total, err
}
func (r *ScrapeTaskRepository) CountByStatus(ctx context.Context) (map[string]int64, error) {
var rows []struct {
Status string
Count int64
}
err := r.db.WithContext(ctx).Model(&model.ScrapeTask{}).
Select("status, count(*) as count").
Group("status").Scan(&rows).Error
if err != nil {
return nil, err
}
out := map[string]int64{}
for _, row := range rows {
out[row.Status] = row.Count
}
return out, nil
}
// ClaimPending picks pending scrape tasks and marks them running.
func (r *ScrapeTaskRepository) ClaimPending(ctx context.Context, limit int) ([]model.ScrapeTask, error) {
scrapeClaimMu.Lock()
defer scrapeClaimMu.Unlock()
var rows []model.ScrapeTask
err := withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Where("status = ?", model.ScrapeTaskPending).
Order("created_at asc").Limit(limit).Find(&rows).Error; err != nil {
return err
}
if len(rows) == 0 {
return nil
}
ids := make([]string, 0, len(rows))
now := time.Now()
for i := range rows {
ids = append(ids, rows[i].ID)
rows[i].Status = model.ScrapeTaskRunning
rows[i].StartedAt = &now
}
return tx.Model(&model.ScrapeTask{}).Where("id IN ?", ids).
Updates(map[string]any{"status": model.ScrapeTaskRunning, "started_at": now}).Error
})
})
if err != nil {
return nil, err
}
return rows, nil
}
func (r *ScrapeTaskRepository) Update(ctx context.Context, t *model.ScrapeTask) error {
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Model(&model.ScrapeTask{}).Where("id = ?", t.ID).Updates(map[string]any{
"status": t.Status,
"error": t.Error,
"provider": t.Provider,
"matched_title": t.MatchedTitle,
"matched_year": t.MatchedYear,
"poster_url": t.PosterURL,
"backdrop_url": t.BackdropURL,
"retry_count": t.RetryCount,
"started_at": t.StartedAt,
"finished_at": t.FinishedAt,
"updated_at": time.Now(),
}).Error
})
}
func (r *ScrapeTaskRepository) Delete(ctx context.Context, id string) error {
return withSQLiteBusyRetry(ctx, func() error {
return r.db.WithContext(ctx).Unscoped().Where("id = ?", id).Delete(&model.ScrapeTask{}).Error
})
}
func (r *ScrapeTaskRepository) DeleteBatch(ctx context.Context, ids []string) (int64, error) {
if len(ids) == 0 {
return 0, nil
}
var count int64
err := withSQLiteBusyRetry(ctx, func() error {
res := r.db.WithContext(ctx).Unscoped().Where("id IN ?", ids).Delete(&model.ScrapeTask{})
count = res.RowsAffected
return res.Error
})
return count, err
}
func (r *ScrapeTaskRepository) RetryBatch(ctx context.Context, ids []string) (int64, error) {
if len(ids) == 0 {
return 0, nil
}
var count int64
err := withSQLiteBusyRetry(ctx, func() error {
res := r.db.WithContext(ctx).Model(&model.ScrapeTask{}).
Where("id IN ? AND status IN ?", ids, []string{model.ScrapeTaskFailed, model.ScrapeTaskCanceled}).
Updates(map[string]any{
"status": model.ScrapeTaskPending,
"error": "",
"retry_count": 0,
"started_at": nil,
"finished_at": nil,
"updated_at": time.Now(),
})
count = res.RowsAffected
return res.Error
})
return count, err
}
func (r *ScrapeTaskRepository) CancelBatch(ctx context.Context, ids []string) (int64, error) {
if len(ids) == 0 {
return 0, nil
}
now := time.Now()
var count int64
err := withSQLiteBusyRetry(ctx, func() error {
res := r.db.WithContext(ctx).Model(&model.ScrapeTask{}).
Where("id IN ? AND status IN ?", ids, []string{model.ScrapeTaskPending, model.ScrapeTaskRunning}).
Updates(map[string]any{
"status": model.ScrapeTaskCanceled,
"error": "已批量取消",
"finished_at": now,
"updated_at": now,
})
count = res.RowsAffected
return res.Error
})
return count, err
}
func (r *ScrapeTaskRepository) ClearDone(ctx context.Context) (int64, error) {
var count int64
err := withSQLiteBusyRetry(ctx, func() error {
res := r.db.WithContext(ctx).Unscoped().Where("status = ?", model.ScrapeTaskDone).Delete(&model.ScrapeTask{})
count = res.RowsAffected
return res.Error
})
return count, err
}
func (r *ScrapeTaskRepository) ClearFinished(ctx context.Context) (int64, error) {
var count int64
err := withSQLiteBusyRetry(ctx, func() error {
res := r.db.WithContext(ctx).Unscoped().Where("status IN ?", []string{model.ScrapeTaskDone, model.ScrapeTaskFailed, model.ScrapeTaskCanceled}).
Delete(&model.ScrapeTask{})
count = res.RowsAffected
return res.Error
})
return count, err
}
func (r *ScrapeTaskRepository) ClearCanceled(ctx context.Context) (int64, error) {
var count int64
err := withSQLiteBusyRetry(ctx, func() error {
res := r.db.WithContext(ctx).Unscoped().Where("status = ?", model.ScrapeTaskCanceled).Delete(&model.ScrapeTask{})
count = res.RowsAffected
return res.Error
})
return count, err
}
func (r *ScrapeTaskRepository) RetryAllFailed(ctx context.Context) (int64, error) {
var count int64
err := withSQLiteBusyRetry(ctx, func() error {
res := r.db.WithContext(ctx).Model(&model.ScrapeTask{}).
Where("status = ?", model.ScrapeTaskFailed).
Updates(map[string]any{
"status": model.ScrapeTaskPending,
"error": "",
"retry_count": 0,
"started_at": nil,
"finished_at": nil,
"updated_at": time.Now(),
})
count = res.RowsAffected
return res.Error
})
return count, err
}
func (r *ScrapeTaskRepository) CancelPending(ctx context.Context) (int64, error) {
now := time.Now()
var count int64
err := withSQLiteBusyRetry(ctx, func() error {
res := r.db.WithContext(ctx).Model(&model.ScrapeTask{}).
Where("status IN ?", []string{model.ScrapeTaskPending, model.ScrapeTaskRunning}).
Updates(map[string]any{
"status": model.ScrapeTaskCanceled,
"error": "已批量取消",
"finished_at": now,
"updated_at": now,
})
count = res.RowsAffected
return res.Error
})
return count, err
}
+8 -5
View File
@@ -32,18 +32,21 @@ func (s *ScraperService) manualTMDbCandidates(ctx context.Context, query string,
for _, typ := range manualTMDbSearchTypes(mediaType) {
switch typ {
case "movie":
if matches, err := s.tmdb.SearchMovieCandidates(ctx, query, year); err == nil {
if matches, err := s.tmdb.SearchMovieCandidates(ctx, query, year); err == nil && len(matches) > 0 {
for _, match := range matches {
out = append(out, manualTMDbCandidate{MediaType: "movie", Match: match})
}
}
case "tv":
if matches, err := s.tmdb.SearchTVCandidates(ctx, query, year); err == nil {
if matches, err := s.tmdb.SearchTVCandidates(ctx, query, year); err == nil && len(matches) > 0 {
for _, match := range matches {
out = append(out, manualTMDbCandidate{MediaType: "tv", Match: match})
}
}
}
if len(out) > 0 {
break
}
}
return out
}
@@ -70,15 +73,15 @@ func manualTMDbIDSearchTypes(mediaType string) []string {
func manualTMDbSearchTypes(mediaType string) []string {
if strings.TrimSpace(mediaType) == "" {
return []string{"movie", "tv"}
return []string{"tv", "movie"}
}
switch normalizeMediaType(mediaType, "", "") {
case "tv", "anime", "variety":
return []string{"tv", "movie"}
case "movie", "adult":
return []string{"movie"}
default:
return []string{"movie", "tv"}
default:
return []string{"tv", "movie"}
}
}
+3 -3
View File
@@ -179,9 +179,9 @@ func TestManualSearchFallsBackToMovieFolderForGenericQuery(t *testing.T) {
if len(results) != 1 || results[0].TMDbID != 27205 {
t.Fatalf("manual search results=%#v, want folder fallback candidate; queries=%v", results, queries)
}
if len(queries) < 2 || queries[0] != "00000" || queries[1] != "inception" {
t.Fatalf("manual search queries=%v, want explicit query then folder fallback", queries)
}
if len(queries) < 2 || queries[0] != "00000" || queries[len(queries)-1] != "inception" {
t.Fatalf("manual search queries=%v, want explicit query then folder fallback", queries)
}
}
func TestManualSearchReturnsMovieFallbackForTVTypedTMDbSearch(t *testing.T) {
+8 -1
View File
@@ -52,7 +52,14 @@ func (s *MediaService) CreateLibraryWithRootsAndCover(ctx context.Context, name,
s.invalidateMediaCache(ctx)
return lib, nil
}
lib := &model.Library{Name: strings.TrimSpace(name), Path: roots[0].Path, Type: kind, CoverURL: strings.TrimSpace(coverURL), Enabled: true}
lib := &model.Library{
Name: strings.TrimSpace(name),
Path: roots[0].Path,
Type: kind,
CoverURL: strings.TrimSpace(coverURL),
Enabled: true,
CarouselEnabled: false,
}
if err := s.repo.Library.CreateWithRoots(ctx, lib, roots); err != nil {
return nil, err
}
+354
View File
@@ -0,0 +1,354 @@
package service
import (
"context"
"errors"
"fmt"
"strings"
"sync"
"time"
"go.uber.org/zap"
"github.com/ShukeBta/MMTL/internal/model"
)
type ScrapeQueueCounts struct {
Pending int64 `json:"pending"`
Running int64 `json:"running"`
Done int64 `json:"done"`
Failed int64 `json:"failed"`
Canceled int64 `json:"canceled"`
}
type ScrapeQueueSnapshot struct {
Counts ScrapeQueueCounts `json:"counts"`
Tasks []model.ScrapeTask `json:"tasks"`
Total int64 `json:"total"`
Page int `json:"page"`
PageSize int `json:"page_size"`
}
// Start 启动刮削任务队列的后台消费者。
func (s *ScraperService) Start(ctx context.Context) {
if s == nil {
return
}
go s.queueWorker(ctx)
}
func (s *ScraperService) queueWorker(ctx context.Context) {
const claimBatch = 4
sem := make(chan struct{}, 2) // 最大并发刮削数:2
for {
select {
case <-ctx.Done():
return
default:
}
tasks, err := s.repo.ScrapeTask.ClaimPending(ctx, claimBatch)
if err != nil {
if s.log != nil {
s.log.Warn("claim pending scrape task failed", zap.Error(err))
}
sleepContext(ctx, 3*time.Second)
continue
}
if len(tasks) == 0 {
sleepContext(ctx, 2*time.Second)
continue
}
var wg sync.WaitGroup
for i := range tasks {
wg.Add(1)
go func(t *model.ScrapeTask) {
defer wg.Done()
select {
case <-ctx.Done():
return
case sem <- struct{}{}:
}
defer func() { <-sem }()
s.processScrapeTask(ctx, t)
}(&tasks[i])
}
wg.Wait()
}
}
func (s *ScraperService) processScrapeTask(ctx context.Context, task *model.ScrapeTask) {
media, err := s.repo.Media.FindByID(ctx, task.MediaID)
if err != nil || media == nil {
now := time.Now()
task.Status = model.ScrapeTaskFailed
task.Error = "媒体项已不存在或被删除"
task.FinishedAt = &now
_ = s.repo.ScrapeTask.Update(ctx, task)
return
}
epArtwork := task.EpisodeImages
options := ScrapeOptions{
EpisodeArtwork: &epArtwork,
IncludeMatched: task.RefreshMatched,
RetryNoMatch: true,
}
enrichErr := s.EnrichOneWithOptions(ctx, media, options)
now := time.Now()
task.FinishedAt = &now
refreshed, _ := s.repo.Media.FindByID(ctx, media.ID)
if refreshed != nil && refreshed.ScrapeStatus == "matched" {
task.Status = model.ScrapeTaskDone
task.Error = ""
task.MatchedTitle = refreshed.Title
task.MatchedYear = refreshed.Year
task.PosterURL = refreshed.PosterURL
task.BackdropURL = refreshed.BackdropURL
if refreshed.TMDbID > 0 {
task.Provider = "tmdb"
} else if strings.TrimSpace(refreshed.DoubanID) != "" {
task.Provider = "douban"
} else if refreshed.BangumiID > 0 {
task.Provider = "bangumi"
} else if strings.TrimSpace(refreshed.TheTVDBID) != "" {
task.Provider = "thetvdb"
} else {
task.Provider = "metatube"
}
} else {
task.Status = model.ScrapeTaskFailed
if enrichErr != nil {
task.Error = enrichErr.Error()
} else if refreshed != nil && refreshed.ScrapeStatus == "no_match" {
task.Error = "未搜索到匹配的元数据"
} else {
task.Error = "刮削未完成匹配"
}
}
_ = s.repo.ScrapeTask.Update(ctx, task)
if s.hub != nil {
s.hub.Publish("scraper_queue", map[string]any{
"task_id": task.ID,
"status": task.Status,
"title": task.MediaTitle,
})
}
}
func (s *ScraperService) mediaKind(m *model.Media, lib *model.Library) string {
if m == nil {
return ""
}
if lib != nil && lib.Type != "" {
return lib.Type
}
if mediaIsEpisodic(m, lib) {
return "tv"
}
return "movie"
}
// EnqueueMedia 把单个媒体项放入刮削队列。
func (s *ScraperService) EnqueueMedia(ctx context.Context, mediaID string, options ScrapeOptions) (*model.ScrapeTask, error) {
if s == nil || s.repo == nil {
return nil, errors.New("scraper service not initialized")
}
media, err := s.repo.Media.FindByID(ctx, mediaID)
if err != nil || media == nil {
return nil, errors.New("media not found")
}
if active, _ := s.repo.ScrapeTask.FindActiveByMediaID(ctx, mediaID); active != nil {
return active, nil
}
libName := ""
var lib *model.Library
if strings.TrimSpace(media.LibraryID) != "" {
lib, _ = s.repo.Library.FindByID(ctx, media.LibraryID)
if lib != nil {
libName = lib.Name
}
}
task := &model.ScrapeTask{
MediaID: media.ID,
LibraryID: media.LibraryID,
LibraryName: libName,
MediaTitle: media.Title,
MediaPath: media.Path,
MediaType: s.mediaKind(media, lib),
Status: model.ScrapeTaskPending,
EpisodeImages: options.episodeArtworkEnabled(),
RefreshMatched: options.IncludeMatched || options.RefreshWeakMatched,
}
if err := s.repo.ScrapeTask.Create(ctx, task); err != nil {
return nil, err
}
return task, nil
}
// EnqueueLibrary 把指定媒体库内的所有候选媒体批量推入刮削队列。
func (s *ScraperService) EnqueueLibrary(ctx context.Context, libraryID string, options ScrapeOptions) (int, error) {
if s == nil || s.repo == nil {
return 0, errors.New("scraper service not initialized")
}
lib, err := s.repo.Library.FindByID(ctx, libraryID)
if err != nil || lib == nil {
return 0, errors.New("library not found")
}
rows, err := s.scrapeCandidateRows(ctx, libraryID, options)
if err != nil {
return 0, err
}
if len(rows) == 0 {
return 0, nil
}
tasks := make([]model.ScrapeTask, 0, len(rows))
for _, m := range rows {
tasks = append(tasks, model.ScrapeTask{
MediaID: m.ID,
LibraryID: lib.ID,
LibraryName: lib.Name,
MediaTitle: m.Title,
MediaPath: m.Path,
MediaType: s.mediaKind(&m, lib),
Status: model.ScrapeTaskPending,
EpisodeImages: options.episodeArtworkEnabled(),
RefreshMatched: options.IncludeMatched || options.RefreshWeakMatched,
})
}
if err := s.repo.ScrapeTask.CreateBatch(ctx, tasks); err != nil {
return 0, err
}
return len(tasks), nil
}
// EnqueueAll 把所有已启用媒体库的媒体推入刮削队列。
func (s *ScraperService) EnqueueAll(ctx context.Context, options ScrapeOptions) (int, error) {
libs, err := s.repo.Library.List(ctx)
if err != nil {
return 0, err
}
total := 0
for _, lib := range libs {
if !lib.Enabled {
continue
}
n, err := s.EnqueueLibrary(ctx, lib.ID, options)
if err != nil {
if s.log != nil {
s.log.Warn("enqueue library for scrape failed", zap.String("library", lib.ID), zap.Error(err))
}
continue
}
total += n
}
return total, nil
}
func (s *ScraperService) ScrapeQueueSnapshot(ctx context.Context, status string, page, pageSize int) (*ScrapeQueueSnapshot, error) {
tasks, total, err := s.repo.ScrapeTask.List(ctx, status, page, pageSize)
if err != nil {
return nil, err
}
countsMap, err := s.repo.ScrapeTask.CountByStatus(ctx)
if err != nil {
return nil, err
}
snap := &ScrapeQueueSnapshot{
Counts: ScrapeQueueCounts{
Pending: countsMap[model.ScrapeTaskPending],
Running: countsMap[model.ScrapeTaskRunning],
Done: countsMap[model.ScrapeTaskDone],
Failed: countsMap[model.ScrapeTaskFailed],
Canceled: countsMap[model.ScrapeTaskCanceled],
},
Tasks: tasks,
Total: total,
Page: page,
PageSize: pageSize,
}
return snap, nil
}
func (s *ScraperService) CancelScrapeTask(ctx context.Context, id string) error {
task, err := s.repo.ScrapeTask.FindByID(ctx, id)
if err != nil || task == nil {
return errors.New("刮削任务不存在")
}
if task.Status != model.ScrapeTaskPending && task.Status != model.ScrapeTaskRunning {
return errors.New("任务已完成或已终止,无法取消")
}
now := time.Now()
task.Status = model.ScrapeTaskCanceled
task.Error = "已取消"
task.FinishedAt = &now
return s.repo.ScrapeTask.Update(ctx, task)
}
func (s *ScraperService) RetryScrapeTask(ctx context.Context, id string) error {
task, err := s.repo.ScrapeTask.FindByID(ctx, id)
if err != nil || task == nil {
return errors.New("刮削任务不存在")
}
if task.Status != model.ScrapeTaskFailed && task.Status != model.ScrapeTaskCanceled {
return errors.New("只有失败或已取消的任务可以重试")
}
task.Status = model.ScrapeTaskPending
task.Error = ""
task.RetryCount = 0
task.StartedAt = nil
task.FinishedAt = nil
return s.repo.ScrapeTask.Update(ctx, task)
}
func (s *ScraperService) DeleteScrapeTask(ctx context.Context, id string) error {
return s.repo.ScrapeTask.Delete(ctx, id)
}
func (s *ScraperService) BatchActionScrapeTasks(ctx context.Context, action string, ids []string) (int64, error) {
switch action {
case "delete":
return s.repo.ScrapeTask.DeleteBatch(ctx, ids)
case "retry":
return s.repo.ScrapeTask.RetryBatch(ctx, ids)
case "cancel":
return s.repo.ScrapeTask.CancelBatch(ctx, ids)
default:
return 0, fmt.Errorf("不支持的操作: %s", action)
}
}
func (s *ScraperService) ClearDoneScrapeTasks(ctx context.Context) (int64, error) {
return s.repo.ScrapeTask.ClearDone(ctx)
}
func (s *ScraperService) ClearFinishedScrapeTasks(ctx context.Context) (int64, error) {
return s.repo.ScrapeTask.ClearFinished(ctx)
}
func (s *ScraperService) ClearCanceledScrapeTasks(ctx context.Context) (int64, error) {
return s.repo.ScrapeTask.ClearCanceled(ctx)
}
func (s *ScraperService) RetryAllFailedScrapeTasks(ctx context.Context) (int64, error) {
return s.repo.ScrapeTask.RetryAllFailed(ctx)
}
func (s *ScraperService) CancelPendingScrapeTasks(ctx context.Context) (int64, error) {
return s.repo.ScrapeTask.CancelPending(ctx)
}
+6 -1
View File
@@ -102,7 +102,12 @@ func (c *Container) Boot() {
c.Strm.Start(c.stopCtx)
}
// Mgo 保号规则巡检:默认关闭,由管理员通过 Telegram Bot 命令开启。
// 启动刮削队列后台消费者
if c.Scraper != nil {
c.Scraper.Start(c.stopCtx)
}
// Mgo 保号规则巡检:默认关闭,由管理员通过 Telegram Bot 命令开启。
// 每天触发一次评估;规则里的窗口可随机,不固定。
if c.Device != nil {
go c.runInactivitySweeper(c.stopCtx)
+16 -4
View File
@@ -57,7 +57,7 @@ func (t *TMDbProvider) searchMovieCandidates(ctx context.Context, query string,
apiKey := t.resolveAPIKey(ctx)
if apiKey == "" {
return nil, nil
return nil, errors.New("TMDb API Key 未配置,请先在「系统设置 → API配置」中填写")
}
base := t.resolveBaseURL(ctx)
@@ -65,7 +65,7 @@ func (t *TMDbProvider) searchMovieCandidates(ctx context.Context, query string,
q.Set("api_key", apiKey)
q.Set("query", query)
q.Set("language", language)
q.Set("include_adult", "false")
q.Set("include_adult", "true")
if year > 0 {
q.Set("year", fmt.Sprintf("%d", year))
}
@@ -79,6 +79,12 @@ func (t *TMDbProvider) searchMovieCandidates(ctx context.Context, query string,
if err := t.getJSON(ctx, u, &p); err != nil {
return nil, err
}
if len(p.Results) == 0 && year > 0 {
// 移除年份限制重试一次,避免年份微小差异(如 2019 vs 2020)导致无搜索结果
q.Del("year")
u = base + "/search/movie?" + q.Encode()
_ = t.getJSON(ctx, u, &p)
}
if len(p.Results) == 0 {
return nil, nil
}
@@ -137,7 +143,7 @@ func (t *TMDbProvider) searchTVCandidates(ctx context.Context, query string, yea
apiKey := t.resolveAPIKey(ctx)
if apiKey == "" {
return nil, nil
return nil, errors.New("TMDb API Key 未配置,请先在「系统设置 → API配置」中填写")
}
base := t.resolveBaseURL(ctx)
@@ -145,7 +151,7 @@ func (t *TMDbProvider) searchTVCandidates(ctx context.Context, query string, yea
q.Set("api_key", apiKey)
q.Set("query", query)
q.Set("language", language)
q.Set("include_adult", "false")
q.Set("include_adult", "true")
if year > 0 {
q.Set("first_air_date_year", fmt.Sprintf("%d", year))
}
@@ -159,6 +165,12 @@ func (t *TMDbProvider) searchTVCandidates(ctx context.Context, query string, yea
if err := t.getJSON(ctx, u, &p); err != nil {
return nil, err
}
if len(p.Results) == 0 && year > 0 {
// 移除年份限制重试一次,避免年份微小差异导致无搜索结果
q.Del("first_air_date_year")
u = base + "/search/tv?" + q.Encode()
_ = t.getJSON(ctx, u, &p)
}
if len(p.Results) == 0 {
return nil, nil
}