mirror of
https://gitlab.com/Chill-Projet/chill-bundles.git
synced 2025-08-30 11:33:49 +00:00
Feature: Create a base importer for geographical units and add index
This commit is contained in:
@@ -0,0 +1,234 @@
|
||||
<?php
|
||||
|
||||
/**
|
||||
* Chill is a software for social workers
|
||||
*
|
||||
* For the full copyright and license information, please view
|
||||
* the LICENSE file that was distributed with this source code.
|
||||
*/
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace Chill\MainBundle\Service\Import;
|
||||
|
||||
use Doctrine\DBAL\Connection;
|
||||
use Doctrine\DBAL\Statement;
|
||||
use Doctrine\DBAL\Types\Type;
|
||||
use Doctrine\DBAL\Types\Types;
|
||||
use Exception;
|
||||
use LogicException;
|
||||
use Psr\Log\LoggerInterface;
|
||||
use function array_key_exists;
|
||||
use function count;
|
||||
|
||||
final class GeographicalUnitBaseImporter
|
||||
{
|
||||
private const INSERT = <<<'SQL'
|
||||
INSERT INTO geographical_unit_temp
|
||||
(layerKey, layerName, unitName, unitKey, geom)
|
||||
SELECT
|
||||
i.layerKey, i.layerName, i.unitName, i.unitKey,
|
||||
ST_Transform(ST_setSrid(ST_GeomFromText(i.wkt), i.srid), 4326)
|
||||
FROM
|
||||
(VALUES
|
||||
{{ values }}
|
||||
) AS i (layerKey, layerName, unitName, unitKey, wkt, srid)
|
||||
SQL;
|
||||
|
||||
private const LOG_PREFIX = '[GeographicalUnitBAseImporter] ';
|
||||
|
||||
private const VALUE = '(?, ?::jsonb, ?, ?, ?, ?::int)';
|
||||
|
||||
/**
|
||||
* @var array<int, Statement>
|
||||
*/
|
||||
private array $cachingStatements = [];
|
||||
|
||||
private Connection $defaultConnection;
|
||||
|
||||
private bool $isInitialized = false;
|
||||
|
||||
private LoggerInterface $logger;
|
||||
|
||||
private array $waitingForInsert = [];
|
||||
|
||||
public function __construct(Connection $defaultConnection, LoggerInterface $logger)
|
||||
{
|
||||
$this->defaultConnection = $defaultConnection;
|
||||
$this->logger = $logger;
|
||||
}
|
||||
|
||||
public function finalize(): void
|
||||
{
|
||||
$this->doInsertPending();
|
||||
|
||||
$this->prepareForFinalize();
|
||||
|
||||
$this->updateGeographicalUnitTable();
|
||||
|
||||
$this->deleteTemporaryTable();
|
||||
|
||||
$this->isInitialized = false;
|
||||
}
|
||||
|
||||
public function importUnit(
|
||||
string $layerKey,
|
||||
array $layerName,
|
||||
string $unitName,
|
||||
string $unitKey,
|
||||
string $geomAsWKT,
|
||||
int $srid = null
|
||||
): void {
|
||||
$this->initialize();
|
||||
|
||||
$this->waitingForInsert[] = [
|
||||
'layerKey' => $layerKey,
|
||||
'layerName' => $layerName,
|
||||
'unitName' => $unitName,
|
||||
'unitKey' => $unitKey,
|
||||
'geomAsWKT' => $geomAsWKT,
|
||||
'srid' => $srid
|
||||
];
|
||||
|
||||
if (100 <= count($this->waitingForInsert)) {
|
||||
$this->doInsertPending();
|
||||
}
|
||||
}
|
||||
|
||||
private function createTemporaryTable(): void
|
||||
{
|
||||
$this->defaultConnection->executeStatement("CREATE TEMPORARY TABLE geographical_unit_temp (
|
||||
layerKey TEXT DEFAULT '' NOT NULL,
|
||||
layerName JSONB DEFAULT '[]'::jsonb NOT NULL,
|
||||
unitName TEXT default '' NOT NULL,
|
||||
unitKey TEXT default '' NOT NULL,
|
||||
geom GEOMETRY(MULTIPOLYGON, 4326)
|
||||
)");
|
||||
|
||||
$this->defaultConnection->executeStatement('SET work_mem TO \'50MB\'');
|
||||
}
|
||||
|
||||
private function deleteTemporaryTable(): void
|
||||
{
|
||||
$this->defaultConnection->executeStatement('DROP TABLE IF EXISTS geographical_unit_temp');
|
||||
}
|
||||
|
||||
private function doInsertPending(): void
|
||||
{
|
||||
$forNumber = count($this->waitingForInsert);
|
||||
|
||||
if (0 === $forNumber) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (!array_key_exists($forNumber, $this->cachingStatements)) {
|
||||
$sql = strtr(self::INSERT, [
|
||||
'{{ values }}' => implode(
|
||||
', ',
|
||||
array_fill(0, $forNumber, self::VALUE)
|
||||
),
|
||||
]);
|
||||
|
||||
$this->logger->debug(self::LOG_PREFIX . ' generated sql for insert', [
|
||||
'sql' => $sql,
|
||||
'forNumber' => $forNumber,
|
||||
]);
|
||||
|
||||
$this->cachingStatements[$forNumber] = $this->defaultConnection->prepare($sql);
|
||||
}
|
||||
|
||||
$this->logger->debug(self::LOG_PREFIX . ' inserting pending addresses', [
|
||||
'number' => $forNumber,
|
||||
'first' => $this->waitingForInsert[0] ?? null,
|
||||
]);
|
||||
|
||||
$statement = $this->cachingStatements[$forNumber];
|
||||
try {
|
||||
$i = 0;
|
||||
foreach ($this->waitingForInsert as $insert) {
|
||||
$statement->bindValue(++$i, $insert['layerKey'], Types::STRING);
|
||||
$statement->bindValue(++$i, $insert['layerName'], Types::JSON);
|
||||
$statement->bindValue(++$i, $insert['unitName'], Types::STRING);
|
||||
$statement->bindValue(++$i, $insert['unitKey'], Types::STRING);
|
||||
$statement->bindValue(++$i, $insert['geomAsWKT'], Types::STRING);
|
||||
$statement->bindValue(++$i, $insert['srid'], Types::INTEGER);
|
||||
}
|
||||
|
||||
$affected = $statement->executeStatement();
|
||||
|
||||
if ($affected === 0) {
|
||||
throw new \RuntimeException('no row affected');
|
||||
}
|
||||
} catch (Exception $e) {
|
||||
throw $e;
|
||||
} finally {
|
||||
$this->waitingForInsert = [];
|
||||
}
|
||||
}
|
||||
|
||||
private function initialize(): void
|
||||
{
|
||||
if ($this->isInitialized) {
|
||||
return;
|
||||
}
|
||||
|
||||
$this->deleteTemporaryTable();
|
||||
$this->createTemporaryTable();
|
||||
$this->isInitialized = true;
|
||||
}
|
||||
|
||||
private function prepareForFinalize(): void
|
||||
{
|
||||
$this->defaultConnection->executeStatement(
|
||||
'CREATE INDEX idx_ref_add_temp ON geographical_unit_temp (unitKey)'
|
||||
);
|
||||
}
|
||||
|
||||
private function updateGeographicalUnitTable(): void
|
||||
{
|
||||
$this->defaultConnection->transactional(
|
||||
function() {
|
||||
// 0) create new layers
|
||||
$this->defaultConnection->executeStatement(
|
||||
"
|
||||
WITH unique_layers AS (
|
||||
SELECT DISTINCT layerKey, layerName FROM geographical_unit_temp
|
||||
)
|
||||
INSERT INTO chill_main_geographical_unit_layer (id, name, refid)
|
||||
SELECT
|
||||
nextval('chill_main_geographical_unit_layer_id_seq'),
|
||||
layerName,
|
||||
layerKey
|
||||
FROM unique_layers
|
||||
ON CONFLICT (refid)
|
||||
DO UPDATE SET name=EXCLUDED.name
|
||||
");
|
||||
|
||||
//1) Add new units
|
||||
$this->logger->info(self::LOG_PREFIX . 'upsert new units');
|
||||
$affected = $this->defaultConnection->executeStatement("INSERT INTO chill_main_geographical_unit
|
||||
(id, geom, unitname, layer_id, unitrefid)
|
||||
SELECT
|
||||
nextval('chill_main_geographical_unit_id_seq'),
|
||||
geom,
|
||||
unitName,
|
||||
layer.id,
|
||||
unitKey
|
||||
FROM geographical_unit_temp JOIN chill_main_geographical_unit_layer AS layer ON layer.refid = layerKey
|
||||
ON CONFLICT (unitrefid)
|
||||
DO UPDATE
|
||||
SET geom = EXCLUDED.geom, unitname = EXCLUDED.unitname
|
||||
");
|
||||
$this->logger->info(self::LOG_PREFIX . 'units upserted', ['upserted' => $affected]);
|
||||
|
||||
//3) Delete units
|
||||
$this->logger->info(self::LOG_PREFIX . 'soft delete adresses');
|
||||
$affected = $this->defaultConnection->executeStatement('DELETE FROM chill_main_geographical_unit
|
||||
WHERE
|
||||
unitrefid NOT IN (SELECT distinct unitKey FROM geographical_unit_temp)
|
||||
');
|
||||
$this->logger->info(self::LOG_PREFIX . 'addresses deleted', ['deleted' => $affected]);
|
||||
}
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user