QWQLwToo
362 lines
9.4 KiB
Go
362 lines
9.4 KiB
Go
package sources
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"net/http"
|
|
"net/url"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"ymhut-box/server/unified-management/internal/config"
|
|
"ymhut-box/server/unified-management/internal/db"
|
|
)
|
|
|
|
type Service struct {
|
|
cfg *config.Config
|
|
store *db.Store
|
|
client *http.Client
|
|
stop chan struct{}
|
|
once sync.Once
|
|
}
|
|
|
|
type legacyMedia struct {
|
|
Categories []legacyCategory `json:"categories"`
|
|
}
|
|
|
|
type legacyCategory struct {
|
|
ID string `json:"id"`
|
|
Name string `json:"name"`
|
|
Enabled *bool `json:"enabled"`
|
|
Subcategories []legacySubcategory `json:"subcategories"`
|
|
}
|
|
|
|
type legacySubcategory struct {
|
|
ID string `json:"id"`
|
|
Name string `json:"name"`
|
|
Description string `json:"description"`
|
|
APIURL string `json:"api_url"`
|
|
ThumbnailURL string `json:"thumbnail_url"`
|
|
RefreshInterval int `json:"refresh_interval"`
|
|
SupportedFormats []string `json:"supported_formats"`
|
|
Downloadable bool `json:"downloadable"`
|
|
}
|
|
|
|
func NewService(cfg *config.Config, store *db.Store) *Service {
|
|
return &Service{
|
|
cfg: cfg,
|
|
store: store,
|
|
client: &http.Client{Timeout: 10 * time.Second},
|
|
stop: make(chan struct{}),
|
|
}
|
|
}
|
|
|
|
func (s *Service) Start(ctx context.Context) {
|
|
s.once.Do(func() {
|
|
go s.loop()
|
|
})
|
|
}
|
|
|
|
func (s *Service) Stop() {
|
|
close(s.stop)
|
|
}
|
|
|
|
func (s *Service) loop() {
|
|
ticker := time.NewTicker(time.Duration(s.cfg.SourceCheckSeconds) * time.Second)
|
|
defer ticker.Stop()
|
|
s.CheckDue(context.Background())
|
|
for {
|
|
select {
|
|
case <-ticker.C:
|
|
s.CheckDue(context.Background())
|
|
case <-s.stop:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Service) ImportLegacyMediaTypesIfEmpty(ctx context.Context) error {
|
|
count, err := s.store.CountSources()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if count > 0 {
|
|
return nil
|
|
}
|
|
return s.ImportLegacyMediaTypes(ctx)
|
|
}
|
|
|
|
func (s *Service) ImportLegacyMediaTypes(ctx context.Context) error {
|
|
data, err := os.ReadFile(filepath.Join(s.cfg.UpdatePublicDir, "media-types.json"))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var legacy legacyMedia
|
|
if err := json.Unmarshal(data, &legacy); err != nil {
|
|
return err
|
|
}
|
|
for _, category := range legacy.Categories {
|
|
for _, sub := range category.Subcategories {
|
|
if strings.TrimSpace(sub.APIURL) == "" {
|
|
continue
|
|
}
|
|
formats, _ := json.Marshal(sub.SupportedFormats)
|
|
_, err := s.store.UpsertSource(db.Source{
|
|
CategoryID: defaultString(category.ID, "media"),
|
|
CategoryName: defaultString(category.Name, category.ID),
|
|
SourceID: defaultString(sub.ID, category.ID+"-"+sub.Name),
|
|
Name: defaultString(sub.Name, sub.ID),
|
|
Description: sub.Description,
|
|
Method: "GET",
|
|
APIURL: sub.APIURL,
|
|
ThumbnailURL: sub.ThumbnailURL,
|
|
ProxyMode: "client_direct",
|
|
TimeoutMS: 8000,
|
|
RetryCount: 1,
|
|
CheckIntervalSec: maxInt(sub.RefreshInterval, 300),
|
|
Enabled: legacyEnabled(category.Enabled),
|
|
ClientVisible: true,
|
|
SupportedFormats: string(formats),
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) Catalog(includeHidden bool) (map[string]any, error) {
|
|
items, err := s.store.ListSources(includeHidden)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
categories := map[string]map[string]any{}
|
|
for _, item := range items {
|
|
cat, ok := categories[item.CategoryID]
|
|
if !ok {
|
|
cat = map[string]any{
|
|
"id": item.CategoryID,
|
|
"name": item.CategoryName,
|
|
"enabled": true,
|
|
"subcategories": []map[string]any{},
|
|
}
|
|
categories[item.CategoryID] = cat
|
|
}
|
|
var formats []string
|
|
_ = json.Unmarshal([]byte(item.SupportedFormats), &formats)
|
|
sub := map[string]any{
|
|
"id": item.SourceID,
|
|
"name": item.Name,
|
|
"description": item.Description,
|
|
"api_url": item.APIURL,
|
|
"urlTemplate": firstNonEmpty(item.URLTemplate, item.APIURL),
|
|
"thumbnail_url": item.ThumbnailURL,
|
|
"method": item.Method,
|
|
"proxy_mode": item.ProxyMode,
|
|
"proxyMode": item.ProxyMode,
|
|
"refresh_interval": item.CheckIntervalSec,
|
|
"cacheSeconds": item.CacheSeconds,
|
|
"supported_formats": formats,
|
|
"downloadable": true,
|
|
"health": map[string]any{
|
|
"status": item.LastStatus,
|
|
"latency_ms": item.LastLatencyMS,
|
|
"last_checked_at": item.LastCheckedAt,
|
|
"last_error": item.LastError,
|
|
"consecutiveFailure": item.ConsecutiveFailure,
|
|
"meta": parseHealthMeta(item.LastError),
|
|
},
|
|
}
|
|
cat["subcategories"] = append(cat["subcategories"].([]map[string]any), sub)
|
|
}
|
|
out := []map[string]any{}
|
|
for _, cat := range categories {
|
|
out = append(out, cat)
|
|
}
|
|
return map[string]any{
|
|
"layout_version": "2.0.0",
|
|
"last_updated": time.Now().UTC().Format(time.RFC3339),
|
|
"categories": out,
|
|
}, nil
|
|
}
|
|
|
|
func (s *Service) Endpoints(includeHidden bool) ([]map[string]any, error) {
|
|
items, err := s.store.ListSources(includeHidden)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out := []map[string]any{}
|
|
for _, item := range items {
|
|
var formats []string
|
|
_ = json.Unmarshal([]byte(item.SupportedFormats), &formats)
|
|
out = append(out, map[string]any{
|
|
"id": item.SourceID,
|
|
"category": item.CategoryID,
|
|
"name": item.Name,
|
|
"method": item.Method,
|
|
"urlTemplate": firstNonEmpty(item.URLTemplate, item.APIURL),
|
|
"proxyMode": item.ProxyMode,
|
|
"clientVisible": item.ClientVisible,
|
|
"enabled": item.Enabled,
|
|
"cacheSeconds": item.CacheSeconds,
|
|
"supportedFormats": formats,
|
|
"health": map[string]any{
|
|
"status": item.LastStatus,
|
|
"latencyMs": item.LastLatencyMS,
|
|
"lastCheckedAt": item.LastCheckedAt,
|
|
"lastError": item.LastError,
|
|
"consecutiveFailure": item.ConsecutiveFailure,
|
|
"meta": parseHealthMeta(item.LastError),
|
|
},
|
|
})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (s *Service) CheckDue(ctx context.Context) {
|
|
items, err := s.store.ListSources(true)
|
|
if err != nil {
|
|
return
|
|
}
|
|
now := time.Now()
|
|
for _, item := range items {
|
|
if !item.Enabled {
|
|
continue
|
|
}
|
|
if item.LastCheckedAt != "" {
|
|
if last, err := time.Parse(time.RFC3339, item.LastCheckedAt); err == nil && now.Sub(last) < time.Duration(item.CheckIntervalSec)*time.Second {
|
|
continue
|
|
}
|
|
}
|
|
_ = s.CheckOne(ctx, item)
|
|
}
|
|
}
|
|
|
|
func (s *Service) CheckSourceID(ctx context.Context, sourceID string) (db.Source, error) {
|
|
item, err := s.store.GetSourceBySourceID(sourceID)
|
|
if err != nil {
|
|
return db.Source{}, err
|
|
}
|
|
return item, s.CheckOne(ctx, item)
|
|
}
|
|
|
|
func (s *Service) CheckOne(ctx context.Context, item db.Source) error {
|
|
if strings.TrimSpace(item.APIURL) == "" {
|
|
return errors.New("source api_url is empty")
|
|
}
|
|
timeout := time.Duration(item.TimeoutMS) * time.Millisecond
|
|
if timeout <= 0 {
|
|
timeout = 8 * time.Second
|
|
}
|
|
ctx, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
method := strings.TrimSpace(item.Method)
|
|
if method == "" {
|
|
method = http.MethodGet
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, method, item.APIURL, nil)
|
|
if err != nil {
|
|
_ = s.store.RecordSourceCheck(item.ID, "error", 0, err.Error())
|
|
return err
|
|
}
|
|
redirects := []string{}
|
|
client := *s.client
|
|
client.Timeout = timeout
|
|
client.CheckRedirect = func(next *http.Request, via []*http.Request) error {
|
|
if next.URL == nil || !isHTTPURL(next.URL) {
|
|
return errors.New("redirect target must be http or https")
|
|
}
|
|
redirects = append(redirects, next.URL.String())
|
|
if len(via) >= 5 {
|
|
return errors.New("too many redirects")
|
|
}
|
|
return nil
|
|
}
|
|
start := time.Now()
|
|
resp, err := client.Do(req)
|
|
latency := int(time.Since(start).Milliseconds())
|
|
if err != nil {
|
|
_ = s.store.RecordSourceCheck(item.ID, "error", latency, err.Error())
|
|
return err
|
|
}
|
|
defer resp.Body.Close()
|
|
status := "ok"
|
|
message := ""
|
|
if len(redirects) > 0 {
|
|
status = "redirected"
|
|
message = healthMetaMessage(map[string]any{
|
|
"redirected": true,
|
|
"redirectCount": len(redirects),
|
|
"finalUrl": resp.Request.URL.String(),
|
|
"finalStatus": resp.StatusCode,
|
|
})
|
|
}
|
|
if resp.StatusCode >= 400 {
|
|
status = "degraded"
|
|
message = healthMetaMessage(map[string]any{
|
|
"redirected": len(redirects) > 0,
|
|
"redirectCount": len(redirects),
|
|
"finalUrl": resp.Request.URL.String(),
|
|
"finalStatus": resp.StatusCode,
|
|
"error": resp.Status,
|
|
})
|
|
}
|
|
return s.store.RecordSourceCheck(item.ID, status, latency, message)
|
|
}
|
|
|
|
func isHTTPURL(value *url.URL) bool {
|
|
scheme := strings.ToLower(value.Scheme)
|
|
return scheme == "http" || scheme == "https"
|
|
}
|
|
|
|
func healthMetaMessage(meta map[string]any) string {
|
|
data, err := json.Marshal(meta)
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
return string(data)
|
|
}
|
|
|
|
func parseHealthMeta(message string) map[string]any {
|
|
var meta map[string]any
|
|
if strings.TrimSpace(message) == "" || json.Unmarshal([]byte(message), &meta) != nil {
|
|
return map[string]any{}
|
|
}
|
|
return meta
|
|
}
|
|
|
|
func defaultString(value, fallback string) string {
|
|
if strings.TrimSpace(value) == "" {
|
|
return fallback
|
|
}
|
|
return value
|
|
}
|
|
|
|
func legacyEnabled(value *bool) bool {
|
|
if value == nil {
|
|
return true
|
|
}
|
|
return *value
|
|
}
|
|
|
|
func maxInt(value, fallback int) int {
|
|
if value > 0 {
|
|
return value
|
|
}
|
|
return fallback
|
|
}
|
|
|
|
func firstNonEmpty(values ...string) string {
|
|
for _, value := range values {
|
|
if strings.TrimSpace(value) != "" {
|
|
return strings.TrimSpace(value)
|
|
}
|
|
}
|
|
return ""
|
|
}
|