diff --git a/README.md b/README.md index 15b3996..b85c8f6 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,33 @@ build unless the destination will never change. ./tgexport sync -c CHAT -r REMOTE:PATH [options] ``` +A run states what it found, what it is about to do, and then shows each file as +it moves: + +``` +reading mychannel + 12,000 messages read in 4m31s +indexing PikPak root 'mychannel' + 11,406 objects listed in 2m10s + + chat holds 12,000 media, 250.0 GiB + archived 11,400 + never fetched 600 + wrong size 6 + + fetching 606 files, 79.0 GiB (largest 2.0 GiB) + into PikPak root 'mychannel' + staging ./staging, capped at 40.0 GiB + concurrency 2 download(s) x 4 thread(s), 2 upload(s) + + total 26/606 files [=> ] 617.5 MiB / 79.0 GiB 2.5 MiB/s 8h47m + ↓ …3214_4242_1000000000000000001.mp4 [=======> ] 41.2 MiB / 96.0 MiB 1.8 MiB/s + ↑ …3214_4243_1000000000000000002.mp4 ⠹ uploading 1.9 GiB +``` + +Redirected output gets the same information as plain periodic lines plus one +line per archived file, with no cursor movement — a captured log stays readable. + `CHAT` accepts a numeric id as printed by `tdl chat ls`, a username with or without `@`, or a `t.me`/`tg://` link. A Bot API `-100…` id is converted automatically. A link to a single *message* is refused — it names a message, not diff --git a/cmd/tgexport/sync.go b/cmd/tgexport/sync.go index 6b3a6a1..23608a5 100644 --- a/cmd/tgexport/sync.go +++ b/cmd/tgexport/sync.go @@ -142,8 +142,7 @@ func syncCmd(ctx context.Context, args []string) error { } warnCollisions(idx) before := verify.Check(items, idx) - fmt.Fprintf(os.Stderr, "%d media messages, %d already archived, %d to fetch\n", - before.Expected, before.Present, len(before.Todo())) + report.Survey(os.Stderr, before) todo := selectTodo(items, before, *limitItems) if len(todo) == 0 { @@ -155,13 +154,24 @@ func syncCmd(ctx context.Context, args []string) error { return err } - var todoBytes int64 + var todoBytes, largest int64 for _, it := range todo { todoBytes += it.Size() + largest = max(largest, it.Size()) } - fmt.Fprintf(os.Stderr, "fetching %d file(s), %.1f GiB\n", len(todo), float64(todoBytes)/(1<<30)) + report.Plan(os.Stderr, report.PlanInfo{ + Files: len(todo), + Bytes: todoBytes, + Largest: largest, + Budget: budget, + Staging: *staging, + Threads: *threads, + Downloads: *limit, + Uploads: *uploads, + Destination: dst.String(), + }) - rep := report.New(os.Stderr, len(todo), todoBytes) + rep := report.Events(os.Stderr, len(todo), todoBytes) var res pipeline.Result res, runErr = pipeline.Run(ctx, sliceSeq(todo), pipeline.Options{ Pool: pool, @@ -173,7 +183,7 @@ func syncCmd(ctx context.Context, args []string) error { Budget: budget, Confirm: *confirm, Takeout: *takeout, - Report: rep.Update, + Events: rep, // Re-checked during the run, not only before it: an archive of this // size runs for hours, and the destination can fill in the middle. FreeBytes: func(ctx context.Context) (int64, bool) { diff --git a/cmd/uidemo/main.go b/cmd/uidemo/main.go new file mode 100644 index 0000000..10e4b7f --- /dev/null +++ b/cmd/uidemo/main.go @@ -0,0 +1,75 @@ +// Command uidemo renders the sync progress UI with fake data, so the layout can +// be checked without a Telegram session or a multi-hour transfer. +package main + +import ( + "fmt" + "os" + "time" + + "github.com/iyear/tdl/core/tmedia" + + "github.com/tiennm99dev/telegram-exporter/internal/pipeline" + "github.com/tiennm99dev/telegram-exporter/internal/report" + "github.com/tiennm99dev/telegram-exporter/internal/tgsource" + "github.com/tiennm99dev/telegram-exporter/internal/verify" +) + +func item(id int, name string, size int64) tgsource.Item { + return tgsource.Item{MessageID: id, Name: name, Media: &tmedia.Media{Size: size}} +} + +func main() { + files := []tgsource.Item{ + item(4242, "1234567890_4242_1000000000000000001.mp4", 96<<20), + item(4243, "1234567890_4243_1000000000000000002.mp4", 1900<<20), + item(4244, "1234567890_4244_short.jpg", 2<<20), + } + var total int64 + for _, f := range files { + total += f.Size() + } + fmt.Fprintln(os.Stderr, "PikPak root 'mychannel' has 1438 GiB free") + fmt.Fprintln(os.Stderr, "reading mychannel") + fmt.Fprintln(os.Stderr, " 12,000 messages read in 4m31s") + fmt.Fprintln(os.Stderr, "indexing PikPak root 'mychannel'") + fmt.Fprintln(os.Stderr, " 11,406 objects listed in 2m10s") + + survey := verify.Report{Expected: 12000, Present: 11400, Bytes: 518 << 30} + for i := range 2607 { + survey.Absent = append(survey.Absent, 4242+i) + } + for i := range 6 { + survey.Mismatched = append(survey.Mismatched, + verify.Mismatch{MessageID: 14290 + i, Want: 2 << 30, Got: 221 << 20}) + } + report.Survey(os.Stderr, survey) + report.Plan(os.Stderr, report.PlanInfo{ + Files: 2613, Bytes: 79 << 30, Largest: 2 << 30, Budget: 40 << 30, + Staging: "./staging", Threads: 4, Downloads: 2, Uploads: 2, + Destination: "PikPak root 'mychannel'", + }) + + ev := report.Events(os.Stderr, 2613, 79<<30) + st := pipeline.Stats{} + for _, f := range files { + ev.DownloadStart(f) + } + for step := range 30 { + for _, f := range files { + ev.DownloadBytes(f, f.Size()*int64(step+1)/30) + } + st.BytesDone += total / 30 + ev.Stats(st) + time.Sleep(60 * time.Millisecond) + } + for i, f := range files { + ev.DownloadDone(f, nil) + ev.UploadStart(f) + st.Done = i + 1 + ev.Stats(st) + time.Sleep(400 * time.Millisecond) + ev.UploadDone(f, nil) + } + ev.Finish(st) +} diff --git a/go.mod b/go.mod index ee2e563..8fbdd96 100644 --- a/go.mod +++ b/go.mod @@ -32,8 +32,10 @@ require ( github.com/ProtonMail/go-srp v0.0.7 // indirect github.com/ProtonMail/gopenpgp/v3 v3.4.1 // indirect github.com/PuerkitoBio/goquery v1.12.0 // indirect + github.com/VividCortex/ewma v1.2.0 // indirect github.com/a1ex3/zstd-seekable-format-go/pkg v0.10.0 // indirect github.com/abbot/go-http-auth v0.4.0 // indirect + github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d // indirect github.com/adrg/xdg v0.5.3 // indirect github.com/anchore/go-lzo v0.1.1 // indirect github.com/andybalholm/brotli v1.2.2 // indirect @@ -157,7 +159,7 @@ require ( github.com/mailru/easyjson v0.9.2 // indirect github.com/mattn/go-colorable v0.1.15 // indirect github.com/mattn/go-isatty v0.0.23 // indirect - github.com/mattn/go-runewidth v0.0.24 // indirect + github.com/mattn/go-runewidth v0.0.28 // indirect github.com/mitchellh/go-homedir v1.1.0 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/ncw/swift/v2 v2.0.5 // indirect @@ -203,6 +205,8 @@ require ( github.com/tyler-smith/go-bip39 v1.1.0 // indirect github.com/ulikunitz/xz v0.5.15 // indirect github.com/unknwon/goconfig v1.0.0 // indirect + github.com/vbauerster/cupwriter v0.0.4 // indirect + github.com/vbauerster/mpb/v8 v8.16.1 // indirect github.com/wk8/go-ordered-map/v2 v2.1.8 // indirect github.com/xanzy/ssh-agent v0.3.3 // indirect github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect diff --git a/go.sum b/go.sum index 9a73884..af02a09 100644 --- a/go.sum +++ b/go.sum @@ -51,12 +51,16 @@ github.com/ProtonMail/gopenpgp/v3 v3.4.1 h1:K7uUhSHSJxORZ+RuHpilTT6S4MA2whCRlXNw github.com/ProtonMail/gopenpgp/v3 v3.4.1/go.mod h1:bGdV9f6edhmd581wzXsQCTKdH8bXBbyhkgDKPjwPc6U= github.com/PuerkitoBio/goquery v1.12.0 h1:pAcL4g3WRXekcB9AU/y1mbKez2dbY2AajVhtkO8RIBo= github.com/PuerkitoBio/goquery v1.12.0/go.mod h1:802ej+gV2y7bbIhOIoPY5sT183ZW0YFofScC4q/hIpQ= +github.com/VividCortex/ewma v1.2.0 h1:f58SaIzcDXrSy3kWaHNvuJgJ3Nmz59Zji6XoJR/q1ow= +github.com/VividCortex/ewma v1.2.0/go.mod h1:nz4BbCtbLyFDeC9SUHbtcT5644juEuWfUAUnGx7j5l4= github.com/a1ex3/zstd-seekable-format-go/pkg v0.10.0 h1:iLDOF0rdGTrol/q8OfPIIs5kLD8XvA2q75o6Uq/tgak= github.com/a1ex3/zstd-seekable-format-go/pkg v0.10.0/go.mod h1:DrEWcQJjz7t5iF2duaiyhg4jyoF0kxOD6LtECNGkZ/Q= github.com/aalpar/deheap v1.1.2 h1:MABHLcnjqsffb8GLkUFDigqpBBxOMz0DoKM9QfELeTw= github.com/aalpar/deheap v1.1.2/go.mod h1:A+nfkD4JbS05sewV0he/MYgR/90vfqyMoNNROgs+rmA= github.com/abbot/go-http-auth v0.4.0 h1:QjmvZ5gSC7jm3Zg54DqWE/T5m1t2AfDu6QlXJT0EVT0= github.com/abbot/go-http-auth v0.4.0/go.mod h1:Cz6ARTIzApMJDzh5bRMSUou6UMSp0IEXg9km/ci7TJM= +github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d h1:licZJFw2RwpHMqeKTCYkitsPqHNxTmd4SNR5r94FGM8= +github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d/go.mod h1:asat636LX7Bqt5lYEZ27JNDcqxfjdBQuJ/MM4CN/Lzo= github.com/adrg/xdg v0.5.3 h1:xRnxJXne7+oWDatRhR1JLnvuccuIeCoBu2rtuLqQB78= github.com/adrg/xdg v0.5.3/go.mod h1:nlTsY+NNiCBGCK2tpm09vRqfVzrc2fLmXGpBLF0zlTQ= github.com/anchore/go-lzo v0.1.1 h1:IwL/fvkdtlIrYIXck6WxZ3nb8WjjHziYYmGxlooyOnM= @@ -384,6 +388,8 @@ github.com/mattn/go-isatty v0.0.23/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI github.com/mattn/go-runewidth v0.0.3/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU= github.com/mattn/go-runewidth v0.0.24 h1:cpokDiIn0MGnhdHwuWnJBITySJ20QyNGnY2kR/ay2DU= github.com/mattn/go-runewidth v0.0.24/go.mod h1:XBkDxAl56ILZc9knddidhrOlY5R/pDhgLpndooCuJAs= +github.com/mattn/go-runewidth v0.0.28 h1:rPyg2ybwEKPebvpzVWe1gKBkH8EQFkxO4Y0hjBeLaBU= +github.com/mattn/go-runewidth v0.0.28/go.mod h1:3qAiGCV4Koz/yuveO58qUefmUTRm8r0IGEXZ9jeHp/8= github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y= github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0= github.com/moby/sys/mountinfo v0.7.2 h1:1shs6aH5s4o5H2zQLn796ADW1wMrIwHsyJ2v9KouLrg= @@ -526,6 +532,10 @@ github.com/ulikunitz/xz v0.5.15 h1:9DNdB5s+SgV3bQ2ApL10xRc35ck0DuIX/isZvIk+ubY= github.com/ulikunitz/xz v0.5.15/go.mod h1:nbz6k7qbPmH4IRqmfOplQw/tblSgqTqBwxkY0oWt/14= github.com/unknwon/goconfig v1.0.0 h1:rS7O+CmUdli1T+oDm7fYj1MwqNWtEJfNj+FqcUHML8U= github.com/unknwon/goconfig v1.0.0/go.mod h1:qu2ZQ/wcC/if2u32263HTVC39PeOQRSmidQk3DuDFQ8= +github.com/vbauerster/cupwriter v0.0.4 h1:9sBPe0uXWLZuWQU5lqVbhyFlxX6c09asST/YfatFAys= +github.com/vbauerster/cupwriter v0.0.4/go.mod h1:IFyzS6Xis5dnBH/rdAhrnuzg3c+KkUqEN6yE8lhJlDw= +github.com/vbauerster/mpb/v8 v8.16.1 h1:gNYmwMip9xRWNGAiblZOgUNXWeU2P0NIGd5x0f8ffbc= +github.com/vbauerster/mpb/v8 v8.16.1/go.mod h1:gnU8zNF/JWltFepqwko/ulMEUIDrydIq7T4UdMN26Nw= github.com/wk8/go-ordered-map/v2 v2.1.8 h1:5h/BUHu93oj4gIdvHHHGsScSTMijfx5PeYkE/fJgbpc= github.com/wk8/go-ordered-map/v2 v2.1.8/go.mod h1:5nJHM5DyteebpVlHnWMV0rPz6Zp7+xBAnxjb1X5vnTw= github.com/xanzy/ssh-agent v0.3.3 h1:+/15pJfg/RsTxqYcX6fHqOXZwwMP+2VyYWJeWM2qQFM= diff --git a/internal/pipeline/download.go b/internal/pipeline/download.go index 7d9bf8a..daa3718 100644 --- a/internal/pipeline/download.go +++ b/internal/pipeline/download.go @@ -24,10 +24,10 @@ type DownloadOptions struct { Limit int // files in flight Takeout bool // use a takeout session, as `tdl dl --takeout` does - // Report, when set, is called with a running summary. It is invoked from - // download worker goroutines, so it must be cheap and safe to call + // Events, when set, receives the run's per-item lifecycle. It is invoked + // from download worker goroutines, so it must be cheap and safe to call // concurrently. - Report func(Stats) + Events Events // acquire reserves staging space before a download starts, blocking until // there is room. Unset means no bound. release hands a reservation back for @@ -87,7 +87,7 @@ func Download(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o Downlo o.onFailed(e.item) } return ferr - }, o.Report) + }, o.Events) err := downloader.New(downloader.Options{ Pool: o.Pool, diff --git a/internal/pipeline/events.go b/internal/pipeline/events.go new file mode 100644 index 0000000..8356ce6 --- /dev/null +++ b/internal/pipeline/events.go @@ -0,0 +1,38 @@ +package pipeline + +import "github.com/tiennm99dev/telegram-exporter/internal/tgsource" + +// Events receives a run's per-item lifecycle. +// +// Aggregate Stats alone cannot say what a run is doing right now — which files +// are moving, how far along each is, whether the time is going into downloads or +// uploads. On an archive that runs for hours those are the questions being +// asked, and a single "26/2613 done" line answers none of them. +// +// Every method is called from a worker goroutine, several at once, and some of +// them fire per network chunk. Implementations must be safe to call +// concurrently and must not block: a slow renderer would throttle the transfers +// it is describing. +type Events interface { + // Stats reports the running totals. + Stats(Stats) + + DownloadStart(it tgsource.Item) + // DownloadBytes reports the total written for it so far, not a delta. + DownloadBytes(it tgsource.Item, done int64) + DownloadDone(it tgsource.Item, err error) + + UploadStart(it tgsource.Item) + UploadDone(it tgsource.Item, err error) +} + +// nopEvents is used when a caller wants no reporting, so nothing on the hot +// path has to nil-check. +type nopEvents struct{} + +func (nopEvents) Stats(Stats) {} +func (nopEvents) DownloadStart(tgsource.Item) {} +func (nopEvents) DownloadBytes(tgsource.Item, int64) {} +func (nopEvents) DownloadDone(tgsource.Item, error) {} +func (nopEvents) UploadStart(tgsource.Item) {} +func (nopEvents) UploadDone(tgsource.Item, error) {} diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index a0d6cc7..def6a24 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -42,7 +42,9 @@ type Options struct { FreeBytes func(context.Context) (int64, bool) MinFree int64 - Report func(Stats) + // Events, when set, receives the run's per-item lifecycle: which files are + // downloading, which are uploading, and how far along each one is. + Events Events } // Result is what a run achieved. @@ -108,6 +110,9 @@ func Run(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o Options) (R if o.MaxFailures <= 0 { o.MaxFailures = 5 } + if o.Events == nil { + o.Events = nopEvents{} + } local, err := fs.NewFs(ctx, o.Staging) if err != nil { @@ -145,7 +150,9 @@ func Run(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o Options) (R // attempt, so the file is dropped rather than uploaded. err := guard.check(ctx) if err == nil { + o.Events.UploadStart(it) err = up.upload(ctx, it) + o.Events.UploadDone(it, err) } if err != nil { @@ -186,7 +193,7 @@ func Run(ctx context.Context, seq iter.Seq2[tgsource.Item, error], o Options) (R Threads: o.Threads, Limit: o.Limit, Takeout: o.Takeout, - Report: o.Report, + Events: o.Events, acquire: budget.acquire, release: budget.release, onReady: func(it tgsource.Item) { uploads <- it }, diff --git a/internal/pipeline/progress.go b/internal/pipeline/progress.go index 8c7c75c..5ad97da 100644 --- a/internal/pipeline/progress.go +++ b/internal/pipeline/progress.go @@ -36,14 +36,17 @@ type progress struct { inFlight map[int]int64 // message id -> bytes written so far finish func(*elem, error) error - report func(Stats) + events Events } -func newProgress(finish func(*elem, error) error, report func(Stats)) *progress { +func newProgress(finish func(*elem, error) error, events Events) *progress { + if events == nil { + events = nopEvents{} + } return &progress{ inFlight: make(map[int]int64), finish: finish, - report: report, + events: events, } } @@ -54,7 +57,8 @@ func (p *progress) OnAdd(e downloader.Elem) { p.stats.BytesTotal += el.item.Size() stats := p.stats p.mu.Unlock() - p.emit(stats) + p.events.DownloadStart(el.item) + p.events.Stats(stats) } func (p *progress) OnDownload(e downloader.Elem, state downloader.ProgressState) { @@ -67,7 +71,8 @@ func (p *progress) OnDownload(e downloader.Elem, state downloader.ProgressState) p.stats.BytesDone += state.Downloaded - prev stats := p.stats p.mu.Unlock() - p.emit(stats) + p.events.DownloadBytes(el.item, state.Downloaded) + p.events.Stats(stats) } func (p *progress) OnDone(e downloader.Elem, err error) { @@ -89,13 +94,8 @@ func (p *progress) OnDone(e downloader.Elem, err error) { p.outcomes = append(p.outcomes, Outcome{Item: el.item, Err: err}) stats := p.stats p.mu.Unlock() - p.emit(stats) -} - -func (p *progress) emit(s Stats) { - if p.report != nil { - p.report(s) - } + p.events.DownloadDone(el.item, err) + p.events.Stats(stats) } func (p *progress) results() ([]Outcome, Stats) { diff --git a/internal/report/live.go b/internal/report/live.go new file mode 100644 index 0000000..d8410a6 --- /dev/null +++ b/internal/report/live.go @@ -0,0 +1,219 @@ +package report + +import ( + "fmt" + "io" + "sync" + "time" + + "github.com/vbauerster/mpb/v8" + "github.com/vbauerster/mpb/v8/decor" + + "github.com/tiennm99dev/telegram-exporter/internal/pipeline" + "github.com/tiennm99dev/telegram-exporter/internal/tgsource" +) + +// nameWidth is how much of a filename a per-file bar shows. Telegram names run +// to 100+ characters and the bar has to fit beside them. +const nameWidth = 34 + +// Live renders a run as a set of progress bars: one overall, plus one for each +// file currently moving. +// +// The aggregate line it replaces could say how much was done but never what was +// happening — which files were in flight, whether a stall was a slow download or +// a slow upload, how long the rest would take. On a run measured in hours those +// are the only questions worth answering. +// +// Bars are for terminals only. A redirected run gets lineEvents instead, because +// this writes ANSI cursor movement continuously and a captured log of it is +// unreadable. +type Live struct { + w io.Writer + p *mpb.Progress + total *mpb.Bar + + mu sync.Mutex + down map[int]*mpb.Bar + up map[int]*mpb.Bar + // done is read by the overall bar's decorator from mpb's render goroutine, + // so it is guarded by the same lock as the maps. + done int + + // seq gives each per-file bar a distinct, increasing priority so bars keep + // their position between frames. Sharing one priority lets mpb reorder them + // on every redraw, which makes a steady transfer look like it is thrashing. + seq int + files int + start time.Time +} + +// NewLive builds a bar renderer over w for a run of the given size. +func NewLive(w io.Writer, files int, bytes int64) *Live { + p := mpb.New( + mpb.WithOutput(w), + mpb.WithWidth(28), + mpb.WithRefreshRate(120*time.Millisecond), + ) + l := &Live{ + w: w, + p: p, + down: make(map[int]*mpb.Bar), + up: make(map[int]*mpb.Bar), + files: files, + start: time.Now(), + } + l.total = p.New(bytes, + mpb.BarStyle().Lbound("[").Filler("=").Tip(">").Padding(" ").Rbound("]"), + mpb.BarPriority(0), + mpb.BarNoPop(), + mpb.PrependDecorators( + decor.Name(" total ", decor.WC{W: 9}), + decor.Any(func(decor.Statistics) string { return l.counts() }, decor.WC{W: 16}), + ), + mpb.AppendDecorators( + decor.CountersKibiByte("% .1f / % .1f", decor.WC{W: 20}), + decor.AverageSpeed(decor.SizeB1024(0), " % .1f", decor.WC{W: 12}), + decor.OnComplete(decor.AverageETA(decor.ET_STYLE_GO, decor.WC{W: 10}), ""), + ), + ) + return l +} + +// Bars are grouped by band: the total on top, then downloads, then uploads. +// Within a band they are ordered by when they started. +const ( + downloadBand = 1 << 20 + uploadBand = 1 << 21 +) + +// next allocates the priority for a new bar in the given band. +func (l *Live) next(band int) int { + l.mu.Lock() + defer l.mu.Unlock() + l.seq++ + return band + l.seq +} + +func (l *Live) counts() string { + l.mu.Lock() + defer l.mu.Unlock() + return fmt.Sprintf("%s/%s files", humanCount(l.done), humanCount(l.files)) +} + +// Stats advances the overall bar. +func (l *Live) Stats(s pipeline.Stats) { + l.mu.Lock() + l.done = s.Done + s.Failed + l.mu.Unlock() + // The bar's own counters, speed and ETA all derive from this, so it is what + // makes the totals move rather than sitting at zero. + l.total.SetCurrent(s.BytesDone) +} + +func (l *Live) DownloadStart(it tgsource.Item) { + bar := l.p.New(it.Size(), + mpb.BarStyle().Lbound("[").Filler("=").Tip(">").Padding(" ").Rbound("]"), + mpb.BarRemoveOnComplete(), + mpb.BarPriority(l.next(downloadBand)), + mpb.PrependDecorators( + decor.Name(" ↓ "), + decor.Name(short(it.Name), decor.WC{W: nameWidth + 2, C: decor.DindentRight}), + ), + mpb.AppendDecorators( + decor.CountersKibiByte("% .1f / % .1f", decor.WC{W: 20}), + decor.AverageSpeed(decor.SizeB1024(0), " % .1f", decor.WC{W: 12}), + ), + ) + l.mu.Lock() + l.down[it.MessageID] = bar + l.mu.Unlock() +} + +func (l *Live) DownloadBytes(it tgsource.Item, done int64) { + l.mu.Lock() + bar := l.down[it.MessageID] + l.mu.Unlock() + if bar != nil { + bar.SetCurrent(done) + } +} + +func (l *Live) DownloadDone(it tgsource.Item, err error) { + l.mu.Lock() + bar := l.down[it.MessageID] + delete(l.down, it.MessageID) + l.mu.Unlock() + if bar == nil { + return + } + // Aborted rather than completed on failure, so a bar for a file that never + // arrived does not linger at 100%. + if err != nil { + bar.Abort(true) + return + } + bar.SetCurrent(it.Size()) +} + +func (l *Live) UploadStart(it tgsource.Item) { + // No byte-level callbacks exist for the upload leg — rclone's MoveFile is a + // single blocking call — so this is a spinner, not a bar. Showing which file + // is uploading is the point: a run that looks stalled is usually waiting on + // one large object, and until now nothing said so. + bar := l.p.New(0, mpb.SpinnerStyle().PositionLeft(), + mpb.BarRemoveOnComplete(), + mpb.BarPriority(l.next(uploadBand)), + mpb.PrependDecorators( + decor.Name(" ↑ "), + decor.Name(short(it.Name), decor.WC{W: nameWidth + 2, C: decor.DindentRight}), + ), + mpb.AppendDecorators( + decor.Any(func(decor.Statistics) string { + return fmt.Sprintf("uploading %s", humanBytes(it.Size())) + }, decor.WC{W: 24}), + ), + ) + l.mu.Lock() + l.up[it.MessageID] = bar + l.mu.Unlock() +} + +func (l *Live) UploadDone(it tgsource.Item, _ error) { + l.mu.Lock() + bar := l.up[it.MessageID] + delete(l.up, it.MessageID) + l.mu.Unlock() + if bar != nil { + bar.Abort(true) + } +} + +// Finish drains the bars and prints the closing summary. +func (l *Live) Finish(s pipeline.Stats) { + l.mu.Lock() + for _, b := range l.down { + b.Abort(true) + } + for _, b := range l.up { + b.Abort(true) + } + clear(l.down) + clear(l.up) + l.mu.Unlock() + + l.total.Abort(true) + l.p.Wait() + writeSummary(l.w, s, time.Since(l.start)) +} + +// short trims a filename to fit beside a bar, keeping the end — the extension +// and the distinguishing digits are there, while the shared dialog-id prefix is +// not. +func short(name string) string { + r := []rune(name) + if len(r) <= nameWidth { + return name + } + return "…" + string(r[len(r)-nameWidth+1:]) +} diff --git a/internal/report/progress.go b/internal/report/progress.go index 4d31ec9..4e73b43 100644 --- a/internal/report/progress.go +++ b/internal/report/progress.go @@ -9,6 +9,7 @@ import ( "time" "github.com/tiennm99dev/telegram-exporter/internal/pipeline" + "github.com/tiennm99dev/telegram-exporter/internal/tgsource" ) // statsInterval is how often a redirected run prints a line. @@ -19,10 +20,14 @@ import ( // else gets a periodic summary. const statsInterval = 30 * time.Second -// Reporter renders progress, adapting to whether it is writing to a terminal. +// Reporter renders progress as periodic plain-text lines. +// +// This is the redirected-output path. Bars are deliberately absent: they are +// continuous ANSI cursor movement, and a captured log of them is megabytes of +// control characters — which is exactly what tdl's progress bar did to the shell +// pipeline's logs. type Reporter struct { w io.Writer - tty bool total int totalBytes int64 @@ -31,35 +36,40 @@ type Reporter struct { lastLine time.Time } -// New builds a reporter for w, which is treated as a terminal when it is one. +// Events builds the renderer suited to w: bars on a terminal, periodic lines +// anywhere else. // // The totals are passed in rather than taken from Stats because Stats.BytesTotal // only counts items the downloader has started, so a progress line built from it // shows a denominator that grows as the run proceeds — "0 B of 52 KiB" on a run // that will move gigabytes. The caller knows the real figures before starting. -func New(w io.Writer, total int, totalBytes int64) *Reporter { - return &Reporter{w: w, tty: isTerminal(w), total: total, totalBytes: totalBytes, started: time.Now()} +func Events(w io.Writer, total int, totalBytes int64) interface { + pipeline.Events + Finish(pipeline.Stats) +} { + if isTerminal(w) { + return NewLive(w, total, totalBytes) + } + return newReporter(w, total, totalBytes) } -// Update renders a snapshot. Safe to call from several goroutines. +func newReporter(w io.Writer, total int, totalBytes int64) *Reporter { + return &Reporter{w: w, total: total, totalBytes: totalBytes, started: time.Now()} +} + +// Stats renders a snapshot. Safe to call from several goroutines. // // A contended update is dropped rather than queued. Every download worker calls // this on each progress callback, so holding the lock across the write would -// make terminal latency — an ssh session with a slow link, say — throttle the -// downloads themselves. A skipped frame costs nothing; the next callback is -// milliseconds away and Finish always prints. -func (r *Reporter) Update(s pipeline.Stats) { +// make write latency throttle the downloads themselves. A skipped frame costs +// nothing; the next callback is milliseconds away and Finish always prints. +func (r *Reporter) Stats(s pipeline.Stats) { if !r.mu.TryLock() { return } defer r.mu.Unlock() now := time.Now() - if r.tty { - // \r rather than \n: one line, redrawn. - fmt.Fprintf(r.w, "\r\033[K%s", r.line(s, now)) - return - } if now.Sub(r.lastLine) < statsInterval { return } @@ -67,25 +77,49 @@ func (r *Reporter) Update(s pipeline.Stats) { fmt.Fprintf(r.w, "%s\n", r.line(s, now)) } -// Finish writes the closing summary, ending the redrawn line if there was one. +// The per-file events are recorded as one line each rather than a bar. At one +// line per file this stays readable in a log, and it is what makes a captured +// run auditable afterwards: which files moved, in what order, and which failed. +func (r *Reporter) DownloadStart(tgsource.Item) {} +func (r *Reporter) DownloadBytes(tgsource.Item, int64) {} +func (r *Reporter) UploadStart(tgsource.Item) {} + +func (r *Reporter) DownloadDone(it tgsource.Item, err error) { + if err != nil { + r.mu.Lock() + defer r.mu.Unlock() + fmt.Fprintf(r.w, " download failed %q: %v\n", it.Name, err) + } +} + +func (r *Reporter) UploadDone(it tgsource.Item, err error) { + r.mu.Lock() + defer r.mu.Unlock() + if err != nil { + fmt.Fprintf(r.w, " upload failed %q: %v\n", it.Name, err) + return + } + fmt.Fprintf(r.w, " archived %-10s %q\n", humanBytes(it.Size()), it.Name) +} + +// Finish writes the closing summary. func (r *Reporter) Finish(s pipeline.Stats) { r.mu.Lock() defer r.mu.Unlock() - if r.tty { - fmt.Fprint(r.w, "\r\033[K") - } - elapsed := time.Since(r.started).Round(time.Second) - fmt.Fprintf(r.w, "%d done, %d failed, %s in %s (%s/s)\n", - s.Done, s.Failed, humanBytes(s.BytesDone), elapsed, - humanBytes(int64(float64(s.BytesDone)/max(elapsed.Seconds(), 1)))) + writeSummary(r.w, s, time.Since(r.started)) } func (r *Reporter) line(s pipeline.Stats, now time.Time) string { elapsed := now.Sub(r.started) rate := float64(s.BytesDone) / max(elapsed.Seconds(), 1) - return fmt.Sprintf("%d/%d done, %d failed, %s of %s, %s/s", - s.Done, r.total, s.Failed, - humanBytes(s.BytesDone), humanBytes(r.totalBytes), humanBytes(int64(rate))) + eta := "—" + if rate > 0 && r.totalBytes > s.BytesDone { + eta = time.Duration(float64(r.totalBytes-s.BytesDone) / rate * float64(time.Second)). + Round(time.Second).String() + } + return fmt.Sprintf(" %s/%s files, %d failed, %s of %s, %s/s, ETA %s", + humanCount(s.Done), humanCount(r.total), s.Failed, + humanBytes(s.BytesDone), humanBytes(r.totalBytes), humanBytes(int64(rate)), eta) } func humanBytes(n int64) string { @@ -117,3 +151,16 @@ func isTerminal(w io.Writer) bool { } return info.Mode()&os.ModeCharDevice != 0 } + +// writeSummary prints the closing line both renderers end with. +func writeSummary(w io.Writer, s pipeline.Stats, elapsed time.Duration) { + elapsed = elapsed.Round(time.Second) + fmt.Fprintf(w, "%s done, %s failed, %s in %s (%s/s)\n", + humanCount(s.Done), humanCount(s.Failed), humanBytes(s.BytesDone), elapsed, + humanBytes(int64(float64(s.BytesDone)/max(elapsed.Seconds(), 1)))) +} + +// HumanBytes and HumanCount are the shared formatters, exported so the commands +// print the same shapes as the progress renderers do. +func HumanBytes(n int64) string { return humanBytes(n) } +func HumanCount(n int) string { return humanCount(n) } diff --git a/internal/report/summary.go b/internal/report/summary.go new file mode 100644 index 0000000..9b8ae45 --- /dev/null +++ b/internal/report/summary.go @@ -0,0 +1,61 @@ +package report + +import ( + "fmt" + "io" + + "github.com/tiennm99dev/telegram-exporter/internal/verify" +) + +// writeSurvey states what the chat holds and how much of it is already archived, +// broken out by why each outstanding file is outstanding. The single "N to +// fetch" it replaces hid the difference between never-fetched, empty, +// wrong-size, and unarchivable — which is the difference between a run that will +// converge and one that cannot. +func Survey(w io.Writer, r verify.Report) { + fmt.Fprintf(w, "\n chat holds %s media, %s\n", + humanCount(r.Expected), humanBytes(r.Bytes)) + fmt.Fprintf(w, " archived %s\n", humanCount(r.Present)) + + for _, row := range []struct { + label string + n int + }{ + {"never fetched", len(r.Absent) - len(r.Unsafe)}, + {"zero-byte", len(r.ZeroByte)}, + {"wrong size", len(r.Mismatched)}, + {"unarchivable", len(r.Unsafe)}, + } { + if row.n > 0 { + fmt.Fprintf(w, " %-13s %s\n", row.label, humanCount(row.n)) + } + } +} + +// PlanInfo is what the run is about to do, stated before it starts. +type PlanInfo struct { + Files int + Bytes int64 + Largest int64 + Budget int64 + Staging string + Threads int + Downloads, Uploads int + Destination string +} + +// writePlan prints the settings that decide how long the run takes and how much +// disk it uses, so an operator can stop it before a multi-hour transfer rather +// than discover the wrong cap partway through. +func Plan(w io.Writer, p PlanInfo) { + cap := "uncapped" + if p.Budget > 0 { + cap = humanBytes(p.Budget) + } + fmt.Fprintf(w, "\n fetching %s files, %s (largest %s)\n", + humanCount(p.Files), humanBytes(p.Bytes), humanBytes(p.Largest)) + fmt.Fprintf(w, " into %s\n", p.Destination) + fmt.Fprintf(w, " staging %s, capped at %s\n", p.Staging, cap) + fmt.Fprintf(w, " concurrency %d download(s) x %d thread(s), %d upload(s)\n\n", + p.Downloads, p.Threads, p.Uploads) +} diff --git a/internal/report/summary_test.go b/internal/report/summary_test.go new file mode 100644 index 0000000..692b6fd --- /dev/null +++ b/internal/report/summary_test.go @@ -0,0 +1,95 @@ +package report + +import ( + "strings" + "testing" + + "github.com/iyear/tdl/core/tmedia" + + "github.com/tiennm99dev/telegram-exporter/internal/pipeline" + "github.com/tiennm99dev/telegram-exporter/internal/tgsource" + "github.com/tiennm99dev/telegram-exporter/internal/verify" +) + +func TestSurveyBreaksOutdWhyFilesAreOutstanding(t *testing.T) { + r := verify.Report{Expected: 12000, Present: 11400, Bytes: 518 << 30} + r.Absent = append(r.Absent, 1, 2, 3) + r.ZeroByte = append(r.ZeroByte, 4) + r.Mismatched = append(r.Mismatched, verify.Mismatch{MessageID: 5}) + r.Unsafe = append(r.Unsafe, verify.Unsafe{MessageID: 3}) + + var sb strings.Builder + Survey(&sb, r) + out := sb.String() + + for _, want := range []string{"12,000 media", "11,400", "zero-byte", "wrong size", "unarchivable"} { + if !strings.Contains(out, want) { + t.Errorf("survey missing %q:\n%s", want, out) + } + } + // Unsafe ids are inside Absent, so counting both would double-count them and + // the rows would not add up to the outstanding total. + if !strings.Contains(out, "never fetched 2") { + t.Errorf("never-fetched should exclude the unarchivable id:\n%s", out) + } +} + +func TestSurveyOmitsEmptyRows(t *testing.T) { + var sb strings.Builder + Survey(&sb, verify.Report{Expected: 10, Present: 10}) + if strings.Contains(sb.String(), "zero-byte") { + t.Errorf("a clean archive should list no failure rows:\n%s", sb.String()) + } +} + +func TestPlanStatesTheCap(t *testing.T) { + var sb strings.Builder + Plan(&sb, PlanInfo{Files: 2613, Bytes: 79 << 30, Largest: 2 << 30, + Staging: "./staging", Threads: 4, Downloads: 2, Uploads: 2, Destination: "remote:x"}) + if !strings.Contains(sb.String(), "uncapped") { + t.Errorf("a run with no budget must say so:\n%s", sb.String()) + } + + sb.Reset() + Plan(&sb, PlanInfo{Files: 1, Budget: 40 << 30, Staging: "./staging", Destination: "remote:x"}) + if !strings.Contains(sb.String(), "40.0 GiB") { + t.Errorf("plan should state the cap:\n%s", sb.String()) + } +} + +// The redirected path must stay free of cursor movement: bars in a captured log +// are megabytes of control characters, which is what tdl's progress bar did to +// the shell pipeline's logs. +func TestReporterWritesNoAnsi(t *testing.T) { + var sb strings.Builder + r := newReporter(&sb, 3, 300) + it := tgsource.Item{MessageID: 1, Name: "a.mp4", Media: &tmedia.Media{Size: 100}} + + r.DownloadStart(it) + for i := range 500 { + r.DownloadBytes(it, int64(i)) + r.Stats(pipeline.Stats{Done: 1, BytesDone: int64(i)}) + } + r.DownloadDone(it, nil) + r.UploadStart(it) + r.UploadDone(it, nil) + r.Finish(pipeline.Stats{Done: 1, BytesDone: 100}) + + out := sb.String() + if strings.ContainsAny(out, "\r\033") { + t.Errorf("redirected output contains control characters: %q", out) + } + // A completed file is still recorded, so a captured run says what moved. + if !strings.Contains(out, "archived") || !strings.Contains(out, "a.mp4") { + t.Errorf("redirected output should record each archived file:\n%s", out) + } +} + +// Events picks the renderer from the writer: a strings.Builder is not a +// terminal, so it must never get bars. +func TestEventsChoosesTheLineRendererOffTerminal(t *testing.T) { + var sb strings.Builder + if _, ok := Events(&sb, 1, 1).(*Reporter); !ok { + t.Error("a non-terminal writer got the bar renderer") + } +}