radioscript

radioscript.go

// radioscript allows to capture redio stream (this is a mostly complete script, taken out of an not yet published project)
package main

import (
	"bufio"
	"crypto/sha1"
	"errors"
	"flag"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"os/signal"
	"path/filepath"
	"sort"
	"strings"
	"syscall"
	"time"

	"github.com/bogem/id3v2/v2"
)

var (
	// ErrMissingStreamURL indicates no stream URL was provided
	ErrMissingStreamURL = errors.New("stream URL is required")

	// ErrInterrupted indicates the recording was interrupted by user signal
	ErrInterrupted = errors.New("interrupted")

	// ErrMaxRetriesReached indicates maximum reconnection attempts were exhausted
	ErrMaxRetriesReached = errors.New("max retries reached")
)

const (
	bufferSize    = 32 * 1024
	fileExtension = ".mp3"
)

// Config holds all application configuration
type Config struct {
	StreamURL     string
	OutputDir     string
	SplitInterval time.Duration
	RetryDelay    time.Duration
	MaxRetries    int
	Verbose       bool
	ConfigDir     string
	ListStreams   bool
}

// NewConfigFromFlags creates a Config from command-line flags
func NewConfigFromFlags() *Config {
	cfg := &Config{}

	flag.StringVar(&cfg.StreamURL, "u", "", "URL of the radio stream, path to m3u file, or @stream-id from config")
	flag.StringVar(&cfg.OutputDir, "o", "recordings", "output directory for recordings")
	flag.DurationVar(&cfg.SplitInterval, "s", 0, "split recording into files at this interval (e.g., 1h, 30m, 0 for no split)")
	flag.DurationVar(&cfg.RetryDelay, "r", 5*time.Second, "delay between reconnection attempts")
	flag.IntVar(&cfg.MaxRetries, "x", 0, "maximum reconnection attempts (0 for infinite)")
	flag.BoolVar(&cfg.Verbose, "v", false, "be verbose")
	flag.StringVar(&cfg.ConfigDir, "c", "/etc/radioscript/streams.d", "config directory for stream definitions")
	flag.BoolVar(&cfg.ListStreams, "l", false, "list all configured streams and exit")

	flag.Parse()

	return cfg
}

// Validate checks if the configuration is valid
func (c *Config) Validate() error {
	if c.StreamURL == "" {
		log.Println("error: stream URL is required")
		log.Println("Usage: -u <stream-url|m3u-file|@stream-id>")
		log.Println("Example: -u @mdr-kultur")
		return ErrMissingStreamURL
	}
	return nil
}

func main() {
	cfg := NewConfigFromFlags()
	switch {
	case cfg.ListStreams:
		registry := NewStreamRegistry(cfg.ConfigDir)
		if err := registry.List(); err != nil {
			log.Fatal(err)
		}
		return
	default:
		if err := cfg.Validate(); err != nil {
			os.Exit(1)
		}
		registry := NewStreamRegistry(cfg.ConfigDir)
		stream, err := registry.Resolve(cfg.StreamURL)
		if err != nil {
			log.Fatal(err)
		}

		recorder := NewRecorder(stream, cfg)
		sigChan := make(chan os.Signal, 1)
		signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)

		if err := recorder.Start(sigChan); err != nil {
			log.Fatal(err)
		}
	}
}

// ==== Stream ====

// Stream represents a radio stream with optional name and URL
type Stream struct {
	Name string // Optional: friendly name for the stream
	URL  string // Required: the actual stream URL
}

// StreamRegistry handles loading and resolving stream configurations
type StreamRegistry struct {
	configDir string
	streams   map[string]string // streamName -> URL
	loaded    bool
}

// NewStreamRegistry creates a new stream registry
func NewStreamRegistry(configDir string) *StreamRegistry {
	return &StreamRegistry{
		configDir: configDir,
		streams:   make(map[string]string),
	}
}

// Resolve resolves an input string to a Stream
// Input can be:
// - @stream-name: references a configured stream
// - path/to/file.m3u: path to an M3U playlist file
// - http://...: direct stream URL
func (r *StreamRegistry) Resolve(input string) (Stream, error) {
	// Handle @stream-name notation
	if strings.HasPrefix(input, "@") {
		streamName := strings.TrimPrefix(input, "@")
		if err := r.ensureLoaded(); err != nil {
			return Stream{}, fmt.Errorf("loading stream configs: %w", err)
		}

		url, exists := r.streams[streamName]
		if !exists {
			return Stream{}, fmt.Errorf("stream '@%s' not found in config", streamName)
		}

		return Stream{Name: streamName, URL: url}, nil
	}

	// Handle M3U file
	if _, err := os.Stat(input); err == nil {
		url, err := parseM3U(input)
		if err != nil {
			return Stream{}, fmt.Errorf("parsing M3U file: %w", err)
		}
		return Stream{URL: url}, nil
	}

	// Assume direct URL
	return Stream{URL: input}, nil
}

