From 0aff87c69d033a57f81709a49fc2e9f27ad96233 Mon Sep 17 00:00:00 2001 From: truewhile <62226914+truewhile@users.noreply.github.com> Date: Mon, 24 Aug 2026 17:00:18 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 优化 --- go.mod | 1 + go.sum | 2 + internal/service/adult_scraper_test.go | 28 ++-- internal/service/api_config.go | 6 +- internal/service/api_config_connection.go | 10 +- internal/service/cloud/cloud_115_test.go | 2 +- internal/service/cloud/pan115_openapi.go | 2 +- internal/service/cloud115/client.go | 113 ++++++++----- internal/service/cloud115/cloud115_test.go | 54 +++++++ internal/service/cloud115/const.go | 4 +- internal/service/cloud115/queue_executor.go | 106 ++++++++++++ internal/service/cloud115/throttle_manager.go | 119 ++++++++++++++ internal/service/cloud115/url_cache.go | 2 +- internal/service/cloud115/url_cache_test.go | 2 +- .../service/cloud_library_display_stubs.go | 1 + internal/service/crypto.go | 2 +- internal/service/danmaku_service.go | 24 +-- internal/service/dlna.go | 16 +- internal/service/download_helpers.go | 2 +- internal/service/emby_compat.go | 8 +- internal/service/emby_media_sources.go | 6 +- internal/service/image_proxy_remote_fetch.go | 14 +- internal/service/organizer_paths.go | 13 +- .../organizer_scrape_providers_test.go | 14 +- internal/service/organizer_target_template.go | 20 +-- internal/service/permission.go | 8 +- internal/service/scanner.go | 14 +- internal/service/scraper_local_metadata.go | 20 +-- internal/service/scraper_test.go | 6 +- .../service/scraper_write_metadata_files.go | 75 ++++----- .../scraper_write_metadata_files_test.go | 12 +- internal/service/sse_hub.go | 12 +- internal/service/strm_queue.go | 6 + internal/service/strm_queue_test.go | 2 +- internal/service/strm_service.go | 153 +++++++++++++++++- internal/service/strm_sync.go | 30 ++-- internal/service/strm_sync_test.go | 138 ++++++++++++++++ internal/service/subtitle_test.go | 10 +- .../service/task_tracker_organize_items.go | 8 +- 39 files changed, 846 insertions(+), 219 deletions(-) create mode 100644 internal/service/cloud115/queue_executor.go create mode 100644 internal/service/cloud115/throttle_manager.go diff --git a/go.mod b/go.mod index 213316e..69560fd 100644 --- a/go.mod +++ b/go.mod @@ -72,6 +72,7 @@ require ( golang.org/x/exp v0.0.0-20230905200255-921286631fa9 // indirect golang.org/x/net v0.21.0 // indirect golang.org/x/text v0.20.0 // indirect + golang.org/x/time v0.15.0 // indirect google.golang.org/protobuf v1.31.0 // indirect gopkg.in/ini.v1 v1.67.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index 31db162..3445a7a 100644 --- a/go.sum +++ b/go.sum @@ -177,6 +177,8 @@ golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y= golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/text v0.20.0 h1:gK/Kv2otX8gz+wn7Rmb3vT96ZwuoxnQlY+HlJVj7Qug= golang.org/x/text v0.20.0/go.mod h1:D4IsuqiFMhST5bX19pQ9ikHC2GsaKyk/oF+pn3ducp4= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8= diff --git a/internal/service/adult_scraper_test.go b/internal/service/adult_scraper_test.go index 4d1a119..cbbbdf3 100644 --- a/internal/service/adult_scraper_test.go +++ b/internal/service/adult_scraper_test.go @@ -36,13 +36,13 @@ func TestParseAdultDetailHTML(t *testing.T) {
日期 2024-05-01
` - got := parseAdultDetailHTML(html, "SSIS-001", "javdb", "https://javdb.com/v/abc") - if got == nil { - t.Fatal("parseAdultDetailHTML returned nil") - } - if got.Title != "SSIS-001-测试标题" || got.OriginalName != "SSIS-001" || !got.NSFW { - t.Fatalf("unexpected metadata: %+v", got) - } + got := parseAdultDetailHTML(html, "SSIS-001", "javdb", "https://javdb.com/v/abc") + if got == nil { + t.Fatal("parseAdultDetailHTML returned nil") + } + if got.Title != "SSIS-001-测试标题" || got.OriginalName != "SSIS-001" || !got.NSFW { + t.Fatalf("unexpected metadata: %+v", got) + } if got.PosterURL != "https://javdb.com/covers/ssis001.jpg" || got.BackdropURL != "https://javdb.com/samples/1.jpg" { t.Fatalf("unexpected artwork: %+v", got) } @@ -98,9 +98,9 @@ func TestAdultProviderUsesConfiguredMultipleSources(t *testing.T) { if err != nil { t.Fatal(err) } - if match == nil || match.Title != "SSIS-001-多源命中标题" || match.OriginalName != "SSIS-001" || !match.NSFW { - t.Fatalf("multi-source adult match = %+v", match) - } + if match == nil || match.Title != "SSIS-001-多源命中标题" || match.OriginalName != "SSIS-001" || !match.NSFW { + t.Fatalf("multi-source adult match = %+v", match) + } } func TestAdultSourceKindRecognizesJavBusMirrors(t *testing.T) { @@ -135,12 +135,12 @@ func TestAdultProviderDefaultBases(t *testing.T) { if len(got) != len(want) { t.Fatalf("resolveBases len = %d, want %d: %v", len(got), len(want), got) } - for i := range want { - if got[i] != want[i] { - t.Fatalf("resolveBases[%d] = %q, want %q", i, got[i], want[i]) - } + for i := range want { + if got[i] != want[i] { + t.Fatalf("resolveBases[%d] = %q, want %q", i, got[i], want[i]) } } +} func TestParseAdultDetailHTMLRejectsAgeVerification(t *testing.T) { verifyHTML := ` diff --git a/internal/service/api_config.go b/internal/service/api_config.go index 6197855..ed8e7c7 100644 --- a/internal/service/api_config.go +++ b/internal/service/api_config.go @@ -41,9 +41,9 @@ func (s *APIConfigService) SeedDefaults(ctx context.Context) error { {Provider: "thetvdb", BaseURL: "https://api4.thetvdb.com/v4", Description: "TheTVDB (tv)", Enabled: true}, {Provider: "fanart", BaseURL: "https://webservice.fanart.tv/v3", Description: "Fanart.tv (artwork)", Enabled: true}, {Provider: "douban", Description: "Douban cookie (zh metadata)", Enabled: true}, - {Provider: "adult", BaseURL: "https://javdb.com", Extra: "https://javbus.sbs,https://www.javbus.com,https://www.cdnbus.cyou,https://www.javsee.cyou,https://www.busjav.cyou", Description: "Adult / 番号元数据（JavDB/JavBus）", Enabled: true}, - {Provider: "metatube", BaseURL: "http://127.0.0.1:7700", Description: "MetaTube Server 番号刮削后端", Enabled: true}, - {Provider: "openai", BaseURL: "https://api.openai.com/v1", Description: "OpenAI-compatible (smart search)", Enabled: true}, + {Provider: "adult", BaseURL: "https://javdb.com", Extra: "https://javbus.sbs,https://www.javbus.com,https://www.cdnbus.cyou,https://www.javsee.cyou,https://www.busjav.cyou", Description: "Adult / 番号元数据（JavDB/JavBus）", Enabled: true}, + {Provider: "metatube", BaseURL: "http://127.0.0.1:7700", Description: "MetaTube Server 番号刮削后端", Enabled: true}, + {Provider: "openai", BaseURL: "https://api.openai.com/v1", Description: "OpenAI-compatible (smart search)", Enabled: true}, } for i := range defaults { var existing model.APIConfig diff --git a/internal/service/api_config_connection.go b/internal/service/api_config_connection.go index 117f1a2..15f53ed 100644 --- a/internal/service/api_config_connection.go +++ b/internal/service/api_config_connection.go @@ -30,11 +30,11 @@ func (s *ApiConfigService) TestConnection(ctx context.Context, provider string) return s.testDeepSeek(cfg) case "siliconflow": return s.testSiliconFlow(cfg) - case "adult": - return s.testAdult(ctx, cfg) - case "metatube": - return s.testMetaTube(ctx, cfg) - default: + case "adult": + return s.testAdult(ctx, cfg) + case "metatube": + return s.testMetaTube(ctx, cfg) + default: return "unknown", fmt.Errorf("no test implemented for provider: %s", provider) } } diff --git a/internal/service/cloud/cloud_115_test.go b/internal/service/cloud/cloud_115_test.go index 0f4d9a7..59c0b24 100644 --- a/internal/service/cloud/cloud_115_test.go +++ b/internal/service/cloud/cloud_115_test.go @@ -131,4 +131,4 @@ func padZero(n int) string { s = "0" + s } return s -} \ No newline at end of file +} diff --git a/internal/service/cloud/pan115_openapi.go b/internal/service/cloud/pan115_openapi.go index 9855f50..5c596ee 100644 --- a/internal/service/cloud/pan115_openapi.go +++ b/internal/service/cloud/pan115_openapi.go @@ -91,4 +91,4 @@ func (p *openAPI115Provider) OpenClient() *cloud115.OpenClient { return p.c } // cloud115.IsRefreshTokenDead(err) 为 true 的错误。 func (p *openAPI115Provider) RefreshToken(refreshToken string) (*cloud115.TokenData, error) { return p.c.RefreshToken(refreshToken) -} \ No newline at end of file +} diff --git a/internal/service/cloud115/client.go b/internal/service/cloud115/client.go index fa9c649..fac3fe1 100644 --- a/internal/service/cloud115/client.go +++ b/internal/service/cloud115/client.go @@ -1,18 +1,16 @@ -// 115 开放平台 HTTP 客户端（移植自 QMediaSync 的 v115open，去掉 resty 依赖， -// 使用 net/http + 简单限流重试；只保留只读能力：授权/列目录/详情/下载直链）。 +// 115 开放平台 HTTP 客户端（集成全局三级令牌桶限流、熔断保护与防 405 重定向策略）。 package cloud115 import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "net/http" "net/url" "strings" - "sync" - "sync/atomic" "time" ) @@ -22,21 +20,35 @@ type OpenClient struct { HTTP *http.Client AccessToken string RefreshTokenStr string - - // 全局 QPS 限流（115 开放平台免费额度较低） - lastSecond int64 - reqInSecond int64 + executor *QueueExecutor } -var openClientMu sync.Mutex +// default115HTTPClient 创建带有防 405 重定向保护的 http.Client。 +func default115HTTPClient() *http.Client { + return &http.Client{ + Timeout: 60 * time.Second, + // 防止 Go 标准库在遇到 301/302/307 重定向时将 POST 降级为 GET 导致 115 报 405 Method Not Allowed + CheckRedirect: func(req *http.Request, via []*http.Request) error { + if len(via) >= 5 { + return errors.New("stopped after 5 redirects") + } + // 如果原请求是 POST，重定向时不允许静默转为 GET（直接终止自动重定向，由业务层处理响应） + if len(via) > 0 && via[len(via)-1].Method == http.MethodPost { + return http.ErrUseLastResponse + } + return nil + }, + } +} // NewOpenClient 构造客户端。 func NewOpenClient(appID, accessToken, refreshToken string) *OpenClient { return &OpenClient{ AppID: appID, - HTTP: &http.Client{Timeout: 60 * time.Second}, + HTTP: default115HTTPClient(), AccessToken: accessToken, RefreshTokenStr: refreshToken, + executor: GetGlobalExecutor(), } } @@ -46,30 +58,6 @@ func (c *OpenClient) SetAuthToken(accessToken, refreshToken string) { c.RefreshTokenStr = refreshToken } -// throttle 简单的每秒限流（默认 4 QPS，115 免费应用限额约 5 QPS）。 -func (c *OpenClient) throttle(n int) { - for i := 0; i < n; i++ { - now := time.Now().Unix() - last := atomic.LoadInt64(&c.lastSecond) - if last != now { - if atomic.CompareAndSwapInt64(&c.lastSecond, last, now) { - atomic.StoreInt64(&c.reqInSecond, 0) - } - } - count := atomic.LoadInt64(&c.reqInSecond) - if count >= 4 { - time.Sleep(300 * time.Millisecond) - i-- - continue - } - if atomic.CompareAndSwapInt64(&c.reqInSecond, count, count+1) { - return - } - time.Sleep(50 * time.Millisecond) - i-- - } -} - // RespState 兼容 115 不同端点返回的 state 类型（proapi 返回布尔、passport 返回数字）。 type RespState bool @@ -101,38 +89,65 @@ type RespBase struct { Raw json.RawMessage `json:"-"` // 原始响应体（外层附加字段用） } -// doJSON 执行 GET 请求并解析为统一响应；带 AccessToken（access=true 时）。 +// doJSON 执行 HTTP 请求并解析为统一响应；带 AccessToken（access=true 时）。 func (c *OpenClient) doJSON(ctx context.Context, method, rawURL string, form map[string]string, access bool, retries int) (*RespBase, error) { + executor := c.executor + if executor == nil { + executor = GetGlobalExecutor() + } + var lastErr error for attempt := 0; attempt <= retries; attempt++ { - c.throttle(1) + // 1. 获取全局三级令牌桶（QPS/QPM/QPH）令牌，若在熔断状态则阻塞等待冷却 + if err := executor.Acquire(ctx); err != nil { + return nil, err + } + req, err := c.buildRequest(ctx, method, rawURL, form, access) if err != nil { return nil, err } + resp, err := c.HTTP.Do(req) if err != nil { lastErr = err if attempt < retries { - time.Sleep(time.Duration(attempt+1) * 500 * time.Millisecond) + time.Sleep(time.Duration(attempt+1) * 1 * time.Second) continue } return nil, err } + body, readErr := io.ReadAll(io.LimitReader(resp.Body, 16<<20)) _ = resp.Body.Close() if readErr != nil { lastErr = readErr continue } + + // 检查 HTTP 状态码：特别处理 405/406/429 等限流和异常阻断 if resp.StatusCode < 200 || resp.StatusCode >= 300 { + if resp.StatusCode == http.StatusMethodNotAllowed || resp.StatusCode == http.StatusNotAcceptable || resp.StatusCode == http.StatusTooManyRequests { + // 触发全局熔断冷却，防止短时间连续重试导致 IP 被完全拉黑 + executor.MarkThrottled() + lastErr = fmt.Errorf("115 接口触发频控/安全拦截（HTTP %d）：%s", resp.StatusCode, strings.TrimSpace(string(body))) + if attempt < retries { + // 阻塞等待 115 限流冷却解封后自动重试当前请求 + if waitErr := executor.throttleManager.WaitThrottleRecovery(ctx); waitErr != nil { + return nil, waitErr + } + continue + } + return nil, lastErr + } lastErr = fmt.Errorf("115 接口返回 HTTP %d：%s", resp.StatusCode, strings.TrimSpace(string(body))) if attempt < retries { - time.Sleep(time.Duration(attempt+1) * 500 * time.Millisecond) + time.Sleep(time.Duration(attempt+1) * 1 * time.Second) continue } return nil, lastErr } + var base RespBase if err := json.Unmarshal(body, &base); err != nil { return nil, fmt.Errorf("115 接口响应解析失败：%w", err) @@ -141,13 +156,29 @@ func (c *OpenClient) doJSON(ctx context.Context, method, rawURL string, form map if base.State { return &base, nil } - // 业务失败：限流/Token 错误不重试，其余按配置重试 - if IsThrottleCode(base.Code) || isTokenCode(base.Code) { + + // 业务返回限流错误码（406 / 770004）：标记全局熔断，等待冷却后重试 + if IsThrottleCode(base.Code) { + executor.MarkThrottled() + lastErr = NewOpenAPIResponseError(base.Code, base.Errno, base.Message, base.Error, "115 访问频率达到上限，已进入冷却") + if attempt < retries { + // 阻塞等待限流冷却解封后自动重试当前请求 + if waitErr := executor.throttleManager.WaitThrottleRecovery(ctx); waitErr != nil { + return nil, waitErr + } + continue + } + return &base, lastErr + } + + // Token 失效不重试 + if isTokenCode(base.Code) { return &base, nil } + lastErr = NewOpenAPIResponseError(base.Code, base.Errno, base.Message, base.Error, "115 接口调用失败") if attempt < retries { - time.Sleep(time.Duration(attempt+1) * 500 * time.Millisecond) + time.Sleep(time.Duration(attempt+1) * 1 * time.Second) continue } return &base, lastErr diff --git a/internal/service/cloud115/cloud115_test.go b/internal/service/cloud115/cloud115_test.go index 7048e18..d1c704d 100644 --- a/internal/service/cloud115/cloud115_test.go +++ b/internal/service/cloud115/cloud115_test.go @@ -7,6 +7,7 @@ import ( "net/url" "strings" "testing" + "time" ) // mockTransport 用 httptest server 替换 API 基址。 @@ -321,3 +322,56 @@ func TestSourceCatalog(t *testing.T) { t.Fatal("relay/thrid-party sources broken") } } + +func TestHTTP405AndThrottleRecovery(t *testing.T) { + tm := NewThrottleManager(100 * time.Millisecond) + qe := NewQueueExecutor(10, 200, 12000) + qe.throttleManager = tm + + mockAPI(t, func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusMethodNotAllowed) + w.Write([]byte("Method Not Allowed")) + }) + + c := NewOpenClient("100195125", "at1", "rt1") + c.executor = qe + + _, err := c.GetDownloadURL(context.Background(), "pickTest") + if err == nil { + t.Fatal("expected 405 error") + } + if !tm.IsThrottled() { + t.Fatal("405 should trigger throttle status") + } + + // 验证熔断冷却后能正常恢复 + ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) + defer cancel() + if err := tm.WaitThrottleRecovery(ctx); err != nil { + t.Fatalf("wait throttle recovery failed: %v", err) + } + if tm.IsThrottled() { + t.Fatal("throttle status should be cleared after duration") + } +} + +func TestThrottleCodeHandling(t *testing.T) { + tm := NewThrottleManager(100 * time.Millisecond) + qe := NewQueueExecutor(10, 200, 12000) + qe.throttleManager = tm + + mockAPI(t, func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(`{"state":false,"code":770004,"message":"访问频率过高"}`)) + }) + + c := NewOpenClient("100195125", "at1", "rt1") + c.executor = qe + + _, _, err := c.GetFsList(context.Background(), "0", 0, 100) + if err == nil { + t.Fatal("expected throttle error") + } + if !tm.IsThrottled() { + t.Fatal("code 770004 should trigger throttle status") + } +} diff --git a/internal/service/cloud115/const.go b/internal/service/cloud115/const.go index b070142..dd92376 100644 --- a/internal/service/cloud115/const.go +++ b/internal/service/cloud115/const.go @@ -37,8 +37,8 @@ const ( RefreshTokenCheckFailed = 40140120 ) -// DefaultUA 请求 UA。 -const DefaultUA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36 MMTL-115-OpenAPI/1.0" +// DefaultUA 请求 UA（标准 API 客户端格式，避免伪装浏览器触发 115 WAF 拦截与 405 错误）。 +const DefaultUA = "MMTL-115-GoClient/1.0" // OpenAPIError 保留 115 开放平台返回的原始错误信息。 type OpenAPIError struct { diff --git a/internal/service/cloud115/queue_executor.go b/internal/service/cloud115/queue_executor.go new file mode 100644 index 0000000..678272f --- /dev/null +++ b/internal/service/cloud115/queue_executor.go @@ -0,0 +1,106 @@ +package cloud115 + +import ( + "context" + "sync" + "time" + + "golang.org/x/time/rate" +) + +// QueueExecutor 全局 115 请求调度执行器，通过三级令牌桶（QPS / QPM / QPH）平滑所有外发请求。 +type QueueExecutor struct { + sync.RWMutex + qpsLimiter *rate.Limiter + qpmLimiter *rate.Limiter + qphLimiter *rate.Limiter + throttleManager *ThrottleManager + + qpsConfig int + qpmConfig int + qphConfig int +} + +var ( + globalExecutor *QueueExecutor + executorOnce sync.Once +) + +// GetGlobalExecutor 获取全局队列执行器单例（默认 QPS=2, QPM=120, QPH=6000，保障 115 API 调用安全不超频）。 +func GetGlobalExecutor() *QueueExecutor { + executorOnce.Do(func() { + globalExecutor = NewQueueExecutor(2, 120, 6000) + }) + return globalExecutor +} + +// NewQueueExecutor 创建新的队列调度执行器。 +func NewQueueExecutor(qps, qpm, qph int) *QueueExecutor { + if qps <= 0 { + qps = 3 + } + if qpm <= 0 { + qpm = 200 + } + if qph <= 0 { + qph = 12000 + } + + return &QueueExecutor{ + qpsLimiter: rate.NewLimiter(rate.Limit(qps), qps), + qpmLimiter: rate.NewLimiter(rate.Every(time.Minute/time.Duration(qpm)), qpm), + qphLimiter: rate.NewLimiter(rate.Every(time.Hour/time.Duration(qph)), qph), + throttleManager: GetGlobalThrottleManager(), + qpsConfig: qps, + qpmConfig: qpm, + qphConfig: qph, + } +} + +// Acquire 统一在发送请求前获取令牌，并等待熔断恢复。 +func (qe *QueueExecutor) Acquire(ctx context.Context) error { + // 1. 如果处于限流状态，先阻塞等待 60s 静默冷却结束 + if err := qe.throttleManager.WaitThrottleRecovery(ctx); err != nil { + return err + } + + // 2. 依次获取小时级、分钟级、秒级令牌 + if err := qe.qphLimiter.Wait(ctx); err != nil { + return err + } + if err := qe.qpmLimiter.Wait(ctx); err != nil { + return err + } + if err := qe.qpsLimiter.Wait(ctx); err != nil { + return err + } + + return nil +} + +// SetRateLimitConfig 动态调整速率限制。 +func (qe *QueueExecutor) SetRateLimitConfig(qps, qpm, qph int) { + qe.Lock() + defer qe.Unlock() + + if qps > 0 { + qe.qpsConfig = qps + qe.qpsLimiter.SetLimit(rate.Limit(qps)) + qe.qpsLimiter.SetBurst(qps) + } + if qpm > 0 { + qe.qpmConfig = qpm + qe.qpmLimiter.SetLimit(rate.Every(time.Minute / time.Duration(qpm))) + qe.qpmLimiter.SetBurst(qpm) + } + if qph > 0 { + qe.qphConfig = qph + qe.qphLimiter.SetLimit(rate.Every(time.Hour / time.Duration(qph))) + qe.qphLimiter.SetBurst(qph) + } +} + +// MarkThrottled 触发限流熔断。 +func (qe *QueueExecutor) MarkThrottled() { + qe.throttleManager.MarkThrottled() +} diff --git a/internal/service/cloud115/throttle_manager.go b/internal/service/cloud115/throttle_manager.go new file mode 100644 index 0000000..14c4257 --- /dev/null +++ b/internal/service/cloud115/throttle_manager.go @@ -0,0 +1,119 @@ +package cloud115 + +import ( + "context" + "sync" + "time" +) + +// ThrottleManager 全局限流/熔断管理器，用于控制 115 API 访问频率与被风控时的全局静默冷却。 +type ThrottleManager struct { + sync.RWMutex + isThrottled bool + throttleStartTime time.Time + throttleNotify chan struct{} + throttleDuration time.Duration +} + +var ( + globalThrottleManager *ThrottleManager + throttleOnce sync.Once +) + +// GetGlobalThrottleManager 获取全局限流管理器单例。 +func GetGlobalThrottleManager() *ThrottleManager { + throttleOnce.Do(func() { + globalThrottleManager = NewThrottleManager(60 * time.Second) + }) + return globalThrottleManager +} + +// NewThrottleManager 创建限流管理器。 +func NewThrottleManager(duration time.Duration) *ThrottleManager { + if duration <= 0 { + duration = 60 * time.Second + } + return &ThrottleManager{ + isThrottled: false, + throttleNotify: make(chan struct{}), + throttleDuration: duration, + } +} + +// IsThrottled 检查是否处于限流冷却状态。 +func (tm *ThrottleManager) IsThrottled() bool { + tm.RLock() + defer tm.RUnlock() + + if !tm.isThrottled { + return false + } + if time.Since(tm.throttleStartTime) >= tm.throttleDuration { + return false + } + return true +} + +// MarkThrottled 标记为限流状态，并启动恢复计时器。 +func (tm *ThrottleManager) MarkThrottled() { + tm.Lock() + defer tm.Unlock() + + if tm.isThrottled && time.Since(tm.throttleStartTime) < tm.throttleDuration { + return + } + + tm.isThrottled = true + tm.throttleStartTime = time.Now() + + go tm.startRecoveryTimer() +} + +func (tm *ThrottleManager) startRecoveryTimer() { + time.Sleep(tm.throttleDuration) + + tm.Lock() + defer tm.Unlock() + + tm.isThrottled = false + + select { + case tm.throttleNotify <- struct{}{}: + default: + } +} + +// WaitThrottleRecovery 如果处于限流状态，挂起等待直到冷却恢复。 +func (tm *ThrottleManager) WaitThrottleRecovery(ctx context.Context) error { + for { + if !tm.IsThrottled() { + return nil + } + + ticker := time.NewTicker(200 * time.Millisecond) + select { + case <-ctx.Done(): + ticker.Stop() + return ctx.Err() + case <-ticker.C: + ticker.Stop() + if !tm.IsThrottled() { + return nil + } + } + } +} + +// Reset 重置限流状态（主要用于单元测试）。 +func (tm *ThrottleManager) Reset() { + tm.Lock() + defer tm.Unlock() + tm.isThrottled = false +} + +// SetDuration 调整限流冷却时间（用于测试或动态配置）。 +func (tm *ThrottleManager) SetDuration(d time.Duration) { + tm.Lock() + defer tm.Unlock() + tm.throttleDuration = d +} diff --git a/internal/service/cloud115/url_cache.go b/internal/service/cloud115/url_cache.go index e0c0841..40b488a 100644 --- a/internal/service/cloud115/url_cache.go +++ b/internal/service/cloud115/url_cache.go @@ -71,4 +71,4 @@ func ClearDownloadURLCache(pickCode string) { urlCacheMu.Lock() defer urlCacheMu.Unlock() delete(urlCache, pickCode) -} \ No newline at end of file +} diff --git a/internal/service/cloud115/url_cache_test.go b/internal/service/cloud115/url_cache_test.go index a331d52..bf5e1bc 100644 --- a/internal/service/cloud115/url_cache_test.go +++ b/internal/service/cloud115/url_cache_test.go @@ -28,4 +28,4 @@ func TestDownloadURLCache(t *testing.T) { if got := GetDownloadURLCache("pick1"); got != "" { t.Fatalf("cleared entry should be a miss, got %q", got) } -} \ No newline at end of file +} diff --git a/internal/service/cloud_library_display_stubs.go b/internal/service/cloud_library_display_stubs.go index 80222d0..514403e 100644 --- a/internal/service/cloud_library_display_stubs.go +++ b/internal/service/cloud_library_display_stubs.go @@ -148,6 +148,7 @@ func joinRemotePath(base, rel string) string { } return path.Clean(path.Join(parts...)) } + // 通用库路径辅助（原云盘库/STRM 生成逻辑使用；网盘后端移除后保留为纯工具函数， // 供库路径构建与既有测试作为稳定夹具使用）。 diff --git a/internal/service/crypto.go b/internal/service/crypto.go index eeffcee..4832f9c 100644 --- a/internal/service/crypto.go +++ b/internal/service/crypto.go @@ -30,7 +30,7 @@ const encPrefix = "enc:v1:" // CryptoService wraps an AES-GCM cipher derived from a stable per-install // secret (the JWT secret). type CryptoService struct { - log *zap.Logger + log *zap.Logger aead cipher.AEAD } diff --git a/internal/service/danmaku_service.go b/internal/service/danmaku_service.go index 4765d1b..18ae7a0 100644 --- a/internal/service/danmaku_service.go +++ b/internal/service/danmaku_service.go @@ -53,17 +53,17 @@ type DanmakuRenderConfig struct { // must ask the user which one to use (disambiguation); Raw is empty then. type DanmakuFetchResult struct { DanmakuRenderConfig - SourceType string `json:"source_type"` - Raw string `json:"raw,omitempty"` - Candidates []DanmakuAnime `json:"candidates,omitempty"` + SourceType string `json:"source_type"` + Raw string `json:"raw,omitempty"` + Candidates []DanmakuAnime `json:"candidates,omitempty"` } // DanmakuAnime is one search hit (an anime) with its episode list, mirroring // the dandanplay SearchEpisodesResponse shape. type DanmakuAnime struct { - AnimeID int64 `json:"animeId"` - AnimeTitle string `json:"animeTitle"` - Episodes []DanmakuEpisode `json:"episodes"` + AnimeID int64 `json:"animeId"` + AnimeTitle string `json:"animeTitle"` + Episodes []DanmakuEpisode `json:"episodes"` } // DanmakuEpisode is one selectable danmaku library inside an anime. @@ -125,9 +125,9 @@ func (s *DanmakuService) Config(ctx context.Context) DanmakuRenderConfig { // through the dandanplay protocol: // // 1. search episodes by title (+ season/episode number) -// 2a. exactly one hit → fetch that episode's comment library -// 2b. several hits → return candidates (Raw empty) so the player asks the user -// 2c. explicit episodeID → fetch it directly +// 2a. exactly one hit → fetch that episode's comment library +// 2b. several hits → return candidates (Raw empty) so the player asks the user +// 2c. explicit episodeID → fetch it directly // 3. fetch the comment library XML // // When danmaku is disabled the result carries Enabled=false so the player can @@ -250,9 +250,9 @@ func (s *DanmakuService) searchCandidates(ctx context.Context, base, name, episo } var resp struct { Animes []struct { - AnimeID int64 `json:"animeId"` + AnimeID int64 `json:"animeId"` AnimeTitle string `json:"animeTitle"` - Episodes []struct { + Episodes []struct { EpisodeID int64 `json:"episodeId"` EpisodeTitle string `json:"episodeTitle"` } `json:"episodes"` @@ -311,4 +311,4 @@ func (s *DanmakuService) fetchBody(ctx context.Context, sourceURL string, follow return "", err } return string(body), nil -} \ No newline at end of file +} diff --git a/internal/service/dlna.go b/internal/service/dlna.go index 479120a..edcb735 100644 --- a/internal/service/dlna.go +++ b/internal/service/dlna.go @@ -31,8 +31,8 @@ import ( type DLNAService struct { log *zap.Logger - mu sync.Mutex - cache []DLNADevice + mu sync.Mutex + cache []DLNADevice cachedAt time.Time } @@ -47,8 +47,8 @@ type DLNADevice struct { FriendlyName string `json:"friendly_name"` Manufacturer string `json:"manufacturer"` ModelName string `json:"model_name"` - Location string `json:"location"` // device description URL - ControlURL string `json:"control_url"` // AVTransport SOAP endpoint + Location string `json:"location"` // device description URL + ControlURL string `json:"control_url"` // AVTransport SOAP endpoint IPAddress string `json:"ip_address"` } @@ -168,10 +168,10 @@ func (d *DLNAService) fetchDescription(ctx context.Context, location string) (*D ControlURL string `xml:"controlURL"` } type device struct { - FriendlyName string `xml:"friendlyName"` - Manufacturer string `xml:"manufacturer"` - ModelName string `xml:"modelName"` - UDN string `xml:"UDN"` + FriendlyName string `xml:"friendlyName"` + Manufacturer string `xml:"manufacturer"` + ModelName string `xml:"modelName"` + UDN string `xml:"UDN"` ServiceList struct { Services []service `xml:"service"` } `xml:"serviceList"` diff --git a/internal/service/download_helpers.go b/internal/service/download_helpers.go index 22df0bd..2cb7b1b 100644 --- a/internal/service/download_helpers.go +++ b/internal/service/download_helpers.go @@ -72,4 +72,4 @@ func detectResolutionScore(titleFold string) int { default: return 1 } -} \ No newline at end of file +} diff --git a/internal/service/emby_compat.go b/internal/service/emby_compat.go index b024b3e..2915d86 100644 --- a/internal/service/emby_compat.go +++ b/internal/service/emby_compat.go @@ -43,10 +43,10 @@ const PlaybackDirectOnlySettingKey = "playback.direct_only" // EmbyService produces Emby-shaped JSON. type EmbyService struct { - cfg *config.Config - log *zap.Logger - repo *repository.Container - cache *RuntimeCacheService + cfg *config.Config + log *zap.Logger + repo *repository.Container + cache *RuntimeCacheService subtitle *SubtitleService virtualMu sync.RWMutex diff --git a/internal/service/emby_media_sources.go b/internal/service/emby_media_sources.go index e35c2a2..d7f39d0 100644 --- a/internal/service/emby_media_sources.go +++ b/internal/service/emby_media_sources.go @@ -234,9 +234,9 @@ func (e *EmbyService) appendSubtitleStreams(ctx context.Context, streams []map[s // MediaSource.Id (== m.ID here). No "mediasource_" prefix: that // prefix only appears in real servers because their MediaSource.Id // value itself starts with it, not because the route template says so. - "IsTextSubtitleStream": true, // external sidecar files are always text subtitles - "SupportsExternalStream": true, - "DeliveryUrl": "/Videos/" + mediaID + "/" + mediaID + "/Subtitles/" + fmt.Sprint(index) + "/Stream." + codec, + "IsTextSubtitleStream": true, // external sidecar files are always text subtitles + "SupportsExternalStream": true, + "DeliveryUrl": "/Videos/" + mediaID + "/" + mediaID + "/Subtitles/" + fmt.Sprint(index) + "/Stream." + codec, }) } return streams diff --git a/internal/service/image_proxy_remote_fetch.go b/internal/service/image_proxy_remote_fetch.go index a7c2088..3972cba 100644 --- a/internal/service/image_proxy_remote_fetch.go +++ b/internal/service/image_proxy_remote_fetch.go @@ -158,13 +158,13 @@ func fetchRemoteImageWithCurl(ctx context.Context, raw, host string) ([]byte, st "--header", "Cache-Control: no-cache", "--header", "Pragma: no-cache", } - if referer := remoteImageReferer(host); referer != "" { - args = append(args, "--referer", referer) - } - if cookie := remoteImageCookie(host); cookie != "" { - args = append(args, "--cookie", cookie) - } - args = append(args, "--", raw) + if referer := remoteImageReferer(host); referer != "" { + args = append(args, "--referer", referer) + } + if cookie := remoteImageCookie(host); cookie != "" { + args = append(args, "--cookie", cookie) + } + args = append(args, "--", raw) cmd := exec.CommandContext(curlCtx, bin, args...) // #nosec G204 -- bin is resolved by LookPath and args are not shell-expanded. stderr := bytes.Buffer{} diff --git a/internal/service/organizer_paths.go b/internal/service/organizer_paths.go index fec4d5b..68f93a1 100644 --- a/internal/service/organizer_paths.go +++ b/internal/service/organizer_paths.go @@ -5,13 +5,22 @@ import ( "strings" ) -// sanitizeFilename removes characters not safe for filesystem names. +// sanitizeFilename removes characters not safe for filesystem names across Windows, Linux, and macOS. func sanitizeFilename(s string) string { r := strings.NewReplacer( "/", " ", "\\", " ", ":", " ", "*", "", "?", "", "\"", "", "<", "", ">", "", "|", "", ) - return strings.TrimSpace(r.Replace(s)) + replaced := r.Replace(s) + var sb strings.Builder + for _, ch := range replaced { + // Strip ASCII control characters (0-31, 127) + if ch < 32 || ch == 127 { + continue + } + sb.WriteRune(ch) + } + return strings.Join(strings.Fields(sb.String()), " ") } func (o *OrganizerService) organizeRoot(libraryPath, mediaType, category string) string { diff --git a/internal/service/organizer_scrape_providers_test.go b/internal/service/organizer_scrape_providers_test.go index cabc471..3a62908 100644 --- a/internal/service/organizer_scrape_providers_test.go +++ b/internal/service/organizer_scrape_providers_test.go @@ -63,13 +63,13 @@ func TestOrganizeDirectoryUsesAdultMetadataBeforeRename(t *testing.T) { if res.Organized != 1 || len(res.Items) != 1 { t.Fatalf("result = %+v, want one organized preview", res) } - if res.Items[0].Title != "SSIS-001-整理成人标题" || res.Items[0].MediaType != "adult" { - t.Fatalf("adult organize item = %+v", res.Items[0]) - } - wantSuffix := filepath.Join("成人", "SSIS-001-整理成人标题", "SSIS-001-整理成人标题.mkv") - if !strings.Contains(res.Items[0].Target, wantSuffix) { - t.Fatalf("adult target = %q, want suffix %q", res.Items[0].Target, wantSuffix) - } + if res.Items[0].Title != "SSIS-001-整理成人标题" || res.Items[0].MediaType != "adult" { + t.Fatalf("adult organize item = %+v", res.Items[0]) + } + wantSuffix := filepath.Join("成人", "SSIS-001-整理成人标题", "SSIS-001-整理成人标题.mkv") + if !strings.Contains(res.Items[0].Target, wantSuffix) { + t.Fatalf("adult target = %q, want suffix %q", res.Items[0].Target, wantSuffix) + } } func TestOrganizeDirectoryUsesBangumiForAnimeRename(t *testing.T) { diff --git a/internal/service/organizer_target_template.go b/internal/service/organizer_target_template.go index 7452c65..b8260d4 100644 --- a/internal/service/organizer_target_template.go +++ b/internal/service/organizer_target_template.go @@ -74,11 +74,11 @@ func renderOrganizeConditionals(template string, data organizeNamingData) string func organizeTemplateTruthy(name string, data organizeNamingData) bool { switch strings.ToLower(strings.TrimSpace(name)) { - case "title", "name": - return data.Title != "" - case "code", "num", "adult_code": - return data.Code != "" - case "year": + case "title", "name": + return data.Title != "" + case "code", "num", "adult_code": + return data.Code != "" + case "year": return data.Year > 0 case "season": return data.Season >= 0 && data.Episode > 0 @@ -105,11 +105,11 @@ func organizeTemplateTruthy(name string, data organizeNamingData) bool { func organizeTemplateValue(name, format string, data organizeNamingData, fallback string) string { switch strings.ToLower(strings.TrimSpace(name)) { - case "title", "name": - return data.Title - case "code", "num", "adult_code": - return data.Code - case "year": + case "title", "name": + return data.Title + case "code", "num", "adult_code": + return data.Code + case "year": if data.Year <= 0 { return "" } diff --git a/internal/service/permission.go b/internal/service/permission.go index 710e82a..0c95ba9 100644 --- a/internal/service/permission.go +++ b/internal/service/permission.go @@ -40,8 +40,8 @@ func DefaultPermissions(userID string) *model.UserPermission { CanRescrape: false, CanUseAI: false, CanCaptureFrames: false, - CanManageDownloads: false, - CanManageSubscriptions: false, + CanManageDownloads: false, + CanManageSubscriptions: false, CanManageSites: false, CanUseAIAssistant: false, CanManageUsers: false, @@ -65,8 +65,8 @@ func adminGrant(userID string) *model.UserPermission { CanRescrape: true, CanUseAI: true, CanCaptureFrames: true, - CanManageDownloads: true, - CanManageSubscriptions: true, + CanManageDownloads: true, + CanManageSubscriptions: true, CanManageSites: true, CanUseAIAssistant: true, CanManageUsers: true, diff --git a/internal/service/scanner.go b/internal/service/scanner.go index a6e7128..521af10 100644 --- a/internal/service/scanner.go +++ b/internal/service/scanner.go @@ -62,12 +62,12 @@ type ScannerService struct { imageProxy *ImageProxy - localMediaProbeOnce sync.Once - localMediaProbeQueue chan localMediaProbeTask - localMediaProbeMu sync.Mutex - localMediaProbing map[string]struct{} - localScanMu sync.Mutex - localScans map[string]struct{} + localMediaProbeOnce sync.Once + localMediaProbeQueue chan localMediaProbeTask + localMediaProbeMu sync.Mutex + localMediaProbing map[string]struct{} + localScanMu sync.Mutex + localScans map[string]struct{} } // NewScannerService is the constructor. @@ -80,7 +80,7 @@ func NewScannerService( scraper *ScraperService, ) *ScannerService { return &ScannerService{ - cfg: cfg, log: log, repo: repo, hub: hub, + cfg: cfg, log: log, repo: repo, hub: hub, probe: probe, scraper: scraper, localMediaProbeQueue: make(chan localMediaProbeTask, 1024), diff --git a/internal/service/scraper_local_metadata.go b/internal/service/scraper_local_metadata.go index 0b767e3..20ef214 100644 --- a/internal/service/scraper_local_metadata.go +++ b/internal/service/scraper_local_metadata.go @@ -248,16 +248,16 @@ func (s *ScraperService) applyLocalMetadataMatch(ctx context.Context, m *model.M if next.NSFW { updates["nsfw"] = true } - if err := s.repo.DB.WithContext(ctx).Model(&model.Media{}). - Where("id = ?", m.ID).Updates(updates).Error; err != nil { - return err - } - var lib *model.Library - if m != nil && s.repo != nil && s.repo.Library != nil { - lib, _ = s.repo.Library.FindByID(ctx, m.LibraryID) - } - s.writeMediaArtworkFilesAfterScrape(ctx, m, lib) - s.invalidateMediaCache(ctx) + if err := s.repo.DB.WithContext(ctx).Model(&model.Media{}). + Where("id = ?", m.ID).Updates(updates).Error; err != nil { + return err + } + var lib *model.Library + if m != nil && s.repo != nil && s.repo.Library != nil { + lib, _ = s.repo.Library.FindByID(ctx, m.LibraryID) + } + s.writeMediaArtworkFilesAfterScrape(ctx, m, lib) + s.invalidateMediaCache(ctx) s.hub.Publish("scrape", map[string]any{ "media_id": m.ID, "title": next.Title, diff --git a/internal/service/scraper_test.go b/internal/service/scraper_test.go index 87ca082..5fc02a4 100644 --- a/internal/service/scraper_test.go +++ b/internal/service/scraper_test.go @@ -372,10 +372,10 @@ func TestApplyManualMatchSkipsTMDbEpisodeStillWhenDisabled(t *testing.T) { if !strings.HasSuffix(got.PosterURL, "/images/w500/poster.jpg") { t.Fatalf("series poster should still be saved when manual episode artwork is disabled: got %q", got.PosterURL) } - if !strings.HasSuffix(got.BackdropURL, "/images/w1280/backdrop.jpg") { - t.Fatalf("series backdrop should still be saved when manual episode artwork is disabled: got %q", got.BackdropURL) - } + if !strings.HasSuffix(got.BackdropURL, "/images/w1280/backdrop.jpg") { + t.Fatalf("series backdrop should still be saved when manual episode artwork is disabled: got %q", got.BackdropURL) } +} func TestEnrichOneAdultScrapesArtwork(t *testing.T) { scraper, repos, closeServer := newTestScraper(t) diff --git a/internal/service/scraper_write_metadata_files.go b/internal/service/scraper_write_metadata_files.go index 19dc063..b49e465 100644 --- a/internal/service/scraper_write_metadata_files.go +++ b/internal/service/scraper_write_metadata_files.go @@ -44,47 +44,47 @@ func (s *ScraperService) writeMediaArtworkFilesAfterScrape(ctx context.Context, if base == "" || base == "." { return } - isAdult := IsAdultMediaPathOrMetadata(refreshed.Path, refreshed.LibraryID, refreshed.NSFW) || IsAdultArtworkURL(refreshed.PosterURL) - if refreshed.PosterURL != "" { - s.downloadArtworkToPathWithOptions(ctx, dir, base+"-poster", refreshed.PosterURL, isAdult) - } - if refreshed.BackdropURL != "" { - s.downloadArtworkToPathWithOptions(ctx, dir, base+"-backdrop", refreshed.BackdropURL, false) - } else if isAdult && refreshed.PosterURL != "" { - // 番号海报原图为完整封套横图，在无独立背景图时直接作为背景图写出 - s.downloadArtworkToPathWithOptions(ctx, dir, base+"-backdrop", refreshed.PosterURL, false) - } + isAdult := IsAdultMediaPathOrMetadata(refreshed.Path, refreshed.LibraryID, refreshed.NSFW) || IsAdultArtworkURL(refreshed.PosterURL) + if refreshed.PosterURL != "" { + s.downloadArtworkToPathWithOptions(ctx, dir, base+"-poster", refreshed.PosterURL, isAdult) } - - func (s *ScraperService) downloadArtworkToPath(ctx context.Context, dir, name, raw string) { - s.downloadArtworkToPathWithOptions(ctx, dir, name, raw, false) + if refreshed.BackdropURL != "" { + s.downloadArtworkToPathWithOptions(ctx, dir, base+"-backdrop", refreshed.BackdropURL, false) + } else if isAdult && refreshed.PosterURL != "" { + // 番号海报原图为完整封套横图，在无独立背景图时直接作为背景图写出 + s.downloadArtworkToPathWithOptions(ctx, dir, base+"-backdrop", refreshed.PosterURL, false) } +} - // downloadArtworkToPathWithOptions fetches an artwork URL via the image proxy cache and - // writes it under dir/.. For adult posters, it crops the right half of the cover. - func (s *ScraperService) downloadArtworkToPathWithOptions(ctx context.Context, dir, name, raw string, cropAdultPoster bool) { - if !isHTTPish(raw) { - return - } - data, ctype, err := s.images.Fetch(ctx, raw) - if err != nil || len(data) == 0 { - s.log.Warn("scrape artwork download failed", - zap.String("name", name), - zap.String("url", raw), - zap.Error(err)) - return - } - if !isImageContentType(ctype) || isTransparentPlaceholderData(data) { - return - } - if cropAdultPoster { - if cropped, croppedType, err := CropAdultCoverPoster(data); err == nil && len(cropped) > 0 { - data = cropped - ctype = croppedType - } - } - s.writeArtworkDataToPath(dir, name, ctype, data) +func (s *ScraperService) downloadArtworkToPath(ctx context.Context, dir, name, raw string) { + s.downloadArtworkToPathWithOptions(ctx, dir, name, raw, false) +} + +// downloadArtworkToPathWithOptions fetches an artwork URL via the image proxy cache and +// writes it under dir/.. For adult posters, it crops the right half of the cover. +func (s *ScraperService) downloadArtworkToPathWithOptions(ctx context.Context, dir, name, raw string, cropAdultPoster bool) { + if !isHTTPish(raw) { + return } + data, ctype, err := s.images.Fetch(ctx, raw) + if err != nil || len(data) == 0 { + s.log.Warn("scrape artwork download failed", + zap.String("name", name), + zap.String("url", raw), + zap.Error(err)) + return + } + if !isImageContentType(ctype) || isTransparentPlaceholderData(data) { + return + } + if cropAdultPoster { + if cropped, croppedType, err := CropAdultCoverPoster(data); err == nil && len(cropped) > 0 { + data = cropped + ctype = croppedType + } + } + s.writeArtworkDataToPath(dir, name, ctype, data) +} // writeArtworkDataToPath writes in-memory artwork bytes to dir/. // using a temp file + rename so readers never observe a partial file. Returns @@ -93,6 +93,7 @@ func (s *ScraperService) writeArtworkDataToPath(dir, name, ctype string, data [] if len(data) == 0 { return "" } + dir = sanitizeLocalPath(dir) if err := os.MkdirAll(dir, 0o755); err != nil { s.log.Warn("scrape artwork mkdir failed", zap.String("dir", dir), zap.Error(err)) return "" diff --git a/internal/service/scraper_write_metadata_files_test.go b/internal/service/scraper_write_metadata_files_test.go index 07522a0..5876853 100644 --- a/internal/service/scraper_write_metadata_files_test.go +++ b/internal/service/scraper_write_metadata_files_test.go @@ -37,12 +37,12 @@ func TestWriteArtworkDataToPathWritesJellyfinSidecar(t *testing.T) { // name Jellyfin sidecar files. func TestImageExtForContentType(t *testing.T) { cases := map[string]string{ - "image/jpeg": ".jpg", - "image/pjpeg": ".jpg", - "image/png": ".png", - "image/webp": ".webp", - "image/gif": ".gif", - "image/avif": ".avif", + "image/jpeg": ".jpg", + "image/pjpeg": ".jpg", + "image/png": ".png", + "image/webp": ".webp", + "image/gif": ".gif", + "image/avif": ".avif", "image/jpeg; charset=binary": ".jpg", "application/octet-stream": ".img", } diff --git a/internal/service/sse_hub.go b/internal/service/sse_hub.go index d3949be..763a40e 100644 --- a/internal/service/sse_hub.go +++ b/internal/service/sse_hub.go @@ -30,12 +30,12 @@ type SSEEvent struct { // SSEEvent 事件类型常量。 const ( - EventTypeScan = "scan" - EventTypeDownload = "download" - EventTypeSubscribe = "subscribe" - EventTypeTask = "task" - EventTypeSystem = "system" - EventTypeAuth = "auth" + EventTypeScan = "scan" + EventTypeDownload = "download" + EventTypeSubscribe = "subscribe" + EventTypeTask = "task" + EventTypeSystem = "system" + EventTypeAuth = "auth" ) // sseTicket 是一次性 OTP 票据。 diff --git a/internal/service/strm_queue.go b/internal/service/strm_queue.go index a779182..7fd9817 100644 --- a/internal/service/strm_queue.go +++ b/internal/service/strm_queue.go @@ -59,6 +59,11 @@ func (s *StrmService) downloadWorker(ctx context.Context) { } func (s *StrmService) processDownloadTask(ctx context.Context, task *model.StrmDownloadTask) { + cleanPath := sanitizeLocalPath(task.LocalPath) + if cleanPath != "" && cleanPath != task.LocalPath { + task.LocalPath = cleanPath + _ = s.repo.StrmDownload.Update(context.Background(), task) + } finish := func(status, message string) { now := time.Now() task.Status = status @@ -210,6 +215,7 @@ func retryTask(retryCount *int, status *string, errMsg *string, nextTryAt **time // downloadToFile 把直链内容下载到目标文件（临时文件 + 原子改名）。 func downloadToFile(ctx context.Context, link *cloud.DirectLink, target string, client *http.Client) error { + target = sanitizeLocalPath(target) if link == nil || link.URL == "" { return errors.New("空下载地址") } diff --git a/internal/service/strm_queue_test.go b/internal/service/strm_queue_test.go index 03dc514..498d57a 100644 --- a/internal/service/strm_queue_test.go +++ b/internal/service/strm_queue_test.go @@ -43,4 +43,4 @@ func TestIsHTTPDownloadFailure(t *testing.T) { t.Errorf("isHTTPDownloadFailure(%v) = %v, want %v", c.err, got, c.want) } } -} \ No newline at end of file +} diff --git a/internal/service/strm_service.go b/internal/service/strm_service.go index aae0c90..5226341 100644 --- a/internal/service/strm_service.go +++ b/internal/service/strm_service.go @@ -572,14 +572,161 @@ const defaultStrmUA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537 // ensureLocalDir 创建本地输出目录。 func ensureLocalDir(dir string) error { - return os.MkdirAll(dir, 0o755) + return os.MkdirAll(sanitizeLocalPath(dir), 0o755) +} + +// windowsReservedNames 包含 Windows 系统底层保留的设备名称（大小写不敏感）。 +var windowsReservedNames = map[string]bool{ + "CON": true, "PRN": true, "AUX": true, "NUL": true, + "COM1": true, "COM2": true, "COM3": true, "COM4": true, "COM5": true, + "COM6": true, "COM7": true, "COM8": true, "COM9": true, + "LPT1": true, "LPT2": true, "LPT3": true, "LPT4": true, "LPT5": true, + "LPT6": true, "LPT7": true, "LPT8": true, "LPT9": true, +} + +func isWindowsReservedName(name string) bool { + return windowsReservedNames[strings.ToUpper(name)] +} + +// truncateStringRuneSafe 安全截断 UTF-8 字符串至指定字节长度，不切断中文字符或多字节 rune。 +func truncateStringRuneSafe(s string, maxBytes int) string { + if len(s) <= maxBytes { + return s + } + b := []byte(s) + if len(b) <= maxBytes { + return s + } + for maxBytes > 0 && (b[maxBytes]&0xC0 == 0x80) { + maxBytes-- + } + return strings.TrimRight(string(b[:maxBytes]), ". ") +} + +// cleanEntryName 清理单个目录名或文件名中的非法字符、控制字符、尾部点空格及 Windows 保留字， +// 确保在 Windows (NTFS/FAT)、Linux (ext4/btrfs/xfs) 及 NAS/SMB 挂载环境下均安全可用。 +func cleanEntryName(name string, isDir bool) string { + name = strings.TrimSpace(name) + if name == "" { + return "unnamed" + } + + if isDir { + clean := sanitizeFilename(name) + clean = strings.Trim(clean, ". ") + if clean == "" { + return "unnamed" + } + if isWindowsReservedName(clean) { + clean = "_" + clean + } + return truncateStringRuneSafe(clean, 255) + } + + ext := filepath.Ext(name) + base := strings.TrimSuffix(name, ext) + + // 清洗扩展名中的非法字符 + cleanExt := sanitizeFilename(ext) + cleanExt = strings.TrimSpace(cleanExt) + if cleanExt != "" && !strings.HasPrefix(cleanExt, ".") { + cleanExt = "." + cleanExt + } + + cleanBase := sanitizeFilename(base) + cleanBase = strings.Trim(cleanBase, ". ") + if cleanBase == "" { + cleanBase = "unnamed" + } + if isWindowsReservedName(cleanBase) { + cleanBase = "_" + cleanBase + } + + maxBaseBytes := 255 - len(cleanExt) + if maxBaseBytes < 10 { + maxBaseBytes = 255 + cleanExt = "" + } + cleanBase = truncateStringRuneSafe(cleanBase, maxBaseBytes) + return cleanBase + cleanExt +} + +// sanitizeRelativePath 清理相对路径中的非法字符与首尾点空格，确保跨操作系统路径合法。 +func sanitizeRelativePath(rel string) string { + rel = strings.TrimSpace(rel) + if rel == "" { + return "" + } + rel = strings.ReplaceAll(rel, "\\", "/") + parts := strings.Split(rel, "/") + out := make([]string, 0, len(parts)) + for i, part := range parts { + part = strings.TrimSpace(part) + if part == "" || part == "." || part == ".." { + continue + } + isDir := i < len(parts)-1 + clean := cleanEntryName(part, isDir) + if clean == "" || clean == "." || clean == ".." { + continue + } + out = append(out, clean) + } + if len(out) == 0 { + return "" + } + return filepath.Join(out...) +} + +// sanitizeLocalPath 对本地全路径中除根目录/盘符/UNC之外的各级目录及文件名进行跨平台非法字符清洗。 +func sanitizeLocalPath(p string) string { + p = strings.TrimSpace(p) + if p == "" || p == "." { + return p + } + vol := filepath.VolumeName(p) + rest := p[len(vol):] + hasRootSlash := len(rest) > 0 && (rest[0] == '/' || rest[0] == '\\') + + parts := strings.FieldsFunc(rest, func(r rune) bool { + return r == '/' || r == '\\' + }) + cleanedParts := make([]string, 0, len(parts)) + for i, part := range parts { + part = strings.TrimSpace(part) + if part == "" || part == "." || part == ".." { + continue + } + isDir := i < len(parts)-1 + clean := cleanEntryName(part, isDir) + if clean == "" || clean == "." || clean == ".." { + continue + } + cleanedParts = append(cleanedParts, clean) + } + prefix := vol + if hasRootSlash { + prefix += string(filepath.Separator) + } + if len(cleanedParts) == 0 { + return filepath.Clean(prefix) + } + if prefix == "" { + return filepath.Clean(filepath.Join(cleanedParts...)) + } + return filepath.Clean(filepath.Join(prefix, filepath.Join(cleanedParts...))) } // joinLocalRel 拼接本地目标路径并校验不越出根目录。 func joinLocalRel(root, rel string) (string, error) { root = filepath.Clean(root) - target := filepath.Clean(filepath.Join(root, filepath.FromSlash(rel))) - if target != root && !strings.HasPrefix(target, root+string(filepath.Separator)) { + cleanRel := sanitizeRelativePath(rel) + if cleanRel == "" { + return "", errors.New("无效相对路径") + } + target := filepath.Clean(filepath.Join(root, cleanRel)) + relToRoot, err := filepath.Rel(root, target) + if err != nil || strings.HasPrefix(relToRoot, "..") || (relToRoot == "." && target != root) { return "", errors.New("本地路径越界") } return target, nil diff --git a/internal/service/strm_sync.go b/internal/service/strm_sync.go index a981b18..f9aa9f1 100644 --- a/internal/service/strm_sync.go +++ b/internal/service/strm_sync.go @@ -253,9 +253,10 @@ func (st *strmSyncState) walkRemote() error { return fmt.Errorf("列出远端目录 %s 失败：%w", task.id, err) } for _, entry := range entries { - rel := entry.Name + cleanName := cleanEntryName(entry.Name, entry.IsDir) + rel := cleanName if task.rel != "" { - rel = task.rel + "/" + entry.Name + rel = task.rel + "/" + cleanName } if entry.IsDir { queue = append(queue, dirTask{id: entry.ID, rel: rel}) @@ -277,8 +278,13 @@ func (st *strmSyncState) processRemoteFile(entry cloud.FileEntry, rel string) { switch { case st.isVideoExt(ext, entry.Size): st.handleVideo(entry, rel, ext) - case st.cfg.DownloadMeta && st.isMetaExt(ext): - st.handleMeta(entry, rel, ext) + case st.isMetaExt(ext): + st.recordRemoteMeta(entry, rel) + if st.cfg.DownloadMeta { + st.handleMeta(entry, rel, ext) + } else { + st.touchProgress() + } default: st.touchProgress() } @@ -414,12 +420,17 @@ func (st *strmSyncState) strmPathParam(rel string) string { } } -// handleMeta 元数据入下载队列（本地已存在且大小一致则跳过）。 -func (st *strmSyncState) handleMeta(entry cloud.FileEntry, rel, ext string) { +// recordRemoteMeta 记录远端存在的元数据索引及文件大小。 +func (st *strmSyncState) recordRemoteMeta(entry cloud.FileEntry, rel string) { st.mu.Lock() st.seenMeta["m:"+rel] = true st.remoteMeta["m:"+rel] = entry.Size st.mu.Unlock() +} + +// handleMeta 元数据入下载队列（本地已存在且大小一致则跳过）。 +func (st *strmSyncState) handleMeta(entry cloud.FileEntry, rel, ext string) { + st.recordRemoteMeta(entry, rel) target, err := joinLocalRel(st.p.LocalPath, rel) if err != nil { @@ -569,12 +580,13 @@ func (st *strmSyncState) scanLocalMetaForUpload() error { return nil } st.mu.Lock() - remoteSize, exists := st.remoteMeta["m:"+rel] + _, exists := st.remoteMeta["m:"+rel] st.mu.Unlock() - if exists && remoteSize == info.Size() { + if exists { + // 网盘端已存在该元数据文件，跳过上传 return nil } - if st.taskExists("upload", st.p.ID, rel) { + if st.taskExists("upload", st.p.ID, path) { return nil } task := &model.StrmUploadTask{ diff --git a/internal/service/strm_sync_test.go b/internal/service/strm_sync_test.go index a7b0801..2b05a2c 100644 --- a/internal/service/strm_sync_test.go +++ b/internal/service/strm_sync_test.go @@ -202,3 +202,141 @@ func TestStrmDefaultBaseURL(t *testing.T) { t.Fatal("effective config base url should fall back to default") } } + +func TestSanitizePathWithSpecialChars(t *testing.T) { + cases := []struct { + input string + isDir bool + wantName string + }{ + {"数码宝贝:拯救者", true, "数码宝贝 拯救者"}, + {"数码宝贝:拯救者.mp4", false, "数码宝贝 拯救者.mp4"}, + {"Season 01.", true, "Season 01"}, + {"file:name?test*|bar\".mkv", false, "file nametestfoobar.mkv"}, + {"poster.jpg", false, "poster.jpg"}, + {" trailing space ", true, "trailing space"}, + {"CON", true, "_CON"}, + {"aux.mp4", false, "_aux.mp4"}, + {"NUL.nfo", false, "_NUL.nfo"}, + {"COM1", false, "_COM1"}, + {"test\x00\x1f\x7fcontrol.mkv", false, "testcontrol.mkv"}, + } + for _, tc := range cases { + got := cleanEntryName(tc.input, tc.isDir) + if got != tc.wantName { + t.Errorf("cleanEntryName(%q, %v) = %q, want %q", tc.input, tc.isDir, got, tc.wantName) + } + } + + // Test sanitizeRelativePath + rel := "动漫/数码宝贝:拯救者/数码宝贝:拯救者.mp4" + wantRel := filepath.Join("动漫", "数码宝贝 拯救者", "数码宝贝 拯救者.mp4") + if got := sanitizeRelativePath(rel); got != wantRel { + t.Errorf("sanitizeRelativePath(%q) = %q, want %q", rel, got, wantRel) + } + + // Test sanitizeLocalPath - Windows Drive + localWin := `D:\test\动漫\数码宝贝:拯救者\poster.jpg` + wantLocalWin := filepath.Join(`D:\`, "test", "动漫", "数码宝贝 拯救者", "poster.jpg") + if got := sanitizeLocalPath(localWin); got != wantLocalWin { + t.Errorf("sanitizeLocalPath(%q) = %q, want %q", localWin, got, wantLocalWin) + } + + // Test sanitizeLocalPath - Linux root + localLinux := `/media/data/动漫/数码宝贝:拯救者/ep01:拯救者.mkv` + cleanedLinux := sanitizeLocalPath(localLinux) + if strings.Contains(cleanedLinux, ":") && !strings.HasPrefix(cleanedLinux, "D:") && !strings.HasPrefix(cleanedLinux, "C:") { + t.Errorf("sanitizeLocalPath should not contain colons in path components: %q", cleanedLinux) + } + + // Test joinLocalRel + root := t.TempDir() + joined, err := joinLocalRel(root, "动漫/数码宝贝:拯救者/poster.jpg") + if err != nil { + t.Fatalf("joinLocalRel failed: %v", err) + } + wantJoined := filepath.Join(root, "动漫", "数码宝贝 拯救者", "poster.jpg") + if joined != wantJoined { + t.Errorf("joinLocalRel = %q, want %q", joined, wantJoined) + } + + // Test joinLocalRel prevents path traversal + if _, err := joinLocalRel(root, "../../"); err == nil { + t.Error("joinLocalRel should error on empty/traversal-only path") + } + traversalCleaned, err := joinLocalRel(root, "../../etc/passwd") + if err != nil { + t.Fatalf("joinLocalRel failed on traversal path: %v", err) + } + if !strings.HasPrefix(traversalCleaned, root) { + t.Errorf("joinLocalRel allowed path escape: %s", traversalCleaned) + } + + // Verify mkdir succeeds on this joined path + if err := os.MkdirAll(filepath.Dir(joined), 0o755); err != nil { + t.Fatalf("MkdirAll on joined path failed: %v", err) + } + if info, err := os.Stat(filepath.Dir(joined)); err != nil || !info.IsDir() { + t.Fatalf("Dir not created: %v", err) + } +} + +// TestScanLocalMetaForUpload 验证元数据上传比对逻辑：网盘已存在跳过，网盘不存在才入队上传。 +func TestScanLocalMetaForUpload(t *testing.T) { + svc := testStrmService(t) + localDir := t.TempDir() + + // 本地有 2 个元数据：poster.jpg 和 fanart.jpg + writeFile(t, filepath.Join(localDir, "动漫", "poster.jpg"), "poster-data") + writeFile(t, filepath.Join(localDir, "动漫", "fanart.jpg"), "fanart-data") + + p := &model.StrmSyncPath{ + Base: model.Base{ID: "test-path-upload"}, + AccountID: "acct-1", + Provider: model.StrmProviderCloudDrive, + RemotePath: "/Media", + LocalPath: localDir, + UploadMeta: true, + } + + st := &strmSyncState{ + s: svc, + ctx: context.Background(), + p: p, + cfg: &strmPathConfig{UploadMeta: true, MetaExt: []string{"jpg", "nfo"}}, + rec: &model.StrmSyncRecord{}, + seenMeta: map[string]bool{}, + remoteMeta: map[string]int64{}, + } + + // 模拟远端已存在 poster.jpg + st.remoteMeta["m:动漫/poster.jpg"] = 1000 + + if err := st.scanLocalMetaForUpload(); err != nil { + t.Fatalf("scanLocalMetaForUpload failed: %v", err) + } + + // 此时应该只有 fanart.jpg 入队上传，poster.jpg 被跳过 + tasks, _, err := svc.repo.StrmUpload.List(context.Background(), "", 1, 10) + if err != nil { + t.Fatal(err) + } + if len(tasks) != 1 { + t.Fatalf("expected 1 upload task (fanart.jpg), got %d", len(tasks)) + } + if tasks[0].FileName != "fanart.jpg" { + t.Errorf("expected upload task for fanart.jpg, got %s", tasks[0].FileName) + } + + // 再次扫描：fanart.jpg 已经在队列中，应自动去重，不重复入队 + if err := st.scanLocalMetaForUpload(); err != nil { + t.Fatalf("second scanLocalMetaForUpload failed: %v", err) + } + tasks, _, err = svc.repo.StrmUpload.List(context.Background(), "", 1, 10) + if err != nil { + t.Fatal(err) + } + if len(tasks) != 1 { + t.Fatalf("expected still 1 upload task after dedup, got %d", len(tasks)) + } +} diff --git a/internal/service/subtitle_test.go b/internal/service/subtitle_test.go index 90a5565..080cfa7 100644 --- a/internal/service/subtitle_test.go +++ b/internal/service/subtitle_test.go @@ -47,12 +47,12 @@ func TestSubtitleDiscoverNoTracksReturnsEmptySlice(t *testing.T) { func TestNormaliseTimecode(t *testing.T) { cases := map[string]string{ - "0:00:01": "00:00:01", - "0:00:01.51": "00:00:01.510", // ASS centiseconds -> 3-digit ms - "0:00:04.123": "00:00:04.123", // already 3 digits - "00:00:12.07": "00:00:12.070", // 2-digit fraction padded + "0:00:01": "00:00:01", + "0:00:01.51": "00:00:01.510", // ASS centiseconds -> 3-digit ms + "0:00:04.123": "00:00:04.123", // already 3 digits + "00:00:12.07": "00:00:12.070", // 2-digit fraction padded "00:01:30.1234": "00:01:30.123", // capped at 3 digits - "00:00:1.5": "00:00:01.500", // single-digit seconds and fraction + "00:00:1.5": "00:00:01.500", // single-digit seconds and fraction } for in, want := range cases { if got := normaliseTimecode(in); got != want { diff --git a/internal/service/task_tracker_organize_items.go b/internal/service/task_tracker_organize_items.go index e95b826..9f5d6ee 100644 --- a/internal/service/task_tracker_organize_items.go +++ b/internal/service/task_tracker_organize_items.go @@ -25,10 +25,10 @@ const ( // exactly one file (organize) or one library (ingest / scrape). type TaskItemRecord struct { ID string `json:"id"` - Kind string `json:"kind"` // organize / scan / scrape - Status string `json:"status"` // pending / running / succeeded / failed - Name string `json:"name"` // display name (file base name or library name) - Source string `json:"source,omitempty"` // source file path (organize) + Kind string `json:"kind"` // organize / scan / scrape + Status string `json:"status"` // pending / running / succeeded / failed + Name string `json:"name"` // display name (file base name or library name) + Source string `json:"source,omitempty"` // source file path (organize) DestPath string `json:"dest_path,omitempty"` LibraryID string `json:"library_id,omitempty"` Error string `json:"error,omitempty"`