Drupal 11：使用批量 API 处理 CSV 文件

这是关于 Drupal 中批量 API 系列文章的第四篇。批量 API 是 Drupal 里的一个系统，它允许以小块的方式处理数据，避免出现超时错误或内存问题。

到目前为止，在这个系列中，我们已经了解了如何使用表单创建批量处理，接着创建批量类以便通过 Drush 运行批量任务，然后使用完成状态来控制批量处理。这些文章共同构成了 Drupal 中批量处理的基础。

在本文中，我们会把这些概念结合起来，执行一项在网站上相当常见的任务——处理逗号分隔值（CSV）文件，我们将借助 Drupal 批量 API 来完成这项任务，这在 Drupal 开发中是一个实用的操作。

处理 CSV 数据在网络上极为常见。虽然与 API 集成很常见，但有时从一个系统生成所需数据的 CSV 文件并上传到网站的表单中会更简单。大多数系统都允许将数据导出为某种类型的 CSV 文件，而且从 Excel 和 Google Sheets 等常用程序中导出 CSV 文件也十分容易。

在 PHP 中处理 CSV 文件相对简单，但一旦记录数量达到一百条，你会发现 PHP 由于超时或内存问题开始抛出错误。解决方案是使用批量 API，将处理负载分散到多个不同的请求中。

我见过一些批量处理 CSV 文件的示例，它们常常在批量处理开始时将整个 CSV 文件处理成一个数组，从而出现问题。在本文的示例中，我们将运用 PHP 内置的文件指针函数来处理 CSV 文件，并使用完成属性告知 Drupal 文件已处理完毕，这在 Drupal 模块开发中是很重要的技巧。

一、CSV 文件

这是我们将在下面示例中使用的 CSV 文件。这个 CSV 文件的第一列包含随机的标题文本，第二列包含随机的句子。

jcDBBwa4tjlCOWv,"Ille ludus paulatim quia saepius sit tincidunt."
rFm8ZPDkeyV39rt,"Conventio decet molior nisl ratis sagaciter sudo suscipit typicus venio."
v5At5x5fYL38dh8,"Magna melior odio quae."
g3Wx7mESABwh5ap,"Caecus consequat cui esse gemino gravis hendrerit nostrud probo quibus."
zRz0gHbRsTl3EXv,"Haero refero te."
rlCXRe5zoxBSZLM,"Erat interdico molior qui."

我们将生成文章内容类型的页面，这可以通过 Drupal 的标准安装配置文件获得。第一列将用于文章的标题，第二列将用于文章的正文内容。

二、文件上传表单

为了让用户能够上传 CSV 文件，我们需要创建一个文件上传表单，该表单将触发批量运行。

以下是表单定义类和 buildForm() 方法。该表单本身包含一个“文件”字段，我们可以在其中上传 CSV 文件，以及一个“提交”字段，我们可以通过它提交表单。

<?php

namespace Drupal\batch_csv_example\Form;

use Drupal\batch_csv_example\Batch\BatchClass;
use Drupal\Core\Batch\BatchBuilder;
use Drupal\Core\Form\FormBase;
use Drupal\Core\Form\FormStateInterface;
use Drupal\file\Entity\File;

class BatchForm extends FormBase {

  public function getFormId() {
    return 'batch_csv_example';
  }

  public function buildForm(array $form, FormStateInterface $form_state) {
    $form['csv_file'] = [
      '#type' => 'file',
      '#title' => $this->t('The CSV file to process'),
    ];

    $form['actions'] = [
      '#type' => 'actions',
      'submit' => [
        '#type' => 'submit',
        '#value' => $this->t('Run batch'),
      ],
    ];

    return $form;
  }
}

当表单提交时，我们首先触发 validateForm() 方法，在这个方法中，我们可以将文件从请求复制到临时存储中。使用 Drupal 文件系统中的 file_save_upload() 函数将文件数据从请求中复制过来。我们将 FileExtension 验证器作为第二个参数，以确保上传的文件是 CSV 文件。

