mirror of
https://github.com/Ernous/TorrServerJellyfin.git
synced 2025-12-20 22:16:09 +05:00
update
This commit is contained in:
YouROK
244
server/tgbot/upload/manager.go
Normal file
244
server/tgbot/upload/manager.go
Normal file
@@ -0,0 +1,244 @@
|
||||
package upload
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
tele "gopkg.in/telebot.v4"
|
||||
"log"
|
||||
"math"
|
||||
"path/filepath"
|
||||
"server/torr"
|
||||
"server/torr/state"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Worker struct {
|
||||
id int
|
||||
c tele.Context
|
||||
msg *tele.Message
|
||||
torrentHash string
|
||||
isCancelled bool
|
||||
from int
|
||||
to int
|
||||
ti *state.TorrentStatus
|
||||
}
|
||||
|
||||
type Manager struct {
|
||||
queue []*Worker
|
||||
working map[int]*Worker
|
||||
ids int
|
||||
wrkSync sync.Mutex
|
||||
queueLock sync.Mutex
|
||||
}
|
||||
|
||||
func (m *Manager) Start() {
|
||||
m.working = make(map[int]*Worker)
|
||||
go m.work()
|
||||
}
|
||||
|
||||
func (m *Manager) AddRange(c tele.Context, hash string, from, to int) {
|
||||
m.queueLock.Lock()
|
||||
defer m.queueLock.Unlock()
|
||||
|
||||
if len(m.queue) > 50 {
|
||||
c.Bot().Send(c.Recipient(), "Очередь переполнена, попробуйте попозже\n\nЭлементов в очереди:"+strconv.Itoa(len(m.queue)))
|
||||
return
|
||||
}
|
||||
|
||||
m.ids++
|
||||
if m.ids > math.MaxInt {
|
||||
m.ids = 0
|
||||
}
|
||||
|
||||
var msg *tele.Message
|
||||
var err error
|
||||
|
||||
for i := 0; i < 20; i++ {
|
||||
msg, err = c.Bot().Send(c.Recipient(), "<b>Подключение к торренту</b>\n<code>"+hash+"</code>")
|
||||
if err == nil {
|
||||
break
|
||||
} else {
|
||||
log.Println("Error send msg, try again:", i+1, "/", 20)
|
||||
}
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
log.Println("Error send msg:", err)
|
||||
return
|
||||
}
|
||||
|
||||
t := torr.GetTorrent(hash)
|
||||
t.WaitInfo()
|
||||
for t.Status().Stat != state.TorrentWorking {
|
||||
time.Sleep(time.Second)
|
||||
t = torr.GetTorrent(hash)
|
||||
}
|
||||
ti := t.Status()
|
||||
|
||||
if from == 1 && to == -1 {
|
||||
to = len(ti.FileStats)
|
||||
}
|
||||
if to > len(ti.FileStats) {
|
||||
to = len(ti.FileStats)
|
||||
}
|
||||
if from < 1 {
|
||||
from = 1
|
||||
}
|
||||
if to >= 0 && to < from {
|
||||
from, to = to, from
|
||||
}
|
||||
if to > len(ti.FileStats) {
|
||||
to = len(ti.FileStats)
|
||||
}
|
||||
|
||||
w := &Worker{
|
||||
id: m.ids,
|
||||
c: c,
|
||||
torrentHash: hash,
|
||||
msg: msg,
|
||||
ti: ti,
|
||||
from: from,
|
||||
to: to,
|
||||
}
|
||||
|
||||
m.queue = append(m.queue, w)
|
||||
}
|
||||
|
||||
func (m *Manager) Cancel(id int) {
|
||||
m.queueLock.Lock()
|
||||
defer m.queueLock.Unlock()
|
||||
var rem []int
|
||||
for i, w := range m.queue {
|
||||
if w.id == id {
|
||||
w.isCancelled = true
|
||||
w.c.Bot().Delete(w.msg)
|
||||
rem = append(rem, i)
|
||||
return
|
||||
}
|
||||
}
|
||||
for _, i := range rem {
|
||||
m.queue = append(m.queue[:i], m.queue[i+1:]...)
|
||||
}
|
||||
if wrk, ok := m.working[id]; ok {
|
||||
wrk.isCancelled = true
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) work() {
|
||||
for {
|
||||
m.queueLock.Lock()
|
||||
if len(m.working) > 0 {
|
||||
m.queueLock.Unlock()
|
||||
m.sendQueueStatus()
|
||||
time.Sleep(time.Second)
|
||||
continue
|
||||
}
|
||||
if len(m.queue) == 0 {
|
||||
m.queueLock.Unlock()
|
||||
time.Sleep(time.Second)
|
||||
continue
|
||||
}
|
||||
wrk := m.queue[0]
|
||||
m.queue = m.queue[1:]
|
||||
m.working[wrk.id] = wrk
|
||||
m.queueLock.Unlock()
|
||||
|
||||
m.sendQueueStatus()
|
||||
|
||||
loading(wrk)
|
||||
|
||||
m.queueLock.Lock()
|
||||
delete(m.working, wrk.id)
|
||||
m.queueLock.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) sendQueueStatus() {
|
||||
m.queueLock.Lock()
|
||||
defer m.queueLock.Unlock()
|
||||
for i, wrk := range m.queue {
|
||||
if wrk.msg == nil {
|
||||
continue
|
||||
}
|
||||
torrKbd := &tele.ReplyMarkup{}
|
||||
torrKbd.Inline([]tele.Row{torrKbd.Row(torrKbd.Data("Отмена", "cancel", strconv.Itoa(wrk.id)))}...)
|
||||
|
||||
msg := "Номер в очереди " + strconv.Itoa(i+1)
|
||||
|
||||
wrk.c.Bot().Edit(wrk.msg, msg, torrKbd)
|
||||
}
|
||||
}
|
||||
|
||||
func loading(wrk *Worker) {
|
||||
iserr := false
|
||||
|
||||
t := torr.GetTorrent(wrk.torrentHash)
|
||||
t.WaitInfo()
|
||||
for t.Status().Stat != state.TorrentWorking {
|
||||
time.Sleep(time.Second)
|
||||
t = torr.GetTorrent(wrk.torrentHash)
|
||||
}
|
||||
wrk.ti = t.Status()
|
||||
|
||||
for i := wrk.from - 1; i <= wrk.to-1; i++ {
|
||||
file := wrk.ti.FileStats[i]
|
||||
if wrk.isCancelled {
|
||||
return
|
||||
}
|
||||
|
||||
err := uploadFile(wrk, file, i+1, len(wrk.ti.FileStats))
|
||||
if err != nil {
|
||||
errstr := fmt.Sprintf("Ошибка загрузки в телеграм: %v", err.Error())
|
||||
wrk.c.Bot().Edit(wrk.msg, errstr)
|
||||
iserr = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !iserr {
|
||||
wrk.c.Bot().Delete(wrk.msg)
|
||||
}
|
||||
}
|
||||
|
||||
func uploadFile(wrk *Worker, file *state.TorrentFileStat, fi, fc int) error {
|
||||
caption := filepath.Base(file.Path)
|
||||
torrFile, err := NewTorrFile(wrk, file)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var wa sync.WaitGroup
|
||||
wa.Add(1)
|
||||
complete := false
|
||||
go func() {
|
||||
for !complete {
|
||||
updateLoadStatus(wrk, torrFile, fi, fc)
|
||||
time.Sleep(1 * time.Second)
|
||||
}
|
||||
wa.Done()
|
||||
}()
|
||||
|
||||
d := &tele.Document{}
|
||||
d.FileName = file.Path
|
||||
d.Caption = caption
|
||||
d.File.FileReader = torrFile
|
||||
|
||||
for i := 0; i < 20; i++ {
|
||||
err = wrk.c.Send(d)
|
||||
if err == nil || errors.Is(err, ERR_STOPPED) {
|
||||
break
|
||||
} else {
|
||||
log.Println("Error send msg, try again:", i+1, "/", 20)
|
||||
}
|
||||
}
|
||||
|
||||
complete = true
|
||||
wa.Wait()
|
||||
torrFile.Close()
|
||||
if errors.Is(err, ERR_STOPPED) {
|
||||
err = nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
127
server/tgbot/upload/queue.go
Normal file
127
server/tgbot/upload/queue.go
Normal file
@@ -0,0 +1,127 @@
|
||||
package upload
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"github.com/dustin/go-humanize"
|
||||
tele "gopkg.in/telebot.v4"
|
||||
"server/torr"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
type DLQueue struct {
|
||||
id int
|
||||
c tele.Context
|
||||
hash string
|
||||
fileID string
|
||||
fileName string
|
||||
updateMsg *tele.Message
|
||||
}
|
||||
|
||||
var (
|
||||
manager = &Manager{}
|
||||
)
|
||||
|
||||
func Start() {
|
||||
manager.Start()
|
||||
}
|
||||
|
||||
func ShowQueue(c tele.Context) error {
|
||||
msg := ""
|
||||
manager.queueLock.Lock()
|
||||
defer manager.queueLock.Unlock()
|
||||
if len(manager.queue) == 0 && len(manager.working) == 0 {
|
||||
return c.Send("Очередь пуста")
|
||||
}
|
||||
if len(manager.working) > 0 {
|
||||
msg += "Закачиваются:\n"
|
||||
i := 0
|
||||
for _, dlQueue := range manager.working {
|
||||
s := "#" + strconv.Itoa(i+1) + ": <code>" + dlQueue.torrentHash + "</code>\n"
|
||||
if len(msg+s) > 1024 {
|
||||
c.Send(msg)
|
||||
msg = ""
|
||||
}
|
||||
msg += s
|
||||
i++
|
||||
}
|
||||
if len(msg) > 0 {
|
||||
c.Send(msg)
|
||||
msg = ""
|
||||
}
|
||||
}
|
||||
if len(manager.queue) > 0 {
|
||||
msg = "В очереди:\n"
|
||||
for i, dlQueue := range manager.queue {
|
||||
s := "#" + strconv.Itoa(i+1) + ": <code>" + dlQueue.torrentHash + "</code>\n"
|
||||
if len(msg+s) > 1024 {
|
||||
c.Send(msg)
|
||||
msg = ""
|
||||
}
|
||||
msg += s
|
||||
}
|
||||
if len(msg) > 0 {
|
||||
c.Send(msg)
|
||||
msg = ""
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func AddRange(c tele.Context, hash string, from, to int) {
|
||||
manager.AddRange(c, hash, from, to)
|
||||
}
|
||||
|
||||
func Cancel(id int) {
|
||||
manager.Cancel(id)
|
||||
}
|
||||
|
||||
func updateLoadStatus(wrk *Worker, file *TorrFile, fi, fc int) {
|
||||
if wrk.msg == nil {
|
||||
return
|
||||
}
|
||||
t := torr.GetTorrent(wrk.torrentHash)
|
||||
ti := t.Status()
|
||||
if wrk.isCancelled {
|
||||
wrk.c.Bot().Edit(wrk.msg, "Остановка...")
|
||||
} else {
|
||||
wrk.c.Send(tele.UploadingVideo)
|
||||
if ti.DownloadSpeed == 0 {
|
||||
ti.DownloadSpeed = 1.0
|
||||
}
|
||||
wait := time.Duration(float64(file.Loaded())/ti.DownloadSpeed) * time.Second
|
||||
speed := humanize.Bytes(uint64(ti.DownloadSpeed)) + "/sec"
|
||||
peers := fmt.Sprintf("%v · %v/%v", ti.ConnectedSeeders, ti.ActivePeers, ti.TotalPeers)
|
||||
prc := fmt.Sprintf("%.2f%% %v / %v", float64(file.offset)*100.0/float64(file.size), humanize.Bytes(uint64(file.offset)), humanize.Bytes(uint64(file.size)))
|
||||
|
||||
name := file.name
|
||||
if name == ti.Title {
|
||||
name = ""
|
||||
}
|
||||
|
||||
msg := "Загрузка торрента:\n" +
|
||||
"<b>" + ti.Title + "</b>\n"
|
||||
if name != "" {
|
||||
msg += "<i>" + name + "</i>\n"
|
||||
}
|
||||
msg += "<b>Хэш:</b> <code>" + file.hash + "</code>\n"
|
||||
if file.offset < file.size {
|
||||
msg += "<b>Скорость: </b>" + speed + "\n" +
|
||||
"<b>Осталось: </b>" + wait.String() + "\n" +
|
||||
"<b>Пиры: </b>" + peers + "\n" +
|
||||
"<b>Загружено: </b>" + prc
|
||||
}
|
||||
if fc > 1 {
|
||||
msg += "\n<b>Файлов: </b>" + strconv.Itoa(fi) + "/" + strconv.Itoa(fc)
|
||||
}
|
||||
if file.offset >= file.size {
|
||||
msg += "\n<b>Завершение загрузки, это займет некоторое время</b>"
|
||||
wrk.c.Bot().Edit(wrk.msg, msg)
|
||||
return
|
||||
}
|
||||
|
||||
torrKbd := &tele.ReplyMarkup{}
|
||||
torrKbd.Inline([]tele.Row{torrKbd.Row(torrKbd.Data("Отмена", "cancel", strconv.Itoa(wrk.id)))}...)
|
||||
wrk.c.Bot().Edit(wrk.msg, msg, torrKbd)
|
||||
}
|
||||
}
|
||||
89
server/tgbot/upload/torrfile.go
Normal file
89
server/tgbot/upload/torrfile.go
Normal file
@@ -0,0 +1,89 @@
|
||||
package upload
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/anacrolix/torrent"
|
||||
"log"
|
||||
"path/filepath"
|
||||
sets "server/settings"
|
||||
"server/tgbot/config"
|
||||
"server/torr"
|
||||
"server/torr/state"
|
||||
"server/torr/storage/torrstor"
|
||||
)
|
||||
|
||||
var ERR_STOPPED = errors.New("stopped")
|
||||
|
||||
type TorrFile struct {
|
||||
hash string
|
||||
name string
|
||||
wrk *Worker
|
||||
offset int64
|
||||
size int64
|
||||
id int
|
||||
|
||||
reader *torrstor.Reader
|
||||
}
|
||||
|
||||
func NewTorrFile(wrk *Worker, stFile *state.TorrentFileStat) (*TorrFile, error) {
|
||||
if config.Cfg.HostTG != "" && stFile.Length > 2*1024*1024*1024 {
|
||||
return nil, errors.New("Размер файла должен быть больше 2GB")
|
||||
}
|
||||
if config.Cfg.HostTG == "" && stFile.Length > 50*1024*1024 {
|
||||
return nil, errors.New("Размер файла должен быть больше 50MB\nЧтобы закачивать файлы до 2GB нужно в tg.cfg указать host к <a href='https://github.com/tdlib/telegram-bot-api'>telegram bot api</a>")
|
||||
}
|
||||
|
||||
tf := new(TorrFile)
|
||||
tf.hash = wrk.torrentHash
|
||||
tf.name = filepath.Base(stFile.Path)
|
||||
tf.wrk = wrk
|
||||
tf.size = stFile.Length
|
||||
|
||||
t := torr.GetTorrent(wrk.torrentHash)
|
||||
t.WaitInfo()
|
||||
|
||||
files := t.Files()
|
||||
var file *torrent.File
|
||||
for _, tfile := range files {
|
||||
if tfile.Path() == stFile.Path {
|
||||
file = tfile
|
||||
break
|
||||
}
|
||||
}
|
||||
if file == nil {
|
||||
return nil, fmt.Errorf("file with id %v not found", stFile.Id)
|
||||
}
|
||||
if int64(sets.MaxSize) > 0 && file.Length() > int64(sets.MaxSize) {
|
||||
log.Println("file", file.DisplayPath(), "size exceeded max allowed", sets.MaxSize, "bytes")
|
||||
return nil, fmt.Errorf("file size exceeded max allowed %d bytes", sets.MaxSize)
|
||||
}
|
||||
|
||||
reader := t.NewReader(file)
|
||||
if sets.BTsets.ResponsiveMode {
|
||||
reader.SetResponsive()
|
||||
}
|
||||
tf.reader = reader
|
||||
|
||||
return tf, nil
|
||||
}
|
||||
|
||||
func (t *TorrFile) Read(p []byte) (n int, err error) {
|
||||
if t.wrk.isCancelled {
|
||||
return 0, ERR_STOPPED
|
||||
}
|
||||
n, err = t.reader.Read(p)
|
||||
t.offset += int64(n)
|
||||
return
|
||||
}
|
||||
|
||||
func (t *TorrFile) Loaded() int64 {
|
||||
return t.size - t.offset
|
||||
}
|
||||
|
||||
func (t *TorrFile) Close() {
|
||||
if t.reader != nil {
|
||||
t.reader.Close()
|
||||
t.reader = nil
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user