Function ExtractQDevS3Data

backend/plugins/q_dev/tasks/s3_data_extractor.go:39–107

ExtractQDevS3Data 从S3下载CSV数据并解析

(taskCtx plugin.SubTaskContext)

// ExtractQDevS3Data 从S3下载CSV数据并解析
func ExtractQDevS3Data(taskCtx plugin.SubTaskContext) errors.Error {
	data := taskCtx.GetData().(*QDevTaskData)
	db := taskCtx.GetDal()

	// 查询未处理的CSV文件元数据（排除.json.gz日志文件）
	cursor, err := db.Cursor(
		dal.From(&models.QDevS3FileMeta{}),
		dal.Where("connection_id = ? AND processed = ? AND file_name LIKE ?",
			data.Options.ConnectionId, false, "%.csv"),
	)
	if err != nil {
		return errors.Default.Wrap(err, "failed to get file metadata cursor")
	}
	defer cursor.Close()

	taskCtx.SetProgress(0, -1)

	// 处理每个文件
	for cursor.Next() {
		fileMeta := &models.QDevS3FileMeta{}
		err = db.Fetch(cursor, fileMeta)
		if err != nil {
			return errors.Default.Wrap(err, "failed to fetch file metadata")
		}

		// 获取文件内容
		getInput := &s3.GetObjectInput{
			Bucket: aws.String(data.S3Client.Bucket),
			Key:    aws.String(fileMeta.S3Path),
		}

		getResult, err := data.S3Client.S3.GetObject(getInput)
		if err != nil {
			return errors.Convert(err)
		}

		// Use a transaction to process the file and update its status
		tx := db.Begin()
		csvErr := processCSVData(taskCtx, tx, getResult.Body, fileMeta)
		if csvErr != nil {
			if rollbackErr := tx.Rollback(); rollbackErr != nil {
				taskCtx.GetLogger().Error(rollbackErr, "failed to rollback transaction")
			}
			return errors.Default.Wrap(csvErr, fmt.Sprintf("failed to process CSV file %s", fileMeta.FileName))
		}

		// Update file processing status within the same transaction
		fileMeta.Processed = true
		now := time.Now()
		fileMeta.ProcessedTime = &now
		err = tx.Update(fileMeta)
		if err != nil {
			if rollbackErr := tx.Rollback(); rollbackErr != nil {
				taskCtx.GetLogger().Error(rollbackErr, "failed to rollback transaction")
			}
			return errors.Default.Wrap(err, "failed to update file metadata")
		}