skong
/
synctv
mirror of https://github.com/synctv-org/synctv.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refecrot: vendor backend

This commit is contained in:
zijiren233 2023-12-14 21:51:18 +08:00
parent b3e498d365
commit c796738e6f
29 changed files with 775 additions and 654 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmd/server.go
View File

@ -32,7 +32,7 @@ var ServerCmd = &cobra.Command{
bootstrap.InitProvider,
bootstrap.InitOp,
bootstrap.InitRtmp,
bootstrap.InitVendor,
bootstrap.InitVendorBackend,
bootstrap.InitSetting,
)
if !flags.DisableUpdateCheck {

2
go.mod
View File

@ -28,7 +28,7 @@ require (
github.com/sirupsen/logrus v1.9.3
github.com/soheilhy/cmux v0.1.5
github.com/spf13/cobra v1.8.0
github.com/synctv-org/vendors v0.1.1-0.20231212054308-d2740988c951
github.com/synctv-org/vendors v0.1.1-0.20231213112456-9743b943a97c
github.com/ulule/limiter/v3 v3.11.2
github.com/zencoder/go-dash/v3 v3.0.3
github.com/zijiren233/gencontainer v0.0.0-20231213075414-f7f4c8261dca

2
go.sum
View File

@ -356,6 +356,8 @@ github.com/synctv-org/vendors v0.1.1-0.20231212053257-8f9b1f19d5a3 h1:DbbFcU5wqn
github.com/synctv-org/vendors v0.1.1-0.20231212053257-8f9b1f19d5a3/go.mod h1:FX5xPnIKQGcadFONDfq30OPgQcTrpZLbW/9ebbz+2qY=
github.com/synctv-org/vendors v0.1.1-0.20231212054308-d2740988c951 h1:xnyL4KYIpNA5TD3c3cL9Kq2FMwpv8r+oGThWO+T9IWM=
github.com/synctv-org/vendors v0.1.1-0.20231212054308-d2740988c951/go.mod h1:hzrMvLeO3iEDYiKM6dk4P6uWe3UwkTFBvYnWf/d/w1o=
github.com/synctv-org/vendors v0.1.1-0.20231213112456-9743b943a97c h1:FhIqTq7CKtqn3xvdDAxM6Po2ImcXDw2KaO/Um34KNAs=
github.com/synctv-org/vendors v0.1.1-0.20231213112456-9743b943a97c/go.mod h1:C0ZGPeF8nYsx60gePPhSrgrjrONj6F72XFHW9Mh8+vU=
github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=
github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=

15
internal/bootstrap/vendor.go
View File

@ -3,10 +3,19 @@ package bootstrap
import (
"context"
"github.com/synctv-org/synctv/internal/conf"
"github.com/synctv-org/synctv/internal/db"
"github.com/synctv-org/synctv/internal/vendor"
)
func InitVendor(ctx context.Context) error {
return vendor.Init(&conf.Conf.Vendor)
func InitVendorBackend(ctx context.Context) error {
vb, err := db.GetAllVendorBackend()
if err != nil {
return err
}
b, err := vendor.NewBackends(ctx, vb)
if err != nil {
return err
}
vendor.StoreBackends(b)
return nil
}

4
internal/cache/alist.go vendored
View File

@ -31,7 +31,7 @@ func NewAlistCache(userID string) *AlistUserCache {
func AlistAuthorizationCacheWithConfigInitFunc(host, username, password, backend string) func(ctx context.Context, args ...string) (*AlistUserCacheData, error) {
return func(ctx context.Context, args ...string) (*AlistUserCacheData, error) {
cli := vendor.AlistClient(backend)
cli := vendor.LoadAlistClient(backend)
if username == "" {
_, err := cli.Me(ctx, &alist.MeReq{
Host: host,
@ -86,7 +86,7 @@ func NewAlistMovieCacheInitFunc(user *AlistUserCache, movie *model.Movie) func(c
if aucd.Host == "" {
return nil, errors.New("not bind alist vendor")
}
cli := vendor.AlistClient(movie.Base.VendorInfo.Backend)
cli := vendor.LoadAlistClient(movie.Base.VendorInfo.Backend)
fg, err := cli.FsGet(ctx, &alist.FsGetReq{
Host: aucd.Host,
Token: aucd.Token,

6
internal/cache/bilibili.go vendored
View File

@ -46,7 +46,7 @@ func BilibiliSharedMpdCacheInitFunc(ctx context.Context, movie *model.Movie, arg
} else {
cookies = utils.MapToHttpCookie(vendorInfo.Cookies)
}
cli := vendor.BilibiliClient(movie.Base.VendorInfo.Backend)
cli := vendor.LoadBilibiliClient(movie.Base.VendorInfo.Backend)
var m, hevcM *mpd.MPD
biliInfo := movie.Base.VendorInfo.Bilibili
switch {
@ -143,7 +143,7 @@ func BilibiliNoSharedMovieCacheInitFunc(ctx context.Context, id string, movie *m
} else {
cookies = utils.MapToHttpCookie(vendorInfo.Cookies)
}
cli := vendor.BilibiliClient(movie.Base.VendorInfo.Backend)
cli := vendor.LoadBilibiliClient(movie.Base.VendorInfo.Backend)
var u string
biliInfo := movie.Base.VendorInfo.Bilibili
switch {
@ -219,7 +219,7 @@ func initBilibiliSubtitleCache(ctx context.Context, movie *model.Movie, args ...
} else {
cookies = utils.MapToHttpCookie(vendorInfo.Cookies)
}
cli := vendor.BilibiliClient(movie.Base.VendorInfo.Backend)
cli := vendor.LoadBilibiliClient(movie.Base.VendorInfo.Backend)
resp, err := cli.GetSubtitles(ctx, &bilibili.GetSubtitlesReq{
Cookies: utils.HttpCookieToMap(cookies),
Bvid: biliInfo.Bvid,

6
internal/conf/config.go
View File

@ -22,9 +22,6 @@ type Config struct {
// RateLimit
RateLimit RateLimitConfig `yaml:"rate_limit"`
// Vendor
Vendor VendorConfig `yaml:"vendor"`
}
func (c *Config) Save(file string) error {
@ -50,8 +47,5 @@ func DefaultConfig() *Config {
// RateLimit
RateLimit: DefaultRateLimitConfig(),
// Vendor
Vendor: DefaultVendorConfig(),
}
}

48
internal/conf/vendor.go
View File

@ -1,48 +0,0 @@
package conf
type VendorConfig struct {
Bilibili map[string]BilibiliConfig `yaml:"bilibili" hc:"default use local vendor"`
Alist map[string]AlistConfig `yaml:"alist" hc:"default use local vendor"`
}
func DefaultVendorConfig() VendorConfig {
return VendorConfig{
Bilibili: nil,
}
}
type Consul struct {
Endpoint string `yaml:"endpoint"`
Token string `yaml:"token,omitempty"`
TokenFile string `yaml:"token_file,omitempty"`
PathPrefix string `yaml:"path_prefix,omitempty"`
Namespace string `yaml:"namespace,omitempty"`
Partition string `yaml:"partition,omitempty"`
}
type Etcd struct {
Endpoints []string `yaml:"endpoints"`
Username string `yaml:"username,omitempty"`
Password string `yaml:"password,omitempty"`
}
type VendorBase struct {
ServerName string `yaml:"server_name" hc:"if use tls and grpc, servername must set the cert server name" env:"BILIBILI_SERVER_NAME"`
Endpoint string `yaml:"endpoint" env:"BILIBILI_ENDPOINT"`
JwtSecret string `yaml:"jwt_secret" env:"BILIBILI_JWT_SECRET"`
Scheme string `yaml:"scheme" lc:"grpc | http" env:"BILIBILI_SCHEME"`
Tls bool `yaml:"tls" env:"BILIBILI_TLS"`
CustomCAFile string `yaml:"custom_ca_file,omitempty" env:"BILIBILI_CUSTOM_CA_FILE"`
TimeOut string `yaml:"time_out" env:"BILIBILI_TIME_OUT"`
Consul Consul `yaml:"consul,omitempty" hc:"if use consul, must set the endpoint"`
Etcd Etcd `yaml:"etcd,omitempty" hc:"if use etcd, must set the endpoints"`
}
type BilibiliConfig struct {
VendorBase `yaml:",inline"`
}
type AlistConfig struct {
VendorBase `yaml:",inline"`
}

1
internal/db/db.go
View File

@ -30,6 +30,7 @@ func Init(d *gorm.DB, t conf.DatabaseType) error {
new(model.Movie),
new(model.BilibiliVendor),
new(model.AlistVendor),
new(model.VendorBackend),
)
}

46
internal/db/vendorBackend.go Normal file
View File

@ -0,0 +1,46 @@
package db
import (
"errors"
"github.com/synctv-org/synctv/internal/model"
"gorm.io/gorm"
)
func GetAllVendorBackend() ([]*model.VendorBackend, error) {
var backends []*model.VendorBackend
err := db.Find(&backends).Error
return backends, HandleNotFound(err, "backends")
}
func CreateVendorBackend(backend *model.VendorBackend) error {
return db.Create(backend).Error
}
func DeleteVendorBackend(endpoint string) error {
return db.Where("backend_endpoint = ?", endpoint).Delete(&model.VendorBackend{}).Error
}
func DeleteVendorBackends(endpoints []string) error {
return db.Where("backend_endpoint IN ?", endpoints).Delete(&model.VendorBackend{}).Error
}
func GetVendorBackend(endpoint string) (*model.VendorBackend, error) {
var backend model.VendorBackend
err := db.Where("backend_endpoint = ?", endpoint).First(&backend).Error
return &backend, HandleNotFound(err, "backend")
}
func CreateOrSaveVendorBackend(backend *model.VendorBackend) (*model.VendorBackend, error) {
return backend, Transactional(func(tx *gorm.DB) error {
if err := tx.Where("backend_endpoint = ?", backend.Backend.Endpoint).First(&model.VendorBackend{}).Error; errors.Is(err, gorm.ErrRecordNotFound) {
return tx.Create(&backend).Error
} else {
return tx.Save(&backend).Error
}
})
}
func SaveVendorBackend(backend *model.VendorBackend) error {
return db.Save(backend).Error
}

0
internal/db/vendor.go → internal/db/vendorRecord.go
View File

3
internal/model/movie.go
View File

@ -47,12 +47,12 @@ type VendorName = string
const (
VendorBilibili VendorName = "bilibili"
VendorAlist VendorName = "alist"
VendorEmby VendorName = "emby"
)
type VendorInfo struct {
Vendor VendorName `json:"vendor"`
Backend string `json:"backend"`
Shared bool `gorm:"not null;default:false" json:"shared"`
Bilibili *BilibiliStreamingInfo `gorm:"embedded;embeddedPrefix:bilibili_" json:"bilibili,omitempty"`
Alist *AlistStreamingInfo `gorm:"embedded;embeddedPrefix:alist_" json:"alist,omitempty"`
}
@ -62,6 +62,7 @@ type BilibiliStreamingInfo struct {
Cid uint64 `json:"cid,omitempty"`
Epid uint64 `json:"epid,omitempty"`
Quality uint64 `json:"quality,omitempty"`
Shared bool `json:"shared,omitempty"`
}
func (b *BilibiliStreamingInfo) Validate() error {

98
internal/model/vendorBackend.go Normal file
View File

@ -0,0 +1,98 @@
package model
import (
"github.com/synctv-org/synctv/utils"
"gorm.io/gorm"
)
type Consul struct {
ServerName string
Token string
TokenFile string
PathPrefix string
Namespace string
Partition string
}
type Etcd struct {
ServerName string
Username string
Password string
}
type Backend struct {
Endpoint string `gorm:"primaryKey" json:"endpoint"`
Comment string `gorm:"type:text" json:"comment"`
Tls bool `gorm:"default:false" json:"tls"`
JwtSecret string `json:"jwtSecret"`
CustomCAFile string `json:"customCaFile"`
TimeOut string `gorm:"default:10s" json:"timeOut"`
Consul Consul `gorm:"embedded;embeddedPrefix:consul_" json:"consul"`
Etcd Etcd `gorm:"embedded;embeddedPrefix:etcd_" json:"etcd"`
}
type VendorBackend struct {
Backend Backend `gorm:"embedded;embeddedPrefix:backend_" json:"backend"`
UsedBy BackendUsedBy `gorm:"embedded;embeddedPrefix:used_by_" json:"usedBy"`
}
type BackendUsedBy struct {
Bilibili bool `gorm:"default:false" json:"bilibili"`
BilibiliBackendName string `json:"bilibiliBackendName"`
Alist bool `gorm:"default:false" json:"alist"`
AlistBackendName string `json:"alistBackendName"`
Emby bool `gorm:"default:false" json:"emby"`
EmbyBackendName string `json:"embyBackendName"`
}
func (v *VendorBackend) BeforeSave(tx *gorm.DB) error {
key := []byte(v.Backend.Endpoint)
var err error
if v.Backend.JwtSecret != "" {
if v.Backend.JwtSecret, err = utils.CryptoToBase64([]byte(v.Backend.JwtSecret), key); err != nil {
return err
}
}
if v.Backend.Consul.Token != "" {
if v.Backend.Consul.Token, err = utils.CryptoToBase64([]byte(v.Backend.Consul.Token), key); err != nil {
return err
}
}
if v.Backend.Etcd.Password != "" {
if v.Backend.Etcd.Password, err = utils.CryptoToBase64([]byte(v.Backend.Etcd.Password), key); err != nil {
return err
}
}
return nil
}
func (v *VendorBackend) AfterFind(tx *gorm.DB) error {
key := []byte(v.Backend.Endpoint)
var (
err error
data []byte
)
if v.Backend.JwtSecret != "" {
if data, err = utils.DecryptoFromBase64(v.Backend.JwtSecret, key); err != nil {
return err
} else {
v.Backend.JwtSecret = string(data)
}
}
if v.Backend.Consul.Token != "" {
if data, err = utils.DecryptoFromBase64(v.Backend.Consul.Token, key); err != nil {
return err
} else {
v.Backend.Consul.Token = string(data)
}
}
if v.Backend.Etcd.Password != "" {
if data, err = utils.DecryptoFromBase64(v.Backend.Etcd.Password, key); err != nil {
return err
} else {
v.Backend.Etcd.Password = string(data)
}
}
return nil
}

0
internal/model/vendor.go → internal/model/vendorRecord.go
View File

279
internal/vendor/alist.go vendored
View File

@ -2,256 +2,41 @@ package vendor
import (
"context"
"crypto/tls"
"crypto/x509"
"errors"
"fmt"
"os"
"time"
"github.com/go-kratos/aegis/circuitbreaker"
"github.com/go-kratos/aegis/circuitbreaker/sre"
consul "github.com/go-kratos/kratos/contrib/registry/consul/v2"
"github.com/go-kratos/kratos/contrib/registry/etcd/v2"
"github.com/go-kratos/kratos/v2/middleware"
"github.com/go-kratos/kratos/v2/middleware/auth/jwt"
kcircuitbreaker "github.com/go-kratos/kratos/v2/middleware/circuitbreaker"
"google.golang.org/grpc"
ggrpc "github.com/go-kratos/kratos/v2/transport/grpc"
"github.com/go-kratos/kratos/v2/transport/http"
jwtv4 "github.com/golang-jwt/jwt/v4"
"github.com/hashicorp/consul/api"
log "github.com/sirupsen/logrus"
"github.com/synctv-org/synctv/internal/conf"
"github.com/synctv-org/vendors/api/alist"
alistService "github.com/synctv-org/vendors/service/alist"
clientv3 "go.etcd.io/etcd/client/v3"
)
type AlistInterface = alist.AlistHTTPServer
func AlistClient(name string) AlistInterface {
if name != "" {
if cli, ok := alistClients[name]; ok {
return cli
}
func LoadAlistClient(name string) AlistInterface {
if cli, ok := backends.Load().alist[name]; ok {
return cli
}
return alistDefaultClient
}
func AlistClients() map[string]AlistInterface {
return alistClients
return alistLocalClient
}
var (
alistClients map[string]AlistInterface
alistDefaultClient AlistInterface
alistLocalClient AlistInterface
)
func InitAlistVendors(conf map[string]conf.AlistConfig) error {
if alistClients == nil {
alistClients = make(map[string]AlistInterface, len(conf))
}
for k, vb := range conf {
cli, err := InitAlist(&vb)
if err != nil {
return err
}
if k == "" {
alistDefaultClient = cli
} else {
alistClients[k] = cli
}
}
if alistDefaultClient == nil {
alistDefaultClient = alistService.NewAlistService(nil)
}
return nil
func init() {
alistLocalClient = alistService.NewAlistService(nil)
}
func InitAlist(conf *conf.AlistConfig) (AlistInterface, error) {
middlewares := []middleware.Middleware{kcircuitbreaker.Client(kcircuitbreaker.WithCircuitBreaker(func() circuitbreaker.CircuitBreaker {
return sre.NewBreaker(
sre.WithRequest(25),
sre.WithWindow(time.Second*15),
)
}))}
func AlistLocalClient() AlistInterface {
return alistLocalClient
}
if conf.JwtSecret != "" {
key := []byte(conf.JwtSecret)
middlewares = append(middlewares, jwt.Client(func(token *jwtv4.Token) (interface{}, error) {
return key, nil
}, jwt.WithSigningMethod(jwtv4.SigningMethodHS256)))
}
switch conf.Scheme {
case "grpc":
opts := []ggrpc.ClientOption{}
opts = append(opts, ggrpc.WithMiddleware(middlewares...))
if conf.TimeOut != "" {
timeout, err := time.ParseDuration(conf.TimeOut)
if err != nil {
return nil, err
}
opts = append(opts, ggrpc.WithTimeout(timeout))
}
if conf.Endpoint != "" {
opts = append(opts, ggrpc.WithEndpoint(conf.Endpoint))
log.Infof("alist client init success with endpoint: %s", conf.Endpoint)
} else if conf.Consul.Endpoint != "" {
if conf.ServerName == "" {
return nil, errors.New("alist server name is empty")
}
c := api.DefaultConfig()
c.Address = conf.Consul.Endpoint
client, err := api.NewClient(c)
if err != nil {
return nil, err
}
endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName)
dis := consul.New(client)
opts = append(opts, ggrpc.WithEndpoint(endpoint), ggrpc.WithDiscovery(dis))
log.Infof("alist client init success with consul: %s", conf.Consul.Endpoint)
} else if len(conf.Etcd.Endpoints) > 0 {
if conf.ServerName == "" {
return nil, errors.New("alist server name is empty")
}
endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName)
cli, err := clientv3.New(clientv3.Config{
Endpoints: conf.Etcd.Endpoints,
Username: conf.Etcd.Username,
Password: conf.Etcd.Password,
})
if err != nil {
return nil, err
}
dis := etcd.New(cli)
opts = append(opts, ggrpc.WithEndpoint(endpoint), ggrpc.WithDiscovery(dis))
log.Infof("alist client init success with etcd: %v", conf.Etcd.Endpoints)
} else {
return nil, errors.New("alist client init failed, endpoint is empty")
}
var (
con *grpc.ClientConn
err error
)
if conf.Tls {
var rootCAs *x509.CertPool
rootCAs, err = x509.SystemCertPool()
if err != nil {
return nil, err
}
if conf.CustomCAFile != "" {
b, err := os.ReadFile(conf.CustomCAFile)
if err != nil {
panic(err)
}
rootCAs.AppendCertsFromPEM(b)
}
opts = append(opts, ggrpc.WithTLSConfig(&tls.Config{
RootCAs: rootCAs,
}))
con, err = ggrpc.Dial(
context.Background(),
opts...,
)
} else {
con, err = ggrpc.DialInsecure(
context.Background(),
opts...,
)
}
if err != nil {
return nil, err
}
return newGrpcAlist(alist.NewAlistClient(con)), nil
case "http":
opts := []http.ClientOption{}
opts = append(opts, http.WithMiddleware(middlewares...))
if conf.TimeOut != "" {
timeout, err := time.ParseDuration(conf.TimeOut)
if err != nil {
return nil, err
}
opts = append(opts, http.WithTimeout(timeout))
}
if conf.Tls {
rootCAs, err := x509.SystemCertPool()
if err != nil {
return nil, err
}
if conf.CustomCAFile != "" {
b, err := os.ReadFile(conf.CustomCAFile)
if err != nil {
panic(err)
}
rootCAs.AppendCertsFromPEM(b)
}
opts = append(opts, http.WithTLSConfig(&tls.Config{
RootCAs: rootCAs,
}))
}
if conf.Endpoint != "" {
opts = append(opts, http.WithEndpoint(conf.Endpoint))
log.Infof("alist client init success with endpoint: %s", conf.Endpoint)
} else if conf.Consul.Endpoint != "" {
if conf.ServerName == "" {
return nil, errors.New("alist server name is empty")
}
c := api.DefaultConfig()
c.Address = conf.Consul.Endpoint
client, err := api.NewClient(c)
if err != nil {
return nil, err
}
c.Token = conf.Consul.Token
c.TokenFile = conf.Consul.TokenFile
c.PathPrefix = conf.Consul.PathPrefix
c.Namespace = conf.Consul.Namespace
c.Partition = conf.Consul.Partition
endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName)
dis := consul.New(client)
opts = append(opts, http.WithEndpoint(endpoint), http.WithDiscovery(dis))
log.Infof("alist client init success with consul: %s", conf.Consul.Endpoint)
} else if len(conf.Etcd.Endpoints) > 0 {
if conf.ServerName == "" {
return nil, errors.New("alist server name is empty")
}
endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName)
cli, err := clientv3.New(clientv3.Config{
Endpoints: conf.Etcd.Endpoints,
Username: conf.Etcd.Username,
Password: conf.Etcd.Password,
})
if err != nil {
return nil, err
}
dis := etcd.New(cli)
opts = append(opts, http.WithEndpoint(endpoint), http.WithDiscovery(dis))
log.Infof("alist client init success with etcd: %v", conf.Etcd.Endpoints)
} else {
return nil, errors.New("alist client init failed, endpoint is empty")
}
con, err := http.NewClient(
context.Background(),
opts...,
)
if err != nil {
return nil, err
}
return newHTTPAlist(alist.NewAlistHTTPClient(con)), nil
default:
return nil, errors.New("unknow alist scheme")
func NewAlistGrpcClient(conn *grpc.ClientConn) (AlistInterface, error) {
if conn == nil {
return nil, errors.New("grpc client conn is nil")
}
conn.GetState()
return newGrpcAlist(alist.NewAlistClient(conn)), nil
}
var _ AlistInterface = (*grpcAlist)(nil)
@ -260,7 +45,7 @@ type grpcAlist struct {
client alist.AlistClient
}
func newGrpcAlist(client alist.AlistClient) *grpcAlist {
func newGrpcAlist(client alist.AlistClient) AlistInterface {
return &grpcAlist{
client: client,
}
@ -285,35 +70,3 @@ func (a *grpcAlist) Login(ctx context.Context, req *alist.LoginReq) (*alist.Logi
func (a *grpcAlist) Me(ctx context.Context, req *alist.MeReq) (*alist.MeResp, error) {
return a.client.Me(ctx, req)
}
var _ AlistInterface = (*httpAlist)(nil)
type httpAlist struct {
client alist.AlistHTTPClient
}
func newHTTPAlist(client alist.AlistHTTPClient) *httpAlist {
return &httpAlist{
client: client,
}
}
func (a *httpAlist) FsGet(ctx context.Context, req *alist.FsGetReq) (*alist.FsGetResp, error) {
return a.client.FsGet(ctx, req)
}
func (a *httpAlist) FsList(ctx context.Context, req *alist.FsListReq) (*alist.FsListResp, error) {
return a.client.FsList(ctx, req)
}
func (a *httpAlist) FsOther(ctx context.Context, req *alist.FsOtherReq) (*alist.FsOtherResp, error) {
return a.client.FsOther(ctx, req)
}
func (a *httpAlist) Login(ctx context.Context, req *alist.LoginReq) (*alist.LoginResp, error) {
return a.client.Login(ctx, req)
}
func (a *httpAlist) Me(ctx context.Context, req *alist.MeReq) (*alist.MeResp, error) {
return a.client.Me(ctx, req)
}

314
internal/vendor/bilibili.go vendored
View File

@ -2,256 +2,40 @@ package vendor
import (
"context"
"crypto/tls"
"crypto/x509"
"errors"
"fmt"
"os"
"time"
"github.com/go-kratos/aegis/circuitbreaker"
"github.com/go-kratos/aegis/circuitbreaker/sre"
consul "github.com/go-kratos/kratos/contrib/registry/consul/v2"
"github.com/go-kratos/kratos/contrib/registry/etcd/v2"
"github.com/go-kratos/kratos/v2/middleware"
"github.com/go-kratos/kratos/v2/middleware/auth/jwt"
kcircuitbreaker "github.com/go-kratos/kratos/v2/middleware/circuitbreaker"
"google.golang.org/grpc"
ggrpc "github.com/go-kratos/kratos/v2/transport/grpc"
"github.com/go-kratos/kratos/v2/transport/http"
jwtv4 "github.com/golang-jwt/jwt/v4"
"github.com/hashicorp/consul/api"
log "github.com/sirupsen/logrus"
"github.com/synctv-org/synctv/internal/conf"
"github.com/synctv-org/vendors/api/bilibili"
bilibiliService "github.com/synctv-org/vendors/service/bilibili"
clientv3 "go.etcd.io/etcd/client/v3"
)
type BilibiliInterface = bilibili.BilibiliHTTPServer
func BilibiliClient(name string) BilibiliInterface {
if name != "" {
if cli, ok := bilibiliClients[name]; ok {
return cli
}
func LoadBilibiliClient(name string) BilibiliInterface {
if cli, ok := backends.Load().bilibili[name]; ok {
return cli
}
return bilibiliDefaultClient
}
func BilibiliClients() map[string]BilibiliInterface {
return bilibiliClients
return bilibiliLocalClient
}
var (
bilibiliClients map[string]BilibiliInterface
bilibiliDefaultClient BilibiliInterface
bilibiliLocalClient BilibiliInterface
)
func InitBilibiliVendors(conf map[string]conf.BilibiliConfig) error {
if bilibiliClients == nil {
bilibiliClients = make(map[string]BilibiliInterface, len(conf))
}
for k, vb := range conf {
cli, err := InitBilibili(&vb)
if err != nil {
return err
}
if k == "" {
bilibiliDefaultClient = cli
} else {
bilibiliClients[k] = cli
}
}
if bilibiliDefaultClient == nil {
bilibiliDefaultClient = bilibiliService.NewBilibiliService(nil)
}
return nil
func init() {
bilibiliLocalClient = bilibiliService.NewBilibiliService(nil)
}
func InitBilibili(conf *conf.BilibiliConfig) (BilibiliInterface, error) {
middlewares := []middleware.Middleware{kcircuitbreaker.Client(kcircuitbreaker.WithCircuitBreaker(func() circuitbreaker.CircuitBreaker {
return sre.NewBreaker(
sre.WithRequest(25),
sre.WithWindow(time.Second*15),
)
}))}
func BilibiliLocalClient() BilibiliInterface {
return bilibiliLocalClient
}
if conf.JwtSecret != "" {
key := []byte(conf.JwtSecret)
middlewares = append(middlewares, jwt.Client(func(token *jwtv4.Token) (interface{}, error) {
return key, nil
}, jwt.WithSigningMethod(jwtv4.SigningMethodHS256)))
}
switch conf.Scheme {
case "grpc":
opts := []ggrpc.ClientOption{}
opts = append(opts, ggrpc.WithMiddleware(middlewares...))
if conf.TimeOut != "" {
timeout, err := time.ParseDuration(conf.TimeOut)
if err != nil {
return nil, err
}
opts = append(opts, ggrpc.WithTimeout(timeout))
}
if conf.Endpoint != "" {
opts = append(opts, ggrpc.WithEndpoint(conf.Endpoint))
log.Infof("bilibili client init success with endpoint: %s", conf.Endpoint)
} else if conf.Consul.Endpoint != "" {
if conf.ServerName == "" {
return nil, errors.New("bilibili server name is empty")
}
c := api.DefaultConfig()
c.Address = conf.Consul.Endpoint
client, err := api.NewClient(c)
if err != nil {
return nil, err
}
endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName)
dis := consul.New(client)
opts = append(opts, ggrpc.WithEndpoint(endpoint), ggrpc.WithDiscovery(dis))
log.Infof("bilibili client init success with consul: %s", conf.Consul.Endpoint)
} else if len(conf.Etcd.Endpoints) > 0 {
if conf.ServerName == "" {
return nil, errors.New("bilibili server name is empty")
}
endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName)
cli, err := clientv3.New(clientv3.Config{
Endpoints: conf.Etcd.Endpoints,
Username: conf.Etcd.Username,
Password: conf.Etcd.Password,
})
if err != nil {
return nil, err
}
dis := etcd.New(cli)
opts = append(opts, ggrpc.WithEndpoint(endpoint), ggrpc.WithDiscovery(dis))
log.Infof("bilibili client init success with etcd: %v", conf.Etcd.Endpoints)
} else {
return nil, errors.New("bilibili client init failed, endpoint is empty")
}
var (
con *grpc.ClientConn
err error
)
if conf.Tls {
var rootCAs *x509.CertPool
rootCAs, err = x509.SystemCertPool()
if err != nil {
return nil, err
}
if conf.CustomCAFile != "" {
b, err := os.ReadFile(conf.CustomCAFile)
if err != nil {
panic(err)
}
rootCAs.AppendCertsFromPEM(b)
}
opts = append(opts, ggrpc.WithTLSConfig(&tls.Config{
RootCAs: rootCAs,
}))
con, err = ggrpc.Dial(
context.Background(),
opts...,
)
} else {
con, err = ggrpc.DialInsecure(
context.Background(),
opts...,
)
}
if err != nil {
return nil, err
}
return newGrpcBilibili(bilibili.NewBilibiliClient(con)), nil
case "http":
opts := []http.ClientOption{}
opts = append(opts, http.WithMiddleware(middlewares...))
if conf.TimeOut != "" {
timeout, err := time.ParseDuration(conf.TimeOut)
if err != nil {
return nil, err
}
opts = append(opts, http.WithTimeout(timeout))
}
if conf.Tls {
rootCAs, err := x509.SystemCertPool()
if err != nil {
return nil, err
}
if conf.CustomCAFile != "" {
b, err := os.ReadFile(conf.CustomCAFile)
if err != nil {
panic(err)
}
rootCAs.AppendCertsFromPEM(b)
}
opts = append(opts, http.WithTLSConfig(&tls.Config{
RootCAs: rootCAs,
}))
}
if conf.Endpoint != "" {
opts = append(opts, http.WithEndpoint(conf.Endpoint))
log.Infof("bilibili client init success with endpoint: %s", conf.Endpoint)
} else if conf.Consul.Endpoint != "" {
if conf.ServerName == "" {
return nil, errors.New("bilibili server name is empty")
}
c := api.DefaultConfig()
c.Address = conf.Consul.Endpoint
client, err := api.NewClient(c)
if err != nil {
return nil, err
}
c.Token = conf.Consul.Token
c.TokenFile = conf.Consul.TokenFile
c.PathPrefix = conf.Consul.PathPrefix
c.Namespace = conf.Consul.Namespace
c.Partition = conf.Consul.Partition
endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName)
dis := consul.New(client)
opts = append(opts, http.WithEndpoint(endpoint), http.WithDiscovery(dis))
log.Infof("bilibili client init success with consul: %s", conf.Consul.Endpoint)
} else if len(conf.Etcd.Endpoints) > 0 {
if conf.ServerName == "" {
return nil, errors.New("bilibili server name is empty")
}
endpoint := fmt.Sprintf("discovery:///%s", conf.ServerName)
cli, err := clientv3.New(clientv3.Config{
Endpoints: conf.Etcd.Endpoints,
Username: conf.Etcd.Username,
Password: conf.Etcd.Password,
})
if err != nil {
return nil, err
}
dis := etcd.New(cli)
opts = append(opts, http.WithEndpoint(endpoint), http.WithDiscovery(dis))
log.Infof("bilibili client init success with etcd: %v", conf.Etcd.Endpoints)
} else {
return nil, errors.New("bilibili client init failed, endpoint is empty")
}
con, err := http.NewClient(
context.Background(),
opts...,
)
if err != nil {
return nil, err
}
return newHTTPBilibili(bilibili.NewBilibiliHTTPClient(con)), nil
default:
return nil, errors.New("unknow bilibili scheme")
func NewBilibiliGrpcClient(conn *grpc.ClientConn) (BilibiliInterface, error) {
if conn == nil {
return nil, errors.New("grpc client conn is nil")
}
return newGrpcBilibili(bilibili.NewBilibiliClient(conn)), nil
}
var _ BilibiliInterface = (*grpcBilibili)(nil)
@ -260,7 +44,7 @@ type grpcBilibili struct {
client bilibili.BilibiliClient
}
func newGrpcBilibili(client bilibili.BilibiliClient) *grpcBilibili {
func newGrpcBilibili(client bilibili.BilibiliClient) BilibiliInterface {
return &grpcBilibili{
client: client,
}
@ -321,71 +105,3 @@ func (g *grpcBilibili) UserInfo(ctx context.Context, in *bilibili.UserInfoReq) (
func (g *grpcBilibili) Match(ctx context.Context, in *bilibili.MatchReq) (*bilibili.MatchResp, error) {
return g.client.Match(ctx, in)
}
var _ BilibiliInterface = (*httpBilibili)(nil)
type httpBilibili struct {
client bilibili.BilibiliHTTPClient
}
func newHTTPBilibili(client bilibili.BilibiliHTTPClient) *httpBilibili {
return &httpBilibili{
client: client,
}
}
func (h *httpBilibili) NewQRCode(ctx context.Context, in *bilibili.Empty) (*bilibili.NewQRCodeResp, error) {
return h.client.NewQRCode(ctx, in)
}
func (h *httpBilibili) LoginWithQRCode(ctx context.Context, in *bilibili.LoginWithQRCodeReq) (*bilibili.LoginWithQRCodeResp, error) {
return h.client.LoginWithQRCode(ctx, in)
}
func (h *httpBilibili) NewCaptcha(ctx context.Context, in *bilibili.Empty) (*bilibili.NewCaptchaResp, error) {
return h.client.NewCaptcha(ctx, in)
}
func (h *httpBilibili) NewSMS(ctx context.Context, in *bilibili.NewSMSReq) (*bilibili.NewSMSResp, error) {
return h.client.NewSMS(ctx, in)
}
func (h *httpBilibili) LoginWithSMS(ctx context.Context, in *bilibili.LoginWithSMSReq) (*bilibili.LoginWithSMSResp, error) {
return h.client.LoginWithSMS(ctx, in)
}
func (h *httpBilibili) ParseVideoPage(ctx context.Context, in *bilibili.ParseVideoPageReq) (*bilibili.VideoPageInfo, error) {
return h.client.ParseVideoPage(ctx, in)
}
func (h *httpBilibili) GetVideoURL(ctx context.Context, in *bilibili.GetVideoURLReq) (*bilibili.VideoURL, error) {
return h.client.GetVideoURL(ctx, in)
}
func (h *httpBilibili) GetDashVideoURL(ctx context.Context, in *bilibili.GetDashVideoURLReq) (*bilibili.GetDashVideoURLResp, error) {
return h.client.GetDashVideoURL(ctx, in)
}
func (h *httpBilibili) GetSubtitles(ctx context.Context, in *bilibili.GetSubtitlesReq) (*bilibili.GetSubtitlesResp, error) {
return h.client.GetSubtitles(ctx, in)
}
func (h *httpBilibili) ParsePGCPage(ctx context.Context, in *bilibili.ParsePGCPageReq) (*bilibili.VideoPageInfo, error) {
return h.client.ParsePGCPage(ctx, in)
}
func (h *httpBilibili) GetPGCURL(ctx context.Context, in *bilibili.GetPGCURLReq) (*bilibili.VideoURL, error) {
return h.client.GetPGCURL(ctx, in)
}
func (h *httpBilibili) GetDashPGCURL(ctx context.Context, in *bilibili.GetDashPGCURLReq) (*bilibili.GetDashPGCURLResp, error) {
return h.client.GetDashPGCURL(ctx, in)
}
func (h *httpBilibili) UserInfo(ctx context.Context, in *bilibili.UserInfoReq) (*bilibili.UserInfoResp, error) {
return h.client.UserInfo(ctx, in)
}
func (h *httpBilibili) Match(ctx context.Context, in *bilibili.MatchReq) (*bilibili.MatchResp, error) {
return h.client.Match(ctx, in)
}

76
internal/vendor/emby.go vendored Normal file
View File

@ -0,0 +1,76 @@
package vendor
import (
"context"
"errors"
"google.golang.org/grpc"
"github.com/synctv-org/vendors/api/emby"
embyService "github.com/synctv-org/vendors/service/emby"
)
type EmbyInterface = emby.EmbyHTTPServer
func LoadEmbyClient(name string) EmbyInterface {
if cli, ok := backends.Load().emby[name]; ok && cli != nil {
return cli
}
return embyLocalClient
}
var (
embyLocalClient EmbyInterface
)
func init() {
embyLocalClient = embyService.NewEmbyService(nil)
}
func EmbyLocalClient() EmbyInterface {
return embyLocalClient
}
func NewEmbyGrpcClient(conn *grpc.ClientConn) (EmbyInterface, error) {
if conn == nil {
return nil, errors.New("grpc client conn is nil")
}
conn.GetState()
return newGrpcEmby(emby.NewEmbyClient(conn)), nil
}
var _ EmbyInterface = (*grpcEmby)(nil)
type grpcEmby struct {
client emby.EmbyClient
}
func newGrpcEmby(client emby.EmbyClient) EmbyInterface {
return &grpcEmby{
client: client,
}
}
func (e *grpcEmby) FsList(ctx context.Context, req *emby.FsListReq) (*emby.FsListResp, error) {
return e.client.FsList(ctx, req)
}
func (e *grpcEmby) GetItem(ctx context.Context, req *emby.GetItemReq) (*emby.Item, error) {
return e.client.GetItem(ctx, req)
}
func (e *grpcEmby) GetItems(ctx context.Context, req *emby.GetItemsReq) (*emby.GetItemsResp, error) {
return e.client.GetItems(ctx, req)
}
func (e *grpcEmby) GetSystemInfo(ctx context.Context, req *emby.Empty) (*emby.SystemInfoResp, error) {
return e.client.GetSystemInfo(ctx, req)
}
func (e *grpcEmby) Login(ctx context.Context, req *emby.LoginReq) (*emby.LoginResp, error) {
return e.client.Login(ctx, req)
}
func (e *grpcEmby) Me(ctx context.Context, req *emby.MeReq) (*emby.MeResp, error) {
return e.client.Me(ctx, req)
}

326
internal/vendor/vendor.go vendored
View File

@ -1,21 +1,329 @@
package vendor
import (
"context"
"crypto/tls"
"crypto/x509"
"errors"
"fmt"
"os"
"sync/atomic"
"time"
"github.com/go-kratos/aegis/circuitbreaker"
"github.com/go-kratos/aegis/circuitbreaker/sre"
consul "github.com/go-kratos/kratos/contrib/registry/consul/v2"
"github.com/go-kratos/kratos/contrib/registry/etcd/v2"
klog "github.com/go-kratos/kratos/v2/log"
"github.com/go-kratos/kratos/v2/middleware"
"github.com/go-kratos/kratos/v2/middleware/auth/jwt"
kcircuitbreaker "github.com/go-kratos/kratos/v2/middleware/circuitbreaker"
"github.com/go-kratos/kratos/v2/selector"
"github.com/go-kratos/kratos/v2/selector/wrr"
ggrpc "github.com/go-kratos/kratos/v2/transport/grpc"
"github.com/go-kratos/kratos/v2/transport/http"
jwtv4 "github.com/golang-jwt/jwt/v4"
"github.com/hashicorp/consul/api"
log "github.com/sirupsen/logrus"
"github.com/synctv-org/synctv/internal/conf"
"github.com/synctv-org/synctv/internal/model"
clientv3 "go.etcd.io/etcd/client/v3"
"google.golang.org/grpc"
)
func Init(conf *conf.VendorConfig) error {
func init() {
klog.SetLogger(klog.NewStdLogger(log.StandardLogger().Writer()))
selector.SetGlobalSelector(wrr.NewBuilder())
if err := InitBilibiliVendors(conf.Bilibili); err != nil {
return err
}
if err := InitAlistVendors(conf.Alist); err != nil {
return err
}
return nil
}
var backends atomic.Pointer[Backends]
type BackendConnInfo struct {
Conn *grpc.ClientConn
Info *model.VendorBackend
}
type Backends struct {
conns map[string]*BackendConnInfo
bilibili map[string]BilibiliInterface
alist map[string]AlistInterface
emby map[string]EmbyInterface
}
func (b *Backends) Conns() map[string]*BackendConnInfo {
return b.conns
}
func (b *Backends) BilibiliClients() map[string]BilibiliInterface {
return b.bilibili
}
func (b *Backends) AlistClients() map[string]AlistInterface {
return b.alist
}
func (b *Backends) EmbyClients() map[string]EmbyInterface {
return b.emby
}
func NewBackends(ctx context.Context, conf []*model.VendorBackend) (*Backends, error) {
newConns := make(map[string]*BackendConnInfo, len(conf))
backends := &Backends{
conns: newConns,
bilibili: make(map[string]BilibiliInterface),
alist: make(map[string]AlistInterface),
emby: make(map[string]EmbyInterface),
}
for _, vb := range conf {
cc, err := NewGrpcClientConn(ctx, &vb.Backend)
if err != nil {
return nil, err
}
if _, ok := newConns[vb.Backend.Endpoint]; ok {
return nil, fmt.Errorf("duplicate endpoint: %s", vb.Backend.Endpoint)
}
newConns[vb.Backend.Endpoint] = &BackendConnInfo{
Conn: cc,
Info: vb,
}
if vb.UsedBy.Bilibili {
if _, ok := backends.bilibili[vb.UsedBy.BilibiliBackendName]; ok {
return nil, fmt.Errorf("duplicate bilibili backend name: %s", vb.UsedBy.BilibiliBackendName)
}
cli, err := NewBilibiliGrpcClient(cc)
if err != nil {
return nil, err
}
backends.bilibili[vb.UsedBy.BilibiliBackendName] = cli
}
if vb.UsedBy.Alist {
if _, ok := backends.alist[vb.UsedBy.AlistBackendName]; ok {
return nil, fmt.Errorf("duplicate alist backend name: %s", vb.UsedBy.AlistBackendName)
}
cli, err := NewAlistGrpcClient(cc)
if err != nil {
return nil, err
}
backends.alist[vb.UsedBy.AlistBackendName] = cli
}
if vb.UsedBy.Emby {
if _, ok := backends.emby[vb.UsedBy.EmbyBackendName]; ok {
return nil, fmt.Errorf("duplicate emby backend name: %s", vb.UsedBy.EmbyBackendName)
}
cli, err := NewEmbyGrpcClient(cc)
if err != nil {
return nil, err
}
backends.emby[vb.UsedBy.EmbyBackendName] = cli
}
}
return backends, nil
}
func LoadBackends() *Backends {
return backends.Load()
}
func StoreBackends(b *Backends) {
old := backends.Swap(b)
if old == nil {
return
}
for k, conn := range old.conns {
conn.Conn.Close()
delete(old.conns, k)
}
}
func NewGrpcClientConn(ctx context.Context, conf *model.Backend) (*grpc.ClientConn, error) {
if conf.Endpoint == "" {
return nil, errors.New("new grpc client failed, endpoint is empty")
}
middlewares := []middleware.Middleware{kcircuitbreaker.Client(kcircuitbreaker.WithCircuitBreaker(func() circuitbreaker.CircuitBreaker {
return sre.NewBreaker(
sre.WithRequest(25),
sre.WithWindow(time.Second*15),
)
}))}
if conf.JwtSecret != "" {
key := []byte(conf.JwtSecret)
middlewares = append(middlewares, jwt.Client(func(token *jwtv4.Token) (interface{}, error) {
return key, nil
}, jwt.WithSigningMethod(jwtv4.SigningMethodHS256)))
}
opts := []ggrpc.ClientOption{
ggrpc.WithMiddleware(middlewares...),
// ggrpc.WithOptions(grpc.WithBlock()),
}
if conf.TimeOut != "" {
timeout, err := time.ParseDuration(conf.TimeOut)
if err != nil {
return nil, err
}
opts = append(opts, ggrpc.WithTimeout(timeout))
}
if conf.Consul.ServerName != "" {
c := api.DefaultConfig()
c.Address = conf.Endpoint
c.Token = conf.Consul.Token
c.TokenFile = conf.Consul.TokenFile
c.PathPrefix = conf.Consul.PathPrefix
c.Namespace = conf.Consul.Namespace
c.Partition = conf.Consul.Partition
client, err := api.NewClient(c)
if err != nil {
return nil, err
}
endpoint := fmt.Sprintf("discovery:///%s", conf.Consul.ServerName)
dis := consul.New(client)
opts = append(opts, ggrpc.WithEndpoint(endpoint), ggrpc.WithDiscovery(dis))
log.Infof("new grpc client with consul: %s", conf.Endpoint)
} else if conf.Etcd.ServerName != "" {
endpoint := fmt.Sprintf("discovery:///%s", conf.Etcd.ServerName)
cli, err := clientv3.New(clientv3.Config{
Endpoints: []string{conf.Endpoint},
Username: conf.Etcd.Username,
Password: conf.Etcd.Password,
})
if err != nil {
return nil, err
}
dis := etcd.New(cli)
opts = append(opts, ggrpc.WithEndpoint(endpoint), ggrpc.WithDiscovery(dis))
log.Infof("new grpc client with etcd: %v", conf.Endpoint)
} else {
opts = append(opts, ggrpc.WithEndpoint(conf.Endpoint))
log.Infof("new grpc client with endpoint: %s", conf.Endpoint)
}
var (
con *grpc.ClientConn
err error
)
if conf.Tls {
var rootCAs *x509.CertPool
rootCAs, err = x509.SystemCertPool()
if err != nil {
return nil, err
}
if conf.CustomCAFile != "" {
b, err := os.ReadFile(conf.CustomCAFile)
if err != nil {
return nil, err
}
rootCAs.AppendCertsFromPEM(b)
}
opts = append(opts, ggrpc.WithTLSConfig(&tls.Config{
RootCAs: rootCAs,
}))
con, err = ggrpc.Dial(
ctx,
opts...,
)
} else {
con, err = ggrpc.DialInsecure(
ctx,
opts...,
)
}
if err != nil {
return nil, err
}
return con, nil
}
func NewHttpClientConn(ctx context.Context, conf *model.Backend) (*http.Client, error) {
if conf.Endpoint == "" {
return nil, errors.New("new http client failed, endpoint is empty")
}
middlewares := []middleware.Middleware{kcircuitbreaker.Client(kcircuitbreaker.WithCircuitBreaker(func() circuitbreaker.CircuitBreaker {
return sre.NewBreaker(
sre.WithRequest(25),
sre.WithWindow(time.Second*15),
)
}))}
if conf.JwtSecret != "" {
key := []byte(conf.JwtSecret)
middlewares = append(middlewares, jwt.Client(func(token *jwtv4.Token) (interface{}, error) {
return key, nil
}, jwt.WithSigningMethod(jwtv4.SigningMethodHS256)))
}
opts := []http.ClientOption{
http.WithMiddleware(middlewares...),
}
if conf.TimeOut != "" {
timeout, err := time.ParseDuration(conf.TimeOut)
if err != nil {
return nil, err
}
opts = append(opts, http.WithTimeout(timeout))
}
if conf.Tls {
rootCAs, err := x509.SystemCertPool()
if err != nil {
return nil, err
}
if conf.CustomCAFile != "" {
b, err := os.ReadFile(conf.CustomCAFile)
if err != nil {
return nil, err
}
rootCAs.AppendCertsFromPEM(b)
}
opts = append(opts, http.WithTLSConfig(&tls.Config{
RootCAs: rootCAs,
}))
}
if conf.Consul.ServerName != "" {
c := api.DefaultConfig()
c.Address = conf.Endpoint
c.Token = conf.Consul.Token
c.TokenFile = conf.Consul.TokenFile
c.PathPrefix = conf.Consul.PathPrefix
c.Namespace = conf.Consul.Namespace
c.Partition = conf.Consul.Partition
client, err := api.NewClient(c)
if err != nil {
return nil, err
}
endpoint := fmt.Sprintf("discovery:///%s", conf.Consul.ServerName)
dis := consul.New(client)
opts = append(opts, http.WithEndpoint(endpoint), http.WithDiscovery(dis))
log.Infof("new http client with consul: %s", conf.Endpoint)
} else if conf.Etcd.ServerName != "" {
endpoint := fmt.Sprintf("discovery:///%s", conf.Etcd.ServerName)
cli, err := clientv3.New(clientv3.Config{
Endpoints: []string{conf.Endpoint},
Username: conf.Etcd.Username,
Password: conf.Etcd.Password,
})
if err != nil {
return nil, err
}
dis := etcd.New(cli)
opts = append(opts, http.WithEndpoint(endpoint), http.WithDiscovery(dis))
log.Infof("new http client with etcd: %v", conf.Endpoint)
} else {
opts = append(opts, http.WithEndpoint(conf.Endpoint))
log.Infof("new http client with endpoint: %s", conf.Endpoint)
}
con, err := http.NewClient(
ctx,
opts...,
)
if err != nil {
return nil, err
}
return con, nil
}

121
server/handlers/admin.go
View File

@ -10,6 +10,7 @@ import (
dbModel "github.com/synctv-org/synctv/internal/model"
"github.com/synctv-org/synctv/internal/op"
"github.com/synctv-org/synctv/internal/settings"
"github.com/synctv-org/synctv/internal/vendor"
"github.com/synctv-org/synctv/server/model"
"gorm.io/gorm"
)
@ -699,3 +700,123 @@ func AdminRoomPassword(ctx *gin.Context) {
ctx.Status(http.StatusNoContent)
}
func AdminGetVendorBackends(ctx *gin.Context) {
// user := ctx.MustGet("user").(*op.User)
conns := vendor.LoadBackends().Conns()
resp := make([]*model.GetVendorBackendResp, 0, len(conns))
for _, conn := range conns {
resp = append(resp, &model.GetVendorBackendResp{
Status: conn.Conn.GetState(),
Info: conn.Info,
})
}
ctx.JSON(http.StatusOK, model.NewApiDataResp(resp))
}
func AdminAddVendorBackends(ctx *gin.Context) {
// user := ctx.MustGet("user").(*op.User)
var req model.AddVendorBackendReq
if err := model.Decode(ctx, &req); err != nil {
ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
return
}
vb, err := db.GetAllVendorBackend()
if err != nil {
ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err))
return
}
vb = append(vb, (*dbModel.VendorBackend)(&req))
backends, err := vendor.NewBackends(ctx, vb)
if err != nil {
ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
return
}
err = db.CreateVendorBackend((*dbModel.VendorBackend)(&req))
if err != nil {
ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
return
}
vendor.StoreBackends(backends)
ctx.Status(http.StatusNoContent)
}
func AdminDeleteVendorBackends(ctx *gin.Context) {
// user := ctx.MustGet("user").(*op.User)
var req model.DeleteVendorBackendsReq
if err := model.Decode(ctx, &req); err != nil {
ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
return
}
err := db.DeleteVendorBackends(req.Endpoints)
if err != nil {
ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
return
}
vb, err := db.GetAllVendorBackend()
if err != nil {
ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err))
return
}
backends, err := vendor.NewBackends(ctx, vb)
if err != nil {
ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
return
}
vendor.StoreBackends(backends)
ctx.Status(http.StatusNoContent)
}
func AdminUpdateVendorBackends(ctx *gin.Context) {
// user := ctx.MustGet("user").(*op.User)
var req model.AddVendorBackendReq
if err := model.Decode(ctx, &req); err != nil {
ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
return
}
vb, err := db.GetAllVendorBackend()
if err != nil {
ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err))
return
}
for i, vb2 := range vb {
if vb2.Backend.Endpoint == req.Backend.Endpoint {
vb[i] = (*dbModel.VendorBackend)(&req)
break
}
}
backends, err := vendor.NewBackends(ctx, vb)
if err != nil {
ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
return
}
err = db.SaveVendorBackend((*dbModel.VendorBackend)(&req))
if err != nil {
ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
return
}
vendor.StoreBackends(backends)
ctx.Status(http.StatusNoContent)
}

8
server/handlers/init.go
View File

@ -38,6 +38,14 @@ func Init(e *gin.Engine) {
admin.POST("/settings", EditAdminSettings)
admin.GET("/vendors", AdminGetVendorBackends)
admin.POST("/vendors", AdminAddVendorBackends)
admin.PUT("/vendors", AdminUpdateVendorBackends)
admin.DELETE("/vendors", AdminDeleteVendorBackends)
{
user := admin.Group("/user")

9
server/handlers/movie.go
View File

@ -732,11 +732,6 @@ func proxyVendorMovie(ctx *gin.Context, movie *op.Movie) {
}
func parse2VendorMovie(ctx context.Context, user *op.User, room *op.Room, movie *dbModel.Movie) (err error) {
userID := user.ID
if movie.Base.VendorInfo.Shared {
userID = movie.CreatorID
}
switch movie.Base.VendorInfo.Vendor {
case dbModel.VendorBilibili:
opM, err := room.GetMovieByID(movie.ID)
@ -748,6 +743,10 @@ func parse2VendorMovie(ctx context.Context, user *op.User, room *op.Room, movie
if err != nil {
return err
}
userID := user.ID
if movie.Base.VendorInfo.Bilibili.Shared {
userID = movie.CreatorID
}
s, err := bmc.NoSharedMovie.LoadOrStore(ctx, userID)
if err != nil {
return err

2
server/handlers/vendors/vendorAlist/login.go vendored
View File

@ -42,7 +42,7 @@ func Login(ctx *gin.Context) {
return
}
cli := vendor.AlistClient("")
cli := vendor.LoadAlistClient("")
if req.Username == "" {
_, err := cli.Me(ctx, &alist.MeReq{

2
server/handlers/vendors/vendorAlist/me.go vendored
View File

@ -17,7 +17,7 @@ type AlistMeResp = model.VendorMeResp[*alist.MeResp]
func Me(ctx *gin.Context) {
user := ctx.MustGet("user").(*op.User)
cli := vendor.AlistClient(ctx.Query("backend"))
cli := vendor.LoadAlistClient(ctx.Query("backend"))
aucd, err := user.AlistCache().Get(ctx, ctx.Query("backend"))
if err != nil {

2
server/handlers/vendors/vendorAlist/parse.go vendored
View File

@ -39,7 +39,7 @@ func List(ctx *gin.Context) {
return
}
var cli = vendor.AlistClient(ctx.Query("backend"))
var cli = vendor.LoadAlistClient(ctx.Query("backend"))
aucd, err := user.AlistCache().Get(ctx, ctx.Query("backend"))
if err != nil {
if errors.Is(err, db.ErrNotFound("vendor")) {

2
server/handlers/vendors/vendorBilibili/bilibili.go vendored
View File

@ -39,7 +39,7 @@ func Parse(ctx *gin.Context) {
return
}
var cli = vendor.BilibiliClient(ctx.Query("backend"))
var cli = vendor.LoadBilibiliClient(ctx.Query("backend"))
resp, err := cli.Match(ctx, &bilibili.MatchReq{
Url: req.URL,

10
server/handlers/vendors/vendorBilibili/login.go vendored
View File

@ -15,7 +15,7 @@ import (
)
func NewQRCode(ctx *gin.Context) {
r, err := vendor.BilibiliClient("").NewQRCode(ctx, &bilibili.Empty{})
r, err := vendor.LoadBilibiliClient("").NewQRCode(ctx, &bilibili.Empty{})
if err != nil {
ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err))
return
@ -47,7 +47,7 @@ func LoginWithQR(ctx *gin.Context) {
return
}
resp, err := vendor.BilibiliClient("").LoginWithQRCode(ctx, &bilibili.LoginWithQRCodeReq{
resp, err := vendor.LoadBilibiliClient("").LoginWithQRCode(ctx, &bilibili.LoginWithQRCodeReq{
Key: req.Key,
})
if err != nil {
@ -89,7 +89,7 @@ func LoginWithQR(ctx *gin.Context) {
}
func NewCaptcha(ctx *gin.Context) {
r, err := vendor.BilibiliClient("").NewCaptcha(ctx, &bilibili.Empty{})
r, err := vendor.LoadBilibiliClient("").NewCaptcha(ctx, &bilibili.Empty{})
if err != nil {
ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err))
return
@ -131,7 +131,7 @@ func NewSMS(ctx *gin.Context) {
return
}
r, err := vendor.BilibiliClient("").NewSMS(ctx, &bilibili.NewSMSReq{
r, err := vendor.LoadBilibiliClient("").NewSMS(ctx, &bilibili.NewSMSReq{
Phone: req.Telephone,
Token: req.Token,
Challenge: req.Challenge,
@ -176,7 +176,7 @@ func LoginWithSMS(ctx *gin.Context) {
return
}
c, err := vendor.BilibiliClient("").LoginWithSMS(ctx, &bilibili.LoginWithSMSReq{
c, err := vendor.LoadBilibiliClient("").LoginWithSMS(ctx, &bilibili.LoginWithSMSReq{
Phone: req.Telephone,
CaptchaKey: req.CaptchaKey,
Code: req.Code,

2
server/handlers/vendors/vendorBilibili/me.go vendored
View File

@ -33,7 +33,7 @@ func Me(ctx *gin.Context) {
}))
return
}
resp, err := vendor.BilibiliClient("").UserInfo(ctx, &bilibili.UserInfoReq{
resp, err := vendor.LoadBilibiliClient("").UserInfo(ctx, &bilibili.UserInfoReq{
Cookies: v.Cookies,
})
if err != nil {

8
server/handlers/vendors/vendors.go vendored
View File

@ -14,11 +14,13 @@ func Backends(ctx *gin.Context) {
var backends []string
switch ctx.Param("vendor") {
case dbModel.VendorBilibili:
backends = maps.Keys(vendor.BilibiliClients())
backends = maps.Keys(vendor.LoadBackends().BilibiliClients())
case dbModel.VendorAlist:
backends = maps.Keys(vendor.AlistClients())
backends = maps.Keys(vendor.LoadBackends().AlistClients())
case dbModel.VendorEmby:
backends = maps.Keys(vendor.LoadBackends().EmbyClients())
default:
ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorStringResp("invalid vendor"))
ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorStringResp("invalid vendor name"))
return
}
ctx.JSON(http.StatusOK, model.NewApiDataResp(backends))

35
server/model/admin.go
View File

@ -5,7 +5,9 @@ import (
"github.com/gin-gonic/gin"
json "github.com/json-iterator/go"
"github.com/synctv-org/synctv/internal/model"
dbModel "github.com/synctv-org/synctv/internal/model"
"google.golang.org/grpc/connectivity"
)
var (
@ -128,3 +130,36 @@ func (aur *AdminRoomPasswordReq) Validate() error {
func (aur *AdminRoomPasswordReq) Decode(ctx *gin.Context) error {
return json.NewDecoder(ctx.Request.Body).Decode(aur)
}
type GetVendorBackendResp struct {
Info *dbModel.VendorBackend `json:"info"`
Status connectivity.State `json:"status"`
}
type AddVendorBackendReq model.VendorBackend
func (avbr *AddVendorBackendReq) Validate() error {
if avbr.Backend.Endpoint == "" {
return errors.New("endpoint is empty")
}
return nil
}
func (avbr *AddVendorBackendReq) Decode(ctx *gin.Context) error {
return json.NewDecoder(ctx.Request.Body).Decode(avbr)
}
type DeleteVendorBackendsReq struct {
Endpoints []string `json:"endpoints"`
}
func (dvbr *DeleteVendorBackendsReq) Validate() error {
if len(dvbr.Endpoints) == 0 {
return errors.New("endpoints is empty")
}
return nil
}
func (dvbr *DeleteVendorBackendsReq) Decode(ctx *gin.Context) error {
return json.NewDecoder(ctx.Request.Body).Decode(dvbr)
}