file_save_upload() 的结果是一个 \Drupal\file\Entity\File 类型的实体（如果文件验证失败或复制失败，则返回 false）。如果一切正常，我们将该实体作为一个名为“temp_csv_file”的值设置在表单状态中，这样我们就可以在提交处理程序中获取并处理它。

  public function validateForm(array &$form, FormStateInterface $form_state): void {
    parent::validateForm($form, $form_state);

    // Create a temporary file entity based on the file form element.
    // Note file_save_upload() might soon be deprecated.
    // https://www.drupal.org/project/drupal/issues/3375423
    $tempFileEntity = file_save_upload(
      'csv_file',
      [
        'FileExtension' => ['extensions' => 'csv'],
      ],
      FALSE,
      0,
    );

    if (!$tempFileEntity instanceof File) {
      $form_state->setErrorByName('csv_file', $this->t('Upload failed.'));
      return;
    }

    // Keep the temporary file entity available for submit handler.
    $form_state->setValue('temp_csv_file', $tempFileEntity);
  }

需要注意的是，file_save_upload() 函数在不远的将来某个时候将会被移除，可能会在 Drupal 11 中被弃用，并在 Drupal 12 中被移除。由于目前还没有开发出替代服务，我们在这里需要使用 file_save_upload()。这里添加一个关于替换此函数的问题提示，以防你想关注其进展情况。

我们在这里将 CSV 文件保存为临时文件，以便在未来某个时候自动清理。临时文件目录的位置以及清理临时文件的频率在 Drupal 中是可配置的。如果你使用的是负载均衡系统，你需要确保临时目录对所有 Web 节点都可用，否则在批量处理文件时会出现文件丢失错误。

这样设置之后，提交处理程序现在就可以访问我们在验证处理程序中创建的临时文件了。

三、启动批量操作

为了处理 CSV 文件，我们需要触发一个批量处理。这里批量运行的设置在一篇关于使用批量 API 进行批量处理的介绍文章中有详细解释，如果你需要更多信息，请查看该文章。

我们在这里需要做的就是将文件的路径传递给批量处理方法，因为我们将在处理方法本身中处理其余的事情。我们只创建一个操作，并不断调用它，直到文件完全处理完毕。

  public function submitForm(array &$form, FormStateInterface $form_state): void {
    $tempFile = $form_state->getValue('temp_csv_file');

    $batch = new BatchBuilder();
    $batch->setTitle('Running batch process.')
      ->setFinishCallback([BatchClass::class, 'batchFinished'])
      ->setInitMessage('Commencing')
      ->setProgressMessage('Processing...')
      ->setErrorMessage('An error occurred during processing.');

    // Process the CSV as a batch.
    $batch->addOperation([BatchClass::class, 'batchProcess'], [$tempFile->getFileUri()]);

    batch_set($batch->toArray());

    $form_state->setRedirectUrl(new Url($this->getFormId()));
  }

要注意，我们不能在这里传递 $tempFile 实体，因为批量操作使用队列系统，而该系统不允许将复杂对象存储为队列属性。因此，我们传递文件名，这样就可以满足我们的所有需求。

四、处理 CSV 文件

batchProcess() 方法是我们在表单提交处理中设置的单一操作，我们将在这个方法中处理 CSV 文件，以在 Drupal 网站上创建文章。

我们需要做的第一件事是设置沙盒和结果数组。结果数组包含我们在本系列其他文章中设置的常规更新、跳过、失败等项。沙盒用于设置属性，以跟踪文件处理的进度。“seek”属性用于确定我们在文件处理过程中的进度。我们将使用 PHP 的 fseek() 函数来查找 CSV 文件中的当前位置，因此需要在沙盒中跟踪 seek 参数。

public static function batchProcess(string $fileName, array &$context): void {
    if (!isset($context['sandbox']['progress'])) {
      $context['sandbox']['progress'] = 0;
      $context['sandbox']['seek'] = 0;
    }
    if (!isset($context['results']['updated'])) {
      $context['results']['updated'] = 0;
      $context['results']['skipped'] = 0;
      $context['results']['failed'] = 0;
      $context['results']['progress'] = 0;
      $context['results']['process'] = 'CSV batch completed';
    }

    // - Continue batch processing...
}

