feat(transport) : synchronisation du référentiel transporteurs QUALIMAT (ERP-39)
Commande console app:qualimat:sync : récupère les opérateurs de transport agréés depuis l'API publique qualimat.org, normalise et synchronise une table référentielle (upsert sur le SIRET + soft-delete des absents + journal). Prévue pour un cron quotidien. - migration : tables qualimat_carrier + qualimat_sync_log (COMMENT ON COLUMN sur chaque colonne) - QualimatRowMapper : normalisation pure (SIRET sans espaces, date dd/mm/yyyy -> ISO, skip sans SIRET) + tests unitaires - SyncQualimatCommand : options --file / --ppp / --dry-run, upsert DBAL transactionnel - activation de framework.http_client - tests fonctionnels de la commande (upsert/normalisation/journal/soft-delete)
This commit is contained in:
@@ -0,0 +1,124 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace App\Module\Transport\Application\Qualimat;
|
||||
|
||||
/**
|
||||
* Mapping pur d'un item brut de l'API QUALIMAT vers une ligne normalisee
|
||||
* prete a l'upsert dans `qualimat_carrier`. Sans dependance (testable en
|
||||
* isolation). Voir ERP-39 § 2 pour les pieges qualite de la source.
|
||||
*/
|
||||
final class QualimatRowMapper
|
||||
{
|
||||
/**
|
||||
* Mappe un lot d'items. Les items sans SIRET exploitable sont ignores et
|
||||
* comptes a part (cf. `rows_skipped` du journal).
|
||||
*
|
||||
* @param array<int, array<string, mixed>> $items
|
||||
*
|
||||
* @return array{rows: list<array<string, mixed>>, skipped: int}
|
||||
*/
|
||||
public static function mapMany(array $items): array
|
||||
{
|
||||
$rows = [];
|
||||
$skipped = 0;
|
||||
|
||||
foreach ($items as $item) {
|
||||
$row = self::mapOne($item);
|
||||
|
||||
if (null === $row) {
|
||||
++$skipped;
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
$rows[] = $row;
|
||||
}
|
||||
|
||||
return ['rows' => $rows, 'skipped' => $skipped];
|
||||
}
|
||||
|
||||
/**
|
||||
* Mappe un item unique. Retourne null si le SIRET est absent ou vide
|
||||
* (ligne inexploitable : pas de cle naturelle pour l'upsert).
|
||||
*
|
||||
* @param array<string, mixed> $item
|
||||
*
|
||||
* @return null|array<string, mixed>
|
||||
*/
|
||||
public static function mapOne(array $item): ?array
|
||||
{
|
||||
$siret = self::normalizeSiret(self::str($item['Siret'] ?? null));
|
||||
|
||||
if (null === $siret) {
|
||||
return null;
|
||||
}
|
||||
|
||||
return [
|
||||
'siret' => $siret,
|
||||
// Nom et Societe sont identiques a la source : une seule colonne.
|
||||
'name' => self::str($item['Nom'] ?? null) ?? '',
|
||||
'address' => self::str($item['Adresse'] ?? null),
|
||||
'postal_code' => self::str($item['CodePostal'] ?? null),
|
||||
'city' => self::str($item['Ville'] ?? null),
|
||||
'phone' => self::str($item['Telephone_1'] ?? null),
|
||||
'department' => self::str($item['Departement'] ?? null),
|
||||
// Statut conserve brut (feed externe, valeurs non contraintes).
|
||||
'status' => self::str($item['Statut'] ?? null) ?? '',
|
||||
'validity_date' => self::parseDate(self::str($item['Validite'] ?? null)),
|
||||
];
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalise un SIRET : ne conserve que les chiffres. Null si vide.
|
||||
* La source est "sale" (longueurs variables 7 a 14) : aucune contrainte
|
||||
* de longueur, on stocke les chiffres tels quels.
|
||||
*/
|
||||
public static function normalizeSiret(?string $raw): ?string
|
||||
{
|
||||
if (null === $raw) {
|
||||
return null;
|
||||
}
|
||||
|
||||
$digits = preg_replace('/\D+/', '', $raw) ?? '';
|
||||
|
||||
return '' === $digits ? null : $digits;
|
||||
}
|
||||
|
||||
/**
|
||||
* Convertit une date "dd/mm/yyyy" en "yyyy-mm-dd". Null si le format ne
|
||||
* correspond pas ou si la date n'est pas un jour calendaire valide
|
||||
* (garde-fou : evite un INSERT en erreur sur une date impossible).
|
||||
*/
|
||||
public static function parseDate(?string $raw): ?string
|
||||
{
|
||||
if (null === $raw || !preg_match('#^(\d{2})/(\d{2})/(\d{4})$#', $raw, $m)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
$day = (int) $m[1];
|
||||
$month = (int) $m[2];
|
||||
$year = (int) $m[3];
|
||||
|
||||
if (!checkdate($month, $day, $year)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
return sprintf('%04d-%02d-%02d', $year, $month, $day);
|
||||
}
|
||||
|
||||
/**
|
||||
* Trim d'une valeur scalaire ; null si la chaine resultante est vide.
|
||||
*/
|
||||
private static function str(mixed $value): ?string
|
||||
{
|
||||
if (null === $value) {
|
||||
return null;
|
||||
}
|
||||
|
||||
$trimmed = trim((string) $value);
|
||||
|
||||
return '' === $trimmed ? null : $trimmed;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,250 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace App\Module\Transport\Infrastructure\Console;
|
||||
|
||||
use App\Module\Transport\Application\Qualimat\QualimatRowMapper;
|
||||
use DateTimeImmutable;
|
||||
use Doctrine\DBAL\Connection;
|
||||
use RuntimeException;
|
||||
use Symfony\Component\Console\Attribute\AsCommand;
|
||||
use Symfony\Component\Console\Command\Command;
|
||||
use Symfony\Component\Console\Input\InputInterface;
|
||||
use Symfony\Component\Console\Input\InputOption;
|
||||
use Symfony\Component\Console\Output\OutputInterface;
|
||||
use Symfony\Component\Console\Style\SymfonyStyle;
|
||||
use Symfony\Contracts\HttpClient\HttpClientInterface;
|
||||
use Throwable;
|
||||
|
||||
use function array_slice;
|
||||
use function count;
|
||||
use function is_array;
|
||||
|
||||
use const JSON_THROW_ON_ERROR;
|
||||
|
||||
/**
|
||||
* ERP-39 : synchronise le referentiel des transporteurs QUALIMAT.
|
||||
*
|
||||
* Recupere la liste des operateurs de transport depuis l'API publique (ou un
|
||||
* fichier local), normalise chaque ligne et synchronise `qualimat_carrier` de
|
||||
* facon transactionnelle : upsert sur le SIRET, soft-delete des absents,
|
||||
* journal dans `qualimat_sync_log`. Idempotente (refresh complet) : prevue
|
||||
* pour un cron quotidien.
|
||||
*/
|
||||
#[AsCommand(
|
||||
name: 'app:qualimat:sync',
|
||||
description: 'Synchronise le referentiel des transporteurs QUALIMAT (upsert + soft-delete + journal).',
|
||||
)]
|
||||
final class SyncQualimatCommand extends Command
|
||||
{
|
||||
private const string API_URL = 'https://www.qualimat.org/wp-json/qualimat/v1/getOperateurs';
|
||||
private const int DEFAULT_PPP = 10000;
|
||||
|
||||
public function __construct(
|
||||
private readonly Connection $connection,
|
||||
private readonly HttpClientInterface $httpClient,
|
||||
) {
|
||||
parent::__construct();
|
||||
}
|
||||
|
||||
protected function configure(): void
|
||||
{
|
||||
$this
|
||||
->addOption('file', null, InputOption::VALUE_REQUIRED, "Chemin d'un JSON local (court-circuite l'appel HTTP, utile pour tests/rejeu).")
|
||||
->addOption('ppp', null, InputOption::VALUE_REQUIRED, "Taille de page demandee a l'API.", (string) self::DEFAULT_PPP)
|
||||
->addOption('dry-run', null, InputOption::VALUE_NONE, 'Analyse sans ecriture en base.')
|
||||
;
|
||||
}
|
||||
|
||||
protected function execute(InputInterface $input, OutputInterface $output): int
|
||||
{
|
||||
$io = new SymfonyStyle($input, $output);
|
||||
$ppp = max(1, (int) $input->getOption('ppp'));
|
||||
$dryRun = (bool) $input->getOption('dry-run');
|
||||
$file = $input->getOption('file');
|
||||
|
||||
// 1. Recuperation des items (fichier local ou API).
|
||||
try {
|
||||
$items = null !== $file ? $this->readLocal((string) $file) : $this->fetchRemote($ppp);
|
||||
} catch (Throwable $e) {
|
||||
$io->error('Recuperation impossible : '.$e->getMessage());
|
||||
|
||||
return Command::FAILURE;
|
||||
}
|
||||
|
||||
$total = count($items);
|
||||
$io->section(sprintf('QUALIMAT — %d items recus', $total));
|
||||
|
||||
// Garde-fou troncature : un retour egal a ppp signale un dataset coupe.
|
||||
if (null === $file && $total === $ppp) {
|
||||
$io->warning(sprintf("Le nombre d'items recus (%d) egale --ppp : resultat potentiellement tronque, augmente --ppp.", $ppp));
|
||||
}
|
||||
|
||||
// 2. Mapping / normalisation (les items sans SIRET sont ignores).
|
||||
['rows' => $rows, 'skipped' => $skipped] = QualimatRowMapper::mapMany($items);
|
||||
$io->writeln(sprintf('%d lignes exploitables, %d ignorees (sans SIRET).', count($rows), $skipped));
|
||||
|
||||
if ($dryRun) {
|
||||
$this->renderPreview($io, $rows);
|
||||
$io->note(sprintf('Dry-run : aucune ecriture. (%d lignes au total)', count($rows)));
|
||||
|
||||
return Command::SUCCESS;
|
||||
}
|
||||
|
||||
// 3. Sync transactionnelle : upsert -> soft-delete -> journal.
|
||||
$run = new DateTimeImmutable()->format('Y-m-d H:i:s.u');
|
||||
|
||||
$this->connection->beginTransaction();
|
||||
|
||||
try {
|
||||
$upserted = $this->upsertAll($rows, $run);
|
||||
$deactivated = $this->deactivateMissing($run);
|
||||
$this->log($run, $total, $upserted, $skipped, $deactivated);
|
||||
$this->connection->commit();
|
||||
} catch (Throwable $e) {
|
||||
$this->connection->rollBack();
|
||||
$io->error('Sync annulee (rollback) : '.$e->getMessage());
|
||||
|
||||
return Command::FAILURE;
|
||||
}
|
||||
|
||||
$io->success(sprintf('%d upsert, %d ignore(s), %d desactive(s).', $upserted, $skipped, $deactivated));
|
||||
|
||||
return Command::SUCCESS;
|
||||
}
|
||||
|
||||
/**
|
||||
* Rejoue l'appel GET de l'API QUALIMAT et retourne le tableau d'items.
|
||||
*
|
||||
* @return array<int, array<string, mixed>>
|
||||
*/
|
||||
private function fetchRemote(int $ppp): array
|
||||
{
|
||||
$response = $this->httpClient->request('GET', self::API_URL, [
|
||||
'query' => ['type' => 'operateur_transport', 'ppp' => $ppp],
|
||||
'timeout' => 60,
|
||||
]);
|
||||
|
||||
// toArray() leve une exception sur un statut non-2xx ou un corps non-JSON.
|
||||
$data = $response->toArray();
|
||||
|
||||
return array_is_list($data) ? $data : [];
|
||||
}
|
||||
|
||||
/**
|
||||
* Lit un export JSON local (tableau d'objets).
|
||||
*
|
||||
* @return array<int, array<string, mixed>>
|
||||
*/
|
||||
private function readLocal(string $path): array
|
||||
{
|
||||
$raw = @file_get_contents($path);
|
||||
|
||||
if (false === $raw) {
|
||||
throw new RuntimeException(sprintf('Fichier illisible : %s', $path));
|
||||
}
|
||||
|
||||
$data = json_decode($raw, true, 512, JSON_THROW_ON_ERROR);
|
||||
|
||||
if (!is_array($data) || !array_is_list($data)) {
|
||||
throw new RuntimeException("Le JSON doit etre un tableau d'objets.");
|
||||
}
|
||||
|
||||
return $data;
|
||||
}
|
||||
|
||||
/**
|
||||
* Upsert de toutes les lignes valides (cle naturelle = siret). Marque
|
||||
* is_active=TRUE et tamponne last_synced_at avec le run courant.
|
||||
*
|
||||
* @param list<array<string, mixed>> $rows
|
||||
*/
|
||||
private function upsertAll(array $rows, string $run): int
|
||||
{
|
||||
$sql = <<<'SQL'
|
||||
INSERT INTO qualimat_carrier
|
||||
(siret, name, address, postal_code, city, phone, department, status, validity_date, is_active, last_synced_at)
|
||||
VALUES
|
||||
(:siret, :name, :address, :postal_code, :city, :phone, :department, :status, :validity_date, TRUE, :run)
|
||||
ON CONFLICT (siret) DO UPDATE SET
|
||||
name = EXCLUDED.name,
|
||||
address = EXCLUDED.address,
|
||||
postal_code = EXCLUDED.postal_code,
|
||||
city = EXCLUDED.city,
|
||||
phone = EXCLUDED.phone,
|
||||
department = EXCLUDED.department,
|
||||
status = EXCLUDED.status,
|
||||
validity_date = EXCLUDED.validity_date,
|
||||
is_active = TRUE,
|
||||
last_synced_at = EXCLUDED.last_synced_at
|
||||
SQL;
|
||||
|
||||
$count = 0;
|
||||
|
||||
foreach ($rows as $r) {
|
||||
$this->connection->executeStatement($sql, [
|
||||
'siret' => $r['siret'],
|
||||
'name' => $r['name'],
|
||||
'address' => $r['address'],
|
||||
'postal_code' => $r['postal_code'],
|
||||
'city' => $r['city'],
|
||||
'phone' => $r['phone'],
|
||||
'department' => $r['department'],
|
||||
'status' => $r['status'],
|
||||
'validity_date' => $r['validity_date'],
|
||||
'run' => $run,
|
||||
]);
|
||||
++$count;
|
||||
}
|
||||
|
||||
return $count;
|
||||
}
|
||||
|
||||
/**
|
||||
* Soft-delete : toute ligne active non revue par ce run (tampon anterieur)
|
||||
* passe a is_active=false.
|
||||
*/
|
||||
private function deactivateMissing(string $run): int
|
||||
{
|
||||
return (int) $this->connection->executeStatement(
|
||||
'UPDATE qualimat_carrier SET is_active = FALSE WHERE is_active = TRUE AND last_synced_at < :run',
|
||||
['run' => $run],
|
||||
);
|
||||
}
|
||||
|
||||
private function log(string $run, int $total, int $upserted, int $skipped, int $deactivated): void
|
||||
{
|
||||
$this->connection->executeStatement(
|
||||
<<<'SQL'
|
||||
INSERT INTO qualimat_sync_log (fetched_at, rows_total, rows_upserted, rows_skipped, rows_deactivated)
|
||||
VALUES (:run, :total, :upserted, :skipped, :deactivated)
|
||||
SQL,
|
||||
[
|
||||
'run' => $run,
|
||||
'total' => $total,
|
||||
'upserted' => $upserted,
|
||||
'skipped' => $skipped,
|
||||
'deactivated' => $deactivated,
|
||||
],
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param list<array<string, mixed>> $rows
|
||||
*/
|
||||
private function renderPreview(SymfonyStyle $io, array $rows): void
|
||||
{
|
||||
$io->table(
|
||||
['SIRET', 'Nom', 'CP', 'Ville', 'Statut', 'Validite'],
|
||||
array_map(static fn (array $r): array => [
|
||||
(string) $r['siret'],
|
||||
mb_strimwidth((string) $r['name'], 0, 40, '…'),
|
||||
(string) ($r['postal_code'] ?? ''),
|
||||
mb_strimwidth((string) ($r['city'] ?? ''), 0, 25, '…'),
|
||||
(string) $r['status'],
|
||||
(string) ($r['validity_date'] ?? ''),
|
||||
], array_slice($rows, 0, 15)),
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user