ZYT/zyt
2
0
Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

feat: add non-destructive EJ medicine sync

Browse Source
This commit is contained in:
long
2026-09-10 11:09:26 +08:00
parent 27fbef9321
commit 17e9e7b6b6
6 changed files with 1087 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/app/command/EjPharmacyPushMedicines.php
+76
View File
@@ -0,0 +1,76 @@
<?php
declare(strict_types=1);
namespace app\command;
use app\common\service\pharmacy\EjMedicineIncrementalPushService;
use think\console\Command;
use think\console\Input;
use think\console\input\Option;
use think\console\Output;
final class EjPharmacyPushMedicines extends Command
{
protected function configure()
{
$this->setName('ej-pharmacy:push-medicines')
->setDescription('非破坏性增量推送 ZYT 药材到恩济药房；保留双方已有药材')
->addOption('apply', null, Option::VALUE_NONE, '执行远端增量导入；缺省仅做 dry-run')
->addOption('confirm', null, Option::VALUE_OPTIONAL, '增量写入确认令牌：INCREMENTAL_NO_DELETE', '')
->addOption('batch-size', null, Option::VALUE_OPTIONAL, '每批药材数量（1-500）', 100)
->addOption('run-id', null, Option::VALUE_OPTIONAL, '幂等运行标识（缺省 zyt-incremental，最长 25 位）', '');
}
protected function execute(Input $input, Output $output)
{
try {
$apply = (bool) $input->getOption('apply');
$confirm = (string) $input->getOption('confirm');
$batchSize = (int) $input->getOption('batch-size');
EjMedicineIncrementalPushService::assertCommandGate($apply, $confirm, $batchSize);
if (!$apply) {
$plan = EjMedicineIncrementalPushService::plan($batchSize);
$output->writeln(sprintf(
'dry-run source=%d candidates=%d batches=%d mapped=%d unmapped=%d '
. 'preserved_inactive=%d remote_delete=0 local_delete=0',
$plan['source_count'],
$plan['candidate_count'],
$plan['batch_count'],
$plan['mapped_count'],
$plan['unmapped_count'],
$plan['preserved_inactive']
));
return 0;
}
$result = EjMedicineIncrementalPushService::execute(
$batchSize,
(string) $input->getOption('run-id')
);
$output->writeln(sprintf(
'incremental 完成 run_id=%s source=%d candidates=%d batches=%d '
. 'remote_created=%d remote_existing=%d preserved_inactive=%d '
. 'catalog_created=%d catalog_updated=%d mapping_created=%d mapping_updated=%d '
. 'mapping_unchanged=%d remote_delete=0 local_delete=0',
$result['run_id'],
$result['source_count'],
$result['candidate_count'],
$result['batch_count'],
$result['remote_created'],
$result['remote_existing'],
$result['preserved_inactive'],
$result['catalog_created'],
$result['catalog_updated'],
$result['mapping_created'],
$result['mapping_updated'],
$result['mapping_unchanged']
));
return 0;
} catch (\Throwable $exception) {
$output->error($exception->getMessage());
return 1;
}
}
}
server/app/common/service/pharmacy/EjMedicineIncrementalPushService.php
+590
View File
@@ -0,0 +1,590 @@
<?php
declare(strict_types=1);
namespace app\common\service\pharmacy;
use DomainException;
use InvalidArgumentException;
use RuntimeException;
use think\facade\Db;
final class EjMedicineIncrementalPushService
{
private const SOURCE_SYSTEM = 'zyt';
private const APPLY_CONFIRMATION = 'INCREMENTAL_NO_DELETE';
private const DEFAULT_RUN_ID = 'zyt-incremental';
private const STATE_ID = 1;
private const LOCK_TTL = 120;
public static function assertCommandGate(bool $apply, string $confirm, int $batchSize): void
{
self::assertBatchSize($batchSize);
if ($apply && !hash_equals(self::APPLY_CONFIRMATION, $confirm)) {
throw new InvalidArgumentException(
'执行增量写入必须提供 --confirm=' . self::APPLY_CONFIRMATION
);
}
}
/**
* @param null|callable():array<int,array<string,mixed>> $sourceLoader
* @param null|callable(array<int,int>):array<int,array<string,mixed>> $mappingLoader
* @return array{source_count:int,candidate_count:int,batch_count:int,mapped_count:int,unmapped_count:int,preserved_inactive:int,remote_delete_count:int,local_delete_count:int}
*/
public static function plan(
int $batchSize = 100,
?callable $sourceLoader = null,
?callable $mappingLoader = null
): array {
self::assertBatchSize($batchSize);
$sourceRows = $sourceLoader === null ? self::loadSourceRows() : $sourceLoader();
$items = EjMedicineBootstrapItem::fromRows($sourceRows);
$localIds = array_map('intval', array_column($items, 'source_medicine_id'));
$mappingRows = $mappingLoader === null
? self::loadMappingRows($localIds)
: $mappingLoader($localIds);
$selection = self::selectCandidates($items, $mappingRows);
return [
'source_count' => count($items),
'candidate_count' => count($selection['candidates']),
'batch_count' => (int) ceil(count($selection['candidates']) / $batchSize),
'mapped_count' => $selection['mapped_count'],
'unmapped_count' => $selection['unmapped_count'],
'preserved_inactive' => $selection['preserved_inactive'],
'remote_delete_count' => 0,
'local_delete_count' => 0,
];
}
/**
* Incrementally imports every active local medicine through EJ's idempotent
* source identity and upserts only the returned ZYT projection rows.
* Existing EJ-only medicines and unrelated local projections are untouched.
*
* @param null|callable():array<int,array<string,mixed>> $sourceLoader
* @param null|callable(array<string,mixed>):array<string,mixed> $importer
* @param null|callable(array<int,array<string,mixed>>):array<string,int> $projectionUpserter
* @param null|callable(array<int,int>):array<int,array<string,mixed>> $mappingLoader
* @param null|callable(string):bool $lockAcquirer
* @param null|callable(string):bool $lockRenewer
* @param null|callable(string):void $lockReleaser
* @return array<string,int|string>
*/
public static function execute(
int $batchSize = 100,
string $runId = '',
?callable $sourceLoader = null,
?callable $importer = null,
?callable $projectionUpserter = null,
?callable $mappingLoader = null,
?callable $lockAcquirer = null,
?callable $lockRenewer = null,
?callable $lockReleaser = null
): array {
self::assertBatchSize($batchSize);
$customLockCallbacks = count(array_filter(
[$lockAcquirer, $lockRenewer, $lockReleaser],
static fn (?callable $callback): bool => $callback !== null
));
if ($customLockCallbacks !== 0 && $customLockCallbacks !== 3) {
throw new InvalidArgumentException('增量同步锁回调必须同时提供 acquire、renew 和 release');
}
$lockAcquirer ??= static fn (string $token): bool => self::acquireLock($token);
$lockRenewer ??= static fn (string $token): bool => self::renewLock($token);
$lockReleaser ??= static function (string $token): void {
self::releaseLock($token);
};
$lockToken = bin2hex(random_bytes(16));
if (!$lockAcquirer($lockToken)) {
throw new DomainException('恩济药材同步正在执行，请稍后重试');
}
try {
return self::executeLocked(
$batchSize,
$runId,
$sourceLoader,
$importer,
$projectionUpserter,
$mappingLoader,
$lockRenewer,
$lockToken
);
} finally {
$lockReleaser($lockToken);
}
}
/**
* @param null|callable():array<int,array<string,mixed>> $sourceLoader
* @param null|callable(array<string,mixed>):array<string,mixed> $importer
* @param null|callable(array<int,array<string,mixed>>):array<string,int> $projectionUpserter
* @param null|callable(array<int,int>):array<int,array<string,mixed>> $mappingLoader
* @param callable(string):bool $lockRenewer
* @return array<string,int|string>
*/
private static function executeLocked(
int $batchSize,
string $runId,
?callable $sourceLoader,
?callable $importer,
?callable $projectionUpserter,
?callable $mappingLoader,
callable $lockRenewer,
string $lockToken
): array {
$sourceRows = $sourceLoader === null ? self::loadSourceRows() : $sourceLoader();
$allItems = EjMedicineBootstrapItem::fromRows($sourceRows);
$localIds = array_map('intval', array_column($allItems, 'source_medicine_id'));
$mappingRows = $mappingLoader === null
? self::loadMappingRows($localIds)
: $mappingLoader($localIds);
$selection = self::selectCandidates($allItems, $mappingRows);
$items = $selection['candidates'];
$runId = self::normalizeRunId($runId);
$batches = self::buildBatches($items, $batchSize, $runId);
if ($items !== [] && $importer === null) {
if (!EjPharmacyClient::isConfigured()) {
throw new RuntimeException('恩济药房接口未启用或配置不完整');
}
$client = new EjPharmacyClient();
$importer = static fn (array $payload): array => $client->importMedicines($payload);
}
$sourceById = [];
foreach ($items as $item) {
$sourceById[(string) $item['source_medicine_id']] = $item;
}
$seenCodes = [];
$seenVersions = [];
$projectionRows = [];
$remoteCreated = 0;
$remoteExisting = 0;
foreach ($batches as $payload) {
self::assertLockLease($lockRenewer, $lockToken);
$response = $importer($payload);
self::assertLockLease($lockRenewer, $lockToken);
$responseItems = self::validateImportResponse(
$response,
$payload,
$seenCodes,
$seenVersions
);
foreach ($responseItems as $responseItem) {
$sourceId = (string) $responseItem['source_medicine_id'];
$source = $sourceById[$sourceId] ?? null;
if (!is_array($source)) {
throw new RuntimeException("恩济增量导入返回未知 source_medicine_id：{$sourceId}");
}
$action = (string) $responseItem['action'];
$medicineCode = (string) $responseItem['medicine_code'];
$expectedCode = $selection['active_mapping_codes'][$sourceId] ?? null;
if ($expectedCode !== null && !hash_equals($expectedCode, $medicineCode)) {
throw new DomainException(
"本地药材 {$sourceId} 的启用映射编码 {$expectedCode} 与恩济返回 {$medicineCode} 不一致"
);
}
$remoteCreated += $action === 'created' ? 1 : 0;
$remoteExisting += $action === 'existing' ? 1 : 0;
$projectionRows[] = [
'local_medicine_id' => $sourceId,
'medicine_code' => $medicineCode,
'name' => (string) $source['name'],
'brand' => (string) ($source['brand'] ?? ''),
'unit' => (string) $source['unit'],
'settlement_price' => (string) $source['settlement_price'],
'retail_price' => (string) $source['retail_price'],
'status' => (int) $source['status'],
'catalog_version' => (int) $responseItem['catalog_version'],
];
}
}
$projectionStats = [
'catalog_created' => 0,
'catalog_updated' => 0,
'mapping_created' => 0,
'mapping_updated' => 0,
'mapping_unchanged' => 0,
];
if ($projectionRows !== []) {
self::assertLockLease($lockRenewer, $lockToken);
$projectionStats = $projectionUpserter === null
? self::upsertProjection($projectionRows, $lockToken)
: $projectionUpserter($projectionRows);
}
return array_merge([
'run_id' => $runId,
'source_count' => count($allItems),
'candidate_count' => count($items),
'batch_count' => count($batches),
'preserved_inactive' => $selection['preserved_inactive'],
'remote_created' => $remoteCreated,
'remote_existing' => $remoteExisting,
'remote_delete_count' => 0,
'local_delete_count' => 0,
], $projectionStats);
}
/**
* @param list<array<string,mixed>> $items
* @return list<array{source_system:string,import_id:string,items:array<int,array<string,mixed>>}>
*/
public static function buildBatches(array $items, int $batchSize, string $runId): array
{
self::assertBatchSize($batchSize);
$runId = self::normalizeRunId($runId);
usort($items, static fn (array $left, array $right): int => EjMedicineBootstrapItem::compareIds(
(string) ($left['source_medicine_id'] ?? ''),
(string) ($right['source_medicine_id'] ?? '')
));
$batches = [];
foreach (array_chunk($items, $batchSize) as $index => $batchItems) {
$contentJson = json_encode(
$batchItems,
JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES
);
$batches[] = [
'source_system' => self::SOURCE_SYSTEM,
'import_id' => sprintf(
'%s-%04d-%s',
$runId,
$index + 1,
substr(hash('sha256', $contentJson), 0, 32)
),
'items' => $batchItems,
];
}
return $batches;
}
private static function assertBatchSize(int $batchSize): void
{
if ($batchSize < 1 || $batchSize > 500) {
throw new InvalidArgumentException('--batch-size 必须在 1 到 500 之间');
}
}
private static function normalizeRunId(string $runId): string
{
$runId = trim($runId);
if ($runId === '') {
$runId = self::DEFAULT_RUN_ID;
}
if (strlen($runId) > 25 || preg_match('/^[A-Za-z0-9][A-Za-z0-9._-]*$/D', $runId) !== 1) {
throw new InvalidArgumentException('--run-id 必须为不超过 25 位的字母、数字、点、下划线或短横线');
}
return $runId;
}
/** @return array<int,array<string,mixed>> */
private static function loadSourceRows(): array
{
return Db::name('doctor_medicine')
->field('id,name,unit,settlement_price,retail_price,status')
->where('status', 1)
->whereNull('delete_time')
->order('id', 'asc')
->select()
->toArray();
}
/** @param array<int,int> $localIds @return array<int,array<string,mixed>> */
private static function loadMappingRows(array $localIds): array
{
if ($localIds === []) {
return [];
}
return Db::name('ej_medicine_mapping')
->whereIn('local_medicine_id', $localIds)
->field('id,local_medicine_id,medicine_code,status,delete_time')
->order('local_medicine_id', 'asc')
->select()
->toArray();
}
/**
* Active mappings are replayed so EJ can recover a missing remote medicine.
* A medicine with an inactive or soft-deleted mapping is intentionally
* excluded; an operator decision must never be undone by synchronization.
*
* @param list<array<string,mixed>> $items
* @param array<int,array<string,mixed>> $mappingRows
* @return array{candidates:list<array<string,mixed>>,mapped_count:int,unmapped_count:int,preserved_inactive:int,active_mapping_codes:array<string,string>}
*/
private static function selectCandidates(array $items, array $mappingRows): array
{
$byLocalId = [];
foreach ($mappingRows as $mapping) {
$localId = (int) ($mapping['local_medicine_id'] ?? 0);
if ($localId < 1 || isset($byLocalId[$localId])) {
throw new DomainException("本地药材 {$localId} 存在重复的恩济映射记录");
}
$byLocalId[$localId] = $mapping;
}
$candidates = [];
$mappedCount = 0;
$unmappedCount = 0;
$preservedInactive = 0;
$activeMappingCodes = [];
foreach ($items as $item) {
$localId = (int) $item['source_medicine_id'];
$mapping = $byLocalId[$localId] ?? null;
if ($mapping === null) {
++$unmappedCount;
$candidates[] = $item;
continue;
}
if ((int) ($mapping['status'] ?? 0) === 1 && ($mapping['delete_time'] ?? null) === null) {
$medicineCode = trim((string) ($mapping['medicine_code'] ?? ''));
if ($medicineCode === '') {
throw new DomainException("本地药材 {$localId} 的启用恩济映射编码为空");
}
++$mappedCount;
$candidates[] = $item;
$activeMappingCodes[(string) $localId] = $medicineCode;
continue;
}
++$preservedInactive;
}
return [
'candidates' => $candidates,
'mapped_count' => $mappedCount,
'unmapped_count' => $unmappedCount,
'preserved_inactive' => $preservedInactive,
'active_mapping_codes' => $activeMappingCodes,
];
}
/**
* @param array<string,mixed> $response
* @param array<string,mixed> $payload
* @param array<string,bool> $seenCodes
* @param array<int,bool> $seenVersions
* @return list<array{source_medicine_id:string,medicine_code:string,catalog_version:int,action:string}>
*/
private static function validateImportResponse(
array $response,
array $payload,
array &$seenCodes,
array &$seenVersions
): array {
$nextSeenCodes = $seenCodes;
$nextSeenVersions = $seenVersions;
$items = EjMedicineBootstrapService::validateImportResponse(
$response,
$payload,
$nextSeenCodes,
$nextSeenVersions
);
$data = $response['body']['data'] ?? null;
if (!is_array($data) || !hash_equals(self::SOURCE_SYSTEM, (string) ($data['source_system'] ?? ''))) {
throw new RuntimeException('恩济药材导入响应 source_system 不匹配');
}
$expectedPayloadHash = hash('sha256', json_encode(
self::canonicalize($payload),
JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES
));
$actualPayloadHash = strtolower(trim((string) ($data['payload_hash'] ?? '')));
if (
preg_match('/^[a-f0-9]{64}$/D', $actualPayloadHash) !== 1
|| !hash_equals($expectedPayloadHash, $actualPayloadHash)
) {
throw new RuntimeException('恩济药材导入响应 payload_hash 不匹配');
}
$seenCodes = $nextSeenCodes;
$seenVersions = $nextSeenVersions;
return $items;
}
private static function canonicalize(mixed $value): mixed
{
if (!is_array($value)) {
return $value;
}
if (array_is_list($value)) {
return array_map([self::class, 'canonicalize'], $value);
}
ksort($value, SORT_STRING);
foreach ($value as $key => $child) {
$value[$key] = self::canonicalize($child);
}
return $value;
}
/** @param callable(string):bool $lockRenewer */
private static function assertLockLease(callable $lockRenewer, string $lockToken): void
{
if (!$lockRenewer($lockToken)) {
throw new DomainException('恩济药材同步锁已失效，请重试');
}
}
private static function acquireLock(string $token): bool
{
$now = time();
return Db::name('ej_pharmacy_sync_state')
->where('id', self::STATE_ID)
->where(function ($query) use ($now): void {
$query->where('lock_token', '')->whereOr('lock_expires_at', '<', $now);
})
->update([
'lock_token' => $token,
'lock_expires_at' => $now + self::LOCK_TTL,
'update_time' => $now,
]) === 1;
}
private static function renewLock(string $token): bool
{
$now = time();
$query = Db::name('ej_pharmacy_sync_state')
->where('id', self::STATE_ID)
->where('lock_token', $token)
->where('lock_expires_at', '>=', $now);
$updated = $query->update([
'lock_expires_at' => $now + self::LOCK_TTL,
'update_time' => $now,
]);
if ($updated === 1) {
return true;
}
$state = Db::name('ej_pharmacy_sync_state')
->where('id', self::STATE_ID)
->where('lock_token', $token)
->find();
return is_array($state) && (int) ($state['lock_expires_at'] ?? 0) >= $now;
}
private static function releaseLock(string $token): void
{
Db::name('ej_pharmacy_sync_state')
->where('id', self::STATE_ID)
->where('lock_token', $token)
->update([
'lock_token' => '',
'lock_expires_at' => 0,
'update_time' => time(),
]);
}
/**
* @param array<int,array<string,mixed>> $rows
* @return array{catalog_created:int,catalog_updated:int,mapping_created:int,mapping_updated:int,mapping_unchanged:int}
*/
private static function upsertProjection(array $rows, string $lockToken): array
{
return Db::transaction(static function () use ($rows, $lockToken): array {
$stats = [
'catalog_created' => 0,
'catalog_updated' => 0,
'mapping_created' => 0,
'mapping_updated' => 0,
'mapping_unchanged' => 0,
];
$state = Db::name('ej_pharmacy_sync_state')
->where('id', self::STATE_ID)
->lock(true)
->find();
if (
!is_array($state)
|| !hash_equals($lockToken, (string) ($state['lock_token'] ?? ''))
|| (int) ($state['lock_expires_at'] ?? 0) < time()
) {
throw new DomainException('恩济药材同步锁已失效，请重试');
}
$now = time();
foreach ($rows as $row) {
$localId = (int) ($row['local_medicine_id'] ?? 0);
$medicineCode = trim((string) ($row['medicine_code'] ?? ''));
if ($localId < 1 || $medicineCode === '') {
throw new RuntimeException('恩济增量导入投影缺少本地药材 ID 或 medicine_code');
}
$conflict = Db::name('ej_medicine_mapping')
->where('medicine_code', $medicineCode)
->where('local_medicine_id', '<>', $localId)
->lock(true)
->find();
if ($conflict) {
throw new DomainException(
"恩济药材编码 {$medicineCode} 已映射到本地药材 " . (int) $conflict['local_medicine_id']
);
}
$catalogValues = [
'name' => (string) $row['name'],
'brand' => (string) ($row['brand'] ?? ''),
'unit' => (string) $row['unit'],
'settlement_price' => (string) $row['settlement_price'],
'retail_price' => (string) $row['retail_price'],
'status' => (int) $row['status'],
'catalog_version' => (int) $row['catalog_version'],
'remote_deleted' => 0,
'update_time' => $now,
];
$catalog = Db::name('ej_medicine_catalog')
->where('medicine_code', $medicineCode)
->lock(true)
->find();
if ($catalog) {
Db::name('ej_medicine_catalog')->where('id', (int) $catalog['id'])->update($catalogValues);
++$stats['catalog_updated'];
} else {
Db::name('ej_medicine_catalog')->insert($catalogValues + [
'medicine_code' => $medicineCode,
'create_time' => $now,
]);
++$stats['catalog_created'];
}
$mapping = Db::name('ej_medicine_mapping')
->where('local_medicine_id', $localId)
->lock(true)
->find();
$mappingValues = [
'medicine_code' => $medicineCode,
'status' => 1,
'operator_id' => 0,
'operator_name' => 'system-incremental',
'delete_time' => null,
'update_time' => $now,
];
if (!$mapping) {
Db::name('ej_medicine_mapping')->insert($mappingValues + [
'local_medicine_id' => $localId,
'create_time' => $now,
]);
++$stats['mapping_created'];
} elseif ((int) $mapping['status'] !== 1 || ($mapping['delete_time'] ?? null) !== null) {
throw new DomainException("本地药材 {$localId} 的恩济映射已被停用，增量同步保持该状态不变");
} elseif (hash_equals((string) $mapping['medicine_code'], $medicineCode)) {
++$stats['mapping_unchanged'];
} else {
throw new DomainException(
"本地药材 {$localId} 的启用映射编码 "
. (string) $mapping['medicine_code']
. " 与恩济返回 {$medicineCode} 不一致，增量同步未改写该映射"
);
}
}
return $stats;
});
}
}