Files
zyt/server/app/common/service/pharmacy/EjMedicineCatalogSyncWorkflow.php
T
2026-07-22 14:50:34 +08:00

96 lines
3.1 KiB
PHP

<?php
declare(strict_types=1);
namespace app\common\service\pharmacy;
use DomainException;
use RuntimeException;
use Throwable;
final class EjMedicineCatalogSyncWorkflow
{
private $acquireLock;
private $releaseLock;
private $loadCursor;
private $fetchPage;
private $mergePage;
private $markSuccess;
private $markFailure;
public function __construct(
callable $acquireLock,
callable $releaseLock,
callable $loadCursor,
callable $fetchPage,
callable $mergePage,
callable $markSuccess,
callable $markFailure
) {
$this->acquireLock = $acquireLock;
$this->releaseLock = $releaseLock;
$this->loadCursor = $loadCursor;
$this->fetchPage = $fetchPage;
$this->mergePage = $mergePage;
$this->markSuccess = $markSuccess;
$this->markFailure = $markFailure;
}
/** @return array{pages:int,pulled:int,received:int,created:int,updated:int,unchanged:int,deactivated:int,cursor:int} */
public function sync(int $limit = 200): array
{
if (!(bool) ($this->acquireLock)()) {
throw new DomainException('洛阳药房目录正在同步,请稍后重试');
}
$cursor = 0;
$stats = [
'pages' => 0,
'pulled' => 0,
'received' => 0,
'created' => 0,
'updated' => 0,
'unchanged' => 0,
'deactivated' => 0,
'cursor' => $cursor,
];
try {
$cursor = max(0, (int) ($this->loadCursor)());
$stats['cursor'] = $cursor;
for ($page = 0; $page < 1000; ++$page) {
$parsed = EjMedicineCatalogSyncPolicy::parsePage(
($this->fetchPage)($cursor, min(max($limit, 1), 500)),
$cursor
);
$merged = ($this->mergePage)($parsed['items'], $parsed['next_cursor']);
++$stats['pages'];
$pulled = count($parsed['items']);
$stats['pulled'] += $pulled;
$stats['received'] += $pulled;
foreach (['created', 'updated', 'unchanged', 'deactivated'] as $key) {
$stats[$key] += (int) ($merged[$key] ?? 0);
}
$cursor = $parsed['next_cursor'];
$stats['cursor'] = $cursor;
if (!$parsed['has_more']) {
($this->markSuccess)($cursor, $stats);
return $stats;
}
}
throw new RuntimeException('洛阳药房药材目录分页超过安全上限');
} catch (Throwable $exception) {
($this->markFailure)($cursor, self::summarizeError($exception->getMessage()));
throw $exception;
} finally {
($this->releaseLock)();
}
}
public static function summarizeError(string $message): string
{
$message = preg_replace('/(app[_-]?secret|signature|token|authorization)\s*[:=]\s*[^\s,;]+/i', '$1=[redacted]', $message) ?? $message;
return mb_substr(trim($message), 0, 500);
}
}