// List prints all configured streams
func (r *StreamRegistry) List() error {
	if err := r.ensureLoaded(); err != nil {
		return err
	}

	if len(r.streams) == 0 {
		fmt.Println("No configured streams found.")
		return nil
	}

	// Sort stream names
	names := make([]string, 0, len(r.streams))
	for name := range r.streams {
		names = append(names, name)
	}
	sort.Strings(names)

	fmt.Printf("Configured streams (%d):\n\n", len(r.streams))
	for _, name := range names {
		fmt.Printf("  @%-20s %s\n", name, r.streams[name])
	}

	return nil
}

// ensureLoaded loads stream configs if not already loaded
func (r *StreamRegistry) ensureLoaded() error {
	if r.loaded {
		return nil
	}

	// Check if config directory exists
	if _, err := os.Stat(r.configDir); os.IsNotExist(err) {
		return fmt.Errorf("config directory does not exist: %s", r.configDir)
	}

	if err := r.loadAll(); err != nil {
		return err
	}

	r.loaded = true
	return nil
}

// loadAll loads all stream configurations from the config directory
func (r *StreamRegistry) loadAll() error {
	entries, err := os.ReadDir(r.configDir)
	if err != nil {
		return fmt.Errorf("reading config directory: %w", err)
	}

	// Collect and sort config files
	var filenames []string
	for _, entry := range entries {
		if entry.IsDir() {
			continue
		}
		name := entry.Name()
		// Support .txt and .conf files
		if strings.HasSuffix(name, ".txt") || strings.HasSuffix(name, ".conf") {
			filenames = append(filenames, name)
		}
	}
	sort.Strings(filenames)

	// Load each file (later files override earlier ones)
	for _, filename := range filenames {
		path := filepath.Join(r.configDir, filename)
		if err := r.loadFile(path); err != nil {
			fmt.Printf("warning: error loading %s: %v\n", filename, err)
			continue
		}
	}

	return nil
}

// loadFile loads streams from a single config file
func (r *StreamRegistry) loadFile(path string) error {
	file, err := os.Open(path)
	if err != nil {
		return err
	}
	defer file.Close()
	var (
		scanner = bufio.NewScanner(file)
		lineNum = 0
	)
	for scanner.Scan() {
		lineNum++
		var line = strings.TrimSpace(scanner.Text())
		if line == "" || strings.HasPrefix(line, "#") {
			continue
		}
		var parts = strings.SplitN(line, ",", 2)
		if len(parts) != 2 {
			fmt.Printf("warning: %s:%d: invalid format (expected: id,url)\n", path, lineNum)
			continue
		}
		var (
			streamName = strings.TrimSpace(parts[0])
			streamURL  = strings.TrimSpace(parts[1])
		)
		if streamName == "" || streamURL == "" {
			fmt.Printf("warning: %s:%d: empty id or url\n", path, lineNum)
			continue
		}
		r.streams[streamName] = streamURL
	}
	return scanner.Err()
}

// parseM3U extracts the first valid URL from an M3U playlist file
func parseM3U(filename string) (string, error) {
	file, err := os.Open(filename)
	if err != nil {
		return "", fmt.Errorf("opening m3u file: %w", err)
	}
	defer file.Close()
	scanner := bufio.NewScanner(file)
	for scanner.Scan() {
		line := strings.TrimSpace(scanner.Text())
		if line == "" || strings.HasPrefix(line, "#") {
			continue
		}
		return line, nil
	}
	if err := scanner.Err(); err != nil {
		return "", err
	}
	return "", fmt.Errorf("no stream URL found in m3u file")
}

// ==== Recorder ====

// Recorder handles the recording of a radio stream
type Recorder struct {
	stream       Stream
	outputDir    string
	cfg          *Config
	sessionStart time.Time
	fileNumber   int
	retryCount   int
}

// NewRecorder creates a new Recorder instance
func NewRecorder(stream Stream, cfg *Config) *Recorder {
	// Use stream name for output directory if available
	outputDir := cfg.OutputDir
	if stream.Name != "" {
		outputDir = filepath.Join(cfg.OutputDir, stream.Name)
	}

	return &Recorder{
		stream:       stream,
		outputDir:    outputDir,
		cfg:          cfg,
		sessionStart: time.Now(),
		fileNumber:   1,
	}
}

