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

feat: add repo sync page

Browse Source
This commit is contained in:
engigu
2025-12-25 17:55:54 +08:00
parent bf47048f8f
commit f096d6dea4
9 changed files with 655 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/controllers/task_controller.go
+13 -3
View File
@@ -49,7 +49,9 @@ func resolveWorkDir(workDir string) string {
func (tc *TaskController) CreateTask(c *gin.Context) {
var req struct {
Name string `json:"name" binding:"required"`
Command string `json:"command" binding:"required"`
Command string `json:"command"`
Type string `json:"type"`
Config string `json:"config"`
Schedule string `json:"schedule" binding:"required"`
Timeout int `json:"timeout"`
WorkDir string `json:"work_dir"`
@@ -62,6 +64,12 @@ func (tc *TaskController) CreateTask(c *gin.Context) {
return
}
// 普通任务需要命令
if req.Type != "repo" && req.Command == "" {
utils.BadRequest(c, "命令不能为空")
return
}
if err := tc.cronService.ValidateCron(req.Schedule); err != nil {
utils.BadRequest(c, "无效的cron表达式: "+err.Error())
return
@@ -70,7 +78,7 @@ func (tc *TaskController) CreateTask(c *gin.Context) {
// 转换为绝对路径
workDir := resolveWorkDir(req.WorkDir)
task := tc.taskService.CreateTask(req.Name, req.Command, req.Schedule, req.Timeout, workDir, req.CleanConfig, req.Envs)
task := tc.taskService.CreateTask(req.Name, req.Command, req.Schedule, req.Timeout, workDir, req.CleanConfig, req.Envs, req.Type, req.Config)
tc.cronService.AddTask(task)
utils.Success(c, task)
@@ -110,6 +118,8 @@ func (tc *TaskController) UpdateTask(c *gin.Context) {
var req struct {
Name string `json:"name"`
Command string `json:"command"`
Type string `json:"type"`
Config string `json:"config"`
Schedule string `json:"schedule"`
Timeout int `json:"timeout"`
WorkDir string `json:"work_dir"`
@@ -130,7 +140,7 @@ func (tc *TaskController) UpdateTask(c *gin.Context) {
}
}
task := tc.taskService.UpdateTask(id, req.Name, req.Command, req.Schedule, req.Timeout, resolveWorkDir(req.WorkDir), req.CleanConfig, req.Envs, req.Enabled)
task := tc.taskService.UpdateTask(id, req.Name, req.Command, req.Schedule, req.Timeout, resolveWorkDir(req.WorkDir), req.CleanConfig, req.Envs, req.Enabled, req.Type, req.Config)
if task == nil {
utils.NotFound(c, "任务不存在")
return
internal/models/task.go
+16 -1
View File
@@ -12,11 +12,26 @@ type CleanConfig struct {
Keep int `json:"keep"` // 保留天数或条数
}
// RepoConfig 仓库同步配置
type RepoConfig struct {
SourceType string `json:"source_type"` // url 或 git
SourceURL string `json:"source_url"` // 源地址
TargetPath string `json:"target_path"` // 目标路径
Branch string `json:"branch"` // Git 分支
SparsePath string `json:"sparse_path"` // 稀疏检出路径（仅拉取指定目录或文件）
SingleFile bool `json:"single_file"` // 单文件模式（直接下载文件而非 sparse-checkout）
Proxy string `json:"proxy"` // 代理类型: none, ghproxy, mirror, custom
ProxyURL string `json:"proxy_url"` // 自定义代理地址
AuthToken string `json:"auth_token"` // 认证 Token
}
// Task represents a scheduled task
type Task struct {
ID uint `json:"id" gorm:"primaryKey"`
Name string `json:"name" gorm:"size:255;not null"`
Command string `json:"command" gorm:"type:text;not null"`
Command string `json:"command" gorm:"type:text"` // 普通任务的命令
Type string `json:"type" gorm:"size:20;default:'task'"` // 任务类型: task(普通任务), repo(仓库同步)
Config string `json:"config" gorm:"type:text"` // 配置 JSON（仓库同步配置等）
Schedule string `json:"schedule" gorm:"size:100"` // cron expression
Timeout int `json:"timeout" gorm:"default:30"` // 超时时间（分钟），默认30分钟
WorkDir string `json:"work_dir" gorm:"size:255;default:''"` // 工作目录，为空则使用 scripts 目录
internal/services/executor_service.go
+93 -8
View File
@@ -12,6 +12,7 @@ import (
"fmt"
"os"
"os/exec"
"strings"
"sync"
"time"
)
@@ -296,6 +297,30 @@ func (es *ExecutorService) executeTaskInternal(taskID int) *ExecutionResult {
es.runningTasks[taskID] = true
es.mu.Unlock()
var result *ExecutionResult
// 根据任务类型执行不同逻辑
if task.Type == "repo" {
result = es.executeRepoTask(task)
} else {
result = es.executeNormalTask(task)
}
result.TaskID = taskID
// 标记任务结束
es.mu.Lock()
delete(es.runningTasks, taskID)
es.mu.Unlock()
// 异步执行回调（日志压缩、统计更新、日志清理）
es.executeCallbacksAsync(uint(taskID), task.Command, result)
return result
}
// executeNormalTask 执行普通任务
func (es *ExecutorService) executeNormalTask(task *models.Task) *ExecutionResult {
// 加载环境变量
envService := NewEnvService()
envVars := envService.GetEnvVarsByIDs(task.Envs)
@@ -311,16 +336,76 @@ func (es *ExecutorService) executeTaskInternal(taskID int) *ExecutionResult {
if timeout <= 0 {
timeout = constant.DefaultTaskTimeout
}
result := es.ExecuteCommandWithOptions(task.Command, time.Duration(timeout)*time.Minute, envVars, workDir)
result.TaskID = taskID
return es.ExecuteCommandWithOptions(task.Command, time.Duration(timeout)*time.Minute, envVars, workDir)
}
// 标记任务结束
es.mu.Lock()
delete(es.runningTasks, taskID)
es.mu.Unlock()
// executeRepoTask 执行仓库同步任务（调用 sync.py）
func (es *ExecutorService) executeRepoTask(task *models.Task) *ExecutionResult {
result := &ExecutionResult{
Success: false,
Start: time.Now(),
}
// 异步执行回调（日志压缩、统计更新、日志清理）
es.executeCallbacksAsync(uint(taskID), task.Command, result)
// 解析仓库配置
var config models.RepoConfig
if err := json.Unmarshal([]byte(task.Config), &config); err != nil {
result.End = time.Now()
result.Error = "解析仓库配置失败: " + err.Error()
return result
}
// 构建 sync.py 命令参数
args := []string{
"/opt/sync.py",
"--source-type", config.SourceType,
"--source-url", config.SourceURL,
"--target-path", config.TargetPath,
}
// Git 分支
if config.Branch != "" {
args = append(args, "--branch", config.Branch)
}
// 稀疏路径
if config.SparsePath != "" {
args = append(args, "--path", config.SparsePath)
}
// 单文件模式
if config.SingleFile {
args = append(args, "--single-file")
}
// 代理设置
if config.Proxy != "" && config.Proxy != "none" {
args = append(args, "--proxy", config.Proxy)
if config.Proxy == "custom" && config.ProxyURL != "" {
args = append(args, "--proxy-url", config.ProxyURL)
}
}
// 认证 Token
if config.AuthToken != "" {
args = append(args, "--auth-token", config.AuthToken)
}
// 构建命令
command := "python3 " + strings.Join(args, " ")
// 使用任务配置的超时时间
timeout := task.Timeout
if timeout <= 0 {
timeout = constant.DefaultTaskTimeout
}
// 执行命令
execResult := es.ExecuteCommandWithOptions(command, time.Duration(timeout)*time.Minute, nil, "/opt")
result.End = time.Now()
result.Output = execResult.Output
result.Success = execResult.Success
result.Error = execResult.Error
return result
}
internal/services/task_service.go
+11 -2
View File
@@ -11,10 +11,15 @@ func NewTaskService() *TaskService {
return &TaskService{}
}
func (ts *TaskService) CreateTask(name, command, schedule string, timeout int, workDir, cleanConfig, envs string) *models.Task {
func (ts *TaskService) CreateTask(name, command, schedule string, timeout int, workDir, cleanConfig, envs, taskType, config string) *models.Task {
if taskType == "" {
taskType = "task"
}
task := &models.Task{
Name: name,
Command: command,
Type: taskType,
Config: config,
Schedule: schedule,
Timeout: timeout,
WorkDir: workDir,
@@ -56,7 +61,7 @@ func (ts *TaskService) GetTaskByID(id int) *models.Task {
return &task
}
func (ts *TaskService) UpdateTask(id int, name, command, schedule string, timeout int, workDir, cleanConfig, envs string, enabled bool) *models.Task {
func (ts *TaskService) UpdateTask(id int, name, command, schedule string, timeout int, workDir, cleanConfig, envs string, enabled bool, taskType, config string) *models.Task {
var task models.Task
if err := database.DB.First(&task, id).Error; err != nil {
return nil
@@ -69,6 +74,10 @@ func (ts *TaskService) UpdateTask(id int, name, command, schedule string, timeou
task.CleanConfig = cleanConfig
task.Envs = envs
task.Enabled = enabled
if taskType != "" {
task.Type = taskType
}
task.Config = config
database.DB.Save(&task)
return &task
}