feat: import products from the MyE&P export API

2026-09-30 14:57:13 +02:00
1 parent 1b6f480bee
commit b510299032
@@ -29,12 +29,9 @@ namespace EP\EpProducts\Command;

 use EP\EpProducts\Service\DateImportService;
 use EP\EpProducts\Service\ProductImportSourceFactory;
-use EP\EpProducts\Service\XmlProductReader;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Output\OutputInterface;
-use TYPO3\CMS\Core\Log\LogManager;
-use TYPO3\CMS\Core\Utility\GeneralUtility;
 use TYPO3\CMS\Extbase\Configuration\ConfigurationManagerInterface;
 use TYPO3\CMS\Extbase\Service\CacheService;

@@ -91,14 +88,14 @@ class DateCommandController extends Command
    public function execute(InputInterface $input, OutputInterface $output): int
    {
        $importSource = $this->importSourceFactory->get();
-        $path = $importSource->acquire();
-        if ($path === null) {
+        $import = $importSource->acquire();
+        if ($import === null) {
            $output->writeln('No new import available.');
            return 0;
        }

-        $count = $this->dateImportService->import(new XmlProductReader($path, GeneralUtility::makeInstance(LogManager::class)->getLogger(XmlProductReader::class)));
-        $importSource->markImported($path);
+        $count = $this->dateImportService->import($import->getRows());
+        $importSource->markImported($import);
        $output->writeln(sprintf('%d rows imported.', $count ));
        $settings = $this->getSettings();
        $resellerPageUid = $settings['resellerExportPageUid'];
@@ -81,7 +81,8 @@ class ApiClient implements LoggerAwareInterface
        }

        try {
-            return new \DateTimeImmutable($data['timestamp']);
+            // The API reports Europe/Berlin local time without an offset.
+            return new \DateTimeImmutable($data['timestamp'], new \DateTimeZone('Europe/Berlin'));
        } catch (\Throwable $e) {
            $message = sprintf('Invalid timestamp returned by MyEP API for GET last-update: %s', $e->getMessage());
            $this->logError($message, ['exception' => $e]);
@@ -89,6 +90,22 @@ class ApiClient implements LoggerAwareInterface
        }
    }

+    /**
+     * One page of the product export: one row per travel date × hotel in `data`,
+     * and `meta` describing the whole export.
+     *
+     * @throws ApiException
+     */
+    public function getProductExport(int $page, int $perPage): array
+    {
+        return $this->requestJson('GET', 'exports/products', [
+            'query' => [
+                'page' => $page,
+                'perPage' => $perPage,
+            ],
+        ]);
+    }
+
    /**
     * @throws ApiException
     */
