Function processCSVData

backend/plugins/q_dev/tasks/s3_data_extractor.go:110–170

处理CSV文件

(taskCtx plugin.SubTaskContext, db dal.Dal, reader io.ReadCloser, fileMeta *models.QDevS3FileMeta)

// 处理CSV文件
func processCSVData(taskCtx plugin.SubTaskContext, db dal.Dal, reader io.ReadCloser, fileMeta *models.QDevS3FileMeta) errors.Error {
	defer reader.Close()

	// Get task data to access Identity Client
	data := taskCtx.GetData().(*QDevTaskData)

	csvReader := csv.NewReader(reader)
	// 使用默认的逗号分隔符，不需要设置 Comma
	csvReader.LazyQuotes = true    // 允许非标准引号处理
	csvReader.FieldsPerRecord = -1 // 允许每行字段数不同

	// 读取标头
	headers, err := csvReader.Read()
	taskCtx.GetLogger().Debug("CSV headers: %+v", headers)
	if err != nil {
		return errors.Convert(err)
	}

	// Auto-detect CSV format from headers
	isNewFormat := detectUserReportFormat(headers)
	if isNewFormat {
		taskCtx.GetLogger().Debug("Detected new user_report CSV format")
	} else {
		taskCtx.GetLogger().Debug("Detected old by_user_analytic CSV format")
	}

	// 逐行读取数据
	for {
		record, err := csvReader.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return errors.Convert(err)
		}

		if isNewFormat {
			reportData, err := createUserReportData(taskCtx.GetLogger(), headers, record, fileMeta, data.IdentityClient)
			if err != nil {
				return errors.Default.Wrap(err, "failed to create user report data")
			}
			err = db.CreateOrUpdate(reportData)
			if err != nil {
				return errors.Default.Wrap(err, "failed to save user report data")
			}
		} else {
			// 创建用户数据对象 (updated to include display name resolution)
			userData, err := createUserDataWithDisplayName(taskCtx.GetLogger(), headers, record, fileMeta, data.IdentityClient)
			if err != nil {
				return errors.Default.Wrap(err, "failed to create user data")
			}

			err = db.CreateOrUpdate(userData)
			if err != nil {
				return errors.Default.Wrap(err, "failed to save user data")
			}
		}
	}