admin/TaskPool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: cron task style fix

Browse Source
This commit is contained in:
engigu
2025-12-21 14:30:59 +08:00
parent f0beb8a654
commit d8b5b75eaf
11 changed files with 174 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/database/migrate.go
+1
View File
@@ -13,5 +13,6 @@ func Migrate() error {
&models.EnvironmentVariable{},
&models.Setting{},
&models.LoginLog{},
&models.SendStats{},
)
}
internal/models/send_stats.go
+20
View File
@@ -0,0 +1,20 @@
package models
import (
"time"
"baihu/internal/constant"
)
// SendStats 任务执行统计
type SendStats struct {
ID uint `json:"id" gorm:"primaryKey"`
TaskID uint `json:"task_id" gorm:"index"`
Status string `json:"status" gorm:"size:20;not null"` // success, failed
Num int `json:"num" gorm:"default:0"`
CreatedAt time.Time `json:"created_at"`
}
func (SendStats) TableName() string {
return constant.TablePrefix + "send_stats"
}
internal/router/router.go
+3 -3
View File
@@ -151,7 +151,7 @@ func Setup(c *Controllers) *gin.Engine {
files.POST("/delete", c.File.DeleteFile)
files.POST("/rename", c.File.RenameFile)
files.POST("/upload", c.File.UploadArchive)
files.POST("/upload-files", c.File.UploadFiles)
files.POST("/uploadfiles", c.File.UploadFiles)
}
// Log routes
@@ -169,11 +169,11 @@ func Setup(c *Controllers) *gin.Engine {
settings := authorized.Group("/settings")
{
settings.POST("/password", c.Settings.ChangePassword)
settings.POST("/clean-logs", c.Settings.CleanLogs)
settings.POST("/cleanlogs", c.Settings.CleanLogs)
settings.GET("/site", c.Settings.GetSiteSettings)
settings.PUT("/site", c.Settings.UpdateSiteSettings)
settings.GET("/about", c.Settings.GetAbout)
settings.GET("/login-logs", c.Settings.GetLoginLogs)
settings.GET("/loginlogs", c.Settings.GetLoginLogs)
}
}
}
internal/services/executor_service.go
+75 -41
View File
@@ -24,20 +24,91 @@ type ExecutionResult struct {
End time.Time
}
// ExecutionCallback 任务执行完成后的回调函数类型
type ExecutionCallback func(taskID uint, command string, result *ExecutionResult)
// ExecutorService handles task execution
type ExecutorService struct {
taskService *TaskService
results []ExecutionResult
runningTasks map[int]bool // 正在运行的任务
runningTasks map[int]bool
callbacks []ExecutionCallback
mu sync.RWMutex
}
// NewExecutorService creates a new executor service
func NewExecutorService(taskService *TaskService) *ExecutorService {
return &ExecutorService{
es := &ExecutorService{
taskService: taskService,
results: make([]ExecutionResult, 0),
runningTasks: make(map[int]bool),
callbacks: make([]ExecutionCallback, 0),
}
// 注册默认回调
es.RegisterCallback(es.saveTaskLogCallback)
es.RegisterCallback(es.updateStatsCallback)
return es
}
// RegisterCallback 注册执行完成回调
func (es *ExecutorService) RegisterCallback(cb ExecutionCallback) {
es.mu.Lock()
es.callbacks = append(es.callbacks, cb)
es.mu.Unlock()
}
// executeCallbacks 执行所有回调
func (es *ExecutorService) executeCallbacks(taskID uint, command string, result *ExecutionResult) {
es.mu.RLock()
callbacks := make([]ExecutionCallback, len(es.callbacks))
copy(callbacks, es.callbacks)
es.mu.RUnlock()
for _, cb := range callbacks {
cb(taskID, command, result)
}
}
// saveTaskLogCallback 保存任务日志的回调
func (es *ExecutorService) saveTaskLogCallback(taskID uint, command string, result *ExecutionResult) {
output := result.Output
if result.Error != "" {
output += "\n[ERROR]\n" + result.Error
}
compressed, err := utils.CompressToBase64(output)
if err != nil {
logger.Errorf("Failed to compress log: %v", err)
compressed = ""
}
status := "success"
if !result.Success {
status = "failed"
}
taskLog := &models.TaskLog{
TaskID: taskID,
Command: command,
Output: compressed,
Status: status,
Duration: result.End.Sub(result.Start).Milliseconds(),
}
if err := database.DB.Create(taskLog).Error; err != nil {
logger.Errorf("Failed to save task log: %v", err)
}
}
// updateStatsCallback 更新统计数据的回调
func (es *ExecutorService) updateStatsCallback(taskID uint, _ string, result *ExecutionResult) {
status := "success"
if !result.Success {
status = "failed"
}
sendStatsService := NewSendStatsService()
if err := sendStatsService.IncrementStats(taskID, status); err != nil {
logger.Errorf("Failed to update stats: %v", err)
}
}
@@ -72,8 +143,8 @@ func (es *ExecutorService) ExecuteTask(taskID int) *ExecutionResult {
delete(es.runningTasks, taskID)
es.mu.Unlock()
// Save log to database
es.saveTaskLog(uint(taskID), task.Command, result)
// 执行回调
es.executeCallbacks(uint(taskID), task.Command, result)
return result
}
@@ -97,11 +168,9 @@ func (es *ExecutorService) ExecuteCommandWithTimeout(command string, timeout tim
Start: time.Now(),
}
// Create a context with timeout
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
// Execute the command
shell, args := utils.GetShellCommand(command)
cmd := exec.CommandContext(ctx, shell, args...)
var stdout, stderr bytes.Buffer
@@ -111,7 +180,6 @@ func (es *ExecutorService) ExecuteCommandWithTimeout(command string, timeout tim
err := cmd.Run()
result.End = time.Now()
// Process results
result.Output = stdout.String()
if err != nil {
if ctx.Err() == context.DeadlineExceeded {
@@ -123,10 +191,8 @@ func (es *ExecutorService) ExecuteCommandWithTimeout(command string, timeout tim
result.Success = true
}
// Store result
es.mu.Lock()
es.results = append(es.results, *result)
// Keep only the last 100 results to prevent memory issues
if len(es.results) > 100 {
es.results = es.results[1:]
}
@@ -149,35 +215,3 @@ func (es *ExecutorService) GetLastResults(count int) []ExecutionResult {
copy(results, es.results[start:])
return results
}
// saveTaskLog saves execution log to database with gzip+base64 compression
func (es *ExecutorService) saveTaskLog(taskID uint, command string, result *ExecutionResult) {
output := result.Output
if result.Error != "" {
output += "\n[ERROR]\n" + result.Error
}
// Compress output
compressed, err := utils.CompressToBase64(output)
if err != nil {
logger.Errorf("Failed to compress log: %v", err)
compressed = ""
}
status := "success"
if !result.Success {
status = "failed"
}
taskLog := &models.TaskLog{
TaskID: taskID,
Command: command,
Output: compressed,
Status: status,
Duration: result.End.Sub(result.Start).Milliseconds(),
}
if err := database.DB.Create(taskLog).Error; err != nil {
logger.Errorf("Failed to save task log: %v", err)
}
}
internal/services/send_stats_service.go
+63
View File
@@ -0,0 +1,63 @@
package services
import (
"time"
"baihu/internal/database"
"baihu/internal/models"
)
type SendStatsService struct{}
func NewSendStatsService() *SendStatsService {
return &SendStatsService{}
}
// IncrementStats 增加任务执行统计
func (s *SendStatsService) IncrementStats(taskID uint, status string) error {
today := time.Now().Format("2006-01-02")
startOfDay, _ := time.ParseInLocation("2006-01-02", today, time.Local)
var stats models.SendStats
result := database.DB.Where("task_id = ? AND status = ? AND created_at >= ?", taskID, status, startOfDay).First(&stats)
if result.Error != nil {
// 不存在则创建
stats = models.SendStats{
TaskID: taskID,
Status: status,
Num: 1,
CreatedAt: time.Now(),
}
return database.DB.Create(&stats).Error
}
// 存在则增加计数
return database.DB.Model(&stats).Update("num", stats.Num+1).Error
}
// GetStatsByTaskID 获取任务的统计数据
func (s *SendStatsService) GetStatsByTaskID(taskID uint) []models.SendStats {
var stats []models.SendStats
database.DB.Where("task_id = ?", taskID).Order("created_at DESC").Find(&stats)
return stats
}
// GetTodayStats 获取今日统计
func (s *SendStatsService) GetTodayStats() []models.SendStats {
today := time.Now().Format("2006-01-02")
startOfDay, _ := time.ParseInLocation("2006-01-02", today, time.Local)
var stats []models.SendStats
database.DB.Where("created_at >= ?", startOfDay).Find(&stats)
return stats
}
// GetRecentStats 获取最近N天的统计
func (s *SendStatsService) GetRecentStats(days int) []models.SendStats {
startDate := time.Now().AddDate(0, 0, -days)
var stats []models.SendStats
database.DB.Where("created_at >= ?", startDate).Order("created_at DESC").Find(&stats)
return stats
}