@@ -0,0 +1,233 @@
+<?php
+
+namespace EP\EpProducts\Service;
+
+use EP\EpProducts\MyEP\ApiClient;
+use EP\EpProducts\MyEP\ApiException;
+use Psr\Log\LoggerAwareInterface;
+use Psr\Log\LoggerAwareTrait;
+use TYPO3\CMS\Core\Core\Environment;
+use TYPO3\CMS\Core\SingletonInterface;
+use TYPO3\CMS\Core\Utility\GeneralUtility;
+
+/**
+ * Reads the product export from GET /api/exports/products on MyE&P. The rows
+ * already have the shape {@see DateImportService::import()} consumes, so they
+ * are passed on unchanged.
+ *
+ * Change detection compares GET /api/last-update against the
+ * `meta.sourceGeneratedAt` of the last successfully imported export, kept in
+ * var/transient/ep_products_import/api-imported.info. Both carry the BusPro
+ * export's own timestamp.
+ */
+final class ApiImportSource implements ProductImportSource, SingletonInterface, LoggerAwareInterface
+{
+    use LoggerAwareTrait;
+
+    /**
+     * The whole export is one page at current volumes; the walk does not rely on it.
+     */
+    private const PER_PAGE = 500;
+
+    /**
+     * How often the walk restarts when MyE&P replaces its copy of the export
+     * mid-walk before giving up until the next run.
+     */
+    private const MAX_WALKS = 3;
+
+    private const STAGING_SUBDIR = 'ep_products_import';
+    private const BASELINE_FILE = 'api-imported.info';
+    private const IMPORT_MARKER = 'fileadmin/products_import.info';
+
+    /**
+     * @var ApiClient|null
+     */
+    private $apiClient;
+
+    public function __construct(?ApiClient $apiClient = null)
+    {
+        $this->apiClient = $apiClient;
+    }
+
+    /**
+     * The whole export is fetched before anything is returned, so a walk that
+     * has to restart never leaves a stitched dataset behind.
+     *
+     * @throws ApiException
+     */
+    public function acquire(): ?ProductImport
+    {
+        if (!$this->isNewImportAvailable()) {
+            return null;
+        }
+
+        for ($walk = 1; $walk <= self::MAX_WALKS; $walk++) {
+            $export = $this->walkPages();
+            if ($export === false) {
+                continue;
+            }
+            if ($export === null) {
+                return null;
+            }
+
+            [$rows, $sourceGeneratedAt] = $export;
+            $baseline = $this->getImportedTimestamp();
+            if ($baseline !== null && $sourceGeneratedAt <= $baseline) {
+                return null;
+            }
+
+            return new ProductImport($rows, $sourceGeneratedAt);
+        }
+
+        throw new ApiException(sprintf(
+            'MyE&P replaced its product export during each of %d page walks, giving up until the next run',
+            self::MAX_WALKS
+        ));
+    }
+
+    public function markImported(ProductImport $import): void
+    {
+        $sourceGeneratedAt = $import->getSourceGeneratedAt();
+        if ($sourceGeneratedAt !== null) {
+            $file = $this->baselinePath();
+            GeneralUtility::mkdir_deep(dirname($file));
+            GeneralUtility::writeFile($file, $sourceGeneratedAt->format(\DateTimeInterface::ATOM), false);
+        }
+
+        (new ImportTimestampService())->writeImportTimestamp(self::IMPORT_MARKER);
+    }
+
+    /**
+     * True when MyE&P reports an export newer than the last imported one, or
+     * nothing has been imported yet. A missing timestamp from MyE&P means
+     * "not importable now", never "import".
+     *
+     * @throws ApiException
+     */
+    public function isNewImportAvailable(): bool
+    {
+        $lastUpdateAt = $this->getApiClient()->getLastUpdateAt();
+        if ($lastUpdateAt === null) {
+            return false;
+        }
+
+        $baseline = $this->getImportedTimestamp();
+
+        return $baseline === null || $lastUpdateAt > $baseline;
+    }
+
+    /**
+     * `meta.sourceGeneratedAt` of the last successfully imported export.
+     */
+    public function getImportedTimestamp(): ?\DateTimeImmutable
+    {
+        $file = $this->baselinePath();
+        if (!is_file($file)) {
+            return null;
+        }
+
+        return $this->parseTimestamp(trim((string) file_get_contents($file)));
+    }
+
+    /**
+     * @return array{0: array[], 1: \DateTimeImmutable}|false|null the rows and
+     *         the export's timestamp; null when the export is not importable now;
+     *         false when the export changed mid-walk and the walk must restart
+     * @throws ApiException
+     */
+    private function walkPages()
+    {
+        $rows = [];
+        $sourceGeneratedAt = null;
+        $page = 1;
+
+        do {
+            $body = $this->getApiClient()->getProductExport($page, self::PER_PAGE);
+            $meta = $body['meta'] ?? null;
+            if (!is_array($meta) || !isset($body['data']) || !is_array($body['data'])) {
+                throw new ApiException(sprintf('MyE&P product export page %d is malformed', $page));
+            }
+
+            $pageGeneratedAt = $this->parseTimestamp($meta['sourceGeneratedAt'] ?? null);
+            if ($page === 1) {
+                if ($pageGeneratedAt === null) {
+                    $this->logWarning('MyE&P product export has no source timestamp; skipping product import');
+                    return null;
+                }
+                $sourceGeneratedAt = $pageGeneratedAt;
+            } elseif ($pageGeneratedAt === null || $pageGeneratedAt != $sourceGeneratedAt) {
+                $this->logWarning('MyE&P product export changed during the page walk; restarting', [
+                    'page' => $page,
+                    'expected' => $sourceGeneratedAt->format(\DateTimeInterface::ATOM),
+                    'actual' => $meta['sourceGeneratedAt'] ?? null,
+                ]);
+                return false;
+            }
+
+            foreach ($body['data'] as $row) {
+                $rows[] = $row;
+            }
+
+            $totalPages = (int) ($meta['totalPages'] ?? 0);
+            $page++;
+        } while ($page <= $totalPages);
+
+        $total = (int) ($meta['total'] ?? -1);
+        if (count($rows) !== $total) {
+            throw new ApiException(sprintf(
+                'MyE&P product export is incomplete: %d rows received, %d announced',
+                count($rows),
+                $total
+            ));
+        }
+
+        if ((int) ($meta['skipped'] ?? 0) > 0) {
+            $this->logWarning('MyE&P skipped travel dates with unparseable dates', ['skipped' => $meta['skipped']]);
+        }
+        if ((int) ($meta['pickupsWithoutMasterData'] ?? 0) > 0) {
+            $this->logWarning(
+                'MyE&P product export has pickups without master data',
+                ['pickupsWithoutMasterData' => $meta['pickupsWithoutMasterData']]
+            );
+        }
+
+        return [$rows, $sourceGeneratedAt];
+    }
+
+    /**
+     * @param mixed $value
+     */
+    private function parseTimestamp($value): ?\DateTimeImmutable
+    {
+        if (!is_string($value) || $value === '') {
+            return null;
+        }
+
+        try {
+            return new \DateTimeImmutable($value);
+        } catch (\Exception $e) {
+            return null;
+        }
+    }
+
+    private function getApiClient(): ApiClient
+    {
+        if ($this->apiClient === null) {
+            $this->apiClient = GeneralUtility::makeInstance(ApiClient::class);
+        }
+
+        return $this->apiClient;
+    }
+
+    private function baselinePath(): string
+    {
+        return Environment::getVarPath() . '/transient/' . self::STAGING_SUBDIR . '/' . self::BASELINE_FILE;
+    }
+
+    private function logWarning(string $message, array $context = []): void
+    {
+        if ($this->logger !== null) {
+            $this->logger->warning($message, $context);
+        }
+    }
+}
@@ -2,6 +2,7 @@

 namespace EP\EpProducts\Service;

