package main import ( "ZhenTuLocalPassiveAdapter/api" "ZhenTuLocalPassiveAdapter/config" "ZhenTuLocalPassiveAdapter/core" "ZhenTuLocalPassiveAdapter/dto" "ZhenTuLocalPassiveAdapter/logger" "ZhenTuLocalPassiveAdapter/telemetry" "context" "fmt" "os" "os/signal" "syscall" "time" "github.com/gin-gonic/gin" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" "go.uber.org/zap" ) var tracer = otel.Tracer("vpt") func startTask(device config.DeviceMapping, task dto.Task) { ctx, span := tracer.Start(context.Background(), "startTask") defer span.End() span.SetAttributes(attribute.String("deviceNo", device.DeviceNo)) span.SetAttributes(attribute.String("taskId", task.TaskID)) span.SetAttributes(attribute.String("scenicId", task.ScenicID)) span.SetAttributes(attribute.String("startTime", task.StartTime.Format("2006-01-02 15:04:05"))) span.SetAttributes(attribute.String("endTime", task.EndTime.Format("2006-01-02 15:04:05"))) fo, err := core.HandleTask(ctx, device, task) if err != nil { span.SetStatus(codes.Error, "处理任务失败") logger.Error("处理任务失败", zap.String("taskID", task.TaskID), zap.String("deviceNo", task.DeviceNo), zap.Error(err)) api.ReportTaskFailure(ctx, task.TaskID) return } span.SetAttributes(attribute.String("fileUrl", fo.URL)) logger.Info("处理任务成功", zap.String("taskID", task.TaskID), zap.String("deviceNo", task.DeviceNo)) err = api.UploadTaskFile(ctx, task, *fo) if err != nil { span.SetStatus(codes.Error, "上传文件失败") logger.Error("上传文件失败", zap.String("taskID", task.TaskID), zap.String("deviceNo", task.DeviceNo), zap.Error(err)) api.ReportTaskFailure(ctx, task.TaskID) return } result := api.ReportTaskSuccess(ctx, task.TaskID, fo) if !result { span.SetStatus(codes.Error, "上报任务成功失败") logger.Error("上报任务成功失败", zap.String("taskID", task.TaskID), zap.String("deviceNo", task.DeviceNo), zap.Error(err)) return } span.SetStatus(codes.Ok, "上传文件成功") logger.Info("上传文件成功", zap.String("taskID", task.TaskID), zap.String("deviceNo", task.DeviceNo)) } func runTaskLoop(ctx context.Context) { ticker := time.NewTicker(2 * time.Second) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: // 执行任务 tasks, err := api.SyncTask() if err == nil { for _, task := range tasks { logger.Info("开始处理任务", zap.String("taskID", task.TaskID), zap.String("deviceNo", task.DeviceNo), zap.String("startTime", task.StartTime.Format("2006-01-02 15:04:05")), zap.String("endTime", task.EndTime.Format("2006-01-02 15:04:05"))) // 处理任务 for _, device := range config.Config.Devices { if device.DeviceNo == task.DeviceNo { // 处理任务 go startTask(device, task) break // 提前返回,避免不必要的循环 } } } } else { logger.Error("同步任务失败", zap.Error(err)) } } } } func startViidServer() { if !config.Config.Viid.Enabled { return } gin.SetMode(gin.ReleaseMode) r := gin.Default() // Register Routes api.RegisterVIIDRoutes(r) addr := fmt.Sprintf(":%d", config.Config.Viid.Port) logger.Info("VIID Server starting", zap.String("addr", addr)) go func() { if err := r.Run(addr); err != nil { logger.Error("VIID Server failed", zap.Error(err)) } }() } func main() { // 初始化日志 err := logger.Init() if err != nil { panic(err) } defer logger.Sync() err = config.LoadConfig() if err != nil { logger.Fatal("加载配置文件失败", zap.Error(err)) return } ctx := context.Background() shutdown, err := telemetry.InitTelemetry(ctx) if err != nil { logger.Fatal("Failed to initialize telemetry", zap.Error(err)) return } defer shutdown(ctx) if config.Config.Record.Storage.Type == "local" { _, err = os.Stat(config.Config.Record.Storage.Path) if err != nil { logger.Error("录像文件夹配置失败", zap.Error(err)) return } else { logger.Info("录像文件夹配置有效") } } else { logger.Info("录像文件夹配置为OSS") } // Start VIID Server startViidServer() // Context for graceful shutdown ctx, cancel := context.WithCancel(context.Background()) defer cancel() // Handle Signals sigChan := make(chan os.Signal, 1) signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) // Start Task Loop go runTaskLoop(ctx) // Wait for signal <-sigChan logger.Info("Received shutdown signal") cancel() logger.Info("Shutdown complete") }