diff --git a/PKGBUILD b/PKGBUILD index c8c63a9..14887e0 100644 --- a/PKGBUILD +++ b/PKGBUILD @@ -1,6 +1,6 @@ # Maintainer: Misti pkgname=compressor-git -pkgver=r6.0d0d2ea +pkgver=r7.ba24d78 pkgrel=1 pkgdesc="Video compressor service that watches input directory and compresses videos using ffmpeg" arch=('x86_64' 'aarch64') diff --git a/cmd/compressor/config.go b/cmd/compressor/config.go index 7acf114..4695e3e 100644 --- a/cmd/compressor/config.go +++ b/cmd/compressor/config.go @@ -30,34 +30,36 @@ const defaultFFMPEGCommandCPU = "-y -i {{input}} -c:v libx265 -preset slow -crf var defaultExtensions = []string{".mp4", ".mkv", ".mov", ".avi", ".flv", ".wmv", ".m4v", ".webm", ".ts"} type config struct { - inputDir string - outputDir string - ffmpegBinary string - ffmpegCommand string - deleteSource bool - processingSuffix string - outputExtension string - httpPort string - rescanInterval time.Duration - stabilityWindow time.Duration - queueSize int - maxConcurrent int - extensions map[string]struct{} + inputDir string + outputDir string + ffmpegBinary string + ffmpegCommand string + deleteSource bool + processingSuffix string + outputExtension string + httpPort string + discordWebhookURL string + rescanInterval time.Duration + stabilityWindow time.Duration + queueSize int + maxConcurrent int + extensions map[string]struct{} } func loadConfig() (config, error) { cfg := config{ - inputDir: getEnv("INPUT_DIR", defaultInputDir), - outputDir: getEnv("OUTPUT_DIR", defaultOutputDir), - ffmpegBinary: getEnv("FFMPEG_BIN", "ffmpeg"), - processingSuffix: getEnv("PROCESSING_SUFFIX", defaultProcessingSuffix), - outputExtension: getEnv("OUTPUT_EXTENSION", defaultOutputExtension), - httpPort: getEnvOrEmpty("PORT"), - queueSize: getEnvInt("QUEUE_SIZE", defaultQueueSize), - maxConcurrent: getEnvInt("MAX_CONCURRENT", defaultMaxConcurrent), - rescanInterval: getEnvDuration("RESCAN_INTERVAL", defaultRescanInterval), - stabilityWindow: getEnvDuration("FILE_STABILITY_DURATION", defaultStabilityDuration), - deleteSource: getEnvBool("DELETE_SOURCE"), + inputDir: getEnv("INPUT_DIR", defaultInputDir), + outputDir: getEnv("OUTPUT_DIR", defaultOutputDir), + ffmpegBinary: getEnv("FFMPEG_BIN", "ffmpeg"), + processingSuffix: getEnv("PROCESSING_SUFFIX", defaultProcessingSuffix), + outputExtension: getEnv("OUTPUT_EXTENSION", defaultOutputExtension), + httpPort: getEnvOrEmpty("PORT"), + discordWebhookURL: getEnvOrEmpty("DISCORD_WEBHOOK_URL"), + queueSize: getEnvInt("QUEUE_SIZE", defaultQueueSize), + maxConcurrent: getEnvInt("MAX_CONCURRENT", defaultMaxConcurrent), + rescanInterval: getEnvDuration("RESCAN_INTERVAL", defaultRescanInterval), + stabilityWindow: getEnvDuration("FILE_STABILITY_DURATION", defaultStabilityDuration), + deleteSource: getEnvBool("DELETE_SOURCE"), } if cfg.maxConcurrent < 1 { diff --git a/cmd/compressor/discord.go b/cmd/compressor/discord.go new file mode 100644 index 0000000..970e4dd --- /dev/null +++ b/cmd/compressor/discord.go @@ -0,0 +1,128 @@ +package main + +import ( + "bytes" + "encoding/json" + "fmt" + "log" + "net/http" + "path/filepath" + "time" +) + +type DiscordWebhook struct { + URL string +} + +type DiscordEmbed struct { + Title string `json:"title"` + Description string `json:"description"` + Color int `json:"color"` + Fields []DiscordEmbedField `json:"fields"` + Timestamp string `json:"timestamp"` +} + +type DiscordEmbedField struct { + Name string `json:"name"` + Value string `json:"value"` + Inline bool `json:"inline"` +} + +type DiscordMessage struct { + Embeds []DiscordEmbed `json:"embeds"` +} + +func sendDiscordSuccess(webhookURL, fileName string, originalSize, compressedSize int64) { + if webhookURL == "" { + return + } + + embed := DiscordEmbed{ + Title: "✅ Compression Successful", + Description: fmt.Sprintf("compressed: **%s**", filepath.Base(fileName)), + Color: 0x00ff00, // Green + Fields: []DiscordEmbedField{ + { + Name: "Original Size", + Value: formatFileSize(originalSize), + Inline: true, + }, + { + Name: "Compressed Size", + Value: formatFileSize(compressedSize), + Inline: true, + }, + { + Name: "Space Saved", + Value: formatFileSize(originalSize - compressedSize), + Inline: true, + }, + { + Name: "Compression Ratio", + Value: fmt.Sprintf("%.1f%%", float64(compressedSize)/float64(originalSize)*100), + Inline: true, + }, + }, + Timestamp: time.Now().Format(time.RFC3339), + } + + sendDiscordMessage(webhookURL, embed) +} + +func sendDiscordFailure(webhookURL, fileName, errorMsg string) { + if webhookURL == "" { + return + } + + embed := DiscordEmbed{ + Title: "❌ Compression Failed", + Description: fmt.Sprintf("Failed to compress **%s**", filepath.Base(fileName)), + Color: 0xff0000, // Red + Fields: []DiscordEmbedField{ + { + Name: "Error", + Value: errorMsg, + Inline: false, + }, + }, + Timestamp: time.Now().Format(time.RFC3339), + } + + sendDiscordMessage(webhookURL, embed) +} + +func sendDiscordMessage(webhookURL string, embed DiscordEmbed) { + message := DiscordMessage{ + Embeds: []DiscordEmbed{embed}, + } + + jsonData, err := json.Marshal(message) + if err != nil { + log.Printf("Failed to marshal Discord message: %v", err) + return + } + + resp, err := http.Post(webhookURL, "application/json", bytes.NewBuffer(jsonData)) + if err != nil { + log.Printf("Failed to send Discord webhook: %v", err) + return + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusOK { + log.Printf("Discord webhook returned status %d", resp.StatusCode) + } +} + +func formatFileSize(bytes int64) string { + const unit = 1024 + if bytes < unit { + return fmt.Sprintf("%d B", bytes) + } + div, exp := int64(unit), 0 + for n := bytes / unit; n >= unit; n /= unit { + div *= unit + exp++ + } + return fmt.Sprintf("%.1f %cB", float64(bytes)/float64(div), "KMGTPE"[exp]) +} diff --git a/cmd/compressor/main.go b/cmd/compressor/main.go index 7fc1a2a..82f8eca 100644 --- a/cmd/compressor/main.go +++ b/cmd/compressor/main.go @@ -34,6 +34,10 @@ func main() { if cfg.httpPort == "" { log.Printf(" HTTP server disabled") } + log.Printf(" Discord Webhook: %s", cfg.discordWebhookURL) + if cfg.discordWebhookURL == "" { + log.Printf(" Discord notifications disabled") + } log.Printf(" Rescan Interval: %v", cfg.rescanInterval) log.Printf(" Stability Window: %v", cfg.stabilityWindow) log.Printf(" Queue Size: %d", cfg.queueSize) diff --git a/cmd/compressor/processor.go b/cmd/compressor/processor.go index eee22e2..b7f3cda 100644 --- a/cmd/compressor/processor.go +++ b/cmd/compressor/processor.go @@ -16,10 +16,18 @@ import ( ) func processFile(ctx context.Context, cfg config, originalPath string) error { + // Get original file size for Discord notifications + originalInfo, err := os.Stat(originalPath) + if err != nil { + return fmt.Errorf("stat original file: %w", err) + } + originalSize := originalInfo.Size() + if err := waitForStability(ctx, originalPath, cfg.stabilityWindow); err != nil { if errors.Is(err, os.ErrNotExist) { return nil } + sendDiscordFailure(cfg.discordWebhookURL, originalPath, fmt.Sprintf("stability check: %v", err)) return fmt.Errorf("stability check: %w", err) } @@ -28,6 +36,7 @@ func processFile(ctx context.Context, cfg config, originalPath string) error { if errors.Is(err, os.ErrNotExist) { return nil } + sendDiscordFailure(cfg.discordWebhookURL, originalPath, fmt.Sprintf("rename for processing: %v", err)) return fmt.Errorf("rename for processing: %w", err) } @@ -54,10 +63,12 @@ func processFile(ctx context.Context, cfg config, originalPath string) error { outputPath, err := buildOutputPath(cfg, originalPath) if err != nil { + sendDiscordFailure(cfg.discordWebhookURL, originalPath, fmt.Sprintf("build output path: %v", err)) return err } if err := runFFMPEG(ctx, cfg, processingPath, outputPath); err != nil { + sendDiscordFailure(cfg.discordWebhookURL, originalPath, err.Error()) if removeErr := os.Remove(outputPath); removeErr != nil && !errors.Is(removeErr, os.ErrNotExist) { log.Printf("remove partial output %s failed: %v", outputPath, removeErr) } @@ -66,6 +77,13 @@ func processFile(ctx context.Context, cfg config, originalPath string) error { success = true log.Printf("processed %s -> %s", originalPath, outputPath) + + // Send Discord success notification + if compressedInfo, err := os.Stat(outputPath); err == nil { + compressedSize := compressedInfo.Size() + sendDiscordSuccess(cfg.discordWebhookURL, originalPath, originalSize, compressedSize) + } + processed.Store(originalPath, time.Now()) return nil } diff --git a/compressor.service b/compressor.service index b1296da..ad3e22d 100644 --- a/compressor.service +++ b/compressor.service @@ -25,6 +25,8 @@ Environment=OUTPUT_DIR=/srv/syncthing/compressor_output # Environment=FILE_STABILITY_DURATION=3s # HTTP server # Environment=PORT=8080 +# Discord notifications +# Environment=DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/YOUR_WEBHOOK_ID/YOUR_WEBHOOK_TOKEN Restart=always RestartSec=5