+use TYPO3\CMS\Core\Log\LogManager;
 use TYPO3\CMS\Core\SingletonInterface;
 use TYPO3\CMS\Core\Utility\GeneralUtility;

@@ -17,16 +18,22 @@ final class LocalImportSource implements ProductImportSource, SingletonInterface
    private const UPLOAD_MARKER = 'fileadmin/xmlexport/uebertragung.info';
    private const IMPORT_MARKER = 'fileadmin/products_import.info';

-    public function acquire(): ?string
+    public function acquire(): ?ProductImport
    {
        if (!$this->isNewImportAvailable()) {
            return null;
        }

-        return GeneralUtility::getFileAbsFileName(self::XML_DIR);
+        $path = GeneralUtility::getFileAbsFileName(self::XML_DIR);
+        $reader = new XmlProductReader(
+            $path,
+            GeneralUtility::makeInstance(LogManager::class)->getLogger(XmlProductReader::class)
+        );
+
+        return new ProductImport($reader, $reader->getSourceGeneratedAt(), $path);
    }

-    public function markImported(string $path): void
+    public function markImported(ProductImport $import): void
    {
        (new ImportTimestampService())->writeImportTimestamp(self::IMPORT_MARKER);
    }
@@ -0,0 +1,56 @@
+<?php
+
+namespace EP\EpProducts\Service;
+
+/**
+ * One acquired product export: the rows to hand to
+ * {@see DateImportService::import()}, the source's own timestamp, and whatever
+ * the source needs to find it again in {@see ProductImportSource::markImported()}.
+ */
+final class ProductImport
+{
+    /**
+     * @var iterable<array>
+     */
+    private $rows;
+
+    /**
+     * @var \DateTimeImmutable|null
+     */
+    private $sourceGeneratedAt;
+
+    /**
+     * @var string|null
+     */
+    private $location;
+
+    /**
+     * @param iterable<array> $rows
+     * @param \DateTimeImmutable|null $sourceGeneratedAt
+     * @param string|null $location local directory of an XML export, null otherwise
+     */
+    public function __construct(iterable $rows, ?\DateTimeImmutable $sourceGeneratedAt, ?string $location = null)
+    {
+        $this->rows = $rows;
+        $this->sourceGeneratedAt = $sourceGeneratedAt;
+        $this->location = $location;
+    }
+
+    /**
+     * @return iterable<array>
+     */
+    public function getRows(): iterable
+    {
+        return $this->rows;
+    }
+
+    public function getSourceGeneratedAt(): ?\DateTimeImmutable
+    {
+        return $this->sourceGeneratedAt;
+    }
+
+    public function getLocation(): ?string
+    {
+        return $this->location;
+    }
+}
@@ -3,25 +3,26 @@
 namespace EP\EpProducts\Service;

 /**
- * A source of the busProNet product export (Ziel_*.xml, zustiege.xml,
- * uebertragung.info). Implementations either read a local drop directory or
- * pull the files from a remote SFTP server; the caller does not care which.
+ * A source of the BusPro product export. Implementations read the XML export
+ * from a local drop directory or a remote SFTP server, or the JSON export from
+ * the MyE&P API; the caller does not care which.
 */
 interface ProductImportSource
 {
    /**
-     * Returns a local directory ready to hand to {@see DateImportService::import()},
+     * Returns the export ready to hand to {@see DateImportService::import()},
     * or null when there is nothing newer than the last successful import.
     *
-     * @throws SftpImportException on an unrecoverable acquisition failure
+     * @throws SftpImportException on an unrecoverable SFTP acquisition failure
+     * @throws \EP\EpProducts\MyEP\ApiException on an unrecoverable API failure
     */
-    public function acquire(): ?string;
+    public function acquire(): ?ProductImport;

    /**
-     * Called once {@see DateImportService::import()} of the acquired path has
+     * Called once {@see DateImportService::import()} of the acquired export has
     * succeeded, so the source can advance its "last imported" bookkeeping.
     */
-    public function markImported(string $path): void;
+    public function markImported(ProductImport $import): void;

    /**
     * Best-effort, side-effect-free check for whether a newer export is available.
@@ -9,18 +9,24 @@ use TYPO3\CMS\Core\Utility\GeneralUtility;
 /**
 * Picks the product import source based on the `productImportSource` extension
 * configuration switch: `sftp` (default) pulls from the remote SFTP server,
- * `local` reads the legacy fileadmin/xmlexport/ drop directory.
+ * `local` reads the legacy fileadmin/xmlexport/ drop directory, `api` reads the
+ * product export of the MyE&P API.
 */
 final class ProductImportSourceFactory implements SingletonInterface
 {
    public const MODE_SFTP = 'sftp';
    public const MODE_LOCAL = 'local';
+    public const MODE_API = 'api';

    public function get(): ProductImportSource
    {
-        if ($this->getMode() === self::MODE_LOCAL) {
+        $mode = $this->getMode();
+        if ($mode === self::MODE_LOCAL) {
            return GeneralUtility::makeInstance(LocalImportSource::class);
        }
+        if ($mode === self::MODE_API) {
+            return GeneralUtility::makeInstance(ApiImportSource::class);
+        }

        return GeneralUtility::makeInstance(SftpImportSource::class);
    }
@@ -30,6 +36,6 @@ final class ProductImportSourceFactory implements SingletonInterface
        $config = GeneralUtility::makeInstance(ExtensionConfiguration::class)->get('ep_products');
        $mode = strtolower(trim((string)(is_array($config) ? ($config['productImportSource'] ?? '') : '')));

-        return $mode === self::MODE_LOCAL ? self::MODE_LOCAL : self::MODE_SFTP;
+        return in_array($mode, [self::MODE_LOCAL, self::MODE_API], true) ? $mode : self::MODE_SFTP;
    }
 }
@@ -12,8 +12,8 @@ use TYPO3\CMS\Core\Utility\GeneralUtility;

 /**
 * Pulls the busProNet product export (Ziel_*.xml, zustiege.xml, uebertragung.info)
- * from a remote SFTP server into a local staging directory so the existing
- * {@see DateImportService::import()} can keep operating on a plain local folder.
+ * from a remote SFTP server into a local staging directory and reads it through
+ * an {@see XmlProductReader}.
 *
 * Change detection compares the remote uebertragung.info timestamp against the
 * staged copy of the last successfully imported export
@@ -42,11 +42,19 @@ final class SftpImportSource implements ProductImportSource, SingletonInterface,
    private $configuration;

    /**
-     * {@see ProductImportSource::acquire()} — alias for {@see fetchIfNewer()}.
+     * {@see ProductImportSource::acquire()} — {@see fetchIfNewer()}, read through
+     * an {@see XmlProductReader}.
     */
-    public function acquire(): ?string
+    public function acquire(): ?ProductImport
    {
-        return $this->fetchIfNewer();
+        $path = $this->fetchIfNewer();
+        if ($path === null) {
+            return null;
+        }
+
+        $reader = new XmlProductReader($path, $this->logger);
+
+        return new ProductImport($reader, $reader->getSourceGeneratedAt(), $path);
    }

    /**
@@ -54,9 +62,9 @@ final class SftpImportSource implements ProductImportSource, SingletonInterface,
     * change-detection baseline (only reached after a successful import), then
     * refresh the display-only marker for the BE module and system-info toolbar.
     */
-    public function markImported(string $path): void
+    public function markImported(ProductImport $import): void
    {
-        $this->commit($path);
+        $this->commit((string)$import->getLocation());
        (new ImportTimestampService())->writeImportTimestamp(self::IMPORT_MARKER);
    }

@@ -19,6 +19,8 @@ final class XmlProductReader implements \IteratorAggregate
 {
    private const PRODUCT_PREFIX = 'Ziel';
    private const PICKUPS_FILE = 'zustiege.xml';
+    private const MARKER_FILE = 'uebertragung.info';
+    private const MARKER_TS_FORMAT = 'd.m.Y H:i:s';
    private const CODE_PSEUDOPRICE = 'PDGS';
    private const TIMEZONE = 'Europe/Berlin';

@@ -54,6 +56,22 @@ final class XmlProductReader implements \IteratorAggregate
        return $this->path;
    }

+    /**
+     * The export's own timestamp from uebertragung.info, or null if unreadable.
+     */
+    public function getSourceGeneratedAt(): ?\DateTimeImmutable
+    {
+        $file = $this->path . '/' . self::MARKER_FILE;
+        if (!is_file($file)) {
+            return null;
+        }
+
+        $line = trim((string) strtok((string) file_get_contents($file), "\r\n"));
+        $date = \DateTimeImmutable::createFromFormat(self::MARKER_TS_FORMAT, $line, $this->timezone);
+
+        return $date ?: null;
+    }
+
    /**
     * @return \Generator<array>
     */
@@ -29,8 +29,6 @@ namespace EP\EpProducts\Task;

 use EP\EpProducts\Service\DateImportService;
 use EP\EpProducts\Service\ProductImportSourceFactory;
-use EP\EpProducts\Service\XmlProductReader;
-use TYPO3\CMS\Core\Log\LogManager;
 use TYPO3\CMS\Core\Utility\GeneralUtility;
 use TYPO3\CMS\Extbase\Configuration\ConfigurationManager;
 use TYPO3\CMS\Extbase\Configuration\ConfigurationManagerInterface;
@@ -46,16 +44,16 @@ class ImportDatesTask extends AbstractTask
        $cacheService = GeneralUtility::makeInstance(CacheService::class);

        // Acquire the export only when it is newer than the one we last imported
-        // (SFTP mode: download to a staging dir; local mode: fileadmin/xmlexport).
-        // An acquisition failure (SFTP connect/auth/transfer) is left to propagate
-        // so the scheduler marks the task failed.
-        $path = $importSource->acquire();
+        // (SFTP mode: download to a staging dir; local mode: fileadmin/xmlexport;
+        // API mode: walk the MyE&P export pages). An acquisition failure is left
+        // to propagate so the scheduler marks the task failed.
+        $import = $importSource->acquire();

-        if ($path !== null) {
-            $importService->import(new XmlProductReader($path, GeneralUtility::makeInstance(LogManager::class)->getLogger(XmlProductReader::class)));
+        if ($import !== null) {
+            $importService->import($import->getRows());
            // Advance the "last imported" bookkeeping only after a successful
            // import, so a crashed import retries on the next run.
-            $importSource->markImported($path);
+            $importSource->markImported($import);
        }

        $settings = $this->getSettings();
@@ -52,7 +52,7 @@ epTeamApiTimeout = 5
 # cat=EpTeamAPI; type=int; label=MyE&P Team API max total request duration in seconds (0 = unlimited)
 epTeamApiMaxDuration = 15

-# cat=SftpImport; type=options[Remote SFTP server=sftp,Local directory (fileadmin/xmlexport)=local]; label=Product/date import source
+# cat=SftpImport; type=options[Remote SFTP server=sftp,Local directory (fileadmin/xmlexport)=local,MyE&P API=api]; label=Product/date import source
 productImportSource = sftp

 # cat=SftpImport; type=string; label=SFTP host for product/date import