// Start begins recording the stream
func (r *Recorder) Start(sigChan chan os.Signal) error {
	if err := os.MkdirAll(r.outputDir, 0755); err != nil {
		return err
	}

	r.logSessionInfo()

	var resp *http.Response

	for {
		filename := r.generateFilename()
		segmentStart := time.Now()

		outFile, err := os.Create(filename)
		if err != nil {
			return err
		}
		log.Printf("writing to %s", filename)

		if resp == nil {
			resp, err = r.connect()
			if err != nil {
				_ = outFile.Close()
				return err
			}
		}

		totalBytes, err := r.writeSegment(outFile, resp, sigChan)
		segmentEnd := time.Now()

		// Always sync and close the file
		if syncErr := outFile.Sync(); syncErr != nil {
			log.Printf("warning: failed to sync file: %v", syncErr)
		}
		_ = outFile.Close()

		// Write metadata if we recorded anything
		if totalBytes > 0 {
			metadata := RecordingMetadata{
				StreamName:   r.stream.Name,
				StreamURL:    r.stream.URL,
				StartTime:    segmentStart,
				EndTime:      segmentEnd,
				SessionStart: r.sessionStart,
				FileNumber:   r.fileNumber,
			}
			if metaErr := writeMetadata(filename, metadata); metaErr != nil {
				log.Printf("error: failed to write metadata: %v", metaErr)
			}
		}

		// Handle errors
		if err != nil {
			if resp != nil {
				_ = resp.Body.Close()
			}
			return err
		}

		r.retryCount = 0
		r.fileNumber++
	}
}

// logSessionInfo logs information about the recording session
func (r *Recorder) logSessionInfo() {
	if r.stream.Name != "" {
		log.Printf("stream: %s", r.stream.Name)
	}
	log.Printf("url: %s", r.stream.URL)
	log.Printf("dir: %s", r.outputDir)
	if r.cfg.SplitInterval > 0 {
		log.Printf("split interval: %s", r.cfg.SplitInterval)
	}
}

// writeSegment writes one segment (file) of the stream
func (r *Recorder) writeSegment(outFile *os.File, resp *http.Response, sigChan chan os.Signal) (int64, error) {
	var (
		segmentStart = time.Now()
		totalBytes   = int64(0)
		lastUpdate   = time.Now()
		buffer       = make([]byte, bufferSize)
	)

	for {
		select {
		case <-sigChan:
			return totalBytes, ErrInterrupted

		default:
			// Check if we should split
			if r.cfg.SplitInterval > 0 && time.Since(segmentStart) >= r.cfg.SplitInterval {
				return totalBytes, nil
			}

			n, err := resp.Body.Read(buffer)
			if n > 0 {
				if _, writeErr := outFile.Write(buffer[:n]); writeErr != nil {
					_ = resp.Body.Close()
					return totalBytes, writeErr
				}

				totalBytes += int64(n)

				// Update display
				if r.cfg.Verbose && time.Since(lastUpdate) >= time.Second {
					r.displayProgress(segmentStart, totalBytes)
					lastUpdate = time.Now()
				}
			}

			// Handle read errors
			if err != nil {
				_ = resp.Body.Close()

				if err == io.EOF || isNetworkError(err) {
					if r.cfg.MaxRetries > 0 && r.retryCount >= r.cfg.MaxRetries {
						return totalBytes, ErrMaxRetriesReached
					}

					time.Sleep(r.cfg.RetryDelay)
					r.retryCount++

					newResp, connErr := r.connect()
					if connErr != nil {
						return totalBytes, connErr
					}

					// Replace the response and continue
					resp = newResp
					continue
				}

				return totalBytes, err
			}
		}
	}
}

// connect establishes a connection to the stream
func (r *Recorder) connect() (*http.Response, error) {
	if r.retryCount > 0 {
		log.Printf("reconnection attempt %d...", r.retryCount)
	}

	client := &http.Client{}
	req, err := http.NewRequest("GET", r.stream.URL, nil)
	if err != nil {
		return nil, err
	}
	req.Header.Set("User-Agent", "radioscript/0.1.0")

	resp, err := client.Do(req)
	if err != nil {
		return nil, err
	}

	if resp.StatusCode != http.StatusOK {
		_ = resp.Body.Close()
		return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
	}

	return resp, nil
}

// generateFilename creates a timestamped filename for a segment
func (r *Recorder) generateFilename() string {
	timestamp := time.Now().Format("20060102-150405")
	var (
		name = "any"
		h    = sha1.New()
	)
	_, _ = h.Write([]byte(r.cfg.StreamURL))
	if r.stream.Name != "" {
		name = r.stream.Name
	}
	return filepath.Join(r.outputDir, fmt.Sprintf("stream-%s-%x-%s%s", name, h.Sum(nil), timestamp, fileExtension))
}