在批量处理中，我们要做的第一件事是使用 PHP 的 filesize() 函数获取文件的大小。该函数返回文件的大小（以字节为单位），这在处理文件时非常重要。为了便于阅读，我们返回给用户的消息以千字节为单位报告文件大小，因此在填充消息参数时需要进行简单的转换。

$filesize = filesize($fileName);

// Message above progress bar.
$percent = round(($context['sandbox']['seek'] / $filesize) * 100);
$context['message'] = t('Processing file, @seek of @filesize complete (@percentage%).', [
  '@seek' => number_format($context['sandbox']['seek'] / 1024) . 'kb',
  '@filesize' => number_format($filesize / 1024) . 'kb',
  '@percentage' => $percent,
]);

我们不会一次性处理整个 CSV 文件，因此需要为单个批量处理的项目数量设置一个限制。使用另一个名为 $count 的变量来统计我们已经处理的项目数量。如果计数达到限制，我们将停止 CSV 处理，并在下一次批量运行时继续处理。

// How many lines to process at once?
$limit = 50;

// Keep track of how many lines we have processed in this batch.
$count = 0;

这两个变量可以简化为一个变量（即限制），但我认为这样能更清楚地展示处理过程。最终，这取决于编码风格以及你希望如何统计已处理项目的数量。

下一部分是批量处理的重要部分。沙盒中的 seek 属性告诉我们文件处理当前所在的位置，因此我们使用 PHP 的 fseek() 函数将文件指针移动到 CSV 文件中的正确位置。这意味着当我们从文件中读取数据时，它将位于正确的位置，并提取我们尚未读取的数据。

我们第一次调用这个批量函数时，seek 属性将为 0，但当我们完成批量处理方法时，seek 属性会更新。这意味着下一次调用批量处理时，我们将前进到 CSV 文件中的不同位置。

fseek($handle, $context['sandbox']['seek']);
while ($line = fgetcsv($handle, 4096)) {
  // Validate the CSV.
  if (count($line) !== 2) {
    // The line in the CSV file won't import correctly. So skip this line.
    $context['results']['skipped']++;
    continue;
  }

  // Extract the data from the CSV file and create the Article here.

  $count++;
  if ($count >= $limit) {
    // We have reached the limit for this run, break out of the loop.
    // If we have more file to process then we will run the batch
    // function again.
    break;
  }
}

PHP 的 fgetcsv() 函数将从 CSV 文件中提取一行数据，并将其作为数组存储在 $line 变量中。由于我们期望 $line 数组有两个元素，因此可以执行一些简单的验证，以防止在尝试使用无效行时抛出错误。

此时，可以调用文章创建代码，将 CSV 数据（在 $line 变量中）传入 Node::create() 方法。这将创建一个节点对象，我们可以使用 save() 方法将其保存到数据库中。

// Process the CSV item.
$node = Node::create([
  'type' => 'article',
  'title' => $line[0],
  'body' => [
    'value' => '<p>' . $line[1] . '</p>',
    'format' => filter_default_format(),
  ],
  'uid' => 1,
  'status' => 1,
]);
$node->save();

完成本次批量处理后，我们需要确定文件指针的位置。我们使用 PHP 的 ftell() 函数来完成这一操作，然后将其存储在沙盒的 seek 属性中。这意味着下一次批量处理开始时，它将从文件的这个位置继续处理，接上上次的进度。

// Update the position of the pointer.
$context['sandbox']['seek'] = ftell($handle);

以下是批量处理方法的完整文件处理代码。

if ($handle = fopen($fileName, 'r')) {
  fseek($handle, $context['sandbox']['seek']);
  while ($line = fgetcsv($handle, 4096)) {

    $context['results']['progress']++;

    // Validate the CSV.
    if (count($line) !== 2) {
      // The line in the CSV file won't import correctly. So skip this line.
      $context['results']['skipped']++;
      continue;
    }

    // Process the CSV item.
    $node = Node::create([
      'type' => 'article',
      'title' => $line[0],
      'body' => [
        'value' => '<p>' . $line[1] . '</p>',
        'format' => filter_default_format(),
      ],
      'uid' => 1,
      'status' => 1,
    ]);
    $node->save();

    $context['results']['updated']++;

    $count++;
    if ($count >= $limit) {
      // We have reached the limit for this run, break out of the loop.
      // If we have more file to process then we will run the batch
      // function again.
      break;
    }
  }
  // Update the position of the pointer.
  $context['sandbox']['seek'] = ftell($handle);

  // Close the file handle.
  fclose($handle);
}

