opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: replace premature media file cleanup with background MediaCleaner

Browse source 
Channel handlers delete downloaded media files via defer os.Remove()
before the async agent loop consumer can read them, causing silent
failures in voice/image/document processing.

Remove the immediate defer cleanup from telegram, line, slack, and
onebot handlers. Introduce a background MediaCleaner goroutine that
periodically removes files older than 30 minutes from the temp media
directory, preventing unbounded accumulation.

discord.go is unchanged because it processes files synchronously.

Closes #619

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
ex-takashima 2026-02-22 14:19:59 +09:00
parent c6865fe852
commit a56d97f233
7 changed files with 164 additions and 77 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
cmd/picoclaw/cmd_gateway.go
View file

@ -25,6 +25,7 @@ import (
"github.com/sipeed/picoclaw/pkg/providers"
"github.com/sipeed/picoclaw/pkg/state"
"github.com/sipeed/picoclaw/pkg/tools"
"github.com/sipeed/picoclaw/pkg/utils"
"github.com/sipeed/picoclaw/pkg/voice"
)
@ -192,6 +193,9 @@ func gatewayCmd() {
fmt.Println("✓ Device event service started")
}
mediaCleaner := utils.NewMediaCleaner()
mediaCleaner.Start()
if err := channelManager.StartAll(ctx); err != nil {
fmt.Printf("Error starting channels: %v\n", err)
}
@ -216,6 +220,7 @@ func gatewayCmd() {
deviceService.Stop()
heartbeatService.Stop()
cronService.Stop()
mediaCleaner.Stop()
agentLoop.Stop()
channelManager.StopAll(ctx)
fmt.Println("✓ Gateway stopped")

22
pkg/channels/line.go
View file

@ -10,7 +10,6 @@ import (
"fmt"
"io"
"net/http"
"os"
"strings"
"sync"
"time"
@ -307,18 +306,6 @@ func (c *LINEChannel) processEvent(event lineEvent) {
var content string
var mediaPaths []string
localFiles := []string{}
defer func() {
for _, file := range localFiles {
if err := os.Remove(file); err != nil {
logger.DebugCF("line", "Failed to cleanup temp file", map[string]any{
"file": file,
"error": err.Error(),
})
}
}
}()
switch msg.Type {
case "text":
@ -330,22 +317,19 @@ func (c *LINEChannel) processEvent(event lineEvent) {
case "image":
localPath := c.downloadContent(msg.ID, "image.jpg")
if localPath != "" {
localFiles = append(localFiles, localPath)
mediaPaths = append(mediaPaths, localPath)
mediaPaths = append(mediaPaths, localPath)
content = "[image]"
}
case "audio":
localPath := c.downloadContent(msg.ID, "audio.m4a")
if localPath != "" {
localFiles = append(localFiles, localPath)
mediaPaths = append(mediaPaths, localPath)
mediaPaths = append(mediaPaths, localPath)
content = "[audio]"
}
case "video":
localPath := c.downloadContent(msg.ID, "video.mp4")
if localPath != "" {
localFiles = append(localFiles, localPath)
mediaPaths = append(mediaPaths, localPath)
mediaPaths = append(mediaPaths, localPath)
content = "[video]"
}
case "file":

20
pkg/channels/onebot.go
View file

@ -4,7 +4,6 @@ import (
"context"
"encoding/json"
"fmt"
"os"
"strconv"
"strings"
"sync"
@ -571,7 +570,6 @@ type parseMessageResult struct {
Text string
IsBotMentioned bool
Media []string
LocalFiles []string
ReplyTo string
}
@ -603,7 +601,6 @@ func (c *OneBotChannel) parseMessageSegments(raw json.RawMessage, selfID int64)
mentioned := false
selfIDStr := strconv.FormatInt(selfID, 10)
var media []string
var localFiles []string
var replyTo string
for _, seg := range segments {
@ -642,7 +639,6 @@ func (c *OneBotChannel) parseMessageSegments(raw json.RawMessage, selfID int64)
})
if localPath != "" {
media = append(media, localPath)
localFiles = append(localFiles, localPath)
textParts = append(textParts, fmt.Sprintf("[%s]", segType))
}
}
@ -656,7 +652,6 @@ func (c *OneBotChannel) parseMessageSegments(raw json.RawMessage, selfID int64)
LoggerPrefix: "onebot",
})
if localPath != "" {
localFiles = append(localFiles, localPath)
if c.transcriber != nil && c.transcriber.IsAvailable() {
tctx, tcancel := context.WithTimeout(c.ctx, 30*time.Second)
result, err := c.transcriber.Transcribe(tctx, localPath)
@ -703,7 +698,6 @@ func (c *OneBotChannel) parseMessageSegments(raw json.RawMessage, selfID int64)
Text: strings.TrimSpace(strings.Join(textParts, "")),
IsBotMentioned: mentioned,
Media: media,
LocalFiles: localFiles,
ReplyTo: replyTo,
}
}
@ -824,20 +818,6 @@ func (c *OneBotChannel) handleMessage(raw *oneBotRawEvent) {
}
}
// Clean up temp files when done
if len(parsed.LocalFiles) > 0 {
defer func() {
for _, f := range parsed.LocalFiles {
if err := os.Remove(f); err != nil {
logger.DebugCF("onebot", "Failed to remove temp file", map[string]any{
"path": f,
"error": err.Error(),
})
}
}
}()
}
if c.isDuplicate(messageID) {
logger.DebugCF("onebot", "Duplicate message, skipping", map[string]any{
"message_id": messageID,

17
pkg/channels/slack.go
View file

@ -3,7 +3,6 @@ package channels
import (
"context"
"fmt"
"os"
"strings"
"sync"
"time"
@ -232,19 +231,6 @@ func (c *SlackChannel) handleMessageEvent(ev *slackevents.MessageEvent) {
content = c.stripBotMention(content)
var mediaPaths []string
localFiles := []string{} // 跟踪需要清理的本地文件
// 确保临时文件在函数返回时被清理
defer func() {
for _, file := range localFiles {
if err := os.Remove(file); err != nil {
logger.DebugCF("slack", "Failed to cleanup temp file", map[string]any{
"file": file,
"error": err.Error(),
})
}
}
}()
if ev.Message != nil && len(ev.Message.Files) > 0 {
for _, file := range ev.Message.Files {
@ -252,8 +238,7 @@ func (c *SlackChannel) handleMessageEvent(ev *slackevents.MessageEvent) {
if localPath == "" {
continue
}
localFiles = append(localFiles, localPath)
mediaPaths = append(mediaPaths, localPath)
mediaPaths = append(mediaPaths, localPath)
if utils.IsAudioFile(file.Name, file.Mimetype) && c.transcriber != nil && c.transcriber.IsAvailable() {
ctx, cancel := context.WithTimeout(c.ctx, 30*time.Second)

25
pkg/channels/telegram.go
View file

@ -221,19 +221,6 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
content := ""
mediaPaths := []string{}
localFiles := []string{} // 跟踪需要清理的本地文件
// 确保临时文件在函数返回时被清理
defer func() {
for _, file := range localFiles {
if err := os.Remove(file); err != nil {
logger.DebugCF("telegram", "Failed to cleanup temp file", map[string]any{
"file": file,
"error": err.Error(),
})
}
}
}()
if message.Text != "" {
content += message.Text
@ -250,8 +237,7 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
photo := message.Photo[len(message.Photo)-1]
photoPath := c.downloadPhoto(ctx, photo.FileID)
if photoPath != "" {
localFiles = append(localFiles, photoPath)
mediaPaths = append(mediaPaths, photoPath)
mediaPaths = append(mediaPaths, photoPath)
if content != "" {
content += "\n"
}
@ -262,8 +248,7 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
if message.Voice != nil {
voicePath := c.downloadFile(ctx, message.Voice.FileID, ".ogg")
if voicePath != "" {
localFiles = append(localFiles, voicePath)
mediaPaths = append(mediaPaths, voicePath)
mediaPaths = append(mediaPaths, voicePath)
transcribedText := ""
if c.transcriber != nil && c.transcriber.IsAvailable() {
@ -297,8 +282,7 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
if message.Audio != nil {
audioPath := c.downloadFile(ctx, message.Audio.FileID, ".mp3")
if audioPath != "" {
localFiles = append(localFiles, audioPath)
mediaPaths = append(mediaPaths, audioPath)
mediaPaths = append(mediaPaths, audioPath)
if content != "" {
content += "\n"
}
@ -309,8 +293,7 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
if message.Document != nil {
docPath := c.downloadFile(ctx, message.Document.FileID, "")
if docPath != "" {
localFiles = append(localFiles, docPath)
mediaPaths = append(mediaPaths, docPath)
mediaPaths = append(mediaPaths, docPath)
if content != "" {
content += "\n"
}

86
pkg/utils/media.go
View file

@ -6,6 +6,7 @@ import (
"os"
"path/filepath"
"strings"
"sync"
"time"
"github.com/google/uuid"
@ -13,6 +14,9 @@ import (
"github.com/sipeed/picoclaw/pkg/logger"
)
// MediaDir is the subdirectory name under os.TempDir() where downloaded media files are stored.
const MediaDir = "picoclaw_media"
// IsAudioFile checks if a file is an audio file based on its filename extension and content type.
func IsAudioFile(filename, contentType string) bool {
audioExtensions := []string{".mp3", ".wav", ".ogg", ".m4a", ".flac", ".aac", ".wma"}
@ -65,7 +69,7 @@ func DownloadFile(url, filename string, opts DownloadOptions) string {
opts.LoggerPrefix = "utils"
}
mediaDir := filepath.Join(os.TempDir(), "picoclaw_media")
mediaDir := filepath.Join(os.TempDir(), MediaDir)
if err := os.MkdirAll(mediaDir, 0o700); err != nil {
logger.ErrorCF(opts.LoggerPrefix, "Failed to create media directory", map[string]any{
"error": err.Error(),
@ -141,3 +145,83 @@ func DownloadFileSimple(url, filename string) string {
LoggerPrefix: "media",
})
}
// MediaCleaner periodically removes old files from the media temp directory.
type MediaCleaner struct {
interval time.Duration
maxAge time.Duration
stop chan struct{}
once sync.Once
}
// NewMediaCleaner creates a new MediaCleaner with default settings
// (scan every 5 minutes, remove files older than 30 minutes).
func NewMediaCleaner() *MediaCleaner {
return &MediaCleaner{
interval: 5 * time.Minute,
maxAge: 30 * time.Minute,
stop: make(chan struct{}),
}
}
// Start begins the background cleanup goroutine. Safe to call multiple times.
func (mc *MediaCleaner) Start() {
mc.once.Do(func() {
go mc.loop()
logger.InfoC("media", "Media cleaner started")
})
}
// Stop signals the cleanup goroutine to exit. Safe to call multiple times.
func (mc *MediaCleaner) Stop() {
select {
case <-mc.stop:
default:
close(mc.stop)
logger.InfoC("media", "Media cleaner stopped")
}
}
func (mc *MediaCleaner) loop() {
ticker := time.NewTicker(mc.interval)
defer ticker.Stop()
for {
select {
case <-mc.stop:
return
case <-ticker.C:
mc.cleanup()
}
}
}
func (mc *MediaCleaner) cleanup() {
mediaDir := filepath.Join(os.TempDir(), MediaDir)
entries, err := os.ReadDir(mediaDir)
if err != nil {
return
}
now := time.Now()
removed := 0
for _, entry := range entries {
if entry.IsDir() {
continue
}
info, err := entry.Info()
if err != nil {
continue
}
if now.Sub(info.ModTime()) > mc.maxAge {
path := filepath.Join(mediaDir, entry.Name())
if err := os.Remove(path); err == nil {
removed++
}
}
}
if removed > 0 {
logger.DebugCF("media", "Cleaned up old media files", map[string]any{
"removed": removed,
})
}
}

66
pkg/utils/media_test.go Normal file
View file

@ -0,0 +1,66 @@
package utils
import (
"os"
"path/filepath"
"testing"
"time"
)
func TestMediaCleanerRemovesOldFiles(t *testing.T) {
// Setup: create a temp media directory with old and new files
mediaDir := filepath.Join(os.TempDir(), MediaDir)
if err := os.MkdirAll(mediaDir, 0700); err != nil {
t.Fatalf("failed to create media dir: %v", err)
}
// Create an "old" file and backdate its modification time
oldFile := filepath.Join(mediaDir, "test_old_file.jpg")
if err := os.WriteFile(oldFile, []byte("old"), 0600); err != nil {
t.Fatalf("failed to create old file: %v", err)
}
oldTime := time.Now().Add(-1 * time.Hour)
if err := os.Chtimes(oldFile, oldTime, oldTime); err != nil {
t.Fatalf("failed to set old file time: %v", err)
}
// Create a "new" file (just created, so modtime is now)
newFile := filepath.Join(mediaDir, "test_new_file.jpg")
if err := os.WriteFile(newFile, []byte("new"), 0600); err != nil {
t.Fatalf("failed to create new file: %v", err)
}
// Cleanup test files at end
defer os.Remove(oldFile)
defer os.Remove(newFile)
// Run cleanup directly
mc := NewMediaCleaner()
mc.cleanup()
// Old file should be gone
if _, err := os.Stat(oldFile); !os.IsNotExist(err) {
t.Errorf("expected old file to be removed, but it still exists")
}
// New file should still exist
if _, err := os.Stat(newFile); err != nil {
t.Errorf("expected new file to still exist, got error: %v", err)
}
}
func TestMediaCleanerStartStop(t *testing.T) {
mc := NewMediaCleaner()
// Start should not panic
mc.Start()
// Second Start should be idempotent (sync.Once)
mc.Start()
// Stop should not panic
mc.Stop()
// Second Stop should be idempotent
mc.Stop()
}