mirror of https://github.com/synctv-org/synctv
Feat: alist vendor
zijiren233
3 years ago
parent
d0eea86cd3
commit
b4cb16de4e
@ -0,0 +1,311 @@
|
||||
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
|
||||
}
|
||||
}
|
||||
return alistDefaultClient
|
||||
}
|
||||
|
||||
func AlistClients() map[string]AlistInterface {
|
||||
return alistClients
|
||||
}
|
||||
|
||||
var (
|
||||
alistClients map[string]AlistInterface
|
||||
alistDefaultClient 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 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),
|
||||
)
|
||||
}))}
|
||||
|
||||
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")
|
||||
}
|
||||
}
|
||||
|
||||
var _ AlistInterface = (*grpcAlist)(nil)
|
||||
|
||||
type grpcAlist struct {
|
||||
client alist.AlistClient
|
||||
}
|
||||
|
||||
func newGrpcAlist(client alist.AlistClient) *grpcAlist {
|
||||
return &grpcAlist{
|
||||
client: client,
|
||||
}
|
||||
}
|
||||
|
||||
func (a *grpcAlist) FsGet(ctx context.Context, req *alist.FsGetReq) (*alist.FsGetResp, error) {
|
||||
return a.client.FsGet(ctx, req)
|
||||
}
|
||||
|
||||
func (a *grpcAlist) FsList(ctx context.Context, req *alist.FsListReq) (*alist.FsListResp, error) {
|
||||
return a.client.FsList(ctx, req)
|
||||
}
|
||||
|
||||
func (a *grpcAlist) FsOther(ctx context.Context, req *alist.FsOtherReq) (*alist.FsOtherResp, error) {
|
||||
return a.client.FsOther(ctx, req)
|
||||
}
|
||||
|
||||
func (a *grpcAlist) Login(ctx context.Context, req *alist.LoginReq) (*alist.LoginResp, error) {
|
||||
return a.client.Login(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)
|
||||
}
|
||||
@ -0,0 +1,68 @@
|
||||
package vendorAlist
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net/http"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
json "github.com/json-iterator/go"
|
||||
"github.com/synctv-org/synctv/internal/db"
|
||||
dbModel "github.com/synctv-org/synctv/internal/model"
|
||||
"github.com/synctv-org/synctv/internal/op"
|
||||
"github.com/synctv-org/synctv/internal/vendor"
|
||||
"github.com/synctv-org/synctv/server/model"
|
||||
"github.com/synctv-org/vendors/api/alist"
|
||||
)
|
||||
|
||||
type LoginReq struct {
|
||||
Host string `json:"host"`
|
||||
Username string `json:"username"`
|
||||
Password string `json:"password"`
|
||||
}
|
||||
|
||||
func (r *LoginReq) Validate() error {
|
||||
if r.Host == "" {
|
||||
return errors.New("host is required")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *LoginReq) Decode(ctx *gin.Context) error {
|
||||
return json.NewDecoder(ctx.Request.Body).Decode(r)
|
||||
}
|
||||
|
||||
func Login(ctx *gin.Context) {
|
||||
user := ctx.MustGet("user").(*op.User)
|
||||
|
||||
req := LoginReq{}
|
||||
if err := model.Decode(ctx, &req); err != nil {
|
||||
ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
|
||||
return
|
||||
}
|
||||
|
||||
if req.Username == "" {
|
||||
_, err := db.AssignFirstOrCreateVendorByUserIDAndVendor(user.ID, dbModel.StreamingVendorAlist, db.WithHost(req.Host))
|
||||
if err != nil {
|
||||
ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err))
|
||||
return
|
||||
}
|
||||
} else {
|
||||
resp, err := vendor.AlistClient("").Login(ctx, &alist.LoginReq{
|
||||
Host: req.Host,
|
||||
Username: req.Username,
|
||||
Password: req.Password,
|
||||
})
|
||||
if err != nil {
|
||||
ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err))
|
||||
return
|
||||
}
|
||||
|
||||
_, err = db.AssignFirstOrCreateVendorByUserIDAndVendor(user.ID, dbModel.StreamingVendorAlist, db.WithAuthorization(resp.Token), db.WithHost(req.Host))
|
||||
if err != nil {
|
||||
ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err))
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
ctx.Status(http.StatusNoContent)
|
||||
}
|
||||
@ -0,0 +1,67 @@
|
||||
package vendorAlist
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
json "github.com/json-iterator/go"
|
||||
"github.com/synctv-org/synctv/internal/db"
|
||||
dbModel "github.com/synctv-org/synctv/internal/model"
|
||||
"github.com/synctv-org/synctv/internal/op"
|
||||
"github.com/synctv-org/synctv/internal/vendor"
|
||||
"github.com/synctv-org/synctv/server/model"
|
||||
"github.com/synctv-org/vendors/api/alist"
|
||||
)
|
||||
|
||||
type ListReq struct {
|
||||
Path string `json:"path"`
|
||||
Password string `json:"password"`
|
||||
}
|
||||
|
||||
func (r *ListReq) Validate() error {
|
||||
if r.Path == "" {
|
||||
r.Password = "/"
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *ListReq) Decode(ctx *gin.Context) error {
|
||||
return json.NewDecoder(ctx.Request.Body).Decode(r)
|
||||
}
|
||||
|
||||
func List(ctx *gin.Context) {
|
||||
user := ctx.MustGet("user").(*op.User)
|
||||
|
||||
req := ListReq{}
|
||||
if err := model.Decode(ctx, &req); err != nil {
|
||||
ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorResp(err))
|
||||
return
|
||||
}
|
||||
|
||||
v, err := db.FirstOrCreateVendorByUserIDAndVendor(user.ID, dbModel.StreamingVendorAlist)
|
||||
if err != nil {
|
||||
ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err))
|
||||
return
|
||||
}
|
||||
|
||||
if v.Authorization == "" {
|
||||
ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorStringResp("unauthorized"))
|
||||
return
|
||||
}
|
||||
|
||||
var cli = vendor.AlistClient(ctx.Query("backend"))
|
||||
|
||||
resp, err := cli.FsList(ctx, &alist.FsListReq{
|
||||
Token: v.Authorization,
|
||||
Password: req.Password,
|
||||
Path: req.Path,
|
||||
Host: v.Host,
|
||||
Refresh: true,
|
||||
})
|
||||
if err != nil {
|
||||
ctx.AbortWithStatusJSON(http.StatusInternalServerError, model.NewApiErrorResp(err))
|
||||
return
|
||||
}
|
||||
|
||||
ctx.JSON(http.StatusOK, model.NewApiDataResp(resp))
|
||||
}
|
||||
@ -1,4 +1,4 @@
|
||||
package Vbilibili
|
||||
package vendorBilibili
|
||||
|
||||
import (
|
||||
"errors"
|
||||
@ -1,4 +1,4 @@
|
||||
package Vbilibili
|
||||
package vendorBilibili
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
@ -0,0 +1,25 @@
|
||||
package vendors
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
dbModel "github.com/synctv-org/synctv/internal/model"
|
||||
"github.com/synctv-org/synctv/internal/vendor"
|
||||
"github.com/synctv-org/synctv/server/model"
|
||||
"golang.org/x/exp/maps"
|
||||
)
|
||||
|
||||
func Backends(ctx *gin.Context) {
|
||||
var backends []string
|
||||
switch dbModel.StreamingVendor(ctx.Param("vendor")) {
|
||||
case dbModel.StreamingVendorBilibili:
|
||||
backends = maps.Keys(vendor.BilibiliClients())
|
||||
case dbModel.StreamingVendorAlist:
|
||||
backends = maps.Keys(vendor.AlistClients())
|
||||
default:
|
||||
ctx.AbortWithStatusJSON(http.StatusBadRequest, model.NewApiErrorStringResp("invalid vendor"))
|
||||
return
|
||||
}
|
||||
ctx.JSON(http.StatusOK, model.NewApiDataResp(backends))
|
||||
}
|
||||
Loading…
Reference in New Issue