最后，在批量处理方法中我们要做的最后一件事是更新“finished”属性。finished 属性会告知 Drupal 批量运行的当前状态，如果该值小于 1，则会再次调用批量处理方法，否则将停止批量处理并触发完成回调。

通过使用 seek 属性和文件大小，我们可以确定何时完成文件处理，并通过将两者相除来告知 Drupal 停止批量处理。

// Update the finished parameter.
$context['finished'] = $context['sandbox']['seek'] / $filesize;

至此，批量处理方法介绍完毕。使用这些代码行，我们可以处理非常长的 CSV 文件，而不会超过 PHP 的超时时间或内存限制，这在 Drupal 模块开发中对于处理大量数据很有帮助。

以下是正在运行的 CSV 批量操作的截图，显示总共 59kb 的文件中已经处理了 15kb。

为了详细说明这里发生的情况，假设我们有一个 29kb 的 CSV 文件。处理过程如下。

• 第一次运行处理方法时，seek 属性被设置为 0。
  • 然后我们使用 fseek() 将 CSV 文件的文件指针设置为 0，并读取前 50 行数据，创建 50 个文章节点。
  • 第一次运行结束时，seek 属性被设置为 3016 字节，然后结束该过程。
  • 我们通过将 3016 字节除以 100kb 来计算 finished 属性，该值远小于 1，因此会再次触发处理调用。
• 第二次触发处理方法，我们使用 fseek() 函数将指针移动到 seek 属性的值（即 3016 字节）。
  • 然后从该点继续处理 CSV 文件，直到达到 50 项的限制。
  • 我们将 seek 参数更新为 5770 字节，该值仍然小于文件大小，因此会再次调用处理方法。
• 第三次调用处理方法，我们再次使用沙盒属性中的 seek 值，通过 fseek() 函数移动文件指针。
  • 再处理 50 项。
  • 然后将 seek 属性设置为 8684 字节。
• 我们继续处理 CSV 文件，直到最后一次运行。
• 最后一次调用处理方法时，fgetcsv() 方法将返回 false，这将跳出 CSV 文件处理循环。当我们获取文件指针位置时，该位置为 29924 字节，与文件大小相同。当我们计算 finished 属性时，该属性值将等于 1，批量操作完成。

将一个 29kb 的 CSV 文件处理成文章将创建 500 篇文章。

五、结论

我公司已经使用这个 CSV 处理代码处理过文件中多达 60000 项的数据，在处理过程中没有遇到任何问题。由于每条记录都有一些验证和数据处理步骤，所以那次处理的批量大小较小，但是使用 fseek() 和 ftell() 来跟踪文件指针进度是这种技术的关键原则。

在研究如何使用批量 API 处理 CSV 文件时，我公司发现一些示例会在处理开始时就从 CSV 文件中提取所有数据，并根据 CSV 文件中的项数创建多个批量操作。这意味着如果文件很大，批量处理的初始化实际上会在处理开始之前就超时并失败。通过将文件名传递给批量处理，然后以小块的方式逐步处理文件，我们消除了任何设置问题，并且仍然可以在不出现任何减速的情况下处理大型 CSV 文件，这在 Drupal 升级和 Drupal 开发中都具有重要意义。

本文的所有代码以及本系列所有其他示例的源代码都可以在 GitHub 上找到。batch_csv_example 模块还包含一个控制器操作，可以用于下载要在批量表单中处理的 CSV 文件。这个控制器操作在表单中提供了链接，因此你可以轻松找到它。在你自己的网站上使用这个模块时要小心，因为它会在几秒钟内创建 1000 篇文章，而且我公司还没有创建删除这些文章的方法！

在本系列的下一篇文章中，我们将探讨如何向正在运行的批量处理添加额外的操作。如果你对批量 API 有任何特定问题需要解答，请随时联系我们。