// displayProgress shows current recording progress
func (r *Recorder) displayProgress(segmentStart time.Time, totalBytes int64) {
	var (
		segmentDuration = time.Since(segmentStart)
		sessionDuration = time.Since(r.sessionStart)
		rate            = float64(totalBytes) / segmentDuration.Seconds() / 1024

		statusLine = fmt.Sprintf("\rfile %d | segment: %s | session: %s | %.2f KB/s | %.2f MB",
			r.fileNumber,
			formatDuration(segmentDuration),
			formatDuration(sessionDuration),
			rate,
			float64(totalBytes)/(1024*1024))
	)

	if r.cfg.SplitInterval > 0 {
		remaining := r.cfg.SplitInterval - segmentDuration
		if remaining > 0 {
			statusLine += fmt.Sprintf(" | next split: %s", formatDuration(remaining))
		}
	}
	fmt.Print(statusLine)
}

// RecordingMetadata contains metadata about a recording segment
type RecordingMetadata struct {
	StreamName   string
	StreamURL    string
	StartTime    time.Time
	EndTime      time.Time
	SessionStart time.Time
	FileNumber   int
}

// writeMetadata writes ID3 tags to the recorded file with comprehensive metadata
func writeMetadata(filename string, meta RecordingMetadata) error {
	tag, err := id3v2.Open(filename, id3v2.Options{Parse: true})
	if err != nil {
		return fmt.Errorf("opening file for tagging: %w", err)
	}
	defer tag.Close()

	tag.SetVersion(3)
	tag.SetDefaultEncoding(id3v2.EncodingUTF8)

	// Title: include stream name if available
	title := fmt.Sprintf("radio recording %s", meta.StartTime.Format("2006-01-02 15:04:05"))
	if meta.StreamName != "" {
		title = fmt.Sprintf("%s - %s", meta.StreamName, title)
	}
	tag.SetTitle(title)

	// Artist: the stream URL
	tag.SetArtist(meta.StreamURL)

	// Album: session identifier
	tag.SetAlbum(fmt.Sprintf("session %s", meta.SessionStart.Format("2006-01-02 15:04:05")))

	// Year
	tag.SetYear(meta.StartTime.Format("2006"))

	// Genre
	tag.SetGenre("Other")

	// Track number
	tag.AddTextFrame(tag.CommonID("Track number"), id3v2.EncodingUTF8, fmt.Sprintf("%d", meta.FileNumber))

	// Comments with structured metadata
	duration := meta.EndTime.Sub(meta.StartTime)

	// Source comment (stream URL)
	tag.AddCommentFrame(id3v2.CommentFrame{
		Encoding:    id3v2.EncodingUTF8,
		Language:    "eng",
		Description: "source",
		Text:        meta.StreamURL,
	})

	// Recording start time
	tag.AddCommentFrame(id3v2.CommentFrame{
		Encoding:    id3v2.EncodingUTF8,
		Language:    "eng",
		Description: "recording_start",
		Text:        meta.StartTime.Format(time.RFC3339),
	})

	// Recording end time
	tag.AddCommentFrame(id3v2.CommentFrame{
		Encoding:    id3v2.EncodingUTF8,
		Language:    "eng",
		Description: "recording_end",
		Text:        meta.EndTime.Format(time.RFC3339),
	})

	// Duration
	tag.AddCommentFrame(id3v2.CommentFrame{
		Encoding:    id3v2.EncodingUTF8,
		Language:    "eng",
		Description: "duration",
		Text:        duration.String(),
	})

	// Stream name if available
	if meta.StreamName != "" {
		tag.AddCommentFrame(id3v2.CommentFrame{
			Encoding:    id3v2.EncodingUTF8,
			Language:    "eng",
			Description: "stream_name",
			Text:        meta.StreamName,
		})
	}

	// Software identifier
	tag.AddCommentFrame(id3v2.CommentFrame{
		Encoding:    id3v2.EncodingUTF8,
		Language:    "eng",
		Description: "software",
		Text:        "radioscript/0.1.0",
	})

	if err := tag.Save(); err != nil {
		return err
	}

	return nil
}

// isNetworkError checks if an error is network-related
func isNetworkError(err error) bool {
	errStr := err.Error()
	return strings.Contains(errStr, "connection") ||
		strings.Contains(errStr, "network") ||
		strings.Contains(errStr, "timeout")
}

// formatDuration formats a duration as HH:MM:SS
func formatDuration(d time.Duration) string {
	d = d.Round(time.Second)
	h := d / time.Hour
	d -= h * time.Hour
	m := d / time.Minute
	d -= m * time.Minute
	s := d / time.Second
	return fmt.Sprintf("%02d:%02d:%02d", h, m, s)
}